Add bundled aggregations

synapse/handlers/sliding_sync.py

@@ -114,6 +114,7 @@ class SlidingSyncHandler:
         self.auth_blocking = hs.get_auth_blocking()
         self.notifier = hs.get_notifier()
         self.event_sources = hs.get_event_sources()
+        self.relations_handler = hs.get_relations_handler()
         self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
 
     async def wait_for_sync_for_user(
@@ -881,6 +882,18 @@ class SlidingSyncHandler:
 
             stripped_state.append(strip_event(invite_or_knock_event))
 
+        # TODO: Handle timeline gaps (`get_timeline_gaps()`)
+
+        # If the timeline is `limited=True`, the client does not have all events
+        # necessary to calculate aggregations themselves.
+        bundled_aggregations = None
+        if limited:
+            bundled_aggregations = (
+                await self.relations_handler.get_bundled_aggregations(
+                    timeline_events, user.to_string()
+                )
+            )
+
         return SlidingSyncResult.RoomResult(
             # TODO: Dummy value
             name=None,
@@ -895,7 +908,8 @@ class SlidingSyncHandler:
             initial=True,
             # TODO: Dummy value
             required_state=[],
-            timeline=timeline_events,
+            timeline_events=timeline_events,
+            bundled_aggregations=bundled_aggregations,
             # TODO: Dummy value
             is_dm=False,
             stripped_state=stripped_state,

synapse/rest/client/sync.py

@@ -976,11 +976,10 @@ class SlidingSyncRestServlet(RestServlet):
         serialized_rooms = {}
         for room_id, room_result in rooms.items():
             serialized_timeline = await self.event_serializer.serialize_events(
-                room_result.timeline,
+                room_result.timeline_events,
                 time_now,
                 config=serialize_options,
-                # TODO
-                # bundle_aggregations=room.timeline.bundled_aggregations,
+                bundle_aggregations=room_result.bundled_aggregations,
             )
 
             serialized_required_state = await self.event_serializer.serialize_events(

synapse/types/handlers/__init__.py

@@ -31,6 +31,7 @@ else:
     from pydantic import Extra
 
 from synapse.events import EventBase
+from synapse.handlers.relations import BundledAggregations
 from synapse.types import JsonDict, JsonMapping, StreamToken, UserID
 from synapse.types.rest.client import SlidingSyncBody
 
@@ -159,7 +160,11 @@ class SlidingSyncResult:
                 entirely and NOT send "initial":false as this is wasteful on bandwidth. The
                 absence of this flag means 'false'.
             required_state: The current state of the room
-            timeline: Latest events in the room. The last event is the most recent
+            timeline: Latest events in the room. The last event is the most recent.
+            bundled_aggregations: A mapping of event ID to the bundled aggregations for
+                the timeline events above. This allows clients to show accurate reaction
+                counts (or edits, threads), even if some of the reaction events were skipped
+                over in a gappy sync.
             is_dm: Flag to specify whether the room is a direct-message room (most likely
                 between two people).
             stripped_state: Stripped state events (for rooms where the usre is
@@ -191,7 +196,8 @@ class SlidingSyncResult:
         heroes: Optional[List[EventBase]]
         initial: bool
         required_state: List[EventBase]
-        timeline: List[EventBase]
+        timeline_events: List[EventBase]
+        bundled_aggregations: Optional[Dict[str, BundledAggregations]]
         is_dm: bool
         stripped_state: Optional[List[JsonDict]]
         prev_batch: StreamToken