Include bundled aggregations for the latest event in a thread

synapse/events/utils.py

@@ -479,7 +479,12 @@ def _inject_bundled_aggregations(
             event: The event being serialized.
             time_now: The current time in milliseconds
             config: Event serialization config
-            bundled_aggregations: The bundled aggregation to serialize.
+            bundled_aggregations: Bundled aggregations to be injected.
+                A map from event_id to aggregation data. Must contain at least an
+                entry for `event`.
+
+                While serializing the bundled aggregations this map may be searched
+                again for additional events in a recursive manner.
             serialized_event: The serialized event which may be modified.
             apply_edits: Whether the content of the event should be modified to reflect
                any replacement in `aggregations.replace`.
@@ -516,15 +521,7 @@ def _inject_bundled_aggregations(
             }
 
         # Include any threaded replies to this event.
-        #
-        # Note that is it not valid to start a thread on an event which already
-        # has a relation. Doing so would cause an infinite loop.
-        event_relation = event.content.get("m.relates_to")
-        include_thread = (
-            not isinstance(event_relation, (dict, frozendict))
-            or event_relation.get("rel_type") is None
-        )
-        if include_thread and event_aggregations.thread:
+        if event_aggregations.thread:
             thread = event_aggregations.thread
 
             serialized_latest_event = self.serialize_event(

synapse/handlers/relations.py

@@ -254,63 +254,6 @@ async def get_annotations_for_event(
 
         return filtered_results
 
-    async def _get_bundled_annotations_and_references_for_event(
-        self, event: EventBase, ignored_users: FrozenSet[str]
-    ) -> Tuple[Optional[JsonDict], Optional[JsonDict]]:
-        """
-        Generate bundled aggregations for annotation and reference relations for an event.
-
-        Note that this does not use a cache, but depends on cached methods.
-
-        Args:
-            event: The event to calculate bundled aggregations for.
-            ignored_users: The users ignored by the requesting user.
-
-        Returns:
-            A tuple of the bundled aggregations for annotation and reference relations.
-            Either or both entries in the tuple might be None if no relations
-            of that type exist.
-        """
-
-        # Do not bundle aggregations for an event which represents an edit or an
-        # annotation. It does not make sense for them to have related events.
-        relates_to = event.content.get("m.relates_to")
-        if isinstance(relates_to, (dict, frozendict)):
-            relation_type = relates_to.get("rel_type")
-            if relation_type in (RelationTypes.ANNOTATION, RelationTypes.REPLACE):
-                return None, None
-
-        event_id = event.event_id
-        room_id = event.room_id
-
-        annotations = await self.get_annotations_for_event(
-            event_id, room_id, ignored_users=ignored_users
-        )
-        serialized_annotations = None
-        if annotations:
-            serialized_annotations = {"chunk": annotations}
-
-        references, next_token = await self.get_relations_for_event(
-            event_id,
-            event,
-            room_id,
-            RelationTypes.REFERENCE,
-            ignored_users=ignored_users,
-        )
-        serialized_references: Optional[JsonDict] = None
-        if references:
-            serialized_references = {
-                "chunk": [{"event_id": event.event_id} for event in references]
-            }
-
-            if next_token:
-                serialized_references["next_batch"] = await next_token.to_string(
-                    self._main_store
-                )
-
-        # Store the bundled aggregations in the event metadata for later use.
-        return serialized_annotations, serialized_references
-
     async def get_threads_for_events(
         self, event_ids: Collection[str], user_id: str, ignored_users: FrozenSet[str]
     ) -> Dict[str, _ThreadAggregation]:
@@ -428,6 +371,15 @@ async def get_bundled_aggregations(
             event.event_id: event for event in events if not event.is_state()
         }
 
+        # A map of event ID to the relation in that event, if there is one.
+        relations_by_id = {}
+        for event_id, event in events_by_id.items():
+            relates_to = event.content.get("m.relates_to")
+            if isinstance(relates_to, (dict, frozendict)):
+                relation_type = relates_to.get("rel_type")
+                if relation_type is not None:
+                    relations_by_id[event_id] = relation_type
+
         # event ID -> bundled aggregation in non-serialized form.
         results: Dict[str, BundledAggregations] = {}
 
@@ -438,11 +390,11 @@ async def get_bundled_aggregations(
         # events to be fetched. Thus, we check those first!
 
         # Fetch thread summaries (but only for the directly requested events).
-        #
-        # Note that you can't have threads off of other related events, but it is
-        # possible for a malicious homeserver to inject them anyway.
         threads = await self.get_threads_for_events(
-            events_by_id.keys(), user_id, ignored_users
+            # It is not valid to start a thread on an event which already has a relation.
+            [eid for eid in events_by_id.keys() if eid not in relations_by_id],
+            user_id,
+            ignored_users,
         )
         for event_id, thread in threads.items():
             results.setdefault(event_id, BundledAggregations()).thread = thread
@@ -453,19 +405,43 @@ async def get_bundled_aggregations(
             latest_thread_event = thread.latest_event
             if latest_thread_event and latest_thread_event.event_id not in events_by_id:
                 events_by_id[latest_thread_event.event_id] = latest_thread_event
+                # The latest event in the thread must have a thread relation.
+                relations_by_id[latest_thread_event.event_id] = RelationTypes.THREAD
 
         # Fetch other relations per event.
         for event in events_by_id.values():
-            (
-                annotations,
-                references,
-            ) = await self._get_bundled_annotations_and_references_for_event(
-                event, ignored_users
+            # Edits and annotations may not have related annotations or references.
+            if relations_by_id.get(event.event_id) in (
+                RelationTypes.ANNOTATION,
+                RelationTypes.REPLACE,
+            ):
+                continue
+
+            annotations = await self.get_annotations_for_event(
+                event.event_id, event.room_id, ignored_users=ignored_users
+            )
+            if annotations:
+                results.setdefault(
+                    event.event_id, BundledAggregations()
+                ).annotations = {"chunk": annotations}
+
+            references, next_token = await self.get_relations_for_event(
+                event.event_id,
+                event,
+                event.room_id,
+                RelationTypes.REFERENCE,
+                ignored_users=ignored_users,
             )
-            if annotations or references:
+            if references:
                 aggregations = results.setdefault(event.event_id, BundledAggregations())
-                aggregations.annotations = annotations
-                aggregations.references = references
+                aggregations.references = {
+                    "chunk": [{"event_id": ev.event_id} for ev in references]
+                }
+
+                if next_token:
+                    aggregations.references["next_batch"] = await next_token.to_string(
+                        self._main_store
+                    )
 
         # Fetch any edits (but not for redacted events).
         #

tests/rest/client/test_relations.py

@@ -1031,7 +1031,6 @@ def assert_thread(bundled_aggregations: JsonDict) -> None:
 
         self._test_bundled_aggregations(RelationTypes.THREAD, assert_thread, 10)
 
-    @unittest.override_config({"experimental_features": {"msc3666_enabled": True}})
     def test_thread_with_bundled_aggregations_for_latest(self) -> None:
         """
         Bundled aggregations should get applied to the latest thread event.
@@ -1078,32 +1077,29 @@ def assert_thread(bundled_aggregations: JsonDict) -> None:
 
         self._test_bundled_aggregations(RelationTypes.THREAD, assert_thread, 10)
 
-    def test_thread_loop(self) -> None:
-        """Ensure that bogus events do not cause the bundled aggregations code to iterate forever."""
-        last_event = self.parent_id
+    def test_nested_thread(self) -> None:
+        """
+        Ensure that a nested thread gets ignored by bundled aggregations, as
+        those are forbidden.
+        """
 
-        # Disable the validation to pretend this came over federation.
+        # Start a thread.
+        channel = self._send_relation(RelationTypes.THREAD, "m.room.test")
+        reply_event_id = channel.json_body["event_id"]
+
+        # Disable the validation to pretend this came over federation, since it is
+        # not an event the Client-Server API will allow..
         with patch(
             "synapse.handlers.message.EventCreationHandler._validate_event_relation",
             new=lambda self, event: make_awaitable(None),
         ):
-            for _ in range(2):
-                channel = self._send_relation(
-                    RelationTypes.THREAD, "m.room.test", parent_id=last_event
-                )
-                last_event = channel.json_body["event_id"]
+            # Create a sub-thread off the thread, which is not allowed.
+            self._send_relation(
+                RelationTypes.THREAD, "m.room.test", parent_id=reply_event_id
+            )
 
         # Fetch the thread root, to get the bundled aggregation for the thread.
-        relations = self._get_bundled_aggregations()
-
-        # The latest event should have bundled aggregations.
-        self.assertIn(RelationTypes.THREAD, relations)
-        thread_summary = relations[RelationTypes.THREAD]
-        self.assertIn("latest_event", thread_summary)
-
-        # The latest event should not have any bundled aggregations (since it
-        # only has a thread attached.
-        self.assertNotIn("m.relations", thread_summary["latest_event"]["unsigned"])
+        relations_from_event = self._get_bundled_aggregations()
 
         # Ensure that requesting the room messages also does not return the sub-thread.
         channel = self.make_request(
@@ -1113,16 +1109,26 @@ def test_thread_loop(self) -> None:
         )
         self.assertEqual(200, channel.code, channel.json_body)
         event = self._find_event_in_chunk(channel.json_body["chunk"])
-        relations = event["unsigned"]["m.relations"]
+        relations_from_messages = event["unsigned"]["m.relations"]
 
-        # The latest event should have bundled aggregations.
-        self.assertIn(RelationTypes.THREAD, relations)
-        thread_summary = relations[RelationTypes.THREAD]
-        self.assertIn("latest_event", thread_summary)
+        # Check the bundled aggregations from each point.
+        for aggregations, desc in (
+            (relations_from_event, "/event"),
+            (relations_from_messages, "/messages"),
+        ):
+            # The latest event should have bundled aggregations.
+            self.assertIn(RelationTypes.THREAD, aggregations, desc)
+            thread_summary = aggregations[RelationTypes.THREAD]
+            self.assertIn("latest_event", thread_summary, desc)
+            self.assertEqual(
+                thread_summary["latest_event"]["event_id"], reply_event_id, desc
+            )
 
-        # The latest event should not have any bundled aggregations (since it
-        # only has a thread attached.
-        self.assertNotIn("m.relations", thread_summary["latest_event"]["unsigned"])
+            # The latest event should not have any bundled aggregations (since the
+            # only relation to it is another thread, which is invalid).
+            self.assertNotIn(
+                "m.relations", thread_summary["latest_event"]["unsigned"], desc
+            )
 
     def test_thread_edit_latest_event(self) -> None:
         """Test that editing the latest event in a thread works."""
@@ -1145,23 +1151,18 @@ def test_thread_edit_latest_event(self) -> None:
         edit_event_id = channel.json_body["event_id"]
 
         # Fetch the thread root, to get the bundled aggregation for the thread.
-        channel = self.make_request(
-            "GET",
-            f"/rooms/{self.room}/event/{self.parent_id}",
-            access_token=self.user_token,
-        )
-        self.assertEqual(200, channel.code, channel.json_body)
+        relations_dict = self._get_bundled_aggregations()
 
         # We expect that the edit message appears in the thread summary in the
         # unsigned relations section.
-        relations_dict = channel.json_body["unsigned"].get("m.relations")
         self.assertIn(RelationTypes.THREAD, relations_dict)
 
         thread_summary = relations_dict[RelationTypes.THREAD]
         self.assertIn("latest_event", thread_summary)
         latest_event_in_thread = thread_summary["latest_event"]
         self.assertEqual(latest_event_in_thread["content"]["body"], "I've been edited!")
-        # The event should also have edit appear under the bundled aggregations.
+        # The latest event in the thread should have the edit appear under the
+        # bundled aggregations.
         self.assertDictContainsSubset(
             {"event_id": edit_event_id, "sender": "@alice:test"},
             latest_event_in_thread["unsigned"]["m.relations"][RelationTypes.REPLACE],