Support thread IDs on receipts (implement MSC3771)

changelog.d/13202.feature

@@ -0,0 +1 @@
+Experimental support for thread-specific receipts ([MSC3771](https://github.com/matrix-org/matrix-spec-proposals/pull/3771)).

synapse/config/experimental.py

@@ -82,6 +82,8 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
         # MSC3786 (Add a default push rule to ignore m.room.server_acl events)
         self.msc3786_enabled: bool = experimental.get("msc3786_enabled", False)
 
+        # MSC3771: Thread read receipts
+        self.msc3771_enabled: bool = experimental.get("msc3771_enabled", False)
         # MSC3772: A push rule for mutual relations.
         self.msc3772_enabled: bool = experimental.get("msc3772_enabled", False)
         # MSC3773: Thread notifications

synapse/handlers/receipts.py

@@ -91,12 +91,22 @@ async def _received_remote_receipt(self, origin: str, content: JsonDict) -> None
                         )
                         continue
 
+                    # Check if these receipts apply to a thread.
+                    thread_id = None
+                    data = user_values.get("data", {})
+                    if isinstance(data, dict):
+                        thread_id = data.get("thread_id")
+                        # If the thread ID is invalid, conider it missing.
+                        if not thread_id or not isinstance(thread_id, str):
+                            thread_id = None
+
                     receipts.append(
                         ReadReceipt(
                             room_id=room_id,
                             receipt_type=receipt_type,
                             user_id=user_id,
                             event_ids=user_values["event_ids"],
+                            thread_id=thread_id,
                             data=user_values.get("data", {}),
                         )
                     )
@@ -114,6 +124,7 @@ async def _handle_new_receipts(self, receipts: List[ReadReceipt]) -> bool:
                 receipt.receipt_type,
                 receipt.user_id,
                 receipt.event_ids,
+                receipt.thread_id,
                 receipt.data,
             )
 
@@ -146,7 +157,12 @@ async def _handle_new_receipts(self, receipts: List[ReadReceipt]) -> bool:
         return True
 
     async def received_client_receipt(
-        self, room_id: str, receipt_type: str, user_id: str, event_id: str
+        self,
+        room_id: str,
+        receipt_type: str,
+        user_id: str,
+        event_id: str,
+        thread_id: Optional[str],
     ) -> None:
         """Called when a client tells us a local user has read up to the given
         event_id in the room.
@@ -156,6 +172,7 @@ async def received_client_receipt(
             receipt_type=receipt_type,
             user_id=user_id,
             event_ids=[event_id],
+            thread_id=thread_id,
             data={"ts": int(self.clock.time_msec())},
         )
 

synapse/replication/tcp/client.py

@@ -426,7 +426,8 @@ async def _on_new_receipts(
                 receipt.receipt_type,
                 receipt.user_id,
                 [receipt.event_id],
-                receipt.data,
+                thread_id=receipt.thread_id,
+                data=receipt.data,
             )
             await self.federation_sender.send_read_receipt(receipt_info)
 

synapse/replication/tcp/streams/_base.py

@@ -361,6 +361,7 @@ class ReceiptsStreamRow:
         receipt_type: str
         user_id: str
         event_id: str
+        thread_id: Optional[str]
         data: dict
 
     NAME = "receipts"

synapse/rest/client/read_marker.py

@@ -85,6 +85,8 @@ async def on_POST(
                     receipt_type,
                     user_id=requester.user.to_string(),
                     event_id=event_id,
+                    # Setting the thread ID is not possible with the /read_markers endpoint.
+                    thread_id=None,
                 )
 
         return 200, {}

synapse/rest/client/receipts.py

@@ -52,6 +52,8 @@ def __init__(self, hs: "HomeServer"):
         if hs.config.experimental.msc2285_enabled:
             self._known_receipt_types.add(ReceiptTypes.UNSTABLE_READ_PRIVATE)
 
+        self._msc3771_enabled = hs.config.experimental.msc3771_enabled
+
     async def on_POST(
         self, request: SynapseRequest, room_id: str, receipt_type: str, event_id: str
     ) -> Tuple[int, JsonDict]:
@@ -63,7 +65,17 @@ async def on_POST(
                 f"Receipt type must be {', '.join(self._known_receipt_types)}",
             )
 
-        parse_json_object_from_request(request, allow_empty_body=False)
+        body = parse_json_object_from_request(request)
+
+        # Pull the thread ID, if one exists.
+        thread_id = None
+        if self._msc3771_enabled:
+            if "thread_id" in body:
+                thread_id = body.get("thread_id")
+                if not thread_id or not isinstance(thread_id, str):
+                    raise SynapseError(
+                        400, "thread_id field must be a non-empty string"
+                    )
 
         await self.presence_handler.bump_presence_active_time(requester.user)
 
@@ -79,6 +91,7 @@ async def on_POST(
                 receipt_type,
                 user_id=requester.user.to_string(),
                 event_id=event_id,
+                thread_id=thread_id,
             )
 
         return 200, {}