Add note

Based on [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575): Sliding Sync

changelog.d/17187.feature

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

changelog.d/17244.feature

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

changelog.d/17248.feature

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

changelog.d/17249.feature

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

synapse/handlers/room_summary.py

@@ -452,11 +452,15 @@ class RoomSummaryHandler:
             return _RoomEntry(room_id, room_entry)
 
         # Otherwise, look for child rooms/spaces.
-        child_events = await self._get_child_events(room_id)
+        space_child_events = await self._get_space_child_events(room_id)
+        # Sort the results for stability.
+        space_child_events = sorted(
+            space_child_events, key=_child_events_comparison_key
+        )
 
         if suggested_only:
             # we only care about suggested children
-            child_events = filter(_is_suggested_child_event, child_events)
+            space_child_events = filter(_is_suggested_child_event, space_child_events)
 
         stripped_events: List[JsonDict] = [
             {
@@ -466,7 +470,7 @@ class RoomSummaryHandler:
                 "sender": e.sender,
                 "origin_server_ts": e.origin_server_ts,
             }
-            for e in child_events
+            for e in space_child_events
         ]
         return _RoomEntry(room_id, room_entry, stripped_events)
 
@@ -763,11 +767,9 @@ class RoomSummaryHandler:
 
         return room_entry
 
