Add thread_root events to threads extension response

synapse/handlers/sliding_sync/extensions.py

@@ -77,6 +77,7 @@ class SlidingSyncExtensionHandler:
         self.event_sources = hs.get_event_sources()
         self.device_handler = hs.get_device_handler()
         self.push_rules_handler = hs.get_push_rules_handler()
+        self.relations_handler = hs.get_relations_handler()
         self._enable_thread_subscriptions = hs.config.experimental.msc4306_enabled
         self._enable_threads_ext = hs.config.experimental.msc4360_enabled
 
@@ -1011,24 +1012,35 @@ class SlidingSyncExtensionHandler:
         # TODO: is the `room_key` the right thing to use here?
         # ie. does it translate into /relations
 
-        # TODO: use new function to get thread updates
         updates, prev_batch = await self.store.get_thread_updates_for_user(
             user_id=sync_config.user.to_string(),
             from_token=from_token.stream_token if from_token else None,
             to_token=to_token,
             limit=limit,
+            include_thread_roots=threads_request.include_roots,
         )
 
         if len(updates) == 0:
             return None
 
-        # TODO: implement
+        # Collect thread root events and get bundled aggregations
+        thread_root_events = [event for _, _, event in updates if event]
+        aggregations_map = {}
+        if thread_root_events:
+            aggregations_map = await self.relations_handler.get_bundled_aggregations(
+                thread_root_events,
+                sync_config.user.to_string(),
+            )
 
         thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {}
-        for thread_root_id, room_id in updates:
+        for thread_root_id, room_id, thread_root_event in updates:
+            bundled_aggs = (
+                aggregations_map.get(thread_root_id) if thread_root_event else None
+            )
             thread_updates.setdefault(room_id, {})[thread_root_id] = _ThreadUpdate(
-                thread_root=None,
+                thread_root=thread_root_event,
                 prev_batch=None,
+                bundled_aggregations=bundled_aggs,
             )
 
         return SlidingSyncResult.Extensions.ThreadsExtension(

synapse/rest/client/sync.py

@@ -31,6 +31,7 @@ from synapse.api.filtering import FilterCollection
 from synapse.api.presence import UserPresenceState
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.events.utils import (
+    EventClientSerializer,
     SerializeEventConfig,
     format_event_for_client_v2_without_room_id,
     format_event_raw,
@@ -852,7 +853,10 @@ class SlidingSyncRestServlet(RestServlet):
             logger.info("Client has disconnected; not serializing response.")
             return 200, {}
 
-        response_content = await self.encode_response(requester, sliding_sync_results)
+        time_now = self.clock.time_msec()
+        response_content = await self.encode_response(
+            requester, sliding_sync_results, time_now
+        )
 
         return 200, response_content
 
@@ -861,6 +865,7 @@ class SlidingSyncRestServlet(RestServlet):
         self,
         requester: Requester,
         sliding_sync_result: SlidingSyncResult,
+        time_now: int,
     ) -> JsonDict:
         response: JsonDict = defaultdict(dict)
 
@@ -869,10 +874,10 @@ class SlidingSyncRestServlet(RestServlet):
         if serialized_lists:
             response["lists"] = serialized_lists
         response["rooms"] = await self.encode_rooms(
-            requester, sliding_sync_result.rooms
+            requester, sliding_sync_result.rooms, time_now
         )
         response["extensions"] = await self.encode_extensions(
-            requester, sliding_sync_result.extensions
+            requester, sliding_sync_result.extensions, time_now
         )
 
         return response
@@ -904,9 +909,8 @@ class SlidingSyncRestServlet(RestServlet):
         self,
         requester: Requester,
         rooms: Dict[str, SlidingSyncResult.RoomResult],
+        time_now: int,
     ) -> JsonDict:
