diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 4eab311007..5defd047aa 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -996,8 +996,8 @@ class SlidingSyncExtensionHandler: """Handle Threads extension (MSC4360) Args: - sync_config: Sync configuration - threads_request: The threads extension from the request + sync_config: Sync configuration. + threads_request: The threads extension from the request. to_token: The point in the stream to sync up to. from_token: The point in the stream to sync from. @@ -1007,24 +1007,30 @@ class SlidingSyncExtensionHandler: if not threads_request.enabled: return None - limit = threads_request.limit - - # TODO: is the `room_key` the right thing to use here? - # ie. does it translate into /relations - - updates, prev_batch = await self.store.get_thread_updates_for_user( + # Fetch thread updates globally across all joined rooms. + # The database layer returns a StreamToken (exclusive) for prev_batch if there + # are more results. + ( + all_thread_updates, + prev_batch_token, + ) = 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, + from_token=from_token.stream_token.room_key if from_token else None, + to_token=to_token.room_key, + limit=threads_request.limit, include_thread_roots=threads_request.include_roots, ) - if len(updates) == 0: + if len(all_thread_updates) == 0: return None - # Collect thread root events and get bundled aggregations - thread_root_events = [event for _, _, event in updates if event] + # Collect thread root events and get bundled aggregations. + # Only fetch bundled aggregations if we have thread root events to attach them to. + thread_root_events = [ + update.thread_root_event + for update in all_thread_updates + if update.thread_root_event + ] aggregations_map = {} if thread_root_events: aggregations_map = await self.relations_handler.get_bundled_aggregations( @@ -1033,17 +1039,23 @@ class SlidingSyncExtensionHandler: ) thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {} - for thread_root_id, room_id, thread_root_event in updates: + for update in all_thread_updates: + # Only look up bundled aggregations if we have a thread root event bundled_aggs = ( - aggregations_map.get(thread_root_id) if thread_root_event else None + aggregations_map.get(update.thread_id) + if update.thread_root_event + else None ) - thread_updates.setdefault(room_id, {})[thread_root_id] = _ThreadUpdate( - thread_root=thread_root_event, - prev_batch=None, - bundled_aggregations=bundled_aggs, + + thread_updates.setdefault(update.room_id, {})[update.thread_id] = ( + _ThreadUpdate( + thread_root=update.thread_root_event, + prev_batch=update.prev_batch, + bundled_aggregations=bundled_aggs, + ) ) return SlidingSyncResult.Extensions.ThreadsExtension( updates=thread_updates, - prev_batch=prev_batch, + prev_batch=prev_batch_token, ) diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index 63d2c390d6..0ddd82d877 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -57,6 +57,7 @@ from synapse.http.servlet import ( from synapse.http.site import SynapseRequest from synapse.logging.opentracing import log_kv, set_tag, trace_with_opname from synapse.rest.admin.experimental_features import ExperimentalFeature +from synapse.storage.databases.main import DataStore from synapse.types import JsonDict, Requester, SlidingSyncStreamToken, StreamToken from synapse.types.rest.client import SlidingSyncBody from synapse.util.caches.lrucache import LruCache @@ -1107,6 +1108,7 @@ class SlidingSyncRestServlet(RestServlet): self.event_serializer, time_now, extensions.threads, + self.store, ) return serialized_extensions @@ -1149,6 +1151,7 @@ async def _serialise_threads( event_serializer: EventClientSerializer, time_now: int, threads: SlidingSyncResult.Extensions.ThreadsExtension, + store: "DataStore", ) -> JsonDict: """ Serialize the threads extension response for sliding sync. @@ -1157,6 +1160,7 @@ async def _serialise_threads( 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. + store: The datastore, needed for serializing stream tokens. Returns: A JSON-serializable dict containing: @@ -1195,7 +1199,7 @@ async def _serialise_threads( # Add prev_batch if present if update.prev_batch is not None: - update_dict["prev_batch"] = str(update.prev_batch) + update_dict["prev_batch"] = await update.prev_batch.to_string(store) room_updates[thread_root_id] = update_dict @@ -1204,7 +1208,7 @@ async def _serialise_threads( out["updates"] = updates_dict if threads.prev_batch: - out["prev_batch"] = str(threads.prev_batch) + out["prev_batch"] = await threads.prev_batch.to_string(store) return out diff --git a/synapse/storage/databases/main/relations.py b/synapse/storage/databases/main/relations.py index 5be960b1c7..5d412d38b3 100644 --- a/synapse/storage/databases/main/relations.py +++ b/synapse/storage/databases/main/relations.py @@ -54,7 +54,12 @@ from synapse.storage.databases.main.stream import ( generate_pagination_where_clause, ) from synapse.storage.engines import PostgresEngine -from synapse.types import JsonDict, StreamKeyType, StreamToken +from synapse.types import ( + JsonDict, + RoomStreamToken, + StreamKeyType, + StreamToken, +) from synapse.util.caches.descriptors import cached, cachedList if TYPE_CHECKING: @@ -96,6 +101,28 @@ class _RelatedEvent: sender: str +@attr.s(slots=True, frozen=True, auto_attribs=True) +class ThreadUpdateInfo: + """ + Information about a thread update for the sliding sync threads extension. + + Attributes: + thread_id: The event ID of the thread root event (the event that started the thread). + room_id: The room ID where this thread exists. + thread_root_event: The actual EventBase object for the thread root event, + if include_thread_roots was True in the request. Otherwise None. + prev_batch: A pagination token (exclusive) for fetching older events in this thread. + Only present if update_count > 1. This token can be used with the /relations + endpoint with dir=b to paginate backwards through the thread's history without + re-receiving the latest event that was already included in the sliding sync response. + """ + + thread_id: str + room_id: str + thread_root_event: Optional[EventBase] + prev_batch: Optional[StreamToken] + + class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): def __init__( self, @@ -1127,31 +1154,35 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): self, *, user_id: str, - from_token: Optional[StreamToken] = None, - to_token: Optional[StreamToken] = None, + from_token: Optional[RoomStreamToken] = None, + to_token: Optional[RoomStreamToken] = None, limit: int = 5, include_thread_roots: bool = False, - ) -> Tuple[Sequence[Tuple[str, str, Optional[EventBase]]], Optional[int]]: + ) -> Tuple[Sequence[ThreadUpdateInfo], Optional[StreamToken]]: """Get a list of updated threads, ordered by stream ordering of their latest reply, filtered to only include threads in rooms where the user is currently joined. Args: - user_id: Only fetch threads for rooms that the user is currently joined to. - 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. + user_id: The user ID to fetch thread updates for. Only threads in rooms + where this user is currently joined will be returned. + from_token: The lower bound (exclusive) for thread updates. If None, + fetch from the start of the room timeline. + to_token: The upper bound (inclusive) for thread updates. If None, + fetch up to the current position in the room timeline. + limit: Maximum number of thread updates to return. include_thread_roots: If True, fetch and return the thread root EventBase - objects. If False, return None for the event. + objects. If False, return None for the thread_root_event field. Returns: A tuple of: - 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. + A list of ThreadUpdateInfo objects containing thread update information, + ordered by stream_ordering descending (most recent first). + A prev_batch StreamToken (exclusive) if there are more results available, + None otherwise. """ # Ensure bad limits aren't being passed in. - assert limit >= 0 + assert limit > 0 # Generate the pagination clause, if necessary. # @@ -1159,18 +1190,35 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): pagination_clause = "" pagination_args: List[str] = [] if from_token: - from_bound = from_token.room_key.stream + from_bound = from_token.stream pagination_clause += " AND stream_ordering > ?" pagination_args.append(str(from_bound)) if to_token: - to_bound = to_token.room_key.stream + to_bound = to_token.stream pagination_clause += " AND stream_ordering <= ?" pagination_args.append(str(to_bound)) - # Filter threads to only those in rooms that the user is currently joined to. + # Build the update count clause - count events in the thread within the sync window + update_count_clause = "" + update_count_args: List[str] = [] + update_count_clause = f""" + (SELECT COUNT(*) + FROM event_relations AS er + INNER JOIN events AS e ON er.event_id = e.event_id + WHERE er.relates_to_id = threads.thread_id + AND er.relation_type = '{RelationTypes.THREAD}'""" + if from_token: + update_count_clause += " AND e.stream_ordering > ?" + update_count_args.append(str(from_token.stream)) + if to_token: + update_count_clause += " AND e.stream_ordering <= ?" + update_count_args.append(str(to_token.stream)) + update_count_clause += ")" + + # Filter threads to only those in rooms where the user is currently joined. sql = f""" - SELECT thread_id, room_id, latest_event_id, stream_ordering + SELECT thread_id, room_id, stream_ordering, {update_count_clause} AS update_count FROM threads WHERE EXISTS ( SELECT 1 @@ -1186,44 +1234,86 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): def _get_thread_updates_for_user_txn( txn: LoggingTransaction, - ) -> Tuple[List[Tuple[str, str]], Optional[int]]: - txn.execute(sql, (user_id, Membership.JOIN, *pagination_args, limit + 1)) + ) -> Tuple[List[Tuple[str, str, int, int]], Optional[int]]: + # Add 1 to the limit as a free way of determining if there are more results + # than the limit amount. If `limit + 1` results are returned, then there are + # more results. Otherwise we would need to do a separate query to determine + # if this was true when exactly `limit` results are returned. + txn.execute( + sql, + ( + *update_count_args, + user_id, + Membership.JOIN, + *pagination_args, + limit + 1, + ), + ) - rows = cast(List[Tuple[str, str, str, int]], txn.fetchall()) - thread_ids = [(r[0], r[1]) for r in rows] + # SQL returns: thread_id, room_id, stream_ordering, update_count + rows = cast(List[Tuple[str, str, int, int]], txn.fetchall()) # If there are more events, generate the next pagination key from the # last thread which will be returned. next_token = None - if len(thread_ids) > limit: - # TODO: why -2? - next_token = rows[-2][3] + if len(rows) > limit: + # Set the next_token to be the second last row in the result set since + # that will be the last row we return from this function. + # This works as an exclusive bound that can be backpaginated from. + # Use the stream_ordering field (index 2 in original rows) + next_token = rows[-2][2] - return thread_ids[:limit], next_token + return rows[:limit], next_token - thread_ids, next_token = await self.db_pool.runInteraction( + thread_infos, next_token_int = await self.db_pool.runInteraction( "get_thread_updates_for_user", _get_thread_updates_for_user_txn ) + # Convert the next_token int (stream ordering) to a StreamToken. + # Use StreamToken.START as base (all other streams at 0) since only room + # position matters. + # Subtract 1 to make it exclusive - the client can paginate from this point without + # receiving the last thread update that was already returned. + next_token = None + if next_token_int is not None: + next_token = StreamToken.START.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=next_token_int - 1) + ) + # Optionally fetch thread root events - if include_thread_roots and thread_ids: - thread_root_ids = [thread_id for thread_id, _ in thread_ids] + event_map = {} + if include_thread_roots and thread_infos: + thread_root_ids = [thread_id for thread_id, _, _, _ in thread_infos] 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, + # Build ThreadUpdateInfo objects with per-thread prev_batch tokens. + thread_update_infos = [] + for thread_id, room_id, stream_ordering, update_count in thread_infos: + # Generate prev_batch token if this thread has more than one update. + per_thread_prev_batch = None + if update_count > 1: + # Create a token pointing to one position before the latest event's + # stream position. + # This makes it exclusive - /relations with dir=b won't return the + # latest event again. + # Use StreamToken.START as base (all other streams at 0) since only room + # position matters. + per_thread_prev_batch = StreamToken.START.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=stream_ordering - 1) + ) + + thread_update_infos.append( + ThreadUpdateInfo( + thread_id=thread_id, + room_id=room_id, + thread_root_event=event_map.get(thread_id), + prev_batch=per_thread_prev_batch, + ) ) + return (thread_update_infos, next_token) + class RelationsStore(RelationsWorkerStore): pass diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index 7c7a00b1d9..3a7a50066d 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -401,28 +401,47 @@ class SlidingSyncResult: @attr.s(slots=True, frozen=True, auto_attribs=True) class ThreadsExtension: - # TODO: comment """The Threads extension (MSC4360) + Provides thread updates for threads that have new activity across all of the + user's joined rooms within the sync window. + Attributes: + updates: A nested mapping of room_id -> thread_root_id -> ThreadUpdate. + Each ThreadUpdate contains information about a thread that has new activity, + including the thread root event (if requested) and a pagination token + for fetching older events in that specific thread. + prev_batch: A pagination token for fetching more thread updates across all rooms. + If present, indicates there are more thread updates available beyond what + was returned in this response. This token can be used with a future request + to paginate through older thread updates. """ @attr.s(slots=True, frozen=True, auto_attribs=True) class ThreadUpdate: - # The thread root event, if requested via include_roots + """Information about a single thread that has new activity. + + Attributes: + thread_root: The thread root event, if requested via include_roots in the + request. This is the event that started the thread. + prev_batch: A pagination token (exclusive) for fetching older events in this + specific thread. Only present if the thread has multiple updates in the + sync window. This token can be used with the /relations endpoint with + dir=b to paginate backwards through the thread's history. + bundled_aggregations: Bundled aggregations for the thread root event, + including the latest_event in the thread (found in + unsigned.m.relations.m.thread). Only present if thread_root is included. + """ + thread_root: Optional[EventBase] - - # 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) updates: Optional[Mapping[str, Mapping[str, ThreadUpdate]]] - prev_batch: Optional[int] + prev_batch: Optional[StreamToken] def __bool__(self) -> bool: return bool(self.updates) or bool(self.prev_batch) @@ -893,6 +912,7 @@ class PerConnectionState: Attributes: rooms: The status of each room for the events stream. receipts: The status of each room for the receipts stream. + account_data: The status of each room for the account data stream. room_configs: Map from room_id to the `RoomSyncConfig` of all rooms that we have previously sent down. """ diff --git a/synapse/types/rest/client/__init__.py b/synapse/types/rest/client/__init__.py index e8f08e434a..3e98fb3def 100644 --- a/synapse/types/rest/client/__init__.py +++ b/synapse/types/rest/client/__init__.py @@ -380,9 +380,9 @@ class SlidingSyncBody(RequestBodyModel): """The Threads extension (MSC4360) Attributes: - enabled + enabled: Whether the threads extension is enabled. include_roots: whether to include thread root events in the extension response. - limit: maximum number of thread updates to return. + limit: maximum number of thread updates to return across all joined rooms. """ enabled: Optional[StrictBool] = False diff --git a/tests/rest/client/sliding_sync/test_extension_threads.py b/tests/rest/client/sliding_sync/test_extension_threads.py index e57bb39f1c..50e34993a7 100644 --- a/tests/rest/client/sliding_sync/test_extension_threads.py +++ b/tests/rest/client/sliding_sync/test_extension_threads.py @@ -17,7 +17,7 @@ from twisted.test.proto_helpers import MemoryReactor import synapse.rest.admin from synapse.api.constants import RelationTypes -from synapse.rest.client import login, room, sync +from synapse.rest.client import login, relations, room, sync from synapse.server import HomeServer from synapse.types import JsonDict from synapse.util.clock import Clock @@ -43,8 +43,7 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): login.register_servlets, room.register_servlets, sync.register_servlets, - # TODO: - # threads.register_servlets, + relations.register_servlets, ] def default_config(self) -> JsonDict: @@ -64,7 +63,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -84,16 +82,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") - initial_sync_body: JsonDict = { - "lists": {}, - } + initial_sync_body: JsonDict = {} # Initial sync response_body, sync_pos = self.do_sync(initial_sync_body, tok=user1_tok) # Incremental sync with extension enabled sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -139,7 +134,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # base = self.store.get_max_thread_subscriptions_stream_id() sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -164,7 +158,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): user1_tok = self.login(user1_id, "pass") room_id = self.helper.create_room_as(user1_id, tok=user1_tok) sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -263,7 +256,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # User2 syncs with threads extension enabled sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -304,7 +296,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Initial sync for user2 sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -389,7 +380,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Sync with include_roots=True sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -444,7 +434,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Sync with include_roots=False (explicitly) sync_body = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -462,7 +451,6 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Also test with include_roots omitted (should behave the same) sync_body_no_param = { - "lists": {}, "extensions": { EXT_NAME: { "enabled": True, @@ -475,3 +463,261 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): "updates" ][room_id][thread_root_id] self.assertNotIn("thread_root", thread_update_no_param) + + def test_per_thread_prev_batch_single_update(self) -> None: + """ + Test that threads with only a single update do NOT get a prev_batch token. + """ + 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"] + + # Initial sync to establish baseline + sync_body = { + "extensions": { + EXT_NAME: { + "enabled": True, + } + }, + } + _, sync_pos = self.do_sync(sync_body, tok=user1_tok) + + # Add ONE reply to thread + self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Single reply", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + + # Incremental sync + response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos) + + # Assert: Thread update should NOT have prev_batch (only 1 update) + thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + thread_root_id + ] + self.assertNotIn( + "prev_batch", + thread_update, + "Threads with single update should not have prev_batch", + ) + + def test_per_thread_prev_batch_multiple_updates(self) -> None: + """ + Test that threads with multiple updates get a prev_batch token that can be + used with /relations endpoint to paginate backwards. + """ + 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"] + + # Initial sync to establish baseline + sync_body = { + "extensions": { + EXT_NAME: { + "enabled": True, + } + }, + } + _, sync_pos = self.do_sync(sync_body, tok=user1_tok) + + # Add MULTIPLE replies to thread + reply1_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "First reply", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply1_id = reply1_resp["event_id"] + + reply2_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Second reply", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply2_id = reply2_resp["event_id"] + + reply3_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Third reply", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply3_id = reply3_resp["event_id"] + + # Incremental sync + response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos) + + # Assert: Thread update SHOULD have prev_batch (3 updates) + thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + thread_root_id + ] + self.assertIn( + "prev_batch", + thread_update, + "Threads with multiple updates should have prev_batch", + ) + + prev_batch = thread_update["prev_batch"] + self.assertIsNotNone(prev_batch, "prev_batch should not be None") + + # Now use the prev_batch token with /relations endpoint to paginate backwards + channel = self.make_request( + "GET", + f"/_matrix/client/v1/rooms/{room_id}/relations/{thread_root_id}?from={prev_batch}&dir=b", + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + relations_response = channel.json_body + returned_event_ids = [ + event["event_id"] for event in relations_response["chunk"] + ] + + # Assert: Only the older replies should be returned (not the latest one we already saw) + # The prev_batch token should be exclusive, pointing just before the latest event + self.assertIn( + reply1_id, + returned_event_ids, + "First reply should be in relations response", + ) + self.assertIn( + reply2_id, + returned_event_ids, + "Second reply should be in relations response", + ) + self.assertNotIn( + reply3_id, + returned_event_ids, + "Third reply (latest) should NOT be in relations response - already returned in sliding sync", + ) + + def test_per_thread_prev_batch_on_initial_sync(self) -> None: + """ + Test that threads with multiple updates get prev_batch tokens on initial sync + so clients can paginate through the full thread history. + """ + 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 with multiple replies BEFORE any sync + thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok) + thread_root_id = thread_root_resp["event_id"] + + reply1_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Reply 1", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply1_id = reply1_resp["event_id"] + + reply2_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Reply 2", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply2_id = reply2_resp["event_id"] + + # Initial sync (no from_token) + sync_body = { + "extensions": { + EXT_NAME: { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # Assert: Thread update SHOULD have prev_batch on initial sync (2+ updates exist) + thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + thread_root_id + ] + self.assertIn( + "prev_batch", + thread_update, + "Threads with multiple updates should have prev_batch even on initial sync", + ) + + prev_batch = thread_update["prev_batch"] + self.assertIsNotNone(prev_batch) + + # Use prev_batch with /relations to fetch the thread history + channel = self.make_request( + "GET", + f"/_matrix/client/v1/rooms/{room_id}/relations/{thread_root_id}?from={prev_batch}&dir=b", + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + relations_response = channel.json_body + returned_event_ids = [ + event["event_id"] for event in relations_response["chunk"] + ] + + # Assert: Only the older reply should be returned (not the latest one we already saw) + # The prev_batch token should be exclusive, pointing just before the latest event + self.assertIn( + reply1_id, + returned_event_ids, + "First reply should be in relations response", + ) + self.assertNotIn( + reply2_id, + returned_event_ids, + "Second reply (latest) should NOT be in relations response - already returned in sliding sync", + )