summary refs log tree commit diff
path: root/synapse/handlers
diff options
context:
space:
mode:
Diffstat (limited to 'synapse/handlers')
-rw-r--r--synapse/handlers/account_data.py21
-rw-r--r--synapse/handlers/admin.py33
-rw-r--r--synapse/handlers/events.py20
-rw-r--r--synapse/handlers/federation.py28
-rw-r--r--synapse/handlers/identity.py14
-rw-r--r--synapse/handlers/message.py62
-rw-r--r--synapse/handlers/presence.py20
-rw-r--r--synapse/handlers/profile.py28
-rw-r--r--synapse/handlers/register.py18
-rw-r--r--synapse/handlers/room.py31
-rw-r--r--synapse/handlers/search.py15
-rw-r--r--synapse/handlers/sync.py82
12 files changed, 252 insertions, 120 deletions
diff --git a/synapse/handlers/account_data.py b/synapse/handlers/account_data.py
index 1d35d3b7dc..fe773bee9b 100644
--- a/synapse/handlers/account_data.py
+++ b/synapse/handlers/account_data.py
@@ -29,9 +29,10 @@ class AccountDataEventSource(object):
         last_stream_id = from_key
 
         current_stream_id = yield self.store.get_max_account_data_stream_id()
-        tags = yield self.store.get_updated_tags(user_id, last_stream_id)
 
         results = []
+        tags = yield self.store.get_updated_tags(user_id, last_stream_id)
+
         for room_id, room_tags in tags.items():
             results.append({
                 "type": "m.tag",
@@ -39,6 +40,24 @@ class AccountDataEventSource(object):
                 "room_id": room_id,
             })
 
+        account_data, room_account_data = (
+            yield self.store.get_updated_account_data_for_user(user_id, last_stream_id)
+        )
+
+        for account_data_type, content in account_data.items():
+            results.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        for room_id, account_data in room_account_data.items():
+            for account_data_type, content in account_data.items():
+                results.append({
+                    "type": account_data_type,
+                    "content": content,
+                    "room_id": room_id,
+                })
+
         defer.returnValue((results, current_stream_id))
 
     @defer.inlineCallbacks
diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py
index d852a18555..04fa58df65 100644
--- a/synapse/handlers/admin.py
+++ b/synapse/handlers/admin.py
@@ -30,34 +30,27 @@ class AdminHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def get_whois(self, user):
-        res = yield self.store.get_user_ip_and_agents(user)
-
-        d = {}
-        for r in res:
-            # Note that device_id is always None
-            device = d.setdefault(r["device_id"], {})
-            session = device.setdefault(r["access_token"], [])
-            session.append({
-                "ip": r["ip"],
-                "user_agent": r["user_agent"],
-                "last_seen": r["last_seen"],
+        connections = []
+
+        sessions = yield self.store.get_user_ip_and_agents(user)
+        for session in sessions:
+            connections.append({
+                "ip": session["ip"],
+                "last_seen": session["last_seen"],
+                "user_agent": session["user_agent"],
             })
 
         ret = {
             "user_id": user.to_string(),
-            "devices": [
-                {
-                    "device_id": k,
+            "devices": {
+                "": {
                     "sessions": [
                         {
-                            # "access_token": x, TODO (erikj)
-                            "connections": y,
+                            "connections": connections,
                         }
-                        for x, y in v.items()
                     ]
-                }
-                for k, v in d.items()
-            ],
+                },
+            },
         }
 
         defer.returnValue(ret)
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 0e4c0d4d06..fe300433e6 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -28,6 +28,18 @@ import random
 logger = logging.getLogger(__name__)
 
 
+def started_user_eventstream(distributor, user):
+    return distributor.fire("started_user_eventstream", user)
+
+
+def stopped_user_eventstream(distributor, user):
+    return distributor.fire("stopped_user_eventstream", user)
+
+
+def user_joined_room(distributor, user, room_id):
+    return distributor.fire("user_joined_room", user, room_id)
+
+
 class EventStreamHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -66,7 +78,7 @@ class EventStreamHandler(BaseHandler):
                 except:
                     logger.exception("Failed to cancel event timer")
             else:
-                yield self.distributor.fire("started_user_eventstream", user)
+                yield started_user_eventstream(self.distributor, user)
 
         self._streams_per_user[user] += 1
 
@@ -89,7 +101,7 @@ class EventStreamHandler(BaseHandler):
 
                 self._stop_timer_per_user.pop(user, None)
 
-                return self.distributor.fire("stopped_user_eventstream", user)
+                return stopped_user_eventstream(self.distributor, user)
 
             logger.debug("Scheduling _later: for %s", user)
             self._stop_timer_per_user[user] = (
@@ -120,9 +132,7 @@ class EventStreamHandler(BaseHandler):
                 timeout = random.randint(int(timeout*0.9), int(timeout*1.1))
 
             if is_guest:
-                yield self.distributor.fire(
-                    "user_joined_room", user=auth_user, room_id=room_id
-                )
+                yield user_joined_room(self.distributor, auth_user, room_id)
 
             events, tokens = yield self.notifier.get_events_for(
                 auth_user, pagin_config, timeout,
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index c1bce07e31..2855f2d7c3 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -44,6 +44,10 @@ import logging
 logger = logging.getLogger(__name__)
 
 
+def user_joined_room(distributor, user, room_id):
+    return distributor.fire("user_joined_room", user, room_id)
+
+
 class FederationHandler(BaseHandler):
     """Handles events that originated from federation.
         Responsible for:
@@ -60,10 +64,7 @@ class FederationHandler(BaseHandler):
 
         self.hs = hs
 
-        self.distributor.observe(
-            "user_joined_room",
-            self._on_user_joined
-        )
+        self.distributor.observe("user_joined_room", self.user_joined_room)
 
         self.waiting_for_join_list = {}
 
@@ -176,7 +177,7 @@ class FederationHandler(BaseHandler):
                 )
 
             try:
-                _, event_stream_id, max_stream_id = yield self._handle_new_event(
+                context, event_stream_id, max_stream_id = yield self._handle_new_event(
                     origin,
                     event,
                     state=state,
@@ -233,10 +234,13 @@ class FederationHandler(BaseHandler):
 
         if event.type == EventTypes.Member:
             if event.membership == Membership.JOIN:
-                user = UserID.from_string(event.state_key)
-                yield self.distributor.fire(
-                    "user_joined_room", user=user, room_id=event.room_id
-                )
+                prev_state = context.current_state.get((event.type, event.state_key))
+                if not prev_state or prev_state.membership != Membership.JOIN:
+                    # Only fire user_joined_room if the user has acutally
+                    # joined the room. Don't bother if the user is just
+                    # changing their profile info.
+                    user = UserID.from_string(event.state_key)
+                    yield user_joined_room(self.distributor, user, event.room_id)
 
     @defer.inlineCallbacks
     def _filter_events_for_server(self, server_name, room_id, events):
@@ -733,9 +737,7 @@ class FederationHandler(BaseHandler):
         if event.type == EventTypes.Member:
             if event.content["membership"] == Membership.JOIN:
                 user = UserID.from_string(event.state_key)
-                yield self.distributor.fire(
-                    "user_joined_room", user=user, room_id=event.room_id
-                )
+                yield user_joined_room(self.distributor, user, event.room_id)
 
         new_pdu = event
 
@@ -1082,7 +1084,7 @@ class FederationHandler(BaseHandler):
         return self.store.get_min_depth(context)
 
     @log_function
-    def _on_user_joined(self, user, room_id):
+    def user_joined_room(self, user, room_id):
         waiters = self.waiting_for_join_list.get(
             (user.to_string(), room_id),
             []
diff --git a/synapse/handlers/identity.py b/synapse/handlers/identity.py
index 2a99921d5f..f1fa562fff 100644
--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -20,7 +20,6 @@ from synapse.api.errors import (
     CodeMessageException
 )
 from ._base import BaseHandler
-from synapse.http.client import SimpleHttpClient
 from synapse.util.async import run_on_reactor
 from synapse.api.errors import SynapseError
 
@@ -35,13 +34,12 @@ class IdentityHandler(BaseHandler):
     def __init__(self, hs):
         super(IdentityHandler, self).__init__(hs)
 
+        self.http_client = hs.get_simple_http_client()
+
     @defer.inlineCallbacks
     def threepid_from_creds(self, creds):
         yield run_on_reactor()
 
-        # TODO: get this from the homeserver rather than creating a new one for
-        # each request
-        http_client = SimpleHttpClient(self.hs)
         # XXX: make this configurable!
         # trustedIdServers = ['matrix.org', 'localhost:8090']
         trustedIdServers = ['matrix.org', 'vector.im']
@@ -67,7 +65,7 @@ class IdentityHandler(BaseHandler):
 
         data = {}
         try:
-            data = yield http_client.get_json(
+            data = yield self.http_client.get_json(
                 "https://%s%s" % (
                     id_server,
                     "/_matrix/identity/api/v1/3pid/getValidated3pid"
@@ -85,7 +83,6 @@ class IdentityHandler(BaseHandler):
     def bind_threepid(self, creds, mxid):
         yield run_on_reactor()
         logger.debug("binding threepid %r to %s", creds, mxid)
-        http_client = SimpleHttpClient(self.hs)
         data = None
 
         if 'id_server' in creds:
@@ -103,7 +100,7 @@ class IdentityHandler(BaseHandler):
             raise SynapseError(400, "No client_secret in creds")
 
         try:
-            data = yield http_client.post_urlencoded_get_json(
+            data = yield self.http_client.post_urlencoded_get_json(
                 "https://%s%s" % (
                     id_server, "/_matrix/identity/api/v1/3pid/bind"
                 ),
@@ -121,7 +118,6 @@ class IdentityHandler(BaseHandler):
     @defer.inlineCallbacks
     def requestEmailToken(self, id_server, email, client_secret, send_attempt, **kwargs):
         yield run_on_reactor()
-        http_client = SimpleHttpClient(self.hs)
 
         params = {
             'email': email,
@@ -131,7 +127,7 @@ class IdentityHandler(BaseHandler):
         params.update(kwargs)
 
         try:
-            data = yield http_client.post_urlencoded_get_json(
+            data = yield self.http_client.post_urlencoded_get_json(
                 "https://%s%s" % (
                     id_server,
                     "/_matrix/identity/api/v1/validate/email/requestToken"
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 64c57375f7..ccdd3d8473 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -26,11 +26,17 @@ from synapse.types import UserID, RoomStreamToken, StreamToken
 
 from ._base import BaseHandler
 
+from canonicaljson import encode_canonical_json
+
 import logging
 
 logger = logging.getLogger(__name__)
 
 
+def collect_presencelike_data(distributor, user, content):
+    return distributor.fire("collect_presencelike_data", user, content)
+
+
 class MessageHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -195,10 +201,8 @@ class MessageHandler(BaseHandler):
             if membership == Membership.JOIN:
                 joinee = UserID.from_string(builder.state_key)
                 # If event doesn't include a display name, add one.
-                yield self.distributor.fire(
-                    "collect_presencelike_data",
-                    joinee,
-                    builder.content
+                yield collect_presencelike_data(
+                    self.distributor, joinee, builder.content
                 )
 
         if token_id is not None:
@@ -211,6 +215,16 @@ class MessageHandler(BaseHandler):
             builder=builder,
         )
 
+        if event.is_state():
+            prev_state = context.current_state.get((event.type, event.state_key))
+            if prev_state and event.user_id == prev_state.user_id:
+                prev_content = encode_canonical_json(prev_state.content)
+                next_content = encode_canonical_json(event.content)
+                if prev_content == next_content:
+                    # Duplicate suppression for state updates with same sender
+                    # and content.
+                    defer.returnValue(prev_state)
+
         if event.type == EventTypes.Member:
             member_handler = self.hs.get_handlers().room_member_handler
             yield member_handler.change_membership(event, context, is_guest=is_guest)
@@ -359,6 +373,10 @@ class MessageHandler(BaseHandler):
 
         tags_by_room = yield self.store.get_tags_for_user(user_id)
 
+        account_data, account_data_by_room = (
+            yield self.store.get_account_data_for_user(user_id)
+        )
+
         public_room_ids = yield self.store.get_public_room_ids()
 
         limit = pagin_config.limit
@@ -436,14 +454,22 @@ class MessageHandler(BaseHandler):
                     for c in current_state.values()
                 ]
 
-                account_data = []
+                account_data_events = []
                 tags = tags_by_room.get(event.room_id)
                 if tags:
-                    account_data.append({
+                    account_data_events.append({
                         "type": "m.tag",
                         "content": {"tags": tags},
                     })
-                d["account_data"] = account_data
+
+                account_data = account_data_by_room.get(event.room_id, {})
+                for account_data_type, content in account_data.items():
+                    account_data_events.append({
+                        "type": account_data_type,
+                        "content": content,
+                    })
+
+                d["account_data"] = account_data_events
             except:
                 logger.exception("Failed to get snapshot")
 
@@ -456,9 +482,17 @@ class MessageHandler(BaseHandler):
                 consumeErrors=True
             ).addErrback(unwrapFirstError)
 
+        account_data_events = []
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
         ret = {
             "rooms": rooms_ret,
             "presence": presence,
+            "account_data": account_data_events,
             "receipts": receipt,
             "end": now_token.to_string(),
         }
@@ -498,14 +532,22 @@ class MessageHandler(BaseHandler):
                 user_id, room_id, pagin_config, membership, member_event_id, is_guest
             )
 
-        account_data = []
+        account_data_events = []
         tags = yield self.store.get_tags_for_room(user_id, room_id)
         if tags:
-            account_data.append({
+            account_data_events.append({
                 "type": "m.tag",
                 "content": {"tags": tags},
             })
-        result["account_data"] = account_data
+
+        account_data = yield self.store.get_account_data_for_room(user_id, room_id)
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        result["account_data"] = account_data_events
 
         defer.returnValue(result)
 
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index aca65096fc..63d6f30a7b 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -62,6 +62,14 @@ def partitionbool(l, func):
     return ret.get(True, []), ret.get(False, [])
 
 
+def user_presence_changed(distributor, user, statuscache):
+    return distributor.fire("user_presence_changed", user, statuscache)
+
+
+def collect_presencelike_data(distributor, user, content):
+    return distributor.fire("collect_presencelike_data", user, content)
+
+
 class PresenceHandler(BaseHandler):
 
     STATE_LEVELS = {
@@ -361,9 +369,7 @@ class PresenceHandler(BaseHandler):
         yield self.store.set_presence_state(
             target_user.localpart, state_to_store
         )
-        yield self.distributor.fire(
-            "collect_presencelike_data", target_user, state
-        )
+        yield collect_presencelike_data(self.distributor, target_user, state)
 
         if now_level > was_level:
             state["last_active"] = self.clock.time_msec()
@@ -467,7 +473,7 @@ class PresenceHandler(BaseHandler):
             )
 
     @defer.inlineCallbacks
-    def send_invite(self, observer_user, observed_user):
+    def send_presence_invite(self, observer_user, observed_user):
         """Request the presence of a local or remote user for a local user"""
         if not self.hs.is_mine(observer_user):
             raise SynapseError(400, "User is not hosted on this Home Server")
@@ -878,7 +884,7 @@ class PresenceHandler(BaseHandler):
             room_ids=room_ids,
             statuscache=statuscache,
         )
-        yield self.distributor.fire("user_presence_changed", user, statuscache)
+        yield user_presence_changed(self.distributor, user, statuscache)
 
     @defer.inlineCallbacks
     def incoming_presence(self, origin, content):
@@ -1116,9 +1122,7 @@ class PresenceHandler(BaseHandler):
                     self._user_cachemap[user].get_state()["last_active"]
                 )
 
-            yield self.distributor.fire(
-                "collect_presencelike_data", user, state
-            )
+            yield collect_presencelike_data(self.distributor, user, state)
 
         if "last_active" in state:
             state = dict(state)
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index 799faffe53..576c6f09b4 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -28,6 +28,14 @@ import logging
 logger = logging.getLogger(__name__)
 
 
+def changed_presencelike_data(distributor, user, state):
+    return distributor.fire("changed_presencelike_data", user, state)
+
+
+def collect_presencelike_data(distributor, user, content):
+    return distributor.fire("collect_presencelike_data", user, content)
+
+
 class ProfileHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -95,11 +103,9 @@ class ProfileHandler(BaseHandler):
             target_user.localpart, new_displayname
         )
 
-        yield self.distributor.fire(
-            "changed_presencelike_data", target_user, {
-                "displayname": new_displayname,
-            }
-        )
+        yield changed_presencelike_data(self.distributor, target_user, {
+            "displayname": new_displayname,
+        })
 
         yield self._update_join_states(target_user)
 
@@ -144,11 +150,9 @@ class ProfileHandler(BaseHandler):
             target_user.localpart, new_avatar_url
         )
 
-        yield self.distributor.fire(
-            "changed_presencelike_data", target_user, {
-                "avatar_url": new_avatar_url,
-            }
-        )
+        yield changed_presencelike_data(self.distributor, target_user, {
+            "avatar_url": new_avatar_url,
+        })
 
         yield self._update_join_states(target_user)
 
@@ -208,9 +212,7 @@ class ProfileHandler(BaseHandler):
                 "membership": Membership.JOIN,
             }
 
-            yield self.distributor.fire(
-                "collect_presencelike_data", user, content
-            )
+            yield collect_presencelike_data(self.distributor, user, content)
 
             msg_handler = self.hs.get_handlers().message_handler
             try:
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index 493a087031..a037da0f70 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -31,6 +31,10 @@ import urllib
 logger = logging.getLogger(__name__)
 
 
+def registered_user(distributor, user):
+    return distributor.fire("registered_user", user)
+
+
 class RegistrationHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -38,6 +42,7 @@ class RegistrationHandler(BaseHandler):
 
         self.distributor = hs.get_distributor()
         self.distributor.declare("registered_user")
+        self.captch_client = CaptchaServerHttpClient(hs)
 
     @defer.inlineCallbacks
     def check_username(self, localpart):
@@ -98,7 +103,7 @@ class RegistrationHandler(BaseHandler):
                 password_hash=password_hash
             )
 
-            yield self.distributor.fire("registered_user", user)
+            yield registered_user(self.distributor, user)
         else:
             # autogen a random user ID
             attempts = 0
@@ -117,7 +122,7 @@ class RegistrationHandler(BaseHandler):
                         token=token,
                         password_hash=password_hash)
 
-                    self.distributor.fire("registered_user", user)
+                    yield registered_user(self.distributor, user)
                 except SynapseError:
                     # if user id is taken, just generate another
                     user_id = None
@@ -167,7 +172,7 @@ class RegistrationHandler(BaseHandler):
             token=token,
             password_hash=""
         )
-        self.distributor.fire("registered_user", user)
+        registered_user(self.distributor, user)
         defer.returnValue((user_id, token))
 
     @defer.inlineCallbacks
@@ -215,7 +220,7 @@ class RegistrationHandler(BaseHandler):
                 token=token,
                 password_hash=None
             )
-            yield self.distributor.fire("registered_user", user)
+            yield registered_user(self.distributor, user)
         except Exception, e:
             yield self.store.add_access_token_to_user(user_id, token)
             # Ignore Registration errors
@@ -302,10 +307,7 @@ class RegistrationHandler(BaseHandler):
         """
         Used only by c/s api v1
         """
-        # TODO: get this from the homeserver rather than creating a new one for
-        # each request
-        client = CaptchaServerHttpClient(self.hs)
-        data = yield client.post_urlencoded_get_raw(
+        data = yield self.captcha_client.post_urlencoded_get_raw(
             "http://www.google.com:80/recaptcha/api/verify",
             args={
                 'privatekey': private_key,
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 023b4001b8..116a998c42 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -41,6 +41,18 @@ logger = logging.getLogger(__name__)
 id_server_scheme = "https://"
 
 
+def collect_presencelike_data(distributor, user, content):
+    return distributor.fire("collect_presencelike_data", user, content)
+
+
+def user_left_room(distributor, user, room_id):
+    return distributor.fire("user_left_room", user=user, room_id=room_id)
+
+
+def user_joined_room(distributor, user, room_id):
+    return distributor.fire("user_joined_room", user=user, room_id=room_id)
+
+
 class RoomCreationHandler(BaseHandler):
 
     PRESETS_DICT = {
@@ -438,9 +450,7 @@ class RoomMemberHandler(BaseHandler):
 
             if prev_state and prev_state.membership == Membership.JOIN:
                 user = UserID.from_string(event.user_id)
-                self.distributor.fire(
-                    "user_left_room", user=user, room_id=event.room_id
-                )
+                user_left_room(self.distributor, user, event.room_id)
 
         defer.returnValue({"room_id": room_id})
 
@@ -458,9 +468,7 @@ class RoomMemberHandler(BaseHandler):
             raise SynapseError(404, "No known servers")
 
         # If event doesn't include a display name, add one.
-        yield self.distributor.fire(
-            "collect_presencelike_data", joinee, content
-        )
+        yield collect_presencelike_data(self.distributor, joinee, content)
 
         content.update({"membership": Membership.JOIN})
         builder = self.event_builder_factory.new({
@@ -517,10 +525,13 @@ class RoomMemberHandler(BaseHandler):
                 do_auth=do_auth,
             )
 
-        user = UserID.from_string(event.user_id)
-        yield self.distributor.fire(
-            "user_joined_room", user=user, room_id=room_id
-        )
+        prev_state = context.current_state.get((event.type, event.state_key))
+        if not prev_state or prev_state.membership != Membership.JOIN:
+            # Only fire user_joined_room if the user has acutally joined the
+            # room. Don't bother if the user is just changing their profile
+            # info.
+            user = UserID.from_string(event.user_id)
+            yield user_joined_room(self.distributor, user, room_id)
 
     @defer.inlineCallbacks
     def get_inviter(self, event):
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 0e58404148..bc79564287 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -240,9 +240,18 @@ class SearchHandler(BaseHandler):
                 last_event_id = room_events[-1].event_id
                 pagination_token = results_map[last_event_id]["pagination_token"]
 
-                global_next_batch = encode_base64("%s\n%s\n%s" % (
-                    "all", "", pagination_token
-                ))
+                # We want to respect the given batch group and group keys so
+                # that if people blindly use the top level `next_batch` token
+                # it returns more from the same group (if applicable) rather
+                # than reverting to searching all results again.
+                if batch_group and batch_group_key:
+                    global_next_batch = encode_base64("%s\n%s\n%s" % (
+                        batch_group, batch_group_key, pagination_token
+                    ))
+                else:
+                    global_next_batch = encode_base64("%s\n%s\n%s" % (
+                        "all", "", pagination_token
+                    ))
 
                 for room_id, group in room_groups.items():
                     group["next_batch"] = encode_base64("%s\n%s\n%s" % (
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 877328b29e..24c2b2fad6 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -100,6 +100,7 @@ class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
 class SyncResult(collections.namedtuple("SyncResult", [
     "next_batch",  # Token for the next sync
     "presence",  # List of presence events for the user.
+    "account_data",  # List of account_data events for the user.
     "joined",  # JoinedSyncResult for each joined room.
     "invited",  # InvitedSyncResult for each invited room.
     "archived",  # ArchivedSyncResult for each archived room.
@@ -185,13 +186,19 @@ class SyncHandler(BaseHandler):
             pagination_config=pagination_config.get_source_config("presence"),
             key=None
         )
+
+        membership_list = (Membership.INVITE, Membership.JOIN)
+        if sync_config.filter.include_leave:
+            membership_list += (Membership.LEAVE, Membership.BAN)
+
         room_list = yield self.store.get_rooms_for_user_where_membership_is(
             user_id=sync_config.user.to_string(),
-            membership_list=(
-                Membership.INVITE,
-                Membership.JOIN,
-                Membership.LEAVE,
-                Membership.BAN
+            membership_list=membership_list
+        )
+
+        account_data, account_data_by_room = (
+            yield self.store.get_account_data_for_user(
+                sync_config.user.to_string()
             )
         )
 
@@ -211,6 +218,7 @@ class SyncHandler(BaseHandler):
                     timeline_since_token=timeline_since_token,
                     ephemeral_by_room=ephemeral_by_room,
                     tags_by_room=tags_by_room,
+                    account_data_by_room=account_data_by_room,
                 )
                 joined.append(room_sync)
             elif event.membership == Membership.INVITE:
@@ -230,11 +238,13 @@ class SyncHandler(BaseHandler):
                     leave_token=leave_token,
                     timeline_since_token=timeline_since_token,
                     tags_by_room=tags_by_room,
+                    account_data_by_room=account_data_by_room,
                 )
                 archived.append(room_sync)
 
         defer.returnValue(SyncResult(
             presence=presence,
+            account_data=self.account_data_for_user(account_data),
             joined=joined,
             invited=invited,
             archived=archived,
@@ -244,7 +254,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def full_state_sync_for_joined_room(self, room_id, sync_config,
                                         now_token, timeline_since_token,
-                                        ephemeral_by_room, tags_by_room):
+                                        ephemeral_by_room, tags_by_room,
+                                        account_data_by_room):
         """Sync a room for a client which is starting without any state
         Returns:
             A Deferred JoinedSyncResult.
@@ -262,19 +273,38 @@ class SyncHandler(BaseHandler):
             state=current_state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         ))
 
-    def account_data_for_room(self, room_id, tags_by_room):
-        account_data = []
+    def account_data_for_user(self, account_data):
+        account_data_events = []
+
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        return account_data_events
+
+    def account_data_for_room(self, room_id, tags_by_room, account_data_by_room):
+        account_data_events = []
         tags = tags_by_room.get(room_id)
         if tags is not None:
-            account_data.append({
+            account_data_events.append({
                 "type": "m.tag",
                 "content": {"tags": tags},
             })
-        return account_data
+
+        account_data = account_data_by_room.get(room_id, {})
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        return account_data_events
 
     @defer.inlineCallbacks
     def ephemeral_by_room(self, sync_config, now_token, since_token=None):
@@ -341,7 +371,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def full_state_sync_for_archived_room(self, room_id, sync_config,
                                           leave_event_id, leave_token,
-                                          timeline_since_token, tags_by_room):
+                                          timeline_since_token, tags_by_room,
+                                          account_data_by_room):
         """Sync a room for a client which is starting without any state
         Returns:
             A Deferred JoinedSyncResult.
@@ -358,7 +389,7 @@ class SyncHandler(BaseHandler):
             timeline=batch,
             state=leave_state,
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         ))
 
@@ -415,6 +446,13 @@ class SyncHandler(BaseHandler):
             since_token.account_data_key,
         )
 
+        account_data, account_data_by_room = (
+            yield self.store.get_updated_account_data_for_user(
+                sync_config.user.to_string(),
+                since_token.account_data_key,
+            )
+        )
+
         joined = []
         archived = []
         if len(room_events) <= timeline_limit:
@@ -469,7 +507,7 @@ class SyncHandler(BaseHandler):
                     state=state,
                     ephemeral=ephemeral_by_room.get(room_id, []),
                     account_data=self.account_data_for_room(
-                        room_id, tags_by_room
+                        room_id, tags_by_room, account_data_by_room
                     ),
                 )
                 logger.debug("Result for room %s: %r", room_id, room_sync)
@@ -492,14 +530,15 @@ class SyncHandler(BaseHandler):
             for room_id in joined_room_ids:
                 room_sync = yield self.incremental_sync_with_gap_for_room(
                     room_id, sync_config, since_token, now_token,
-                    ephemeral_by_room, tags_by_room
+                    ephemeral_by_room, tags_by_room, account_data_by_room
                 )
                 if room_sync:
                     joined.append(room_sync)
 
         for leave_event in leave_events:
             room_sync = yield self.incremental_sync_for_archived_room(
-                sync_config, leave_event, since_token, tags_by_room
+                sync_config, leave_event, since_token, tags_by_room,
+                account_data_by_room
             )
             archived.append(room_sync)
 
@@ -510,6 +549,7 @@ class SyncHandler(BaseHandler):
 
         defer.returnValue(SyncResult(
             presence=presence,
+            account_data=self.account_data_for_user(account_data),
             joined=joined,
             invited=invited,
             archived=archived,
@@ -566,7 +606,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def incremental_sync_with_gap_for_room(self, room_id, sync_config,
                                            since_token, now_token,
-                                           ephemeral_by_room, tags_by_room):
+                                           ephemeral_by_room, tags_by_room,
+                                           account_data_by_room):
         """ Get the incremental delta needed to bring the client up to date for
         the room. Gives the client the most recent events and the changes to
         state.
@@ -606,7 +647,7 @@ class SyncHandler(BaseHandler):
             state=state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         )
 
@@ -616,7 +657,8 @@ class SyncHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def incremental_sync_for_archived_room(self, sync_config, leave_event,
-                                           since_token, tags_by_room):
+                                           since_token, tags_by_room,
+                                           account_data_by_room):
         """ Get the incremental delta needed to bring the client up to date for
         the archived room.
         Returns:
@@ -654,7 +696,7 @@ class SyncHandler(BaseHandler):
             timeline=batch,
             state=state_events_delta,
             account_data=self.account_data_for_room(
-                leave_event.room_id, tags_by_room
+                leave_event.room_id, tags_by_room, account_data_by_room
             ),
         )