-        time_now = self.clock.time_msec()
-
         serialize_options = SerializeEventConfig(
             event_format=format_event_for_client_v2_without_room_id,
             requester=requester,
@@ -1022,7 +1026,10 @@ class SlidingSyncRestServlet(RestServlet):
 
     @trace_with_opname("sliding_sync.encode_extensions")
     async def encode_extensions(
-        self, requester: Requester, extensions: SlidingSyncResult.Extensions
+        self,
+        requester: Requester,
+        extensions: SlidingSyncResult.Extensions,
+        time_now: int,
     ) -> JsonDict:
         serialized_extensions: JsonDict = {}
 
@@ -1094,8 +1101,12 @@ class SlidingSyncRestServlet(RestServlet):
 
         # excludes both None and falsy `threads`
         if extensions.threads:
-            serialized_extensions["io.element.msc4360.threads"] = _serialise_threads(
-                extensions.threads
+            serialized_extensions[
+                "io.element.msc4360.threads"
+            ] = await _serialise_threads(
+                self.event_serializer,
+                time_now,
+                extensions.threads,
             )
 
         return serialized_extensions
@@ -1134,21 +1145,63 @@ def _serialise_thread_subscriptions(
     return out
 
 
-def _serialise_threads(
+async def _serialise_threads(
+    event_serializer: EventClientSerializer,
+    time_now: int,
     threads: SlidingSyncResult.Extensions.ThreadsExtension,
 ) -> JsonDict:
+    """
+    Serialize the threads extension response for sliding sync.
+
+    Args:
+        event_serializer: The event serializer to use for serializing thread root events.
+        time_now: The current time in milliseconds, used for event serialization.
+        threads: The threads extension data containing thread updates and pagination tokens.
+
+    Returns:
+        A JSON-serializable dict containing:
+        - "updates": A nested dict mapping room_id -> thread_root_id -> thread update.
+          Each thread update may contain:
+          - "thread_root": The serialized thread root event (if include_roots was True),
+            with bundled aggregations including the latest_event in unsigned.m.relations.m.thread.
+          - "prev_batch": A pagination token for fetching older events in the thread.
+        - "prev_batch": A pagination token for fetching older thread updates (if available).
+    """
     out: JsonDict = {}
 
     if threads.updates:
-        out["updates"] = {
-            room_id: {
-                thread_root_id: attr.asdict(
-                    update, filter=lambda _attr, v: v is not None
-                )
-                for thread_root_id, update in thread_updates.items()
-            }
-            for room_id, thread_updates in threads.updates.items()
-        }
+        updates_dict: JsonDict = {}
+        for room_id, thread_updates in threads.updates.items():
+            room_updates: JsonDict = {}
+            for thread_root_id, update in thread_updates.items():
+                # Serialize the update
+                update_dict: JsonDict = {}
+
+                # Serialize the thread_root event if present
+                if update.thread_root is not None:
+                    # Create a mapping of event_id to bundled_aggregations
+                    bundle_aggs_map = (
+                        {thread_root_id: update.bundled_aggregations}
+                        if update.bundled_aggregations
+                        else None
+                    )
+                    serialized_events = await event_serializer.serialize_events(
+                        [update.thread_root],
+                        time_now,
+                        bundle_aggregations=bundle_aggs_map,
+                    )
+                    if serialized_events:
+                        update_dict["thread_root"] = serialized_events[0]
+
+                # Add prev_batch if present
+                if update.prev_batch is not None:
+                    update_dict["prev_batch"] = str(update.prev_batch)
+
+                room_updates[thread_root_id] = update_dict
+
+            updates_dict[room_id] = room_updates
+
+        out["updates"] = updates_dict
 
     if threads.prev_batch:
         out["prev_batch"] = str(threads.prev_batch)

synapse/storage/databases/main/relations.py

@@ -47,6 +47,7 @@ from synapse.storage.database import (
     LoggingTransaction,
     make_in_list_sql_clause,
 )
+from synapse.storage.databases.main.events_worker import EventsWorkerStore
 from synapse.storage.databases.main.stream import (
     generate_next_token,
     generate_pagination_bounds,
@@ -95,7 +96,7 @@ class _RelatedEvent:
     sender: str
 
 
-class RelationsWorkerStore(SQLBaseStore):
+class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore):
     def __init__(
         self,
         database: DatabasePool,
@@ -591,7 +592,7 @@ class RelationsWorkerStore(SQLBaseStore):
             "get_applicable_edits", _get_applicable_edits_txn
         )
 
-        edits = await self.get_events(edit_ids.values())  # type: ignore[attr-defined]
+        edits = await self.get_events(edit_ids.values())
 
         # Map to the original event IDs to the edit events.
         #
@@ -706,7 +707,7 @@ class RelationsWorkerStore(SQLBaseStore):
             "get_thread_summaries", _get_thread_summaries_txn
         )
 
-        latest_events = await self.get_events(latest_event_ids.values())  # type: ignore[attr-defined]
+        latest_events = await self.get_events(latest_event_ids.values())
 
         # Map to the event IDs to the thread summary.
         #
@@ -1125,7 +1126,8 @@ class RelationsWorkerStore(SQLBaseStore):
         from_token: Optional[StreamToken] = None,
         to_token: Optional[StreamToken] = None,
         limit: int = 5,
-    ) -> Tuple[Sequence[Tuple[str, str]], Optional[int]]:
+        include_thread_roots: bool = False,
+    ) -> Tuple[Sequence[Tuple[str, str, Optional[EventBase]]], Optional[int]]:
         """Get a list of updated threads, ordered by stream ordering of their
         latest reply, filtered to only include threads in rooms where the user
         was joined at the time of the thread's latest update.
@@ -1150,10 +1152,13 @@ class RelationsWorkerStore(SQLBaseStore):
             from_token: Fetch rows from a previous next_batch, or from the start if None.
             to_token: Fetch rows from a previous prev_batch, or from the stream end if None.
             limit: Only fetch the most recent `limit` threads.
+            include_thread_roots: If True, fetch and return the thread root EventBase
+                objects. If False, return None for the event.
 
         Returns:
             A tuple of:
-                A list of (thread_id, room_id) tuples.
+                A list of (thread_id, room_id, thread_root_event) tuples.
+                    thread_root_event will be None if include_thread_roots=False.
                 The next_batch, if one exists.
         """
         # Ensure bad limits aren't being passed in.
@@ -1228,10 +1233,29 @@ class RelationsWorkerStore(SQLBaseStore):
 
             return thread_ids[:limit], next_token
 
-        return await self.db_pool.runInteraction(
+        thread_ids, next_token = await self.db_pool.runInteraction(
             "get_thread_updates_for_user", _get_thread_updates_for_user_txn
         )
 
+        # Optionally fetch thread root events
+        if include_thread_roots and thread_ids:
+            thread_root_ids = [thread_id for thread_id, _ in thread_ids]
+            thread_root_events = await self.get_events_as_list(thread_root_ids)
+            event_map = {e.event_id: e for e in thread_root_events}
+
+            return (
+                [
+                    (thread_id, room_id, event_map.get(thread_id))
+                    for thread_id, room_id in thread_ids
+                ],
+                next_token,
+            )
+        else:
+            return (
+                [(thread_id, room_id, None) for thread_id, room_id in thread_ids],
+                next_token,
+            )
+
 
 class RelationsStore(RelationsWorkerStore):
     pass

synapse/types/handlers/sliding_sync.py

@@ -40,6 +40,9 @@ import attr
 from synapse._pydantic_compat import Extra
 from synapse.api.constants import EventTypes
 from synapse.events import EventBase
+
+if TYPE_CHECKING:
+    from synapse.handlers.relations import BundledAggregations
 from synapse.types import (
     DeviceListUpdates,
     JsonDict,
@@ -406,12 +409,15 @@ class SlidingSyncResult:
 
             @attr.s(slots=True, frozen=True, auto_attribs=True)
             class ThreadUpdate:
-                # TODO: comment
+                # The thread root event, if requested via include_roots
                 thread_root: Optional[EventBase]
 
-                # TODO: comment
+                # Pagination token for this thread
                 prev_batch: Optional[StreamToken]
 
+                # Bundled aggregations for the thread root event (includes latest_event)
+                bundled_aggregations: Optional["BundledAggregations"] = None
+
                 def __bool__(self) -> bool:
                     return bool(self.thread_root) or bool(self.prev_batch)
 

tests/rest/client/sliding_sync/test_extension_threads.py

@@ -357,3 +357,121 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
             response_body["extensions"],
             "User2 should not see thread updates after leaving the room",
         )
+
+    def test_threads_with_include_roots_true(self) -> None:
+        """
+        Test that include_roots=True returns thread root events with latest_event
+        in the unsigned field.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Create thread root
+        thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok)
+        thread_root_id = thread_root_resp["event_id"]
+
+        # Add reply to thread
+        latest_event_resp = self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Latest reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+        latest_event_id = latest_event_resp["event_id"]
+
+        # Sync with include_roots=True
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                    "include_roots": True,
+                }
+            },
+        }
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok)
+
+        # Assert thread root is present
+        thread_root = response_body["extensions"][EXT_NAME]["updates"][room_id][
+            thread_root_id
+        ]["thread_root"]
+
+        # Verify it's the correct event
+        self.assertEqual(thread_root["event_id"], thread_root_id)
+        self.assertEqual(thread_root["content"]["body"], "Thread root")
+
+        # Verify latest_event is in unsigned.m.relations.m.thread
+        latest_event = thread_root["unsigned"]["m.relations"]["m.thread"][
+            "latest_event"
+        ]
+        self.assertEqual(latest_event["event_id"], latest_event_id)
+        self.assertEqual(latest_event["content"]["body"], "Latest reply")
+
+    def test_threads_with_include_roots_false(self) -> None:
+        """
+        Test that include_roots=False (or omitted) does not return thread root events.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Create thread
+        thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok)
+        thread_root_id = thread_root_resp["event_id"]
+
+        # Add reply
+        self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": "Reply",
+                "m.relates_to": {
+                    "rel_type": RelationTypes.THREAD,
+                    "event_id": thread_root_id,
+                },
+            },
+            tok=user1_tok,
+        )
+
+        # Sync with include_roots=False (explicitly)
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                    "include_roots": False,
+                }
+            },
+        }
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok)
+
+        # Assert thread update exists but has no thread_root
+        thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][
+            thread_root_id
+        ]
+        self.assertNotIn("thread_root", thread_update)
+
+        # Also test with include_roots omitted (should behave the same)
+        sync_body_no_param = {
+            "lists": {},
+            "extensions": {
+                EXT_NAME: {
+                    "enabled": True,
+                }
+            },
+        }
+        response_body_no_param, _ = self.do_sync(sync_body_no_param, tok=user1_tok)
+
+        thread_update_no_param = response_body_no_param["extensions"][EXT_NAME][
+            "updates"
+        ][room_id][thread_root_id]
+        self.assertNotIn("thread_root", thread_update_no_param)