-    async def _get_child_events(self, room_id: str) -> Iterable[EventBase]:
+    async def _get_space_child_events(self, room_id: str) -> Iterable[EventBase]:
         """
-        Get the child events for a given room.
-
-        The returned results are sorted for stability.
+        Get the space child events for a given room.
 
         Args:
             room_id: The room id to get the children of.
@@ -791,7 +793,9 @@ class RoomSummaryHandler:
 
         # filter out any events without a "via" (which implies it has been redacted),
         # and order to ensure we return stable results.
-        return sorted(filter(_has_valid_via, events), key=_child_events_comparison_key)
+        filtered_events = filter(_has_valid_via, events)
+
+        return filtered_events
 
     async def get_room_summary(
         self,

synapse/handlers/sliding_sync.py

@@ -0,0 +1,617 @@
+import logging
+from enum import Enum
+from typing import TYPE_CHECKING, AbstractSet, Dict, Final, List, Optional, Set, Tuple
+
+import attr
+
+from synapse._pydantic_compat import HAS_PYDANTIC_V2
+
+if TYPE_CHECKING or HAS_PYDANTIC_V2:
+    from pydantic.v1 import Extra
+else:
+    from pydantic import Extra
+
+from synapse.api.constants import AccountDataTypes, EventTypes, Membership
+from synapse.events import EventBase
+from synapse.rest.client.models import SlidingSyncBody
+from synapse.types import JsonMapping, Requester, RoomStreamToken, StreamToken, UserID
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+
+# Everything except `Membership.LEAVE`
+MEMBERSHIP_TO_DISPLAY_IN_SYNC = (
+    Membership.INVITE,
+    Membership.JOIN,
+    Membership.KNOCK,
+    Membership.BAN,
+)
+
+
+class SlidingSyncConfig(SlidingSyncBody):
+    """
+    Inherit from `SlidingSyncBody` since we need all of the same fields and add a few
+    extra fields that we need in the handler
+    """
+
+    user: UserID
+    device_id: Optional[str]
+
+    # Pydantic config
+    class Config:
+        # By default, ignore fields that we don't recognise.
+        extra = Extra.ignore
+        # By default, don't allow fields to be reassigned after parsing.
+        allow_mutation = False
+        # Allow custom types like `UserID` to be used in the model
+        arbitrary_types_allowed = True
+
+
+class OperationType(Enum):
+    """
+    Represents the operation types in a Sliding Sync window.
+
+    Attributes:
+        SYNC: Sets a range of entries. Clients SHOULD discard what they previous knew about
+            entries in this range.
+        INSERT: Sets a single entry. If the position is not empty then clients MUST move
+            entries to the left or the right depending on where the closest empty space is.
+        DELETE: Remove a single entry. Often comes before an INSERT to allow entries to move
+            places.
+        INVALIDATE: Remove a range of entries. Clients MAY persist the invalidated range for
+            offline support, but they should be treated as empty when additional operations
+            which concern indexes in the range arrive from the server.
+    """
+
+    SYNC: Final = "SYNC"
+    INSERT: Final = "INSERT"
+    DELETE: Final = "DELETE"
+    INVALIDATE: Final = "INVALIDATE"
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class SlidingSyncResult:
+    """
+    The Sliding Sync result to be serialized to JSON for a response.
+
+    Attributes:
+        next_pos: The next position token in the sliding window to request (next_batch).
+        lists: Sliding window API. A map of list key to list results.
+        rooms: Room subscription API. A map of room ID to room subscription to room results.
+        extensions: Extensions API. A map of extension key to extension results.
+    """
+
+    @attr.s(slots=True, frozen=True, auto_attribs=True)
+    class RoomResult:
+        """
+        Attributes:
+            name: Room name or calculated room name.
+            avatar: Room avatar
+            heroes: List of stripped membership events (containing `user_id` and optionally
+                `avatar_url` and `displayname`) for the users used to calculate the room name.
+            initial: Flag which is set when this is the first time the server is sending this
+                data on this connection. Clients can use this flag to replace or update
+                their local state. When there is an update, servers MUST omit this flag
+                entirely and NOT send "initial":false as this is wasteful on bandwidth. The
+                absence of this flag means 'false'.
+            required_state: The current state of the room
+            timeline: Latest events in the room. The last event is the most recent
+            is_dm: Flag to specify whether the room is a direct-message room (most likely
+                between two people).
+            invite_state: Stripped state events. Same as `rooms.invite.$room_id.invite_state`
+                in sync v2, absent on joined/left rooms
+            prev_batch: A token that can be passed as a start parameter to the
+                `/rooms/<room_id>/messages` API to retrieve earlier messages.
+            limited: True if their are more events than fit between the given position and now.
+                Sync again to get more.
+            joined_count: The number of users with membership of join, including the client's
+                own user ID. (same as sync `v2 m.joined_member_count`)
+            invited_count: The number of users with membership of invite. (same as sync v2
+                `m.invited_member_count`)
+            notification_count: The total number of unread notifications for this room. (same
+                as sync v2)
+            highlight_count: The number of unread notifications for this room with the highlight
+                flag set. (same as sync v2)
+            num_live: The number of timeline events which have just occurred and are not historical.
+                The last N events are 'live' and should be treated as such. This is mostly
+                useful to determine whether a given @mention event should make a noise or not.
+                Clients cannot rely solely on the absence of `initial: true` to determine live
+                events because if a room not in the sliding window bumps into the window because
+                of an @mention it will have `initial: true` yet contain a single live event
+                (with potentially other old events in the timeline).
+        """
+
+        name: str
+        avatar: Optional[str]
+        heroes: Optional[List[EventBase]]
+        initial: bool
+        required_state: List[EventBase]
+        timeline: List[EventBase]
+        is_dm: bool
+        invite_state: List[EventBase]
+        prev_batch: StreamToken
+        limited: bool
+        joined_count: int
+        invited_count: int
+        notification_count: int
+        highlight_count: int
+        num_live: int
+
+    @attr.s(slots=True, frozen=True, auto_attribs=True)
+    class SlidingWindowList:
+        """
+        Attributes:
+            count: The total number of entries in the list. Always present if this list
+                is.
+            ops: The sliding list operations to perform.
+        """
+
+        @attr.s(slots=True, frozen=True, auto_attribs=True)
+        class Operation:
+            """
+            Attributes:
+                op: The operation type to perform.
+                range: Which index positions are affected by this operation. These are
+                    both inclusive.
+                room_ids: Which room IDs are affected by this operation. These IDs match
+                    up to the positions in the `range`, so the last room ID in this list
+                    matches the 9th index. The room data is held in a separate object.
+            """
+
+            op: OperationType
+            range: Tuple[int, int]
+            room_ids: List[str]
+
+        count: int
+        ops: List[Operation]
+
+    next_pos: StreamToken
+    lists: Dict[str, SlidingWindowList]
+    rooms: List[RoomResult]
+    extensions: JsonMapping
+
+    def __bool__(self) -> bool:
+        """Make the result appear empty if there are no updates. This is used
+        to tell if the notifier needs to wait for more events when polling for
+        events.
+        """
+        return bool(self.lists or self.rooms or self.extensions)
+
+
+class SlidingSyncHandler:
+    def __init__(self, hs: "HomeServer"):
+        self.hs_config = hs.config
+        self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
+        self.store = hs.get_datastores().main
+        self.storage_controllers = hs.get_storage_controllers()
+        self.auth_blocking = hs.get_auth_blocking()
+        self.notifier = hs.get_notifier()
+        self.event_sources = hs.get_event_sources()
+        self.room_summary_handler = hs.get_room_summary_handler()
+
+    async def wait_for_sync_for_user(
+        self,
+        requester: Requester,
+        sync_config: SlidingSyncConfig,
+        from_token: Optional[StreamToken] = None,
+        timeout: int = 0,
+    ) -> SlidingSyncResult:
+        """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.
+        """
+        # If the user is not part of the mau group, then check that limits have
+        # not been exceeded (if not part of the group by this point, almost certain
+        # auth_blocking will occur)
+        await self.auth_blocking.check_auth_blocking(requester=requester)
+
+        # TODO: If the To-Device extension is enabled and we have a `from_token`, delete
+        # any to-device messages before that token (since we now know that the device
+        # has received them). (see sync v2 for how to do this)
+
+        if timeout == 0 or from_token is None:
+            now_token = self.event_sources.get_current_token()
+            result = await self.current_sync_for_user(
+                sync_config,
+                from_token=from_token,
+                to_token=now_token,
+            )
+        else:
+            # Otherwise, we wait for something to happen and report it to the user.
+            async def current_sync_callback(
+                before_token: StreamToken, after_token: StreamToken
+            ) -> SlidingSyncResult:
+                return await self.current_sync_for_user(
+                    sync_config,
+                    from_token=from_token,
+                    to_token=after_token,
+                )
+
+            result = await self.notifier.wait_for_events(
+                sync_config.user.to_string(),
+                timeout,
+                current_sync_callback,
+                from_token=from_token,
+            )
+
+        return result
+
+    async def current_sync_for_user(
+        self,
+        sync_config: SlidingSyncConfig,
+        to_token: StreamToken,
+        from_token: Optional[StreamToken] = None,
+    ) -> SlidingSyncResult:
+        """
+        Generates the response body of a Sliding Sync result, represented as a
+        `SlidingSyncResult`.
+        """
+        user_id = sync_config.user.to_string()
+        app_service = self.store.get_app_service_by_user_id(user_id)
+        if app_service:
+            # We no longer support AS users using /sync directly.
+            # See https://github.com/matrix-org/matrix-doc/issues/1144
+            raise NotImplementedError()
+
+        # Get all of the room IDs that the user should be able to see in the sync
+        # response
+        room_id_set = await self.get_sync_room_ids_for_user(
+            sync_config.user,
+            from_token=from_token,
+            to_token=to_token,
+        )
+
+        # Assemble sliding window lists
+        lists: Dict[str, SlidingSyncResult.SlidingWindowList] = {}
+        if sync_config.lists:
+            for list_key, list_config in sync_config.lists.items():
+                # Apply filters
+                filtered_room_ids = room_id_set
+                if list_config.filters is not None:
+                    # TODO: To be absolutely correct, this could also take into account
+                    # from/to tokens
+                    filtered_room_ids = await self.filter_rooms(
+                        sync_config.user, room_id_set, list_config.filters
+                    )
+                # TODO: Apply sorts
+                sorted_room_ids = sorted(filtered_room_ids)
+
+                ops: List[SlidingSyncResult.SlidingWindowList.Operation] = []
+                if list_config.ranges:
+                    for range in list_config.ranges:
+                        ops.append(
+                            SlidingSyncResult.SlidingWindowList.Operation(
+                                op=OperationType.SYNC,
+                                range=range,
+                                room_ids=sorted_room_ids[range[0] : range[1]],
+                            )
+                        )
+
+                lists[list_key] = SlidingSyncResult.SlidingWindowList(
+                    count=len(sorted_room_ids),
+                    ops=ops,
+                )
+
+        return SlidingSyncResult(
+            next_pos=to_token,
+            lists=lists,
+            # TODO: Gather room data for rooms in lists and `sync_config.room_subscriptions`
+            rooms=[],
+            extensions={},
+        )
+
+    async def get_sync_room_ids_for_user(
+        self,
+        user: UserID,
+        to_token: StreamToken,
+        from_token: Optional[StreamToken] = None,
+    ) -> AbstractSet[str]:
+        """
+        Fetch room IDs that should be listed for this user in the sync response.
+
+        We're looking for rooms that the user has not left (`invite`, `knock`, `join`,
+        and `ban`) or newly_left rooms that are > `from_token` and <= `to_token`.
+        """
+        user_id = user.to_string()
+
+        # First grab a current snapshot rooms for the user
+        room_for_user_list = await self.store.get_rooms_for_local_user_where_membership_is(
+            user_id=user_id,
+            # We want to fetch any kind of membership (joined and left rooms) in order
+            # to get the `stream_ordering` of the latest room membership event for the
+            # user.
+            #
+            # We will filter out the rooms that the user has left below (see
+            # `MEMBERSHIP_TO_DISPLAY_IN_SYNC`)
+            membership_list=Membership.LIST,
+            excluded_rooms=self.rooms_to_exclude_globally,
+        )
+
+        # If the user has never joined any rooms before, we can just return an empty list
+        if not room_for_user_list:
+            return set()
+
+        # Our working list of rooms that can show up in the sync response
+        sync_room_id_set = {
+            room_for_user.room_id
+            for room_for_user in room_for_user_list
+            if room_for_user.membership in MEMBERSHIP_TO_DISPLAY_IN_SYNC
+        }
+
+        # Find the stream_ordering of the latest room membership event which will mark
+        # the spot we queried up to.
+        max_stream_ordering_from_room_list = max(
+            room_for_user.stream_ordering for room_for_user in room_for_user_list
+        )
+
+        # If our `to_token` is already the same or ahead of the latest room membership
+        # for the user, we can just straight-up return the room list (nothing has
+        # changed)
+        if max_stream_ordering_from_room_list <= to_token.room_key.stream:
+            return sync_room_id_set
+
+        # We assume the `from_token` is before or at-least equal to the `to_token`
+        assert (
+            from_token is None or from_token.room_key.stream <= to_token.room_key.stream
+        ), f"{from_token.room_key.stream if from_token else None} <= {to_token.room_key.stream}"
+
+        # We assume the `from_token`/`to_token` is before the `max_stream_ordering_from_room_list`
+        assert (
+            from_token is None
+            or from_token.room_key.stream < max_stream_ordering_from_room_list
+        ), f"{from_token.room_key.stream if from_token else None} < {max_stream_ordering_from_room_list}"
+        assert (
+            to_token.room_key.stream < max_stream_ordering_from_room_list
+        ), f"{to_token.room_key.stream} < {max_stream_ordering_from_room_list}"
+
+        # Since we fetched the users room list at some point in time after the from/to
+        # tokens, we need to revert/rewind some membership changes to match the point in
+        # time of the `to_token`.
+        #
+        # - 1) Add back newly_left rooms (> `from_token` and <= `to_token`)
+        # - 2a) Remove rooms that the user joined after the `to_token`
+        # - 2b) Add back rooms that the user left after the `to_token`
+        membership_change_events = await self.store.get_membership_changes_for_user(
+            user_id,
+            # Start from the `from_token` if given, otherwise from the `to_token` so we
+            # can still do the 2) fixups.
+            from_key=from_token.room_key if from_token else to_token.room_key,
+            # Fetch up to our membership snapshot
+            to_key=RoomStreamToken(stream=max_stream_ordering_from_room_list),
+            excluded_rooms=self.rooms_to_exclude_globally,
+        )
+
+        # Assemble a list of the last membership events in some given ranges. Someone
+        # could have left and joined multiple times during the given range but we only
+        # care about end-result so we grab the last one.
+        last_membership_change_by_room_id_in_from_to_range: Dict[str, EventBase] = {}
+        last_membership_change_by_room_id_after_to_token: Dict[str, EventBase] = {}
+        # We also need the first membership event after the `to_token` so we can step
+        # backward to the previous membership that would apply to the from/to range.
+        first_membership_change_by_room_id_after_to_token: Dict[str, EventBase] = {}
+        for event in membership_change_events:
+            assert event.internal_metadata.stream_ordering
+
+            if (
+                (
+                    from_token is None
+                    or event.internal_metadata.stream_ordering
+                    > from_token.room_key.stream
+                )
+                and event.internal_metadata.stream_ordering <= to_token.room_key.stream
+            ):
+                last_membership_change_by_room_id_in_from_to_range[event.room_id] = (
+                    event
+                )
+            elif (
+                event.internal_metadata.stream_ordering > to_token.room_key.stream
+                and event.internal_metadata.stream_ordering
+                <= max_stream_ordering_from_room_list
+            ):
+                last_membership_change_by_room_id_after_to_token[event.room_id] = event
+                # Only set if we haven't already set it
+                first_membership_change_by_room_id_after_to_token.setdefault(
+                    event.room_id, event
+                )
+            else:
+                # We don't expect this to happen since we should only be fetching
+                # `membership_change_events` that fall in the given ranges above. It
+                # doesn't hurt anything to ignore an event we don't need but may
+                # indicate a bug in the logic above.
+                raise AssertionError(
+                    "Membership event with stream_ordering=%s should fall in the given ranges above"
+                    + " (%d > x <= %d) or (%d > x <= %d). We shouldn't be fetching extra membership"
+                    + " events that aren't used.",
+                    event.internal_metadata.stream_ordering,
+                    from_token.room_key.stream if from_token else None,
+                    to_token.room_key.stream,
+                    to_token.room_key.stream,
+                    max_stream_ordering_from_room_list,
+                )
+
+        # 1)
+        for (
+            last_membership_change_in_from_to_range
+        ) in last_membership_change_by_room_id_in_from_to_range.values():
+            room_id = last_membership_change_in_from_to_range.room_id
+
+            # 1) Add back newly_left rooms (> `from_token` and <= `to_token`). We
+            # include newly_left rooms because the last event that the user should see
+            # is their own leave event
+            if last_membership_change_in_from_to_range.membership == Membership.LEAVE:
+                sync_room_id_set.add(room_id)
+
+        # 2)
+        for (
+            last_membership_change_after_to_token
+        ) in last_membership_change_by_room_id_after_to_token.values():
+            room_id = last_membership_change_after_to_token.room_id
+
+            # We want to find the first membership change after the `to_token` then step
+            # backward to know the membership in the from/to range.
+            first_membership_change_after_to_token = (
+                first_membership_change_by_room_id_after_to_token.get(room_id)
+            )
+            assert first_membership_change_after_to_token is not None, (
+                "If there was a `last_membership_change_after_to_token` that we're iterating over, "
+                + "then there should be corresponding a first change. For example, even if there "
+                + "is only one event after the `to_token`, the first and last event will be same event. "
+                + "This is probably a mistake in assembling the `last_membership_change_by_room_id_after_to_token`"
+                + "/`first_membership_change_by_room_id_after_to_token` dicts above."
+            )
+            prev_content = first_membership_change_after_to_token.unsigned.get(
+                "prev_content", {}
+            )
+            prev_membership = prev_content.get("membership", None)
+
+            # 2a) Add back rooms that the user left after the `to_token`
+            #
+            # If the last membership event after the `to_token` is a leave event, then
+            # the room was excluded from the
+            # `get_rooms_for_local_user_where_membership_is()` results. We should add
+            # these rooms back as long as the user was part of the room before the
+            # `to_token`.
+            if (
+                last_membership_change_after_to_token.membership == Membership.LEAVE
+                and prev_membership is not None
+                and prev_membership != Membership.LEAVE
+            ):
+                sync_room_id_set.add(room_id)
+            # 2b) Remove rooms that the user joined (hasn't left) after the `to_token`
+            #
+            # If the last membership event after the `to_token` is a "join" event, then
+            # the room was included in the `get_rooms_for_local_user_where_membership_is()`
+            # results. We should remove these rooms as long as the user wasn't part of
+            # the room before the `to_token`.
+            elif (
+                last_membership_change_after_to_token.membership != Membership.LEAVE
+                and (prev_membership is None or prev_membership == Membership.LEAVE)
+            ):
+                sync_room_id_set.discard(room_id)
+
+        return sync_room_id_set
+
+    async def filter_rooms(
+        self,
+        user: UserID,
+        room_id_set: AbstractSet[str],
+        filters: SlidingSyncConfig.SlidingSyncList.Filters,
+    ) -> AbstractSet[str]:
+        """
+        Filter rooms based on the sync request.
+        """
+        user_id = user.to_string()
+
+        # TODO: Re-order filters so that the easiest, most likely to eliminate rooms,
+        # are first. This way when people use multiple filters, we can eliminate rooms
+        # and do less work for the subsequent filters.
+        #
+        # TODO: Exclude partially stated rooms unless the `required_state` has
+        # `["m.room.member", "$LAZY"]`
+
+        filtered_room_id_set = set(room_id_set)
+
+        # Filter for Direct-Message (DM) rooms
+        if filters.is_dm is not None:
+            # We're using global account data (`m.direct`) instead of checking for
+            # `is_direct` on membership events because that property only appears for
+            # the invitee membership event (doesn't show up for the inviter). Account
+            # data is set by the client so it needs to be scrutinized.
+            dm_map = await self.store.get_global_account_data_by_type_for_user(
+                user_id, AccountDataTypes.DIRECT
+            )
+            logger.warn("dm_map: %s", dm_map)
+            # Flatten out the map
+            dm_room_id_set = set()
+            if dm_map:
+                for room_ids in dm_map.values():
+                    # Account data should be a list of room IDs. Ignore anything else
+                    if isinstance(room_ids, list):
+                        for room_id in room_ids:
+                            if isinstance(room_id, str):
+                                dm_room_id_set.add(room_id)
+
+            if filters.is_dm:
+                # Only DM rooms please
+                filtered_room_id_set = filtered_room_id_set.intersection(dm_room_id_set)
+            else:
+                # Only non-DM rooms please
+                filtered_room_id_set = filtered_room_id_set.difference(dm_room_id_set)
+
+        # Filter the room based on the space they belong to according to `m.space.child`
+        # state events. If multiple spaces are present, a room can be part of any one of
+        # the listed spaces (OR'd).
+        if filters.spaces:
+            # Only use spaces that we're joined to to avoid leaking private space
+            # information that the user is not part of. We could probably allow
+            # public spaces here but the spec says "joined" only.
+            joined_space_room_ids = set()
+            for space_room_id in set(filters.spaces):
+                # TODO: Is there a good method to look up all space rooms at once? (N+1 query problem)
+                is_user_in_room = await self.store.check_local_user_in_room(
+                    user_id=user.to_string(), room_id=space_room_id
+                )
+
+                if is_user_in_room:
+                    joined_space_room_ids.add(space_room_id)
+
+            # Flatten the child rooms in the spaces
+            space_child_room_ids: Set[str] = set()
+            for space_room_id in joined_space_room_ids:
+                space_child_events = (
+                    await self.room_summary_handler._get_space_child_events(
+                        space_room_id
+                    )
+                )
+                space_child_room_ids.update(
+                    event.state_key for event in space_child_events
+                )
+                # TODO: The spec says that if the child room has a `m.room.tombstone`
+                # event, we should recursively navigate until we find the latest room
+                # and include those IDs (although this point is under scrutiny).
+
+            # Only rooms in the spaces please
+            filtered_room_id_set = filtered_room_id_set.intersection(
+                space_child_room_ids
+            )
+
+        # Filter for encrypted rooms
+        if filters.is_encrypted is not None:
+            # Make a copy so we don't run into an error: `Set changed size during iteration`
+            for room_id in list(filtered_room_id_set):
+                # TODO: Is there a good method to look up all rooms at once? (N+1 query problem)
+                is_encrypted = (
+                    await self.storage_controllers.state.get_current_state_event(
+                        room_id, EventTypes.RoomEncryption, ""
+                    )
+                )
+
+                # If we're looking for encrypted rooms, filter out rooms that are not
+                # encrypted and vice versa
+                if (filters.is_encrypted and not is_encrypted) or (
+                    not filters.is_encrypted and is_encrypted
+                ):
+                    filtered_room_id_set.remove(room_id)
+
+        if filters.is_invite:
+            raise NotImplementedError()
+
+        if filters.room_types:
+            raise NotImplementedError()
+
+        if filters.not_room_types:
+            raise NotImplementedError()
+
+        if filters.room_name_like:
+            raise NotImplementedError()
+
+        if filters.tags:
+            raise NotImplementedError()
+
+        if filters.not_tags:
+            raise NotImplementedError()
+
+        return filtered_room_id_set

synapse/handlers/sync.py

@@ -1977,7 +1977,7 @@ class SyncHandler:
         """
         user_id = sync_config.user.to_string()
 
