lazyload aware /messages

changelog.d/3331.feature

@@ -0,0 +1 @@
+add support for the include_redundant_members filter param as per MSC1227

changelog.d/3567.feature

@@ -0,0 +1 @@
+make the /context API filter & lazy-load aware as per MSC1227

changelog.d/3568.feature

@@ -0,0 +1 @@
+speed up /members API and add `at` and `membership` params as per MSC1227

changelog.d/3574.feature

@@ -0,0 +1 @@
+implement `summary` block in /sync response as per MSC688

changelog.d/3589.feature

@@ -0,0 +1 @@
+Add lazy-loading support to /messages as per MSC1227

synapse/handlers/message.py

@@ -82,14 +82,49 @@ def get_room_data(self, user_id=None, room_id=None,
         defer.returnValue(data)
 
     @defer.inlineCallbacks
-    def get_state_events(self, user_id, room_id, is_guest=False):
+    def _check_in_room_or_world_readable(self, room_id, user_id):
+        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:
+            visibility = yield self.state_handler.get_current_state(
+                room_id, EventTypes.RoomHistoryVisibility, ""
+            )
+            if (
+                visibility and
+                visibility.content["history_visibility"] == "world_readable"
+            ):
+                defer.returnValue((Membership.JOIN, None))
+                return
+            raise AuthError(
+                403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN
+            )
+
+    @defer.inlineCallbacks
+    def get_state_events(
+        self, user_id, room_id, types=None, filtered_types=None,
+        at_event=None, is_guest=False
+    ):
         """Retrieve all state events for a given room. If the user is
         joined to the room then return the current state. If the user has
-        left the room return the state events from when they left.
+        left the room return the state events from when they left. If an explicit
+        'at' parameter is passed, return the state events as of that event.
 
         Args:
             user_id(str): The user requesting state events.
             room_id(str): The room ID to get all state events from.
+            types(list[(Str, (Str|None))]): the (type, state_key)s to return
+                results for.
+            filtered_types(list[Str]): the list of types to apply the types filter
+                to.
+            at_event(str): the event_id we are requesting the state as of
+            is_guest(Boolean): whether this user is a guest
         Returns:
             A list of dicts representing state events. [{}, {}, {}]
         """
@@ -98,10 +133,19 @@ def get_state_events(self, user_id, room_id, is_guest=False):
         )
 
         if membership == Membership.JOIN:
-            room_state = yield self.state.get_current_state(room_id)
+            if at_event:
+                room_state = yield self.store.get_state_for_events(
+                    [at_event], types, filtered_types=filtered_types
+                )
+                room_state = room_state[at_event]
+            else:
+                state_ids = yield self.store.get_filtered_current_state_ids(
+                    room_id, types, filtered_types=filtered_types
+                )
+                room_state = yield self.store.get_events(state_ids.values())
         elif membership == Membership.LEAVE:
             room_state = yield self.store.get_state_for_events(
-                [membership_event_id], None
+                [membership_event_id], types, filtered_types=filtered_types
             )
             room_state = room_state[membership_event_id]
 

synapse/handlers/pagination.py

@@ -18,7 +18,7 @@
 from twisted.internet import defer
 from twisted.python.failure import Failure
 
-from synapse.api.constants import Membership
+from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import SynapseError
 from synapse.events.utils import serialize_event
 from synapse.types import RoomStreamToken
@@ -251,6 +251,33 @@ def get_messages(self, requester, room_id=None, pagin_config=None,
             is_peeking=(member_event_id is None),
         )
 
+        state = None
+        if event_filter and event_filter.lazy_load_members():
+            # TODO: remove redundant members
+
+            types = [
+                (EventTypes.Member, state_key)
+                for state_key in set(
+                    event.sender  # FIXME: we also care about invite targets etc.
+                    for event in events
+                )
+            ]
+
+            state_ids = yield self.store.get_state_ids_for_event(
+                events[0].event_id, types=types,
+            )
+
+            if state_ids:
+                state = yield self.store.get_events(list(state_ids.values()))
+
+            if state:
+                state = yield filter_events_for_client(
+                    self.store,
+                    user_id,
+                    state.values(),
+                    is_peeking=(member_event_id is None),
+                )
+
         time_now = self.clock.time_msec()
 
         chunk = {
@@ -262,4 +289,10 @@ def get_messages(self, requester, room_id=None, pagin_config=None,
             "end": next_token.to_string(),
         }
 
+        if state:
+            chunk["state"] = [
+                serialize_event(e, time_now, as_client_event)
+                for e in state
+            ]
+
         defer.returnValue(chunk)

synapse/handlers/sync.py

@@ -75,6 +75,7 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [
     "ephemeral",
     "account_data",
     "unread_notifications",
+    "summary",
 ])):
     __slots__ = []
 
