Fix stream token multiple devices

2 files changed, 33 insertions, 0 deletions

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 2cc229abcc..e76e40b3b4 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -296,6 +296,30 @@ class BasePresenceHandler(abc.ABC):
         for destinations, states in hosts_and_states:
             self._federation.send_presence_to_destinations(states, destinations)
 
+    async def resend_current_presence_for_users(self, user_ids: Iterable[str]):
+        """
+        Grabs the current presence state for a given set of users and adds it
+        to the top of the presence stream.
+
+        Args:
+            user_ids: The IDs of the users to use.
+        """
+        # Get the current presence state for each user (defaults to offline if not found)
+        current_presence_for_users = await self.current_state_for_users(user_ids)
+
+        for user_id, current_presence_state in current_presence_for_users.items():
+            # Convert the UserPresenceState object into a serializable dict
+            state = {
+                "presence": current_presence_state.state,
+                "status_message": current_presence_state.status_msg,
+            }
+
+            # Copy the presence state to the tip of the presence stream
+            print(f"Adding a presence update for {user_id}: {state}")
+            await self.set_state(UserID.from_string(user_id), state)
+
+        print("bla")
+
 
 class _NullContextManager(ContextManager[None]):
     """A context manager which does nothing."""
diff --git a/synapse/storage/databases/main/presence.py b/synapse/storage/databases/main/presence.py
index 3e72d10062..9079b5f592 100644
--- a/synapse/storage/databases/main/presence.py
+++ b/synapse/storage/databases/main/presence.py
@@ -57,6 +57,7 @@ class PresenceStore(SQLBaseStore):
                 db_conn, "presence_stream", "stream_id"
             )
 
+        self.hs = hs
         self._presence_on_startup = self._get_active_presence(db_conn)
 
         presence_cache_prefill, min_presence_val = self.db_pool.get_cache_dict(
@@ -266,6 +267,14 @@ class PresenceStore(SQLBaseStore):
             desc="add_users_to_send_full_presence_to",
         )
 
+        # Add a new entry to the presence stream. Since we use stream tokens to determine whether a
+        # local user should receive a full snapshot presence when they sync, we need to bump the
+        # presence stream so that subsequent syncs with no presence activity in between won't result
+        # in the client receiving multiple full snapshots of presence.
+        # If we bump the stream ID, then the user will get a higher stream token next sync, and thus
+        # won't receive another snapshot.
+        await self.hs.get_presence_handler().resend_current_presence_for_users(user_ids)
+
     async def get_presence_for_all_users(
         self,
         include_offline: bool = True,