diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 6855d7a708..5805ccfd55 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -25,7 +25,12 @@ from typing import ( from typing_extensions import TypeAlias, assert_never -from synapse.api.constants import AccountDataTypes, EduTypes, StickyEvent +from synapse.api.constants import ( + AccountDataTypes, + EduTypes, + ProfileUpdateAction, + StickyEvent, +) from synapse.events.utils import FilteredEvent from synapse.handlers.receipts import ReceiptEventSource from synapse.logging.opentracing import trace @@ -33,12 +38,15 @@ from synapse.storage.databases.main.receipts import ReceiptInRoom from synapse.types import ( Absent, DeviceListUpdates, + JsonDict, JsonMapping, + JsonValue, MultiWriterStreamToken, SlidingSyncStreamToken, StrCollection, StreamToken, ThreadSubscriptionsToken, + UserID, ) from synapse.types.handlers.sliding_sync import ( HaveSentRoomFlag, @@ -1070,6 +1078,52 @@ class SlidingSyncExtensionHandler: ), ) + async def _get_profiles_extension_initial_sync_response( + self, + user_id: UserID, + fields: set[str], + ) -> dict[str, JsonDict | None]: + """ + Build an initial sync response for the profiles extension. + + Args: + user_id: The syncing user UserID + fields: A set of fields to include in the response. + + Returns: + A dictionary containing the profile updates in an `updated` dictionary. + """ + response: dict[str, JsonDict | None] = {} + + # TODO should be filtered for the rooms for this sync + profile_user_ids = await self.store.get_local_users_who_share_room_with_user( + user_id.to_string(), + ) + # Ensure we're in the list even if we don't belong to any rooms + profile_user_ids.add(user_id.to_string()) + + profile_data_by_user = await self.store.get_profile_data_for_users( + profile_user_ids + ) + # Serialise the profile updates into the sync response format. + for profile_user_id in profile_user_ids: + profile_data = profile_data_by_user.get(profile_user_id) + if profile_data is None: + # Don't generate anything for users with no profile data + # in initial sync. + continue + per_user_updates: dict[str, JsonValue | dict[str, JsonValue]] = {} + for field_name in fields: + if field_name in profile_data.keys(): + per_user_updates[field_name] = profile_data[field_name] + + if per_user_updates: + response[profile_user_id] = { + "updated": per_user_updates, + } + + return response + async def get_profiles_extension_response( self, sync_config: SlidingSyncConfig, @@ -1078,9 +1132,118 @@ class SlidingSyncExtensionHandler: to_token: StreamToken, from_token: SlidingSyncStreamToken | None, ) -> SlidingSyncResult.Extensions.ProfilesExtension | None: + """ + Generate a response for the profiles extension. + + Args: + sync_config: The Sliding Sync config. + profiles_request: The profiles extension request. + all_interested_room_ids: Set of rooms the sync request is interested in. + to_token: The stream token to generate a response until. + from_token: The stream token to generate a response from. + + Returns: + A SlidingSyncResult.Extensions.ProfilesExtension object containing + all the users who have profile updates. + """ if not profiles_request.enabled: return None - return SlidingSyncResult.Extensions.ProfilesExtension( - users={}, + user_id = sync_config.user.to_string() + + if not profiles_request.fields: + return SlidingSyncResult.Extensions.ProfilesExtension( + users={}, + ) + fields = set(profiles_request.fields) + + response: dict[str, JsonDict | None] = {} + + if from_token is None: + # Initial sync + return SlidingSyncResult.Extensions.ProfilesExtension( + users=await self._get_profiles_extension_initial_sync_response( + user_id=sync_config.user, + fields=fields, + ), + ) + + # Incremental sync + updates = await self.store.get_profile_updates_for_user_and_fields( + from_id=from_token.stream_token.profile_updates_key, + to_id=to_token.profile_updates_key, + user_id=user_id, + field_names=set(fields), + ) + profile_user_ids = set() + updated_users = { + update.user_id + for update in updates + if update.action == ProfileUpdateAction.UPDATE.value + } + # Add users with updates + profile_user_ids.update(updated_users) + + updated_user_fields: dict[str, set[str]] = {} + # Set fields from updates + for update in updates: + # Skip the update if there is no field update (a joined or left room + # action), the client didn't ask for this field, or we're not + # interested in this user. + if ( + not update.field_name + or update.field_name not in fields + or update.user_id not in profile_user_ids + ): + continue + updated_user_fields.setdefault(update.user_id, set()).add(update.field_name) + + profile_data_by_user = await self.store.get_profile_data_for_users( + profile_user_ids + ) + + # TODO lazy loading + is_lazy = False + + # Serialise the profile updates into the sync response format. + for profile_user_id in profile_user_ids: + profile_data = profile_data_by_user.get(profile_user_id) + if profile_data is None: + # No profile data for this user, just return a blank dictionary + # in incremental sync, telling the clients to remove all profile + # information for this user. + response[profile_user_id] = None + continue + + per_user_updates: dict[str, JsonValue | dict[str, JsonValue]] = {} + if is_lazy: + # TODO lazy cache + + # Include all the fields the client asked for + fields = set(profile_data.keys()).intersection(fields) + for field_name in fields: + per_user_updates[field_name] = profile_data.get(field_name) + else: + # Include only the diff, unless the user recently joined, + # then send all the fields the client asked for. + # We don't use a cache here as for non-lazy sync we always + # send changes and/or fields the client asked for, if relevant + # as above joined condition. + fields = ( + fields + # TODO joined_room_user_ids + if profile_user_id in [] + else set(updated_user_fields.get(profile_user_id, [])) + ) + fields = set(profile_data.keys()).intersection(fields) + for field_name in fields: + per_user_updates[field_name] = profile_data[field_name] + + if per_user_updates: + response[profile_user_id] = { + "updated": per_user_updates, + } + + return SlidingSyncResult.Extensions.ProfilesExtension( + users=response, ) diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index a7036d440f..afc44116b1 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -1149,18 +1149,28 @@ class SlidingSyncRestServlet(RestServlet): serialized_extensions[ "org.matrix.msc4262.profiles" ] = await self._serialise_profiles( - requester, extensions.profiles, ref_rooms_results + extensions.profiles, ) return serialized_extensions async def _serialise_profiles( self, - requester: Requester, profiles: SlidingSyncResult.Extensions.ProfilesExtension, - ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult], ) -> JsonMapping: - return profiles.users + """ + Serialise the profiles extension response. + + Args: + profiles: The generated profiles response object. + + Returns: + A dictionary containing the response `users` with the + generated profile updates. + """ + return { + "users": profiles.users, + } async def _serialise_sticky_events( self, diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index 9730fcf956..d27793d9a7 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -450,7 +450,7 @@ class SlidingSyncResult: users: map (user_id -> [profile_updates]) """ - users: Mapping[str, JsonMapping] + users: Mapping[str, JsonMapping | None] def __bool__(self) -> bool: return bool(self.users) diff --git a/tests/rest/client/sliding_sync/test_extension_profiles.py b/tests/rest/client/sliding_sync/test_extension_profiles.py index 3c024e77c5..5fe98872ee 100644 --- a/tests/rest/client/sliding_sync/test_extension_profiles.py +++ b/tests/rest/client/sliding_sync/test_extension_profiles.py @@ -20,6 +20,7 @@ from twisted.internet.testing import MemoryReactor import synapse.rest.admin from synapse.rest.client import login, profile, room, sync from synapse.server import HomeServer +from synapse.types import UserID, create_requester from synapse.util.clock import Clock from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase @@ -55,7 +56,15 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.store = hs.get_datastores().main - + self.profile_handler = self.hs.get_profile_handler() + self.user = self.register_user("user", "password") + self.tok = self.login("user", "password") + self.other_user = self.register_user("other_user", "password") + self.other_tok = self.login("other_user", "password") + self.joined_room = self.helper.create_room_as(self.user, tok=self.tok) + self.helper.join( + room=self.joined_room, user=self.other_user, tok=self.other_tok + ) super().prepare(reactor, clock, hs) @parameterized.expand( @@ -69,26 +78,45 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): Test that no profile extension response is returned if the feature is not enabled. """ - user1_id = self.register_user("user1", "pass") - user1_tok = self.login(user1_id, "pass") - + if is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) # Make an initial Sliding Sync request with the profiles extension enabled sync_body = { "lists": {}, "extensions": { "org.matrix.msc4262.profiles": { "enabled": True, + "fields": ["field"], }, }, } - response_body, from_token = self.do_sync(sync_body, tok=user1_tok) - self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles")) + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + self.assertIsNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) if not is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) # Make an incremental Sliding Sync request - response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) - self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles")) + self.assertIsNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) @override_config({"include_profile_updates_in_sync": True}) def test_no_data_initial_sync(self) -> None: @@ -105,11 +133,14 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): "extensions": { "org.matrix.msc4262.profiles": { "enabled": True, + "fields": ["field"], }, }, } response_body, _ = self.do_sync(sync_body, tok=user1_tok) - self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles")) + self.assertIsNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) @override_config({"include_profile_updates_in_sync": True}) def test_no_data_incremental_sync(self) -> None: @@ -125,6 +156,7 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): "extensions": { "org.matrix.msc4262.profiles": { "enabled": True, + "fields": ["field"], } }, } @@ -133,4 +165,124 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): # Make an incremental Sliding Sync request with the profiles extension enabled response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) - self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles")) + self.assertIsNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) + + @parameterized.expand( + [ + True, + False, + ] + ) + @override_config({"include_profile_updates_in_sync": True}) + def test_updated_fields_are_sent(self, is_initial: bool) -> None: + """ + Test that profile extension response returns field updates + in incremental and initial sync. + """ + if is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) + # Make an initial Sliding Sync request with the profiles extension enabled + sync_body = { + "lists": {}, + "extensions": { + "org.matrix.msc4262.profiles": { + "enabled": True, + "fields": ["field"], + }, + }, + } + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + if is_initial: + self.assertEqual( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][ + "@other_user:test" + ], + { + "updated": { + "field": "value", + } + }, + ) + + if not is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) + # Make an incremental Sliding Sync request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) + + self.assertEqual( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][ + "@other_user:test" + ], + { + "updated": { + "field": "value", + } + }, + ) + + @parameterized.expand( + [ + True, + False, + ] + ) + @override_config({"include_profile_updates_in_sync": True}) + def test_updated_fields_are_not_sent_if_not_requested( + self, is_initial: bool + ) -> None: + """ + Test that profile extension response doesn't return field updates we didn't + request in initial and incremental sync. + """ + if is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="anotherfield", + new_value="value", + ) + ) + # Make an initial Sliding Sync request with the profiles extension enabled + sync_body = { + "lists": {}, + "extensions": { + "org.matrix.msc4262.profiles": { + "enabled": True, + "fields": ["field"], + }, + }, + } + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + if is_initial: + response_body["extensions"].get("org.matrix.msc4262.profiles") + + if not is_initial: + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="anotherfield", + new_value="value", + ) + ) + # Make an incremental Sliding Sync request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) + + response_body["extensions"].get("org.matrix.msc4262.profiles")