Do not consider events by ignored users for bundled aggregations

changelog.d/12235.bugfix

@@ -0,0 +1 @@
+Fix a long-standing bug where events from ignored users were still considered for bundled aggregations.

changelog.d/12338.bugfix

@@ -0,0 +1 @@
+Fix a long-standing bug where events from ignored users were still considered for bundled aggregations.

synapse/handlers/relations.py

@@ -12,15 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import TYPE_CHECKING, Dict, Iterable, Optional
+from typing import (
+    TYPE_CHECKING,
+    Collection,
+    Dict,
+    FrozenSet,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+)
 
 import attr
 from frozendict import frozendict
 
 from synapse.api.constants import RelationTypes
 from synapse.api.errors import SynapseError
 from synapse.events import EventBase
-from synapse.types import JsonDict, Requester, StreamToken
+from synapse.storage.databases.main.relations import _RelatedEvent
+from synapse.types import JsonDict, Requester, StreamToken, UserID
 from synapse.visibility import filter_events_for_client
 
 if TYPE_CHECKING:
@@ -115,6 +125,9 @@ async def get_relations(
         if event is None:
             raise SynapseError(404, "Unknown parent event.")
 
+        # Note that ignored users are not passed into get_relations_for_event
+        # below. Ignored users are handled in filter_events_for_client (and by
+        # not passing them in here we should get a better cache hit rate).
         related_events, next_token = await self._main_store.get_relations_for_event(
             event_id=event_id,
             event=event,
@@ -128,7 +141,9 @@ async def get_relations(
             to_token=to_token,
         )
 
-        events = await self._main_store.get_events_as_list(related_events)
+        events = await self._main_store.get_events_as_list(
+            [e.event_id for e in related_events]
+        )
 
         events = await filter_events_for_client(
             self._storage, user_id, events, is_peeking=(member_event_id is None)
@@ -162,16 +177,95 @@ async def get_relations(
 
         return return_value
 
+    async def get_relations_for_event(
+        self,
+        event_id: str,
+        event: EventBase,
+        room_id: str,
+        relation_type: str,
+        ignored_users: FrozenSet[str] = frozenset(),
+    ) -> Tuple[List[_RelatedEvent], Optional[StreamToken]]:
+        """Get a list of events which relate to an event, ordered by topological ordering.
+
+        Args:
+            event_id: Fetch events that relate to this event ID.
+            event: The matching EventBase to event_id.
+            room_id: The room the event belongs to.
+            relation_type: The type of relation.
+            ignored_users: The users ignored by the requesting user.
+
+        Returns:
+            List of event IDs that match relations requested. The rows are of
+            the form `{"event_id": "..."}`.
+        """
+
+        # Call the underlying storage method, which is cached.
+        related_events, next_token = await self._main_store.get_relations_for_event(
+            event_id, event, room_id, relation_type, direction="f"
+        )
+
+        # Filter out ignored users and convert to the expected format.
+        related_events = [
+            event for event in related_events if event.sender not in ignored_users
+        ]
+
+        return related_events, next_token
+
+    async def get_annotations_for_event(
+        self,
+        event_id: str,
+        room_id: str,
+        limit: int = 5,
+        ignored_users: FrozenSet[str] = frozenset(),
+    ) -> List[JsonDict]:
+        """Get a list of annotations on the event, grouped by event type and
+        aggregation key, sorted by count.
+
+        This is used e.g. to get the what and how many reactions have happend
+        on an event.
+
+        Args:
+            event_id: Fetch events that relate to this event ID.
+            room_id: The room the event belongs to.
+            limit: Only fetch the `limit` groups.
+            ignored_users: The users ignored by the requesting user.
+
+        Returns:
+            List of groups of annotations that match. Each row is a dict with
+            `type`, `key` and `count` fields.
+        """
+        # Get the base results for all users.
+        full_results = await self._main_store.get_aggregation_groups_for_event(
+            event_id, room_id, limit
+        )
+
+        # Then subtract off the results for any ignored users.
+        ignored_results = await self._main_store.get_aggregation_groups_for_users(
+            event_id, room_id, limit, ignored_users
+        )
+
+        filtered_results = []
+        for result in full_results:
+            key = (result["type"], result["key"])
+            if key in ignored_results:
+                result = result.copy()
+                result["count"] -= ignored_results[key]
+                if result["count"] <= 0:
+                    continue
+            filtered_results.append(result)
+
+        return filtered_results
+
     async def _get_bundled_aggregation_for_event(
-        self, event: EventBase, user_id: str
+        self, event: EventBase, ignored_users: FrozenSet[str]
     ) -> Optional[BundledAggregations]:
         """Generate bundled aggregations 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.
-            user_id: The user requesting the bundled aggregations.
+            ignored_users: The users ignored by the requesting user.
 
         Returns:
             The bundled aggregations for an event, if bundled aggregations are
@@ -194,18 +288,22 @@ async def _get_bundled_aggregation_for_event(
         # while others need more processing during serialization.
         aggregations = BundledAggregations()
 
-        annotations = await self._main_store.get_aggregation_groups_for_event(
-            event_id, room_id
+        annotations = await self.get_annotations_for_event(
+            event_id, room_id, ignored_users=ignored_users
         )
         if annotations:
             aggregations.annotations = {"chunk": annotations}
 
-        references, next_token = await self._main_store.get_relations_for_event(
-            event_id, event, room_id, RelationTypes.REFERENCE, direction="f"
+        references, next_token = await self.get_relations_for_event(
+            event_id,
+            event,
+            room_id,
+            RelationTypes.REFERENCE,
+            ignored_users=ignored_users,
         )
         if references:
             aggregations.references = {
-                "chunk": [{"event_id": event_id} for event_id in references]
+                "chunk": [{"event_id": event.event_id} for event in references]
             }
 
             if next_token:
@@ -216,6 +314,99 @@ async def _get_bundled_aggregation_for_event(
         # Store the bundled aggregations in the event metadata for later use.
         return aggregations
 
+    async def get_threads_for_events(
+        self, event_ids: Collection[str], user_id: str, ignored_users: FrozenSet[str]
+    ) -> Dict[str, _ThreadAggregation]:
+        """Get the bundled aggregations for threads for the requested events.
+
+        Args:
+            event_ids: Events to get aggregations for threads.
+            user_id: The user requesting the bundled aggregations.
+            ignored_users: The users ignored by the requesting user.
+
+        Returns:
+            A dictionary mapping event ID to the thread information.
+
+            May not contain a value for all requested event IDs.
+        """
+        user = UserID.from_string(user_id)
+
+        # Fetch thread summaries.
+        summaries = await self._main_store.get_thread_summaries(event_ids)
+
+        # Only fetch participated for a limited selection based on what had
+        # summaries.
+        thread_event_ids = [
+            event_id for event_id, summary in summaries.items() if summary
+        ]
+        participated = await self._main_store.get_threads_participated(
+            thread_event_ids, user_id
+        )
+
+        # Then subtract off the results for any ignored users.
+        ignored_results = await self._main_store.get_threaded_messages_per_user(
+            thread_event_ids, ignored_users
+        )
+
+        # A map of event ID to the thread aggregation.
+        results = {}
+
+        for event_id, summary in summaries.items():
+            if summary:
+                thread_count, latest_thread_event, edit = summary
+
+                # Subtract off the count of any ignored users.
+                for ignored_user in ignored_users:
+                    thread_count -= ignored_results.get((event_id, ignored_user), 0)
+
+                # This is gnarly, but if the latest event is from an ignored user,
+                # attempt to find one that isn't from an ignored user.
+                if latest_thread_event.sender in ignored_users:
+                    room_id = latest_thread_event.room_id
+
+                    # If the root event is not found, something went wrong, do
+                    # not include a summary of the thread.
+                    event = await self._event_handler.get_event(user, room_id, event_id)
+                    if event is None:
+                        continue
+
+                    potential_events, _ = await self.get_relations_for_event(
+                        event_id,
+                        event,
+                        room_id,
+                        RelationTypes.THREAD,
+                        ignored_users,
+                    )
+
+                    # If all found events are from ignored users, do not include
+                    # a summary of the thread.
+                    if not potential_events:
+                        continue
+
+                    # The *last* event returned is the one that is cared about.
+                    event = await self._event_handler.get_event(
+                        user, room_id, potential_events[-1].event_id
+                    )
+                    # It is unexpected that the event will not exist.
+                    if event is None:
+                        logger.warning(
+                            "Unable to fetch latest event in a thread with event ID: %s",
+                            potential_events[-1].event_id,
+                        )
+                        continue
+                    latest_thread_event = event
+
+                results[event_id] = _ThreadAggregation(
+                    latest_event=latest_thread_event,
+                    latest_edit=edit,
+                    count=thread_count,
+                    # If there's a thread summary it must also exist in the
+                    # participated dictionary.
+                    current_user_participated=participated[event_id],
+                )
+
+        return results
+
     async def get_bundled_aggregations(
         self, events: Iterable[EventBase], user_id: str
     ) -> Dict[str, BundledAggregations]:
@@ -239,13 +430,21 @@ async def get_bundled_aggregations(
         # event ID -> bundled aggregation in non-serialized form.
         results: Dict[str, BundledAggregations] = {}
 
+        # Fetch any ignored users of the requesting user.
+        ignored_users = await self._main_store.ignored_users(user_id)
+
         # Fetch other relations per event.
         for event in events_by_id.values():
-            event_result = await self._get_bundled_aggregation_for_event(event, user_id)
+            event_result = await self._get_bundled_aggregation_for_event(
+                event, ignored_users
+            )
             if event_result:
                 results[event.event_id] = event_result
 
         # Fetch any edits (but not for redacted events).
+        #
+        # Note that there is no use in limiting edits by ignored users since the
+        # parent event should be ignored in the first place if the user is ignored.
         edits = await self._main_store.get_applicable_edits(
             [
                 event_id
@@ -256,25 +455,10 @@ async def get_bundled_aggregations(
         for event_id, edit in edits.items():
             results.setdefault(event_id, BundledAggregations()).replace = edit
 
-        # Fetch thread summaries.
-        summaries = await self._main_store.get_thread_summaries(events_by_id.keys())
-        # Only fetch participated for a limited selection based on what had
-        # summaries.
-        participated = await self._main_store.get_threads_participated(
-            [event_id for event_id, summary in summaries.items() if summary], user_id
+        threads = await self.get_threads_for_events(
+            events_by_id.keys(), user_id, ignored_users
         )
-        for event_id, summary in summaries.items():
-            if summary:
-                thread_count, latest_thread_event, edit = summary
-                results.setdefault(
-                    event_id, BundledAggregations()
-                ).thread = _ThreadAggregation(
-                    latest_event=latest_thread_event,
-                    latest_edit=edit,
-                    count=thread_count,
-                    # If there's a thread summary it must also exist in the
-                    # participated dictionary.
-                    current_user_participated=participated[event_id],
-                )
+        for event_id, thread in threads.items():
+            results.setdefault(event_id, BundledAggregations()).thread = thread
 
         return results