Allow limiting threads by participation.

Author: clokep

synapse/handlers/relations.py

@@ -485,6 +485,7 @@ async def get_threads(
         self,
         requester: Requester,
         room_id: str,
+        filter: str,
         limit: int = 5,
         from_token: Optional[StreamToken] = None,
         to_token: Optional[StreamToken] = None,
@@ -497,7 +498,11 @@ async def get_threads(
             requester: The user requesting the relations.
             event_id: Fetch events that relate to this event ID.
             room_id: The room the event belongs to.
+            filter: One of "all" or "participated" to indicate which threads should
+                be returned.
             limit: Only fetch the most recent `limit` events.
+            include_all: True if all threads should be included, false to limit
+                to only threads the user has participated in.
             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.
 
@@ -516,7 +521,11 @@ async def get_threads(
         # 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).
         thread_roots, next_token = await self._main_store.get_threads(
-            room_id=room_id, limit=limit, from_token=from_token, to_token=to_token
+            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)

synapse/rest/client/relations.py

@@ -113,6 +113,9 @@ async def on_GET(
         limit = parse_integer(request, "limit", default=5)
         from_token_str = parse_string(request, "from")
         to_token_str = parse_string(request, "to")
+        filter_str = parse_string(
+            request, "filter", default="all", allowed_values=["all", "participated"]
+        )
 
         # Return the relations
         from_token = None
@@ -125,6 +128,7 @@ async def on_GET(
         result = await self._relations_handler.get_threads(
             requester=requester,
             room_id=room_id,
+            filter=filter_str,
             limit=limit,
             from_token=from_token,
             to_token=to_token,

synapse/storage/databases/main/relations.py

@@ -818,6 +818,7 @@ 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,
@@ -827,6 +828,8 @@ 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.
@@ -837,9 +840,21 @@ 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=("topological_ordering", "stream_ordering"),
+            column_names=("child.topological_ordering", "child.stream_ordering"),
             from_token=from_token.room_key.as_historical_tuple()
             if from_token
             else None,
@@ -848,25 +863,25 @@ async def get_threads(
         )
 
         if pagination_clause:
-            pagination_clause = "AND " + pagination_clause
+            where_clause.append(pagination_clause)
 
         sql = f"""
-            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}
+            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)}
             GROUP BY relates_to_id
-            ORDER BY topological_ordering DESC, stream_ordering DESC
+            ORDER BY child.topological_ordering DESC, child.stream_ordering DESC
             LIMIT ?
         """
 
         def _get_threads_txn(
             txn: LoggingTransaction,
         ) -> Tuple[List[str], Optional[StreamToken]]:
-            txn.execute(sql, [room_id, limit + 1])
+            txn.execute(sql, where_args + [limit + 1])
 
             last_topo_id = None
             last_stream_id = None

tests/rest/client/test_relations.py

@@ -1750,4 +1750,50 @@ def test_pagination(self) -> None:
 
         self.assertNotIn("next_batch", channel.json_body, channel.json_body)
 
+    @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."""
+        # Thread 1 has the user as the root event.
+        thread_1 = self.parent_id
+        self._send_relation(
+            RelationTypes.THREAD, "m.room.test", access_token=self.user2_token
+        )
+
+        # Thread 2 has the user replying.
+        res = self.helper.send(self.room, body="Thread Root!", tok=self.user2_token)
+        thread_2 = res["event_id"]
+        self._send_relation(RelationTypes.THREAD, "m.room.test", parent_id=thread_2)
+
+        # Thread 3 has the user not participating in.
+        res = self.helper.send(self.room, body="Another thread!", tok=self.user2_token)
+        thread_3 = res["event_id"]
+        self._send_relation(
+            RelationTypes.THREAD,
+            "m.room.test",
+            access_token=self.user2_token,
+            parent_id=thread_3,
+        )
+
+        # All threads in the room.
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/unstable/org.matrix.msc3856/rooms/{self.room}/threads",
+            access_token=self.user_token,
+        )
+        self.assertEquals(200, channel.code, channel.json_body)
+        thread_roots = [ev["event_id"] for ev in channel.json_body["chunk"]]
+        self.assertEqual(
+            thread_roots, [thread_3, thread_2, thread_1], channel.json_body
+        )
+
+        # Only participated threads.
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/unstable/org.matrix.mscxxxx/rooms/{self.room}/threads?filter=participated",
+            access_token=self.user_token,
+        )
+        self.assertEquals(200, channel.code, channel.json_body)
+        thread_roots = [ev["event_id"] for ev in channel.json_body["chunk"]]
+        self.assertEqual(thread_roots, [thread_2, thread_1], channel.json_body)
+
     # XXX Test ignoring users.