From dbeed36dec021df3036e088910c72d5727910dd3 Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Mon, 15 Feb 2016 14:38:27 +0000
Subject: Merge some room joining codepaths

Force joining by alias to go through the send_membership_event checks,
rather than bypassing them straight into _do_join. This is the first of
many stages of cleanup.
---
 synapse/rest/client/v1/room.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 81bfe377bd..76025213dc 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -246,7 +246,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
         if is_room_alias:
             handler = self.handlers.room_member_handler
             ret_dict = yield handler.join_room_alias(
-                requester.user,
+                requester,
                 identifier,
             )
             defer.returnValue((200, ret_dict))
-- 
cgit 1.4.1


From e71095801fc376aac30ff9408ae7f0203684024d Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Mon, 15 Feb 2016 15:39:16 +0000
Subject: Merge implementation of /join by alias or ID

This code is kind of rough (passing the remote servers down a long
chain), but is a step towards improvement.
---
 synapse/handlers/_base.py      |  5 +++-
 synapse/handlers/message.py    | 20 ++++++++-----
 synapse/handlers/room.py       | 40 ++++++++++---------------
 synapse/rest/client/v1/room.py | 68 +++++++++++++++++++-----------------------
 synapse/types.py               |  8 +++++
 5 files changed, 71 insertions(+), 70 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 064e8723c8..8508ecdd49 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -188,9 +188,12 @@ class BaseHandler(object):
         )
 
     @defer.inlineCallbacks
-    def handle_new_client_event(self, event, context, extra_users=[]):
+    def handle_new_client_event(self, event, context, ratelimit=True, extra_users=[]):
         # We now need to go and hit out to wherever we need to hit out to.
 
+        if ratelimit:
+            self.ratelimit(event.sender)
+
         self.auth.check(event, auth_events=context.current_state)
 
         yield self.maybe_kick_guest_users(event, context.current_state.values())
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 82c8cb5f0c..a94fad1735 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -216,7 +216,7 @@ class MessageHandler(BaseHandler):
         defer.returnValue((event, context))
 
     @defer.inlineCallbacks
-    def send_event(self, event, context, ratelimit=True, is_guest=False):
+    def send_event(self, event, context, ratelimit=True, is_guest=False, room_hosts=None):
         """
         Persists and notifies local clients and federation of an event.
 
@@ -230,9 +230,6 @@ class MessageHandler(BaseHandler):
 
         assert self.hs.is_mine(user), "User must be our own: %s" % (user,)
 
-        if ratelimit:
-            self.ratelimit(event.sender)
-
         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:
@@ -245,11 +242,18 @@ class MessageHandler(BaseHandler):
 
         if event.type == EventTypes.Member:
             member_handler = self.hs.get_handlers().room_member_handler
-            yield member_handler.send_membership_event(event, context, is_guest=is_guest)
+            yield member_handler.send_membership_event(
+                event,
+                context,
+                is_guest=is_guest,
+                ratelimit=ratelimit,
+                room_hosts=room_hosts
+            )
         else:
             yield self.handle_new_client_event(
                 event=event,
                 context=context,
+                ratelimit=ratelimit,
             )
 
         if event.type == EventTypes.Message:
@@ -259,7 +263,8 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def create_and_send_event(self, event_dict, ratelimit=True,
-                              token_id=None, txn_id=None, is_guest=False):
+                              token_id=None, txn_id=None, is_guest=False,
+                              room_hosts=None):
         """
         Creates an event, then sends it.
 
@@ -274,7 +279,8 @@ class MessageHandler(BaseHandler):
             event,
             context,
             ratelimit=ratelimit,
-            is_guest=is_guest
+            is_guest=is_guest,
+            room_hosts=room_hosts,
         )
         defer.returnValue(event)
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 89695cc0cf..b748e81d20 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -455,7 +455,9 @@ class RoomMemberHandler(BaseHandler):
             yield self.forget(requester.user, room_id)
 
     @defer.inlineCallbacks
-    def send_membership_event(self, event, context, is_guest=False, room_hosts=None):
+    def send_membership_event(
+            self, event, context, is_guest=False, room_hosts=None, ratelimit=True
+    ):
         """ Change the membership status of a user in a room.
 
         Args:
@@ -527,8 +529,17 @@ class RoomMemberHandler(BaseHandler):
         defer.returnValue({"room_id": room_id})
 
     @defer.inlineCallbacks
-    def join_room_alias(self, requester, room_alias, content={}):
-        joinee = requester.user
+    def lookup_room_alias(self, room_alias):
+        """
+        Get the room ID associated with a room alias.
+
+        Args:
+            room_alias (RoomAlias): The alias to look up.
+        Returns:
+            The room ID as a RoomID object.
+        Raises:
+            SynapseError if room alias could not be found.
+        """
         directory_handler = self.hs.get_handlers().directory_handler
         mapping = yield directory_handler.get_association(room_alias)
 
@@ -540,28 +551,7 @@ class RoomMemberHandler(BaseHandler):
         if not hosts:
             raise SynapseError(404, "No known servers")
 
-        # If event doesn't include a display name, add one.
-        yield collect_presencelike_data(self.distributor, joinee, content)
-
-        content.update({"membership": Membership.JOIN})
-        builder = self.event_builder_factory.new({
-            "type": EventTypes.Member,
-            "state_key": joinee.to_string(),
-            "room_id": room_id,
-            "sender": joinee.to_string(),
-            "membership": Membership.JOIN,
-            "content": content,
-        })
-        event, context = yield self._create_new_client_event(builder)
-
-        yield self.send_membership_event(
-            event,
-            context,
-            is_guest=requester.is_guest,
-            room_hosts=hosts
-        )
-
-        defer.returnValue({"room_id": room_id})
+        defer.returnValue((RoomID.from_string(room_id), hosts))
 
     @defer.inlineCallbacks
     def _do_join(self, event, context, room_hosts=None):
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 76025213dc..340c24635d 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -229,46 +229,40 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
             allow_guest=True,
         )
 
-        # the identifier could be a room alias or a room id. Try one then the
-        # other if it fails to parse, without swallowing other valid
-        # SynapseErrors.
-
-        identifier = None
-        is_room_alias = False
-        try:
-            identifier = RoomAlias.from_string(room_identifier)
-            is_room_alias = True
-        except SynapseError:
-            identifier = RoomID.from_string(room_identifier)
+        if RoomID.is_valid(room_identifier):
+            room_id = room_identifier
+            room_hosts = None
+        elif RoomAlias.is_valid(room_identifier):
+            handler = self.handlers.room_member_handler
+            room_alias = RoomAlias.from_string(room_identifier)
+            room_id, room_hosts = yield handler.lookup_room_alias(room_alias)
+            room_id = room_id.to_string()
+        else:
+            raise SynapseError(400, "%s was not legal room ID or room alias" % (
+                room_identifier,
+            ))
 
-        # TODO: Support for specifying the home server to join with?
+        msg_handler = self.handlers.message_handler
+        content = {"membership": Membership.JOIN}
+        if requester.is_guest:
+            content["kind"] = "guest"
+        yield msg_handler.create_and_send_event(
+            {
+                "type": EventTypes.Member,
+                "content": content,
+                "room_id": room_id,
+                "sender": requester.user.to_string(),
+                "state_key": requester.user.to_string(),
 
-        if is_room_alias:
-            handler = self.handlers.room_member_handler
-            ret_dict = yield handler.join_room_alias(
-                requester,
-                identifier,
-            )
-            defer.returnValue((200, ret_dict))
-        else:  # room id
-            msg_handler = self.handlers.message_handler
-            content = {"membership": Membership.JOIN}
-            if requester.is_guest:
-                content["kind"] = "guest"
-            yield msg_handler.create_and_send_event(
-                {
-                    "type": EventTypes.Member,
-                    "content": content,
-                    "room_id": identifier.to_string(),
-                    "sender": requester.user.to_string(),
-                    "state_key": requester.user.to_string(),
-                },
-                token_id=requester.access_token_id,
-                txn_id=txn_id,
-                is_guest=requester.is_guest,
-            )
+                "membership": Membership.JOIN,  # For backwards compatibility
+            },
+            token_id=requester.access_token_id,
+            txn_id=txn_id,
+            is_guest=requester.is_guest,
+            room_hosts=room_hosts,
+        )
 
-            defer.returnValue((200, {"room_id": identifier.to_string()}))
+        defer.returnValue((200, {"room_id": room_id}))
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_identifier, txn_id):
diff --git a/synapse/types.py b/synapse/types.py
index 2095837ba6..d5bd95cbd3 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -73,6 +73,14 @@ class DomainSpecificString(
         """Return a string encoding the fields of the structure object."""
         return "%s%s:%s" % (self.SIGIL, self.localpart, self.domain)
 
+    @classmethod
+    def is_valid(cls, s):
+        try:
+            cls.from_string(s)
+            return True
+        except:
+            return False
+
     __str__ = to_string
 
     @classmethod
-- 
cgit 1.4.1


From 150fcde0dce02670c2180f9d4657783eb204daa8 Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Mon, 15 Feb 2016 16:16:03 +0000
Subject: Reuse update_membership from /join

---
 synapse/handlers/room.py       | 12 +++++++++---
 synapse/rest/client/v1/room.py | 21 +++++----------------
 2 files changed, 14 insertions(+), 19 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index d17e5c1b7b..04916d4e24 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -403,7 +403,9 @@ class RoomMemberHandler(BaseHandler):
                     remotedomains.add(member.domain)
 
     @defer.inlineCallbacks
-    def update_membership(self, requester, target, room_id, action, txn_id=None):
+    def update_membership(
+            self, requester, target, room_id, action, txn_id=None, room_hosts=None
+    ):
         effective_membership_state = action
         if action in ["kick", "unban"]:
             effective_membership_state = "leave"
@@ -412,7 +414,7 @@ class RoomMemberHandler(BaseHandler):
 
         msg_handler = self.hs.get_handlers().message_handler
 
-        content = {"membership": unicode(effective_membership_state)}
+        content = {"membership": effective_membership_state}
         if requester.is_guest:
             content["kind"] = "guest"
 
@@ -423,6 +425,9 @@ class RoomMemberHandler(BaseHandler):
                 "room_id": room_id,
                 "sender": requester.user.to_string(),
                 "state_key": target.to_string(),
+
+                # For backwards compatibility:
+                "membership": effective_membership_state,
             },
             token_id=requester.access_token_id,
             txn_id=txn_id,
@@ -447,7 +452,8 @@ class RoomMemberHandler(BaseHandler):
             event,
             context,
             ratelimit=True,
-            is_guest=requester.is_guest
+            is_guest=requester.is_guest,
+            room_hosts=room_hosts,
         )
 
         if action == "forget":
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 340c24635d..f8cd746a88 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -242,23 +242,12 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
                 room_identifier,
             ))
 
-        msg_handler = self.handlers.message_handler
-        content = {"membership": Membership.JOIN}
-        if requester.is_guest:
-            content["kind"] = "guest"
-        yield msg_handler.create_and_send_event(
-            {
-                "type": EventTypes.Member,
-                "content": content,
-                "room_id": room_id,
-                "sender": requester.user.to_string(),
-                "state_key": requester.user.to_string(),
-
-                "membership": Membership.JOIN,  # For backwards compatibility
-            },
-            token_id=requester.access_token_id,
+        yield self.handlers.room_member_handler.update_membership(
+            requester=requester,
+            target=requester.user,
+            room_id=room_id,
+            action="join",
             txn_id=txn_id,
-            is_guest=requester.is_guest,
             room_hosts=room_hosts,
         )
 
-- 
cgit 1.4.1


From 1a2197d7bf62437208643f750ee757b8b85e2db6 Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Mon, 15 Feb 2016 18:13:10 +0000
Subject: Simplify room creation code

---
 synapse/handlers/room.py       | 62 +++++++++++++++++-------------------------
 synapse/rest/client/v1/room.py | 18 ++----------
 2 files changed, 28 insertions(+), 52 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 505fb383ec..bdaa05e0b6 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -76,13 +76,11 @@ class RoomCreationHandler(BaseHandler):
     }
 
     @defer.inlineCallbacks
-    def create_room(self, user_id, room_id, config):
+    def create_room(self, requester, config):
         """ Creates a new room.
 
         Args:
             user_id (str): The ID of the user creating the new room.
-            room_id (str): The proposed ID for the new room. Can be None, in
-            which case one will be created for you.
             config (dict) : A dict of configuration options.
         Returns:
             The new room ID.
@@ -90,6 +88,8 @@ class RoomCreationHandler(BaseHandler):
             SynapseError if the room ID was taken, couldn't be stored, or
             something went horribly wrong.
         """
+        user_id = requester.user.to_string()
+
         self.ratelimit(user_id)
 
         if "room_alias_name" in config:
@@ -121,40 +121,28 @@ class RoomCreationHandler(BaseHandler):
 
         is_public = config.get("visibility", None) == "public"
 
-        if room_id:
-            # Ensure room_id is the correct type
-            room_id_obj = RoomID.from_string(room_id)
-            if not self.hs.is_mine(room_id_obj):
-                raise SynapseError(400, "Room id must be local")
-
-            yield self.store.store_room(
-                room_id=room_id,
-                room_creator_user_id=user_id,
-                is_public=is_public
-            )
-        else:
-            # autogen room IDs and try to create it. We may clash, so just
-            # try a few times till one goes through, giving up eventually.
-            attempts = 0
-            room_id = None
-            while attempts < 5:
-                try:
-                    random_string = stringutils.random_string(18)
-                    gen_room_id = RoomID.create(
-                        random_string,
-                        self.hs.hostname,
-                    )
-                    yield self.store.store_room(
-                        room_id=gen_room_id.to_string(),
-                        room_creator_user_id=user_id,
-                        is_public=is_public
-                    )
-                    room_id = gen_room_id.to_string()
-                    break
-                except StoreError:
-                    attempts += 1
-            if not room_id:
-                raise StoreError(500, "Couldn't generate a room ID.")
+        # autogen room IDs and try to create it. We may clash, so just
+        # try a few times till one goes through, giving up eventually.
+        attempts = 0
+        room_id = None
+        while attempts < 5:
+            try:
+                random_string = stringutils.random_string(18)
+                gen_room_id = RoomID.create(
+                    random_string,
+                    self.hs.hostname,
+                )
+                yield self.store.store_room(
+                    room_id=gen_room_id.to_string(),
+                    room_creator_user_id=user_id,
+                    is_public=is_public
+                )
+                room_id = gen_room_id.to_string()
+                break
+            except StoreError:
+                attempts += 1
+        if not room_id:
+            raise StoreError(500, "Couldn't generate a room ID.")
 
         if room_alias:
             directory_handler = self.hs.get_handlers().directory_handler
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index f8cd746a88..5f5c26a91c 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -63,24 +63,12 @@ class RoomCreateRestServlet(ClientV1RestServlet):
     def on_POST(self, request):
         requester = yield self.auth.get_user_by_req(request)
 
-        room_config = self.get_room_config(request)
-        info = yield self.make_room(
-            room_config,
-            requester.user,
-            None,
-        )
-        room_config.update(info)
-        defer.returnValue((200, info))
-
-    @defer.inlineCallbacks
-    def make_room(self, room_config, auth_user, room_id):
         handler = self.handlers.room_creation_handler
         info = yield handler.create_room(
-            user_id=auth_user.to_string(),
-            room_id=room_id,
-            config=room_config
+            requester, self.get_room_config(request)
         )
-        defer.returnValue(info)
+
+        defer.returnValue((200, info))
 
     def get_room_config(self, request):
         try:
-- 
cgit 1.4.1


From 4bfb32f685cff919141a3fc0cd9179447febc765 Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Mon, 15 Feb 2016 18:21:30 +0000
Subject: Branch off member and non member sends

Unclean, needs tidy-up, but works
---
 synapse/handlers/directory.py  |  2 +-
 synapse/handlers/federation.py |  4 +--
 synapse/handlers/message.py    | 66 +++++++++++++++++-----------------
 synapse/handlers/room.py       | 80 ++++++++++++++++++++++++------------------
 synapse/rest/client/v1/room.py | 21 ++++++++---
 5 files changed, 99 insertions(+), 74 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py
index 4efecb1ffd..e0a778e7ff 100644
--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -216,7 +216,7 @@ class DirectoryHandler(BaseHandler):
         aliases = yield self.store.get_aliases_for_room(room_id)
 
         msg_handler = self.hs.get_handlers().message_handler
-        yield msg_handler.create_and_send_event({
+        yield msg_handler.create_and_send_nonmember_event({
             "type": EventTypes.Aliases,
             "state_key": self.hs.hostname,
             "room_id": room_id,
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index da55d43541..ac15f9e5dd 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -1658,7 +1658,7 @@ class FederationHandler(BaseHandler):
             self.auth.check(event, context.current_state)
             yield self._validate_keyserver(event, auth_events=context.current_state)
             member_handler = self.hs.get_handlers().room_member_handler
-            yield member_handler.send_membership_event(event, context)
+            yield member_handler.send_membership_event(event, context, from_client=False)
         else:
             destinations = set([x.split(":", 1)[-1] for x in (sender, room_id)])
             yield self.replication_layer.forward_third_party_invite(
@@ -1687,7 +1687,7 @@ class FederationHandler(BaseHandler):
         # TODO: Make sure the signatures actually are correct.
         event.signatures.update(returned_invite.signatures)
         member_handler = self.hs.get_handlers().room_member_handler
-        yield member_handler.send_membership_event(event, context)
+        yield member_handler.send_membership_event(event, context, from_client=False)
 
     @defer.inlineCallbacks
     def add_display_name_to_third_party_invite(self, event_dict, event, context):
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index a94fad1735..05dab172b8 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -16,7 +16,7 @@
 from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
-from synapse.api.errors import AuthError, Codes
+from synapse.api.errors import AuthError, Codes, SynapseError
 from synapse.streams.config import PaginationConfig
 from synapse.events.utils import serialize_event
 from synapse.events.validator import EventValidator
@@ -216,7 +216,7 @@ class MessageHandler(BaseHandler):
         defer.returnValue((event, context))
 
     @defer.inlineCallbacks
-    def send_event(self, event, context, ratelimit=True, is_guest=False, room_hosts=None):
+    def send_nonmember_event(self, event, context, ratelimit=True):
         """
         Persists and notifies local clients and federation of an event.
 
