Extend sync to inform clients about the progress of their knocks

So we've got federation so that homeservers can communicate knocking
information between them - but how does that information actually get
down to the client? The client knows that it knocked successfully from a
200 in its original request, but what else does it need? This commit
adds a new "knock" section to /sync (in addition to "invite", "join",
and "leave") all help give the client the information it needs.

The new "knock" section is used for sending down the stripped state
events we collected earlier. The client will use these to display the
room and its metadata in a little "pending knocks" section or similar.

This is all this commit adds. If the user's knock has been accepted or
rejected, they will receive that information in the "join" or "leave"
sections of /sync.

Most of this code is just cribbing off the invite and join sync code yet
again, with some minor differences. For instance, we don't need to
exclude knock events from sync if the sender is in your ignore list, as
you are the only ones that can send knocks for yourself.

The structure of the "knock" dict in sync is modeled after "invite", as
clients also receive stripped state in that. The structure can be viewed
in the linked MSC.

3 files changed, 127 insertions, 23 deletions

diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index 14f7f1156f..589fa877d2 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -228,6 +228,7 @@ def format_event_for_client_v1(d):
         "replaces_state",
         "prev_content",
         "invite_room_state",
+        "knock_room_state",
     )
     for key in copy_keys:
         if key in d["unsigned"]:
