diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index bcad918d96..4eab311007 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/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( diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index cfb662c9a2..63d2c390d6 100644 --- a/synapse/rest/client/sync.py +++ b/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) diff --git a/synapse/storage/databases/main/relations.py b/synapse/storage/databases/main/relations.py index a8543040f3..48f6b0cbf1 100644 --- a/synapse/storage/databases/main/relations.py +++ b/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 diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index 01679f4fb0..7c7a00b1d9 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/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) diff --git a/tests/rest/client/sliding_sync/test_extension_threads.py b/tests/rest/client/sliding_sync/test_extension_threads.py index 9e7d000167..8a07d7207c 100644 --- a/tests/rest/client/sliding_sync/test_extension_threads.py +++ b/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)