Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Implement changes to MSC2285
Browse files Browse the repository at this point in the history
Signed-off-by: Šimon Brandner <simon.bra.ag@gmail.com>
  • Loading branch information
SimonBrandner committed Mar 5, 2022
1 parent bc2a3db commit 926bd84
Show file tree
Hide file tree
Showing 11 changed files with 134 additions and 142 deletions.
6 changes: 2 additions & 4 deletions synapse/api/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
51 changes: 22 additions & 29 deletions synapse/handlers/receipts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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)

Expand All @@ -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:
Expand Down
3 changes: 1 addition & 2 deletions synapse/handlers/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
3 changes: 1 addition & 2 deletions synapse/push/push_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)

Expand Down
3 changes: 1 addition & 2 deletions synapse/rest/client/notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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]
Expand Down
26 changes: 12 additions & 14 deletions synapse/rest/client/read_marker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
47 changes: 26 additions & 21 deletions synapse/rest/client/receipts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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, {}

Expand Down
46 changes: 24 additions & 22 deletions synapse/storage/databases/main/receipts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,"
Expand Down Expand Up @@ -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))

Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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
)
Expand Down
Loading

0 comments on commit 926bd84

Please sign in to comment.