@@ -494,10 +495,128 @@ def get_state_at(self, room_id, stream_position, types=None, filtered_types=None
             state = {}
         defer.returnValue(state)
 
+    @defer.inlineCallbacks
+    def compute_summary(self, room_id, sync_config, batch, state, now_token):
+        """ Works out a room summary block for this room, summarising the number
+        of joined members in the room, and providing the 'hero' members if the
+        room has no name so clients can consistently name rooms.  Also adds
+        state events to 'state' if needed to describe the heroes.
+
+        Args:
+            room_id(str):
+            sync_config(synapse.handlers.sync.SyncConfig):
+            batch(synapse.handlers.sync.TimelineBatch): The timeline batch for
+                the room that will be sent to the user.
+            state(dict): dict of (type, state_key) -> Event as returned by
+                compute_state_delta
+            now_token(str): Token of the end of the current batch.
+
+        Returns:
+             A deferred dict describing the room summary
+        """
+
+        # FIXME: this promulgates https://github.com/matrix-org/synapse/issues/3305
+        last_events, _ = yield self.store.get_recent_event_ids_for_room(
+            room_id, end_token=now_token.room_key, limit=1,
+        )
+
+        if not last_events:
+            defer.returnValue(None)
+            return
+
+        last_event = last_events[-1]
+        state_ids = yield self.store.get_state_ids_for_event(
+            last_event.event_id, [
+                (EventTypes.Member, None),
+                (EventTypes.Name, ''),
+                (EventTypes.CanonicalAlias, ''),
+            ]
+        )
+
+        member_ids = {
+            state_key: event_id
+            for (t, state_key), event_id in state_ids.iteritems()
+            if t == EventTypes.Member
+        }
+        name_id = state_ids.get((EventTypes.Name, ''))
+        canonical_alias_id = state_ids.get((EventTypes.CanonicalAlias, ''))
+
+        summary = {}
+
+        # FIXME: it feels very heavy to load up every single membership event
+        # just to calculate the counts.
+        member_events = yield self.store.get_events(member_ids.values())
+
+        joined_user_ids = []
+        invited_user_ids = []
+
+        for ev in member_events.values():
+            if ev.content.get("membership") == Membership.JOIN:
+                joined_user_ids.append(ev.state_key)
+            elif ev.content.get("membership") == Membership.INVITE:
+                invited_user_ids.append(ev.state_key)
+
+        # TODO: only send these when they change.
+        summary["m.joined_member_count"] = len(joined_user_ids)
+        summary["m.invited_member_count"] = len(invited_user_ids)
+
+        if not name_id and not canonical_alias_id:
+            # FIXME: order by stream ordering, not alphabetic
+
+            me = sync_config.user.to_string()
+            if (joined_user_ids or invited_user_ids):
+                summary['m.heroes'] = sorted(
+                    [
+                        user_id
+                        for user_id in (joined_user_ids + invited_user_ids)
+                        if user_id != me
+                    ]
+                )[0:5]
+            else:
+                summary['m.heroes'] = sorted(
+                    [user_id for user_id in member_ids.keys() if user_id != me]
+                )[0:5]
+
+            if sync_config.filter_collection.lazy_load_members():
+                # ensure we send membership events for heroes if needed
+                cache_key = (sync_config.user.to_string(), sync_config.device_id)
+                cache = self.lazy_loaded_members_cache.get(cache_key)
+
+                # track which members the client should already know about via LL:
+                # Ones which are already in state...
+                existing_members = set(
+                    user_id for (typ, user_id) in state.keys()
+                    if typ == EventTypes.Member
+                )
+
+                # ...or ones which are in the timeline...
+                for ev in batch.events:
+                    if ev.type == EventTypes.Member:
+                        existing_members.add(ev.state_key)
+
+                # ...and then ensure any missing ones get included in state.
+                missing_hero_event_ids = [
+                    member_ids[hero_id]
+                    for hero_id in summary['m.heroes']
+                    if (
+                        cache.get(hero_id) != member_ids[hero_id] and
+                        hero_id not in existing_members
+                    )
+                ]
+
+                missing_hero_state = yield self.store.get_events(missing_hero_event_ids)
+                missing_hero_state = missing_hero_state.values()
+
+                for s in missing_hero_state:
+                    cache.set(s.state_key, s.event_id)
+                    state[(EventTypes.Member, s.state_key)] = s
+
+        defer.returnValue(summary)
+
     @defer.inlineCallbacks
     def compute_state_delta(self, room_id, batch, sync_config, since_token, now_token,
                             full_state):
-        """ Works out the differnce in state between the start of the timeline
+        """ Works out the difference in state between the start of the timeline
         and the previous sync.
 
         Args:
@@ -511,7 +630,7 @@ def compute_state_delta(self, room_id, batch, sync_config, since_token, now_toke
             full_state(bool): Whether to force returning the full state.
 
         Returns:
-             A deferred new event dictionary
+             A deferred dict of (type, state_key) -> Event
         """
         # TODO(mjark) Check if the state events were received by the server
         # after the previous sync, since we need to include those state
@@ -1416,7 +1535,6 @@ def _generate_room_entry(self, sync_result_builder, ignored_users,
             if events == [] and tags is None:
                 return
 
-        since_token = sync_result_builder.since_token
         now_token = sync_result_builder.now_token
         sync_config = sync_result_builder.sync_config
 
@@ -1459,6 +1577,18 @@ def _generate_room_entry(self, sync_result_builder, ignored_users,
             full_state=full_state
         )
 
+        summary = {}
+        if (
+            sync_config.filter_collection.lazy_load_members() and
+            (
+                any(ev.type == EventTypes.Member for ev in batch.events) or
+                since_token is None
+            )
+        ):
+            summary = yield self.compute_summary(
+                room_id, sync_config, batch, state, now_token
+            )
+
         if room_builder.rtype == "joined":
             unread_notifications = {}
             room_sync = JoinedSyncResult(
@@ -1468,6 +1598,7 @@ def _generate_room_entry(self, sync_result_builder, ignored_users,
                 ephemeral=ephemeral,
                 account_data=account_data_events,
                 unread_notifications=unread_notifications,
+                summary=summary,
             )
 
             if room_sync or always_include:

synapse/rest/client/v1/room.py

@@ -384,15 +384,30 @@ def __init__(self, hs):
     def on_GET(self, request, room_id):
         # TODO support Pagination stream API (limit/tokens)
         requester = yield self.auth.get_user_by_req(request)
-        events = yield self.message_handler.get_state_events(
+        handler = self.message_handler
+
+        # request the state as of a given event
+        # useful for synchronising with /messages
+        at_event = parse_string(request, "at")
+
+        # let you filter down on particular memberships
+        membership = parse_string(request, "membership")
+        not_membership = parse_string(request, "not_membership")
+
+        events = yield handler.get_state_events(
             room_id=room_id,
             user_id=requester.user.to_string(),
+            at_event=at_event,
+            types=[(EventTypes.Member, None)],
         )
 
         chunk = []
 
         for event in events:
-            if event["type"] != EventTypes.Member:
+            if (
+                (membership and event['content'].get("membership") != membership) or
+                (not_membership and event['content'].get("membership") == not_membership)
+            ):
                 continue
             chunk.append(event)
 
@@ -401,6 +416,8 @@ def on_GET(self, request, room_id):
         }))
 
 
+# deprecated in favour of /members?membership=join?
+# except it does custom AS logic and has a simpler return format
 class JoinedRoomMemberListRestServlet(ClientV1RestServlet):
     PATTERNS = client_path_patterns("/rooms/(?P<room_id>[^/]*)/joined_members$")
 

synapse/rest/client/v2_alpha/sync.py

@@ -370,6 +370,7 @@ def serialize(event):
             ephemeral_events = room.ephemeral
             result["ephemeral"] = {"events": ephemeral_events}
             result["unread_notifications"] = room.unread_notifications
+            result["summary"] = room.summary
 
         return result
 

synapse/rest/client/versions.py

@@ -27,11 +27,17 @@ class VersionsRestServlet(RestServlet):
     def on_GET(self, request):
         return (200, {
             "versions": [
-                "r0.0.1",
-                "r0.1.0",
-                "r0.2.0",
+                # we don't need to include the minor values any more, as we've defined
+                # X.Y.Z to be backwards-compatible with X.(Y-1).* where Y>1
+                # "r0.0.1",
+                # "r0.1.0",
+                # "r0.2.0",
                 "r0.3.0",
-            ]
+            ],
+            # as per MSC1497:
+            "unstable_features": {
+                "m.lazy_load_members": True,
+            }
         })