@@ -226,61 +226,63 @@ class MessageHandler(BaseHandler):
             ratelimit (bool): Whether to rate limit this send.
             is_guest (bool): Whether the sender is a guest.
         """
+        if event.type == EventTypes.Member:
+            raise SynapseError(
+                500,
+                "Tried to send member even through non-member codepath"
+            )
+
         user = UserID.from_string(event.sender)
 
         assert self.hs.is_mine(user), "User must be our own: %s" % (user,)
 
         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)
+            prev_state = self.deduplicate_state_event(event, context)
+            if prev_state is not None:
+                defer.returnValue(prev_state)
 
-        if event.type == EventTypes.Member:
-            member_handler = self.hs.get_handlers().room_member_handler
-            yield member_handler.send_membership_event(
-                event,
-                context,
-                is_guest=is_guest,
-                ratelimit=ratelimit,
-                room_hosts=room_hosts
-            )
-        else:
-            yield self.handle_new_client_event(
-                event=event,
-                context=context,
-                ratelimit=ratelimit,
-            )
+        yield self.handle_new_client_event(
+            event=event,
+            context=context,
+            ratelimit=ratelimit,
+        )
 
         if event.type == EventTypes.Message:
             presence = self.hs.get_handlers().presence_handler
             with PreserveLoggingContext():
                 presence.bump_presence_active_time(user)
 
+    def deduplicate_state_event(self, event, context):
+        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:
+                return prev_state
+        return None
+
     @defer.inlineCallbacks
-    def create_and_send_event(self, event_dict, ratelimit=True,
-                              token_id=None, txn_id=None, is_guest=False,
-                              room_hosts=None):
+    def create_and_send_nonmember_event(
+        self,
+        event_dict,
+        ratelimit=True,
+        token_id=None,
+        txn_id=None
+    ):
         """
         Creates an event, then sends it.
 
-        See self.create_event and self.send_event.
+        See self.create_event and self.send_nonmember_event.
         """
         event, context = yield self.create_event(
             event_dict,
             token_id=token_id,
             txn_id=txn_id
         )
-        yield self.send_event(
+        yield self.send_nonmember_event(
             event,
             context,
             ratelimit=ratelimit,
-            is_guest=is_guest,
-            room_hosts=room_hosts,
         )
         defer.returnValue(event)
 
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index bdaa05e0b6..5d4e87b3b0 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -179,13 +179,24 @@ class RoomCreationHandler(BaseHandler):
         )
 
         msg_handler = self.hs.get_handlers().message_handler
+        room_member_handler = self.hs.get_handlers().room_member_handler
 
         for event in creation_events:
-            yield msg_handler.create_and_send_event(event, ratelimit=False)
+            if event["type"] == EventTypes.Member:
+                # TODO(danielwh): This is hideous
+                yield room_member_handler.update_membership(
+                    requester,
+                    user,
+                    room_id,
+                    "join",
+                    ratelimit=False,
+                )
+            else:
+                yield msg_handler.create_and_send_nonmember_event(event, ratelimit=False)
 
         if "name" in config:
             name = config["name"]
-            yield msg_handler.create_and_send_event({
+            yield msg_handler.create_and_send_nonmember_event({
                 "type": EventTypes.Name,
                 "room_id": room_id,
                 "sender": user_id,
@@ -195,7 +206,7 @@ class RoomCreationHandler(BaseHandler):
 
         if "topic" in config:
             topic = config["topic"]
-            yield msg_handler.create_and_send_event({
+            yield msg_handler.create_and_send_nonmember_event({
                 "type": EventTypes.Topic,
                 "room_id": room_id,
                 "sender": user_id,
@@ -204,13 +215,13 @@ class RoomCreationHandler(BaseHandler):
             }, ratelimit=False)
 
         for invitee in invite_list:
-            yield msg_handler.create_and_send_event({
-                "type": EventTypes.Member,
-                "state_key": invitee,
-                "room_id": room_id,
-                "sender": user_id,
-                "content": {"membership": Membership.INVITE},
-            }, ratelimit=False)
+            room_member_handler.update_membership(
+                requester,
+                UserID.from_string(invitee),
+                room_id,
+                "invite",
+                ratelimit=False,
+            )
 
         for invite_3pid in invite_3pid_list:
             id_server = invite_3pid["id_server"]
@@ -222,7 +233,7 @@ class RoomCreationHandler(BaseHandler):
                 medium,
                 address,
                 id_server,
-                token_id=None,
+                requester,
                 txn_id=None,
             )
 
@@ -439,12 +450,14 @@ class RoomMemberHandler(BaseHandler):
                 errcode=Codes.BAD_STATE
             )
 
-        yield msg_handler.send_event(
+        member_handler = self.hs.get_handlers().room_member_handler
+        yield member_handler.send_membership_event(
             event,
             context,
-            ratelimit=ratelimit,
             is_guest=requester.is_guest,
+            ratelimit=ratelimit,
             room_hosts=room_hosts,
+            from_client=True,
         )
 
         if action == "forget":
@@ -452,7 +465,7 @@ class RoomMemberHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def send_membership_event(
-            self, event, context, is_guest=False, room_hosts=None, ratelimit=True
+            self, event, context, is_guest=False, room_hosts=None, ratelimit=True, from_client=True,
     ):
         """ Change the membership status of a user in a room.
 
@@ -461,6 +474,16 @@ class RoomMemberHandler(BaseHandler):
         Raises:
             SynapseError if there was a problem changing the membership.
         """
+        if from_client:
+            user = UserID.from_string(event.sender)
+
+            assert self.hs.is_mine(user), "User must be our own: %s" % (user,)
+
+        if event.is_state():
+            prev_state = self.hs.get_handlers().message_handler.deduplicate_state_event(event, context)
+            if prev_state is not None:
+                return
+
         target_user_id = event.state_key
         target_user = UserID.from_string(event.state_key)
 
@@ -549,13 +572,11 @@ class RoomMemberHandler(BaseHandler):
                         room_id,
                         event.user_id
                     )
-                    defer.returnValue({"room_id": room_id})
                     return
 
             # FIXME: This isn't idempotency.
             if prev_state and prev_state.membership == event.membership:
                 # double same action, treat this event as a NOOP.
-                defer.returnValue({})
                 return
 
             yield self.handle_new_client_event(
@@ -569,8 +590,6 @@ class RoomMemberHandler(BaseHandler):
                 user = UserID.from_string(event.user_id)
                 user_left_room(self.distributor, user, event.room_id)
 
-        defer.returnValue({"room_id": room_id})
-
     @defer.inlineCallbacks
     def lookup_room_alias(self, room_alias):
         """
@@ -657,7 +676,7 @@ class RoomMemberHandler(BaseHandler):
             medium,
             address,
             id_server,
-            token_id,
+            requester,
             txn_id
     ):
         invitee = yield self._lookup_3pid(
@@ -665,19 +684,12 @@ class RoomMemberHandler(BaseHandler):
         )
 
         if invitee:
-            # make sure it looks like a user ID; it'll throw if it's invalid.
-            UserID.from_string(invitee)
-            yield self.hs.get_handlers().message_handler.create_and_send_event(
-                {
-                    "type": EventTypes.Member,
-                    "content": {
-                        "membership": unicode("invite")
-                    },
-                    "room_id": room_id,
-                    "sender": inviter.to_string(),
-                    "state_key": invitee,
-                },
-                token_id=token_id,
+            handler = self.hs.get_handlers().room_member_handler
+            yield handler.update_membership(
+                requester,
+                UserID.from_string(invitee),
+                room_id,
+                "invite",
                 txn_id=txn_id,
             )
         else:
@@ -687,7 +699,7 @@ class RoomMemberHandler(BaseHandler):
                 address,
                 room_id,
                 inviter,
-                token_id,
+                requester.access_token_id,
                 txn_id=txn_id
             )
 
@@ -798,7 +810,7 @@ class RoomMemberHandler(BaseHandler):
             )
         )
         msg_handler = self.hs.get_handlers().message_handler
