diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py index 62f55416b5..ee20e3d6c8 100644 --- a/synapse/api/filtering.py +++ b/synapse/api/filtering.py @@ -227,7 +227,7 @@ class FilterCollection: self.event_fields = filter_json.get("event_fields", []) self.event_format = filter_json.get("event_format", "client") - self.profile_fields: list[str] = [] + self.profile_fields: set[str] = set() if hs.config.experimental.msc4429_enabled: profile_fields_filter = filter_json.get("org.matrix.msc4429.profile_fields") @@ -235,7 +235,7 @@ class FilterCollection: ids = profile_fields_filter.get("ids", []) if ids is None: ids = [] - self.profile_fields = list(ids) + self.profile_fields = set(ids) def __repr__(self) -> str: return "" % (json.dumps(self._filter_json),) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 4dd0f34750..19c63945c8 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -2142,7 +2142,7 @@ class SyncHandler: *, user_id: str, sync_result_builder: "SyncResultBuilder", - profile_fields: list[str], + profile_fields: set[str], include_users: set[str] | None, ) -> None: """ @@ -2159,21 +2159,35 @@ class SyncHandler: for when we have calculated a list of users in our lazy loading sync and want to only return those. """ - # Currently, limited to only local profiles, so filter remote servers out - user_ids = await self.store.get_local_users_who_share_room_with_user(user_id) + updates = await self.store.get_profile_updates_for_user_and_fields( + from_id=0, + to_id=sync_result_builder.now_token.profile_updates_key, + user_id=user_id, + field_names=profile_fields, + include_users=include_users, + ) - if include_users: - # Filter down to selected included users - user_ids = {user_id for user_id in user_ids if user_id in include_users} + user_fields: dict[str, set[str]] = {} + for update in updates: + if ( + update.action != ProfileUpdateAction.UPDATE.value + # TODO: When would field_name be None? + or update.field_name is None + ): + continue - if not user_ids: + user_fields.setdefault(update.user_id, set()).add(update.field_name) + + if not user_fields: return - profile_data_by_user = await self.store.get_profile_data_for_users(user_ids) + profile_data_by_user = await self.store.get_profile_data_for_users( + user_fields.keys() + ) # Serialise the profile updates into the sync response format. profile_updates: dict[str, dict[str, JsonValue | None]] = {} - for other_user_id in user_ids: + for other_user_id, fields in user_fields.items(): profile_data = profile_data_by_user.get(other_user_id) if profile_data is None: # Don't generate anything for users with no profile data @@ -2181,7 +2195,7 @@ class SyncHandler: continue per_user_updates: dict[str, JsonValue] = {} - for field_name in profile_fields: + for field_name in fields: if profile_data.get(field_name): per_user_updates[field_name] = cast( JsonValue, profile_data[field_name] @@ -2247,42 +2261,23 @@ class SyncHandler: if since_token.profile_updates_key == now_token.profile_updates_key: return - updates = await self.store.get_profile_updates_for_fields( - from_id=since_token.profile_updates_key, - to_id=now_token.profile_updates_key, - field_names=profile_fields, - ) - interesting_updates = await self.store.get_profile_updates_per_user_for_user( + updates = await self.store.get_profile_updates_for_user_and_fields( from_id=since_token.profile_updates_key, to_id=now_token.profile_updates_key, user_id=user_id, + field_names=profile_fields, ) - if include_users: - # Further filter down to selected included users - updates = [update for update in updates if update.user_id in include_users] left_room_user_ids = { update.user_id for update in updates if update.action == ProfileUpdateAction.LEFT_ROOM.value } - # Only include updates we've got for us specifically for field updates - updates = [ - update for update in updates if update.stream_id in interesting_updates - ] updated_user_ids = { update.user_id for update in updates if update.action == ProfileUpdateAction.UPDATE.value } - no_longer_sharing_rooms_user_ids: set[str] = set() - if left_room_user_ids: - shared_left_user_ids = await self.store.do_users_share_a_room( - user_id, left_room_user_ids - ) - no_longer_sharing_rooms_user_ids = set(left_room_user_ids) - set( - shared_left_user_ids - ) if not updated_user_ids and not left_room_user_ids: return @@ -2293,7 +2288,6 @@ class SyncHandler: # Process field updates if updated_user_ids: - updated_user_ids.add(user_id) user_fields: dict[str, set[str]] = {} for update in updates: if not update.field_name or update.user_id not in updated_user_ids: @@ -2331,8 +2325,8 @@ class SyncHandler: profile_updates[other_user_id] = per_user_updates # Process left rooms - if no_longer_sharing_rooms_user_ids: - for other_user_id in no_longer_sharing_rooms_user_ids: + if left_room_user_ids: + for other_user_id in left_room_user_ids: # Return an empty dictionary to the client profile_updates[other_user_id] = {} diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index c7f8aea48f..3a8a54da5e 100644 --- a/synapse/storage/databases/main/profile.py +++ b/synapse/storage/databases/main/profile.py @@ -411,7 +411,7 @@ class ProfileWorkerStore(SQLBaseStore): " OR action != ?) " " ORDER BY stream_id ASC" ) - txn.execute(sql, (from_id, to_id, ProfileUpdateAction.UPDATE.value, *args)) + txn.execute(sql, (from_id, to_id, *args, ProfileUpdateAction.UPDATE.value)) rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall()) updates: list[ProfileUpdate] = [] @@ -431,6 +431,104 @@ class ProfileWorkerStore(SQLBaseStore): "get_profile_updates_for_fields", _get_profile_updates_for_fields_txn ) + async def get_profile_updates_for_user_and_fields( + self, + *, + from_id: int, + to_id: int, + user_id: str, + field_names: set[str], + include_users: set[str] | None = None, + ) -> list[ProfileUpdate]: + """Get profile update markers for a user in a stream range. + + The returned profile update rows are restricted to those with a + corresponding `profile_updates_per_user` row for the syncing user. + + Bounds: from_id < ... <= to_id + + Args: + from_id: The starting stream ID (exclusive). + to_id: The ending stream ID (inclusive). + user_id: The full user ID to filter on. + field_names: Set of field names to filter update actions against. + include_users: If given, only include updates for these user IDs. + + Returns: + A list of ProfileUpdates update rows. + """ + if from_id == to_id: + return [] + + if len(field_names) == 0: + return [] + + if include_users is not None and len(include_users) == 0: + # All updates have been filtered out by lazy-loading. + return [] + + def _get_profile_updates_for_user_and_fields_txn( + txn: LoggingTransaction, + ) -> list[ProfileUpdate]: + field_clause, field_args = make_in_list_sql_clause( + txn.database_engine, "pu.field_name", field_names + ) + user_clause = "" + user_args: list[str] = [] + if include_users is not None: + # Filter out rows that aren't in `include_users`, if defined. + # This is only relevant when lazy-loading. + user_clause, user_args = make_in_list_sql_clause( + txn.database_engine, "pu.user_id", include_users + ) + user_clause = f"AND {user_clause}" + + # Retrieve profile updates where there's a corresponding row in + # `profile_updates_per_user` within the given `stream_id` bounds + # and the `user_id` and `field_names` match. + sql = f""" + SELECT pu.stream_id, pu.user_id, pu.action, pu.field_name + FROM profile_updates AS pu + INNER JOIN profile_updates_per_user AS puf + ON pu.stream_id = puf.stream_id + WHERE ? < pu.stream_id AND pu.stream_id <= ? + AND puf.user_id = ? + {user_clause} + AND ({field_clause} OR pu.action != ?) + ORDER BY pu.stream_id ASC + """ + + txn.execute( + sql, + ( + from_id, + to_id, + user_id, + *user_args, + *field_args, + ProfileUpdateAction.UPDATE.value, + ), + ) + rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall()) + + updates: list[ProfileUpdate] = [] + for stream_id, updated_user_id, action, field_name in rows: + updates.append( + ProfileUpdate( + stream_id=stream_id, + user_id=updated_user_id, + action=action, + field_name=field_name, + ) + ) + + return updates + + return await self.db_pool.runInteraction( + "get_profile_updates_for_user_and_fields", + _get_profile_updates_for_user_and_fields_txn, + ) + async def get_profile_data_for_users( self, user_ids: Collection[str] ) -> dict[str, dict[str, str | JsonDict | None]]: