diff --git a/changelog.d/20412.misc b/changelog.d/20412.misc new file mode 100644 index 0000000000..00a9b4f284 --- /dev/null +++ b/changelog.d/20412.misc @@ -0,0 +1 @@ +Deliver server announcement rooms to users who are not members of them in legacy `/sync`. diff --git a/rust/src/events/mod.rs b/rust/src/events/mod.rs index 3fc2d5ad69..6e0dc95135 100644 --- a/rust/src/events/mod.rs +++ b/rust/src/events/mod.rs @@ -119,6 +119,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> child_module.add_function(wrap_pyfunction!(redact_event_py, m)?)?; child_module.add_function(wrap_pyfunction!(redact_event_dict, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::serialize_events, m)?)?; + child_module.add_function(wrap_pyfunction!(serialize::serialize_synthetic_event, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_raw, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_for_client_v1, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_for_client_v2, m)?)?; diff --git a/rust/src/events/serialize.rs b/rust/src/events/serialize.rs index 20d4d9bf52..0d1a0fda91 100644 --- a/rust/src/events/serialize.rs +++ b/rust/src/events/serialize.rs @@ -36,7 +36,7 @@ use pyo3::{ types::{PyAnyMethods, PyDict, PyDictMethods}, Bound, PyAny, PyResult, Python, }; -use pythonize::pythonize; +use pythonize::{depythonize, pythonize}; use serde_json::{Map, Number, Value}; use crate::{ @@ -636,6 +636,33 @@ fn format_for_client_v2(d: &mut Map) { } } +/// Serialize an event that only exists as JSON, such as a synthetic event that +/// is never persisted. +/// +/// Only the event format and the field allowlist from `config` apply: there is +/// no stored event to take the age, transaction ID, redaction or bundled +/// aggregations from. +#[pyfunction] +pub fn serialize_synthetic_event<'py>( + py: Python<'py>, + event: Bound<'py, PyAny>, + config: &SerializeEventConfig, +) -> PyResult> { + let mut d: Map = depythonize(&event)?; + + if config.as_client_event { + apply_event_format(config.event_format, &mut d); + } + + if let Some(fields) = &config.event_field_allowlist { + if !fields.is_empty() { + d = only_fields(&d, fields)?; + } + } + + Ok(pythonize(py, &Value::Object(d))?) +} + // Standalone versions of the client format transforms, re-exported from // `synapse.events.utils` purely as a backwards compatibility hack: they have // never been part of the module API and modules shouldn't be pulling them in, diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py index cbae9133c6..0d7b63ef07 100644 --- a/synapse/api/filtering.py +++ b/synapse/api/filtering.py @@ -292,6 +292,9 @@ class FilterCollection: await self._room_filter.filter(events) ) + def filter_rooms(self, room_ids: Iterable[str]) -> set[str]: + return self._room_filter.filter_rooms(room_ids) + def blocks_all_rooms(self) -> bool: return self._room_filter.filters_all_rooms() diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 6014d328ba..bddc3334e3 100644 --- a/synapse/app/generic_worker.py +++ b/synapse/app/generic_worker.py @@ -98,6 +98,9 @@ from synapse.storage.databases.main.relations import RelationsWorkerStore from synapse.storage.databases.main.room import RoomWorkerStore from synapse.storage.databases.main.roommember import RoomMemberWorkerStore from synapse.storage.databases.main.search import SearchStore +from synapse.storage.databases.main.server_announcements import ( + ServerAnnouncementsStore, +) from synapse.storage.databases.main.session import SessionStore from synapse.storage.databases.main.signatures import SignatureWorkerStore from synapse.storage.databases.main.sliding_sync import SlidingSyncStore @@ -171,6 +174,7 @@ class GenericWorkerStore( ExperimentalFeaturesStore, SlidingSyncStore, DelayedEventsStore, + ServerAnnouncementsStore, ): # Properties that multiple storage classes define. Tell mypy what the # expected type is. diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 365b7b1559..d6a7d8cb6c 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -51,6 +51,7 @@ from synapse.synapse_rust.events import ( format_event_raw, redact_event, serialize_events, + serialize_synthetic_event, ) from synapse.synapse_rust.types import Requester from synapse.types import JsonDict @@ -423,6 +424,22 @@ class EventClientSerializer: ) return [event for event in stitched if event is not None] + def serialize_synthetic_event( + self, event: JsonDict, config: SerializeEventConfig + ) -> JsonDict: + """Serializes an event that only exists as JSON, such as a synthetic + event that is never persisted, applying the event format and field + allowlist from `config`. + + Args: + event: The event's JSON, in federation format. + config: Event serialization config + + Returns: + The serialized event + """ + return serialize_synthetic_event(event, config) + def register_add_extra_fields_to_unsigned_client_event_callback( self, callback: ADD_EXTRA_FIELDS_TO_UNSIGNED_CLIENT_EVENT_CALLBACK ) -> None: diff --git a/synapse/handlers/server_announcements.py b/synapse/handlers/server_announcements.py index a8a4b7f78c..6adfe49ee3 100644 --- a/synapse/handlers/server_announcements.py +++ b/synapse/handlers/server_announcements.py @@ -20,12 +20,57 @@ from synapse.config import ConfigError from synapse.config.server_announcements import ServerAnnouncementRoom from synapse.types import JsonDict, UserID, create_requester from synapse.types.state import StateFilter +from synapse.util.hash import sha256_and_url_safe_base64 if TYPE_CHECKING: from synapse.server import HomeServer logger = logging.getLogger(__name__) +SYNTHETIC_JOIN_EVENT_ID_PREFIX = "$fake_announcement_join" + + +def make_synthetic_join_event( + room_id: str, user_id: str, origin_server_ts: int +) -> JsonDict: + """Build the join event shown to a user who is not actually in an announcement + room, so that clients list the room as joined. + + The event is never persisted. Its event ID is derived from the room and user, + so the same user always receives the same event and clients drop repeats. The + prefix makes it obvious that the event ID is not a real one. + + It is injected into the `/sync` responses of every user of the homeserver who + is not one of the room's senders. + + Args: + room_id: The announcement room. + user_id: The user receiving the event. + origin_server_ts: The timestamp to give the event. Pass the same value + every time, e.g. that of the room's create event. + + Returns: + The event's JSON in federation format. + """ + return { + "room_id": room_id, + "type": EventTypes.Member, + "event_id": "-".join( + [ + SYNTHETIC_JOIN_EVENT_ID_PREFIX, + # 8 first chars of room_id and user_id localpart for debugging + room_id[1:9], + UserID.from_string(user_id).localpart[:8], + # hash of room_id + user_id for uniqueness & staying under 255 bytes length + sha256_and_url_safe_base64(f"{room_id}\0{user_id}"), + ] + ), + "sender": user_id, + "state_key": user_id, + "origin_server_ts": origin_server_ts, + "content": {"membership": Membership.JOIN}, + } + class ServerAnnouncementsHandler: """Reconcile configured rooms on the main process, without deleting removed entries.""" diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index c35e8f01fb..b4b260d156 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -276,6 +276,7 @@ class JoinedSyncResult: unread_thread_notifications: JsonDict summary: JsonDict | None unread_count: int + is_announcement: bool = False def __bool__(self) -> bool: """Make the result appear empty if there are no updates. This is used @@ -2364,12 +2365,27 @@ class SyncHandler: if room_id not in mutable_rooms_to_exclude ) + last_membership_by_room_id = { + event.room_id: event.membership for event in membership_change_events + } + synthetic_join_announcement_room_ids = frozenset( + room_id + for room_id in sync_config.filter_collection.filter_rooms( + await self.store.get_server_announcement_room_ids() + ) + if room_id not in mutable_joined_room_ids + and room_id not in mutable_rooms_to_exclude + and last_membership_by_room_id.get(room_id) + in (None, Membership.LEAVE, Membership.BAN) + ) + sync_result_builder = SyncResultBuilder( sync_config, full_state, since_token=since_token, now_token=now_token, joined_room_ids=joined_room_ids, + synthetic_join_announcement_room_ids=synthetic_join_announcement_room_ids, excluded_room_ids=frozenset(mutable_rooms_to_exclude), forced_newly_joined_room_ids=frozenset(forced_newly_joined_room_ids), membership_change_events=membership_change_events, @@ -2966,7 +2982,9 @@ 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, + account_data_room_ids=account_data_by_room.keys(), ) tags_by_room = await self.store.get_updated_tags( user_id, since_token.account_data_key @@ -3034,7 +3052,10 @@ class SyncHandler: return True stream_id = since_token.room_key.stream - for room_id in sync_result_builder.joined_room_ids: + for room_id in ( + sync_result_builder.joined_room_ids + | sync_result_builder.synthetic_join_announcement_room_ids + ): if self.store.has_room_changed_since(room_id, stream_id): return True return False @@ -3043,6 +3064,7 @@ class SyncHandler: self, sync_result_builder: "SyncResultBuilder", ignored_users: frozenset[str], + account_data_room_ids: AbstractSet[str], ) -> _RoomChanges: """Determine the changes in rooms to report to the user. @@ -3081,6 +3103,7 @@ class SyncHandler: sync_result_builder.forced_newly_joined_room_ids ) newly_left_rooms: list[str] = [] + left_announcement_room_ids: set[str] = set() room_entries: list[RoomSyncResultBuilder] = [] invited: list[InvitedSyncResult] = [] knocked: list[KnockedSyncResult] = [] @@ -3184,6 +3207,10 @@ class SyncHandler: if knock_room_sync: knocked.append(knock_room_sync) + if room_id in sync_result_builder.synthetic_join_announcement_room_ids: + left_announcement_room_ids.add(room_id) + continue + # Always include leave/ban events. Just take the last one. # TODO: How do we handle ban -> leave in same batch? leave_events = [ @@ -3291,6 +3318,15 @@ class SyncHandler: room_entries.append(entry) + room_entries.extend( + await self._get_announcement_room_entries( + sync_result_builder, + sync_result_builder.synthetic_join_announcement_room_ids, + left_room_ids=left_announcement_room_ids, + account_data_room_ids=account_data_room_ids, + ) + ) + return _RoomChanges( room_entries, invited, @@ -3299,6 +3335,144 @@ class SyncHandler: newly_left_rooms, ) + async def _get_announcement_room_entries( + self, + sync_result_builder: "SyncResultBuilder", + room_ids: AbstractSet[str], + left_room_ids: AbstractSet[str] = frozenset(), + account_data_room_ids: AbstractSet[str] = frozenset(), + ) -> list["RoomSyncResultBuilder"]: + """Returns entries for the given announcement rooms that have something + to show. + + A room only shows up once it has had an announcement, i.e. a non-state + 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, or a change + to the user's room account data for it. 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 + timeline. + + Rooms in `left_room_ids` are included even without an announcement, so + that the client replaces the user's real membership with the synthetic + join and gets the room's current state. + + An entry carries the full state only if any of: + - there is no `since_token` + - the first announcement is after the `since_token` + - the user left or was removed from the room + Otherwise it carries the changes since the `since_token`, as for a + joined room: any sync covering a room's first announcement delivers it + with full state, so the client already has the room. + + Sending an announcement does not wake up waiting syncs, so that an + announcement does not wake every user's sync at once. A waiting sync + picks announcements up when it next runs: at its timeout, or when + something else wakes it. With no other activity, an announcement is up + to one sync timeout late. It can be up to about two timeouts late when + something woke the sync without producing anything for it: the sync + then returns that empty result at its timeout without running again, + and the announcement waits for the next sync. + + Args: + sync_result_builder + 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` of `sync_result_builder`. + account_data_room_ids: Rooms whose room account data for the user + changed since the `since_token` of `sync_result_builder`. + They may include rooms that are not in `room_ids`. + """ + since_token = sync_result_builder.since_token + now_token = sync_result_builder.now_token + timeline_limit = ( + sync_result_builder.sync_config.filter_collection.timeline_limit() + ) + + room_to_events = await self.store.get_room_events_stream_for_rooms( + room_ids=room_ids, + from_key=now_token.room_key, + to_key=since_token.room_key if since_token else None, + limit=timeline_limit + 1, + direction=Direction.BACKWARDS, + ) + + # Rooms with an announcement since the `since_token`. + announced_room_ids = { + room_id + for room_id, (events, _, _) in room_to_events.items() + if not all(event.is_state() for event in events) + } + + # Rooms the user left are included even without an announcement, so + # their real membership is replaced by the synthetic join. + room_ids_to_include = announced_room_ids | left_room_ids + + # On an incremental sync, `room_ids_to_include` should also include + # rooms that have new state events but no new announcement in the + # window, or a room account data change, if they had an announcement + # before the window. + # + # On an initial or `full_state` sync, a room is included whenever it + # has ever had an announcement, even if it isn't among the recent + # events fetched above. + if not since_token or sync_result_builder.full_state: + room_ids_to_check = room_ids - room_ids_to_include + else: + room_ids_to_check = { + room_id + for room_id, (events, _, _) in room_to_events.items() + if events and room_id not in left_room_ids + } + # Also consider rooms whose account data has changed. + room_ids_to_check |= account_data_room_ids & ( + room_ids - room_ids_to_include + ) + first_announcement_positions = ( + await self.store.get_first_non_state_event_positions( + room_ids_to_check, now_token.room_key.get_max_stream_pos() + ) + ) + room_ids_to_include.update(first_announcement_positions) + + room_entries = [] + for room_id in room_ids_to_include: + if room_id in room_to_events: + events, start_key, _ = room_to_events[room_id] + events.reverse() + upto_token = now_token.copy_and_replace(StreamKeyType.ROOM, start_key) + else: + # Without a `since_token`, `room_to_events` has an entry for every + # room in `room_ids`. + assert since_token is not None + events = [] + upto_token = since_token + + first_announcement_position = first_announcement_positions.get(room_id) + full_state = ( + since_token is None + or first_announcement_position is None + or first_announcement_position.persisted_after(since_token.room_key) + ) + + room_entries.append( + RoomSyncResultBuilder( + room_id=room_id, + rtype="joined", + events=events, + newly_joined=False, + full_state=full_state, + since_token=since_token, + upto_token=upto_token, + end_token=now_token, + ) + ) + + return room_entries + async def _get_room_changes_for_initial_sync( self, sync_result_builder: "SyncResultBuilder", @@ -3360,6 +3534,14 @@ class SyncHandler: knock = await self.store.get_event(event.event_id) knocked.append(KnockedSyncResult(room_id=event.room_id, knock=knock)) elif event.membership in (Membership.LEAVE, Membership.BAN): + # Former members of an announcement room get the synthetic join + # instead of this leave or ban. + if ( + event.room_id + in sync_result_builder.synthetic_join_announcement_room_ids + ): + continue + # Always send down rooms we were banned from or kicked from. if not sync_config.filter_collection.include_leave: if event.membership == Membership.LEAVE: @@ -3382,6 +3564,19 @@ class SyncHandler: ) ) + room_ids_in_response = ( + {entry.room_id for entry in room_entries} + | {result.room_id for result in invited} + | {result.room_id for result in knocked} + ) + room_entries.extend( + await self._get_announcement_room_entries( + sync_result_builder, + sync_result_builder.synthetic_join_announcement_room_ids + - room_ids_in_response, + ) + ) + return _RoomChanges(room_entries, invited, knocked, [], []) async def _generate_room_entry( @@ -3615,10 +3810,19 @@ class SyncHandler: summary=summary, unread_count=0, sticky=sticky_events, + is_announcement=( + room_id + in sync_result_builder.synthetic_join_announcement_room_ids + ), ) if room_sync or always_include: - notifs = await self.unread_notifs_for_room_id(room_id, sync_config) + if room_sync.is_announcement: + notifs = RoomNotifCounts.empty() + else: + notifs = await self.unread_notifs_for_room_id( + room_id, sync_config + ) # Notifications for the main timeline. notify_count = notifs.main_timeline.notify_count @@ -3805,6 +4009,11 @@ class SyncResultBuilder: since_token: The token supplied by user, or None. now_token: The token to sync up to. joined_room_ids: List of rooms the user is joined to + synthetic_join_announcement_room_ids: Server announcement rooms the user + is not a member of, whose new announcements should still be sent with + a synthetic join. Rooms the user really joined are not included. + They are kept apart from `joined_room_ids` because most code that + reads `joined_room_ids` relies on the user being a member. excluded_room_ids: Set of room ids we should omit from the /sync response. forced_newly_joined_room_ids: Rooms that should be presented in the /sync response as if they were @@ -3829,6 +4038,7 @@ class SyncResultBuilder: since_token: StreamToken | None now_token: StreamToken joined_room_ids: frozenset[str] + synthetic_join_announcement_room_ids: frozenset[str] excluded_room_ids: frozenset[str] forced_newly_joined_room_ids: frozenset[str] membership_change_events: list[EventBase] @@ -3855,6 +4065,9 @@ class SyncResultBuilder: newly_left_users = set() if self.since_token: for joined_sync in self.joined: + if joined_sync.room_id in self.synthetic_join_announcement_room_ids: + continue + it = itertools.chain( joined_sync.state.values(), (e.event for e in joined_sync.timeline.events), diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index 6e6eb6d633..da56f46632 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -24,7 +24,13 @@ from typing import TYPE_CHECKING, Any, Literal, Mapping import attr -from synapse.api.constants import AccountDataTypes, EduTypes, Membership, PresenceState +from synapse.api.constants import ( + AccountDataTypes, + EduTypes, + EventTypes, + Membership, + PresenceState, +) from synapse.api.errors import Codes, StoreError, SynapseError from synapse.api.filtering import FilterCollection from synapse.api.presence import UserPresenceState @@ -37,6 +43,7 @@ from synapse.events.utils import ( strip_event, ) from synapse.handlers.presence import format_user_presence_state +from synapse.handlers.server_announcements import make_synthetic_join_event from synapse.handlers.sliding_sync import SlidingSyncConfig, SlidingSyncResult from synapse.handlers.sync import ( ArchivedSyncResult, @@ -598,6 +605,26 @@ class SyncRestServlet(RestServlet): state_dict = room.state timeline_events = room.timeline.events + user_id = sync_config.user.to_string() + is_announcement = isinstance(room, JoinedSyncResult) and room.is_announcement + # A former member gets the room with `is_announcement` set, as they are + # no longer a member, but their real leave or ban can still be in the + # state or the timeline. Remove it, as the synthetic join replaces it. + if is_announcement: + state_dict = { + key: event + for key, event in state_dict.items() + if key != (EventTypes.Member, user_id) + } + timeline_events = [ + filtered_event + for filtered_event in timeline_events + if not ( + filtered_event.event.type == EventTypes.Member + and filtered_event.event.state_key == user_id + ) + ] + state_events = state_dict.values() for event in state_events: @@ -626,6 +653,16 @@ class SyncRestServlet(RestServlet): time_now, config=serialize_options, ) + if is_announcement: + create_event = await self.store.get_create_event_for_room(room.room_id) + serialized_state.append( + self._event_serializer.serialize_synthetic_event( + make_synthetic_join_event( + room.room_id, user_id, create_event.origin_server_ts + ), + serialize_options, + ) + ) serialized_timeline = await self._event_serializer.serialize_events( timeline_events, time_now, diff --git a/synapse/storage/databases/main/cache.py b/synapse/storage/databases/main/cache.py index 45fe605d8b..5ca908ac5b 100644 --- a/synapse/storage/databases/main/cache.py +++ b/synapse/storage/databases/main/cache.py @@ -612,6 +612,7 @@ class CacheInvalidationWorkerStore(SQLBaseStore): self._attempt_to_invalidate_cache("get_room_version_id", (room_id,)) self._attempt_to_invalidate_cache("get_room_type", (room_id,)) self._attempt_to_invalidate_cache("get_room_encryption", (room_id,)) + self._attempt_to_invalidate_cache("get_server_announcement_room_ids", None) self._attempt_to_invalidate_cache("_get_max_event_pos", (room_id,)) diff --git a/synapse/storage/databases/main/server_announcements.py b/synapse/storage/databases/main/server_announcements.py index bb466ca97a..444fd9a12f 100644 --- a/synapse/storage/databases/main/server_announcements.py +++ b/synapse/storage/databases/main/server_announcements.py @@ -12,17 +12,18 @@ # . # -from synapse.storage._base import SQLBaseStore +from synapse.storage.database import LoggingTransaction +from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore +from synapse.util.caches.descriptors import cached -class ServerAnnouncementsStore(SQLBaseStore): +class ServerAnnouncementsStore(CacheInvalidationWorkerStore): """Stores details about an announcement room. The unique announcement `id` identifies the same room even when its configured details change. It is separate from the actual room ID. - Lookups are cheap and deliberately uncached, avoiding cross-worker cache - invalidation for mapping and avatar changes. + Only the set of configured room IDs is cached, as it is read on every sync. """ async def get_server_announcement_room(self, config_id: str) -> str | None: @@ -45,14 +46,41 @@ class ServerAnnouncementsStore(SQLBaseStore): desc="get_server_announcement_id", ) + @cached(num_args=0) + async def get_server_announcement_room_ids(self) -> frozenset[str]: + """Return the room IDs of the currently configured announcement rooms. + + Rooms whose config ID is no longer in the config are left out, as + are configured rooms that have not been created yet. + """ + rows = await self.db_pool.simple_select_many_batch( + table="server_announcement_rooms", + column="config_id", + iterable=[ + room.config_id for room in self.hs.config.serverannouncements.rooms + ], + retcols=("room_id",), + desc="get_server_announcement_room_ids", + ) + return frozenset(room_id for (room_id,) in rows) + async def store_server_announcement_room( self, config_id: str, room_id: str ) -> None: """Store a new mapping.""" - await self.db_pool.simple_insert( - table="server_announcement_rooms", - values={"config_id": config_id, "room_id": room_id}, - desc="store_server_announcement_room", + + def _store_server_announcement_room_txn(txn: LoggingTransaction) -> None: + self.db_pool.simple_insert_txn( + txn, + table="server_announcement_rooms", + values={"config_id": config_id, "room_id": room_id}, + ) + self._invalidate_cache_and_stream( + txn, self.get_server_announcement_room_ids, () + ) + + await self.db_pool.runInteraction( + "store_server_announcement_room", _store_server_announcement_room_txn ) async def get_server_announcement_avatar( diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 59048c2fc6..d278985832 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1841,6 +1841,46 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore): return results + async def get_first_non_state_event_positions( + self, room_ids: Collection[str], max_stream_ordering: int + ) -> dict[str, PersistedEventPosition]: + """Return the position of the first non-state, non-outlier event at or + before `max_stream_ordering` in each of `room_ids`, leaving out rooms that + have none. + """ + + def _get_first_non_state_event_positions_txn( + txn: LoggingTransaction, + ) -> dict[str, PersistedEventPosition]: + positions = {} + for room_id in room_ids: + txn.execute( + """ + SELECT instance_name, stream_ordering FROM events + WHERE room_id = ? AND stream_ordering <= ? + AND state_key IS NULL + AND NOT outlier + ORDER BY stream_ordering + LIMIT 1 + """, + (room_id, max_stream_ordering), + ) + row = txn.fetchone() + if row is not None: + instance_name, stream_ordering = row + positions[room_id] = PersistedEventPosition( + instance_name or "master", stream_ordering + ) + return positions + + if not room_ids: + return {} + + return await self.db_pool.runInteraction( + "get_first_non_state_event_positions", + _get_first_non_state_event_positions_txn, + ) + async def get_current_room_stream_token_for_room_id( self, room_id: str ) -> RoomStreamToken: diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi index bac7af5a13..1690d71753 100644 --- a/synapse/synapse_rust/events.pyi +++ b/synapse/synapse_rust/events.pyi @@ -482,6 +482,24 @@ def serialize_events( The serialized events, in the same order as `events`. """ +def serialize_synthetic_event( + event: JsonDict, config: SerializeEventConfig +) -> JsonDict: + """Serialize an event that only exists as JSON, such as a synthetic event + that is never persisted. + + Only the event format and the field allowlist from `config` apply: there is + no stored event to take the age, transaction ID, redaction or bundled + aggregations from. + + Args: + event: The event's JSON, in federation format. + config: The serialization config. + + Returns: + The serialized event. + """ + # The standalone `format_event_*` transforms below are a backwards # compatibility hack: they have never been part of the module API and modules # shouldn't be pulling them in, but some in the wild import them (via diff --git a/tests/handlers/test_server_announcements.py b/tests/handlers/test_server_announcements.py index c3d423746e..d69fcecf1c 100644 --- a/tests/handlers/test_server_announcements.py +++ b/tests/handlers/test_server_announcements.py @@ -21,7 +21,10 @@ from synapse.api.constants import EventTypes from synapse.api.errors import SynapseError from synapse.config import ConfigError from synapse.config.server_announcements import ServerAnnouncementRoom -from synapse.handlers.server_announcements import ServerAnnouncementsHandler +from synapse.handlers.server_announcements import ( + ServerAnnouncementsHandler, + make_synthetic_join_event, +) from synapse.rest import admin from synapse.rest.client import login, register, room, sync from synapse.server import HomeServer @@ -350,3 +353,36 @@ class ServerAnnouncementsTestCase(unittest.HomeserverTestCase): self.get_success(self.store.get_server_announcement_room("updates")), room_id, ) + + +class SyntheticJoinEventTestCase(unittest.TestCase): + def test_event_id(self) -> None: + """The event ID is the prefix, the first 8 characters of the room ID and + of the user localpart, and a hash of both IDs. + """ + self.assertEqual( + make_synthetic_join_event( + "!Q8bOeGjhyXNtCVmX7Zvvh-W5KiVu2Q8vBK_2ljoZW7s", + "@alexander:example.com", + 0, + )["event_id"], + "$fake_announcement_join-Q8bOeGjh-alexande-SVJqkL5sx8x0TXrBVujJXBUw_Kc-c10PNz_sZV5FTnk", + ) + + def test_event_id_short_localpart(self) -> None: + """A localpart shorter than 8 characters is included in full.""" + self.assertEqual( + make_synthetic_join_event( + "!Q8bOeGjhyXNtCVmX7Zvvh-W5KiVu2Q8vBK_2ljoZW7s", "@bob:example.com", 0 + )["event_id"], + "$fake_announcement_join-Q8bOeGjh-bob-OkzlBBrE1j2T2pq5aF4ubO-Z1hmIo2cun6rrIXjw2QE", + ) + + def test_event_id_room_id_with_domain(self) -> None: + """For a room ID with a domain, only the start of its localpart is included.""" + self.assertEqual( + make_synthetic_join_event( + "!abcdefghijklmnopqr:example.com", "@alexander:example.com", 0 + )["event_id"], + "$fake_announcement_join-abcdefgh-alexande-QdoyrQ6WLSZZBDWVTw4cVhH61QX8Z90v8YIhfWkHTlg", + ) diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index 4f80758033..418c2810dd 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -21,6 +21,7 @@ import json import logging +import attr from parameterized import parameterized from twisted.internet.testing import MemoryReactor @@ -34,8 +35,19 @@ from synapse.api.constants import ( ReceiptTypes, RelationTypes, ) +from synapse.config.server_announcements import ServerAnnouncementRoom from synapse.events import EventBase -from synapse.rest.client import devices, knock, login, read_marker, receipts, room, sync +from synapse.handlers.server_announcements import SYNTHETIC_JOIN_EVENT_ID_PREFIX +from synapse.rest.client import ( + account_data, + devices, + knock, + login, + read_marker, + receipts, + room, + sync, +) from synapse.server import HomeServer from synapse.types import JsonDict, RoomStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock @@ -2187,3 +2199,493 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase): if e["type"] == EventTypes.Member ] self.assertIn(self.alice, members) + + +class ServerAnnouncementsSyncTestCase(unittest.HomeserverTestCase): + """Tests that server announcement rooms are delivered to users who are not + members of them, with a synthetic join for the syncing user. + """ + + servlets = [ + synapse.rest.admin.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + account_data.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.announcer = self.register_user("announcer", "pass") + self.announcer_tok = self.login("announcer", "pass") + self.former = self.register_user("former", "pass") + self.former_tok = self.login("former", "pass") + self.bob = self.register_user("bob", "pass") + self.bob_tok = self.login("bob", "pass") + + self.config = hs.config.serverannouncements + self.config.rooms = [ + ServerAnnouncementRoom( + config_id="news", + room_name="Server News", + room_topic="News from the server", + room_avatar_url=None, + room_avatar_path=None, + allowed_user_ids=[self.announcer, self.former], + ) + ] + self.handler = hs.get_server_announcements_handler() + self.store = hs.get_datastores().main + self.get_success(self.handler.reconcile_rooms()) + room_id = self.get_success(self.store.get_server_announcement_room("news")) + assert room_id is not None + self.room_id = room_id + + def _sync( + self, + tok: str, + since: str | None = None, + full_state: bool = False, + filter: JsonDict | None = None, + use_state_after: bool = False, + ) -> JsonDict: + params = [] + if since is not None: + params.append(f"since={since}") + if full_state: + params.append("full_state=true") + if use_state_after: + params.append("use_state_after=true") + if filter is not None: + params.append(f"filter={json.dumps(filter)}") + channel = self.make_request( + "GET", "/sync?" + "&".join(params), access_token=tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + return channel.json_body + + def _announce(self, body: str) -> str: + return self.helper.send(self.room_id, body, tok=self.announcer_tok)["event_id"] + + def _assert_not_delivered(self, response: JsonDict) -> None: + for section in ("join", "invite", "leave", "knock"): + self.assertNotIn( + self.room_id, response.get("rooms", {}).get(section, {}), section + ) + + def _assert_delivered(self, response: JsonDict, user_id: str) -> JsonDict: + """Assert that the announcement room is in the `join` section with a + synthetic join for `user_id` and no unread counts, and return it. + """ + room = response["rooms"]["join"][self.room_id] + state = room["state_after"] if "state_after" in room else room["state"] + + own_memberships = [ + event + for event in state["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Member and event["state_key"] == user_id + ] + self.assertEqual(len(own_memberships), 1, own_memberships) + membership = own_memberships[0] + self.assertTrue( + membership["event_id"].startswith(SYNTHETIC_JOIN_EVENT_ID_PREFIX), + membership, + ) + self.assertEqual(membership["content"], {"membership": Membership.JOIN}) + create_event = self.get_success( + self.store.get_create_event_for_room(self.room_id) + ) + self.assertEqual(membership["origin_server_ts"], create_event.origin_server_ts) + + self.assertEqual( + room["unread_notifications"], + {"highlight_count": 0, "notification_count": 0}, + ) + return room + + def _timeline_ids(self, room: JsonDict) -> list[str]: + return [event["event_id"] for event in room["timeline"]["events"]] + + def _timeline_types(self, room: JsonDict) -> list[str]: + return [event["type"] for event in room["timeline"]["events"]] + + def _state_types(self, room: JsonDict) -> set[str]: + return {event["type"] for event in room["state"]["events"]} + + def test_incremental_sync_delivers_new_announcement(self) -> None: + """A non-member syncing incrementally gets an announcement sent since + their token, with the room's full state. + """ + since = self._sync(self.bob_tok)["next_batch"] + event_id = self._announce("hello") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertTrue( + {EventTypes.Create, EventTypes.Name, EventTypes.Topic} + <= self._state_types(room) + ) + + def test_incremental_sync_without_new_announcement(self) -> None: + """An announcement room with nothing new since the token is left out of + an incremental sync. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + self._assert_not_delivered(self._sync(self.bob_tok, since)) + + def test_incremental_sync_after_first_announcement(self) -> None: + """A non-member who has synced past an announcement room's first + announcement gets a later announcement with only the synthetic join in + `state`, as nothing else changed since their token. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + event_id = self._announce("again") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertEqual( + [event["type"] for event in room["state"]["events"]], [EventTypes.Member] + ) + + def test_members_get_their_real_membership(self) -> None: + """Members of an announcement room sync it as a regular room.""" + self._announce("hello") + + room = self._sync(self.announcer_tok)["rooms"]["join"][self.room_id] + memberships = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Member + ] + self.assertIn(self.announcer, {event["state_key"] for event in memberships}) + for event in memberships: + self.assertFalse( + event["event_id"].startswith(SYNTHETIC_JOIN_EVENT_ID_PREFIX), event + ) + + def test_full_state_sync_delivers_earlier_announcement(self) -> None: + """An incremental sync with `full_state` delivers the state of an + announcement room even when there is no announcement since the token. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + room = self._assert_delivered( + self._sync(self.bob_tok, since, full_state=True), self.bob + ) + self.assertEqual(self._timeline_ids(room), []) + self.assertTrue( + {EventTypes.Create, EventTypes.Name, EventTypes.Topic} + <= self._state_types(room) + ) + + def test_full_state_sync_without_announcement(self) -> None: + """An announcement room without any announcement is left out of an + incremental sync with `full_state`. + """ + since = self._sync(self.bob_tok)["next_batch"] + + self._assert_not_delivered(self._sync(self.bob_tok, since, full_state=True)) + + def test_initial_sync_delivers_announcements(self) -> None: + """An initial sync delivers an announcement room with its most recent + announcements. + """ + event_ids = [self._announce(f"message {i}") for i in range(3)] + + room = self._assert_delivered( + self._sync(self.bob_tok, filter={"room": {"timeline": {"limit": 2}}}), + self.bob, + ) + self.assertEqual(self._timeline_ids(room), event_ids[-2:]) + self.assertTrue(room["timeline"]["limited"]) + + def test_state_changes_after_last_announcement(self) -> None: + """An announcement room is delivered on initial, incremental and + `full_state` syncs even when state changes push its last announcement + out of the timeline. The incremental sync has the state changes in its + timeline. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve( + self.config.rooms[0], room_name="New name", room_topic="New topic" + ) + ] + self.get_success(self.handler.reconcile_rooms()) + filter = {"room": {"timeline": {"limit": 1}}} + + for response in ( + self._sync(self.bob_tok, filter=filter), + self._sync(self.bob_tok, since, filter=filter), + self._sync(self.bob_tok, since, full_state=True, filter=filter), + ): + room = self._assert_delivered(response, self.bob) + self.assertNotIn(EventTypes.Message, self._timeline_types(room)) + topic = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Topic + ][-1] + self.assertEqual(topic["content"]["topic"], "New topic") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertIn(EventTypes.Topic, self._timeline_types(room)) + + def test_state_change_after_announcement(self) -> None: + """Once an announcement room has had an announcement, a state change + such as a sender leaving delivers the room to non-members with the + change in the timeline, as it does for members. + """ + self._announce("hello") + bob_since = self._sync(self.bob_tok)["next_batch"] + announcer_since = self._sync(self.announcer_tok)["next_batch"] + + leave_id = self.helper.leave(self.room_id, self.former, tok=self.former_tok)[ + "event_id" + ] + + room = self._assert_delivered(self._sync(self.bob_tok, bob_since), self.bob) + self.assertEqual(self._timeline_ids(room), [leave_id]) + room = self._sync(self.announcer_tok, announcer_since)["rooms"]["join"][ + self.room_id + ] + self.assertEqual(self._timeline_ids(room), [leave_id]) + + def test_state_change_before_first_announcement(self) -> None: + """Before an announcement room's first announcement, a state change such + as a sender leaving does not deliver the room on incremental sync. + """ + since = self._sync(self.bob_tok)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + + self._assert_not_delivered(self._sync(self.bob_tok, since)) + + def test_initial_sync_without_announcement(self) -> None: + """An announcement room without any announcement is left out of an + initial sync. + """ + self._assert_not_delivered(self._sync(self.bob_tok)) + + def test_filtered_out_room(self) -> None: + """An announcement room excluded by the sync filter is not delivered.""" + self._announce("hello") + filter = {"room": {"not_rooms": [self.room_id]}} + + since = self._sync(self.bob_tok, filter=filter)["next_batch"] + self._announce("again") + + self._assert_not_delivered(self._sync(self.bob_tok, filter=filter)) + self._assert_not_delivered(self._sync(self.bob_tok, since, filter=filter)) + self._assert_not_delivered( + self._sync(self.bob_tok, since, full_state=True, filter=filter) + ) + + def test_removed_member(self) -> None: + """A member removed from the room by the config switches straight to the + synthetic join and the room's current power levels, then gets later + announcements as a non-member, on initial sync too. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve(self.config.rooms[0], allowed_user_ids=[self.announcer]) + ] + self.get_success(self.handler.reconcile_rooms()) + response = self._sync(self.former_tok, since) + room = self._assert_delivered(response, self.former) + self.assertNotIn(EventTypes.Message, self._timeline_types(room)) + power_levels = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.PowerLevels + ][-1] + self.assertNotIn(self.former, power_levels["content"]["users"]) + + event_id = self._announce("hello") + room = self._assert_delivered( + self._sync(self.former_tok, response["next_batch"]), self.former + ) + self.assertEqual(self._timeline_ids(room), [event_id]) + + room = self._assert_delivered(self._sync(self.former_tok), self.former) + self.assertEqual(self._timeline_ids(room)[-1], event_id) + + def test_removed_member_syncing_after_announcements(self) -> None: + """A member removed from the room by the config while offline gets the + announcements sent since their removal on their next incremental sync. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve(self.config.rooms[0], allowed_user_ids=[self.announcer]) + ] + self.get_success(self.handler.reconcile_rooms()) + event_ids = [self._announce(f"message {i}") for i in range(2)] + + response = self._sync(self.former_tok, since) + self.assertNotIn(self.room_id, response["rooms"].get("leave", {})) + room = self._assert_delivered(response, self.former) + self.assertEqual(self._timeline_ids(room)[-2:], event_ids) + + def test_member_leaving(self) -> None: + """A member who leaves the room on their own switches straight to the + synthetic join, even when no announcement is sent. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + + room = self._assert_delivered(self._sync(self.former_tok, since), self.former) + self.assertEqual(self._timeline_ids(room), []) + + def test_initial_sync_after_leaving(self) -> None: + """A former member who left on their own gets a single, synthetic join + on initial sync, even when their real leave is among the recent events. + """ + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + self._announce("hello") + + self._assert_delivered(self._sync(self.former_tok), self.former) + + def test_no_device_list_changes(self) -> None: + """Delivering an announcement room does not report its members as users + whose device lists changed. + """ + since = self._sync(self.bob_tok)["next_batch"] + self._announce("hello") + + response = self._sync(self.bob_tok, since) + self._assert_delivered(response, self.bob) + self.assertEqual(response.get("device_lists", {}).get("changed", []), []) + + def test_marked_unread_without_new_events(self) -> None: + """A non-member who marks an announcement room as unread gets + `m.marked_unread` on their next incremental sync, even when nothing + else happened in the room. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + channel = self.make_request( + "PUT", + f"/user/{self.bob}/rooms/{self.room_id}/account_data/m.marked_unread", + {"unread": True}, + access_token=self.bob_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), []) + self.assertEqual( + room["account_data"]["events"], + [{"type": "m.marked_unread", "content": {"unread": True}}], + ) + + def test_marked_unread_with_new_announcement(self) -> None: + """A non-member who marks an announcement room as unread, followed by a + new announcement, gets both the announcement and `m.marked_unread` in + the room on their next incremental sync. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + channel = self.make_request( + "PUT", + f"/user/{self.bob}/rooms/{self.room_id}/account_data/m.marked_unread", + {"unread": True}, + access_token=self.bob_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + event_id = self._announce("again") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertEqual( + room["account_data"]["events"], + [{"type": "m.marked_unread", "content": {"unread": True}}], + ) + + def test_state_after(self) -> None: + """With `use_state_after`, a non-member gets the synthetic join in + `state_after`, including a former member whose real leave is in the + window. + """ + bob_since = self._sync(self.bob_tok, use_state_after=True)["next_batch"] + former_since = self._sync(self.former_tok, use_state_after=True)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + self._announce("hello") + + for tok, user_id, since in ( + (self.bob_tok, self.bob, bob_since), + (self.former_tok, self.former, former_since), + ): + room = self._assert_delivered( + self._sync(tok, since, use_state_after=True), user_id + ) + self.assertNotIn("state", room) + + def test_lazy_loaded_members(self) -> None: + """The synthetic join is sent when members are lazy-loaded.""" + self._announce("hello") + + self._assert_delivered( + self._sync( + self.bob_tok, filter={"room": {"state": {"lazy_load_members": True}}} + ), + self.bob, + ) + + def test_event_fields(self) -> None: + """With an `event_fields` filter, the synthetic join is trimmed to the + requested fields like every other event. + """ + self._announce("hello") + event_fields = ["type", "content", "state_key"] + + response = self._sync(self.bob_tok, filter={"event_fields": event_fields}) + room = response["rooms"]["join"][self.room_id] + + events = room["state"]["events"] + room["timeline"]["events"] + own_memberships = [ + event + for event in events + if event["type"] == EventTypes.Member and event["state_key"] == self.bob + ] + self.assertEqual( + own_memberships, + [ + { + "type": EventTypes.Member, + "content": {"membership": Membership.JOIN}, + "state_key": self.bob, + } + ], + ) + for event in events: + self.assertLessEqual(set(event), set(event_fields), event) + + def test_federation_event_format(self) -> None: + """With the `federation` event format, the synthetic join carries the + room ID like every other event. + """ + self._announce("hello") + + response = self._sync(self.bob_tok, filter={"event_format": "federation"}) + room = response["rooms"]["join"][self.room_id] + + events = room["state"]["events"] + room["timeline"]["events"] + own_memberships = [ + event + for event in events + if event["type"] == EventTypes.Member and event["state_key"] == self.bob + ] + self.assertEqual(len(own_memberships), 1, own_memberships) + for event in events: + self.assertEqual(event.get("room_id"), self.room_id, event)