Send down full profile of user if user has joined during incremental sync

Also refactor the "user left" to not listen to the "user_left_room" dispatch hook since we only want to do this once, thus call the profile handler immediately in the instance responsible for receiving the event.

synapse/api/constants.py

@@ -411,6 +411,7 @@ class ProfileFields:
 
 
 class ProfileUpdateAction(enum.Enum):
+    JOINED_ROOM = "joined_room"
     LEFT_ROOM = "left_room"
     UPDATE = "update"
 

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(

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:
         """

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)
                         )

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."""