diff --git a/rust/src/sliding_sync/encode.rs b/rust/src/sliding_sync/encode.rs new file mode 100644 index 0000000000..dcf2e3bd3b --- /dev/null +++ b/rust/src/sliding_sync/encode.rs @@ -0,0 +1,187 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + */ + +//! The servlet's `encode_rooms` for the rooms Rust computed, so that their +//! events never become Python objects. + +use std::collections::HashMap; + +use pyo3::prelude::*; +use pythonize::pythonize; +use serde_json::{json, Map, Value}; + +use crate::events::json_object::JsonObject; +use crate::events::serialize::{ + serialize_event, BundleRef, SerializeEventConfig, SerializeLookups, +}; +use crate::events::Event; +use crate::sliding_sync::result::RoomResult; +use crate::sliding_sync::room_entry::RoomEntry; +use crate::sync::relations::Bundle; + +/// The lookups of one room: the bundles and redactions Rust loaded for it, +/// and the module callbacks' unsigned additions. +struct EntryLookups<'a> { + entry: &'a RoomEntry, + unsigned_additions: &'a HashMap, +} + +impl SerializeLookups for EntryLookups<'_> { + fn redaction(&self, event_id: &str) -> Option<&Event> { + self.entry.redactions.get(event_id) + } + + fn bundle(&self, event_id: &str) -> Option> { + self.entry + .bundles + .as_ref()? + .get(event_id) + .map(Bundle::as_ref) + } + + fn unsigned_additions(&self, event_id: &str) -> Option<&JsonObject> { + self.unsigned_additions.get(event_id) + } +} + +/// `encode_rooms` for one room. +pub fn encode_room( + entry: &RoomEntry, + time_now_ms: i64, + config: &SerializeEventConfig, + unsigned_additions: &HashMap, +) -> PyResult { + let lookups = EntryLookups { + entry, + unsigned_additions, + }; + + // Notifications are only done properly by clients, so these are + // placeholders. + let mut room = Map::new(); + room.insert("notification_count".to_owned(), json!(0)); + room.insert("highlight_count".to_owned(), json!(0)); + if let Some(bump_stamp) = entry.bump_stamp { + room.insert("bump_stamp".to_owned(), json!(bump_stamp)); + } + if let Some(joined_count) = entry.joined_count { + room.insert("joined_count".to_owned(), json!(joined_count)); + } + if let Some(invited_count) = entry.invited_count { + room.insert("invited_count".to_owned(), json!(invited_count)); + } + if let Some(name) = entry.name.as_deref().filter(|name| !name.is_empty()) { + room.insert("name".to_owned(), json!(name)); + } + if let Some(avatar) = entry.avatar.as_deref().filter(|avatar| !avatar.is_empty()) { + room.insert("avatar".to_owned(), json!(avatar)); + } + if !entry.heroes.is_empty() { + let heroes: Vec = entry + .heroes + .iter() + .map(|hero| { + let mut serialized = Map::new(); + serialized.insert("user_id".to_owned(), json!(hero.user_id)); + // The spec spells it "displayname". + if let Some(display_name) = &hero.display_name { + serialized.insert("displayname".to_owned(), json!(display_name)); + } + if let Some(avatar_url) = &hero.avatar_url { + serialized.insert("avatar_url".to_owned(), json!(avatar_url)); + } + Value::Object(serialized) + }) + .collect(); + room.insert("heroes".to_owned(), Value::Array(heroes)); + } + // Only sent when true, to save bandwidth. + if entry.initial { + room.insert("initial".to_owned(), json!(true)); + } + if entry.unstable_expanded_timeline { + room.insert("unstable_expanded_timeline".to_owned(), json!(true)); + } + if !entry.required_state.is_empty() { + let state = entry + .required_state + .iter() + .map(|event| { + serialize_event(event, time_now_ms, config, None, false, &lookups) + .map(Value::Object) + }) + .collect::>>()?; + room.insert("required_state".to_owned(), Value::Array(state)); + } + if !entry.timeline.is_empty() { + let with_bundles = entry.bundles.is_some(); + let timeline = entry + .timeline + .iter() + .map(|filtered| { + serialize_event( + &filtered.event, + time_now_ms, + config, + Some(&filtered.membership), + with_bundles, + &lookups, + ) + .map(Value::Object) + }) + .collect::>>()?; + room.insert("timeline".to_owned(), Value::Array(timeline)); + } + if let Some(limited) = entry.limited { + room.insert("limited".to_owned(), json!(limited)); + } + if let Some(prev_batch) = &entry.prev_batch { + room.insert("prev_batch".to_owned(), json!(prev_batch)); + } + if let Some(num_live) = entry.num_live { + room.insert("num_live".to_owned(), json!(num_live)); + } + // Absent on rooms that are not DMs. + if entry.is_dm { + room.insert("is_dm".to_owned(), json!(true)); + } + if !entry.stripped_state.is_empty() { + room.insert( + "invite_state".to_owned(), + Value::Array(entry.stripped_state.clone()), + ); + } + Ok(Value::Object(room)) +} + +/// `encode_rooms` for `rooms`, in order. `unsigned_additions` are the +/// module callbacks' additions to the rooms' events, by event ID. +#[pyfunction] +#[pyo3(signature = (rooms, time_now_ms, config, *, unsigned_additions = None))] +pub fn encode_rooms<'py>( + py: Python<'py>, + rooms: Vec>, + time_now_ms: i64, + config: &SerializeEventConfig, + unsigned_additions: Option>, +) -> PyResult>> { + let unsigned_additions = unsigned_additions.unwrap_or_default(); + rooms + .iter() + .map(|room| { + let encoded = encode_room(&room.get().entry, time_now_ms, config, &unsigned_additions)?; + Ok(pythonize(py, &encoded)?) + }) + .collect() +} diff --git a/rust/src/sliding_sync/mod.rs b/rust/src/sliding_sync/mod.rs index ebe8f43657..b86bdaee30 100644 --- a/rust/src/sliding_sync/mod.rs +++ b/rust/src/sliding_sync/mod.rs @@ -26,22 +26,28 @@ //! ([`SlidingSyncHandler::defer`]), and a `_blocking` twin runs to completion //! for tests ([`SlidingSyncHandler::block_on`]). +use std::collections::HashMap; use std::sync::Arc; use futures::future::BoxFuture; use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyModule}; +use request::SlidingSyncRequest; +use result::{RoomInput, RoomsResult}; use crate::database::postgres::pool::{ConnectionPool, PyConnectionPool}; use crate::deferred::{create_deferred_with, OnCancel}; use crate::runtime::RustRuntime; -use crate::storage::sync_store::{SyncDbStats, SyncStore}; +use crate::storage::sync_store::{anyhow_to_py, SyncDbStats, SyncStore}; pub mod config; pub mod connection; +pub mod encode; pub mod extensions; pub mod request; pub mod required_state; +pub mod result; pub mod room; +pub mod room_entry; pub mod room_lists; /// Computes sliding sync responses with the Rust implementation, reading @@ -102,6 +108,62 @@ impl SlidingSyncHandler { let handle = self.runtime.tokio_handle()?; py.detach(|| handle.block_on(future)) } + + /// Start computing the entries of `rooms`: read them from Python, and + /// return the future that computes them off the GIL. + fn start_room_entries( + &self, + py: Python<'_>, + request: &Bound<'_, SlidingSyncRequest>, + rooms: Vec>, + instance_ids: HashMap, + ) -> PyResult>> { + let pool = self.pool(py)?; + let request = Arc::clone(&request.get().data); + let rooms: Vec<_> = rooms.iter().map(|room| room.get().data.clone()).collect(); + let store = Arc::clone(&self.store); + let now_ms = self.runtime.clock().now_millis() as i64; + Ok(Box::pin(async move { + let reader = store.sliding_sync_reader(pool); + let outcomes = + room_entry::generate_room_entries(&reader, &request, &instance_ids, now_ms, &rooms) + .await + .map_err(anyhow_to_py)?; + let stats = SlidingSyncStats::from_db(&reader.stats()); + Ok(RoomsResult::new(outcomes, request.from.is_some(), stats)) + })) + } +} + +#[pymethods] +impl SlidingSyncHandler { + /// `get_room_sync_data` for each of `rooms`, twenty at a time: a + /// deferred resolving to a `RoomsResult`. `instance_ids` are + /// `get_id_for_instance` for the writers the rooms' tokens may name. + #[pyo3(signature = (request, rooms, *, instance_ids))] + fn generate_room_entries<'py>( + &self, + py: Python<'py>, + request: &Bound<'py, SlidingSyncRequest>, + rooms: Vec>, + instance_ids: HashMap, + ) -> PyResult> { + let future = self.start_room_entries(py, request, rooms, instance_ids)?; + self.defer(py, future) + } + + /// As [`Self::generate_room_entries`], but blocks. + #[pyo3(signature = (request, rooms, *, instance_ids))] + fn generate_room_entries_blocking( + &self, + py: Python<'_>, + request: &Bound<'_, SlidingSyncRequest>, + rooms: Vec>, + instance_ids: HashMap, + ) -> PyResult { + let future = self.start_room_entries(py, request, rooms, instance_ids)?; + self.block_on(py, future) + } } /// What the Rust implementation did for one request, for the @@ -167,6 +229,10 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> child_module.add_class::()?; child_module.add_class::()?; child_module.add_class::()?; + child_module.add_class::()?; + child_module.add_class::()?; + child_module.add_class::()?; + child_module.add_function(wrap_pyfunction!(encode::encode_rooms, &child_module)?)?; child_module.add_class::()?; config::register_test_functions(&child_module)?; connection::register_test_functions(&child_module)?; diff --git a/rust/src/sliding_sync/result.rs b/rust/src/sliding_sync/result.rs new file mode 100644 index 0000000000..0c88c664ba --- /dev/null +++ b/rust/src/sliding_sync/result.rs @@ -0,0 +1,376 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + */ + +//! What crosses to Python for the rooms of a request: [`RoomInput`] in, +//! [`RoomsResult`] out, and a [`RoomResult`] per room Rust computed. +//! +//! A `RoomResult`'s events never become Python objects, except for module +//! callbacks: Python reads only what `current_sync_for_user` and the +//! extensions need, and the servlet encodes the room in Rust. + +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::sync::{Arc, Mutex}; + +use pyo3::{ + exceptions::PyValueError, + intern, + prelude::*, + types::{PyList, PyTuple}, +}; + +use crate::caches::lock_ignoring_poison; +use crate::events::Event; +use crate::sliding_sync::config::{RequiredStateMap, RoomSyncConfig, MEMBER}; +use crate::sliding_sync::connection::HaveSentRoom; +use crate::sliding_sync::room_entry::{RoomEntry, RoomInputData, RoomOutcome}; +use crate::sliding_sync::room_lists::RoomForUser; +use crate::sliding_sync::SlidingSyncStats; +use crate::storage::sync_store::PersistedEventPosition; +use crate::sync::relations::Bundle; +use crate::types::stream_token::RoomStreamToken; + +/// A `required_state_map` from Python: a mapping of types to sets of +/// state keys. +pub fn required_state_map_from_py(map: &Bound<'_, PyAny>) -> PyResult { + let py = map.py(); + let mut result = RequiredStateMap::new(); + for item in map.call_method0(intern!(py, "items"))?.try_iter()? { + let (state_type, state_keys): (String, Bound<'_, PyAny>) = item?.extract()?; + let mut keys = BTreeSet::new(); + for key in state_keys.try_iter()? { + keys.insert(key?.extract::()?); + } + result.insert(state_type, keys); + } + Ok(result) +} + +/// The input of one room's `get_room_sync_data`. See [`RoomInputData`]. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomInput { + pub(crate) data: RoomInputData, +} + +#[pymethods] +impl RoomInput { + /// `room_status` is a `HaveSentRoomFlag` value, with its token for + /// "previously". + #[new] + #[pyo3(signature = (*, room_id, sender, membership, event_id, instance_name, stream, room_version_id, timeline_limit, required_state_map, prev_room_sync_config, room_status, room_status_token, newly_joined, newly_left, is_dm))] + #[allow(clippy::too_many_arguments)] + fn py_new( + room_id: &str, + sender: Option<&str>, + membership: &str, + event_id: Option<&str>, + instance_name: &str, + stream: i64, + room_version_id: Option<&str>, + timeline_limit: i64, + required_state_map: &Bound<'_, PyAny>, + prev_room_sync_config: Option<(i64, Bound<'_, PyAny>)>, + room_status: &str, + room_status_token: Option, + newly_joined: bool, + newly_left: bool, + is_dm: bool, + ) -> PyResult { + let room_status = match (room_status, room_status_token) { + ("never", _) => HaveSentRoom::Never, + ("live", _) => HaveSentRoom::Live, + ("previously", Some(token)) => HaveSentRoom::Previously(token), + (status, _) => { + return Err(PyValueError::new_err(format!( + "bad room status {status:?} or missing token" + ))) + } + }; + let prev_room_sync_config = prev_room_sync_config + .map(|(timeline_limit, map)| { + Ok::<_, PyErr>(RoomSyncConfig { + timeline_limit, + required_state_map: required_state_map_from_py(&map)?, + }) + }) + .transpose()?; + Ok(RoomInput { + data: RoomInputData { + membership: RoomForUser { + room_id: Arc::from(room_id), + sender: sender.map(Arc::from), + membership: Arc::from(membership), + event_id: event_id.map(Arc::from), + event_pos: PersistedEventPosition { + instance_name: Arc::from(instance_name), + stream, + }, + room_version_id: room_version_id.map(Arc::from), + // Not read by the room's computation. + has_known_state: true, + room_type: None, + is_encrypted: false, + }, + room_sync_config: RoomSyncConfig { + timeline_limit, + required_state_map: required_state_map_from_py(required_state_map)?, + }, + prev_room_sync_config, + room_status, + newly_joined, + newly_left, + is_dm, + }, + }) + } + + #[getter] + fn room_id(&self) -> &str { + &self.data.membership.room_id + } +} + +/// One room's entry, computed in Rust: `SlidingSyncResult.RoomResult` as +/// far as `current_sync_for_user` and the extensions read it. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomResult { + pub(crate) entry: Arc, +} + +#[pymethods] +impl RoomResult { + fn __bool__(&self) -> bool { + self.entry.has_updates() + } + + #[getter] + fn room_id(&self) -> &str { + &self.entry.room_id + } + + #[getter] + fn initial(&self) -> bool { + self.entry.initial + } + + #[getter] + fn unstable_expanded_timeline(&self) -> bool { + self.entry.unstable_expanded_timeline + } + + #[getter] + fn timeline_event_ids(&self) -> Vec<&str> { + self.entry + .timeline + .iter() + .map(|filtered| filtered.event.event_id()) + .collect() + } + + #[getter] + fn timeline_senders(&self) -> Vec<&str> { + self.entry + .timeline + .iter() + .map(|filtered| filtered.event.sender()) + .collect() + } + + /// The state keys of the membership events in `required_state`. + #[getter] + fn required_state_member_keys(&self) -> Vec<&str> { + self.entry + .required_state + .iter() + .filter(|event| event.type_() == MEMBER) + .filter_map(Event::state_key) + .collect() + } + + #[getter] + fn hero_user_ids(&self) -> Vec<&str> { + self.entry + .heroes + .iter() + .map(|hero| hero.user_id.as_str()) + .collect() + } + + /// Every event the room's encoding serializes, for the module + /// callbacks that add to their unsigned data. + fn events_to_serialize<'py>(&self, py: Python<'py>) -> PyResult> { + let bundled = self + .entry + .bundles + .iter() + .flat_map(|bundles| bundles.values().flat_map(Bundle::events)); + let events = self + .entry + .timeline + .iter() + .map(|filtered| &filtered.event) + .chain(&self.entry.required_state) + .chain(bundled) + .map(|event| Bound::new(py, event.shallow_clone())) + .collect::>>()?; + PyList::new(py, events) + } +} + +/// A [`RequiredStateMap`] for Python: types to sets of state keys. +type PyRequiredStateMap = BTreeMap>; + +/// A room's `RoomLazyMembershipChanges`: `(room_id, +/// returned_user_id_to_last_seen_ts_map, invalidated_user_ids)`. +type LazyMembership = (Arc, HashMap>, HashSet); + +/// What [`crate::sliding_sync::room_entry::generate_room_entries`] made of +/// the rooms. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomsResult { + /// The entries to send. Taken by [`RoomsResult::take_entries`]. + entries: Mutex>>, + /// The rooms computed with nothing to send. + empty: Vec>, + /// `room_configs[room_id]` for the computed rooms whose config changes. + room_config_updates: Vec<(Arc, RoomSyncConfig)>, + /// `room_lazy_membership[room_id]` for every computed room. + lazy_memberships: Vec, + fallbacks: Vec<(Arc, String)>, + errors: Vec<(Arc, String)>, + stats: SlidingSyncStats, +} + +impl RoomsResult { + /// The result of `outcomes`. An entry with nothing to send is dropped + /// from an incremental response. + pub fn new( + outcomes: Vec<(Arc, RoomOutcome)>, + incremental: bool, + mut stats: SlidingSyncStats, + ) -> Self { + let mut entries = Vec::new(); + let mut empty = Vec::new(); + let mut room_config_updates = Vec::new(); + let mut lazy_memberships = Vec::new(); + let mut fallbacks = Vec::new(); + let mut errors = Vec::new(); + for (room_id, outcome) in outcomes { + match outcome { + RoomOutcome::Computed(computed) => { + stats.rooms_computed += 1; + if computed.lazy_member_lookup { + stats.lazy_member_lookups += 1; + } + let mutations = computed.mutations; + if let Some(config) = mutations.room_config { + room_config_updates.push((Arc::clone(&room_id), config)); + } + let lazy = mutations.lazy_membership; + lazy_memberships.push(( + Arc::clone(&room_id), + lazy.returned_user_id_to_last_seen_ts_map, + lazy.invalidated_user_ids, + )); + if computed.entry.has_updates() || !incremental { + entries.push(Arc::new(computed.entry)); + } else { + empty.push(room_id); + } + } + RoomOutcome::Fallback(reason) => fallbacks.push((room_id, reason)), + RoomOutcome::Error(error) => errors.push((room_id, error)), + } + } + RoomsResult { + entries: Mutex::new(entries), + empty, + room_config_updates, + lazy_memberships, + fallbacks, + errors, + stats, + } + } +} + +#[pymethods] +impl RoomsResult { + /// The entries to send as `RoomResult`s. A second call returns an + /// empty list. + fn take_entries<'py>(&self, py: Python<'py>) -> PyResult> { + let entries = std::mem::take(&mut *lock_ignoring_poison(&self.entries)); + let items = entries + .into_iter() + .map(|entry| Bound::new(py, RoomResult { entry })) + .collect::>>()?; + PyList::new(py, items) + } + + #[getter] + fn empty(&self) -> Vec<&str> { + self.empty.iter().map(|room_id| &**room_id).collect() + } + + /// `(room_id, timeline_limit, required_state_map)` for each room whose + /// config the connection records. + #[getter] + fn room_config_updates<'py>(&self, py: Python<'py>) -> PyResult> { + let items = self + .room_config_updates + .iter() + .map(|(room_id, config)| { + let map: PyRequiredStateMap = config.required_state_map.clone(); + (&**room_id, config.timeline_limit, map).into_pyobject(py) + }) + .collect::>>>()?; + PyList::new(py, items) + } + + /// `(room_id, returned_user_id_to_last_seen_ts_map, + /// invalidated_user_ids)` for each room computed. + #[getter] + fn lazy_memberships<'py>(&self, py: Python<'py>) -> PyResult> { + let items = self + .lazy_memberships + .iter() + .map(|(room_id, returned, invalidated)| { + (&**room_id, returned, invalidated).into_pyobject(py) + }) + .collect::>>>()?; + PyList::new(py, items) + } + + /// `(room_id, reason)` for each room Python must compute. + #[getter] + fn fallbacks(&self) -> Vec<(&str, &str)> { + self.fallbacks + .iter() + .map(|(room_id, reason)| (&**room_id, reason.as_str())) + .collect() + } + + /// `(room_id, error)` for each room Rust failed on. + #[getter] + fn errors(&self) -> Vec<(&str, &str)> { + self.errors + .iter() + .map(|(room_id, error)| (&**room_id, error.as_str())) + .collect() + } + + #[getter] + fn stats(&self) -> SlidingSyncStats { + self.stats + } +} diff --git a/rust/src/sliding_sync/room_entry.rs b/rust/src/sliding_sync/room_entry.rs new file mode 100644 index 0000000000..5ec367c342 --- /dev/null +++ b/rust/src/sliding_sync/room_entry.rs @@ -0,0 +1,925 @@ +/* + * This file is licensed under the Affero General Public License (AGPL) version 3. + * + * Copyright (C) 2026 Element Creations Ltd. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * See the GNU Affero General Public License for more details: + * . + */ + +//! `get_room_sync_data`: one room's entry in a sliding sync response, in +//! the order of the Python, and the batch of rooms Python hands over in +//! one call. +//! +//! The pure decisions are in [`crate::sliding_sync::room`] and +//! [`crate::sliding_sync::required_state`]. This module does the reads +//! between them. + +use std::collections::{HashMap, HashSet}; +use std::panic::AssertUnwindSafe; +use std::sync::Arc; + +use anyhow::{bail, Context}; +use futures::{stream, FutureExt, StreamExt}; +use indexmap::IndexMap; +use serde_json::{Map, Value}; + +use crate::events::Event; +use crate::sliding_sync::config::{RoomSyncConfig, MEMBER}; +use crate::sliding_sync::connection::{HaveSentRoom, RoomLazyMembershipChanges}; +use crate::sliding_sync::request::SlidingSyncRequestData; +use crate::sliding_sync::required_state::{ + required_state_changes, required_state_request, state_filter_with_meta, TimelineMember, +}; +use crate::sliding_sync::room::{ + always_return_bump_stamp, bump_stamp_from_last_event, final_bump_stamp, needs_heroes, num_live, + room_config_to_persist, store_bump_stamp, strip_event, stripped_state, timeline_bounds, + timeline_bump_stamp, StoreBumpStamp, StrippedHero, TimelineBounds, BUMP_EVENT_TYPES, +}; +use crate::sliding_sync::room_lists::RoomForUser; +use crate::state::{lookup, StateKey, StateMap}; +use crate::storage::events::RedactBehaviour; +use crate::storage::sliding_sync_store::SlidingSyncReader; +use crate::storage::sync_store::{PaginationOrder, RoomSummary}; +use crate::sync::client_events::{filter_events_for_client, ClientFilterContext}; +use crate::sync::relations::{bundled_aggregations, Bundle}; +use crate::sync::room::Fallback; +use crate::sync::rooms::panic_message; +use crate::sync::summary::extract_heroes_from_room_summary; +use crate::types::state_filter::StateFilter; +use crate::types::stream_token::{Direction, RoomStreamToken}; +use crate::visibility::FilteredEvent; + +const M_ROOM_NAME: &str = "m.room.name"; +const M_ROOM_AVATAR: &str = "m.room.avatar"; + +/// How many rooms are computed at once, as `concurrently_execute` does. +const ROOM_ENTRY_CONCURRENCY: usize = 20; + +/// What `current_sync_for_user` passes `get_room_sync_data` for one room, +/// besides the request. +#[derive(Debug, Clone)] +pub struct RoomInputData { + /// The user's membership in the room at the `to_token`. + pub membership: RoomForUser, + pub room_sync_config: RoomSyncConfig, + /// What the connection last recorded for the room. + pub prev_room_sync_config: Option, + /// Whether the connection sent the room before. + pub room_status: HaveSentRoom, + pub newly_joined: bool, + pub newly_left: bool, + pub is_dm: bool, +} + +/// `SlidingSyncResult.RoomResult`, holding the events Rust loaded. +pub struct RoomEntry { + pub room_id: Arc, + /// May be empty, which counts as an update but is not sent. + pub name: Option, + pub avatar: Option, + pub heroes: Vec, + pub is_dm: bool, + pub initial: bool, + pub unstable_expanded_timeline: bool, + pub required_state: Vec, + pub timeline: Vec, + /// `None` unless the timeline is limited. + pub bundles: Option, Bundle>>, + /// The redactions of the redacted events among the room's, which the + /// serializer adds to their unsigned data. + pub redactions: HashMap, Event>, + pub stripped_state: Vec, + /// The `prev_batch` token as a string. + pub prev_batch: Option, + pub limited: Option, + pub num_live: Option, + pub bump_stamp: Option, + pub joined_count: Option, + pub invited_count: Option, +} + +impl RoomEntry { + /// `RoomResult.__bool__`: whether the entry holds an update to return + /// at once. + pub fn has_updates(&self) -> bool { + self.initial + || self.name.is_some() + || self.avatar.is_some() + || !self.heroes.is_empty() + || self.joined_count.is_some() + || self.invited_count.is_some() + || !self.required_state.is_empty() + || !self.timeline.is_empty() + || !self.stripped_state.is_empty() + } +} + +/// What `get_room_sync_data` writes to the new connection state for the +/// room. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RoomMutations { + /// `room_configs[room_id]`, if it changes. + pub room_config: Option, + /// `room_lazy_membership[room_id]`, always set. + pub lazy_membership: RoomLazyMembershipChanges, +} + +/// One room computed. +pub struct RoomComputed { + pub entry: RoomEntry, + pub mutations: RoomMutations, + /// Whether the room's lazy-loaded members were looked up in the + /// connection state. + pub lazy_member_lookup: bool, +} + +/// How one room came out. +pub enum RoomOutcome { + Computed(Box), + /// Python must compute the room: [`Fallback`]. + Fallback(String), + /// Computing the room failed. + Error(String), +} + +/// What computing one room reads, shared by the rooms of a request. +pub struct RoomContext<'r, 'a> { + pub reader: &'r SlidingSyncReader<'a>, + pub request: &'r SlidingSyncRequestData, + pub client_filter: ClientFilterContext, + /// `get_id_for_instance` for the writers the rooms' tokens may name. + pub instance_ids: &'r HashMap, + /// The time the visibility checks compare with. + pub now_ms: i64, +} + +/// One room's view of the request: the room, and where the user's view of +/// it ends. +struct Room<'r> { + input: &'r RoomInputData, + room_id: &'r str, + membership: &'r str, + /// Where a leave or ban stops what the user sees: the membership's + /// position, as `event_pos.to_room_stream_token()`. + membership_token: RoomStreamToken, +} + +impl Room<'_> { + fn has_left(&self) -> bool { + matches!(self.membership, "leave" | "ban") + } +} + +/// The events of `event_ids`, by state key. Events that are not found or +/// rejected are left out. +/// +/// An event of the room's timeline is that event, with the previous content +/// the timeline's load gave it. Python's loads find the same objects through +/// its map of the events still held, whatever the event cache holds. +async fn state_events( + reader: &SlidingSyncReader<'_>, + timeline: &HashMap<&str, &Arc>, + event_ids: &[Arc], +) -> anyhow::Result> { + let (held, to_load): (Vec<&Arc>, Vec<&Arc>) = event_ids + .iter() + .partition(|event_id| timeline.contains_key(&***event_id)); + let to_load: Vec> = to_load.into_iter().cloned().collect(); + let loaded = reader + .events(&to_load, RedactBehaviour::Redact, false, false) + .await?; + let held = held + .into_iter() + .map(|event_id| timeline[&**event_id].shallow_clone()); + Ok(loaded + .into_iter() + .chain(held) + .filter_map(|event| { + let key = StateKey::new(event.type_(), event.state_key()?); + Some((key, event)) + }) + .collect()) +} + +/// `get_event`: the event, which must exist. +async fn one_event(reader: &SlidingSyncReader<'_>, event_id: &Arc) -> anyhow::Result { + reader + .events( + std::slice::from_ref(event_id), + RedactBehaviour::Redact, + false, + false, + ) + .await? + .into_iter() + .next() + .with_context(|| format!("Could not find event {event_id}")) +} + +/// `get_current_state_ids_at`: the room's state as the user sees it, at +/// their leave or ban, or the current state. +async fn state_ids_at( + context: &RoomContext<'_, '_>, + room: &Room<'_>, + state_filter: &StateFilter, +) -> anyhow::Result { + if room.has_left() { + context + .reader + .state_ids_at(room.room_id, &room.membership_token, state_filter) + .await + } else { + context + .reader + .current_state_ids(room.room_id, state_filter) + .await + } +} + +/// `get_current_state_at`: as [`state_ids_at`], with the events. +async fn state_at( + context: &RoomContext<'_, '_>, + room: &Room<'_>, + timeline: &HashMap<&str, &Arc>, + state_filter: &StateFilter, +) -> anyhow::Result> { + let state_ids = state_ids_at(context, room, state_filter).await?; + let event_ids: Vec> = state_ids.into_values().collect(); + state_events(context.reader, timeline, &event_ids).await +} + +/// `_stripped_state` of an invite or knock event. +fn stripped_state_of(event: &Event) -> anyhow::Result> { + let mut unsigned = Map::new(); + let unsigned_ref = event.unsigned_ref(); + if let Some(state) = unsigned_ref.invite_room_state()? { + unsigned.insert("invite_room_state".to_owned(), Value::Array(state)); + } + if let Some(state) = unsigned_ref.knock_room_state()? { + unsigned.insert("knock_room_state".to_owned(), Value::Array(state)); + } + let content = Value::Object( + event + .content() + .as_map() + .iter() + .map(|(key, value)| (key.to_string(), value.clone())) + .collect(), + ); + let membership = event + .content() + .get_field("membership") + .and_then(Value::as_str); + let stripped = strip_event( + event.type_(), + event.state_key().unwrap_or_default(), + &content, + event.sender(), + ); + Ok(stripped_state(membership, &unsigned, stripped)) +} + +/// A string field of an event's content, as the Python's `isinstance(value, +/// str)` checks read it. +fn content_str(event: &Event, key: &str) -> Option { + event + .content() + .get_field(key) + .and_then(Value::as_str) + .map(str::to_owned) +} + +/// `_get_bump_stamp`. +async fn bump_stamp( + context: &RoomContext<'_, '_>, + room_id: &str, + timeline: &[FilteredEvent], + check_outside_timeline: bool, +) -> anyhow::Result> { + let in_timeline = timeline_bump_stamp(timeline.iter().filter_map(|filtered| { + let stream = filtered.event.internal_metadata().stream_ordering()?; + Some((filtered.event.type_(), stream.get())) + })); + if in_timeline.is_some() { + return Ok(in_timeline); + } + if !check_outside_timeline { + return Ok(None); + } + let request = context.request; + let latest = context.reader.latest_bump_stamp(room_id).await?; + match store_bump_stamp( + latest, + request.have_finished_bg_jobs, + request.to_room_key.stream, + ) { + StoreBumpStamp::Done(stamp) => Ok(stamp), + StoreBumpStamp::LookupLastEvent => { + let types: Vec = BUMP_EVENT_TYPES.iter().map(|t| t.to_string()).collect(); + let last = context + .reader + .last_event_of_types_before(room_id, &request.to_room_key, &types) + .await?; + Ok(bump_stamp_from_last_event(last.map(|(_, pos)| pos.stream))) + } + } +} + +/// The redactions of the redacted events of `entry`. +async fn load_redactions( + reader: &SlidingSyncReader<'_>, + entry: &RoomEntry, +) -> anyhow::Result, Event>> { + let mut redaction_ids: Vec> = Vec::new(); + let bundled = entry + .bundles + .iter() + .flat_map(|bundles| bundles.values().flat_map(Bundle::events)); + let events = entry + .timeline + .iter() + .map(|filtered| &filtered.event) + .chain(&entry.required_state) + .chain(bundled); + for event in events { + if let Some(redacted_by) = event.internal_metadata().redacted_by()? { + let redacted_by: Arc = Arc::from(redacted_by); + if !redaction_ids.contains(&redacted_by) { + redaction_ids.push(redacted_by); + } + } + } + Ok(reader + .events(&redaction_ids, RedactBehaviour::Redact, false, false) + .await? + .into_iter() + .map(|event| (Arc::from(event.event_id()), event)) + .collect()) +} + +/// `get_room_sync_data`. +pub async fn room_sync_data( + context: &RoomContext<'_, '_>, + input: &RoomInputData, +) -> anyhow::Result { + let reader = context.reader; + let request = context.request; + let user_id = &*request.user_id; + let member = &input.membership; + let room = Room { + input, + room_id: &member.room_id, + membership: &member.membership, + membership_token: RoomStreamToken::live(member.event_pos.stream), + }; + let room_id = room.room_id; + let membership = room.membership; + let config = &input.room_sync_config; + let prev_config = input.prev_room_sync_config.as_ref(); + let from = request.from.as_ref(); + + // Only a state reset out of the room leaves a membership with no event. + let state_reset_out_of_room = member.event_id.is_none(); + if state_reset_out_of_room && membership != "leave" { + bail!("{room_id}: membership {membership} with no event"); + } + + let TimelineBounds { + from_bound, + initial, + ignore_timeline_bound, + } = timeline_bounds( + &room.input.room_status, + from.map(|from| &from.room_key), + input.newly_joined || state_reset_out_of_room, + prev_config, + config, + ); + + // The timeline. Invites and knocks only have stripped state. + let mut timeline: Vec = Vec::new(); + // The timeline's events as loaded, before visibility. + let mut loaded_timeline: Vec> = Vec::new(); + let mut bundles = None; + let mut limited = None; + let mut prev_batch = None; + let mut live = None; + if config.timeline_limit > 0 && !matches!(membership, "invite" | "knock") { + let to_bound = if room.has_left() { + &room.membership_token + } else { + &request.to_room_key + }; + let timeline_from_bound = if ignore_timeline_bound { + None + } else { + from_bound.as_ref() + }; + // A historical section of the timeline in topological order, as + // `/messages` returns it, or the updates since the bound in the + // order the server received them. + let order = if timeline_from_bound.is_none() { + PaginationOrder::Topological + } else { + PaginationOrder::Stream + }; + let (mut events, new_room_key, page_limited) = reader + .paginate_room_events( + room_id, + to_bound, + timeline_from_bound, + Direction::Backwards, + config.timeline_limit, + order, + ) + .await?; + events.reverse(); + loaded_timeline = events.into_iter().map(Arc::new).collect(); + timeline = filter_events_for_client( + reader, + &context.client_filter, + &loaded_timeline, + &HashSet::new(), + membership != "join", + context.now_ms, + ) + .await?; + + let mut positions = Vec::new(); + if from.is_some() { + for filtered in &timeline { + let metadata = filtered.event.internal_metadata(); + let stream = metadata.stream_ordering().with_context(|| { + format!("event {} has no stream ordering", filtered.event.event_id()) + })?; + let instance_name = metadata.instance_name().with_context(|| { + format!("event {} has no instance name", filtered.event.event_id()) + })?; + positions.push((instance_name, stream.get())); + } + } + live = Some(num_live( + positions + .iter() + .map(|(instance_name, stream)| (instance_name.as_str(), *stream)), + from.map(|from| &from.room_key), + )); + + // With a limited timeline, the client lacks the events it would + // need to aggregate relations itself. + if page_limited { + bundles = Some( + bundled_aggregations( + reader, + &timeline, + user_id, + &context.client_filter.ignored_users, + ) + .await?, + ); + } + limited = Some(page_limited); + + let room_key = new_room_key + .to_string_with_instances(|name| context.instance_ids.get(name).copied()) + .map_err(|e| anyhow::anyhow!("{room_id}: prev_batch: {e}"))?; + prev_batch = Some(format!("{room_key}_{}", request.to_suffix)); + } + + // Python holds no reference to a redacted event, which is loaded again. + let held: HashMap<&str, &Arc> = loaded_timeline + .iter() + .filter(|event| matches!(event.internal_metadata().redacted_by(), Ok(None))) + .map(|event| (event.event_id(), event)) + .collect(); + + // Invites and knocks carry the stripped state the potential joiner + // identifies the room by. + let mut stripped = Vec::new(); + if matches!(membership, "invite" | "knock") { + let event_id = member + .event_id + .as_ref() + .with_context(|| format!("{room_id}: {membership} with no event"))?; + stripped = stripped_state_of(&one_event(reader, event_id).await?)?; + } + + // The changes to the current state in the token range, in the order of + // Python's dict. + let mut deltas: IndexMap> = IndexMap::new(); + let mut membership_changed = false; + let mut name_changed = false; + let mut avatar_changed = false; + if !initial { + let from_bound = from_bound + .as_ref() + .with_context(|| format!("{room_id}: incremental with no from bound"))?; + + // An out-of-band leave, such as a rejected remote invite, is not in + // the state deltas: it is passed in as newly left. + if input.newly_left { + if let Some(event_id) = &member.event_id { + membership_changed = true; + let leave = one_event(reader, event_id).await?; + if let Some(state_key) = leave.state_key() { + deltas.insert( + StateKey::new(leave.type_(), state_key), + Arc::clone(event_id), + ); + } + } + } + + let to_bound = match membership { + "leave" | "ban" => Some(&room.membership_token), + "join" => Some(&request.to_room_key), + "invite" | "knock" => None, + other => bail!("Unexpected membership {other} that we don't know how to handle yet"), + }; + let state_deltas = match to_bound { + Some(to_bound) => { + reader + .state_deltas_by_stream_id(room_id, Some(from_bound), Some(to_bound)) + .await? + } + None => Vec::new(), + }; + for delta in state_deltas { + // A state reset leaves a delta with no event, which is not + // handled yet. + if let Some(event_id) = &delta.event_id { + deltas.insert( + StateKey(Arc::clone(&delta.event_type), Arc::clone(&delta.state_key)), + Arc::clone(event_id), + ); + } + if &*delta.event_type == MEMBER { + membership_changed = true; + } else if &*delta.event_type == M_ROOM_NAME && delta.state_key.is_empty() { + name_changed = true; + } else if &*delta.event_type == M_ROOM_AVATAR && delta.state_key.is_empty() { + avatar_changed = true; + } + } + } + + let mut room_name = None; + if initial || name_changed { + let name_state = state_at( + context, + &room, + &held, + &StateFilter::from_types([(M_ROOM_NAME, Some(""))]), + ) + .await?; + room_name = + lookup(&name_state, (M_ROOM_NAME, "")).and_then(|event| content_str(event, "name")); + } + + // Heroes, when the room has no name, on the first send or when the + // membership changed. The summary also gives the member counts. + let mut summary: Option> = None; + let mut hero_user_ids: Vec> = Vec::new(); + if needs_heroes(room_name.as_deref(), initial, membership_changed) { + // The summary of a room left, invited to or knocked on is not known. + let details = if membership == "join" { + reader.room_summary(room_id).await? + } else { + Arc::new(RoomSummary::new()) + }; + hero_user_ids = extract_heroes_from_room_summary(&details, user_id); + summary = Some(details); + } + + let mut joined_count = None; + let mut invited_count = None; + if (initial || membership_changed) && membership == "join" { + let counts = match &summary { + Some(details) => { + let count = |m: &str| details.get(m).map_or(0, |summary| summary.count); + (count("join"), count("invite")) + } + None => { + let counts = reader.cached_member_counts(room_id).await?; + let count = |m: &str| counts.get(m).copied().unwrap_or(0); + (count("join"), count("invite")) + } + }; + joined_count = Some(counts.0); + invited_count = Some(counts.1); + } + + // The state the request asks for, from the timeline and the deltas + // before any are added below. + let delta_keys: Vec<(String, String)> = deltas + .keys() + .map(|key| (key.0.to_string(), key.1.to_string())) + .collect(); + let timeline_members: Vec> = timeline + .iter() + .map(|filtered| TimelineMember { + sender: filtered.event.sender(), + event_type: filtered.event.type_(), + state_key: filtered.event.state_key(), + }) + .collect(); + let state_request = required_state_request( + user_id, + membership, + &config.required_state_map, + &timeline_members, + limited, + initial, + delta_keys.iter().map(|(t, k)| (t.as_str(), k.as_str())), + ); + drop(timeline_members); + let required_state_filter = &state_request.required_state_filter; + let lazy_load_user_ids = &state_request.lazy_load_user_ids; + + let state_filter = state_filter_with_meta( + required_state_filter, + &hero_user_ids, + initial || avatar_changed, + ); + + // Every lazy-loaded member counts as sent first in this request, until + // the connection says otherwise. + let mut returned: HashMap> = if state_request.lazy_load_room_members { + lazy_load_user_ids + .iter() + .map(|user_id| (user_id.clone(), None)) + .collect() + } else { + HashMap::new() + }; + let mut invalidated = HashSet::new(); + let mut changed_required_state_map = None; + let mut lazy_member_lookup = false; + + let room_state: HashMap; + // The heroes' memberships, when not in `room_state`. + let mut hero_state: Option> = None; + if initial { + room_state = state_at(context, &room, &held, &state_filter).await?; + } else { + let from = from.with_context(|| format!("{room_id}: incremental with no from token"))?; + if let Some(prev_config) = prev_config { + // The members whose state may be sent: explicit, lazy-loaded, + // and those whose membership changed. + let mut all_required_user_state: HashSet<&str> = state_request + .explicit_user_state + .iter() + .chain(lazy_load_user_ids) + .map(String::as_str) + .collect(); + for (state_type, state_key) in &delta_keys { + if state_type == MEMBER { + all_required_user_state.insert(state_key); + } + } + + let mut previously_returned: HashSet = HashSet::new(); + if !all_required_user_state.is_empty() { + lazy_member_lookup = true; + let user_ids: Vec> = all_required_user_state + .iter() + .map(|u| Arc::from(*u)) + .collect(); + let last_seen = reader + .connection_lazy_members(from.connection_position, room_id, &user_ids) + .await?; + for (user_id, last_seen_ts) in last_seen { + if lazy_load_user_ids.contains(&*user_id) { + returned.insert(user_id.to_string(), Some(last_seen_ts)); + } + previously_returned.insert(user_id.to_string()); + } + } + + let changes = required_state_changes( + user_id, + &prev_config.required_state_map, + &config.required_state_map, + &previously_returned, + lazy_load_user_ids, + &delta_keys, + ); + changed_required_state_map = changes.changed_required_state_map; + invalidated = changes.lazy_members_invalidated; + // Explicit members converted to lazy ones: when they were last + // sent is not known, so a timestamp already there is kept. + for user_id in changes.extra_users_to_add_to_lazy_cache { + returned.entry(user_id).or_insert(None); + } + if changes.added_state_filter != StateFilter::none() { + // State the request newly asks for, on top of the deltas. + let added = state_ids_at(context, &room, &changes.added_state_filter).await?; + deltas.extend(added); + } + } + + let delta_state: StateMap = deltas + .iter() + .map(|(key, event_id)| (key.clone(), Arc::clone(event_id))) + .collect(); + let event_ids: Vec> = state_filter + .filter_state(&delta_state) + .into_values() + .collect(); + room_state = state_events(reader, &held, &event_ids).await?; + + if !hero_user_ids.is_empty() { + let hero_filter = + StateFilter::from_types(hero_user_ids.iter().map(|u| (MEMBER, Some(&**u)))); + hero_state = Some(state_at(context, &room, &held, &hero_filter).await?); + } + } + + let room_avatar = + lookup(&room_state, (M_ROOM_AVATAR, "")).and_then(|event| content_str(event, "url")); + + let hero_membership_state = hero_state.as_ref().unwrap_or(&room_state); + let heroes: Vec = hero_user_ids + .iter() + .filter_map(|hero| { + let event = lookup(hero_membership_state, (MEMBER, hero))?; + Some(StrippedHero { + user_id: hero.to_string(), + display_name: content_str(event, "displayname"), + avatar_url: content_str(event, "avatar_url"), + }) + }) + .collect(); + + let mut required_state: Vec = Vec::new(); + if *required_state_filter != StateFilter::none() { + let mut keys: Vec<&StateKey> = room_state + .keys() + .filter(|key| required_state_filter.matches(&key.0, &key.1)) + .collect(); + keys.sort(); + required_state = keys + .into_iter() + .map(|key| room_state[key].shallow_clone()) + .collect(); + } + + // The bump stamp: the last bump event for a joined room, else the + // membership's position if the client needs one. + let always_return = always_return_bump_stamp(membership, limited, initial); + let new_bump_stamp = if membership == "join" { + bump_stamp(context, room_id, &timeline, always_return).await? + } else { + None + }; + let bump_stamp = final_bump_stamp(new_bump_stamp, always_return, member.event_pos.stream); + + let room_config = room_config_to_persist( + config, + prev_config, + changed_required_state_map.as_ref(), + ignore_timeline_bound, + limited, + ); + + let mut entry = RoomEntry { + room_id: Arc::clone(&member.room_id), + name: room_name, + avatar: room_avatar, + heroes, + is_dm: input.is_dm, + initial, + unstable_expanded_timeline: ignore_timeline_bound, + required_state, + timeline, + bundles, + redactions: HashMap::new(), + stripped_state: stripped, + prev_batch, + limited, + num_live: live, + bump_stamp, + joined_count, + invited_count, + }; + entry.redactions = load_redactions(reader, &entry).await?; + + Ok(RoomComputed { + entry, + mutations: RoomMutations { + room_config, + lazy_membership: RoomLazyMembershipChanges { + returned_user_id_to_last_seen_ts_map: returned, + invalidated_user_ids: invalidated, + }, + }, + lazy_member_lookup, + }) +} + +/// One room's outcome. A panic is this room's error, not the request's. +async fn room_outcome( + context: &RoomContext<'_, '_>, + input: &RoomInputData, +) -> (Arc, RoomOutcome) { + let outcome = match AssertUnwindSafe(room_sync_data(context, input)) + .catch_unwind() + .await + { + Ok(Ok(computed)) => RoomOutcome::Computed(Box::new(computed)), + Ok(Err(err)) => match err.downcast::() { + Ok(fallback) => RoomOutcome::Fallback(fallback.0), + Err(err) => RoomOutcome::Error(format!("{err:#}")), + }, + Err(panic) => RoomOutcome::Error(format!("panicked: {}", panic_message(&*panic))), + }; + (Arc::clone(&input.membership.room_id), outcome) +} + +/// The entries of `rooms`, twenty at a time, in the order they finish. +pub async fn generate_room_entries( + reader: &SlidingSyncReader<'_>, + request: &SlidingSyncRequestData, + instance_ids: &HashMap, + now_ms: i64, + rooms: &[RoomInputData], +) -> anyhow::Result, RoomOutcome)>> { + let ignored_users = reader.ignored_users(&request.user_id).await?; + let context = RoomContext { + reader, + request, + client_filter: ClientFilterContext { + user_id: Arc::clone(&request.user_id), + ignored_users: Arc::new(ignored_users), + admin_policy: request.admin_policy, + msc4354_enabled: request.msc4354_enabled, + }, + instance_ids, + now_ms, + }; + let outcomes: Vec<_> = rooms + .iter() + .map(|room| room_outcome(&context, room)) + .collect(); + Ok(stream::iter(outcomes) + .buffer_unordered(ROOM_ENTRY_CONCURRENCY) + .collect() + .await) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn empty_entry() -> RoomEntry { + RoomEntry { + room_id: Arc::from("!r:test"), + name: None, + avatar: None, + heroes: Vec::new(), + is_dm: true, + initial: false, + unstable_expanded_timeline: true, + required_state: Vec::new(), + timeline: Vec::new(), + bundles: None, + redactions: HashMap::new(), + stripped_state: Vec::new(), + prev_batch: Some("s1_0".to_owned()), + limited: Some(true), + num_live: Some(0), + bump_stamp: Some(5), + joined_count: None, + invited_count: None, + } + } + + #[test] + fn what_counts_as_an_update() { + // As in Python: a DM flag, an expanded timeline, a token or a bump + // stamp alone are not updates. + assert!(!empty_entry().has_updates()); + // An empty name is, though it is not sent. + let entry = RoomEntry { + name: Some(String::new()), + ..empty_entry() + }; + assert!(entry.has_updates()); + let entry = RoomEntry { + joined_count: Some(0), + ..empty_entry() + }; + assert!(entry.has_updates()); + let entry = RoomEntry { + initial: true, + ..empty_entry() + }; + assert!(entry.has_updates()); + } +}