diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 19280a2edf..dc70d8cf12 100644 --- a/synapse/storage/databases/main/events.py +++ b/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( diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index af0daf55d1..b87e99f92a 100644 --- a/synapse/storage/databases/main/profile.py +++ b/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( diff --git a/synapse/storage/databases/main/roommember.py b/synapse/storage/databases/main/roommember.py index cafdf315b0..952de88908 100644 --- a/synapse/storage/databases/main/roommember.py +++ b/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 )