diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 44c84252fc..80505da6d1 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -411,6 +411,7 @@ class ProfileFields: class ProfileUpdateAction(enum.Enum): + JOINED_ROOM = "joined_room" LEFT_ROOM = "left_room" UPDATE = "update" diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 296b7ed57d..ba5a0ceab4 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -102,8 +102,6 @@ class ProfileHandler: ) self._worker_locks = hs.get_worker_locks_handler() - hs.get_distributor().observe("user_left_room", self.user_left_room) - async def _record_profile_updates( self, user_id: UserID, updated_fields: list[str] ) -> None: @@ -431,18 +429,18 @@ class ProfileHandler: if propagate: await self._update_join_states(requester, target_user) - async def _user_left_room(self, user_id: UserID, room_id: str) -> None: + async def user_left_room(self, user_id: UserID, room_id: str) -> None: """ A user left a room. We now: - * Add a row to `profile_updates` stating the the user left a certain room. + * Add a row to `profile_updates` stating that the user left a room. * Check if this user no longer shares any rooms with certain users. * Insert a row for each of those users into `profile_updates_per_user`. * Now, when any of those users sync, the sync code will check `profile_updates` and see that the user left a room. And thus a "clear this user's profile" instruction will be sent down to the client. - If that is the case, + If that is the case, """ user_id_str = user_id.to_string() stream_id = await self.store.add_profile_updates( @@ -467,13 +465,35 @@ class ProfileHandler: user_ids=users_to_update, ) - def user_left_room(self, user: UserID, room_id: str) -> None: - if self.hs.is_mine_id(user.to_string()): - self.hs.run_as_background_process( - "profile._user_left_room", - self._user_left_room, - user_id=user, - room_id=room_id, + async def user_joined_room(self, user_id: UserID, room_id: str) -> None: + """ + A user joined a room. We now: + + * Add a row to `profile_updates` stating that the user joined a room. + * Get list of users in that room. + * Insert a row for each of those users into `profile_updates_per_user`. + * Now, when any of those users sync, the sync code will check + `profile_updates` and see that the user joined a room. Thus, we can + include the users full profile in the case that we need to do so. + + If that is the case, + """ + user_id_str = user_id.to_string() + stream_id = await self.store.add_profile_updates( + user_id=user_id, + action=ProfileUpdateAction.JOINED_ROOM.value, + updated_fields=None, + ) + + users_in_room = set(await self.store.get_local_users_in_room(room_id)) + users_in_room.discard(user_id_str) + if not users_in_room: + return + + if users_in_room: + await self.store.track_profile_updates_per_user( + stream_id=stream_id, + user_ids=users_in_room, ) async def delete_profile_upon_deactivation( diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index 5152d0b522..2ba29fbe48 100644 --- a/synapse/handlers/room_member.py +++ b/synapse/handlers/room_member.py @@ -526,6 +526,14 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): ) if prev_member_event.membership == Membership.JOIN: await self._user_left_room(target, room_id) + # Notify the profile handler. We only want to do this once + # in a multi-worker setup, so we can't listen on the dispatched + # event above. + await self.profile_handler.user_left_room(target, room_id) + elif event.membership == Membership.JOIN: + # Notify the profile handler. We only want to do this once + # in a multi-worker setup, so we can't dispatch a hook to all workers. + await self.profile_handler.user_joined_room(target, room_id) break except PartialStateConflictError as e: @@ -1541,6 +1549,14 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): prev_member_event = await self.store.get_event(prev_member_event_id) if prev_member_event.membership == Membership.JOIN: await self._user_left_room(target_user, room_id) + # Notify the profile handler. We only want to do this once + # in a multi-worker setup, so we can't listen on the dispatched + # event above. + await self.profile_handler.user_left_room(target_user, room_id) + elif event.membership == Membership.JOIN: + # Notify the profile handler. We only want to do this once + # in a multi-worker setup, so we can't dispatch a hook to all workers. + await self.profile_handler.user_joined_room(target_user, room_id) async def _can_guest_join(self, partial_current_state_ids: StateMap[str]) -> bool: """ diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 1f31b53786..eb9b186775 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -2307,6 +2307,11 @@ class SyncHandler: for update in updates if update.action == ProfileUpdateAction.LEFT_ROOM.value } + joined_room_user_ids = { + update.user_id + for update in updates + if update.action == ProfileUpdateAction.JOINED_ROOM.value + } users = set() updated_users = { update.user_id @@ -2318,6 +2323,8 @@ class SyncHandler: users.update(include_users) # Add users with updates users.update(updated_users) + # Add any newly joined users + users.update(joined_room_user_ids) if not users and not left_room_user_ids: return @@ -2387,9 +2394,14 @@ class SyncHandler: cast(str, profile_data.get(field_name)), ) else: - # Include only the diff + # Include only the diff, unless the user recently joined # We don't use a cache here as changes are always sent - for field_name in user_fields.get(other_user_id, []): + fields = ( + list(profile_data.keys()) + if other_user_id in joined_room_user_ids + else user_fields.get(other_user_id, []) + ) + for field_name in fields: per_user_updates[field_name] = cast( JsonValue, profile_data.get(field_name) ) diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index fbec22ab9b..4ee74a4721 100644 --- a/tests/handlers/test_sync.py +++ b/tests/handlers/test_sync.py @@ -1634,6 +1634,86 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): incremental_result.profile_updates["@other_user:test"], ) + @override_config({"experimental_features": {"msc4429_enabled": True}}) + def test_incremental_sync_sends_down_full_profile_for_users_who_have_joined( + self, + ) -> None: + requester = create_requester(self.user) + initial_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json={ + "org.matrix.msc4429.profile_fields": { + "ids": ["m.status", "displayname", "avatar_url"] + }, + }, + ), + ), + request_key=generate_request_key(), + ) + ) + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=initial_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json={ + "org.matrix.msc4429.profile_fields": { + "ids": ["m.status", "displayname", "avatar_url"] + }, + }, + ), + ), + request_key=generate_request_key(), + ) + ) + + third_user = self.register_user("third_user", "password") + third_tok = self.login("third_user", "password") + self.helper.join( + room=self.joined_room, + user=third_user, + tok=third_tok, + ) + + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=incremental_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json={ + "org.matrix.msc4429.profile_fields": { + "ids": ["m.status", "displayname", "avatar_url"] + }, + }, + ), + ), + request_key=generate_request_key(), + ) + ) + assert incremental_result.profile_updates["@third_user:test"] is not None + self.assertCountEqual( + incremental_result.profile_updates.keys(), + [third_user], + ) + self.assertEqual( + incremental_result.profile_updates["@third_user:test"]["displayname"], + "third_user", + ) + self.assertIsNone( + incremental_result.profile_updates["@third_user:test"]["avatar_url"], + ) + class SyncStateAfterTestCase(tests.unittest.HomeserverTestCase): """Tests Sync Handler state behavior when using `use_state_after."""