Merge branch 'develop' into release-v0.11.0

5 files changed, 172 insertions, 118 deletions

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index f4ade1f594..6519f183df 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -29,6 +29,12 @@ logger = logging.getLogger(__name__)
 
 
 class BaseHandler(object):
+    """
+    Common base class for the event handlers.
+
+    :type store: synapse.storage.events.StateStore
+    :type state_handler: synapse.state.StateHandler
+    """
 
     def __init__(self, hs):
         self.store = hs.get_datastore()
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index d1589334a5..c1bce07e31 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -564,7 +564,7 @@ class FederationHandler(BaseHandler):
 
     @log_function
     @defer.inlineCallbacks
-    def do_invite_join(self, target_hosts, room_id, joinee):
+    def do_invite_join(self, target_hosts, room_id, joinee, content):
         """ Attempts to join the `joinee` to the room `room_id` via the
         server `target_host`.
 
@@ -584,7 +584,8 @@ class FederationHandler(BaseHandler):
             target_hosts,
             room_id,
             joinee,
-            "join"
+            "join",
+            content,
         )
 
         self.room_queues[room_id] = []
@@ -840,12 +841,14 @@ class FederationHandler(BaseHandler):
         defer.returnValue(None)
 
     @defer.inlineCallbacks
-    def _make_and_verify_event(self, target_hosts, room_id, user_id, membership):
+    def _make_and_verify_event(self, target_hosts, room_id, user_id, membership,
+                               content={},):
         origin, pdu = yield self.replication_layer.make_membership_event(
             target_hosts,
             room_id,
             user_id,
-            membership
+            membership,
+            content,
         )
 
         logger.debug("Got response to make_%s: %s", membership, pdu)
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 7d31ff8d46..14051aee99 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -258,20 +258,30 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def _check_in_room_or_world_readable(self, room_id, user_id, is_guest):
-        if is_guest:
+        try:
+            # check_user_was_in_room will return the most recent membership
+            # event for the user if:
+            #  * The user is a non-guest user, and was ever in the room
+            #  * The user is a guest user, and has joined the room
+            # else it will throw.
+            member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+            defer.returnValue((member_event.membership, member_event.event_id))
+            return
+        except AuthError, auth_error:
             visibility = yield self.state_handler.get_current_state(
                 room_id, EventTypes.RoomHistoryVisibility, ""
             )
-            if visibility.content["history_visibility"] == "world_readable":
+            if (
+                visibility and
+                visibility.content["history_visibility"] == "world_readable"
+            ):
                 defer.returnValue((Membership.JOIN, None))
                 return
-            else:
-                raise AuthError(
-                    403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN
-                )
-        else:
-            member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
-            defer.returnValue((member_event.membership, member_event.event_id))
+            if not is_guest:
+                raise auth_error
+            raise AuthError(
+                403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN
+            )
 
     @defer.inlineCallbacks
     def get_state_events(self, user_id, room_id, is_guest=False):
@@ -456,7 +466,7 @@ class MessageHandler(BaseHandler):
         defer.returnValue(ret)
 
     @defer.inlineCallbacks
-    def room_initial_sync(self, user_id, room_id, pagin_config=None):
+    def room_initial_sync(self, user_id, room_id, pagin_config=None, is_guest=False):
         """Capture the a snapshot of a room. If user is currently a member of
         the room this will be what is currently in the room. If the user left
         the room this will be what was in the room when they left.
