Implement changes to MSC2285 (hidden read receipts)

changelog.d/12168.feature

@@ -0,0 +1 @@
+Implement [changes](https://github.com/matrix-org/matrix-spec-proposals/pull/2285/commits/4a77139249c2e830aec3c7d6bd5501a514d1cc27) to [MSC2285 (hidden read receipts)](https://github.com/matrix-org/matrix-spec-proposals/pull/2285). Contributed by @SimonBrandner.

synapse/api/constants.py

@@ -255,7 +255,5 @@ class GuestAccess:
 
 class ReceiptTypes:
     READ: Final = "m.read"
-
-
-class ReadReceiptEventFields:
-    MSC2285_HIDDEN: Final = "org.matrix.msc2285.hidden"
+    READ_PRIVATE: Final = "org.matrix.msc2285.read.private"
+    FULLY_READ: Final = "m.fully_read"

synapse/handlers/receipts.py

@@ -14,7 +14,7 @@
 import logging
 from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple
 
-from synapse.api.constants import ReadReceiptEventFields, ReceiptTypes
+from synapse.api.constants import ReceiptTypes
 from synapse.appservice import ApplicationService
 from synapse.streams import EventSource
 from synapse.types import JsonDict, ReadReceipt, UserID, get_domain_from_id
@@ -138,7 +138,7 @@ 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, hidden: bool
+        self, room_id: str, receipt_type: str, user_id: str, event_id: str
     ) -> None:
         """Called when a client tells us a local user has read up to the given
         event_id in the room.
@@ -148,15 +148,16 @@ async def received_client_receipt(
             receipt_type=receipt_type,
             user_id=user_id,
             event_ids=[event_id],
-            data={"ts": int(self.clock.time_msec()), "hidden": hidden},
+            data={"ts": int(self.clock.time_msec())},
         )
 
         is_new = await self._handle_new_receipts([receipt])
         if not is_new:
             return
 
         if self.federation_sender and not (
-            self.hs.config.experimental.msc2285_enabled and hidden
+            self.hs.config.experimental.msc2285_enabled
+            and receipt_type == ReceiptTypes.READ_PRIVATE
         ):
             await self.federation_sender.send_read_receipt(receipt)
 
@@ -178,35 +179,27 @@ def filter_out_hidden(events: List[JsonDict], user_id: str) -> List[JsonDict]:
 
             for event_id in content.keys():
                 event_content = content.get(event_id, {})
-                m_read = event_content.get(ReceiptTypes.READ, {})
 
-                # If m_read is missing copy over the original event_content as there is nothing to process here
-                if not m_read:
-                    new_event["content"][event_id] = event_content.copy()
+                m_read = event_content.get(ReceiptTypes.READ, None)
+                if m_read:
+                    new_event["content"][event_id] = {ReceiptTypes.READ: m_read}
                     continue
 
-                new_users = {}
-                for rr_user_id, user_rr in m_read.items():
-                    try:
-                        hidden = user_rr.get("hidden")
-                    except AttributeError:
-                        # Due to https://github.com/matrix-org/synapse/issues/10376
-                        # there are cases where user_rr is a string, in those cases
-                        # we just ignore the read receipt
-                        continue
+                m_read_private = event_content.get(ReceiptTypes.READ_PRIVATE, None)
+                if m_read_private:
+                    new_users = {}
+                    for rr_user_id, user_rr in m_read_private.items():
+                        if rr_user_id == user_id:
+                            new_users[rr_user_id] = user_rr.copy()
+
+                    # Set new users unless empty
+                    if len(new_users.keys()) > 0:
+                        new_event["content"][event_id] = {
+                            ReceiptTypes.READ_PRIVATE: new_users
+                        }
+                    continue
 
-                    if hidden is not True or rr_user_id == user_id:
-                        new_users[rr_user_id] = user_rr.copy()
-                        # If hidden has a value replace hidden with the correct prefixed key
-                        if hidden is not None:
-                            new_users[rr_user_id].pop("hidden")
-                            new_users[rr_user_id][
-                                ReadReceiptEventFields.MSC2285_HIDDEN
-                            ] = hidden
-
-                # Set new users unless empty
-                if len(new_users.keys()) > 0:
-                    new_event["content"][event_id] = {ReceiptTypes.READ: new_users}
+                new_event["content"][event_id] = event_content
 
             # Append new_event to visible_events unless empty
             if len(new_event["content"].keys()) > 0:

synapse/handlers/sync.py

@@ -28,7 +28,7 @@
 import attr
 from prometheus_client import Counter
 
-from synapse.api.constants import AccountDataTypes, EventTypes, Membership, ReceiptTypes
+from synapse.api.constants import AccountDataTypes, EventTypes, Membership
 from synapse.api.filtering import FilterCollection
 from synapse.api.presence import UserPresenceState
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
@@ -1065,7 +1065,6 @@ async def unread_notifs_for_room_id(
             last_unread_event_id = await self.store.get_last_receipt_event_id_for_user(
                 user_id=sync_config.user.to_string(),
                 room_id=room_id,
-                receipt_type=ReceiptTypes.READ,
             )
 
             return await self.store.get_unread_event_push_actions_by_room_for_user(

synapse/push/push_tools.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 from typing import Dict
 
-from synapse.api.constants import ReceiptTypes
 from synapse.events import EventBase
 from synapse.push.presentable_names import calculate_room_name, name_from_member_event
 from synapse.storage import Storage
@@ -24,7 +23,7 @@ async def get_badge_count(store: DataStore, user_id: str, group_by_room: bool) -
     invites = await store.get_invited_rooms_for_local_user(user_id)
     joins = await store.get_rooms_for_user(user_id)
 
-    my_receipts_by_room = await store.get_receipts_for_user(user_id, ReceiptTypes.READ)
+    my_receipts_by_room = await store.get_receipts_for_user(user_id)
 
     badge = len(invites)
 

synapse/rest/client/notifications.py

@@ -15,7 +15,6 @@
 import logging
 from typing import TYPE_CHECKING, Tuple
 
-from synapse.api.constants import ReceiptTypes
 from synapse.events.utils import (
     SerializeEventConfig,
     format_event_for_client_v2_without_room_id,
@@ -58,7 +57,7 @@ async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
         )
 
         receipts_by_room = await self.store.get_receipts_for_user_with_orderings(
-            user_id, ReceiptTypes.READ
+            user_id
         )
 
         notif_event_ids = [pa.event_id for pa in push_actions]

synapse/rest/client/read_marker.py

@@ -15,8 +15,7 @@
 import logging
 from typing import TYPE_CHECKING, Tuple
 
-from synapse.api.constants import ReadReceiptEventFields, ReceiptTypes
-from synapse.api.errors import Codes, SynapseError
+from synapse.api.constants import ReceiptTypes
 from synapse.http.server import HttpServer
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
 from synapse.http.site import SynapseRequest
@@ -48,27 +47,26 @@ async def on_POST(
         await self.presence_handler.bump_presence_active_time(requester.user)
 
         body = parse_json_object_from_request(request)
-        read_event_id = body.get(ReceiptTypes.READ, None)
-        hidden = body.get(ReadReceiptEventFields.MSC2285_HIDDEN, False)
-
-        if not isinstance(hidden, bool):
-            raise SynapseError(
-                400,
-                "Param %s must be a boolean, if given"
-                % ReadReceiptEventFields.MSC2285_HIDDEN,
-                Codes.BAD_JSON,
-            )
 
+        read_event_id = body.get(ReceiptTypes.READ, None)
         if read_event_id:
             await self.receipts_handler.received_client_receipt(
                 room_id,
                 ReceiptTypes.READ,
                 user_id=requester.user.to_string(),
                 event_id=read_event_id,
-                hidden=hidden,
             )
 
-        read_marker_event_id = body.get("m.fully_read", None)
+        read_private_event_id = body.get(ReceiptTypes.READ_PRIVATE, None)
+        if read_private_event_id:
+            await self.receipts_handler.received_client_receipt(
+                room_id,
+                ReceiptTypes.READ_PRIVATE,
+                user_id=requester.user.to_string(),
+                event_id=read_private_event_id,
+            )
+
+        read_marker_event_id = body.get(ReceiptTypes.FULLY_READ, None)
         if read_marker_event_id:
             await self.read_marker_handler.received_client_read_marker(
                 room_id,

synapse/rest/client/receipts.py

@@ -16,8 +16,8 @@
 import re
 from typing import TYPE_CHECKING, Tuple
 
-from synapse.api.constants import ReadReceiptEventFields, ReceiptTypes
-from synapse.api.errors import Codes, SynapseError
+from synapse.api.constants import ReceiptTypes
+from synapse.api.errors import SynapseError
 from synapse.http import get_request_user_agent
 from synapse.http.server import HttpServer
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
@@ -46,42 +46,47 @@ def __init__(self, hs: "HomeServer"):
         self.hs = hs
         self.auth = hs.get_auth()
         self.receipts_handler = hs.get_receipts_handler()
+        self.read_marker_handler = hs.get_read_marker_handler()
         self.presence_handler = hs.get_presence_handler()
 
     async def on_POST(
         self, request: SynapseRequest, room_id: str, receipt_type: str, event_id: str
     ) -> Tuple[int, JsonDict]:
         requester = await self.auth.get_user_by_req(request)
 
-        if receipt_type != ReceiptTypes.READ:
-            raise SynapseError(400, "Receipt type must be 'm.read'")
+        if receipt_type not in [
+            ReceiptTypes.READ,
+            ReceiptTypes.READ_PRIVATE,
+            ReceiptTypes.FULLY_READ,
+        ]:
+            raise SynapseError(
+                400,
+                "Receipt type must be 'm.read', 'org.matrix.msc2285.read.private' or 'm.fully_read'",
+            )
 
         # Do not allow older SchildiChat and Element Android clients (prior to Element/1.[012].x) to send an empty body.
         user_agent = get_request_user_agent(request)
         allow_empty_body = False
         if "Android" in user_agent:
             if pattern.match(user_agent) or "Riot" in user_agent:
                 allow_empty_body = True
-        body = parse_json_object_from_request(request, allow_empty_body)
-        hidden = body.get(ReadReceiptEventFields.MSC2285_HIDDEN, False)
-
-        if not isinstance(hidden, bool):
-            raise SynapseError(
-                400,
-                "Param %s must be a boolean, if given"
-                % ReadReceiptEventFields.MSC2285_HIDDEN,
-                Codes.BAD_JSON,
-            )
+        parse_json_object_from_request(request, allow_empty_body)
 
         await self.presence_handler.bump_presence_active_time(requester.user)
 
-        await self.receipts_handler.received_client_receipt(
-            room_id,
-            receipt_type,
-            user_id=requester.user.to_string(),
-            event_id=event_id,
-            hidden=hidden,
-        )
+        if receipt_type == ReceiptTypes.FULLY_READ:
+            await self.read_marker_handler.received_client_read_marker(
+                room_id,
+                user_id=requester.user.to_string(),
+                event_id=event_id,
+            )
+        else:
+            await self.receipts_handler.received_client_receipt(
+                room_id,
+                receipt_type,
+                user_id=requester.user.to_string(),
+                event_id=event_id,
+            )
 
         return 200, {}
 

synapse/storage/databases/main/receipts.py

@@ -118,38 +118,33 @@ async def get_receipts_for_room(
             desc="get_receipts_for_room",
         )
 
-    @cached(num_args=3)
+    @cached(num_args=2)
     async def get_last_receipt_event_id_for_user(
-        self, user_id: str, room_id: str, receipt_type: str
+        self, user_id: str, room_id: str
     ) -> Optional[str]:
         return await self.db_pool.simple_select_one_onecol(
             table="receipts_linearized",
             keyvalues={
                 "room_id": room_id,
-                "receipt_type": receipt_type,
                 "user_id": user_id,
             },
             retcol="event_id",
             desc="get_own_receipt_for_user",
             allow_none=True,
         )
 
-    @cached(num_args=2)
-    async def get_receipts_for_user(
-        self, user_id: str, receipt_type: str
-    ) -> Dict[str, str]:
+    @cached(num_args=1)
+    async def get_receipts_for_user(self, user_id: str) -> Dict[str, str]:
         rows = await self.db_pool.simple_select_list(
             table="receipts_linearized",
-            keyvalues={"user_id": user_id, "receipt_type": receipt_type},
+            keyvalues={"user_id": user_id},
             retcols=("room_id", "event_id"),
             desc="get_receipts_for_user",
         )
 
         return {row["room_id"]: row["event_id"] for row in rows}
 
-    async def get_receipts_for_user_with_orderings(
-        self, user_id: str, receipt_type: str
-    ) -> JsonDict:
+    async def get_receipts_for_user_with_orderings(self, user_id: str) -> JsonDict:
         def f(txn: LoggingTransaction) -> List[Tuple[str, str, int, int]]:
             sql = (
                 "SELECT rl.room_id, rl.event_id,"
@@ -490,9 +485,7 @@ def invalidate_caches_for_receipt(
     ) -> None:
         self.get_receipts_for_user.invalidate((user_id, receipt_type))
         self._get_linearized_receipts_for_room.invalidate((room_id,))
-        self.get_last_receipt_event_id_for_user.invalidate(
-            (user_id, room_id, receipt_type)
-        )
+        self.get_last_receipt_event_id_for_user.invalidate((user_id, room_id))
         self._invalidate_get_users_with_receipts_in_room(room_id, receipt_type, user_id)
         self.get_receipts_for_room.invalidate((room_id, receipt_type))
 
@@ -541,14 +534,20 @@ def insert_linearized_receipt_txn(
         # have to compare orderings of existing receipts
         if stream_ordering is not None:
             sql = (
-                "SELECT stream_ordering, event_id FROM events"
-                " INNER JOIN receipts_linearized as r USING (event_id, room_id)"
-                " WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ?"
+                "SELECT e.stream_ordering, e.event_id, r.receipt_type FROM events AS e"
+                " INNER JOIN receipts_linearized AS r USING (event_id, room_id)"
+                " WHERE r.room_id = ? AND r.user_id = ?"
             )
-            txn.execute(sql, (room_id, receipt_type, user_id))
-
-            for so, eid in txn:
-                if int(so) >= stream_ordering:
+            txn.execute(sql, (room_id, user_id))
+
+            for so, eid, rt in txn:
+                if int(so) >= stream_ordering and (
+                    receipt_type == rt
+                    or (
+                        rt == ReceiptTypes.READ
+                        and receipt_type == ReceiptTypes.READ_PRIVATE
+                    )
+                ):
                     logger.debug(
                         "Ignoring new receipt for %s in favour of existing "
                         "one for later event %s",
@@ -583,7 +582,10 @@ def insert_linearized_receipt_txn(
             lock=False,
         )
 
-        if receipt_type == ReceiptTypes.READ and stream_ordering is not None:
+        if (
+            receipt_type in [ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE]
+            and stream_ordering is not None
+        ):
             self._remove_old_push_actions_before_txn(
                 txn, room_id=room_id, user_id=user_id, stream_ordering=stream_ordering
             )