Return some room data in Sliding Sync /sync

changelog.d/17320.feature

@@ -0,0 +1 @@
+Add `rooms` data to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

synapse/events/utils.py

@@ -836,3 +836,21 @@ def maybe_upsert_event_field(
             del container[key]
 
     return upsert_okay
+
+
+def strip_event(event: EventBase) -> JsonDict:
+    """
+    Used for "stripped state" events which provide a simplified view of the state of a
+    room intended to help a potential joiner identify the room (relevant when the user
+    is invited or knocked).
+
+    Stripped state events can only have the `sender`, `type`, `state_key` and `content`
+    properties present.
+    """
+
+    return {
+        "type": event.type,
+        "state_key": event.state_key,
+        "content": event.content,
+        "sender": event.sender,
+    }

synapse/handlers/sliding_sync.py

@@ -18,22 +18,27 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple
 
+import attr
 from immutabledict import immutabledict
 
-from synapse.api.constants import AccountDataTypes, EventTypes, Membership
+from synapse.api.constants import AccountDataTypes, Direction, EventTypes, Membership
 from synapse.events import EventBase
+from synapse.events.utils import strip_event
 from synapse.storage.roommember import RoomsForUser
 from synapse.types import (
+    JsonDict,
     PersistedEventPosition,
     Requester,
     RoomStreamToken,
+    StreamKeyType,
     StreamToken,
     UserID,
 )
 from synapse.types.handlers import OperationType, SlidingSyncConfig, SlidingSyncResult
-from synapse.types.state import StateFilter
+from synapse.types.state import StateFilter, StateKey
+from synapse.visibility import filter_events_for_client
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -82,6 +87,24 @@ def filter_membership_for_sync(*, membership: str, user_id: str, sender: str) ->
     return membership != Membership.LEAVE or sender != user_id
 
 
+# We can't freeze this class because we want to update it in place with the
+# de-duplicated data.
+@attr.s(slots=True, auto_attribs=True)
+class RoomSyncConfig:
+    """
+    Holds the config for what data we should fetch for a room in the sync response.
+
+    Attributes:
+        timeline_limit: The maximum number of events to return in the timeline.
+        required_state: The minimum set of state events requested for the room. The
+            values are close to `StateKey` but actually use a syntax where you can provide
+            `*` and `$LAZY` as the state key part of the tuple (type, state_key).
+    """
+
+    timeline_limit: int
+    required_state: Set[Tuple[str, str]]
+
+
 class SlidingSyncHandler:
     def __init__(self, hs: "HomeServer"):
         self.clock = hs.get_clock()
@@ -201,6 +224,7 @@ async def current_sync_for_user(
 
         # Assemble sliding window lists
         lists: Dict[str, SlidingSyncResult.SlidingWindowList] = {}
+        relevant_room_map: Dict[str, RoomSyncConfig] = {}
         if sync_config.lists:
             # Get all of the room IDs that the user should be able to see in the sync
             # response
@@ -225,29 +249,66 @@ async def current_sync_for_user(
                 ops: List[SlidingSyncResult.SlidingWindowList.Operation] = []
                 if list_config.ranges:
                     for range in list_config.ranges:
+                        sliced_room_ids = [
+                            room_id
+                            for room_id, _ in sorted_room_info[range[0] : range[1]]
+                        ]
+
                         ops.append(
                             SlidingSyncResult.SlidingWindowList.Operation(
                                 op=OperationType.SYNC,
                                 range=range,
-                                room_ids=[
-                                    room_id
-                                    for room_id, _ in sorted_room_info[
-                                        range[0] : range[1]
-                                    ]
-                                ],
+                                room_ids=sliced_room_ids,
                             )
                         )
 
+                        # Update the relevant room map
+                        for room_id in sliced_room_ids:
+                            if relevant_room_map.get(room_id) is not None:
+                                # Take the highest timeline limit
+                                if (
+                                    relevant_room_map[room_id].timeline_limit
+                                    < list_config.timeline_limit
+                                ):
+                                    relevant_room_map[room_id].timeline_limit = (
+                                        list_config.timeline_limit
+                                    )
+
+                                # Union the required state
+                                relevant_room_map[room_id].required_state.update(
+                                    list_config.required_state
+                                )
+                            else:
+                                relevant_room_map[room_id] = RoomSyncConfig(
+                                    timeline_limit=list_config.timeline_limit,
+                                    required_state=set(list_config.required_state),
+                                )
+
                 lists[list_key] = SlidingSyncResult.SlidingWindowList(
                     count=len(sorted_room_info),
                     ops=ops,
                 )
 
+        # TODO: if (sync_config.room_subscriptions):
+
+        # Fetch room data
+        rooms: Dict[str, SlidingSyncResult.RoomResult] = {}
+        for room_id, room_sync_config in relevant_room_map.items():
+            room_sync_result = await self.get_room_sync_data(
+                user=sync_config.user,
+                room_id=room_id,
+                room_sync_config=room_sync_config,
+                rooms_for_user_membership_at_to_token=sync_room_map[room_id],
+                from_token=from_token,
+                to_token=to_token,
+            )
+
+            rooms[room_id] = room_sync_result
+
         return SlidingSyncResult(
             next_pos=to_token,
             lists=lists,
-            # TODO: Gather room data for rooms in lists and `sync_config.room_subscriptions`
-            rooms={},
+            rooms=rooms,
             extensions={},
         )
 
@@ -665,3 +726,154 @@ async def sort_rooms(
             # We want descending order
             reverse=True,
         )
+
+    async def get_room_sync_data(
+        self,
+        user: UserID,
+        room_id: str,
+        room_sync_config: RoomSyncConfig,
+        rooms_for_user_membership_at_to_token: RoomsForUser,
+        from_token: Optional[StreamToken],
+        to_token: StreamToken,
+    ) -> SlidingSyncResult.RoomResult:
+        """
+        Fetch room data for a room.
+
+        We fetch data according to the token range (> `from_token` and <= `to_token`).
+
+        Args:
+            user: User to fetch data for
+            room_id: The room ID to fetch data for
+            room_sync_config: Config for what data we should fetch for a room in the
+                sync response.
+            rooms_for_user_membership_at_to_token: Membership information for the user
+                in the room at the time of `to_token`.
+            from_token: The point in the stream to sync from.
+            to_token: The point in the stream to sync up to.
+        """
+
+        timeline_events: List[EventBase] = []
+        limited = False
+        # We want to use `to_token` (vs `from_token`) because we look backwards from the
+        # `to_token` up to the `timeline_limit` and we might not reach `from_token`
+        # before we hit the limit. We will update the room stream position once we've
+        # fetched the events.
+        prev_batch_token = to_token
+        if room_sync_config.timeline_limit > 0:
+            timeline_events, new_room_key = await self.store.paginate_room_events(
+                room_id=room_id,
+                # We're going to paginate backwards from the `to_token`
+                from_key=to_token.room_key,
+                # We should always return historical messages (outside token range) in
+                # these cases because clients want to be able to show a basic screen of
+                # information:
+                #  - Initial sync (because no `from_token`)
+                #  - When users newly_join
+                #  - TODO: For incremental sync where we haven't sent it down this
+                #    connection before
+                to_key=from_token.room_key if from_token is not None else None,
+                direction=Direction.BACKWARDS,
+                # We add one so we can determine if there are enough events to saturate
+                # the limit or not (see `limited`)
+                limit=room_sync_config.timeline_limit + 1,
+                event_filter=None,
+            )
+
+            # We want to return the events in ascending order (the last event is the
+            # most recent).
+            timeline_events.reverse()
+
+            timeline_events = await filter_events_for_client(
+                self.storage_controllers,
+                user.to_string(),
+                timeline_events,
+                is_peeking=rooms_for_user_membership_at_to_token.membership
+                != Membership.JOIN,
+                filter_send_to_client=True,
+            )
+
+            # Determine our `limited` status
+            if len(timeline_events) > room_sync_config.timeline_limit:
+                limited = True
+                # Get rid of that extra "+ 1" event because we only used it to determine
+                # if we hit the limit or not
+                timeline_events = timeline_events[-room_sync_config.timeline_limit :]
+                assert timeline_events[0].internal_metadata.stream_ordering
+                new_room_key = RoomStreamToken(
+                    stream=timeline_events[0].internal_metadata.stream_ordering - 1
+                )
+
+            prev_batch_token = prev_batch_token.copy_and_replace(
+                StreamKeyType.ROOM, new_room_key
+            )
+
+        # Figure out any stripped state events for invite/knocks
+        stripped_state: List[JsonDict] = []
+        if rooms_for_user_membership_at_to_token.membership in {
+            Membership.INVITE,
+            Membership.KNOCK,
+        }:
+            invite_or_knock_event = await self.store.get_event(
+                rooms_for_user_membership_at_to_token.event_id
+            )
+
+            stripped_state = []
+            if invite_or_knock_event.membership == Membership.INVITE:
+                stripped_state.extend(
+                    invite_or_knock_event.unsigned.get("invite_room_state", [])
+                )
+            elif invite_or_knock_event.membership == Membership.KNOCK:
+                stripped_state.extend(
+                    invite_or_knock_event.unsigned.get("knock_room_state", [])
+                )
+
+            stripped_state.append(strip_event(invite_or_knock_event))
+
+        required_state = []
+        if len(room_sync_config.required_state) > 0:
+            await self.storage_controllers.state.get_current_state(
+                room_id,
+                state_filter=StateFilter.from_types(TODO),
+                await_full_state=False,
+            )
+
+            # TODO: rewind
+
+            # required_state = await self.storage_controllers.state.get_state_at(
+            #     room_id,
+            #     to_token,
+            #     state_filter=StateFilter.from_types(TODO),
+            # )
+
+        return SlidingSyncResult.RoomResult(
+            # TODO: Dummy value
+            # TODO: Make this optional because a computed name doesn't make sense for translated cases
+            name="TODO",
+            # TODO: Dummy value
+            avatar=None,
+            # TODO: Dummy value
+            heroes=None,
+            # Since we can't determine whether we've already sent a room down this
+            # Sliding Sync connection before (we plan to add this optimization in the
+            # future), we're always returning the requested room state instead of
+            # updates.
+            initial=True,
+            # TODO: Dummy value
+            required_state=[],
+            timeline=timeline_events,
+            # TODO: Dummy value
+            is_dm=False,
+            stripped_state=stripped_state,
+            prev_batch=prev_batch_token,
+            limited=limited,
+            # TODO: Dummy values
+            joined_count=0,
+            invited_count=0,
+            # TODO: These are just dummy values. We could potentially just remove these
+            # since notifications can only really be done correctly on the client anyway
+            # (encrypted rooms).
+            notification_count=0,
+            highlight_count=0,
+            # TODO: Dummy value
+            num_live=0,
+        )