@@ -473,15 +483,19 @@ class MessageHandler(BaseHandler):
             A JSON serialisable dict with the snapshot of the room.
         """
 
-        member_event = yield self.auth.check_user_was_in_room(room_id, user_id)
+        membership, member_event_id = yield self._check_in_room_or_world_readable(
+            room_id,
+            user_id,
+            is_guest
+        )
 
-        if member_event.membership == Membership.JOIN:
+        if membership == Membership.JOIN:
             result = yield self._room_initial_sync_joined(
-                user_id, room_id, pagin_config, member_event
+                user_id, room_id, pagin_config, membership, is_guest
             )
-        elif member_event.membership == Membership.LEAVE:
+        elif membership == Membership.LEAVE:
             result = yield self._room_initial_sync_parted(
-                user_id, room_id, pagin_config, member_event
+                user_id, room_id, pagin_config, membership, member_event_id, is_guest
             )
 
         private_user_data = []
@@ -497,19 +511,19 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def _room_initial_sync_parted(self, user_id, room_id, pagin_config,
-                                  member_event):
+                                  membership, member_event_id, is_guest):
         room_state = yield self.store.get_state_for_events(
-            [member_event.event_id], None
+            [member_event_id], None
         )
 
-        room_state = room_state[member_event.event_id]
+        room_state = room_state[member_event_id]
 
         limit = pagin_config.limit if pagin_config else None
         if limit is None:
             limit = 10
 
         stream_token = yield self.store.get_stream_token_for_event(
-            member_event.event_id
+            member_event_id
         )
 
         messages, token = yield self.store.get_recent_events_for_room(
@@ -519,7 +533,7 @@ class MessageHandler(BaseHandler):
         )
 
         messages = yield self._filter_events_for_client(
-            user_id, messages
+            user_id, messages, is_guest=is_guest
         )
 
         start_token = StreamToken(token[0], 0, 0, 0, 0)
@@ -528,7 +542,7 @@ class MessageHandler(BaseHandler):
         time_now = self.clock.time_msec()
 
         defer.returnValue({
-            "membership": member_event.membership,
+            "membership": membership,
             "room_id": room_id,
             "messages": {
                 "chunk": [serialize_event(m, time_now) for m in messages],
@@ -542,7 +556,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def _room_initial_sync_joined(self, user_id, room_id, pagin_config,
-                                  member_event):
+                                  membership, is_guest):
         current_state = yield self.state.get_current_state(
             room_id=room_id,
         )
@@ -574,12 +588,14 @@ 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,
-                as_event=True,
-                check_auth=False,
-            )
+            states = {}
+            if not is_guest:
+                states = yield presence_handler.get_states(
+                    target_users=[UserID.from_string(m.user_id) for m in room_members],
+                    auth_user=auth_user,
+                    as_event=True,
+                    check_auth=False,
+                )
 
             defer.returnValue(states.values())
 
@@ -599,7 +615,7 @@ class MessageHandler(BaseHandler):
         ).addErrback(unwrapFirstError)
 
         messages = yield self._filter_events_for_client(
-            user_id, messages
+            user_id, messages, is_guest=is_guest, require_all_visible_for_guests=False
         )
 
         start_token = now_token.copy_and_replace("room_key", token[0])
@@ -607,8 +623,7 @@ class MessageHandler(BaseHandler):
 
         time_now = self.clock.time_msec()
 
-        defer.returnValue({
-            "membership": member_event.membership,
+        ret = {
             "room_id": room_id,
             "messages": {
                 "chunk": [serialize_event(m, time_now) for m in messages],
@@ -618,4 +633,8 @@ class MessageHandler(BaseHandler):
             "state": state,
             "presence": presence,
             "receipts": receipts,
-        })
+        }
+        if not is_guest:
+            ret["membership"] = membership
+
+        defer.returnValue(ret)
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 0266926fc7..3f04752581 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -504,7 +504,8 @@ class RoomMemberHandler(BaseHandler):
             yield handler.do_invite_join(
                 room_hosts,
                 room_id,
-                event.user_id
+                event.user_id,
+                event.content,
             )
         else:
             logger.debug("Doing normal join")
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 492c1c17d5..6dc9d0fb92 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -47,9 +47,9 @@ class TimelineBatch(collections.namedtuple("TimelineBatch", [
 
 
 class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [
-    "room_id",
-    "timeline",
-    "state",
+    "room_id",           # str
+    "timeline",          # TimelineBatch
+    "state",             # dict[(str, str), FrozenEvent]
     "ephemeral",
     "private_user_data",
 ])):
@@ -68,9 +68,9 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [
 
 
 class ArchivedSyncResult(collections.namedtuple("JoinedSyncResult", [
-    "room_id",
-    "timeline",
-    "state",
+    "room_id",            # str
+    "timeline",           # TimelineBatch
+    "state",              # dict[(str, str), FrozenEvent]
     "private_user_data",
 ])):
     __slots__ = []
@@ -87,8 +87,8 @@ class ArchivedSyncResult(collections.namedtuple("JoinedSyncResult", [
 
 
 class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
-    "room_id",
-    "invite",
+    "room_id",   # str
+    "invite",    # FrozenEvent: the invite event
 ])):
     __slots__ = []
 
@@ -254,15 +254,12 @@ class SyncHandler(BaseHandler):
             room_id, sync_config, now_token, since_token=timeline_since_token
         )
 
-        current_state = yield self.state_handler.get_current_state(
-            room_id
-        )
-        current_state_events = current_state.values()
+        current_state = yield self.get_state_at(room_id, now_token)
 
         defer.returnValue(JoinedSyncResult(
             room_id=room_id,
             timeline=batch,
-            state=current_state_events,
+            state=current_state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             private_user_data=self.private_user_data_for_room(
                 room_id, tags_by_room
@@ -354,14 +351,12 @@ class SyncHandler(BaseHandler):
             room_id, sync_config, leave_token, since_token=timeline_since_token
         )
 
-        leave_state = yield self.store.get_state_for_events(
-            [leave_event_id], None
-        )
+        leave_state = yield self.store.get_state_for_event(leave_event_id)
 
         defer.returnValue(ArchivedSyncResult(
             room_id=room_id,
             timeline=batch,
-            state=leave_state[leave_event_id].values(),
+            state=leave_state,
             private_user_data=self.private_user_data_for_room(
                 room_id, tags_by_room
             ),
@@ -425,6 +420,9 @@ class SyncHandler(BaseHandler):
         if len(room_events) <= timeline_limit:
             # There is no gap in any of the rooms. Therefore we can just
             # partition the new events by room and return them.
+            logger.debug("Got %i events for incremental sync - not limited",
+                         len(room_events))
+
             invite_events = []
             leave_events = []
             events_by_room_id = {}
@@ -440,7 +438,12 @@ class SyncHandler(BaseHandler):
 
             for room_id in joined_room_ids:
                 recents = events_by_room_id.get(room_id, [])
-                state = [event for event in recents if event.is_state()]
+                logger.debug("Events for room %s: %r", room_id, recents)
+                state = {
+                    (event.type, event.state_key): event
+                    for event in recents if event.is_state()}
+                limited = False
+
                 if recents:
                     prev_batch = now_token.copy_and_replace(
                         "room_key", recents[0].internal_metadata.before
@@ -448,9 +451,13 @@ class SyncHandler(BaseHandler):
                 else:
                     prev_batch = now_token
 
-                state, limited = yield self.check_joined_room(
-                    sync_config, room_id, state
-                )
+                just_joined = yield self.check_joined_room(sync_config, state)
+                if just_joined:
+                    logger.debug("User has just joined %s: needs full state",
+                                 room_id)
+                    state = yield self.get_state_at(room_id, now_token)
+                    # the timeline is inherently limited if we've just joined
+                    limited = True
 
                 room_sync = JoinedSyncResult(
                     room_id=room_id,
@@ -465,10 +472,15 @@ class SyncHandler(BaseHandler):
                         room_id, tags_by_room
                     ),
                 )
+                logger.debug("Result for room %s: %r", room_id, room_sync)
+
                 if room_sync:
                     joined.append(room_sync)
 
         else:
+            logger.debug("Got %i events for incremental sync - hit limit",
+                         len(room_events))
+
             invite_events = yield self.store.get_invites_for_user(
                 sync_config.user.to_string()
             )
@@ -507,6 +519,9 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def load_filtered_recents(self, room_id, sync_config, now_token,
                               since_token=None):
+        """
+        :returns a Deferred TimelineBatch
+        """
         limited = True
         recents = []
         filtering_factor = 2
@@ -558,6 +573,8 @@ class SyncHandler(BaseHandler):
         Returns:
             A Deferred JoinedSyncResult
         """