-        yield msg_handler.create_and_send_event(
+        yield msg_handler.create_and_send_nonmember_event(
             {
                 "type": EventTypes.ThirdPartyInvite,
                 "content": {
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 5f5c26a91c..179fe9a010 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -150,10 +150,21 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
             event_dict["state_key"] = state_key
 
         msg_handler = self.handlers.message_handler
-        yield msg_handler.create_and_send_event(
-            event_dict, token_id=requester.access_token_id, txn_id=txn_id,
+        event, context = yield msg_handler.create_event(
+            event_dict,
+            token_id=requester.access_token_id,
+            txn_id=txn_id,
         )
 
+        if event_type == EventTypes.Member:
+            yield self.handlers.room_member_handler.send_membership_event(
+                event,
+                context,
+                is_guest=requester.is_guest,
+            )
+        else:
+            yield msg_handler.send_nonmember_event(event, context)
+
         defer.returnValue((200, {}))
 
 
@@ -171,7 +182,7 @@ class RoomSendEventRestServlet(ClientV1RestServlet):
         content = _parse_json(request)
 
         msg_handler = self.handlers.message_handler
-        event = yield msg_handler.create_and_send_event(
+        event = yield msg_handler.create_and_send_nonmember_event(
             {
                 "type": event_type,
                 "content": content,
@@ -434,7 +445,7 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
                 content["medium"],
                 content["address"],
                 content["id_server"],
-                requester.access_token_id,
+                requester,
                 txn_id
             )
             defer.returnValue((200, {}))
@@ -490,7 +501,7 @@ class RoomRedactEventRestServlet(ClientV1RestServlet):
         content = _parse_json(request)
 
         msg_handler = self.handlers.message_handler
-        event = yield msg_handler.create_and_send_event(
+        event = yield msg_handler.create_and_send_nonmember_event(
             {
                 "type": EventTypes.Redaction,
                 "content": content,
-- 
cgit 1.4.1


From e5999bfb1a4aab56acecb59ed6d068442f5b11a0 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 15 Feb 2016 17:10:40 +0000
Subject: Initial cut

---
 synapse/handlers/events.py                         |   43 +-
 synapse/handlers/message.py                        |   14 +-
 synapse/handlers/presence.py                       | 1662 ++++++++------------
 synapse/handlers/profile.py                        |    3 +
 synapse/handlers/sync.py                           |   22 +
 synapse/rest/client/v1/presence.py                 |   26 +-
 synapse/rest/client/v1/room.py                     |   18 +-
 synapse/rest/client/v2_alpha/receipts.py           |    3 +
 synapse/rest/client/v2_alpha/sync.py               |   16 +-
 synapse/storage/__init__.py                        |   50 +-
 synapse/storage/prepare_database.py                |    2 +-
 synapse/storage/presence.py                        |  170 +-
 .../storage/schema/delta/30/presence_stream.sql    |   30 +
 synapse/storage/util/id_generators.py              |    4 +-
 synapse/util/__init__.py                           |    2 +-
 tests/utils.py                                     |    4 +-
 16 files changed, 933 insertions(+), 1136 deletions(-)
 create mode 100644 synapse/storage/schema/delta/30/presence_stream.sql

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 4933c31c19..72a31a9755 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -19,6 +19,8 @@ from synapse.util.logutils import log_function
 from synapse.types import UserID
 from synapse.events.utils import serialize_event
 from synapse.util.logcontext import preserve_context_over_fn
+from synapse.api.constants import Membership, EventTypes
+from synapse.events import EventBase
 
 from ._base import BaseHandler
 
@@ -126,11 +128,12 @@ class EventStreamHandler(BaseHandler):
         If `only_keys` is not None, events from keys will be sent down.
         """
         auth_user = UserID.from_string(auth_user_id)
+        presence_handler = self.hs.get_handlers().presence_handler
 
-        try:
-            if affect_presence:
-                yield self.started_stream(auth_user)
-
+        context = yield presence_handler.user_syncing(
+            auth_user_id, affect_presence=affect_presence,
+        )
+        with context:
             if timeout:
                 # If they've set a timeout set a minimum limit.
                 timeout = max(timeout, 500)
@@ -145,6 +148,34 @@ class EventStreamHandler(BaseHandler):
                 is_guest=is_guest, explicit_room_id=room_id
             )
 
+            # When the user joins a new room, or another user joins a currently
+            # joined room, we need to send down presence for those users.
+            to_add = []
+            for event in events:
+                if not isinstance(event, EventBase):
+                    continue
+                if event.type == EventTypes.Member:
+                    if event.membership != Membership.JOIN:
+                        continue
+                    # Send down presence.
+                    if event.state_key == auth_user_id:
+                        # Send down presence for everyone in the room.
+                        users = yield self.store.get_users_in_room(event.room_id)
+                        states = yield presence_handler.get_states(
+                            users,
+                            as_event=True,
+                        )
+                        to_add.extend(states)
+                    else:
+
+                        ev = yield presence_handler.get_state(
+                            UserID.from_string(event.state_key),
+                            as_event=True,
+                        )
+                        to_add.append(ev)
+
+            events.extend(to_add)
+
             time_now = self.clock.time_msec()
 
             chunks = [
@@ -159,10 +190,6 @@ class EventStreamHandler(BaseHandler):
 
             defer.returnValue(chunk)
 
-        finally:
-            if affect_presence:
-                self.stopped_stream(auth_user)
-
 
 class EventHandler(BaseHandler):
 
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 82c8cb5f0c..77894d9132 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -21,7 +21,6 @@ from synapse.streams.config import PaginationConfig
 from synapse.events.utils import serialize_event
 from synapse.events.validator import EventValidator
 from synapse.util import unwrapFirstError
-from synapse.util.logcontext import PreserveLoggingContext
 from synapse.util.caches.snapshot_cache import SnapshotCache
 from synapse.types import UserID, RoomStreamToken, StreamToken
 
@@ -254,8 +253,7 @@ class MessageHandler(BaseHandler):
 
         if event.type == EventTypes.Message:
             presence = self.hs.get_handlers().presence_handler
-            with PreserveLoggingContext():
-                presence.bump_presence_active_time(user)
+            yield presence.bump_presence_active_time(user)
 
     @defer.inlineCallbacks
     def create_and_send_event(self, event_dict, ratelimit=True,
@@ -660,10 +658,6 @@ class MessageHandler(BaseHandler):
             room_id=room_id,
         )
 
-        # TODO(paul): I wish I was called with user objects not user_id
-        #   strings...
-        auth_user = UserID.from_string(user_id)
-
         # TODO: These concurrently
         time_now = self.clock.time_msec()
         state = [
@@ -688,13 +682,11 @@ class MessageHandler(BaseHandler):
         @defer.inlineCallbacks
         def get_presence():
             states = yield presence_handler.get_states(
-                target_users=[UserID.from_string(m.user_id) for m in room_members],
-                auth_user=auth_user,
+                [m.user_id for m in room_members],
                 as_event=True,
-                check_auth=False,
             )
 
-            defer.returnValue(states.values())
+            defer.returnValue(states)
 
         @defer.inlineCallbacks
         def get_receipts():
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index b61394f2b5..26f2e669ce 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -13,13 +13,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from twisted.internet import defer
+"""This module is responsible for keeping track of presence status of local
+and remote users.
 
-from synapse.api.errors import SynapseError, AuthError
+The methods that define policy are:
+    - PresenceHandler._update_states
+    - PresenceHandler._handle_timeouts
+    - should_notify
+"""
+
+from twisted.internet import defer, reactor
+from contextlib import contextmanager
+
+from synapse.api.errors import SynapseError
 from synapse.api.constants import PresenceState
+from synapse.storage.presence import UserPresenceState
 
-from synapse.util.logcontext import PreserveLoggingContext
+from synapse.util.logcontext import preserve_fn
 from synapse.util.logutils import log_function
+from synapse.util.wheel_timer import WheelTimer
 from synapse.types import UserID
 import synapse.metrics
 
@@ -33,33 +45,24 @@ logger = logging.getLogger(__name__)
 metrics = synapse.metrics.get_metrics_for(__name__)
 
 
-# Don't bother bumping "last active" time if it differs by less than 60 seconds
+# If a user was last active in the last LAST_ACTIVE_GRANULARITY, consider them
+# "currently_active"
 LAST_ACTIVE_GRANULARITY = 60 * 1000
 
-# Keep no more than this number of offline serial revisions
-MAX_OFFLINE_SERIALS = 1000
-
-
-# TODO(paul): Maybe there's one of these I can steal from somewhere
-def partition(l, func):
-    """Partition the list by the result of func applied to each element."""
-    ret = {}
+# How long to wait until a new /events or /sync request before assuming
+# the client has gone.
+SYNC_ONLINE_TIMEOUT = 30 * 1000
 
-    for x in l:
-        key = func(x)
-        if key not in ret:
-            ret[key] = []
-        ret[key].append(x)
+# How long to wait before marking the user as idle. Compared against last active
+IDLE_TIMER = 5 * 60 * 1000
 
-    return ret
+# How often we expect remote servers to resend us presence.
+FEDERATION_TIMEOUT = 30 * 60 * 1000
 
+# How often to resend presence to remote servers
+FEDERATION_PING_INTERVAL = 25 * 60 * 1000
 
-def partitionbool(l, func):
-    def boolfunc(x):
-        return bool(func(x))
-
-    ret = partition(l, boolfunc)
-    return ret.get(True, []), ret.get(False, [])
+assert LAST_ACTIVE_GRANULARITY < IDLE_TIMER
 
 
 def user_presence_changed(distributor, user, statuscache):
@@ -72,45 +75,13 @@ def collect_presencelike_data(distributor, user, content):
 
 class PresenceHandler(BaseHandler):
 
-    STATE_LEVELS = {
-        PresenceState.OFFLINE: 0,
-        PresenceState.UNAVAILABLE: 1,
-        PresenceState.ONLINE: 2,
-        PresenceState.FREE_FOR_CHAT: 3,
-    }
-
     def __init__(self, hs):
         super(PresenceHandler, self).__init__(hs)
-
-        self.homeserver = hs
-
+        self.hs = hs
         self.clock = hs.get_clock()
-
-        distributor = hs.get_distributor()
-        distributor.observe("registered_user", self.registered_user)
-
-        distributor.observe(
-            "started_user_eventstream", self.started_user_eventstream
-        )
-        distributor.observe(
-            "stopped_user_eventstream", self.stopped_user_eventstream
-        )
-
-        distributor.observe("user_joined_room", self.user_joined_room)
-
-        distributor.declare("collect_presencelike_data")
-
-        distributor.declare("changed_presencelike_data")
-        distributor.observe(
-            "changed_presencelike_data", self.changed_presencelike_data
-        )
-
-        # outbound signal from the presence module to advertise when a user's
-        # presence has changed
-        distributor.declare("user_presence_changed")
-
-        self.distributor = distributor
-
+        self.store = hs.get_datastore()
+        self.wheel_timer = WheelTimer()
+        self.notifier = hs.get_notifier()
         self.federation = hs.get_replication_layer()
 
         self.federation.register_edu_handler(
@@ -138,348 +109,574 @@ class PresenceHandler(BaseHandler):
             )
         )
 
-        # IN-MEMORY store, mapping local userparts to sets of local users to
-        # be informed of state changes.
-        self._local_pushmap = {}
-        # map local users to sets of remote /domain names/ who are interested
-        # in them
-        self._remote_sendmap = {}
-        # map remote users to sets of local users who're interested in them
-        self._remote_recvmap = {}
-        # list of (serial, set of(userids)) tuples, ordered by serial, latest
-        # first
-        self._remote_offline_serials = []
-
-        # map any user to a UserPresenceCache
-        self._user_cachemap = {}
-        self._user_cachemap_latest_serial = 0
-
-        # map room_ids to the latest presence serial for a member of that
-        # room
-        self._room_serials = {}
-
-        metrics.register_callback(
-            "userCachemap:size",
-            lambda: len(self._user_cachemap),
+        distributor = hs.get_distributor()
+        distributor.observe("user_joined_room", self.user_joined_room)
+
+        active_presence = self.store.take_presence_startup_info()
+
+        # A dictionary of the current state of users. This is prefilled with
+        # non-offline presence from the DB. We should fetch from the DB if
+        # we can't find a users presence in here.
+        self.user_to_current_state = {
+            state.user_id: state
+            for state in active_presence
+        }
+
+        now = self.clock.time_msec()
+        for state in active_presence:
+            self.wheel_timer.insert(
+                now=now,
+                obj=state.user_id,
+                then=state.last_active + IDLE_TIMER,
+            )
+            self.wheel_timer.insert(
+                now=now,
+                obj=state.user_id,
+                then=state.last_user_sync + SYNC_ONLINE_TIMEOUT,
+            )
+            if self.hs.is_mine_id(state.user_id):
+                self.wheel_timer.insert(
+                    now=now,
+                    obj=state.user_id,
+                    then=state.last_federation_update + FEDERATION_PING_INTERVAL,
+                )
+            else:
+                self.wheel_timer.insert(
+                    now=now,
+                    obj=state.user_id,
+                    then=state.last_federation_update + FEDERATION_TIMEOUT,
+                )
+
+        # Set of users who have presence in the `user_to_current_state` that
+        # have not yet been persisted
+        self.unpersisted_users_changes = set()
+
+        reactor.addSystemEventTrigger("before", "shutdown", self._on_shutdown)
+
+        self.serial_to_user = {}
+        self._next_serial = 1
+
+        # Keeps track of the number of *ongoing* syncs. While this is non zero
+        # a user will never go offline.
+        self.user_to_num_current_syncs = {}
+
+        # Start a LoopingCall in 30s that fires every 5s.
+        # The initial delay is to allow disconnected clients a chance to
+        # reconnect before we treat them as offline.
+        self.clock.call_later(
+            0 * 1000,
+            self.clock.looping_call,
+            self._handle_timeouts,
+            5000,
         )
 
-    def _get_or_make_usercache(self, user):
-        """If the cache entry doesn't exist, initialise a new one."""
-        if user not in self._user_cachemap:
-            self._user_cachemap[user] = UserPresenceCache()
-        return self._user_cachemap[user]
-
-    def _get_or_offline_usercache(self, user):
-        """If the cache entry doesn't exist, return an OFFLINE one but do not
-        store it into the cache."""
-        if user in self._user_cachemap:
-            return self._user_cachemap[user]
-        else:
-            return UserPresenceCache()
+    @defer.inlineCallbacks
+    def _on_shutdown(self):
+        """Gets called when shutting down. This lets us persist any updates that
+        we haven't yet persisted, e.g. updates that only changes some internal
+        timers. This allows changes to persist across startup without having to
+        persist every single change.
+
+        If this does not run it simply means that some of the timers will fire
+        earlier than they should when synapse is restarted. This affect of this
+        is some spurious presence changes that will self-correct.
+        """
+        logger.info(
+            "Performing _on_shutdown. Persiting %d unpersisted changes",
+            len(self.user_to_current_state)
+        )
 
-    def registered_user(self, user):
-        return self.store.create_presence(user.localpart)
+        if self.unpersisted_users_changes:
+            yield self.store.update_presence([
+                self.user_to_current_state[user_id]
+                for user_id in self.unpersisted_users_changes
+            ])
+        logger.info("Finished _on_shutdown")
 
     @defer.inlineCallbacks
-    def is_presence_visible(self, observer_user, observed_user):
-        assert(self.hs.is_mine(observed_user))
+    def _update_states(self, new_states):
+        """Updates presence of users. Sets the appropriate timeouts. Pokes
+        the notifier and federation if and only if the changed presence state
+        should be sent to clients/servers.
+        """
+        now = self.clock.time_msec()
 
-        if observer_user == observed_user:
-            defer.returnValue(True)
+        # NOTE: We purposefully don't yield between now and when we've
+        # calculated what we want to do with the new states, to avoid races.
 
-        if (yield self.store.user_rooms_intersect(
-                [u.to_string() for u in observer_user, observed_user])):
-            defer.returnValue(True)
+        to_notify = {}  # Changes we want to notify everyone about
+        to_federation_ping = {}  # These need sending keep-alives
+        for new_state in new_states:
+            user_id = new_state.user_id
+            prev_state = self.user_to_current_state.get(
+                user_id, UserPresenceState.default(user_id)
+            )
 
-        if (yield self.store.is_presence_visible(
-                observed_localpart=observed_user.localpart,
-                observer_userid=observer_user.to_string())):
-            defer.returnValue(True)
+            # If the users are ours then we want to set up a bunch of timers
+            # to time things out.
+            if self.hs.is_mine_id(user_id):
+                if new_state.state == PresenceState.ONLINE:
+                    # Idle timer
+                    self.wheel_timer.insert(
+                        now=now,
+                        obj=user_id,
+                        then=new_state.last_active + IDLE_TIMER
+                    )
 
-        defer.returnValue(False)
+                if new_state.state != PresenceState.OFFLINE:
+                    # User has stopped syncing
+                    self.wheel_timer.insert(
+                        now=now,
+                        obj=user_id,
+                        then=new_state.last_user_sync + SYNC_ONLINE_TIMEOUT
+                    )
 
-    @defer.inlineCallbacks
-    def get_state(self, target_user, auth_user, as_event=False, check_auth=True):
-        """Get the current presence state of the given user.
+                    last_federate = new_state.last_federation_update
+                    if now - last_federate > FEDERATION_PING_INTERVAL:
+                        # Been a while since we've poked remote servers
+                        new_state = new_state.copy_and_replace(
+                            last_federation_update=now,
+                        )
+                        to_federation_ping[user_id] = new_state
 
-        Args:
-            target_user (UserID): The user whose presence we want
-            auth_user (UserID): The user requesting the presence, used for
-                checking if said user is allowed to see the persence of the
-                `target_user`
-            as_event (bool): Format the return as an event or not?
-            check_auth (bool): Perform the auth checks or not?
+            else:
+                self.wheel_timer.insert(
+                    now=now,
+                    obj=user_id,
+                    then=new_state.last_federation_update + FEDERATION_TIMEOUT
+                )
 
-        Returns:
-            dict: The presence state of the `target_user`, whose format depends
-            on the `as_event` argument.
-        """
-        if self.hs.is_mine(target_user):
-            if check_auth:
-                visible = yield self.is_presence_visible(
-                    observer_user=auth_user,
-                    observed_user=target_user
+            if new_state.state == PresenceState.ONLINE:
+                currently_active = now - new_state.last_active < LAST_ACTIVE_GRANULARITY
+                new_state = new_state.copy_and_replace(
+                    currently_active=currently_active,
                 )
 
-                if not visible:
-                    raise SynapseError(404, "Presence information not visible")
+            # Check whether the change was something worth notifying about
+            if should_notify(prev_state, new_state):
+                new_state.copy_and_replace(
+                    last_federation_update=now,
+                )
+                to_notify[user_id] = new_state
 
-            if target_user in self._user_cachemap:
-                state = self._user_cachemap[target_user].get_state()
-            else:
-                state = yield self.store.get_presence_state(target_user.localpart)
-                if "mtime" in state:
-                    del state["mtime"]
-                state["presence"] = state.pop("state")
-        else:
-            # TODO(paul): Have remote server send us permissions set
-            state = self._get_or_offline_usercache(target_user).get_state()
+            self.user_to_current_state[user_id] = new_state
+
+        # TODO: We should probably ensure there are no races hereafter
 
-        if "last_active" in state:
-            state["last_active_ago"] = int(
-                self.clock.time_msec() - state.pop("last_active")
+        if to_notify:
+            yield self._persist_and_notify(to_notify.values())
+
+        self.unpersisted_users_changes |= set(s.user_id for s in new_states)
+        self.unpersisted_users_changes -= set(to_notify.keys())
+
+        to_federation_ping = {
+            user_id: state for user_id, state in to_federation_ping.items()
+            if user_id not in to_notify
+        }
+        if to_federation_ping:
+            _, _, hosts_to_states = yield self._get_interested_parties(
+                to_federation_ping.values()
             )
 
-        if as_event:
-            content = state
+            self._push_to_remotes(hosts_to_states)
 
-            content["user_id"] = target_user.to_string()
+    def _handle_timeouts(self):
+        """Checks the presence of users that have timed out and updates as
+        appropriate.
+        """
+        now = self.clock.time_msec()
 
-            if "last_active" in content:
-                content["last_active_ago"] = int(
-                    self._clock.time_msec() - content.pop("last_active")
-                )
+        # Fetch the list of users that *may* have timed out. Things may have
+        # changed since the timeout was set, so we won't necessarily have to
+        # take any action.
+        users_to_check = self.wheel_timer.fetch(now)
 
-            defer.returnValue({"type": "m.presence", "content": content})
-        else:
-            defer.returnValue(state)
+        changes = {}  # Actual changes we need to notify people about
 
-    @defer.inlineCallbacks
-    def get_states(self, target_users, auth_user, as_event=False, check_auth=True):
-        """A batched version of the `get_state` method that accepts a list of
-        `target_users`
+        for user_id in set(users_to_check):
+            state = self.user_to_current_state.get(user_id, None)
+            if not state:
+                continue
 
-        Args:
-            target_users (list): The list of UserID's whose presence we want
-            auth_user (UserID): The user requesting the presence, used for
-                checking if said user is allowed to see the persence of the
-                `target_users`
-            as_event (bool): Format the return as an event or not?
-            check_auth (bool): Perform the auth checks or not?
+            if self.hs.is_mine_id(user_id):
+                if state.state == PresenceState.OFFLINE:
+                    continue
 
-        Returns:
-            dict: A mapping from user -> presence_state
+                if state.state == PresenceState.ONLINE:
+                    if now - state.last_active > IDLE_TIMER:
+                        # Currently online, but last activity ages ago so auto
+                        # idle
+                        changes[user_id] = state.copy_and_replace(
+                            state=PresenceState.UNAVAILABLE,
+                        )
+                    elif now - state.last_active > LAST_ACTIVE_GRANULARITY:
+                        # So that we send down a notification that we've
+                        # stopped updating.
+                        changes[user_id] = state
+
+                if now - state.last_federation_update > FEDERATION_PING_INTERVAL:
+                    # Need to send ping to other servers to ensure they don't
+                    # timeout and set us to offline
+                    changes[user_id] = state
+
+                # If there are have been no sync for a while (and none ongoing),
+                # set presence to offline
+                if not self.user_to_num_current_syncs.get(user_id, 0):
+                    if now - state.last_user_sync > SYNC_ONLINE_TIMEOUT:
+                        changes[user_id] = state.copy_and_replace(
+                            state=PresenceState.OFFLINE,
+                        )
+            else:
+                # We expect to be poked occaisonally by the other side.
+                # This is to protect against forgetful/buggy servers, so that
+                # no one gets stuck online forever.
+                if now - state.last_federation_update > FEDERATION_TIMEOUT:
+                    if state.state != PresenceState.OFFLINE:
+                        # The other side seems to have disappeared.
+                        changes[user_id] = state.copy_and_replace(
+                            state=PresenceState.OFFLINE,
+                        )
+
+        preserve_fn(self._update_states)(changes.values())
+
+    @defer.inlineCallbacks
+    def bump_presence_active_time(self, user):
+        """We've seen the user do something that indicates they're interacting
+        with the app.
         """
-        local_users, remote_users = partitionbool(
-            target_users,
-            lambda u: self.hs.is_mine(u)
-        )
+        user_id = user.to_string()
 
-        if check_auth:
-            for user in local_users:
-                visible = yield self.is_presence_visible(
-                    observer_user=auth_user,
-                    observed_user=user
-                )
+        prev_state = yield self.current_state_for_user(user_id)
 
-                if not visible:
-                    raise SynapseError(404, "Presence information not visible")
+        yield self._update_states([prev_state.copy_and_replace(
+            state=PresenceState.ONLINE,
+            last_active=self.clock.time_msec(),
+        )])
 
-        results = {}
-        if local_users:
-            for user in local_users:
-                if user in self._user_cachemap:
-                    results[user] = self._user_cachemap[user].get_state()
+    @defer.inlineCallbacks
+    def user_syncing(self, user_id, affect_presence=True):
+        """Returns a context manager that should surround any stream requests
+        from the user.
 
-            local_to_user = {u.localpart: u for u in local_users}
+        This allows us to keep track of who is currently streaming and who isn't
+        without having to have timers outside of this module to avoid flickering
+        when users disconnect/reconnect.
 
-            states = yield self.store.get_presence_states(
-                [u.localpart for u in local_users if u not in results]
-            )
+        Args:
+            user_id (str)
+            affect_presence (bool): If false this function will be a no-op.
+                Useful for streams that are not associated with an actual
+                client that is being used by a user.
+        """
+        if affect_presence:
+            curr_sync = self.user_to_num_current_syncs.get(user_id, 0)
+            self.user_to_num_current_syncs[user_id] = curr_sync + 1
+
+            prev_state = yield self.current_state_for_user(user_id)
+            if prev_state.state == PresenceState.OFFLINE:
+                # If they're currently offline then bring them online, otherwise
+                # just update the last sync times.
+                yield self._update_states([prev_state.copy_and_replace(
+                    state=PresenceState.ONLINE,
+                    last_active=self.clock.time_msec(),
+                    last_user_sync=self.clock.time_msec(),
+                )])
+            else:
+                yield self._update_states([prev_state.copy_and_replace(
+                    last_user_sync=self.clock.time_msec(),
+                )])
 
-            for local_part, state in states.items():
-                if state is None:
-                    continue
-                res = {"presence": state["state"]}
-                if "status_msg" in state and state["status_msg"]:
-                    res["status_msg"] = state["status_msg"]
-                results[local_to_user[local_part]] = res
-
-        for user in remote_users:
-            # TODO(paul): Have remote server send us permissions set
-            results[user] = self._get_or_offline_usercache(user).get_state()
-
-        for state in results.values():
-            if "last_active" in state:
-                state["last_active_ago"] = int(
-                    self.clock.time_msec() - state.pop("last_active")
-                )
+        @defer.inlineCallbacks
+        def _end():
+            if affect_presence:
+                self.user_to_num_current_syncs[user_id] -= 1
 
-        if as_event:
-            for user, state in results.items():
-                content = state
-                content["user_id"] = user.to_string()
+                prev_state = yield self.current_state_for_user(user_id)
+                yield self._update_states([prev_state.copy_and_replace(
+                    last_user_sync=self.clock.time_msec(),
+                )])
 
-                if "last_active" in content:
-                    content["last_active_ago"] = int(
-                        self._clock.time_msec() - content.pop("last_active")
-                    )
+        @contextmanager
+        def _user_syncing():
+            try:
+                yield
+            finally:
+                preserve_fn(_end)()
 
-                results[user] = {"type": "m.presence", "content": content}
+        defer.returnValue(_user_syncing())
 
-        defer.returnValue(results)
+    @defer.inlineCallbacks
+    def current_state_for_user(self, user_id):
+        """Get the current presence state for a user.
+        """
+        res = yield self.current_state_for_users([user_id])
+        defer.returnValue(res[user_id])
 
     @defer.inlineCallbacks
-    @log_function
-    def set_state(self, target_user, auth_user, state):
-        # return
-        # TODO (erikj): Turn this back on. Why did we end up sending EDUs
-        # everywhere?
+    def current_state_for_users(self, user_ids):
+        """Get the current presence state for multiple users.
 
-        if not self.hs.is_mine(target_user):
-            raise SynapseError(400, "User is not hosted on this Home Server")
+        Returns:
+            dict: `user_id` -> `UserPresenceState`
+        """
+        states = {
+            user_id: self.user_to_current_state.get(user_id, None)
+            for user_id in user_ids
+        }
+
+        missing = [user_id for user_id, state in states.items() if not state]
+        if missing:
+            # There are things not in our in memory cache. Lets pull them out of
+            # the database.
+            res = yield self.store.get_presence_for_users(missing)
+            states.update({state.user_id: state for state in res})
+
+            missing = [user_id for user_id, state in states.items() if not state]
+            if missing:
+                states.update({
+                    user_id: UserPresenceState.default(user_id)
+                    for user_id in missing
+                })
 
-        if target_user != auth_user:
-            raise AuthError(400, "Cannot set another user's presence")
+        defer.returnValue(states)
 
-        if "status_msg" not in state:
-            state["status_msg"] = None
+    @defer.inlineCallbacks
+    def _get_interested_parties(self, states):
+        """Given a list of states return which entities (rooms, users, servers)
+        are interested in the given states.
 
-        for k in state.keys():
-            if k not in ("presence", "status_msg"):
-                raise SynapseError(
-                    400, "Unexpected presence state key '%s'" % (k,)
-                )
+        Returns:
+            3-tuple: `(room_ids_to_states, users_to_states, hosts_to_states)`,
+            with each item being a dict of `entity_name` -> `[UserPresenceState]`
+        """
+        room_ids_to_states = {}
+        users_to_states = {}
+        for state in states:
+            events = yield self.store.get_rooms_for_user(state.user_id)
+            for e in events:
+                room_ids_to_states.setdefault(e.room_id, []).append(state)
+
+            plist = yield self.store.get_presence_list_observers_accepted(state.user_id)
+            for u in plist:
+                users_to_states.setdefault(u, []).append(state)
+
+            # Always notify self
+            users_to_states.setdefault(state.user_id, []).append(state)
+
+        hosts_to_states = {}
+        for room_id, states in room_ids_to_states.items():
+            hosts = yield self.store.get_joined_hosts_for_room(room_id)
+            for host in hosts:
+                hosts_to_states.setdefault(host, []).extend(states)
 
-        if state["presence"] not in self.STATE_LEVELS:
-            raise SynapseError(400, "'%s' is not a valid presence state" % (
-                state["presence"],
-            ))
+        for user_id, states in users_to_states.items():
+            host = UserID.from_string(user_id).domain
+            hosts_to_states.setdefault(host, []).extend(states)
 
-        logger.debug("Updating presence state of %s to %s",
-                     target_user.localpart, state["presence"])
+        # TODO: de-dup hosts_to_states, as a single host might have multiple
+        # of same presence
 
-        state_to_store = dict(state)
-        state_to_store["state"] = state_to_store.pop("presence")
+        defer.returnValue((room_ids_to_states, users_to_states, hosts_to_states))
+
+    @defer.inlineCallbacks
+    def _persist_and_notify(self, states):
+        """Persist states in the database, poke the notifier and send to
+        interested remote servers
+        """
+        stream_id, max_token = yield self.store.update_presence(states)
 
-        statuscache = self._get_or_offline_usercache(target_user)
-        was_level = self.STATE_LEVELS[statuscache.get_state()["presence"]]
-        now_level = self.STATE_LEVELS[state["presence"]]
+        parties = yield self._get_interested_parties(states)
+        room_ids_to_states, users_to_states, hosts_to_states = parties
 
-        yield self.store.set_presence_state(
-            target_user.localpart, state_to_store
+        self.notifier.on_new_event(
+            "presence_key", stream_id, rooms=room_ids_to_states.keys(),
+            users=[UserID.from_string(u) for u in users_to_states.keys()]
         )
-        yield collect_presencelike_data(self.distributor, target_user, state)
 
-        if now_level > was_level:
-            state["last_active"] = self.clock.time_msec()
+        self._push_to_remotes(hosts_to_states)
+
+    def _push_to_remotes(self, hosts_to_states):
+        """Sends state updates to remote servers.
+
+        Args:
+            hosts_to_states (dict): Mapping `server_name` -> `[UserPresenceState]`
+        """
+        now = self.clock.time_msec()
+        for host, states in hosts_to_states.items():
+            self.federation.send_edu(
+                destination=host,
+                edu_type="m.presence",
+                content={
+                    "push": [
+                        _format_user_presence_state(state, now)
+                        for state in states
+                    ]
+                }
+            )
+
+    @defer.inlineCallbacks
+    def incoming_presence(self, origin, content):
+        """Called when we receive a `m.presence` EDU from a remote server.
+        """
+        now = self.clock.time_msec()
+        updates = []
+        for push in content.get("push", []):
+            # A "push" contains a list of presence that we are probably interested
+            # in.
+            # TODO: Actually check if we're interested, rather than blindly
+            # accepting presence updates.
+            user_id = push.get("user_id", None)
+            if not user_id:
+                logger.info(
+                    "Got presence update from %r with no 'user_id': %r",
+                    origin, push,
+                )
+                continue
 
-        now_online = state["presence"] != PresenceState.OFFLINE
-        was_polling = target_user in self._user_cachemap
+            presence_state = push.get("presence", None)
+            if not presence_state:
+                logger.info(
+                    "Got presence update from %r with no 'presence_state': %r",
+                    origin, push,
+                )
+                continue
 
-        if now_online and not was_polling:
-            yield self.start_polling_presence(target_user, state=state)
-        elif not now_online and was_polling:
-            yield self.stop_polling_presence(target_user)
+            new_fields = {
+                "state": presence_state,
+                "last_federation_update": now,
+            }
 
-        # TODO(paul): perform a presence push as part of start/stop poll so
-        #   we don't have to do this all the time
-        yield self.changed_presencelike_data(target_user, state)
+            last_active_ago = push.get("last_active_ago", None)
+            if last_active_ago is not None:
+                new_fields["last_active"] = now - last_active_ago
 
-    def bump_presence_active_time(self, user, now=None):
-        if now is None:
-            now = self.clock.time_msec()
+            new_fields["status_msg"] = push.get("status_msg", None)
 
-        prev_state = self._get_or_make_usercache(user)
-        if now - prev_state.state.get("last_active", 0) < LAST_ACTIVE_GRANULARITY:
-            return
+            prev_state = yield self.current_state_for_user(user_id)
+            updates.append(prev_state.copy_and_replace(**new_fields))
 
-        with PreserveLoggingContext():
-            self.changed_presencelike_data(user, {"last_active": now})
+        if updates:
+            yield self._update_states(updates)
 
-    def get_joined_rooms_for_user(self, user):
-        """Get the list of rooms a user is joined to.
+    @defer.inlineCallbacks
+    def get_state(self, target_user, as_event=False):
+        results = yield self.get_states(
+            [target_user.to_string()],
+            as_event=as_event,
+        )
+
+        defer.returnValue(results[0])
+
+    @defer.inlineCallbacks
+    def get_states(self, target_user_ids, as_event=False):
+        """Get the presence state for users.
 
         Args:
-            user(UserID): The user.
+            target_user_ids (list)
+            as_event (bool): Whether to format it as a client event or not.
+
         Returns:
-            A Deferred of a list of room id strings.
+            list
         """
-        rm_handler = self.homeserver.get_handlers().room_member_handler
-        return rm_handler.get_joined_rooms_for_user(user)
 
-    def get_joined_users_for_room_id(self, room_id):
-        rm_handler = self.homeserver.get_handlers().room_member_handler
-        return rm_handler.get_room_members(room_id)
+        updates = yield self.current_state_for_users(target_user_ids)
+        updates = updates.values()
 
-    @defer.inlineCallbacks
-    def changed_presencelike_data(self, user, state):
-        """Updates the presence state of a local user.
+        for user_id in set(target_user_ids) - set(u.user_id for u in updates):
+            updates.append(UserPresenceState.default(user_id))
 
-        Args:
-            user(UserID): The user being updated.
-            state(dict): The new presence state for the user.
-        Returns:
-            A Deferred
+        now = self.clock.time_msec()
+        if as_event:
+            defer.returnValue([
+                {
+                    "type": "m.presence",
+                    "content": _format_user_presence_state(state, now),
+                }
+                for state in updates
+            ])
+        else:
+            defer.returnValue([
+                _format_user_presence_state(state, now) for state in updates
+            ])
+
+    @defer.inlineCallbacks
+    def set_state(self, target_user, state):
+        """Set the presence state of the user.
         """
-        self._user_cachemap_latest_serial += 1
-        statuscache = yield self.update_presence_cache(user, state)
-        yield self.push_presence(user, statuscache=statuscache)
+        status_msg = state.get("status_msg", None)
+        presence = state["presence"]
 
-    @log_function
-    def started_user_eventstream(self, user):
-        # TODO(paul): Use "last online" state
-        return self.set_state(user, user, {"presence": PresenceState.ONLINE})
+        user_id = target_user.to_string()
 
-    @log_function
-    def stopped_user_eventstream(self, user):
-        # TODO(paul): Save current state as "last online" state
-        return self.set_state(user, user, {"presence": PresenceState.OFFLINE})
+        prev_state = yield self.current_state_for_user(user_id)
+
+        new_fields = {
+            "state": presence,
+            "status_msg": status_msg
+        }
+
+        if presence == PresenceState.ONLINE:
+            new_fields["last_active"] = self.clock.time_msec()
+
+        yield self._update_states([prev_state.copy_and_replace(**new_fields)])
 
     @defer.inlineCallbacks
     def user_joined_room(self, user, room_id):
-        """Called via the distributor whenever a user joins a room.
-        Notifies the new member of the presence of the current members.
-        Notifies the current members of the room of the new member's presence.
-
-        Args:
-            user(UserID): The user who joined the room.
-            room_id(str): The room id the user joined.
+        """Called (via the distributor) when a user joins a room. This funciton
+        sends presence updates to servers, either:
+            1. the joining user is a local user and we send their presence to
+               all servers in the room.
+            2. the joining user is a remote user and so we send presence for all
+               local users in the room.
         """
+        # We only need to send presence to servers that don't have it yet. We
+        # don't need to send to local clients here, as that is done as part
+        # of the event stream/sync.
+        # TODO: Only send to servers not already in the room.
         if self.hs.is_mine(user):
-            # No actual update but we need to bump the serial anyway for the
-            # event source
-            self._user_cachemap_latest_serial += 1
-            statuscache = yield self.update_presence_cache(
-                user, room_ids=[room_id]
-            )
-            self.push_update_to_local_and_remote(
-                observed_user=user,
-                room_ids=[room_id],
-                statuscache=statuscache,
-            )
+            state = yield self.current_state_for_user(user.to_string())
 
-        # We also want to tell them about current presence of people.
-        curr_users = yield self.get_joined_users_for_room_id(room_id)
+            hosts = yield self.store.get_joined_hosts_for_room(room_id)
+            self._push_to_remotes({host: (state,) for host in hosts})
+        else:
+            user_ids = yield self.store.get_users_in_room(room_id)
+            user_ids = filter(self.hs.is_mine_id, user_ids)
 
-        for local_user in [c for c in curr_users if self.hs.is_mine(c)]:
-            statuscache = yield self.update_presence_cache(
-                local_user, room_ids=[room_id], add_to_cache=False
-            )
+            states = yield self.current_state_for_users(user_ids)
 
-            with PreserveLoggingContext():
-                self.push_update_to_local_and_remote(
-                    observed_user=local_user,
-                    users_to_push=[user],
-                    statuscache=statuscache,
-                )
+            self._push_to_remotes({user.domain: states.values()})
 
     @defer.inlineCallbacks
-    def send_presence_invite(self, observer_user, observed_user):
-        """Request the presence of a local or remote user for a local user"""
+    def get_presence_list(self, observer_user, accepted=None):
+        """Returns the presence for all users in their presence list.
+        """
         if not self.hs.is_mine(observer_user):
             raise SynapseError(400, "User is not hosted on this Home Server")
 
+        presence_list = yield self.store.get_presence_list(
+            observer_user.localpart, accepted=accepted
+        )
+
+        results = yield self.get_states(
+            target_user_ids=[row["observed_user_id"] for row in presence_list],
+            as_event=False,
+        )
+
+        is_accepted = {
+            row["observed_user_id"]: row["accepted"] for row in presence_list
+        }
+
+        for result in results:
+            result.update({
+                "accepted": is_accepted,
+            })
+
+        defer.returnValue(results)
+
+    @defer.inlineCallbacks
+    def send_presence_invite(self, observer_user, observed_user):
+        """Sends a presence invite.
+        """
         yield self.store.add_presence_list_pending(
             observer_user.localpart, observed_user.to_string()
         )
@@ -496,60 +693,41 @@ class PresenceHandler(BaseHandler):
                 }
             )
 
-    @defer.inlineCallbacks
-    def _should_accept_invite(self, observed_user, observer_user):
-        if not self.hs.is_mine(observed_user):
-            defer.returnValue(False)
-
-        row = yield self.store.has_presence_state(observed_user.localpart)
-        if not row:
-            defer.returnValue(False)
-
-        # TODO(paul): Eventually we'll ask the user's permission for this
-        # before accepting. For now just accept any invite request
-        defer.returnValue(True)
-
     @defer.inlineCallbacks
     def invite_presence(self, observed_user, observer_user):
-        """Handles a m.presence_invite EDU. A remote or local user has
-        requested presence updates for a local user. If the invite is accepted
-        then allow the local or remote user to see the presence of the local
-        user.
-
-        Args:
-            observed_user(UserID): The local user whose presence is requested.
-            observer_user(UserID): The remote or local user requesting presence.
+        """Handles new presence invites.
         """
-        accept = yield self._should_accept_invite(observed_user, observer_user)
-
-        if accept:
-            yield self.store.allow_presence_visible(
-                observed_user.localpart, observer_user.to_string()
-            )
+        if not self.hs.is_mine(observed_user):
+            raise SynapseError(400, "User is not hosted on this Home Server")
 
+        # TODO: Don't auto accept
         if self.hs.is_mine(observer_user):
-            if accept:
-                yield self.accept_presence(observed_user, observer_user)
-            else:
-                yield self.deny_presence(observed_user, observer_user)
+            yield self.accept_presence(observed_user, observer_user)
         else:
-            edu_type = "m.presence_accept" if accept else "m.presence_deny"
-
-            yield self.federation.send_edu(
+            self.federation.send_edu(
                 destination=observer_user.domain,
-                edu_type=edu_type,
+                edu_type="m.presence_accept",
                 content={
                     "observed_user": observed_user.to_string(),
                     "observer_user": observer_user.to_string(),
                 }
             )
 
+            state_dict = yield self.get_state(observed_user, as_event=False)
+
+            self.federation.send_edu(
+                destination=observer_user.domain,
+                edu_type="m.presence",
+                content={
+                    "push": [state_dict]
+                }
+            )
+
     @defer.inlineCallbacks
     def accept_presence(self, observed_user, observer_user):
         """Handles a m.presence_accept EDU. Mark a presence invite from a
         local or remote user as accepted in a local user's presence list.
         Starts polling for presence updates from the local or remote user.
-
         Args:
             observed_user(UserID): The user to update in the presence list.
             observer_user(UserID): The owner of the presence list to update.
@@ -558,15 +736,10 @@ class PresenceHandler(BaseHandler):
             observer_user.localpart, observed_user.to_string()
         )
 
-        yield self.start_polling_presence(
-            observer_user, target_user=observed_user
-        )
-
     @defer.inlineCallbacks
     def deny_presence(self, observed_user, observer_user):
         """Handle a m.presence_deny EDU. Removes a local or remote user from a
         local user's presence list.
-
         Args:
             observed_user(UserID): The local or remote user to remove from the
                 list.
@@ -584,7 +757,6 @@ class PresenceHandler(BaseHandler):
     def drop(self, observed_user, observer_user):
         """Remove a local or remote user from a local user's presence list and
         unsubscribe the local user from updates that user.
-
         Args:
             observed_user(UserId): The local or remote user to remove from the
                 list.
@@ -599,710 +771,138 @@ class PresenceHandler(BaseHandler):
             observer_user.localpart, observed_user.to_string()
         )
 
-        self.stop_polling_presence(
-            observer_user, target_user=observed_user
-        )
-
-    @defer.inlineCallbacks
-    def get_presence_list(self, observer_user, accepted=None):
-        """Get the presence list for a local user. The retured list includes
-        the current presence state for each user listed.
-
-        Args:
-            observer_user(UserID): The local user whose presence list to fetch.
-            accepted(bool or None): If not none then only include users who
-                have or have not accepted the presence invite request.
-        Returns:
-            A Deferred list of presence state events.
-        """
-        if not self.hs.is_mine(observer_user):
-            raise SynapseError(400, "User is not hosted on this Home Server")
-
-        presence_list = yield self.store.get_presence_list(
-            observer_user.localpart, accepted=accepted
-        )
-
-        results = []
-        for row in presence_list:
-            observed_user = UserID.from_string(row["observed_user_id"])
-            result = {
-                "observed_user": observed_user, "accepted": row["accepted"]
-            }
-            result.update(
-                self._get_or_offline_usercache(observed_user).get_state()
-            )
-            if "last_active" in result:
-                result["last_active_ago"] = int(
-                    self.clock.time_msec() - result.pop("last_active")
-                )
-            results.append(result)
-
-        defer.returnValue(results)
-
-    @defer.inlineCallbacks
-    @log_function
-    def start_polling_presence(self, user, target_user=None, state=None):
-        """Subscribe a local user to presence updates from a local or remote
-        user. If no target_user is supplied then subscribe to all users stored
-        in the presence list for the local user.
-
-        Additonally this pushes the current presence state of this user to all
-        target_users. That state can be provided directly or will be read from
-        the stored state for the local user.
-
-        Also this attempts to notify the local user of the current state of
-        any local target users.
-
-        Args:
-            user(UserID): The local user that whishes for presence updates.
-            target_user(UserID): The local or remote user whose updates are
-                wanted.
-            state(dict): Optional presence state for the local user.
-        """
-        logger.debug("Start polling for presence from %s", user)
-
-        if target_user:
-            target_users = set([target_user])
-            room_ids = []
-        else:
-            presence = yield self.store.get_presence_list(
-                user.localpart, accepted=True
-            )
-            target_users = set([
-                UserID.from_string(x["observed_user_id"]) for x in presence
-            ])
-
-            # Also include people in all my rooms
-
-            room_ids = yield self.get_joined_rooms_for_user(user)
-
-        if state is None:
-            state = yield self.store.get_presence_state(user.localpart)
-        else:
-            # statuscache = self._get_or_make_usercache(user)
-            # self._user_cachemap_latest_serial += 1
-            # statuscache.update(state, self._user_cachemap_latest_serial)
-            pass
-
-        yield self.push_update_to_local_and_remote(
-            observed_user=user,
-            users_to_push=target_users,
-            room_ids=room_ids,
-            statuscache=self._get_or_make_usercache(user),
-        )
-
-        for target_user in target_users:
-            if self.hs.is_mine(target_user):
-                self._start_polling_local(user, target_user)
-
-                # We want to tell the person that just came online
-                # presence state of people they are interested in?
-                self.push_update_to_clients(
-                    users_to_push=[user],
-                )
-
-        deferreds = []
-        remote_users = [u for u in target_users if not self.hs.is_mine(u)]
-        remoteusers_by_domain = partition(remote_users, lambda u: u.domain)
-        # Only poll for people in our get_presence_list
-        for domain in remoteusers_by_domain:
-            remoteusers = remoteusers_by_domain[domain]
-
-            deferreds.append(self._start_polling_remote(
-                user, domain, remoteusers
-            ))
-
-        yield defer.DeferredList(deferreds, consumeErrors=True)
-
-    def _start_polling_local(self, user, target_user):
-        """Subscribe a local user to presence updates for a local user
-
-        Args:
-            user(UserId): The local user that wishes for updates.
-            target_user(UserId): The local users whose updates are wanted.
-        """
-        target_localpart = target_user.localpart
-
-        if target_localpart not in self._local_pushmap:
-            self._local_pushmap[target_localpart] = set()
-
-        self._local_pushmap[target_localpart].add(user)
-
-    def _start_polling_remote(self, user, domain, remoteusers):
-        """Subscribe a local user to presence updates for remote users on a
-        given remote domain.
-
-        Args:
-            user(UserID): The local user that wishes for updates.
-            domain(str): The remote server the local user wants updates from.
-            remoteusers(UserID): The remote users that local user wants to be
-                told about.
-        Returns:
-            A Deferred.
-        """
-        to_poll = set()
-
-        for u in remoteusers:
-            if u not in self._remote_recvmap:
-                self._remote_recvmap[u] = set()
-                to_poll.add(u)
-
-            self._remote_recvmap[u].add(user)
-
-        if not to_poll:
-            return defer.succeed(None)
-
-        return self.federation.send_edu(
-            destination=domain,
-            edu_type="m.presence",
-            content={"poll": [u.to_string() for u in to_poll]}
-        )
-
-    @log_function
-    def stop_polling_presence(self, user, target_user=None):
-        """Unsubscribe a local user from presence updates from a local or
-        remote user. If no target user is supplied then unsubscribe the user
-        from all presence updates that the user had subscribed to.
-
-        Args:
-            user(UserID): The local user that no longer wishes for updates.
-            target_user(UserID or None): The user whose updates are no longer
-                wanted.
-        Returns:
-            A Deferred.
-        """
-        logger.debug("Stop polling for presence from %s", user)
-
-        if not target_user or self.hs.is_mine(target_user):
-            self._stop_polling_local(user, target_user=target_user)
-
-        deferreds = []
-
-        if target_user:
-            if target_user not in self._remote_recvmap:
-                return
-            target_users = set([target_user])
-        else:
-            target_users = self._remote_recvmap.keys()
-
-        remoteusers = [u for u in target_users
-                       if user in self._remote_recvmap[u]]
-        remoteusers_by_domain = partition(remoteusers, lambda u: u.domain)
-
-        for domain in remoteusers_by_domain:
-            remoteusers = remoteusers_by_domain[domain]
-
-            deferreds.append(
-                self._stop_polling_remote(user, domain, remoteusers)
-            )
-
-        return defer.DeferredList(deferreds, consumeErrors=True)
-
-    def _stop_polling_local(self, user, target_user):
-        """Unsubscribe a local user from presence updates from a local user on
-        this server.
-
-        Args:
-            user(UserID): The local user that no longer wishes for updates.
-            target_user(UserID): The user whose updates are no longer wanted.
-        """
-        for localpart in self._local_pushmap.keys():
-            if target_user and localpart != target_user.localpart:
-                continue
-
-            if user in self._local_pushmap[localpart]:
-                self._local_pushmap[localpart].remove(user)
-
-            if not self._local_pushmap[localpart]:
-                del self._local_pushmap[localpart]
-
-    @log_function
-    def _stop_polling_remote(self, user, domain, remoteusers):
-        """Unsubscribe a local user from presence updates from remote users on
-        a given domain.
-
-        Args:
-            user(UserID): The local user that no longer wishes for updates.
-            domain(str): The remote server to unsubscribe from.
-            remoteusers([UserID]): The users on that remote server that the
-                local user no longer wishes to be updated about.
-        Returns:
-            A Deferred.
-        """
-        to_unpoll = set()
-
-        for u in remoteusers:
-            self._remote_recvmap[u].remove(user)
-
-            if not self._remote_recvmap[u]:
-                del self._remote_recvmap[u]
-                to_unpoll.add(u)
-
-        if not to_unpoll:
-            return defer.succeed(None)
-
-        return self.federation.send_edu(
-            destination=domain,
-            edu_type="m.presence",
-            content={"unpoll": [u.to_string() for u in to_unpoll]}
-        )
-
-    @defer.inlineCallbacks
-    @log_function
-    def push_presence(self, user, statuscache):
-        """
-        Notify local and remote users of a change in presence of a local user.
-        Pushes the update to local clients and remote domains that are directly
-        subscribed to the presence of the local user.
-        Also pushes that update to any local user or remote domain that shares
-        a room with the local user.
-
-        Args:
-            user(UserID): The local user whose presence was updated.
-            statuscache(UserPresenceCache): Cache of the user's presence state
-        Returns:
-            A Deferred.
-        """
-        assert(self.hs.is_mine(user))
-
-        logger.debug("Pushing presence update from %s", user)
-
-        localusers = set(self._local_pushmap.get(user.localpart, set()))
-        remotedomains = set(self._remote_sendmap.get(user.localpart, set()))
-
-        # Reflect users' status changes back to themselves, so UIs look nice
-        # and also user is informed of server-forced pushes
-        localusers.add(user)
-
-        room_ids = yield self.get_joined_rooms_for_user(user)
-
-        if not localusers and not room_ids:
-            defer.returnValue(None)
-
-        yield self.push_update_to_local_and_remote(
-            observed_user=user,
-            users_to_push=localusers,
-            remote_domains=remotedomains,
-            room_ids=room_ids,
-            statuscache=statuscache,
-        )
-        yield user_presence_changed(self.distributor, user, statuscache)
-
-    @defer.inlineCallbacks
-    def incoming_presence(self, origin, content):
-        """Handle an incoming m.presence EDU.
-        For each presence update in the "push" list update our local cache and
-        notify the appropriate local clients. Only clients that share a room
-        or are directly subscribed to the presence for a user should be
-        notified of the update.
-        For each subscription request in the "poll" list start pushing presence
-        updates to the remote server.
-        For unsubscribe request in the "unpoll" list stop pushing presence
-        updates to the remote server.
-
-        Args:
-            orgin(str): The source of this m.presence EDU.
-            content(dict): The content of this m.presence EDU.
-        Returns:
-            A Deferred.
-        """
-        deferreds = []
-
-        for push in content.get("push", []):
-            user = UserID.from_string(push["user_id"])
-
-            logger.debug("Incoming presence update from %s", user)
-
-            observers = set(self._remote_recvmap.get(user, set()))
-            if observers:
-                logger.debug(
-                    " | %d interested local observers %r", len(observers), observers
-                )
-
-            room_ids = yield self.get_joined_rooms_for_user(user)
-            if room_ids:
-                logger.debug(" | %d interested room IDs %r", len(room_ids), room_ids)
-
-            state = dict(push)
-            del state["user_id"]
-
-            if "presence" not in state:
-                logger.warning(
-                    "Received a presence 'push' EDU from %s without a"
-                    " 'presence' key", origin
-                )
-                continue
-
-            if "last_active_ago" in state:
-                state["last_active"] = int(
-                    self.clock.time_msec() - state.pop("last_active_ago")
-                )
-
-            self._user_cachemap_latest_serial += 1
-            yield self.update_presence_cache(user, state, room_ids=room_ids)
-
-            if not observers and not room_ids:
-                logger.debug(" | no interested observers or room IDs")
-                continue
-
-            self.push_update_to_clients(
-                users_to_push=observers, room_ids=room_ids
-            )
-
-            user_id = user.to_string()
-
-            if state["presence"] == PresenceState.OFFLINE:
-                self._remote_offline_serials.insert(
-                    0,
-                    (self._user_cachemap_latest_serial, set([user_id]))
-                )
-                while len(self._remote_offline_serials) > MAX_OFFLINE_SERIALS:
-                    self._remote_offline_serials.pop()  # remove the oldest
-                if user in self._user_cachemap:
-                    del self._user_cachemap[user]
-            else:
-                # Remove the user from remote_offline_serials now that they're
-                # no longer offline
-                for idx, elem in enumerate(self._remote_offline_serials):
-                    (_, user_ids) = elem
-                    user_ids.discard(user_id)
-                    if not user_ids:
-                        self._remote_offline_serials.pop(idx)
-
-        for poll in content.get("poll", []):
-            user = UserID.from_string(poll)
-
-            if not self.hs.is_mine(user):
-                continue
-
-            # TODO(paul) permissions checks
-
-            if user not in self._remote_sendmap:
-                self._remote_sendmap[user] = set()
-
-            self._remote_sendmap[user].add(origin)
-
-            deferreds.append(self._push_presence_remote(user, origin))
-
-        for unpoll in content.get("unpoll", []):
-            user = UserID.from_string(unpoll)
-
-            if not self.hs.is_mine(user):
-                continue
-
-            if user in self._remote_sendmap:
-                self._remote_sendmap[user].remove(origin)
-
-                if not self._remote_sendmap[user]:
-                    del self._remote_sendmap[user]
-
-        yield defer.DeferredList(deferreds, consumeErrors=True)
-
-    @defer.inlineCallbacks
-    def update_presence_cache(self, user, state={}, room_ids=None,
-                              add_to_cache=True):
-        """Update the presence cache for a user with a new state and bump the
-        serial to the latest value.
-
-        Args:
-            user(UserID): The user being updated
-            state(dict): The presence state being updated
-            room_ids(None or list of str): A list of room_ids to update. If
-                room_ids is None then fetch the list of room_ids the user is
-                joined to.
-            add_to_cache: Whether to add an entry to the presence cache if the
-                user isn't already in the cache.
-        Returns:
-            A Deferred UserPresenceCache for the user being updated.
-        """
-        if room_ids is None:
-            room_ids = yield self.get_joined_rooms_for_user(user)
-
-        for room_id in room_ids:
-            self._room_serials[room_id] = self._user_cachemap_latest_serial
-        if add_to_cache:
-            statuscache = self._get_or_make_usercache(user)
-        else:
-            statuscache = self._get_or_offline_usercache(user)
-        statuscache.update(state, serial=self._user_cachemap_latest_serial)
-        defer.returnValue(statuscache)
+        # TODO: Inform the remote that we've dropped the presence list.
 
     @defer.inlineCallbacks
-    def push_update_to_local_and_remote(self, observed_user, statuscache,
-                                        users_to_push=[], room_ids=[],
-                                        remote_domains=[]):
-        """Notify local clients and remote servers of a change in the presence
-        of a user.
-
-        Args:
-            observed_user(UserID): The user to push the presence state for.
-            statuscache(UserPresenceCache): The cache for the presence state to
-                push.
-            users_to_push([UserID]): A list of local and remote users to
-                notify.
-            room_ids([str]): Notify the local and remote occupants of these
-                rooms.
-            remote_domains([str]): A list of remote servers to notify in
-                addition to those implied by the users_to_push and the
-                room_ids.
-        Returns:
-            A Deferred.
-        """
+    def is_visible(self, observed_user, observer_user):
+        observer_rooms = yield self.store.get_rooms_for_user(observer_user.to_string())
+        observed_rooms = yield self.store.get_rooms_for_user(observed_user.to_string())
 
-        localusers, remoteusers = partitionbool(
-            users_to_push,
-            lambda u: self.hs.is_mine(u)
-        )
+        observer_room_ids = set(r.room_id for r in observer_rooms)
+        observed_room_ids = set(r.room_id for r in observed_rooms)
 
-        localusers = set(localusers)
+        if observer_room_ids & observed_room_ids:
+            defer.returnValue(True)
 
-        self.push_update_to_clients(
-            users_to_push=localusers, room_ids=room_ids
+        accepted_observers = yield self.store.get_presence_list_observers_accepted(
+            observed_user.to_string()
         )
 
-        remote_domains = set(remote_domains)
-        remote_domains |= set([r.domain for r in remoteusers])
-        for room_id in room_ids:
-            remote_domains.update(
-                (yield self.store.get_joined_hosts_for_room(room_id))
-            )
+        defer.returnValue(observer_user.to_string() in accepted_observers)
 
-        remote_domains.discard(self.hs.hostname)
-
-        deferreds = []
-        for domain in remote_domains:
-            logger.debug(" | push to remote domain %s", domain)
-            deferreds.append(
-                self._push_presence_remote(
-                    observed_user, domain, state=statuscache.get_state()
-                )
-            )
 
-        yield defer.DeferredList(deferreds, consumeErrors=True)
+def should_notify(old_state, new_state):
+    """Decides if a presence state change should be sent to interested parties.
+    """
+    if old_state.status_msg != new_state.status_msg:
+        return True
 
-        defer.returnValue((localusers, remote_domains))
+    if old_state.state == PresenceState.ONLINE:
+        if new_state.state != PresenceState.ONLINE:
+            # Always notify for online -> anything
+            return True
 
-    def push_update_to_clients(self, users_to_push=[], room_ids=[]):
-        """Notify clients of a new presence event.
+        if new_state.currently_active != old_state.currently_active:
+            return True
 
-        Args:
-            users_to_push([UserID]): List of users to notify.
-            room_ids([str]): List of room_ids to notify.
-        """
-        with PreserveLoggingContext():
-            self.notifier.on_new_event(
-                "presence_key",
-                self._user_cachemap_latest_serial,
-                users_to_push,
-                room_ids,
-            )
+    if new_state.last_active - old_state.last_active > LAST_ACTIVE_GRANULARITY:
+        # Always notify for a transition where last active gets bumped.
+        return True
 
-    @defer.inlineCallbacks
-    def _push_presence_remote(self, user, destination, state=None):
-        """Push a user's presence to a remote server. If a presence state event
-        that event is sent. Otherwise a new state event is constructed from the
-        stored presence state.
-        The last_active is replaced with last_active_ago in case the wallclock
-        time on the remote server is different to the time on this server.
-        Sends an EDU to the remote server with the current presence state.
+    if old_state.state != new_state.state:
+        # Nothing to report.
+        return True
 
-        Args:
-            user(UserID): The user to push the presence state for.
-            destination(str): The remote server to send state to.
-            state(dict): The state to push, or None to use the current stored
-                state.
-        Returns:
-            A Deferred.
-        """
-        if state is None:
-            state = yield self.store.get_presence_state(user.localpart)
-            del state["mtime"]
-            state["presence"] = state.pop("state")
-
-            if user in self._user_cachemap:
-                state["last_active"] = (
-                    self._user_cachemap[user].get_state()["last_active"]
-                )
+    return False
 
-            yield collect_presencelike_data(self.distributor, user, state)
 
-        if "last_active" in state:
-            state = dict(state)
-            state["last_active_ago"] = int(
-                self.clock.time_msec() - state.pop("last_active")
-            )
-
-        user_state = {"user_id": user.to_string(), }
-        user_state.update(state)
+def _format_user_presence_state(state, now):
+    """Convert UserPresenceState to a format that can be sent down to clients
+    and to other servers.
+    """
+    content = {
+        "presence": state.state,
+        "user_id": state.user_id,
+    }
+    if state.last_active:
+        content["last_active_ago"] = now - state.last_active
+    if state.status_msg and state.state != PresenceState.OFFLINE:
+        content["status_msg"] = state.status_msg
+    if state.state == PresenceState.ONLINE:
+        content["currently_active"] = state.currently_active
 
-        yield self.federation.send_edu(
-            destination=destination,
-            edu_type="m.presence",
-            content={"push": [user_state, ], }
-        )
+    return content
 
 
 class PresenceEventSource(object):
     def __init__(self, hs):
         self.hs = hs
         self.clock = hs.get_clock()
+        self.store = hs.get_datastore()
 
     @defer.inlineCallbacks
     @log_function
-    def get_new_events(self, user, from_key, room_ids=None, **kwargs):
-        from_key = int(from_key)
+    def get_new_events(self, user, from_key, room_ids=None, include_offline=True,
+                       **kwargs):
+        # The process for getting presence events are:
+        #  1. Get the rooms the user is in.
+        #  2. Get the list of user in the rooms.
+        #  3. Get the list of users that are in the user's presence list.
+        #  4. If there is a from_key set, cross reference the list of users
+        #     with the `presence_stream_cache` to see which ones we actually
+        #     need to check.
+        #  5. Load current state for the users.
+        #
+        # We don't try and limit the presence updates by the current token, as
+        # sending down the rare duplicate is not a concern.
+
+        user_id = user.to_string()
+        if from_key is not None:
+            from_key = int(from_key)
         room_ids = room_ids or []
 
         presence = self.hs.get_handlers().presence_handler
-        cachemap = presence._user_cachemap
-
-        max_serial = presence._user_cachemap_latest_serial
-
-        clock = self.clock
-        latest_serial = 0
-
-        user_ids_to_check = {user}
-        presence_list = yield presence.store.get_presence_list(
-            user.localpart, accepted=True
-        )
-        if presence_list is not None:
-            user_ids_to_check |= set(
-                UserID.from_string(p["observed_user_id"]) for p in presence_list
-            )
-        for room_id in set(room_ids) & set(presence._room_serials):
-            if presence._room_serials[room_id] > from_key:
-                joined = yield presence.get_joined_users_for_room_id(room_id)
-                user_ids_to_check |= set(joined)
 
-        updates = []
-        for observed_user in user_ids_to_check & set(cachemap):
-            cached = cachemap[observed_user]
-
-            if cached.serial <= from_key or cached.serial > max_serial:
-                continue
-
-            latest_serial = max(cached.serial, latest_serial)
-            updates.append(cached.make_event(user=observed_user, clock=clock))
+        if not room_ids:
+            rooms = yield self.store.get_rooms_for_user(user_id)
+            room_ids = set(e.room_id for e in rooms)
 
-        # TODO(paul): limit
-
-        for serial, user_ids in presence._remote_offline_serials:
-            if serial <= from_key:
-                break
-
-            if serial > max_serial:
-                continue
-
-            latest_serial = max(latest_serial, serial)
-            for u in user_ids:
-                updates.append({
-                    "type": "m.presence",
-                    "content": {"user_id": u, "presence": PresenceState.OFFLINE},
-                })
-        # TODO(paul): For the v2 API we want to tell the client their from_key
-        #   is too old if we fell off the end of the _remote_offline_serials
-        #   list, and get them to invalidate+resync. In v1 we have no such
-        #   concept so this is a best-effort result.
-
-        if updates:
-            defer.returnValue((updates, latest_serial))
-        else:
-            defer.returnValue(([], presence._user_cachemap_latest_serial))
-
-    def get_current_key(self):
-        presence = self.hs.get_handlers().presence_handler
-        return presence._user_cachemap_latest_serial
+        user_ids_to_check = set()
+        for room_id in room_ids:
+            users = yield self.store.get_users_in_room(room_id)
+            user_ids_to_check.update(users)
 
-    @defer.inlineCallbacks
-    def get_pagination_rows(self, user, pagination_config, key):
-        # TODO (erikj): Does this make sense? Ordering?
+        plist = yield self.store.get_presence_list_accepted(user.localpart)
+        user_ids_to_check.update([row["observed_user_id"] for row in plist])
 
-        from_key = int(pagination_config.from_key)
+        # Always include yourself. Only really matters for when the user is
+        # not in any rooms, but still.
+        user_ids_to_check.add(user_id)
 
-        if pagination_config.to_key:
-            to_key = int(pagination_config.to_key)
-        else:
-            to_key = -1
+        max_token = self.store.get_current_presence_token()
 
-        presence = self.hs.get_handlers().presence_handler
-        cachemap = presence._user_cachemap
-
-        user_ids_to_check = {user}
-        presence_list = yield presence.store.get_presence_list(
-            user.localpart, accepted=True
-        )
-        if presence_list is not None:
-            user_ids_to_check |= set(
-                UserID.from_string(p["observed_user_id"]) for p in presence_list
+        if from_key:
+            user_ids_changed = self.store.presence_stream_cache.get_entities_changed(
+                user_ids_to_check, from_key,
             )
-        room_ids = yield presence.get_joined_rooms_for_user(user)
-        for room_id in set(room_ids) & set(presence._room_serials):
-            if presence._room_serials[room_id] >= from_key:
-                joined = yield presence.get_joined_users_for_room_id(room_id)
-                user_ids_to_check |= set(joined)
-
-        updates = []
-        for observed_user in user_ids_to_check & set(cachemap):
-            if not (to_key < cachemap[observed_user].serial <= from_key):
-                continue
-
-            updates.append((observed_user, cachemap[observed_user]))
-
-        # TODO(paul): limit
-
-        if updates:
-            clock = self.clock
-
-            earliest_serial = max([x[1].serial for x in updates])
-            data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
-
-            defer.returnValue((data, earliest_serial))
         else:
-            defer.returnValue(([], 0))
-
+            user_ids_changed = user_ids_to_check
 
-class UserPresenceCache(object):
-    """Store an observed user's state and status message.
+        updates = yield presence.current_state_for_users(user_ids_changed)
 
-    Includes the update timestamp.
-    """
-    def __init__(self):
-        self.state = {"presence": PresenceState.OFFLINE}
-        self.serial = None
-
-    def __repr__(self):
-        return "UserPresenceCache(state=%r, serial=%r)" % (
-            self.state, self.serial
-        )
-
-    def update(self, state, serial):
-        assert("mtime_age" not in state)
+        now = self.clock.time_msec()
 
-        self.state.update(state)
-        # Delete keys that are now 'None'
-        for k in self.state.keys():
-            if self.state[k] is None:
-                del self.state[k]
-
-        self.serial = serial
-
-        if "status_msg" in state:
-            self.status_msg = state["status_msg"]
-        else:
-            self.status_msg = None
-
-    def get_state(self):
-        # clone it so caller can't break our cache
-        state = dict(self.state)
-        return state
-
-    def make_event(self, user, clock):
-        content = self.get_state()
-        content["user_id"] = user.to_string()
+        defer.returnValue(([
+            {
+                "type": "m.presence",
+                "content": _format_user_presence_state(s, now),
+            }
+            for s in updates.values()
+            if include_offline or s.state != PresenceState.OFFLINE
+        ], max_token))
 
-        if "last_active" in content:
-            content["last_active_ago"] = int(
-                clock.time_msec() - content.pop("last_active")
-            )
+    def get_current_key(self):
+        return self.store.get_current_presence_token()
 
-        return {"type": "m.presence", "content": content}
+    def get_pagination_rows(self, user, pagination_config, key):
+        return self.get_new_events(user, from_key=None, include_offline=False)
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index 629e6e3594..7084a7396f 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -49,6 +49,9 @@ class ProfileHandler(BaseHandler):
         distributor = hs.get_distributor()
         self.distributor = distributor
 
+        distributor.declare("collect_presencelike_data")
+        distributor.declare("changed_presencelike_data")
+
         distributor.observe("registered_user", self.registered_user)
 
         distributor.observe(
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 1d0f0058a2..c5c13e085b 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -582,6 +582,28 @@ class SyncHandler(BaseHandler):
             if room_sync:
                 joined.append(room_sync)
 
+        # For each newly joined room, we want to send down presence of
+        # existing users.
+        presence_handler = self.hs.get_handlers().presence_handler
+        extra_presence_users = set()
+        for room_id in newly_joined_rooms:
+            users = yield self.store.get_users_in_room(event.room_id)
+            extra_presence_users.update(users)
+
+        # For each new member, send down presence.
+        for joined_sync in joined:
+            it = itertools.chain(joined_sync.timeline.events, joined_sync.state.values())
+            for event in it:
+                if event.type == EventTypes.Member:
+                    if event.membership == Membership.JOIN:
+                        extra_presence_users.add(event.state_key)
+
+        states = yield presence_handler.get_states(
+            [u for u in extra_presence_users if u != user_id],
+            as_event=True,
+        )
+        presence.extend(states)
+
         account_data_for_user = sync_config.filter_collection.filter_account_data(
             self.account_data_for_user(account_data)
         )
diff --git a/synapse/rest/client/v1/presence.py b/synapse/rest/client/v1/presence.py
index a6f8754e32..27ea5f2a43 100644
--- a/synapse/rest/client/v1/presence.py
+++ b/synapse/rest/client/v1/presence.py
@@ -17,7 +17,7 @@
 """
 from twisted.internet import defer
 
-from synapse.api.errors import SynapseError
+from synapse.api.errors import SynapseError, AuthError
 from synapse.types import UserID
 from .base import ClientV1RestServlet, client_path_patterns
 
@@ -35,8 +35,15 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
         requester = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
-        state = yield self.handlers.presence_handler.get_state(
-            target_user=user, auth_user=requester.user)
+        if requester.user != user:
+            allowed = yield self.handlers.presence_handler.is_visible(
+                observed_user=user, observer_user=requester.user,
+            )
+
+            if not allowed:
+                raise AuthError(403, "You are allowed to see their presence.")
+
+        state = yield self.handlers.presence_handler.get_state(target_user=user)
 
         defer.returnValue((200, state))
 
@@ -45,6 +52,9 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
         requester = yield self.auth.get_user_by_req(request)
         user = UserID.from_string(user_id)
 
+        if requester.user != user:
+            raise AuthError(403, "Can only set your own presence state")
+
         state = {}
         try:
             content = json.loads(request.content.read())
@@ -63,8 +73,7 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
         except:
             raise SynapseError(400, "Unable to parse state")
 
-        yield self.handlers.presence_handler.set_state(
-            target_user=user, auth_user=requester.user, state=state)
+        yield self.handlers.presence_handler.set_state(user, state)
 
         defer.returnValue((200, {}))
 
@@ -87,11 +96,8 @@ class PresenceListRestServlet(ClientV1RestServlet):
             raise SynapseError(400, "Cannot get another user's presence list")
 
         presence = yield self.handlers.presence_handler.get_presence_list(
-            observer_user=user, accepted=True)
-
-        for p in presence:
-            observed_user = p.pop("observed_user")
-            p["user_id"] = observed_user.to_string()
+            observer_user=user, accepted=True
+        )
 
         defer.returnValue((200, presence))
 
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 24706f9383..a8e89c7fe9 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -304,18 +304,6 @@ class RoomMemberListRestServlet(ClientV1RestServlet):
             if event["type"] != EventTypes.Member:
                 continue
             chunk.append(event)
-            # FIXME: should probably be state_key here, not user_id
-            target_user = UserID.from_string(event["user_id"])
-            # Presence is an optional cache; don't fail if we can't fetch it
-            try:
-                presence_handler = self.handlers.presence_handler
-                presence_state = yield presence_handler.get_state(
-                    target_user=target_user,
-                    auth_user=requester.user,
-                )
-                event["content"].update(presence_state)
-            except:
-                pass
 
         defer.returnValue((200, {
             "chunk": chunk
@@ -541,6 +529,10 @@ class RoomTypingRestServlet(ClientV1RestServlet):
         "/rooms/(?P<room_id>[^/]*)/typing/(?P<user_id>[^/]*)$"
     )
 
+    def __init__(self, hs):
+        super(RoomTypingRestServlet, self).__init__(hs)
+        self.presence_handler = hs.get_handlers().presence_handler
+
     @defer.inlineCallbacks
     def on_PUT(self, request, room_id, user_id):
         requester = yield self.auth.get_user_by_req(request)
@@ -552,6 +544,8 @@ class RoomTypingRestServlet(ClientV1RestServlet):
 
         typing_handler = self.handlers.typing_notification_handler
 
+        yield self.presence_handler.bump_presence_active_time(requester.user)
+
         if content["typing"]:
             yield typing_handler.started_typing(
                 target_user=target_user,
diff --git a/synapse/rest/client/v2_alpha/receipts.py b/synapse/rest/client/v2_alpha/receipts.py
index eb4b369a3d..b831d8c95e 100644
--- a/synapse/rest/client/v2_alpha/receipts.py
+++ b/synapse/rest/client/v2_alpha/receipts.py
@@ -37,6 +37,7 @@ class ReceiptRestServlet(RestServlet):
         self.hs = hs
         self.auth = hs.get_auth()
         self.receipts_handler = hs.get_handlers().receipts_handler
+        self.presence_handler = hs.get_handlers().presence_handler
 
     @defer.inlineCallbacks
     def on_POST(self, request, room_id, receipt_type, event_id):
@@ -45,6 +46,8 @@ class ReceiptRestServlet(RestServlet):
         if receipt_type != "m.read":
             raise SynapseError(400, "Receipt type must be 'm.read'")
 
+        yield self.presence_handler.bump_presence_active_time(requester.user)
+
         yield self.receipts_handler.received_client_receipt(
             room_id,
             receipt_type,
diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py
index accbc6cfac..de4a020ad4 100644
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -25,6 +25,7 @@ from synapse.events.utils import (
 )
 from synapse.api.filtering import FilterCollection, DEFAULT_FILTER_COLLECTION
 from synapse.api.errors import SynapseError
+from synapse.api.constants import PresenceState
 from ._base import client_v2_patterns
 
 import copy
@@ -82,6 +83,7 @@ class SyncRestServlet(RestServlet):
         self.sync_handler = hs.get_handlers().sync_handler
         self.clock = hs.get_clock()
         self.filtering = hs.get_filtering()
+        self.presence_handler = hs.get_handlers().presence_handler
 
     @defer.inlineCallbacks
     def on_GET(self, request):
@@ -139,17 +141,19 @@ class SyncRestServlet(RestServlet):
         else:
             since_token = None
 
-        if set_presence == "online":
-            yield self.event_stream_handler.started_stream(user)
+        affect_presence = set_presence != PresenceState.OFFLINE
 
-        try:
+        if affect_presence:
+            yield self.presence_handler.set_state(user, {"presence": set_presence})
+
+        context = yield self.presence_handler.user_syncing(
+            user.to_string(), affect_presence=affect_presence,
+        )
+        with context:
             sync_result = yield self.sync_handler.wait_for_sync_for_user(
                 sync_config, since_token=since_token, timeout=timeout,
                 full_state=full_state
             )
-        finally:
-            if set_presence == "online":
-                self.event_stream_handler.stopped_stream(user)
 
         time_now = self.clock.time_msec()
 
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 5a9e7720d9..8c3cf9e801 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -20,7 +20,7 @@ from .appservice import (
 from ._base import Cache
 from .directory import DirectoryStore
 from .events import EventsStore
-from .presence import PresenceStore
+from .presence import PresenceStore, UserPresenceState
 from .profile import ProfileStore
 from .registration import RegistrationStore
 from .room import RoomStore
@@ -47,6 +47,7 @@ from .account_data import AccountDataStore
 
 from util.id_generators import IdGenerator, StreamIdGenerator
 
+from synapse.api.constants import PresenceState
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 
 
@@ -110,6 +111,9 @@ class DataStore(RoomMemberStore, RoomStore,
         self._account_data_id_gen = StreamIdGenerator(
             db_conn, "account_data_max_stream_id", "stream_id"
         )
+        self._presence_id_gen = StreamIdGenerator(
+            db_conn, "presence_stream", "stream_id"
+        )
 
         self._transaction_id_gen = IdGenerator("sent_transactions", "id", self)
         self._state_groups_id_gen = IdGenerator("state_groups", "id", self)
@@ -119,7 +123,7 @@ class DataStore(RoomMemberStore, RoomStore,
         self._push_rule_id_gen = IdGenerator("push_rules", "id", self)
         self._push_rules_enable_id_gen = IdGenerator("push_rules_enable", "id", self)
 
-        events_max = self._stream_id_gen.get_max_token(None)
+        events_max = self._stream_id_gen.get_max_token()
         event_cache_prefill, min_event_val = self._get_cache_dict(
             db_conn, "events",
             entity_column="room_id",
@@ -135,13 +139,31 @@ class DataStore(RoomMemberStore, RoomStore,
             "MembershipStreamChangeCache", events_max,
         )
 
-        account_max = self._account_data_id_gen.get_max_token(None)
+        account_max = self._account_data_id_gen.get_max_token()
         self._account_data_stream_cache = StreamChangeCache(
             "AccountDataAndTagsChangeCache", account_max,
         )
 
+        self.__presence_on_startup = self._get_active_presence(db_conn)
+
+        presence_cache_prefill, min_presence_val = self._get_cache_dict(
+            db_conn, "presence_stream",
+            entity_column="user_id",
+            stream_column="stream_id",
+            max_value=self._presence_id_gen.get_max_token(),
+        )
+        self.presence_stream_cache = StreamChangeCache(
+            "PresenceStreamChangeCache", min_presence_val,
+            prefilled_cache=presence_cache_prefill
+        )
+
         super(DataStore, self).__init__(hs)
 
+    def take_presence_startup_info(self):
+        active_on_startup = self.__presence_on_startup
+        self.__presence_on_startup = None
+        return active_on_startup
+
     def _get_cache_dict(self, db_conn, table, entity_column, stream_column, max_value):
         # Fetch a mapping of room_id -> max stream position for "recent" rooms.
         # It doesn't really matter how many we get, the StreamChangeCache will
@@ -161,6 +183,7 @@ class DataStore(RoomMemberStore, RoomStore,
         txn = db_conn.cursor()
         txn.execute(sql, (int(max_value),))
         rows = txn.fetchall()
+        txn.close()
 
         cache = {
             row[0]: int(row[1])
@@ -174,6 +197,27 @@ class DataStore(RoomMemberStore, RoomStore,
 
         return cache, min_val
 
+    def _get_active_presence(self, db_conn):
+        """Fetch non-offline presence from the database so that we can register
+        the appropriate time outs.
+        """
+
+        sql = (
+            "SELECT user_id, state, last_active, last_federation_update,"
+            " last_user_sync, status_msg, currently_active FROM presence_stream"
+            " WHERE state != ?"
+        )
+        sql = self.database_engine.convert_param_style(sql)
+
+        txn = db_conn.cursor()
+        txn.execute(sql, (PresenceState.OFFLINE,))
+        rows = self.cursor_to_dict(txn)
+
+        for row in rows:
+            row["currently_active"] = bool(row["currently_active"])
+
+        return [UserPresenceState(**row) for row in rows]
+
     @defer.inlineCallbacks
     def insert_client_ip(self, user, access_token, ip, user_agent):
         now = int(self._clock.time_msec())
diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py
index 850736c85e..0fd5d497ab 100644
--- a/synapse/storage/prepare_database.py
+++ b/synapse/storage/prepare_database.py
@@ -25,7 +25,7 @@ logger = logging.getLogger(__name__)
 
 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 29
+SCHEMA_VERSION = 30
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 
diff --git a/synapse/storage/presence.py b/synapse/storage/presence.py
index ef525f34c5..b133979102 100644
--- a/synapse/storage/presence.py
+++ b/synapse/storage/presence.py
@@ -14,73 +14,128 @@
 # limitations under the License.
 
 from ._base import SQLBaseStore
-from synapse.util.caches.descriptors import cached, cachedList
+from synapse.api.constants import PresenceState
+from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
 
+from collections import namedtuple
 from twisted.internet import defer
 
 
-class PresenceStore(SQLBaseStore):
-    def create_presence(self, user_localpart):
-        res = self._simple_insert(
-            table="presence",
-            values={"user_id": user_localpart},
-            desc="create_presence",
+class UserPresenceState(namedtuple("UserPresenceState",
+                        ("user_id", "state", "last_active", "last_federation_update",
+                            "last_user_sync", "status_msg", "currently_active"))):
+    """Represents the current presence state of the user.
+
+    user_id (str)
+    last_active (int): Time in msec that the user last interacted with server.
+    last_federation_update (int): Time in msec since either a) we sent a presence
+        update to other servers or b) we received a presence update, depending
+        on if is a local user or not.
+    last_user_sync (int): Time in msec that the user last *completed* a sync
+        (or event stream).
+    status_msg (str): User set status message.
+    """
+
+    def copy_and_replace(self, **kwargs):
+        return self._replace(**kwargs)
+
+    @classmethod
+    def default(cls, user_id):
+        """Returns a default presence state.
+        """
+        return cls(
+            user_id=user_id,
+            state=PresenceState.OFFLINE,
+            last_active=0,
+            last_federation_update=0,
+            last_user_sync=0,
+            status_msg=None,
+            currently_active=False,
         )
 
-        self.get_presence_state.invalidate((user_localpart,))
-        return res
 
-    def has_presence_state(self, user_localpart):
-        return self._simple_select_one(
-            table="presence",
-            keyvalues={"user_id": user_localpart},
-            retcols=["user_id"],
-            allow_none=True,
-            desc="has_presence_state",
+class PresenceStore(SQLBaseStore):
+    @defer.inlineCallbacks
+    def update_presence(self, presence_states):
+        stream_id_manager = yield self._presence_id_gen.get_next(self)
+        with stream_id_manager as stream_id:
+            yield self.runInteraction(
+                "update_presence",
+                self._update_presence_txn, stream_id, presence_states,
+            )
+
+        defer.returnValue((stream_id, self._presence_id_gen.get_max_token()))
+
+    def _update_presence_txn(self, txn, stream_id, presence_states):
+        for state in presence_states:
+            txn.call_after(
+                self.presence_stream_cache.entity_has_changed,
+                state.user_id, stream_id,
+            )
+
+        # Actually insert new rows
+        self._simple_insert_many_txn(
+            txn,
+            table="presence_stream",
+            values=[
+                {
+                    "stream_id": stream_id,
+                    "user_id": state.user_id,
+                    "state": state.state,
+                    "last_active": state.last_active,
+                    "last_federation_update": state.last_federation_update,
+                    "last_user_sync": state.last_user_sync,
+                    "status_msg": state.status_msg,
+                    "currently_active": state.currently_active,
+                }
+                for state in presence_states
+            ],
         )
 
-    @cached(max_entries=2000)
-    def get_presence_state(self, user_localpart):
-        return self._simple_select_one(
-            table="presence",
-            keyvalues={"user_id": user_localpart},
-            retcols=["state", "status_msg", "mtime"],
-            desc="get_presence_state",
+        # Delete old rows to stop database from getting really big
+        sql = (
+            "DELETE FROM presence_stream WHERE"
+            " stream_id < ?"
+            " AND user_id IN (%s)"
         )
 
-    @cachedList(get_presence_state.cache, list_name="user_localparts",
-                inlineCallbacks=True)
-    def get_presence_states(self, user_localparts):
+        batches = (
+            presence_states[i:i + 50]
+            for i in xrange(0, len(presence_states), 50)
+        )
+        for states in batches:
+            args = [stream_id]
+            args.extend(s.user_id for s in states)
+            txn.execute(
+                sql % (",".join("?" for _ in states),),
+                args
+            )
+
+    @defer.inlineCallbacks
+    def get_presence_for_users(self, user_ids):
         rows = yield self._simple_select_many_batch(
-            table="presence",
+            table="presence_stream",
             column="user_id",
-            iterable=user_localparts,
-            retcols=("user_id", "state", "status_msg", "mtime",),
-            desc="get_presence_states",
+            iterable=user_ids,
+            keyvalues={},
+            retcols=(
+                "user_id",
+                "state",
+                "last_active",
+                "last_federation_update",
+                "last_user_sync",
+                "status_msg",
+                "currently_active",
+            ),
         )
 
-        defer.returnValue({
-            row["user_id"]: {
-                "state": row["state"],
-                "status_msg": row["status_msg"],
-                "mtime": row["mtime"],
-            }
-            for row in rows
-        })
+        for row in rows:
+            row["currently_active"] = bool(row["currently_active"])
 
-    @defer.inlineCallbacks
-    def set_presence_state(self, user_localpart, new_state):
-        res = yield self._simple_update_one(
-            table="presence",
-            keyvalues={"user_id": user_localpart},
-            updatevalues={"state": new_state["state"],
-                          "status_msg": new_state["status_msg"],
-                          "mtime": self._clock.time_msec()},
-            desc="set_presence_state",
-        )
+        defer.returnValue([UserPresenceState(**row) for row in rows])
 
-        self.get_presence_state.invalidate((user_localpart,))
-        defer.returnValue(res)
+    def get_current_presence_token(self):
+        return self._presence_id_gen.get_max_token()
 
     def allow_presence_visible(self, observed_localpart, observer_userid):
         return self._simple_insert(
@@ -128,6 +183,7 @@ class PresenceStore(SQLBaseStore):
             desc="set_presence_list_accepted",
         )
         self.get_presence_list_accepted.invalidate((observer_localpart,))
+        self.get_presence_list_observers_accepted.invalidate((observed_userid,))
         defer.returnValue(result)
 
     def get_presence_list(self, observer_localpart, accepted=None):
@@ -154,6 +210,19 @@ class PresenceStore(SQLBaseStore):
             desc="get_presence_list_accepted",
         )
 
+    @cachedInlineCallbacks()
+    def get_presence_list_observers_accepted(self, observed_userid):
+        user_localparts = yield self._simple_select_onecol(
+            table="presence_list",
+            keyvalues={"observed_user_id": observed_userid, "accepted": True},
+            retcol="user_id",
+            desc="get_presence_list_accepted",
+        )
+
+        defer.returnValue([
+            "@%s:%s" % (u, self.hs.hostname,) for u in user_localparts
+        ])
+
     @defer.inlineCallbacks
     def del_presence_list(self, observer_localpart, observed_userid):
         yield self._simple_delete_one(
@@ -163,3 +232,4 @@ class PresenceStore(SQLBaseStore):
             desc="del_presence_list",
         )
         self.get_presence_list_accepted.invalidate((observer_localpart,))
+        self.get_presence_list_observers_accepted.invalidate((observed_userid,))
diff --git a/synapse/storage/schema/delta/30/presence_stream.sql b/synapse/storage/schema/delta/30/presence_stream.sql
new file mode 100644
index 0000000000..14f5e3d30a
--- /dev/null
+++ b/synapse/storage/schema/delta/30/presence_stream.sql
@@ -0,0 +1,30 @@
+/* Copyright 2016 OpenMarket Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+ CREATE TABLE presence_stream(
+     stream_id BIGINT,
+     user_id TEXT,
+     state TEXT,
+     last_active BIGINT,
+     last_federation_update BIGINT,
+     last_user_sync BIGINT,
+     status_msg TEXT,
+     currently_active BOOLEAN
+ );
+
+ CREATE INDEX presence_stream_id ON presence_stream(stream_id, user_id);
+ CREATE INDEX presence_stream_user_id ON presence_stream(user_id);
+ CREATE INDEX presence_stream_state ON presence_stream(state);
diff --git a/synapse/storage/util/id_generators.py b/synapse/storage/util/id_generators.py
index 5c522f4ab9..5ce54f76de 100644
--- a/synapse/storage/util/id_generators.py
+++ b/synapse/storage/util/id_generators.py
@@ -130,9 +130,11 @@ class StreamIdGenerator(object):
 
         return manager()
 
-    def get_max_token(self, store):
+    def get_max_token(self, *args):
         """Returns the maximum stream id such that all stream ids less than or
         equal to it have been successfully persisted.
+
+        Used to take a DataStore param, which is no longer needed.
         """
         with self._lock:
             if self._unfinished_ids:
diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py
index 133671e238..3b9da5b34a 100644
--- a/synapse/util/__init__.py
+++ b/synapse/util/__init__.py
@@ -42,7 +42,7 @@ class Clock(object):
 
     def time_msec(self):
         """Returns the current system time in miliseconds since epoch."""
-        return self.time() * 1000
+        return int(self.time() * 1000)
 
     def looping_call(self, f, msec):
         l = task.LoopingCall(f)
diff --git a/tests/utils.py b/tests/utils.py
index 3b1eb50d8d..f71125042b 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -224,12 +224,12 @@ class MockClock(object):
     def time_msec(self):
         return self.time() * 1000
 
-    def call_later(self, delay, callback):
+    def call_later(self, delay, callback, *args, **kwargs):
         current_context = LoggingContext.current_context()
 
         def wrapped_callback():
             LoggingContext.thread_local.current_context = current_context
-            callback()
+            callback(*args, **kwargs)
 
         t = [self.now + delay, wrapped_callback, False]
         self.timers.append(t)
-- 
cgit 1.4.1


From 591af2d074044a70a48b033c4dfc322f58189d3e Mon Sep 17 00:00:00 2001
From: Daniel Wagner-Hall <daniel@matrix.org>
Date: Wed, 17 Feb 2016 15:50:13 +0000
Subject: Some cleanup

I'm not particularly happy with the "action" switching, but there's no
convenient way to defer the work that needs to happen after it, so... :(
---
 synapse/handlers/room.py       | 124 +++++++++++++++++++----------------------
 synapse/rest/client/v1/room.py |   6 +-
 2 files changed, 61 insertions(+), 69 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index f85a5f2677..cd04ac09fa 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -397,7 +397,7 @@ class RoomMemberHandler(BaseHandler):
             room_id,
             action,
             txn_id=None,
-            room_hosts=None,
+            remote_room_hosts=None,
             ratelimit=True,
     ):
         effective_membership_state = action
@@ -448,7 +448,7 @@ class RoomMemberHandler(BaseHandler):
             context,
             is_guest=requester.is_guest,
             ratelimit=ratelimit,
-            room_hosts=room_hosts,
+            remote_room_hosts=remote_room_hosts,
             from_client=True,
         )
 
@@ -461,11 +461,12 @@ class RoomMemberHandler(BaseHandler):
             event,
             context,
             is_guest=False,
-            room_hosts=None,
+            remote_room_hosts=None,
             ratelimit=True,
             from_client=True,
     ):
-        """ Change the membership status of a user in a room.
+        """
+        Change the membership status of a user in a room.
 
         Args:
             event (SynapseEvent): The membership event.
@@ -482,78 +483,64 @@ class RoomMemberHandler(BaseHandler):
         Raises:
             SynapseError if there was a problem changing the membership.
         """
-        user = UserID.from_string(event.sender)
+        target_user = UserID.from_string(event.state_key)
+        room_id = event.room_id
 
         if from_client:
-            assert self.hs.is_mine(user), "User must be our own: %s" % (user,)
+            sender = UserID.from_string(event.sender)
+            assert self.hs.is_mine(sender), "Sender must be our own: %s" % (sender,)
 
         if event.is_state():
             message_handler = self.hs.get_handlers().message_handler
-            prev_state = message_handler.deduplicate_state_event(event, context)
-            if prev_state is not None:
+            prev_event = message_handler.deduplicate_state_event(event, context)
+            if prev_event is not None:
                 return
 
-        target_user = UserID.from_string(event.state_key)
-
-        prev_state = context.current_state.get(
-            (EventTypes.Member, target_user.to_string()),
-            None
-        )
-
-        room_id = event.room_id
-
-        handled = False
+        action = "send"
 
         if event.membership == Membership.JOIN:
             if is_guest and not self._can_guest_join(context.current_state):
                 # This should be an auth check, but guests are a local concept,
                 # so don't really fit into the general auth process.
                 raise AuthError(403, "Guest access not allowed")
-
-            should_do_dance, room_hosts = self._should_do_dance(
+            do_remote_join_dance, remote_room_hosts = self._should_do_dance(
                 context,
-                (self.get_inviter(target_user.to_string(), context.current_state)),
-                room_hosts,
+                (self.get_inviter(event.state_key, context.current_state)),
+                remote_room_hosts,
             )
-
-            if should_do_dance:
-                if len(room_hosts) == 0:
-                    # return the same error as join_room_alias does
-                    raise SynapseError(404, "No known servers")
-
-                # We don't do an auth check if we are doing an invite
-                # join dance for now, since we're kinda implicitly checking
-                # that we are allowed to join when we decide whether or not we
-                # need to do the invite/join dance.
-                yield self.hs.get_handlers().federation_handler.do_invite_join(
-                    room_hosts,
-                    room_id,
-                    event.user_id,
-                    event.content,
-                )
-                handled = True
-        if event.membership == Membership.LEAVE:
+            if do_remote_join_dance:
+                action = "remote_join"
+        elif event.membership == Membership.LEAVE:
             is_host_in_room = self.is_host_in_room(context.current_state)
             if not is_host_in_room:
-                # Rejecting an invite, rather than leaving a joined room
-                handler = self.hs.get_handlers().federation_handler
-                inviter = self.get_inviter(target_user.to_string(), context.current_state)
-                if not inviter:
-                    # return the same error as join_room_alias does
-                    raise SynapseError(404, "No known servers")
-                yield handler.do_remotely_reject_invite(
-                    [inviter.domain],
-                    room_id,
-                    event.user_id
-                )
-                handled = True
-
-            # FIXME: This isn't idempotency.
-            if prev_state and prev_state.membership == event.membership:
-                # double same action, treat this event as a NOOP.
-                return
-
-        if not handled:
+                action = "remote_reject"
+
+        federation_handler = self.hs.get_handlers().federation_handler
+
+        if action == "remote_join":
+            if len(remote_room_hosts) == 0:
+                raise SynapseError(404, "No known servers")
+
+            # We don't do an auth check if we are doing an invite
+            # join dance for now, since we're kinda implicitly checking
+            # that we are allowed to join when we decide whether or not we
+            # need to do the invite/join dance.
+            yield federation_handler.do_invite_join(
+                remote_room_hosts,
+                event.room_id,
+                event.user_id,
+                event.content,
+            )
+        elif action == "remote_reject":
+            inviter = self.get_inviter(target_user.to_string(), context.current_state)
+            if not inviter:
+                raise SynapseError(404, "No known servers")
+            yield federation_handler.do_remotely_reject_invite(
+                [inviter.domain],
+                room_id,
+                event.user_id
+            )
+        else:
             yield self.handle_new_client_event(
                 event,
                 context,
@@ -561,14 +548,19 @@ class RoomMemberHandler(BaseHandler):
                 ratelimit=ratelimit,
             )
 
+        prev_member_event = context.current_state.get(
+            (EventTypes.Member, target_user.to_string()),
+            None
+        )
+
         if event.membership == Membership.JOIN:
-            if not prev_state or prev_state.membership != Membership.JOIN:
+            if not prev_member_event or prev_member_event.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.
                 yield user_joined_room(self.distributor, target_user, room_id)
         elif event.membership == Membership.LEAVE:
-            if prev_state and prev_state.membership == Membership.JOIN:
+            if prev_member_event and prev_member_event.membership == Membership.JOIN:
                 user_left_room(self.distributor, target_user, room_id)
 
     def _can_guest_join(self, current_state):
@@ -604,7 +596,9 @@ class RoomMemberHandler(BaseHandler):
         Args:
             room_alias (RoomAlias): The alias to look up.
         Returns:
-            The room ID as a RoomID object.
+            A tuple of:
+                The room ID as a RoomID object.
+                Hosts likely to be participating in the room ([str]).
         Raises:
             SynapseError if room alias could not be found.
         """
@@ -615,11 +609,9 @@ class RoomMemberHandler(BaseHandler):
             raise SynapseError(404, "No such room alias")
 
         room_id = mapping["room_id"]
-        hosts = mapping["servers"]
-        if not hosts:
-            raise SynapseError(404, "No known servers")
+        servers = mapping["servers"]
 
-        defer.returnValue((RoomID.from_string(room_id), hosts))
+        defer.returnValue((RoomID.from_string(room_id), servers))
 
     def get_inviter(self, user_id, current_state):
         prev_state = current_state.get((EventTypes.Member, user_id))
diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py
index 179fe9a010..1f5ee09dcc 100644
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -230,11 +230,11 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
 
         if RoomID.is_valid(room_identifier):
             room_id = room_identifier
-            room_hosts = None
+            remote_room_hosts = None
         elif RoomAlias.is_valid(room_identifier):
             handler = self.handlers.room_member_handler
             room_alias = RoomAlias.from_string(room_identifier)
-            room_id, room_hosts = yield handler.lookup_room_alias(room_alias)
+            room_id, remote_room_hosts = yield handler.lookup_room_alias(room_alias)
             room_id = room_id.to_string()
         else:
             raise SynapseError(400, "%s was not legal room ID or room alias" % (
@@ -247,7 +247,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
             room_id=room_id,
             action="join",
             txn_id=txn_id,
-            room_hosts=room_hosts,
+            remote_room_hosts=remote_room_hosts,
         )
 
         defer.returnValue((200, {"room_id": room_id}))
-- 
cgit 1.4.1


From b9977ea667889f6cf89464c92fc57cbcae7cca28 Mon Sep 17 00:00:00 2001
From: Mark Haines <mjark@negativecurvature.net>
Date: Thu, 18 Feb 2016 16:05:13 +0000
Subject: Remove dead code for setting device specific rules.

It wasn't possible to hit the code from the API because of a typo
in parsing the request path. Since no-one was using the feature
we might as well remove the dead code.
---
 synapse/push/__init__.py                 |  7 ++-
 synapse/push/action_generator.py         |  2 +-
 synapse/push/bulk_push_rule_evaluator.py |  2 +-
 synapse/push/httppusher.py               |  3 +-
 synapse/push/push_rule_evaluator.py      | 15 ++----
 synapse/push/pusherpool.py               | 48 +++++++----------
 synapse/rest/client/v1/push_rule.py      | 90 ++------------------------------
 synapse/rest/client/v1/pusher.py         |  6 +--
 synapse/storage/event_push_actions.py    |  7 ++-
 synapse/storage/pusher.py                |  6 +--
 10 files changed, 45 insertions(+), 141 deletions(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py
index 8da2d8716c..4c6c3b83a2 100644
--- a/synapse/push/__init__.py
+++ b/synapse/push/__init__.py
@@ -47,14 +47,13 @@ class Pusher(object):
     MAX_BACKOFF = 60 * 60 * 1000
     GIVE_UP_AFTER = 24 * 60 * 60 * 1000
 
-    def __init__(self, _hs, profile_tag, user_id, app_id,
+    def __init__(self, _hs, user_id, app_id,
                  app_display_name, device_display_name, pushkey, pushkey_ts,
                  data, last_token, last_success, failing_since):
         self.hs = _hs
         self.evStreamHandler = self.hs.get_handlers().event_stream_handler
         self.store = self.hs.get_datastore()
         self.clock = self.hs.get_clock()
-        self.profile_tag = profile_tag
         self.user_id = user_id
         self.app_id = app_id
         self.app_display_name = app_display_name
@@ -186,8 +185,8 @@ class Pusher(object):
         processed = False
 
         rule_evaluator = yield \
-            push_rule_evaluator.evaluator_for_user_id_and_profile_tag(
-                self.user_id, self.profile_tag, single_event['room_id'], self.store
+            push_rule_evaluator.evaluator_for_user_id(
+                self.user_id, single_event['room_id'], self.store
             )
 
         actions = yield rule_evaluator.actions_for_event(single_event)
diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py
index e0da0868ec..c6c1dc769e 100644
--- a/synapse/push/action_generator.py
+++ b/synapse/push/action_generator.py
@@ -44,5 +44,5 @@ class ActionGenerator:
         )
 
         context.push_actions = [
-            (uid, None, actions) for uid, actions in actions_by_user.items()
+            (uid, actions) for uid, actions in actions_by_user.items()
         ]
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index 8ac5ceb9ef..0a23b3f102 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -152,7 +152,7 @@ def _condition_checker(evaluator, conditions, uid, display_name, cache):
             elif res is True:
                 continue
 
-        res = evaluator.matches(cond, uid, display_name, None)
+        res = evaluator.matches(cond, uid, display_name)
         if _id:
             cache[_id] = bool(res)
 
diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py
index cdc4494928..9be4869360 100644
--- a/synapse/push/httppusher.py
+++ b/synapse/push/httppusher.py
@@ -23,12 +23,11 @@ logger = logging.getLogger(__name__)
 
 
 class HttpPusher(Pusher):
-    def __init__(self, _hs, profile_tag, user_id, app_id,
+    def __init__(self, _hs, user_id, app_id,
                  app_display_name, device_display_name, pushkey, pushkey_ts,
                  data, last_token, last_success, failing_since):
         super(HttpPusher, self).__init__(
             _hs,
-            profile_tag,
             user_id,
             app_id,
             app_display_name,
diff --git a/synapse/push/push_rule_evaluator.py b/synapse/push/push_rule_evaluator.py
index 2a2b4437dc..98e2a2015e 100644
--- a/synapse/push/push_rule_evaluator.py
+++ b/synapse/push/push_rule_evaluator.py
@@ -33,7 +33,7 @@ INEQUALITY_EXPR = re.compile("^([=<>]*)([0-9]*)$")
 
 
 @defer.inlineCallbacks
-def evaluator_for_user_id_and_profile_tag(user_id, profile_tag, room_id, store):
+def evaluator_for_user_id(user_id, room_id, store):
     rawrules = yield store.get_push_rules_for_user(user_id)
     enabled_map = yield store.get_push_rules_enabled_for_user(user_id)
     our_member_event = yield store.get_current_state(
@@ -43,7 +43,7 @@ def evaluator_for_user_id_and_profile_tag(user_id, profile_tag, room_id, store):
     )
 
     defer.returnValue(PushRuleEvaluator(
-        user_id, profile_tag, rawrules, enabled_map,
+        user_id, rawrules, enabled_map,
         room_id, our_member_event, store
     ))
 
@@ -77,10 +77,9 @@ def _room_member_count(ev, condition, room_member_count):
 class PushRuleEvaluator:
     DEFAULT_ACTIONS = []
 
-    def __init__(self, user_id, profile_tag, raw_rules, enabled_map, room_id,
+    def __init__(self, user_id, raw_rules, enabled_map, room_id,
                  our_member_event, store):
         self.user_id = user_id
-        self.profile_tag = profile_tag
         self.room_id = room_id
         self.our_member_event = our_member_event
         self.store = store
@@ -152,7 +151,7 @@ class PushRuleEvaluator:
             matches = True
             for c in conditions:
                 matches = evaluator.matches(
-                    c, self.user_id, my_display_name, self.profile_tag
+                    c, self.user_id, my_display_name
                 )
                 if not matches:
                     break
@@ -189,13 +188,9 @@ class PushRuleEvaluatorForEvent(object):
         # Maps strings of e.g. 'content.body' -> event["content"]["body"]
         self._value_cache = _flatten_dict(event)
 
-    def matches(self, condition, user_id, display_name, profile_tag):
+    def matches(self, condition, user_id, display_name):
         if condition['kind'] == 'event_match':
             return self._event_match(condition, user_id)
-        elif condition['kind'] == 'device':
-            if 'profile_tag' not in condition:
-                return True
-            return condition['profile_tag'] == profile_tag
         elif condition['kind'] == 'contains_display_name':
             return self._contains_display_name(display_name)
         elif condition['kind'] == 'room_member_count':
diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py
index d7dcb2de4b..a05aa5f661 100644
--- a/synapse/push/pusherpool.py
+++ b/synapse/push/pusherpool.py
@@ -29,6 +29,7 @@ class PusherPool:
     def __init__(self, _hs):
         self.hs = _hs
         self.store = self.hs.get_datastore()
+        self.clock = self.hs.get_clock()
         self.pushers = {}
         self.last_pusher_started = -1
 
@@ -38,8 +39,11 @@ class PusherPool:
         self._start_pushers(pushers)
 
     @defer.inlineCallbacks
-    def add_pusher(self, user_id, access_token, profile_tag, kind, app_id,
-                   app_display_name, device_display_name, pushkey, lang, data):
+    def add_pusher(self, user_id, access_token, kind, app_id,
+                   app_display_name, device_display_name, pushkey, lang, data,
+                   profile_tag=""):
+        time_now_msec = self.clock.time_msec()
+
         # we try to create the pusher just to validate the config: it
         # will then get pulled out of the database,
         # recreated, added and started: this means we have only one
@@ -47,23 +51,31 @@ class PusherPool:
         self._create_pusher({
             "user_name": user_id,
             "kind": kind,
-            "profile_tag": profile_tag,
             "app_id": app_id,
             "app_display_name": app_display_name,
             "device_display_name": device_display_name,
             "pushkey": pushkey,
-            "ts": self.hs.get_clock().time_msec(),
+            "ts": time_now_msec,
             "lang": lang,
             "data": data,
             "last_token": None,
             "last_success": None,
             "failing_since": None
         })
-        yield self._add_pusher_to_store(
-            user_id, access_token, profile_tag, kind, app_id,
-            app_display_name, device_display_name,
-            pushkey, lang, data
+        yield self.store.add_pusher(
+            user_id=user_id,
+            access_token=access_token,
+            kind=kind,
+            app_id=app_id,
+            app_display_name=app_display_name,
+            device_display_name=device_display_name,
+            pushkey=pushkey,
+            pushkey_ts=time_now_msec,
+            lang=lang,
+            data=data,
+            profile_tag=profile_tag,
         )
+        yield self._refresh_pusher(app_id, pushkey, user_id)
 
     @defer.inlineCallbacks
     def remove_pushers_by_app_id_and_pushkey_not_user(self, app_id, pushkey,
@@ -94,30 +106,10 @@ class PusherPool:
                 )
                 yield self.remove_pusher(p['app_id'], p['pushkey'], p['user_name'])
 
-    @defer.inlineCallbacks
-    def _add_pusher_to_store(self, user_id, access_token, profile_tag, kind,
-                             app_id, app_display_name, device_display_name,
-                             pushkey, lang, data):
-        yield self.store.add_pusher(
-            user_id=user_id,
-            access_token=access_token,
-            profile_tag=profile_tag,
-            kind=kind,
-            app_id=app_id,
-            app_display_name=app_display_name,
-            device_display_name=device_display_name,
-            pushkey=pushkey,
-            pushkey_ts=self.hs.get_clock().time_msec(),
-            lang=lang,
-            data=data,
-        )
-        yield self._refresh_pusher(app_id, pushkey, user_id)
-
     def _create_pusher(self, pusherdict):
         if pusherdict['kind'] == 'http':
             return HttpPusher(
                 self.hs,
-                profile_tag=pusherdict['profile_tag'],
                 user_id=pusherdict['user_name'],
                 app_id=pusherdict['app_id'],
                 app_display_name=pusherdict['app_display_name'],
diff --git a/synapse/rest/client/v1/push_rule.py b/synapse/rest/client/v1/push_rule.py
index 7766b8be1d..5db2805d68 100644
--- a/synapse/rest/client/v1/push_rule.py
+++ b/synapse/rest/client/v1/push_rule.py
@@ -60,7 +60,6 @@ class PushRuleRestServlet(ClientV1RestServlet):
                 spec['template'],
                 spec['rule_id'],
                 content,
-                device=spec['device'] if 'device' in spec else None
             )
         except InvalidRuleException as e:
             raise SynapseError(400, e.message)
@@ -153,23 +152,7 @@ class PushRuleRestServlet(ClientV1RestServlet):
                 elif pattern_type == "user_localpart":
                     c["pattern"] = user.localpart
 
-            if r['priority_class'] > PRIORITY_CLASS_MAP['override']:
-                # per-device rule
-                profile_tag = _profile_tag_from_conditions(r["conditions"])
-                r = _strip_device_condition(r)
-                if not profile_tag:
-                    continue
-                if profile_tag not in rules['device']:
-                    rules['device'][profile_tag] = {}
-                    rules['device'][profile_tag] = (
-                        _add_empty_priority_class_arrays(
-                            rules['device'][profile_tag]
-                        )
-                    )
-
-                rulearray = rules['device'][profile_tag][template_name]
-            else:
-                rulearray = rules['global'][template_name]
+            rulearray = rules['global'][template_name]
 
             template_rule = _rule_to_template(r)
             if template_rule:
@@ -195,24 +178,6 @@ class PushRuleRestServlet(ClientV1RestServlet):
             path = path[1:]
             result = _filter_ruleset_with_path(rules['global'], path)
             defer.returnValue((200, result))
-        elif path[0] == 'device':
-            path = path[1:]
-            if path == []:
-                raise UnrecognizedRequestError(
-                    PushRuleRestServlet.SLIGHTLY_PEDANTIC_TRAILING_SLASH_ERROR
-                )
-            if path[0] == '':
-                defer.returnValue((200, rules['device']))
-
-            profile_tag = path[0]
-            path = path[1:]
-            if profile_tag not in rules['device']:
-                ret = {}
-                ret = _add_empty_priority_class_arrays(ret)
-                defer.returnValue((200, ret))
-            ruleset = rules['device'][profile_tag]
-            result = _filter_ruleset_with_path(ruleset, path)
-            defer.returnValue((200, result))
         else:
             raise UnrecognizedRequestError()
 
@@ -252,16 +217,9 @@ def _rule_spec_from_path(path):
 
     scope = path[1]
     path = path[2:]
-    if scope not in ['global', 'device']:
+    if scope != 'global':
         raise UnrecognizedRequestError()
 
-    device = None
-    if scope == 'device':
-        if len(path) == 0:
-            raise UnrecognizedRequestError()
-        device = path[0]
-        path = path[1:]
-
     if len(path) == 0:
         raise UnrecognizedRequestError()
 
@@ -278,8 +236,6 @@ def _rule_spec_from_path(path):
         'template': template,
         'rule_id': rule_id
     }
-    if device:
-        spec['profile_tag'] = device
 
     path = path[1:]
 
@@ -289,7 +245,7 @@ def _rule_spec_from_path(path):
     return spec
 
 
-def _rule_tuple_from_request_object(rule_template, rule_id, req_obj, device=None):
+def _rule_tuple_from_request_object(rule_template, rule_id, req_obj):
     if rule_template in ['override', 'underride']:
         if 'conditions' not in req_obj:
             raise InvalidRuleException("Missing 'conditions'")
@@ -322,12 +278,6 @@ def _rule_tuple_from_request_object(rule_template, rule_id, req_obj, device=None
     else:
         raise InvalidRuleException("Unknown rule template: %s" % (rule_template,))
 
-    if device:
-        conditions.append({
-            'kind': 'device',
-            'profile_tag': device
-        })
-
     if 'actions' not in req_obj:
         raise InvalidRuleException("No actions found")
     actions = req_obj['actions']
@@ -349,17 +299,6 @@ def _add_empty_priority_class_arrays(d):
     return d
 
 
-def _profile_tag_from_conditions(conditions):
-    """
-    Given a list of conditions, return the profile tag of the
-    device rule if there is one
-    """
-    for c in conditions:
-        if c['kind'] == 'device':
-            return c['profile_tag']
-    return None
-
-
 def _filter_ruleset_with_path(ruleset, path):
     if path == []:
         raise UnrecognizedRequestError(
@@ -403,19 +342,11 @@ def _priority_class_from_spec(spec):
         raise InvalidRuleException("Unknown template: %s" % (spec['template']))
     pc = PRIORITY_CLASS_MAP[spec['template']]
 
-    if spec['scope'] == 'device':
-        pc += len(PRIORITY_CLASS_MAP)
-
     return pc
 
 
 def _priority_class_to_template_name(pc):
-    if pc > PRIORITY_CLASS_MAP['override']:
-        # per-device
-        prio_class_index = pc - len(PRIORITY_CLASS_MAP)
-        return PRIORITY_CLASS_INVERSE_MAP[prio_class_index]
-    else:
-        return PRIORITY_CLASS_INVERSE_MAP[pc]
+    return PRIORITY_CLASS_INVERSE_MAP[pc]
 
 
 def _rule_to_template(rule):
@@ -445,23 +376,12 @@ def _rule_to_template(rule):
     return templaterule
 
 
-def _strip_device_condition(rule):
-    for i, c in enumerate(rule['conditions']):
-        if c['kind'] == 'device':
-            del rule['conditions'][i]
-    return rule
-
-
 def _namespaced_rule_id_from_spec(spec):
     return _namespaced_rule_id(spec, spec['rule_id'])
 
 
 def _namespaced_rule_id(spec, rule_id):
-    if spec['scope'] == 'global':
-        scope = 'global'
-    else:
-        scope = 'device/%s' % (spec['profile_tag'])
-    return "%s/%s/%s" % (scope, spec['template'], rule_id)
+    return "global/%s/%s" % (spec['template'], rule_id)
 
 
 def _rule_id_from_namespaced(in_rule_id):
diff --git a/synapse/rest/client/v1/pusher.py b/synapse/rest/client/v1/pusher.py
index 5547f1b112..4c662e6e3c 100644
--- a/synapse/rest/client/v1/pusher.py
+++ b/synapse/rest/client/v1/pusher.py
@@ -45,7 +45,7 @@ class PusherRestServlet(ClientV1RestServlet):
             )
             defer.returnValue((200, {}))
 
-        reqd = ['profile_tag', 'kind', 'app_id', 'app_display_name',
+        reqd = ['kind', 'app_id', 'app_display_name',
                 'device_display_name', 'pushkey', 'lang', 'data']
         missing = []
         for i in reqd:
@@ -73,14 +73,14 @@ class PusherRestServlet(ClientV1RestServlet):
             yield pusher_pool.add_pusher(
                 user_id=user.to_string(),
                 access_token=requester.access_token_id,
-                profile_tag=content['profile_tag'],
                 kind=content['kind'],
                 app_id=content['app_id'],
                 app_display_name=content['app_display_name'],
                 device_display_name=content['device_display_name'],
                 pushkey=content['pushkey'],
                 lang=content['lang'],
-                data=content['data']
+                data=content['data'],
+                profile_tag=content.get('profile_tag', ""),
             )
         except PusherConfigException as pce:
             raise SynapseError(400, "Config Error: " + pce.message,
diff --git a/synapse/storage/event_push_actions.py b/synapse/storage/event_push_actions.py
index d77a817682..5820539a92 100644
--- a/synapse/storage/event_push_actions.py
+++ b/synapse/storage/event_push_actions.py
@@ -27,15 +27,14 @@ class EventPushActionsStore(SQLBaseStore):
     def _set_push_actions_for_event_and_users_txn(self, txn, event, tuples):
         """
         :param event: the event set actions for
-        :param tuples: list of tuples of (user_id, profile_tag, actions)
+        :param tuples: list of tuples of (user_id, actions)
         """
         values = []
-        for uid, profile_tag, actions in tuples:
+        for uid, actions in tuples:
             values.append({
                 'room_id': event.room_id,
                 'event_id': event.event_id,
                 'user_id': uid,
-                'profile_tag': profile_tag,
                 'actions': json.dumps(actions),
                 'stream_ordering': event.internal_metadata.stream_ordering,
                 'topological_ordering': event.depth,
@@ -43,7 +42,7 @@ class EventPushActionsStore(SQLBaseStore):
                 'highlight': 1 if _action_has_highlight(actions) else 0,
             })
 
-        for uid, _, __ in tuples:
+        for uid, __ in tuples:
             txn.call_after(
                 self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
                 (event.room_id, uid)
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
index 8ec706178a..c23648cdbc 100644
--- a/synapse/storage/pusher.py
+++ b/synapse/storage/pusher.py
@@ -80,9 +80,9 @@ class PusherStore(SQLBaseStore):
         defer.returnValue(rows)
 
     @defer.inlineCallbacks
-    def add_pusher(self, user_id, access_token, profile_tag, kind, app_id,
+    def add_pusher(self, user_id, access_token, kind, app_id,
                    app_display_name, device_display_name,
-                   pushkey, pushkey_ts, lang, data):
+                   pushkey, pushkey_ts, lang, data, profile_tag=""):
         try:
             next_id = yield self._pushers_id_gen.get_next()
             yield self._simple_upsert(
@@ -95,12 +95,12 @@ class PusherStore(SQLBaseStore):
                 dict(
                     access_token=access_token,
                     kind=kind,
-                    profile_tag=profile_tag,
                     app_display_name=app_display_name,
                     device_display_name=device_display_name,
                     ts=pushkey_ts,
                     lang=lang,
                     data=encode_canonical_json(data),
+                    profile_tag=profile_tag,
                 ),
                 insertion_values=dict(
                     id=next_id,
-- 
cgit 1.4.1


From e12ec335a58bb7957cb7abfc1c96500bb4fb2627 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 18 Feb 2016 17:01:53 +0000
Subject: "You are not..."

---
 synapse/rest/client/v1/presence.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'synapse/rest/client/v1')

diff --git a/synapse/rest/client/v1/presence.py b/synapse/rest/client/v1/presence.py
index 27ea5f2a43..bbfa1d6ac4 100644
--- a/synapse/rest/client/v1/presence.py
+++ b/synapse/rest/client/v1/presence.py
@@ -41,7 +41,7 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
             )
 
             if not allowed:
-                raise AuthError(403, "You are allowed to see their presence.")
+                raise AuthError(403, "You are not allowed to see their presence.")
 
         state = yield self.handlers.presence_handler.get_state(target_user=user)
 
-- 
cgit 1.4.1