Start implementing incremental initial sync

3 files changed, 241 insertions, 34 deletions

diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index b7f1ad4b40..42fb0371e5 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -137,6 +137,7 @@ def serialize_event(e, time_now_ms, client_event=True, strip_ids=False):
     d.pop("depth", None)
     d.pop("unsigned", None)
     d.pop("origin", None)
+    d.pop("prev_state", None)
 
     if strip_ids:
         d.pop("room_id", None)
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index bbabaf3df1..f8629a588f 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -37,14 +37,18 @@ SyncConfig = collections.namedtuple("SyncConfig", [
 ])
 
 
-RoomSyncResult = collections.namedtuple("RoomSyncResult", [
+class RoomSyncResult(collections.namedtuple("RoomSyncResult", [
     "room_id",
     "limited",
     "published",
-    "events", # dict of event
+    "events",
     "state",
     "prev_batch",
-])
+])):
+    __slots__ = []
+
+    def __nonzero__(self):
+        return bool(self.events or self.state)
 
 
 class SyncResult(collections.namedtuple("SyncResult", [
@@ -56,7 +60,9 @@ class SyncResult(collections.namedtuple("SyncResult", [
     __slots__ = []
 
     def __nonzero__(self):
-        return self.private_user_data or self.public_user_data or self.rooms
+        return bool(
+            self.private_user_data or self.public_user_data or self.rooms
+        )
 
 
 class SyncHandler(BaseHandler):
@@ -67,7 +73,13 @@ class SyncHandler(BaseHandler):
         self.clock = hs.get_clock()
 
     def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0):
-        if timeout == 0:
+        """Get the sync for a client if we have new data for it now. Otherwise
+        wait for new data to arrive on the server. If the timeout expires, then
+        return an empty sync result.
+        Returns:
+            A Deferred SyncResult.
+        """
+        if timeout == 0 or since_token is None:
             return self.current_sync_for_user(sync_config, since_token)
         else:
             def current_sync_callback(since_token):
@@ -79,13 +91,25 @@ class SyncHandler(BaseHandler):
             )
 
     def current_sync_for_user(self, sync_config, since_token=None):
+        """Get the sync for client needed to match what the server has now.
+        Returns:
+            A Deferred SyncResult.
+        """
         if since_token is None:
             return self.initial_sync(sync_config)
         else:
-            return self.incremental_sync(sync_config)
+            if sync_config.gap:
+                return self.incremental_sync_with_gap(sync_config, since_token)
+            else:
+                #TODO(mjark): Handle gapless sync
+                pass
 
     @defer.inlineCallbacks
     def initial_sync(self, sync_config):
+        """Get a sync for a client which is starting without any state
+        Returns:
+            A Deferred SyncResult.
+        """
         if sync_config.sort == "timeline,desc":
             # TODO(mjark): Handle going through events in reverse order?.
             # What does "most recent events" mean when applying the limits mean
@@ -114,25 +138,86 @@ class SyncHandler(BaseHandler):
 
         rooms = []
         for event in room_list:
-            #TODO (mjark): Apply the event filter in sync_config.
-            recent_events, token = yield self.store.get_recent_events_for_room(
-                event.room_id,
-                limit=sync_config.limit,
-                end_token=now_token.room_key,
-            )
-            prev_batch_token = now_token.copy_and_replace("room_key", token[0])
-            current_state_events = yield self.state_handler.get_current_state(
-                event.room_id
+            room_sync = yield self.initial_sync_for_room(
+                event.room_id, sync_config, now_token, published_room_ids
             )
+            rooms.append(room_sync)
+
+        defer.returnValue(SyncResult(
+            public_user_data=presence,
+            private_user_data=[],
+            rooms=rooms,
+            next_batch=now_token,
+        ))
+
+    @defer.inlineCallbacks
+    def intial_sync_for_room(self, room_id, sync_config, now_token,
+                             published_room_ids):
+        """Sync a room for a client which is starting without any state
+        Returns:
+            A Deferred RoomSyncResult.
+        """
+        recent_events, token = yield self.store.get_recent_events_for_room(
+            room_id,
+            limit=sync_config.limit,
+            end_token=now_token.room_key,
+        )
+        prev_batch_token = now_token.copy_and_replace("room_key", token[0])
+        current_state_events = yield self.state_handler.get_current_state(
+            room_id
+        )
+
+        defer.returnValue(RoomSyncResult(
+            room_id=room_id,
+            published=room_id in published_room_ids,
+            events=recent_events,
+            prev_batch=prev_batch_token,
+            state=current_state_events,
+            limited=True,
+        ))
+
+
+    @defer.inlineCallbacks
+    def incremental_sync_with_gap(self, sync_config, since_token):
+        """ Get the incremental delta needed to bring the client up to
+        date with the server.
+        Returns:
+            A Deferred SyncResult.
+        """
+        if sync_config.sort == "timeline,desc":
+            # TODO(mjark): Handle going through events in reverse order?.
+            # What does "most recent events" mean when applying the limits mean
+            # in this case?
+            raise NotImplementedError()
+
+        now_token = yield self.event_sources.get_current_token()
+
+        presence_stream = self.event_sources.sources["presence"]
+        pagination_config = PaginationConfig(
+            from_token=since_token, to_token=now_token
+        )
+        presence, _ = yield presence_stream.get_pagination_rows(
+            user=sync_config.user,
+            pagination_config=pagination_config.get_source_config("presence"),
+            key=None
+        )
+        room_list = yield self.store.get_rooms_for_user_where_membership_is(
+            user_id=sync_config.user.to_string(),
+            membership_list=[Membership.INVITE, Membership.JOIN]
+        )
+
+        # TODO (mjark): Does public mean "published"?
+        published_rooms = yield self.store.get_rooms(is_public=True)
+        published_room_ids = set(r["room_id"] for r in published_rooms)
 
-            rooms.append(RoomSyncResult(
-                room_id=event.room_id,
-                published=event.room_id in published_room_ids,
-                events=recent_events,
-                prev_batch=prev_batch_token,
-                state=current_state_events,
-                limited=True,
-            ))
+        rooms = []
+        for event in room_list:
+            room_sync = yield self.incremental_sync_with_gap_for_room(
+                event.room_id, sync_config, since_token, now_token,
+                published_room_ids
+            )
+            if room_sync:
+                rooms.append(room_sync)
 
         defer.returnValue(SyncResult(
             public_user_data=presence,
@@ -143,5 +228,103 @@ class SyncHandler(BaseHandler):
 
 
     @defer.inlineCallbacks
-    def incremental_sync(self, sync_config):
-        pass
+    def incremental_sync_with_gap_for_room(self, room_id, sync_config,
+                                           since_token, now_token,
+                                           published_room_ids):
+        """ Get the incremental delta needed to bring the client up to date for
+        the room. Gives the client the most recent events and the changes to
+        state.
+        Returns:
+            A Deferred RoomSyncResult
+        """
+        # TODO(mjark): Check if they have joined the room between
+        # the previous sync and this one.
+        # TODO(mjark): Apply the event filter in sync_config
+        # TODO(mjark): Check for redactions we might have missed.
+        # TODO(mjark): Typing notifications.
+        recents, token = yield self.store.get_recent_events_for_room(
+            room_id,
+            limit=sync_config.limit + 1,
+            from_token=since_token.room_key,
+            end_token=now_token.room_key,
+        )
+
+        logging.debug("Recents %r", recents)
+
+        if len(recents) > sync_config.limit:
+            limited = True
+            recents = recents[1:]
+        else:
+            limited = False
+
+        prev_batch_token = now_token.copy_and_replace("room_key", token[0])
+
+        # TODO(mjark): This seems racy since this isn't being passed a
+        # token to indicate what point in the stream this is
+        current_state_events = yield self.state_handler.get_current_state(
+            room_id
+        )
+
+        state_at_previous_sync = yield self.get_state_at_previous_sync(
+            room_id, since_token=since_token
+        )
+
+        state_events_delta = yield self.compute_state_delta(
+            since_token=since_token,
+            previous_state=state_at_previous_sync,
+            current_state=current_state_events,
+        )
+
+        room_sync = RoomSyncResult(
+            room_id=room_id,
+            published=room_id in published_room_ids,
+            events=recents,
+            prev_batch=prev_batch_token,
+            state=state_events_delta,
+            limited=limited,
+        )
+
+        logging.debug("Room sync: %r", room_sync)
+
+        defer.returnValue(room_sync)
+
+    @defer.inlineCallbacks
+    def get_state_at_previous_sync(self, room_id, since_token):
+        """ Get the room state at the previous sync the client made.
+        Returns:
+            A Deferred list of Events.
+        """
+        last_events, token = yield self.store.get_recent_events_for_room(
+            room_id, end_token=since_token.room_key, limit=1,
+        )
+
+        if last_events:
+            last_event = last_events[0]
+            last_context = yield self.state_handler.compute_event_context(
+                last_event
+            )
+            if last_event.is_state():
+                state = [last_event] + last_context.current_state.values()
+            else:
+                state = last_context.current_state.values()
+        else:
+            state = ()
+        defer.returnValue(state)
+
+
+    def compute_state_delta(self, since_token, previous_state, current_state):
+        """ Works out the differnce in state between the current state and the
+        state the client got when it last performed a sync.
+        Returns:
+            A list of events.
+        """
+        # TODO(mjark) Check if the state events were received by the server
+        # after the previous sync, since we need to include those state
+        # updates even if they occured logically before the previous event.
+        # TODO(mjark) Check for new redactions in the state events.
+        previous_dict = {event.event_id:event for event in previous_state}
+        state_delta = []
+        for event in current_state:
+            if event.event_id not in previous_dict:
+                state_delta.append(event)
+        return state_delta
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 8ac2adab05..06aca1a4e5 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -265,17 +265,38 @@ class StreamStore(SQLBaseStore):
         return self.runInteraction("paginate_room_events", f)
 
     def get_recent_events_for_room(self, room_id, limit, end_token,
-                                   with_feedback=False):
+                                   with_feedback=False, from_token=None):
         # TODO (erikj): Handle compressed feedback
 
-        sql = (
-            "SELECT stream_ordering, topological_ordering, event_id FROM events "
-            "WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 "
-            "ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
-        )
+        end_token = _StreamToken.parse_stream_token(end_token)
 
-        def f(txn):
-            txn.execute(sql, (room_id, end_token, limit,))
+        if from_token is None:
+            sql = (
+                "SELECT stream_ordering, topological_ordering, event_id"
+                " FROM events"
+                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC"
+                " LIMIT ?"
+            )
+        else:
+            from_token = _StreamToken.parse_stream_token(from_token)
+            sql = (
+                "SELECT stream_ordering, topological_ordering, event_id"
+                " FROM events"
+                " WHERE room_id = ? AND stream_ordering > ?"
+                " AND stream_ordering <= ? AND outlier = 0"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC"
+                " LIMIT ?"
+            )
+
+
+        def get_recent_events_for_room_txn(txn):
+            if from_token is None:
+                txn.execute(sql, (room_id, end_token.stream, limit,))
+            else:
+                txn.execute(sql, (
+                    room_id, from_token.stream, end_token.stream, limit
+                ))
 
             rows = self.cursor_to_dict(txn)
 
@@ -303,7 +324,9 @@ class StreamStore(SQLBaseStore):
 
             return events, token
 
-        return self.runInteraction("get_recent_events_for_room", f)
+        return self.runInteraction(
+            "get_recent_events_for_room", get_recent_events_for_room_txn
+        )
 
     def get_room_events_max_id(self):
         return self.runInteraction(