dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-29 07:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Filter events from extension if in timeline

Browse Source
This commit is contained in:
Devon Hudson
2025-10-08 17:01:40 -06:00
parent ab7e5a2b17
commit 4d7826b006
2 changed files with 170 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/sliding_sync/extensions.py
+39 -1
View File
@@ -24,12 +24,13 @@ from typing import (
Optional,
Sequence,
Set,
Tuple,
cast,
)
from typing_extensions import TypeAlias, assert_never
from synapse.api.constants import AccountDataTypes, EduTypes
from synapse.api.constants import AccountDataTypes, EduTypes, RelationTypes
from synapse.handlers.receipts import ReceiptEventSource
from synapse.logging.opentracing import trace
from synapse.storage.databases.main.receipts import ReceiptInRoom
@@ -185,6 +186,7 @@ class SlidingSyncExtensionHandler:
threads_coro = self.get_threads_extension_response(
sync_config=sync_config,
threads_request=sync_config.extensions.threads,
actual_room_response_map=actual_room_response_map,
to_token=to_token,
from_token=from_token,
)
@@ -990,6 +992,7 @@ class SlidingSyncExtensionHandler:
self,
sync_config: SlidingSyncConfig,
threads_request: SlidingSyncConfig.Extensions.ThreadsExtension,
actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult],
to_token: StreamToken,
from_token: Optional[SlidingSyncStreamToken],
) -> Optional[SlidingSyncResult.Extensions.ThreadsExtension]:
@@ -998,6 +1001,9 @@ class SlidingSyncExtensionHandler:
Args:
sync_config: Sync configuration.
threads_request: The threads extension from the request.
actual_room_response_map: A map of room ID to room results in the
sliding sync response. Used to determine which threads already have
events in the room timeline.
to_token: The point in the stream to sync up to.
from_token: The point in the stream to sync from.
@@ -1024,6 +1030,29 @@ class SlidingSyncExtensionHandler:
if len(all_thread_updates) == 0:
return None
# Identify which threads already have events in the room timelines.
# If include_roots=False, we'll omit these threads from the extension response
# since the client already sees the thread activity in the timeline.
# If include_roots=True, we include all threads regardless, because the client
# wants the thread root events.
threads_in_timeline: Set[Tuple[str, str]] = set() # (room_id, thread_id)
if not threads_request.include_roots:
for room_id, room_result in actual_room_response_map.items():
if room_result.timeline_events:
for event in room_result.timeline_events:
# Check if this event is part of a thread
relates_to = event.content.get("m.relates_to")
if not isinstance(relates_to, dict):
continue
rel_type = relates_to.get("rel_type")
# If this is a thread reply, track the thread
if rel_type == RelationTypes.THREAD:
thread_id = relates_to.get("event_id")
if thread_id:
threads_in_timeline.add((room_id, thread_id))
# 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 = [
@@ -1040,6 +1069,11 @@ class SlidingSyncExtensionHandler:
thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {}
for update in all_thread_updates:
# Skip this thread if it already has events in the room timeline
# (unless include_roots=True, in which case we always include it)
if (update.room_id, update.thread_id) in threads_in_timeline:
continue
# Only look up bundled aggregations if we have a thread root event
bundled_aggs = (
aggregations_map.get(update.thread_id)
@@ -1055,6 +1089,10 @@ class SlidingSyncExtensionHandler:
)
)
# If after filtering we have no thread updates, return None to omit the extension
if not thread_updates:
return None
return SlidingSyncResult.Extensions.ThreadsExtension(
updates=thread_updates,
prev_batch=prev_batch_token,
tests/rest/client/sliding_sync/test_extension_threads.py
+131 -18
View File
@@ -587,16 +587,9 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
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][
prev_batch = 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"]
]["prev_batch"]
self.assertIsNotNone(prev_batch, "prev_batch should not be None")
# Now use the prev_batch token with /relations endpoint to paginate backwards
@@ -684,16 +677,9 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
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][
prev_batch = 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"]
]["prev_batch"]
self.assertIsNotNone(prev_batch)
# Use prev_batch with /relations to fetch the thread history
@@ -721,3 +707,130 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase):
returned_event_ids,
"Second reply (latest) should NOT be in relations response - already returned in sliding sync",
)
def test_thread_in_timeline_omitted_without_include_roots(self) -> None:
"""
Test that threads with events in the room timeline are omitted from the
extension response when include_roots=False. When all threads are filtered out,
the entire extension should be omitted from the response.
"""
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: JsonDict = {
"lists": {
"foo-list": {
"ranges": [[0, 1]],
"required_state": [],
"timeline_limit": 5,
}
},
"extensions": {
EXT_NAME: {
"enabled": True,
"include_roots": False,
}
},
}
_, sync_pos = self.do_sync(sync_body, tok=user1_tok)
# Send a reply to the thread
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,
)
# Incremental sync - the reply should be in the timeline
response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos)
# Assert: Extension should be omitted entirely since the only thread with updates
# is already visible in the timeline (include_roots=False)
self.assertNotIn(
EXT_NAME,
response_body.get("extensions", {}),
"Extension should be omitted when all threads are filtered out (in timeline with include_roots=False)",
)
def test_thread_in_timeline_included_with_include_roots(self) -> None:
"""
Test that threads with events in the room timeline are still included in the
extension response when include_roots=True, because the client wants the root event.
"""
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: JsonDict = {
"lists": {
"foo-list": {
"ranges": [[0, 1]],
"required_state": [],
"timeline_limit": 5,
}
},
"extensions": {
EXT_NAME: {
"enabled": True,
"include_roots": True,
}
},
}
_, sync_pos = self.do_sync(sync_body, tok=user1_tok)
# Send a reply to the thread
reply_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,
)
reply_id = reply_resp["event_id"]
# Incremental sync - the reply should be in the timeline
response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos)
# Assert: The thread reply should be in the room timeline
room_response = response_body["rooms"][room_id]
timeline_event_ids = [event["event_id"] for event in room_response["timeline"]]
self.assertIn(
reply_id,
timeline_event_ids,
"Thread reply should be in the room timeline",
)
# Assert: Thread SHOULD be in extension (include_roots=True)
thread_updates = response_body["extensions"][EXT_NAME]["updates"][room_id]
self.assertIn(
thread_root_id,
thread_updates,
"Thread should be included in extension when include_roots=True, even if in timeline",
)
# Verify the thread root event is present
self.assertIn("thread_root", thread_updates[thread_root_id])