@@ -311,6 +312,10 @@ def serialize_event(
     # If this is an invite for somebody else, then we don't care about the
     # invite_room_state as that's meant solely for the invitee. Other clients
     # will already have the state since they're in the room.
+    #
+    # Note that is_invite only seems to be set to False in the case of appservices
+    # that are not interested in the target user of the invite.
+    # TODO: Figure out what to do here w.r.t knocking
     if not is_invite:
         d["unsigned"].pop("invite_room_state", None)
 
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 32e53c2d25..71685cc9f6 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -150,6 +150,16 @@ class InvitedSyncResult:
 
 
 @attr.s(slots=True, frozen=True)
+class KnockedSyncResult:
+    room_id = attr.ib(type=str)
+    knock = attr.ib(type=EventBase)
+
+    def __bool__(self) -> bool:
+        """Knocked rooms should always be reported to the client"""
+        return True
+
+
+@attr.s(slots=True, frozen=True)
 class GroupsSyncResult:
     join = attr.ib(type=JsonDict)
     invite = attr.ib(type=JsonDict)
@@ -182,6 +192,7 @@ class _RoomChanges:
 
     room_entries = attr.ib(type=List["RoomSyncResultBuilder"])
     invited = attr.ib(type=List[InvitedSyncResult])
+    knocked = attr.ib(type=List[KnockedSyncResult])
     newly_joined_rooms = attr.ib(type=List[str])
     newly_left_rooms = attr.ib(type=List[str])
 
@@ -195,6 +206,7 @@ class SyncResult:
         account_data: List of account_data events for the user.
         joined: JoinedSyncResult for each joined room.
         invited: InvitedSyncResult for each invited room.
+        knocked: KnockedSyncResult for each knocked on room.
         archived: ArchivedSyncResult for each archived room.
         to_device: List of direct messages for the device.
         device_lists: List of user_ids whose devices have changed
@@ -210,6 +222,7 @@ class SyncResult:
     account_data = attr.ib(type=List[JsonDict])
     joined = attr.ib(type=List[JoinedSyncResult])
     invited = attr.ib(type=List[InvitedSyncResult])
+    knocked = attr.ib(type=List[KnockedSyncResult])
     archived = attr.ib(type=List[ArchivedSyncResult])
     to_device = attr.ib(type=List[JsonDict])
     device_lists = attr.ib(type=DeviceLists)
@@ -226,6 +239,7 @@ class SyncResult:
             self.presence
             or self.joined
             or self.invited
+            or self.knocked
             or self.archived
             or self.account_data
             or self.to_device
@@ -997,7 +1011,7 @@ class SyncHandler:
         res = await self._generate_sync_entry_for_rooms(
             sync_result_builder, account_data_by_room
         )
-        newly_joined_rooms, newly_joined_or_invited_users, _, _ = res
+        newly_joined_rooms, newly_joined_or_invited_or_knocking_users, _, _ = res
         _, _, newly_left_rooms, newly_left_users = res
 
         block_all_presence_data = (
@@ -1006,7 +1020,9 @@ class SyncHandler:
         if self.hs_config.use_presence and not block_all_presence_data:
             logger.debug("Fetching presence data")
             await self._generate_sync_entry_for_presence(
-                sync_result_builder, newly_joined_rooms, newly_joined_or_invited_users
+                sync_result_builder,
+                newly_joined_rooms,
+                newly_joined_or_invited_or_knocking_users,
             )
 
         logger.debug("Fetching to-device data")
@@ -1015,7 +1031,7 @@ class SyncHandler:
         device_lists = await self._generate_sync_entry_for_device_list(
             sync_result_builder,
             newly_joined_rooms=newly_joined_rooms,
-            newly_joined_or_invited_users=newly_joined_or_invited_users,
+            newly_joined_or_invited_or_knocking_users=newly_joined_or_invited_or_knocking_users,
             newly_left_rooms=newly_left_rooms,
             newly_left_users=newly_left_users,
         )
@@ -1049,6 +1065,7 @@ class SyncHandler:
             account_data=sync_result_builder.account_data,
             joined=sync_result_builder.joined,
             invited=sync_result_builder.invited,
+            knocked=sync_result_builder.knocked,
             archived=sync_result_builder.archived,
             to_device=sync_result_builder.to_device,
             device_lists=device_lists,
@@ -1108,7 +1125,7 @@ class SyncHandler:
         self,
         sync_result_builder: "SyncResultBuilder",
         newly_joined_rooms: Set[str],
-        newly_joined_or_invited_users: Set[str],
+        newly_joined_or_invited_or_knocking_users: Set[str],
         newly_left_rooms: Set[str],
         newly_left_users: Set[str],
     ) -> DeviceLists:
@@ -1117,8 +1134,9 @@ class SyncHandler:
         Args:
             sync_result_builder
             newly_joined_rooms: Set of rooms user has joined since previous sync
-            newly_joined_or_invited_users: Set of users that have joined or
-                been invited to a room since previous sync.
+            newly_joined_or_invited_or_knocking_users: Set of users that have joined,
+                been invited to a room or are knocking on a room since
+                previous sync.
             newly_left_rooms: Set of rooms user has left since previous sync
             newly_left_users: Set of users that have left a room we're in since
                 previous sync
@@ -1129,7 +1147,9 @@ class SyncHandler:
 
         # We're going to mutate these fields, so lets copy them rather than
         # assume they won't get used later.
-        newly_joined_or_invited_users = set(newly_joined_or_invited_users)
+        newly_joined_or_invited_or_knocking_users = set(
+            newly_joined_or_invited_or_knocking_users
+        )
         newly_left_users = set(newly_left_users)
 
         if since_token and since_token.device_list_key:
@@ -1168,11 +1188,11 @@ class SyncHandler:
             # Step 1b, check for newly joined rooms
             for room_id in newly_joined_rooms:
                 joined_users = await self.state.get_current_users_in_room(room_id)
-                newly_joined_or_invited_users.update(joined_users)
+                newly_joined_or_invited_or_knocking_users.update(joined_users)
 
             # TODO: Check that these users are actually new, i.e. either they
             # weren't in the previous sync *or* they left and rejoined.
-            users_that_have_changed.update(newly_joined_or_invited_users)
+            users_that_have_changed.update(newly_joined_or_invited_or_knocking_users)
 
             user_signatures_changed = await self.store.get_users_whose_signatures_changed(
                 user_id, since_token.device_list_key
@@ -1417,6 +1437,7 @@ class SyncHandler:
 
         room_entries = room_changes.room_entries
         invited = room_changes.invited
+        knocked = room_changes.knocked
         newly_joined_rooms = room_changes.newly_joined_rooms
         newly_left_rooms = room_changes.newly_left_rooms
 
@@ -1437,9 +1458,10 @@ class SyncHandler:
         await concurrently_execute(handle_room_entries, room_entries, 10)
 
         sync_result_builder.invited.extend(invited)
+        sync_result_builder.knocked.extend(knocked)
 
-        # Now we want to get any newly joined or invited users
-        newly_joined_or_invited_users = set()
+        # Now we want to get any newly joined, invited or knocking users
+        newly_joined_or_invited_or_knocking_users = set()
         newly_left_users = set()
         if since_token:
             for joined_sync in sync_result_builder.joined:
@@ -1451,19 +1473,22 @@ class SyncHandler:
                         if (
                             event.membership == Membership.JOIN
                             or event.membership == Membership.INVITE
+                            or event.membership == Membership.KNOCK
                         ):
-                            newly_joined_or_invited_users.add(event.state_key)
+                            newly_joined_or_invited_or_knocking_users.add(
+                                event.state_key
+                            )
                         else:
                             prev_content = event.unsigned.get("prev_content", {})
                             prev_membership = prev_content.get("membership", None)
                             if prev_membership == Membership.JOIN:
                                 newly_left_users.add(event.state_key)
 
-        newly_left_users -= newly_joined_or_invited_users
+        newly_left_users -= newly_joined_or_invited_or_knocking_users
 
         return (
             set(newly_joined_rooms),
-            newly_joined_or_invited_users,
+            newly_joined_or_invited_or_knocking_users,
             set(newly_left_rooms),
             newly_left_users,
         )
@@ -1519,6 +1544,7 @@ class SyncHandler:
         newly_left_rooms = []
         room_entries = []
         invited = []
+        knocked = []
         for room_id, events in mem_change_events_by_room_id.items():
             logger.debug(
                 "Membership changes in %s: [%s]",
@@ -1598,9 +1624,17 @@ class SyncHandler:
             should_invite = non_joins[-1].membership == Membership.INVITE
             if should_invite:
                 if event.sender not in ignored_users:
-                    room_sync = InvitedSyncResult(room_id, invite=non_joins[-1])
-                    if room_sync:
-                        invited.append(room_sync)
+                    invite_room_sync = InvitedSyncResult(room_id, invite=non_joins[-1])
+                    if invite_room_sync:
+                        invited.append(invite_room_sync)
+
+            # Only bother if our latest membership in the room is knock (and we haven't
+            # been accepted/rejected in the meantime).
+            should_knock = non_joins[-1].membership == Membership.KNOCK
+            if should_knock:
+                knock_room_sync = KnockedSyncResult(room_id, knock=non_joins[-1])
+                if knock_room_sync:
+                    knocked.append(knock_room_sync)
 
             # Always include leave/ban events. Just take the last one.
             # TODO: How do we handle ban -> leave in same batch?
@@ -1704,7 +1738,9 @@ class SyncHandler:
                 )
             room_entries.append(entry)
 
-        return _RoomChanges(room_entries, invited, newly_joined_rooms, newly_left_rooms)
+        return _RoomChanges(
+            room_entries, invited, knocked, newly_joined_rooms, newly_left_rooms,
+        )
 
     async def _get_all_rooms(
         self, sync_result_builder: "SyncResultBuilder", ignored_users: FrozenSet[str]
@@ -1724,6 +1760,7 @@ class SyncHandler:
 
         membership_list = (
             Membership.INVITE,
+            Membership.KNOCK,
             Membership.JOIN,
             Membership.LEAVE,
             Membership.BAN,
@@ -1735,6 +1772,7 @@ class SyncHandler:
 
         room_entries = []
         invited = []
+        knocked = []
 
         for event in room_list:
             if event.membership == Membership.JOIN:
@@ -1754,8 +1792,11 @@ class SyncHandler:
                     continue
                 invite = await self.store.get_event(event.event_id)
                 invited.append(InvitedSyncResult(room_id=event.room_id, invite=invite))
+            elif event.membership == Membership.KNOCK:
+                knock = await self.store.get_event(event.event_id)
+                knocked.append(KnockedSyncResult(room_id=event.room_id, knock=knock))
             elif event.membership in (Membership.LEAVE, Membership.BAN):
-                # Always send down rooms we were banned or kicked from.
+                # Always send down rooms we were banned from or kicked from.
                 if not sync_config.filter_collection.include_leave:
                     if event.membership == Membership.LEAVE:
                         if user_id == event.sender:
@@ -1776,7 +1817,7 @@ class SyncHandler:
                     )
                 )
 
-        return _RoomChanges(room_entries, invited, [], [])
+        return _RoomChanges(room_entries, invited, knocked, [], [])
 
     async def _generate_room_entry(
         self,
@@ -2065,6 +2106,7 @@ class SyncResultBuilder:
         account_data (list)
         joined (list[JoinedSyncResult])
         invited (list[InvitedSyncResult])
+        knocked (list[KnockedSyncResult])
         archived (list[ArchivedSyncResult])
         groups (GroupsSyncResult|None)
         to_device (list)
@@ -2080,6 +2122,7 @@ class SyncResultBuilder:
     account_data = attr.ib(type=List[JsonDict], default=attr.Factory(list))
     joined = attr.ib(type=List[JoinedSyncResult], default=attr.Factory(list))
     invited = attr.ib(type=List[InvitedSyncResult], default=attr.Factory(list))
+    knocked = attr.ib(type=List[KnockedSyncResult], default=attr.Factory(list))
     archived = attr.ib(type=List[ArchivedSyncResult], default=attr.Factory(list))
     groups = attr.ib(type=Optional[GroupsSyncResult], default=None)
     to_device = attr.ib(type=List[JsonDict], default=attr.Factory(list))
diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py
index 2b84eb89c0..2c9c5add72 100644
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -15,6 +15,7 @@
 
 import itertools
 import logging
+from typing import Any, Callable, Dict, List
 
 from synapse.api.constants import PresenceState
 from synapse.api.errors import Codes, StoreError, SynapseError
@@ -24,7 +25,7 @@ from synapse.events.utils import (
     format_event_raw,
 )
 from synapse.handlers.presence import format_user_presence_state
-from synapse.handlers.sync import SyncConfig
+from synapse.handlers.sync import KnockedSyncResult, SyncConfig
 from synapse.http.servlet import RestServlet, parse_boolean, parse_integer, parse_string
 from synapse.types import StreamToken
 from synapse.util import json_decoder
@@ -212,6 +213,10 @@ class SyncRestServlet(RestServlet):
             sync_result.invited, time_now, access_token_id, event_formatter
         )
 
+        knocked = await self.encode_knocked(
+            sync_result.knocked, time_now, access_token_id, event_formatter
+        )
+
         archived = await self.encode_archived(
             sync_result.archived,
             time_now,
@@ -229,7 +234,12 @@ class SyncRestServlet(RestServlet):
                 "left": list(sync_result.device_lists.left),
             },
             "presence": SyncRestServlet.encode_presence(sync_result.presence, time_now),
-            "rooms": {"join": joined, "invite": invited, "leave": archived},
+            "rooms": {
+                "join": joined,
+                "invite": invited,
+                "knock": knocked,
+                "leave": archived,
+            },
             "groups": {
                 "join": sync_result.groups.join,
                 "invite": sync_result.groups.invite,
@@ -295,7 +305,7 @@ class SyncRestServlet(RestServlet):
 
         Args:
             rooms(list[synapse.handlers.sync.InvitedSyncResult]): list of
-                sync results for rooms this user is joined to
+                sync results for rooms this user is invited to
             time_now(int): current time - used as a baseline for age
                 calculations
             token_id(int): ID of the user's auth token - used for namespacing
@@ -324,6 +334,52 @@ class SyncRestServlet(RestServlet):
 
         return invited
 
+    async def encode_knocked(
+        self,
+        rooms: List[KnockedSyncResult],
+        time_now: int,
+        token_id: int,
+        event_formatter: Callable[[Dict], Dict],
+    ) -> Dict[str, Dict[str, Any]]:
+        """
+        Encode the rooms we've knocked on in a sync result.
+
+        Args:
+            rooms: list of sync results for rooms this user is knocking on
+            time_now: current time - used as a baseline for age calculations
+            token_id: ID of the user's auth token - used for namespacing of transaction IDs
+            event_formatter: function to convert from federation format to client format
+
+        Returns:
+            The list of rooms the user has knocked on, in our response format.
+        """
+        knocked = {}
+        for room in rooms:
+            knock = await self._event_serializer.serialize_event(
+                room.knock, time_now, token_id=token_id, event_format=event_formatter,
+            )
+
+            # Extract the `unsigned` key from the knock event.
+            # This is where we (cheekily) store the knock state events
+            unsigned = knock.setdefault("unsigned", {})
+
+            # Extract the stripped room state from the unsigned dict
+            # This is for clients to get a little bit of information about
+            # the room they've knocked on, without revealing any sensitive information
+            knocked_state = list(unsigned.pop("knock_room_state", []))
+
+            # Append the actual knock membership event itself as well
+            # TODO: I *believe* this is just for the client's sake of track its membership
+            # state in each room, but I could be wrong. This certainly doesn't seem like it
+            # could have any negative effects besides resource usage
+            knocked_state.append(knock)
+
+            # Build the `knock_state` dictionary, which will contain the state of the
+            # room that the client has knocked on
+            knocked[room.room_id] = {"knock_state": {"events": knocked_state}}
+
+        return knocked
+
     async def encode_archived(
         self, rooms, time_now, token_id, event_fields, event_formatter
     ):