Re-use get_threads_participated instead of custom SQL.

Author: clokep

synapse/handlers/relations.py

@@ -522,14 +522,28 @@ async def get_threads(
         # not passing them in here we should get a better cache hit rate).
         thread_roots, next_token = await self._main_store.get_threads(
             room_id=room_id,
-            participating_user_id=None if filter == "all" else user_id,
             limit=limit,
             from_token=from_token,
             to_token=to_token,
         )
 
         events = await self._main_store.get_events_as_list(thread_roots)
 
+        if filter == "participated":
+            # Pre-seed thread participation with whether the requester sent the event.
+            participated = {event.event_id: event.sender == user_id for event in events}
+            # For events the requester did not send, check the database for whether
+            # the requester sent a threaded reply.
+            participated.update(
+                await self._main_store.get_threads_participated(
+                    [eid for eid, p in participated.items() if not p],
+                    user_id,
+                )
+            )
+
+            # Limit the returned threads to those the user has participated in.
+            events = [event for event in events if participated[event.event_id]]
+
         events = await filter_events_for_client(
             self._storage_controllers,
             user_id,

synapse/storage/databases/main/relations.py

@@ -818,7 +818,6 @@ def _get_event_relations(
     async def get_threads(
         self,
         room_id: str,
-        participating_user_id: Optional[str] = None,
         limit: int = 5,
         from_token: Optional[StreamToken] = None,
         to_token: Optional[StreamToken] = None,
@@ -828,8 +827,6 @@ async def get_threads(
 
         Args:
             room_id: The room the event belongs to.
-            participating_user_id: A user who must have participated in the thread.
-                Provide None to return all threads.
             limit: Only fetch the most recent `limit` events.
             from_token: Fetch rows from the given token, or from the start if None.
             to_token: Fetch rows up to the given token, or up to the end if None.
@@ -840,21 +837,9 @@ async def get_threads(
 
                 The next stream token, if one exists.
         """
-        where_clause = [
-            "child.room_id = ?",
-            f"relation_type = '{RelationTypes.THREAD}'",
-        ]
-        where_args: List[Union[str, int]] = [room_id]
-
-        # The user has participated if they sent the thread root or have any reply
-        # to the thread.
-        if participating_user_id:
-            where_clause.append("parent.sender = ?")
-            where_args.append(participating_user_id)
-
         pagination_clause = generate_pagination_where_clause(
             direction="b",
-            column_names=("child.topological_ordering", "child.stream_ordering"),
+            column_names=("topological_ordering", "stream_ordering"),
             from_token=from_token.room_key.as_historical_tuple()
             if from_token
             else None,
@@ -863,25 +848,25 @@ async def get_threads(
         )
 
         if pagination_clause:
-            where_clause.append(pagination_clause)
+            pagination_clause = "AND " + pagination_clause
 
         sql = f"""
-            SELECT relates_to_id, MAX(child.topological_ordering), MAX(child.stream_ordering)
-            FROM events AS child
-            INNER JOIN event_relations USING (event_id)
-            INNER JOIN events AS parent ON
-                parent.event_id = relates_to_id
-                AND parent.room_id = child.room_id
-            WHERE {" AND ".join(where_clause)}
+            SELECT relates_to_id, MAX(topological_ordering), MAX(stream_ordering)
+            FROM event_relations
+            INNER JOIN events USING (event_id)
+            WHERE
+                room_id = ? AND
+                relation_type = '{RelationTypes.THREAD}'
+                {pagination_clause}
             GROUP BY relates_to_id
-            ORDER BY child.topological_ordering DESC, child.stream_ordering DESC
+            ORDER BY topological_ordering DESC, stream_ordering DESC
             LIMIT ?
         """
 
         def _get_threads_txn(
             txn: LoggingTransaction,
         ) -> Tuple[List[str], Optional[StreamToken]]:
-            txn.execute(sql, where_args + [limit + 1])
+            txn.execute(sql, [room_id, limit + 1])
 
             last_topo_id = None
             last_stream_id = None

tests/rest/client/test_relations.py

@@ -1752,7 +1752,7 @@ def test_pagination(self) -> None:
 
     @unittest.override_config({"experimental_features": {"msc3856_enabled": True}})
     def test_filter(self) -> None:
-        """Create threads and ensure the ordering is due to their latest event."""
+        """Filtering threads to all or participated in should work."""
         # Thread 1 has the user as the root event.
         thread_1 = self.parent_id
         self._send_relation(