Cleanup and docstring updates

synapse/storage/databases/main/events.py

@@ -2159,22 +2159,25 @@ class PersistEventsStore:
         room_id: str,
     ) -> None:
         """
-        Record updates into the profile updates stream for when a user leaves
-        the last room with a set of users.
+        Record updates into the profile updates stream for when a user leaves a room.
 
-        Doing this clears all old rows from the `profile_updates_per_user` table,
-        to avoid exposing any profile field changes past the point of not being
-        in any common rooms with the user.
+        If this was the last shared room with a set of users, clear all old rows from
+        the `profile_updates_per_user` table relating to those users, to avoid exposing
+        any profile field changes past the point of not being in any common rooms with
+        the user.
+
+        Currently, updates are only recorded for local users.
 
         Note, this method lives here in the events store file due to the profile
-        store not having access to the membership store, which we need to re-use
-        the `do_users_share_a_room_txn` method there.
+        store not having access to the membership store (which the events store does),
+        which we need to re-use the `do_users_share_a_room_txn` method there.
 
         Args:
-            user_id: The user who left the last common room with a set of users.
-            users_to_update: The set of users who no longer share rooms with the user.
+            user_id: The user who left the room.
+            room_id: The room that was left.
         """
-        rows = self.db_pool.simple_select_onecol_txn(
+        # Get the local members of the room
+        room_members = self.db_pool.simple_select_onecol_txn(
             txn=txn,
             table="local_current_membership",
             retcol="user_id",
@@ -2187,9 +2190,11 @@ class PersistEventsStore:
         users_sharing_rooms = self.store.do_users_share_a_room_txn(
             txn=txn,
             user_id=user_id.to_string(),
-            other_user_ids=set(rows),
+            other_user_ids=set(room_members),
+        )
+        users_no_longer_sharing_rooms = set(room_members) - set(
+            users_sharing_rooms.keys()
         )
-        users_no_longer_sharing_rooms = set(rows) - set(users_sharing_rooms.keys())
 
         # First clear the previous rows from the table
         user_clause, user_args = make_in_list_sql_clause(

synapse/storage/databases/main/profile.py

@@ -768,29 +768,31 @@ class ProfileWorkerStore(SQLBaseStore):
         self, *, txn: LoggingTransaction, joined_users: set[str]
     ) -> None:
         """
-        Record profile updates from membership changes.
-
-        Note, currently only local user changes are reported.
+        Record profile updates from membership join events.
 
         Adds "joined_room" and "left_room" profile update actions into the
         profile updates stream tables.
 
+        Currently, updates are only recorded for local users.
+
         Args:
             txn: The transaction to use.
-            changes: A list of changes to room membership state, a tuple of
-                "user ID" and "membership".
+            joined_users: A list of users who have joined.
         """
         if not self._msc4429_enabled:
             return
 
         assert self._is_events_writer
 
-        # Get rooms for all the users in the changes
+        # Ensure we're working with local users only
+        users = {user_id for user_id in joined_users if self.hs.is_mine_id(user_id)}
+
+        # Get rooms for all the users in the changes in one go to save on queries.
         rows = self.db_pool.simple_select_many_txn(
             txn=txn,
             table="current_state_events",
             column="state_key",
-            iterable=joined_users,
+            iterable=users,
             retcols=(
                 "state_key",
                 "room_id",
@@ -800,11 +802,12 @@ class ProfileWorkerStore(SQLBaseStore):
                 "membership": Membership.JOIN,
             },
         )
-        user_rooms: dict[str, set[str]] = {user_id: set() for user_id in joined_users}
+        user_rooms: dict[str, set[str]] = {user_id: set() for user_id in users}
         for state_key, room_id in rows:
             user_rooms[state_key].add(room_id)
 
-        for user_id in joined_users:
+        # Record the profile updates for each user
+        for user_id in users:
             self.record_profile_updates_txn(
                 txn=txn,
                 user_id=UserID.from_string(user_id),
@@ -820,19 +823,25 @@ class ProfileWorkerStore(SQLBaseStore):
         user_id: UserID,
         action: ProfileUpdateAction,
         field_names: list[str] | None,
-        target_users: set[str] | None = None,
         user_rooms: set[str] | None = None,
+        target_users: set[str] | None = None,
     ) -> int | None:
         """
         Record updates into the profile updates stream tables.
 
+        Currently, updates are only recorded for local users.
+
         Args:
             txn: Transaction to use
             user_id: User ID that made the profile update
             action: The profile update action, either `update`, `left_room` or
                 `joined_room`
             field_names: A list of fields that were set, if ProfileUpdateAction.UPDATE
-            target_users: Set of users to create profile update stream rows for
+            user_rooms: Optionally, a set of rooms that the update concerns. If not
+                given, a database lookup will be done to fetch all the users rooms.
+            target_users: Optionally, set of users to create profile update stream rows
+                for. If not given, a database lookup will be done based on `user_rooms`,
+                or if that is not set, the result of the rooms lookup.
 
         Returns:
             The latest stream ID created in this transaction
@@ -870,15 +879,19 @@ class ProfileWorkerStore(SQLBaseStore):
                 },
             )
             target_users = {row[0] for row in rows}
+
+        # Ensure we only write updates for local users
+        users = {user for user in target_users if self.hs.is_mine_id(user)}
+
         if action in (ProfileUpdateAction.JOINED_ROOM, ProfileUpdateAction.LEFT_ROOM):
-            target_users.discard(user_id.to_string())
-            if not target_users:
+            users.discard(user_id.to_string())
+            if not users:
                 # No point writing an update for ourselves, if a membership change and no
                 # other users interested
                 return None
         elif action == ProfileUpdateAction.UPDATE:
             # Always include ourselves when updating field values
-            target_users.add(user_id.to_string())
+            users.add(user_id.to_string())
 
         # Record the profile update
         inserted_ts = self.clock.time_msec()
@@ -927,7 +940,7 @@ class ProfileWorkerStore(SQLBaseStore):
         inserted_ts = self.clock.time_msec()
         per_user_values = [
             (stream_id, user_id, inserted_ts)
-            for user_id in target_users
+            for user_id in users
             for stream_id in stream_ids
         ]
         self.db_pool.simple_insert_many_txn(

synapse/storage/databases/main/roommember.py

@@ -858,6 +858,14 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
         other_user_ids: Collection[str],
         exclude_room_id: str | None = None,
     ) -> dict[str, bool]:
+        """Return mapping from user ID to whether they share a room with the
+        given user.
+
+        Optionally, exclude a room when querying the database.
+
+        Note: `None` and `False` are equivalent and mean they don't share a
+        room.
+        """
         clause, args = make_in_list_sql_clause(
             self.database_engine, "state_key", other_user_ids
         )