Merge branch 'develop' into matthew/preview_urls

1 files changed, 52 insertions, 124 deletions

diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py
index 4114a7e430..de4a020ad4 100644
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -20,15 +20,16 @@ from synapse.http.servlet import (
 )
 from synapse.handlers.sync import SyncConfig
 from synapse.types import StreamToken
-from synapse.events import FrozenEvent
 from synapse.events.utils import (
     serialize_event, format_event_for_client_v2_without_room_id,
 )
-from synapse.api.filtering import FilterCollection
+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
+import itertools
 import logging
 
 import ujson as json
@@ -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):
@@ -113,24 +115,24 @@ class SyncRestServlet(RestServlet):
             )
         )
 
-        if filter_id and filter_id.startswith('{'):
-            try:
-                filter_object = json.loads(filter_id)
-            except:
-                raise SynapseError(400, "Invalid filter JSON")
-            self.filtering._check_valid_filter(filter_object)
-            filter = FilterCollection(filter_object)
-        else:
-            try:
+        if filter_id:
+            if filter_id.startswith('{'):
+                try:
+                    filter_object = json.loads(filter_id)
+                except:
+                    raise SynapseError(400, "Invalid filter JSON")
+                self.filtering.check_valid_filter(filter_object)
+                filter = FilterCollection(filter_object)
+            else:
                 filter = yield self.filtering.get_user_filter(
                     user.localpart, filter_id
                 )
-            except:
-                filter = FilterCollection({})
+        else:
+            filter = DEFAULT_FILTER_COLLECTION
 
         sync_config = SyncConfig(
             user=user,
-            filter=filter,
+            filter_collection=filter,
             is_guest=requester.is_guest,
         )
 
