diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index d590578c5d..11de5f00f1 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -39,6 +39,7 @@ from prometheus_client import Counter from synapse.api.constants import ( AccountDataTypes, Direction, + EduTypes, EventContentFields, EventTypes, Membership, @@ -62,6 +63,7 @@ from synapse.logging.opentracing import ( ) from synapse.metrics import SERVER_NAME_LABEL from synapse.storage.databases.main.event_push_actions import RoomNotifCounts +from synapse.storage.databases.main.receipts import ReceiptInRoom from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary from synapse.storage.databases.main.stream import PaginateFunction from synapse.storage.invite_rule import InviteRule @@ -778,6 +780,23 @@ class SyncHandler: room_id = event["room_id"] ephemeral_by_room.setdefault(room_id, []).append(event) + announcement_receipts = ( + await self.store.get_linearized_receipts_for_user_in_rooms( + user_id=sync_config.user.to_string(), + room_ids=sync_result_builder.announcement_room_ids, + to_key=receipt_key, + from_key=since_token.receipt_key if since_token else None, + ) + ) + for room_id, room_receipts in announcement_receipts.items(): + ephemeral_by_room.setdefault(room_id, []).append( + { + "room_id": room_id, + "type": EduTypes.RECEIPT, + "content": ReceiptInRoom.merge_to_content(room_receipts), + } + ) + return now_token, ephemeral_by_room async def sticky_events_by_room( @@ -2978,7 +2997,7 @@ class SyncHandler: ignored_users = await self.store.ignored_users(user_id) if since_token: room_changes = await self._get_room_changes_for_incremental_sync( - sync_result_builder, ignored_users + sync_result_builder, ignored_users, frozenset(ephemeral_by_room) ) tags_by_room = await self.store.get_updated_tags( user_id, since_token.account_data_key @@ -3058,6 +3077,7 @@ class SyncHandler: self, sync_result_builder: "SyncResultBuilder", ignored_users: frozenset[str], + ephemeral_room_ids: AbstractSet[str], ) -> _RoomChanges: """Determine the changes in rooms to report to the user. @@ -3316,6 +3336,7 @@ class SyncHandler: sync_result_builder, sync_result_builder.announcement_room_ids, left_room_ids=left_announcement_room_ids, + ephemeral_room_ids=ephemeral_room_ids, ) ) @@ -3332,6 +3353,7 @@ class SyncHandler: sync_result_builder: "SyncResultBuilder", room_ids: AbstractSet[str], left_room_ids: AbstractSet[str] = frozenset(), + ephemeral_room_ids: AbstractSet[str] = frozenset(), ) -> list["RoomSyncResultBuilder"]: """Returns entries for the given announcement rooms that have something to show. @@ -3340,7 +3362,8 @@ class SyncHandler: event, so that creating the room and joining its senders does not show it to every user. On an incremental sync, a room is included if it has had an announcement and there is any new event in it since the - `since_token`, such as a sender leaving or a name change. With + `since_token`, such as a sender leaving or a name change, or any new + ephemeral event for the user, such as their own read receipt. With `full_state` set, a room is also included without any new event. On an initial sync, a room is included if it has ever had an announcement, even when later state changes push that announcement out of the @@ -3369,6 +3392,8 @@ class SyncHandler: room_ids: The announcement rooms to consider. left_room_ids: Rooms among `room_ids` that the user left or was removed from since the `since_token`. + ephemeral_room_ids: Rooms with new ephemeral events for the user + since the `since_token`. """ since_token = sync_result_builder.since_token now_token = sync_result_builder.now_token @@ -3396,7 +3421,7 @@ class SyncHandler: room_id for room_id, (events, _, _) in room_to_events.items() if events and room_id not in announced_room_ids - } + } | ((ephemeral_room_ids & room_ids) - announced_room_ids) announced_room_ids.update( await self.store.get_server_announcement_room_ids_with_announcement( room_ids_to_check, now_token.room_key.stream diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index 3720d25c1d..efdbfe2137 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -674,9 +674,14 @@ class ReceiptsWorkerStore(SQLBaseStore): return results, reached_token async def get_linearized_receipts_for_user_in_rooms( - self, user_id: str, room_ids: StrCollection, to_key: MultiWriterStreamToken + self, + user_id: str, + room_ids: StrCollection, + to_key: MultiWriterStreamToken, + from_key: MultiWriterStreamToken | None = None, ) -> Mapping[str, Sequence[ReceiptInRoom]]: - """Fetch all receipts for the user in the given room. + """Fetch all receipts for the user in the given room, after `from_key` + if given. Returns: A dict from room ID to receipts in the room. @@ -699,13 +704,17 @@ class ReceiptsWorkerStore(SQLBaseStore): args.append(user_id) args.append(to_key.get_max_stream_pos()) + if from_key: + sql += " AND stream_id > ?" + args.append(from_key.stream) + txn.execute(sql, args) return [ (room_id, receipt_type, user_id, event_id, thread_id, data) for instance_name, stream_id, room_id, receipt_type, user_id, event_id, thread_id, data in txn if MultiWriterStreamToken.is_stream_position_in_range( - low=None, + low=from_key, high=to_key, instance_name=instance_name, pos=stream_id,