+        logger.debug("Doing incremental sync for room %s between %s and %s",
+                     room_id, since_token, now_token)
 
         # TODO(mjark): Check for redactions we might have missed.
 
@@ -567,31 +584,26 @@ class SyncHandler(BaseHandler):
 
         logging.debug("Recents %r", batch)
 
-        # TODO(mjark): This seems racy since this isn't being passed a
-        # token to indicate what point in the stream this is
-        current_state = yield self.state_handler.get_current_state(
-            room_id
-        )
-        current_state_events = current_state.values()
+        current_state = yield self.get_state_at(room_id, now_token)
 
-        state_at_previous_sync = yield self.get_state_at_previous_sync(
-            room_id, since_token=since_token
+        state_at_previous_sync = yield self.get_state_at(
+            room_id, stream_position=since_token
         )
 
-        state_events_delta = yield self.compute_state_delta(
+        state = yield self.compute_state_delta(
             since_token=since_token,
             previous_state=state_at_previous_sync,
-            current_state=current_state_events,
+            current_state=current_state,
         )
 
-        state_events_delta, _ = yield self.check_joined_room(
-            sync_config, room_id, state_events_delta
-        )
+        just_joined = yield self.check_joined_room(sync_config, state)
+        if just_joined:
+            state = yield self.get_state_at(room_id, now_token)
 
         room_sync = JoinedSyncResult(
             room_id=room_id,
             timeline=batch,
-            state=state_events_delta,
+            state=state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             private_user_data=self.private_user_data_for_room(
                 room_id, tags_by_room
@@ -623,16 +635,12 @@ class SyncHandler(BaseHandler):
 
         logging.debug("Recents %r", batch)
 
-        # TODO(mjark): This seems racy since this isn't being passed a
-        # token to indicate what point in the stream this is
-        leave_state = yield self.store.get_state_for_events(
-            [leave_event.event_id], None
+        state_events_at_leave = yield self.store.get_state_for_event(
+            leave_event.event_id
         )
 
-        state_events_at_leave = leave_state[leave_event.event_id].values()
-
-        state_at_previous_sync = yield self.get_state_at_previous_sync(
-            leave_event.room_id, since_token=since_token
+        state_at_previous_sync = yield self.get_state_at(
+            leave_event.room_id, stream_position=since_token
         )
 
         state_events_delta = yield self.compute_state_delta(
@@ -655,60 +663,77 @@ class SyncHandler(BaseHandler):
         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.
+    def get_state_after_event(self, event):
+        """
+        Get the room state after the given event
+
+        :param synapse.events.EventBase event: event of interest
+        :return: A Deferred map from ((type, state_key)->Event)
+        """
+        state = yield self.store.get_state_for_event(event.event_id)
+        if event.is_state():
+            state = state.copy()
+            state[(event.type, event.state_key)] = event
+        defer.returnValue(state)
+
+    @defer.inlineCallbacks
+    def get_state_at(self, room_id, stream_position):
+        """ Get the room state at a particular stream position
+        :param str room_id: room for which to get state
+        :param StreamToken stream_position: point at which to get state
+        :returns: A Deferred map from ((type, state_key)->Event)
         """
         last_events, token = yield self.store.get_recent_events_for_room(
-            room_id, end_token=since_token.room_key, limit=1,
+            room_id, end_token=stream_position.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()
+            last_event = last_events[-1]
+            state = yield self.get_state_after_event(last_event)
+
         else:
-            state = ()
+            # no events in this room - so presumably no state
+            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.
+
+        :param str since_token: the point we are comparing against
+        :param dict[(str,str), synapse.events.FrozenEvent] previous_state: the
+            state to compare to
+        :param dict[(str,str), synapse.events.FrozenEvent] current_state: the
+            new state
+
+        :returns A new event dictionary
         """
         # 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)
+
+        state_delta = {}
+        for key, event in current_state.iteritems():
+            if (key not in previous_state or
+                    previous_state[key].event_id != event.event_id):
+                state_delta[key] = event
         return state_delta
 
-    @defer.inlineCallbacks
-    def check_joined_room(self, sync_config, room_id, state_delta):
-        joined = False
-        limited = False
-        for event in state_delta:
-            if (
-                event.type == EventTypes.Member
-                and event.state_key == sync_config.user.to_string()
-            ):
-                if event.content["membership"] == Membership.JOIN:
-                    joined = True
-
-        if joined:
-            res = yield self.state_handler.get_current_state(room_id)
-            state_delta = res.values()
-            limited = True
+    def check_joined_room(self, sync_config, state_delta):
+        """
+        Check if the user has just joined the given room (so should
+        be given the full state)
 
-        defer.returnValue((state_delta, limited))
+        :param sync_config:
+        :param dict[(str,str), synapse.events.FrozenEvent] state_delta: the
+           difference in state since the last sync
+
+        :returns A deferred Tuple (state_delta, limited)
+        """
+        join_event = state_delta.get((
+            EventTypes.Member, sync_config.user.to_string()), None)
+        if join_event is not None:
+            if join_event.content["membership"] == Membership.JOIN:
+                return True
+        return False