@@ -139,38 +141,38 @@ 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()
 
         joined = self.encode_joined(
-            sync_result.joined, filter, time_now, requester.access_token_id
+            sync_result.joined, time_now, requester.access_token_id
         )
 
         invited = self.encode_invited(
-            sync_result.invited, filter, time_now, requester.access_token_id
+            sync_result.invited, time_now, requester.access_token_id
         )
 
         archived = self.encode_archived(
-            sync_result.archived, filter, time_now, requester.access_token_id
+            sync_result.archived, time_now, requester.access_token_id
         )
 
         response_content = {
-            "account_data": self.encode_account_data(
-                sync_result.account_data, filter, time_now
-            ),
+            "account_data": {"events": sync_result.account_data},
             "presence": self.encode_presence(
-                sync_result.presence, filter, time_now
+                sync_result.presence, time_now
             ),
             "rooms": {
                 "join": joined,
@@ -182,24 +184,20 @@ class SyncRestServlet(RestServlet):
 
         defer.returnValue((200, response_content))
 
-    def encode_presence(self, events, filter, time_now):
+    def encode_presence(self, events, time_now):
         formatted = []
         for event in events:
             event = copy.deepcopy(event)
             event['sender'] = event['content'].pop('user_id')
             formatted.append(event)
-        return {"events": filter.filter_presence(formatted)}
-
-    def encode_account_data(self, events, filter, time_now):
-        return {"events": filter.filter_account_data(events)}
+        return {"events": formatted}
 
-    def encode_joined(self, rooms, filter, time_now, token_id):
+    def encode_joined(self, rooms, time_now, token_id):
         """
         Encode the joined rooms in a sync result
 
         :param list[synapse.handlers.sync.JoinedSyncResult] rooms: list of sync
             results for rooms this user is joined to
-        :param FilterCollection filter: filters to apply to the results
         :param int time_now: current time - used as a baseline for age
             calculations
         :param int token_id: ID of the user's auth token - used for namespacing
@@ -211,18 +209,17 @@ class SyncRestServlet(RestServlet):
         joined = {}
         for room in rooms:
             joined[room.room_id] = self.encode_room(
-                room, filter, time_now, token_id
+                room, time_now, token_id
             )
 
         return joined
 
-    def encode_invited(self, rooms, filter, time_now, token_id):
+    def encode_invited(self, rooms, time_now, token_id):
         """
         Encode the invited rooms in a sync result
 
         :param list[synapse.handlers.sync.InvitedSyncResult] rooms: list of
              sync results for rooms this user is joined to
-        :param FilterCollection filter: filters to apply to the results
         :param int time_now: current time - used as a baseline for age
             calculations
         :param int token_id: ID of the user's auth token - used for namespacing
@@ -237,7 +234,9 @@ class SyncRestServlet(RestServlet):
                 room.invite, time_now, token_id=token_id,
                 event_format=format_event_for_client_v2_without_room_id,
             )
-            invited_state = invite.get("unsigned", {}).pop("invite_room_state", [])
+            unsigned = dict(invite.get("unsigned", {}))
+            invite["unsigned"] = unsigned
+            invited_state = list(unsigned.pop("invite_room_state", []))
             invited_state.append(invite)
             invited[room.room_id] = {
                 "invite_state": {"events": invited_state}
@@ -245,13 +244,12 @@ class SyncRestServlet(RestServlet):
 
         return invited
 
-    def encode_archived(self, rooms, filter, time_now, token_id):
+    def encode_archived(self, rooms, time_now, token_id):
         """
         Encode the archived rooms in a sync result
 
         :param list[synapse.handlers.sync.ArchivedSyncResult] rooms: list of
              sync results for rooms this user is joined to
-        :param FilterCollection filter: filters to apply to the results
         :param int time_now: current time - used as a baseline for age
             calculations
         :param int token_id: ID of the user's auth token - used for namespacing
@@ -263,17 +261,16 @@ class SyncRestServlet(RestServlet):
         joined = {}
         for room in rooms:
             joined[room.room_id] = self.encode_room(
-                room, filter, time_now, token_id, joined=False
+                room, time_now, token_id, joined=False
             )
 
         return joined
 
     @staticmethod
-    def encode_room(room, filter, time_now, token_id, joined=True):
+    def encode_room(room, time_now, token_id, joined=True):
         """
         :param JoinedSyncResult|ArchivedSyncResult room: sync result for a
             single room
-        :param FilterCollection filter: filters to apply to the results
         :param int time_now: current time - used as a baseline for age
             calculations
         :param int token_id: ID of the user's auth token - used for namespacing
@@ -292,19 +289,23 @@ class SyncRestServlet(RestServlet):
             )
 
         state_dict = room.state
-        timeline_events = filter.filter_room_timeline(room.timeline.events)
+        timeline_events = room.timeline.events
 
-        state_dict = SyncRestServlet._rollback_state_for_timeline(
-            state_dict, timeline_events)
+        state_events = state_dict.values()
 
-        state_events = filter.filter_room_state(state_dict.values())
+        for event in itertools.chain(state_events, timeline_events):
+            # We've had bug reports that events were coming down under the
+            # wrong room.
+            if event.room_id != room.room_id:
+                logger.warn(
+                    "Event %r is under room %r instead of %r",
+                    event.event_id, room.room_id, event.room_id,
+                )
 
         serialized_state = [serialize(e) for e in state_events]
         serialized_timeline = [serialize(e) for e in timeline_events]
 
-        account_data = filter.filter_room_account_data(
-            room.account_data
-        )
+        account_data = room.account_data
 
         result = {
             "timeline": {
@@ -317,85 +318,12 @@ class SyncRestServlet(RestServlet):
         }
 
         if joined:
-            ephemeral_events = filter.filter_room_ephemeral(room.ephemeral)
+            ephemeral_events = room.ephemeral
             result["ephemeral"] = {"events": ephemeral_events}
             result["unread_notifications"] = room.unread_notifications
 
         return result
 
-    @staticmethod
-    def _rollback_state_for_timeline(state, timeline):
-        """
-        Wind the state dictionary backwards, so that it represents the
-        state at the start of the timeline, rather than at the end.
-
-        :param dict[(str, str), synapse.events.EventBase] state: the
-            state dictionary. Will be updated to the state before the timeline.
-        :param list[synapse.events.EventBase] timeline: the event timeline
-        :return: updated state dictionary
-        """
-        logger.debug("Processing state dict %r; timeline %r", state,
-                     [e.get_dict() for e in timeline])
-
-        result = state.copy()
-
-        for timeline_event in reversed(timeline):
-            if not timeline_event.is_state():
-                continue
-
-            event_key = (timeline_event.type, timeline_event.state_key)
-
-            logger.debug("Considering %s for removal", event_key)
-
-            state_event = result.get(event_key)
-            if (state_event is None or
-                    state_event.event_id != timeline_event.event_id):
-                # the event in the timeline isn't present in the state
-                # dictionary.
-                #
-                # the most likely cause for this is that there was a fork in
-                # the event graph, and the state is no longer valid. Really,
-                # the event shouldn't be in the timeline. We're going to ignore
-                # it for now, however.
-                logger.warn("Found state event %r in timeline which doesn't "
-                            "match state dictionary", timeline_event)
-                continue
-
-            prev_event_id = timeline_event.unsigned.get("replaces_state", None)
-
-            prev_content = timeline_event.unsigned.get('prev_content')
-            prev_sender = timeline_event.unsigned.get('prev_sender')
-            # Empircally it seems possible for the event to have a
-            # "replaces_state" key but not a prev_content or prev_sender
-            # markjh conjectures that it could be due to the server not
-            # having a copy of that event.
-            # If this is the case the we ignore the previous event. This will
-            # cause the displayname calculations on the client to be incorrect
-            if prev_event_id is None or not prev_content or not prev_sender:
-                logger.debug(
-                    "Removing %r from the state dict, as it is missing"
-                    " prev_content (prev_event_id=%r)",
-                    timeline_event.event_id, prev_event_id
-                )
-                del result[event_key]
-            else:
-                logger.debug(
-                    "Replacing %r with %r in state dict",
-                    timeline_event.event_id, prev_event_id
-                )
-                result[event_key] = FrozenEvent({
-                    "type": timeline_event.type,
-                    "state_key": timeline_event.state_key,
-                    "content": prev_content,
-                    "sender": prev_sender,
-                    "event_id": prev_event_id,
-                    "room_id": timeline_event.room_id,
-                })
-
-            logger.debug("New value: %r", result.get(event_key))
-
-        return result
-
 
 def register_servlets(hs, http_server):
     SyncRestServlet(hs).register(http_server)