-        # Note: we get the users room list *before* we get the current token, this
+        # Note: we get the users room list *before* we get the `now_token`, this
         # avoids checking back in history if rooms are joined after the token is fetched.
         token_before_rooms = self.event_sources.get_current_token()
         mutable_joined_room_ids = set(await self.store.get_rooms_for_user(user_id))
@@ -1989,10 +1989,10 @@ class SyncHandler:
         now_token = self.event_sources.get_current_token()
         log_kv({"now_token": now_token})
 
-        # Since we fetched the users room list before the token, there's a small window
-        # during which membership events may have been persisted, so we fetch these now
-        # and modify the joined room list for any changes between the get_rooms_for_user
-        # call and the get_current_token call.
+        # Since we fetched the users room list before calculating the `now_token` (see
+        # above), there's a small window during which membership events may have been
+        # persisted, so we fetch these now and modify the joined room list for any
+        # changes between the get_rooms_for_user call and the get_current_token call.
         membership_change_events = []
         if since_token:
             membership_change_events = await self.store.get_membership_changes_for_user(
@@ -2002,16 +2002,19 @@ class SyncHandler:
                 self.rooms_to_exclude_globally,
             )
 
-            mem_last_change_by_room_id: Dict[str, EventBase] = {}
+            last_membership_change_by_room_id: Dict[str, EventBase] = {}
             for event in membership_change_events:
-                mem_last_change_by_room_id[event.room_id] = event
+                last_membership_change_by_room_id[event.room_id] = event
 
             # For the latest membership event in each room found, add/remove the room ID
             # from the joined room list accordingly. In this case we only care if the
             # latest change is JOIN.
 
-            for room_id, event in mem_last_change_by_room_id.items():
+            for room_id, event in last_membership_change_by_room_id.items():
                 assert event.internal_metadata.stream_ordering
+                # As a shortcut, skip any events that happened before we got our
+                # `get_rooms_for_user()` snapshot (any changes are already represented
+                # in that list).
                 if (
                     event.internal_metadata.stream_ordering
                     < token_before_rooms.room_key.stream

synapse/rest/client/models.py

@@ -18,14 +18,30 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-from typing import TYPE_CHECKING, Dict, Optional
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
 
 from synapse._pydantic_compat import HAS_PYDANTIC_V2
 
 if TYPE_CHECKING or HAS_PYDANTIC_V2:
-    from pydantic.v1 import Extra, StrictInt, StrictStr, constr, validator
+    from pydantic.v1 import (
+        Extra,
+        StrictBool,
+        StrictInt,
+        StrictStr,
+        conint,
+        constr,
+        validator,
+    )
 else:
-    from pydantic import Extra, StrictInt, StrictStr, constr, validator
+    from pydantic import (
+        Extra,
+        StrictBool,
+        StrictInt,
+        StrictStr,
+        conint,
+        constr,
+        validator,
+    )
 
 from synapse.rest.models import RequestBodyModel
 from synapse.util.threepids import validate_email
@@ -97,3 +113,219 @@ else:
 class MsisdnRequestTokenBody(ThreepidRequestTokenBody):
     country: ISO3116_1_Alpha_2
     phone_number: StrictStr
+
+
+class SlidingSyncBody(RequestBodyModel):
+    """
+    Sliding Sync API request body.
+
+    Attributes:
+        lists: Sliding window API. A map of list key to list information
+            (:class:`SlidingSyncList`). Max lists: 100. The list keys should be
+            arbitrary strings which the client is using to refer to the list. Keep this
+            small as it needs to be sent a lot. Max length: 64 bytes.
+        room_subscriptions: Room subscription API. A map of room ID to room subscription
+            information. Used to subscribe to a specific room. Sometimes clients know
+            exactly which room they want to get information about e.g by following a
+            permalink or by refreshing a webapp currently viewing a specific room. The
+            sliding window API alone is insufficient for this use case because there's
+            no way to say "please track this room explicitly".
+        extensions: Extensions API. A map of extension key to extension config.
+    """
+
+    class CommonRoomParameters(RequestBodyModel):
+        """
+        Common parameters shared between the sliding window and room subscription APIs.
+
+        Attributes:
+            required_state: Required state for each room returned. An array of event
+                type and state key tuples. Elements in this array are ORd together to
+                produce the final set of state events to return. One unique exception is
+                when you request all state events via `["*", "*"]`. When used, all state
+                events are returned by default, and additional entries FILTER OUT the
+                returned set of state events. These additional entries cannot use `*`
+                themselves. For example, `["*", "*"], ["m.room.member",
+                "@alice:example.com"]` will *exclude* every `m.room.member` event
+                *except* for `@alice:example.com`, and include every other state event.
+                In addition, `["*", "*"], ["m.space.child", "*"]` is an error, the
+                `m.space.child` filter is not required as it would have been returned
+                anyway.
+            timeline_limit: The maximum number of timeline events to return per response.
+                (Max 1000 messages)
+            include_old_rooms: Determines if `predecessor` rooms are included in the
+                `rooms` response. The user MUST be joined to old rooms for them to show up
+                in the response.
+        """
+
+        class IncludeOldRooms(RequestBodyModel):
+            timeline_limit: StrictInt
+            required_state: List[Tuple[StrictStr, StrictStr]]
+
+        required_state: List[Tuple[StrictStr, StrictStr]]
+        # mypy workaround via https://github.com/pydantic/pydantic/issues/156#issuecomment-1130883884
+        if TYPE_CHECKING:
+            timeline_limit: int
+        else:
+            timeline_limit: conint(le=1000, strict=True)  # type: ignore[valid-type]
+        include_old_rooms: Optional[IncludeOldRooms] = None
+
+    class SlidingSyncList(CommonRoomParameters):
+        """
+        Attributes:
+            ranges: Sliding window ranges. If this field is missing, no sliding window
+                is used and all rooms are returned in this list. Integers are
+                *inclusive*.
+            sort: How the list should be sorted on the server. The first value is
+                applied first, then tiebreaks are performed with each subsequent sort
+                listed.
+
+                    FIXME: Furthermore, it's not currently defined how servers should behave
+                    if they encounter a filter or sort operation they do not recognise. If
+                    the server rejects the request with an HTTP 400 then that will break
+                    backwards compatibility with new clients vs old servers. However, the
+                    client would be otherwise unaware that only some of the sort/filter
+                    operations have taken effect. We may need to include a "warnings"
+                    section to indicate which sort/filter operations are unrecognised,
+                    allowing for some form of graceful degradation of service.
+                    -- https://github.com/matrix-org/matrix-spec-proposals/blob/kegan/sync-v3/proposals/3575-sync.md#filter-and-sort-extensions
+
+            slow_get_all_rooms: Just get all rooms (for clients that don't want to deal with
+                sliding windows). When true, the `ranges` and `sort` fields are ignored.
+            required_state: Required state for each room returned. An array of event
+                type and state key tuples. Elements in this array are ORd together to
+                produce the final set of state events to return.
+
+                One unique exception is when you request all state events via `["*",
+                "*"]`. When used, all state events are returned by default, and
+                additional entries FILTER OUT the returned set of state events. These
+                additional entries cannot use `*` themselves. For example, `["*", "*"],
+                ["m.room.member", "@alice:example.com"]` will *exclude* every
+                `m.room.member` event *except* for `@alice:example.com`, and include
+                every other state event. In addition, `["*", "*"], ["m.space.child",
+                "*"]` is an error, the `m.space.child` filter is not required as it
+                would have been returned anyway.
+
+                Room members can be lazily-loaded by using the special `$LAZY` state key
+                (`["m.room.member", "$LAZY"]`). Typically, when you view a room, you
+                want to retrieve all state events except for m.room.member events which
+                you want to lazily load. To get this behaviour, clients can send the
+                following::
+
+                    {
+                        "required_state": [
+                            // activate lazy loading
+                            ["m.room.member", "$LAZY"],
+                            // request all state events _except_ for m.room.member
+                            events which are lazily loaded
+                            ["*", "*"]
+                        ]
+                    }
+
+            timeline_limit: The maximum number of timeline events to return per response.
+            include_old_rooms: Determines if `predecessor` rooms are included in the
+                `rooms` response. The user MUST be joined to old rooms for them to show up
+                in the response.
+            include_heroes: Return a stripped variant of membership events (containing
+                `user_id` and optionally `avatar_url` and `displayname`) for the users used
+                to calculate the room name.
+            filters: Filters to apply to the list before sorting.
+            bump_event_types: Allowlist of event types which should be considered recent activity
+                when sorting `by_recency`. By omitting event types from this field,
+                clients can ensure that uninteresting events (e.g. a profile rename) do
+                not cause a room to jump to the top of its list(s). Empty or omitted
+                `bump_event_types` have no effect—all events in a room will be
+                considered recent activity.
+        """
+
+        class Filters(RequestBodyModel):
+            """
+            All fields are applied with AND operators, hence if `is_dm: True` and
+            `is_encrypted: True` then only Encrypted DM rooms will be returned. The absence
+            of fields implies no filter on that criteria: it does NOT imply `False`.
+            These fields may be expanded through use of extensions.
+
+            Attributes:
+                is_dm: Flag which only returns rooms present (or not) in the DM section
+                    of account data. If unset, both DM rooms and non-DM rooms are returned.
+                    If False, only non-DM rooms are returned. If True, only DM rooms are
+                    returned.
+                spaces: Filter the room based on the space they belong to according to
+                    `m.space.child` state events. If multiple spaces are present, a room can
+                    be part of any one of the listed spaces (OR'd). The server will inspect
+                    the `m.space.child` state events for the JOINED space room IDs given.
+                    Servers MUST NOT navigate subspaces. It is up to the client to give a
+                    complete list of spaces to navigate. Only rooms directly mentioned as
+                    `m.space.child` events in these spaces will be returned. Unknown spaces
+                    or spaces the user is not joined to will be ignored.
+                is_encrypted: Flag which only returns rooms which have an
+                    `m.room.encryption` state event. If unset, both encrypted and
+                    unencrypted rooms are returned. If `False`, only unencrypted rooms are
+                    returned. If `True`, only encrypted rooms are returned.
+                is_invite: Flag which only returns rooms the user is currently invited
+                    to. If unset, both invited and joined rooms are returned. If `False`, no
+                    invited rooms are returned. If `True`, only invited rooms are returned.
+                room_types: If specified, only rooms where the `m.room.create` event has
+                    a `type` matching one of the strings in this array will be returned. If
+                    this field is unset, all rooms are returned regardless of type. This can
+                    be used to get the initial set of spaces for an account. For rooms which
+                    do not have a room type, use `null`/`None` to include them.
+                not_room_types: Same as `room_types` but inverted. This can be used to
+                    filter out spaces from the room list. If a type is in both `room_types`
+                    and `not_room_types`, then `not_room_types` wins and they are not included
+                    in the result.
+                room_name_like: Filter the room name. Case-insensitive partial matching
+                    e.g 'foo' matches 'abFooab'. The term 'like' is inspired by SQL 'LIKE',
+                    and the text here is similar to '%foo%'.
+                tags: Filter the room based on its room tags. If multiple tags are
+                    present, a room can have any one of the listed tags (OR'd).
+                not_tags: Filter the room based on its room tags. Takes priority over
+                    `tags`. For example, a room with tags A and B with filters `tags: [A]`
+                    `not_tags: [B]` would NOT be included because `not_tags` takes priority over
+                    `tags`. This filter is useful if your rooms list does NOT include the
+                    list of favourite rooms again.
+            """
+
+            is_dm: Optional[StrictBool] = None
+            spaces: Optional[List[StrictStr]] = None
+            is_encrypted: Optional[StrictBool] = None
+            is_invite: Optional[StrictBool] = None
+            room_types: Optional[List[Union[StrictStr, None]]] = None
+            not_room_types: Optional[List[StrictStr]] = None
+            room_name_like: Optional[StrictStr] = None
+            tags: Optional[List[StrictStr]] = None
+            not_tags: Optional[List[StrictStr]] = None
+
+        # mypy workaround via https://github.com/pydantic/pydantic/issues/156#issuecomment-1130883884
+        if TYPE_CHECKING:
+            ranges: Optional[List[Tuple[int, int]]] = None
+        else:
+            ranges: Optional[List[Tuple[conint(ge=0, strict=True), conint(ge=0, strict=True)]]] = None  # type: ignore[valid-type]
+        sort: Optional[List[StrictStr]] = None
+        slow_get_all_rooms: Optional[StrictBool] = False
+        include_heroes: Optional[StrictBool] = False
+        filters: Optional[Filters] = None
+        bump_event_types: Optional[List[StrictStr]] = None
+
+    class RoomSubscription(CommonRoomParameters):
+        pass
+
+    class Extension(RequestBodyModel):
+        enabled: Optional[StrictBool] = False
+        lists: Optional[List[StrictStr]] = None
+        rooms: Optional[List[StrictStr]] = None
+
+    # mypy workaround via https://github.com/pydantic/pydantic/issues/156#issuecomment-1130883884
+    if TYPE_CHECKING:
+        lists: Optional[Dict[str, SlidingSyncList]] = None
+    else:
+        lists: Optional[Dict[constr(max_length=64, strict=True), SlidingSyncList]] = None  # type: ignore[valid-type]
+    room_subscriptions: Optional[Dict[StrictStr, RoomSubscription]] = None
+    extensions: Optional[Dict[StrictStr, Extension]] = None
+
+    @validator("lists")
+    def lists_length_check(
+        cls, value: Optional[Dict[str, SlidingSyncList]]
+    ) -> Optional[Dict[str, SlidingSyncList]]:
+        if value is not None:
+            assert len(value) <= 100, f"Max lists: 100 but saw {len(value)}"
+        return value

synapse/rest/client/sync.py

@@ -33,6 +33,7 @@ from synapse.events.utils import (
     format_event_raw,
 )
 from synapse.handlers.presence import format_user_presence_state
+from synapse.handlers.sliding_sync import SlidingSyncConfig, SlidingSyncResult
 from synapse.handlers.sync import (
     ArchivedSyncResult,
     InvitedSyncResult,
@@ -43,9 +44,16 @@ from synapse.handlers.sync import (
     SyncVersion,
 )
 from synapse.http.server import HttpServer
-from synapse.http.servlet import RestServlet, parse_boolean, parse_integer, parse_string
+from synapse.http.servlet import (
+    RestServlet,
+    parse_and_validate_json_object_from_request,
+    parse_boolean,
+    parse_integer,
+    parse_string,
+)
 from synapse.http.site import SynapseRequest
 from synapse.logging.opentracing import trace_with_opname
+from synapse.rest.client.models import SlidingSyncBody
 from synapse.types import JsonDict, Requester, StreamToken
 from synapse.util import json_decoder
 from synapse.util.caches.lrucache import LruCache
@@ -735,8 +743,222 @@ class SlidingSyncE2eeRestServlet(RestServlet):
         return 200, response
 
 
+class SlidingSyncRestServlet(RestServlet):
+    """
+    API endpoint for MSC3575 Sliding Sync `/sync`. Allows for clients to request a
+    subset (sliding window) of rooms, state, and timeline events (just what they need)
+    in order to bootstrap quickly and subscribe to only what the client cares about.
+    Because the client can specify what it cares about, we can respond quickly and skip
+    all of the work we would normally have to do with a sync v2 response.
+
+    Request query parameters:
+        timeout: How long to wait for new events in milliseconds.
+        pos: Stream position token when asking for incremental deltas.
+
+    Request body::
+        {
+            // Sliding Window API
+            "lists": {
+                "foo-list": {
+                    "ranges": [ [0, 99] ],
+                    "sort": [ "by_notification_level", "by_recency", "by_name" ],
+                    "required_state": [
+                        ["m.room.join_rules", ""],
+                        ["m.room.history_visibility", ""],
+                        ["m.space.child", "*"]
+                    ],
+                    "timeline_limit": 10,
+                    "filters": {
+                        "is_dm": true
+                    },
+                    "bump_event_types": [ "m.room.message", "m.room.encrypted" ],
+                }
+            },
+            // Room Subscriptions API
+            "room_subscriptions": {
+                "!sub1:bar": {
+                    "required_state": [ ["*","*"] ],
+                    "timeline_limit": 10,
+                    "include_old_rooms": {
+                        "timeline_limit": 1,
+                        "required_state": [ ["m.room.tombstone", ""], ["m.room.create", ""] ],
+                    }
+                }
+            },
+            // Extensions API
+            "extensions": {}
+        }
+
+    Response JSON::
+        {
+            "next_pos": "s58_224_0_13_10_1_1_16_0_1",
+            "lists": {
+                "foo-list": {
+                    "count": 1337,
+                    "ops": [{
+                        "op": "SYNC",
+                        "range": [0, 99],
+                        "room_ids": [
+                            "!foo:bar",
+                            // ... 99 more room IDs
+                        ]
+                    }]
+                }
+            },
+            // Aggregated rooms from lists and room subscriptions
+            "rooms": {
+                // Room from room subscription
+                "!sub1:bar": {
+                    "name": "Alice and Bob",
+                    "avatar": "mxc://...",
+                    "initial": true,
+                    "required_state": [
+                        {"sender":"@alice:example.com","type":"m.room.create", "state_key":"", "content":{"creator":"@alice:example.com"}},
+                        {"sender":"@alice:example.com","type":"m.room.join_rules", "state_key":"", "content":{"join_rule":"invite"}},
+                        {"sender":"@alice:example.com","type":"m.room.history_visibility", "state_key":"", "content":{"history_visibility":"joined"}},
+                        {"sender":"@alice:example.com","type":"m.room.member", "state_key":"@alice:example.com", "content":{"membership":"join"}}
+                    ],
+                    "timeline": [
+                        {"sender":"@alice:example.com","type":"m.room.create", "state_key":"", "content":{"creator":"@alice:example.com"}},
+                        {"sender":"@alice:example.com","type":"m.room.join_rules", "state_key":"", "content":{"join_rule":"invite"}},
+                        {"sender":"@alice:example.com","type":"m.room.history_visibility", "state_key":"", "content":{"history_visibility":"joined"}},
+                        {"sender":"@alice:example.com","type":"m.room.member", "state_key":"@alice:example.com", "content":{"membership":"join"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"A"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"B"}},
+                    ],
+                    "prev_batch": "t111_222_333",
+                    "joined_count": 41,
+                    "invited_count": 1,
+                    "notification_count": 1,
+                    "highlight_count": 0
+                },
+                // rooms from list
+                "!foo:bar": {
+                    "name": "The calculated room name",
+                    "avatar": "mxc://...",
+                    "initial": true,
+                    "required_state": [
+                        {"sender":"@alice:example.com","type":"m.room.join_rules", "state_key":"", "content":{"join_rule":"invite"}},
+                        {"sender":"@alice:example.com","type":"m.room.history_visibility", "state_key":"", "content":{"history_visibility":"joined"}},
+                        {"sender":"@alice:example.com","type":"m.space.child", "state_key":"!foo:example.com", "content":{"via":["example.com"]}},
+                        {"sender":"@alice:example.com","type":"m.space.child", "state_key":"!bar:example.com", "content":{"via":["example.com"]}},
+                        {"sender":"@alice:example.com","type":"m.space.child", "state_key":"!baz:example.com", "content":{"via":["example.com"]}}
+                    ],
+                    "timeline": [
+                        {"sender":"@alice:example.com","type":"m.room.join_rules", "state_key":"", "content":{"join_rule":"invite"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"A"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"B"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"C"}},
+                        {"sender":"@alice:example.com","type":"m.room.message", "content":{"body":"D"}},
+                    ],
+                    "prev_batch": "t111_222_333",
+                    "joined_count": 4,
+                    "invited_count": 0,
+                    "notification_count": 54,
+                    "highlight_count": 3
+                },
+                 // ... 99 more items
+            },
+            "extensions": {}
+        }
+    """
+
+    PATTERNS = client_patterns(
+        "/org.matrix.msc3575/sync$", releases=[], v1=False, unstable=True
+    )
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.auth = hs.get_auth()
+        self.store = hs.get_datastores().main
+        self.filtering = hs.get_filtering()
+        self.sliding_sync_handler = hs.get_sliding_sync_handler()
+
+    # TODO: Update this to `on_GET` once we figure out how we want to handle params
+    async def on_POST(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
+        requester = await self.auth.get_user_by_req(request, allow_guest=True)
+        user = requester.user
+        device_id = requester.device_id
+
+        timeout = parse_integer(request, "timeout", default=0)
+        # Position in the stream
+        from_token_string = parse_string(request, "pos")
+
+        from_token = None
+        if from_token_string is not None:
+            from_token = await StreamToken.from_string(self.store, from_token_string)
+
+        # TODO: We currently don't know whether we're going to use sticky params or
+        # maybe some filters like sync v2  where they are built up once and referenced
+        # by filter ID. For now, we will just prototype with always passing everything
+        # in.
+        body = parse_and_validate_json_object_from_request(request, SlidingSyncBody)
+        logger.info("Sliding sync request: %r", body)
+
+        sync_config = SlidingSyncConfig(
+            user=user,
+            device_id=device_id,
+            # FIXME: Currently, we're just manually copying the fields from the
+            # `SlidingSyncBody` into the config. How can we gurantee into the future
+            # that we don't forget any? I would like something more structured like
+            # `copy_attributes(from=body, to=config)`
+            lists=body.lists,
+            room_subscriptions=body.room_subscriptions,
+            extensions=body.extensions,
+        )
+
+        sliding_sync_results = await self.sliding_sync_handler.wait_for_sync_for_user(
+            requester,
+            sync_config,
+            from_token,
+            timeout,
+        )
+
+        response_content = await self.encode_response(sliding_sync_results)
+
+        return 200, response_content
+
+    # TODO: Is there a better way to encode things?
+    async def encode_response(
+        self,
+        sliding_sync_result: SlidingSyncResult,
+    ) -> JsonDict:
+        response: JsonDict = defaultdict(dict)
+
+        response["next_pos"] = await sliding_sync_result.next_pos.to_string(self.store)
+        serialized_lists = self.encode_lists(sliding_sync_result.lists)
+        if serialized_lists:
+            response["lists"] = serialized_lists
+        response["rooms"] = {}  # TODO: sliding_sync_result.rooms
+        response["extensions"] = {}  # TODO: sliding_sync_result.extensions
+
+        return response
+
+    def encode_lists(
+        self, lists: Dict[str, SlidingSyncResult.SlidingWindowList]
+    ) -> JsonDict:
+        def encode_operation(
+            operation: SlidingSyncResult.SlidingWindowList.Operation,
+        ) -> JsonDict:
+            return {
+                "op": operation.op.value,
+                "range": operation.range,
+                "room_ids": operation.room_ids,
+            }
+
+        serialized_lists = {}
+        for list_key, list_result in lists.items():
+            serialized_lists[list_key] = {
+                "count": list_result.count,
+                "ops": [encode_operation(op) for op in list_result.ops],
+            }
+
+        return serialized_lists
+
+
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     SyncRestServlet(hs).register(http_server)
 
     if hs.config.experimental.msc3575_enabled:
+        SlidingSyncRestServlet(hs).register(http_server)
         SlidingSyncE2eeRestServlet(hs).register(http_server)

synapse/server.py

@@ -109,6 +109,7 @@ from synapse.handlers.room_summary import RoomSummaryHandler
 from synapse.handlers.search import SearchHandler
 from synapse.handlers.send_email import SendEmailHandler
 from synapse.handlers.set_password import SetPasswordHandler
+from synapse.handlers.sliding_sync import SlidingSyncHandler
 from synapse.handlers.sso import SsoHandler
 from synapse.handlers.stats import StatsHandler
 from synapse.handlers.sync import SyncHandler
@@ -554,6 +555,9 @@ class HomeServer(metaclass=abc.ABCMeta):
     def get_sync_handler(self) -> SyncHandler:
         return SyncHandler(self)
 
+    def get_sliding_sync_handler(self) -> SlidingSyncHandler:
+        return SlidingSyncHandler(self)
+
     @cache_in_self
     def get_room_list_handler(self) -> RoomListHandler:
         return RoomListHandler(self)

tests/handlers/test_sliding_sync.py

@@ -0,0 +1,937 @@
+import logging
+from typing import List, Optional
+
+from twisted.test.proto_helpers import MemoryReactor
+
+from synapse.api.constants import (
+    AccountDataTypes,
+    EventContentFields,
+    EventTypes,
+    JoinRules,
+    RoomTypes,
+)
+from synapse.api.room_versions import RoomVersions
+from synapse.rest import admin
+from synapse.rest.client import knock, login, room
+from synapse.server import HomeServer
+from synapse.types import JsonDict, UserID
+from synapse.util import Clock
+
+from tests.unittest import HomeserverTestCase
+
+logger = logging.getLogger(__name__)
+
+
+class GetSyncRoomIdsForUserTestCase(HomeserverTestCase):
+    """
+    Tests Sliding Sync handler `get_sync_room_ids_for_user()` to make sure it returns
+    the correct list of rooms IDs.
+    """
+
+    servlets = [
+        admin.register_servlets,
+        knock.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        # Enable sliding sync
+        config["experimental_features"] = {"msc3575_enabled": True}
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.sliding_sync_handler = self.hs.get_sliding_sync_handler()
+        self.store = self.hs.get_datastores().main
+        self.event_sources = hs.get_event_sources()
+
+    def test_no_rooms(self) -> None:
+        """
+        Test when the user has never joined any rooms before
+        """
+        user1_id = self.register_user("user1", "pass")
+        # user1_tok = self.login(user1_id, "pass")
+
+        now_token = self.event_sources.get_current_token()
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=now_token,
+                to_token=now_token,
+            )
+        )
+
+        self.assertEqual(room_id_results, set())
+
+    def test_get_newly_joined_room(self) -> None:
+        """
+        Test that rooms that the user has newly_joined show up. newly_joined is when you
+        join after the `from_token` and <= `to_token`.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        before_room_token = self.event_sources.get_current_token()
+
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok, is_public=True)
+
+        after_room_token = self.event_sources.get_current_token()
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room_token,
+                to_token=after_room_token,
+            )
+        )
+
+        self.assertEqual(room_id_results, {room_id})
+
+    def test_get_already_joined_room(self) -> None:
+        """
+        Test that rooms that the user is already joined show up.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok, is_public=True)
+
+        after_room_token = self.event_sources.get_current_token()
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room_token,
+                to_token=after_room_token,
+            )
+        )
+
+        self.assertEqual(room_id_results, {room_id})
+
+    def test_get_invited_banned_knocked_room(self) -> None:
+        """
+        Test that rooms that the user is invited to, banned from, and knocked on show
+        up.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        before_room_token = self.event_sources.get_current_token()
+
+        # Setup the invited room (user2 invites user1 to the room)
+        invited_room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.invite(invited_room_id, targ=user1_id, tok=user2_tok)
+
+        # Setup the ban room (user2 bans user1 from the room)
+        ban_room_id = self.helper.create_room_as(
+            user2_id, tok=user2_tok, is_public=True
+        )
+        self.helper.join(ban_room_id, user1_id, tok=user1_tok)
+        self.helper.ban(ban_room_id, src=user2_id, targ=user1_id, tok=user2_tok)
+
+        # Setup the knock room (user1 knocks on the room)
+        knock_room_id = self.helper.create_room_as(
+            user2_id, tok=user2_tok, room_version=RoomVersions.V7.identifier
+        )
+        self.helper.send_state(
+            knock_room_id,
+            EventTypes.JoinRules,
+            {"join_rule": JoinRules.KNOCK},
+            tok=user2_tok,
+        )
+        # User1 knocks on the room
+        channel = self.make_request(
+            "POST",
+            "/_matrix/client/r0/knock/%s" % (knock_room_id,),
+            b"{}",
+            user1_tok,
+        )
+        self.assertEqual(200, channel.code, channel.result)
+
+        after_room_token = self.event_sources.get_current_token()
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room_token,
+                to_token=after_room_token,
+            )
+        )
+
+        # Ensure that the invited, ban, and knock rooms show up
+        self.assertEqual(
+            room_id_results,
+            {
+                invited_room_id,
+                ban_room_id,
+                knock_room_id,
+            },
+        )
+
+    def test_only_newly_left_rooms_show_up(self) -> None:
+        """
+        Test that newly_left rooms still show up in the sync response but rooms that
+        were left before the `from_token` don't show up. See condition "1)" comments in
+        the `get_sync_room_ids_for_user` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        # Leave before we calculate the `from_token`
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Leave during the from_token/to_token range (newly_left)
+        room_id2 = self.helper.create_room_as(user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        after_room2_token = self.event_sources.get_current_token()
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room1_token,
+                to_token=after_room2_token,
+            )
+        )
+
+        # Only the newly_left room should show up
+        self.assertEqual(room_id_results, {room_id2})
+
+    def test_no_joins_after_to_token(self) -> None:
+        """
+        Rooms we join after the `to_token` should *not* show up. See condition "2b)"
+        comments in the `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        before_room1_token = self.event_sources.get_current_token()
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Room join after after our `to_token` shouldn't show up
+        room_id2 = self.helper.create_room_as(user1_id, tok=user1_tok)
+        _ = room_id2
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_join_during_range_and_left_room_after_to_token(self) -> None:
+        """
+        Room still shows up if we left the room but were joined during the
+        from_token/to_token. See condition "2b)" comments in the
+        `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        before_room1_token = self.event_sources.get_current_token()
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Leave the room after we already have our tokens
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # We should still see the room because we were joined during the
+        # from_token/to_token time period.
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_join_before_range_and_left_room_after_to_token(self) -> None:
+        """
+        Room still shows up if we left the room but were joined before the `from_token`
+        so it should show up. See condition "2b)" comments in the
+        `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Leave the room after we already have our tokens
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # We should still see the room because we were joined before the `from_token`
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_newly_left_during_range_and_join_leave_after_to_token(self) -> None:
+        """
+        Newly left room should show up. But we're also testing that joining and leaving
+        after the `to_token` doesn't mess with the results. See condition "2a)" comments
+        in the `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        before_room1_token = self.event_sources.get_current_token()
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        # Join and leave the room during the from/to range
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Join and leave the room after we already have our tokens
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # Room should still show up because it's newly_left during the from/to range
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_leave_before_range_and_join_leave_after_to_token(self) -> None:
+        """
+        Old left room shouldn't show up. But we're also testing that joining and leaving
+        after the `to_token` doesn't mess with the results. See condition "2a)" comments
+        in the `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        # Join and leave the room before the from/to range
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Join and leave the room after we already have our tokens
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # Room shouldn't show up because it was left before the `from_token`
+        self.assertEqual(room_id_results, set())
+
+    def test_join_leave_multiple_times_during_range_and_after_to_token(
+        self,
+    ) -> None:
+        """
+        Join and leave multiple times shouldn't affect rooms from showing up. It just
+        matters that we were joined or newly_left in the from/to range. But we're also
+        testing that joining and leaving after the `to_token` doesn't mess with the
+        results.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        before_room1_token = self.event_sources.get_current_token()
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        # Join, leave, join back to the room before the from/to range
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Leave and Join the room multiple times after we already have our tokens
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # Room should show up because it was newly_left and joined during the from/to range
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_join_leave_multiple_times_before_range_and_after_to_token(
+        self,
+    ) -> None:
+        """
+        Join and leave multiple times before the from/to range shouldn't affect rooms
+        from showing up. It just matters that we were joined or newly_left in the
+        from/to range. But we're also testing that joining and leaving after the
+        `to_token` doesn't mess with the results.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        # Join, leave, join back to the room before the from/to range
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Leave and Join the room multiple times after we already have our tokens
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # Room should show up because we were joined before the from/to range
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_invite_before_range_and_join_leave_after_to_token(
+        self,
+    ) -> None:
+        """
+        Make it look like we joined after the token range but we were invited before the
+        from/to range so the room should still show up. See condition "2a)" comments in
+        the `get_sync_room_ids_for_user()` method.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+
+        # Invited to the room before the token
+        self.helper.invite(room_id1, src=user2_id, targ=user1_id, tok=user2_tok)
+
+        after_room1_token = self.event_sources.get_current_token()
+
+        # Join and leave the room after we already have our tokens
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=after_room1_token,
+                to_token=after_room1_token,
+            )
+        )
+
+        # Room should show up because we were invited before the from/to range
+        self.assertEqual(room_id_results, {room_id1})
+
+    def test_multiple_rooms_are_not_confused(
+        self,
+    ) -> None:
+        """
+        Test that multiple rooms are not confused as we fixup the list. This test is
+        spawning from a real world bug in the code where I was accidentally using
+        `event.room_id` in one of the fix-up loops but the `event` being referenced was
+        actually from a different loop.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # We create the room with user2 so the room isn't left with no members when we
+        # leave and can still re-join.
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        room_id2 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+
+        # Invited and left the room before the token
+        self.helper.invite(room_id1, src=user2_id, targ=user1_id, tok=user2_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        # Invited to room2
+        self.helper.invite(room_id2, src=user2_id, targ=user1_id, tok=user2_tok)
+
+        before_room3_token = self.event_sources.get_current_token()
+
+        # Invited and left room3 during the from/to range
+        room_id3 = self.helper.create_room_as(user2_id, tok=user2_tok, is_public=True)
+        self.helper.invite(room_id3, src=user2_id, targ=user1_id, tok=user2_tok)
+        self.helper.leave(room_id3, user1_id, tok=user1_tok)
+
+        after_room3_token = self.event_sources.get_current_token()
+
+        # Join and leave the room after we already have our tokens
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.leave(room_id1, user1_id, tok=user1_tok)
+        # Leave room2
+        self.helper.leave(room_id2, user1_id, tok=user1_tok)
+        # Leave room3
+        self.helper.leave(room_id3, user1_id, tok=user1_tok)
+
+        room_id_results = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=before_room3_token,
+                to_token=after_room3_token,
+            )
+        )
+
+        self.assertEqual(
+            room_id_results,
+            {
+                # `room_id1` shouldn't show up because we left before the from/to range
+                #
+                # Room should show up because we were invited before the from/to range
+                room_id2,
+                # Room should show up because it was newly_left during the from/to range
+                room_id3,
+            },
+        )
+
+
+class FilterRoomsTestCase(HomeserverTestCase):
+    """
+    Tests Sliding Sync handler `filter_rooms()` to make sure it includes/excludes rooms
+    correctly.
+    """
+
+    servlets = [
+        admin.register_servlets,
+        knock.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        # Enable sliding sync
+        config["experimental_features"] = {"msc3575_enabled": True}
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.sliding_sync_handler = self.hs.get_sliding_sync_handler()
+        self.store = self.hs.get_datastores().main
+
+    def _create_dm_room(
+        self,
+        inviter_user_id: str,
+        inviter_tok: str,
+        invitee_user_id: str,
+        invitee_tok: str,
+    ) -> str:
+        """
+        Helper to create a DM room as the "inviter" and invite the "invitee" user to the room. The
+        "invitee" user also will join the room. The `m.direct` account data will be set
+        for both users.
+        """
+
+        # Create a room and send an invite the other user
+        room_id = self.helper.create_room_as(
+            inviter_user_id,
+            is_public=False,
+            tok=inviter_tok,
+        )
+        self.helper.invite(
+            room_id,
+            src=inviter_user_id,
+            targ=invitee_user_id,
+            tok=inviter_tok,
+            extra_data={"is_direct": True},
+        )
+        # Person that was invited joins the room
+        self.helper.join(room_id, invitee_user_id, tok=invitee_tok)
+
+        # Mimic the client setting the room as a direct message in the global account
+        # data
+        self.get_success(
+            self.store.add_account_data_for_user(
+                invitee_user_id,
+                AccountDataTypes.DIRECT,
+                {inviter_user_id: [room_id]},
+            )
+        )
+        self.get_success(
+            self.store.add_account_data_for_user(
+                inviter_user_id,
+                AccountDataTypes.DIRECT,
+                {invitee_user_id: [room_id]},
+            )
+        )
+
+        return room_id
+
+    def _add_space_child(
+        self,
+        space_id: str,
+        room_id: str,
+        token: str,
+        order: Optional[str] = None,
+        via: Optional[List[str]] = None,
+    ) -> None:
+        """
+        Helper to add a child room to a space.
+        """
+
+        if via is None:
+            via = [self.hs.hostname]
+
+        content: JsonDict = {"via": via}
+        if order is not None:
+            content["order"] = order
+        self.helper.send_state(
+            space_id,
+            event_type=EventTypes.SpaceChild,
+            body=content,
+            tok=token,
+            state_key=room_id,
+        )
+
+    def test_filter_dm_rooms(self) -> None:
+        """
+        Test `filter.is_dm` for DM rooms
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # Create a normal room
+        room_id = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+
+        # Create a DM room
+        dm_room_id = self._create_dm_room(
+            inviter_user_id=user1_id,
+            inviter_tok=user1_tok,
+            invitee_user_id=user2_id,
+            invitee_tok=user2_tok,
+        )
+
+        # TODO: Better way to avoid the circular import? (see
+        # https://github.com/element-hq/synapse/pull/17187#discussion_r1619492779)
+        from synapse.handlers.sliding_sync import SlidingSyncConfig
+
+        # Try with `is_dm=True`
+        # -----------------------------
+        truthy_filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            is_dm=True,
+        )
+
+        # Try filtering the rooms
+        truthy_filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id), {room_id, dm_room_id}, truthy_filters
+            )
+        )
+
+        self.assertEqual(truthy_filtered_room_ids, {dm_room_id})
+
+        # Try with `is_dm=True`
+        # -----------------------------
+        falsy_filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            is_dm=False,
+        )
+
+        # Try filtering the rooms
+        falsy_filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id), {room_id, dm_room_id}, falsy_filters
+            )
+        )
+
+        self.assertEqual(falsy_filtered_room_ids, {room_id})
+
+    def test_filter_space_rooms(self) -> None:
+        """
+        Test `filter.spaces` for rooms in spaces
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        space_a = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            extra_content={
+                "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE}
+            },
+        )
+        space_b = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            extra_content={
+                "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE}
+            },
+        )
+        space_c = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            extra_content={
+                "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE}
+            },
+        )
+
+        room_id1 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to space_a
+        self._add_space_child(space_a, room_id1, user1_tok)
+
+        room_id2 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to space_a and space_b
+        self._add_space_child(space_a, room_id2, user1_tok)
+        self._add_space_child(space_c, room_id2, user1_tok)
+
+        room_id3 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to all spaces
+        self._add_space_child(space_a, room_id3, user1_tok)
+        self._add_space_child(space_b, room_id3, user1_tok)
+        self._add_space_child(space_c, room_id3, user1_tok)
+
+        room_id4 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to space_c
+        self._add_space_child(space_c, room_id3, user1_tok)
+
+        room_not_in_space1 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+
+        # TODO: Better way to avoid the circular import? (see
+        # https://github.com/element-hq/synapse/pull/17187#discussion_r1619492779)
+        from synapse.handlers.sliding_sync import SlidingSyncConfig
+
+        filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            spaces=[
+                space_a,
+                space_b,
+            ],
+        )
+
+        # Try filtering the rooms
+        filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id),
+                {
+                    # a
+                    room_id1,
+                    # a, c
+                    room_id2,
+                    # a, b, c
+                    room_id3,
+                    # c
+                    room_id4,
+                    # not in any space
+                    room_not_in_space1,
+                },
+                filters,
+            )
+        )
+
+        self.assertEqual(filtered_room_ids, {room_id1, room_id2, room_id3})
+
+    def test_filter_only_joined_spaces(self) -> None:
+        """
+        Test `filter.spaces` to make sure the filter only takes into account spaces we
+        are joined to.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # space_a created by user1
+        space_a = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+            extra_content={
+                "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE}
+            },
+        )
+        # space_b created by user2
+        space_b = self.helper.create_room_as(
+            user2_id,
+            tok=user2_tok,
+            extra_content={
+                "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE}
+            },
+        )
+
+        room_id1 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to space_a
+        self._add_space_child(space_a, room_id1, user1_tok)
+
+        room_id2 = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        # Add to space_b
+        self._add_space_child(space_b, room_id2, user2_tok)
+
+        # TODO: Better way to avoid the circular import? (see
+        # https://github.com/element-hq/synapse/pull/17187#discussion_r1619492779)
+        from synapse.handlers.sliding_sync import SlidingSyncConfig
+
+        filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            spaces=[
+                space_a,
+                space_b,
+            ],
+        )
+
+        # Try filtering the rooms
+        filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id),
+                {
+                    # a
+                    room_id1,
+                    # b (but user1 isn't in space_b)
+                    room_id2,
+                },
+                filters,
+            )
+        )
+
+        self.assertEqual(filtered_room_ids, {room_id1})
+
+    def test_filter_encrypted_rooms(self) -> None:
+        """
+        Test `filter.is_encrypted` for encrypted rooms
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        # Create a normal room
+        room_id = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+
+        # Create an encrypted room
+        encrypted_room_id = self.helper.create_room_as(
+            user1_id,
+            is_public=False,
+            tok=user1_tok,
+        )
+        self.helper.send_state(
+            encrypted_room_id,
+            EventTypes.RoomEncryption,
+            {"algorithm": "m.megolm.v1.aes-sha2"},
+            tok=user1_tok,
+        )
+
+        # TODO: Better way to avoid the circular import? (see
+        # https://github.com/element-hq/synapse/pull/17187#discussion_r1619492779)
+        from synapse.handlers.sliding_sync import SlidingSyncConfig
+
+        # Try with `is_encrypted=True`
+        # -----------------------------
+        truthy_filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            is_encrypted=True,
+        )
+
+        # Try filtering the rooms
+        truthy_filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id),
+                {room_id, encrypted_room_id},
+                truthy_filters,
+            )
+        )
+
+        self.assertEqual(truthy_filtered_room_ids, {encrypted_room_id})
+
+        # Try with `is_encrypted=False`
+        # -----------------------------
+        falsy_filters = SlidingSyncConfig.SlidingSyncList.Filters(
+            is_encrypted=False,
+        )
+
+        # Try filtering the rooms
+        falsy_filtered_room_ids = self.get_success(
+            self.sliding_sync_handler.filter_rooms(
+                UserID.from_string(user1_id),
+                {room_id, encrypted_room_id},
+                falsy_filters,
+            )
+        )
+
+        self.assertEqual(falsy_filtered_room_ids, {room_id})

tests/rest/client/test_sync.py

@@ -27,6 +27,7 @@ from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
 from synapse.api.constants import (
+    AccountDataTypes,
     EventContentFields,
     EventTypes,
     ReceiptTypes,
@@ -1204,3 +1205,205 @@ class ExcludeRoomTestCase(unittest.HomeserverTestCase):
 
         self.assertNotIn(self.excluded_room_id, channel.json_body["rooms"]["join"])
         self.assertIn(self.included_room_id, channel.json_body["rooms"]["join"])
+
+
+class SlidingSyncTestCase(unittest.HomeserverTestCase):
+    """
+    Tests regarding MSC3575 Sliding Sync `/sync` endpoint.
+    """
+
+    servlets = [
+        synapse.rest.admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+        devices.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        # Enable sliding sync
+        config["experimental_features"] = {"msc3575_enabled": True}
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+        self.sync_endpoint = "/_matrix/client/unstable/org.matrix.msc3575/sync"
+
+    def _create_dm_room(
+        self,
+        inviter_user_id: str,
+        inviter_tok: str,
+        invitee_user_id: str,
+        invitee_tok: str,
+    ) -> str:
+        """
+        Helper to create a DM room as the "inviter" and invite the "invitee" user to the
+        room. The "invitee" user also will join the room. The `m.direct` account data
+        will be set for both users.
+        """
+
+        # Create a room and send an invite the other user
+        room_id = self.helper.create_room_as(
+            inviter_user_id,
+            is_public=False,
+            tok=inviter_tok,
+        )
+        self.helper.invite(
+            room_id,
+            src=inviter_user_id,
+            targ=invitee_user_id,
+            tok=inviter_tok,
+            extra_data={"is_direct": True},
+        )
+        # Person that was invited joins the room
+        self.helper.join(room_id, invitee_user_id, tok=invitee_tok)
+
+        # Mimic the client setting the room as a direct message in the global account
+        # data
+        self.get_success(
+            self.store.add_account_data_for_user(
+                invitee_user_id,
+                AccountDataTypes.DIRECT,
+                {inviter_user_id: [room_id]},
+            )
+        )
+        self.get_success(
+            self.store.add_account_data_for_user(
+                inviter_user_id,
+                AccountDataTypes.DIRECT,
+                {invitee_user_id: [room_id]},
+            )
+        )
+
+        return room_id
+
+    def test_sync_list(self) -> None:
+        """
+        Test that room IDs show up in the Sliding Sync lists
+        """
+        alice_user_id = self.register_user("alice", "correcthorse")
+        alice_access_token = self.login(alice_user_id, "correcthorse")
+
+        room_id = self.helper.create_room_as(
+            alice_user_id, tok=alice_access_token, is_public=True
+        )
+
+        # Make the Sliding Sync request
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint,
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 99]],
+                        "sort": ["by_notification_level", "by_recency", "by_name"],
+                        "required_state": [
+                            ["m.room.join_rules", ""],
+                            ["m.room.history_visibility", ""],
+                            ["m.space.child", "*"],
+                        ],
+                        "timeline_limit": 1,
+                    }
+                }
+            },
+            access_token=alice_access_token,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Make sure it has the foo-list we requested
+        self.assertListEqual(
+            list(channel.json_body["lists"].keys()),
+            ["foo-list"],
+            channel.json_body["lists"].keys(),
+        )
+
+        # Make sure the list includes the room we are joined to
+        self.assertListEqual(
+            list(channel.json_body["lists"]["foo-list"]["ops"]),
+            [
+                {
+                    "op": "SYNC",
+                    "range": [0, 99],
+                    "room_ids": [room_id],
+                }
+            ],
+            channel.json_body["lists"]["foo-list"],
+        )
+
+    def test_filter_list(self) -> None:
+        """
+        Test that filters apply to lists
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # Create a DM room
+        dm_room_id = self._create_dm_room(
+            inviter_user_id=user1_id,
+            inviter_tok=user1_tok,
+            invitee_user_id=user2_id,
+            invitee_tok=user2_tok,
+        )
+
+        # Create a normal room
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok, is_public=True)
+
+        # Make the Sliding Sync request
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint,
+            {
+                "lists": {
+                    "dms": {
+                        "ranges": [[0, 99]],
+                        "sort": ["by_recency"],
+                        "required_state": [],
+                        "timeline_limit": 1,
+                        "filters": {"is_dm": True},
+                    },
+                    "foo-list": {
+                        "ranges": [[0, 99]],
+                        "sort": ["by_recency"],
+                        "required_state": [],
+                        "timeline_limit": 1,
+                        "filters": {"is_dm": False},
+                    },
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Make sure it has the foo-list we requested
+        self.assertListEqual(
+            list(channel.json_body["lists"].keys()),
+            ["dms", "foo-list"],
+            channel.json_body["lists"].keys(),
+        )
+
+        # Make sure the list includes the room we are joined to
+        self.assertListEqual(
+            list(channel.json_body["lists"]["dms"]["ops"]),
+            [
+                {
+                    "op": "SYNC",
+                    "range": [0, 99],
+                    "room_ids": [dm_room_id],
+                }
+            ],
+            list(channel.json_body["lists"]["dms"]),
+        )
+        self.assertListEqual(
+            list(channel.json_body["lists"]["foo-list"]["ops"]),
+            [
+                {
+                    "op": "SYNC",
+                    "range": [0, 99],
+                    "room_ids": [room_id],
+                }
+            ],
+            list(channel.json_body["lists"]["foo-list"]),
+        )