use std::{ collections::{BTreeMap, BTreeSet}, mem, sync::{Arc, Mutex}, }; use ruma::{ api::client::{ device::Device, filter::FilterDefinition, sync::sync_events::{ self, v4::{ExtensionsConfig, SyncRequestList}, }, }, encryption::{CrossSigningKey, DeviceKeys, OneTimeKey}, events::AnyToDeviceEvent, serde::Raw, DeviceId, OneTimeKeyAlgorithm, OneTimeKeyName, OwnedDeviceId, OwnedKeyId, OwnedMxcUri, OwnedOneTimeKeyId, OwnedRoomId, OwnedUserId, UInt, UserId, }; use crate::{services, Error, Result}; mod data; pub(crate) use data::Data; pub(crate) struct SlidingSyncCache { lists: BTreeMap, subscriptions: BTreeMap, // For every room, the roomsince number known_rooms: BTreeMap>, extensions: ExtensionsConfig, } pub(crate) struct Service { pub(crate) db: &'static dyn Data, #[allow(clippy::type_complexity)] pub(crate) connections: Mutex< BTreeMap< (OwnedUserId, OwnedDeviceId, String), Arc>, >, >, } impl Service { pub(crate) fn new(db: &'static dyn Data) -> Self { Self { db, connections: Mutex::new(BTreeMap::new()), } } /// Check if a user has an account on this homeserver. pub(crate) fn exists(&self, user_id: &UserId) -> Result { self.db.exists(user_id) } pub(crate) fn forget_sync_request_connection( &self, user_id: OwnedUserId, device_id: OwnedDeviceId, conn_id: String, ) { self.connections.lock().unwrap().remove(&(user_id, device_id, conn_id)); } #[allow(clippy::too_many_lines)] pub(crate) fn update_sync_request_with_cache( &self, user_id: OwnedUserId, device_id: OwnedDeviceId, request: &mut sync_events::v4::Request, ) -> BTreeMap> { let Some(conn_id) = request.conn_id.clone() else { return BTreeMap::new(); }; let mut cache = self.connections.lock().unwrap(); let cached = Arc::clone( cache.entry((user_id, device_id, conn_id)).or_insert_with(|| { Arc::new(Mutex::new(SlidingSyncCache { lists: BTreeMap::new(), subscriptions: BTreeMap::new(), known_rooms: BTreeMap::new(), extensions: ExtensionsConfig::default(), })) }), ); let cached = &mut cached.lock().unwrap(); drop(cache); for (list_id, list) in &mut request.lists { if let Some(cached_list) = cached.lists.get(list_id) { if list.sort.is_empty() { list.sort.clone_from(&cached_list.sort); }; if list.room_details.required_state.is_empty() { list.room_details .required_state .clone_from(&cached_list.room_details.required_state); }; list.room_details.timeline_limit = list .room_details .timeline_limit .or(cached_list.room_details.timeline_limit); list.include_old_rooms = list .include_old_rooms .clone() .or(cached_list.include_old_rooms.clone()); match (&mut list.filters, cached_list.filters.clone()) { (Some(list_filters), Some(cached_filters)) => { list_filters.is_dm = list_filters.is_dm.or(cached_filters.is_dm); if list_filters.spaces.is_empty() { list_filters.spaces = cached_filters.spaces; } list_filters.is_encrypted = list_filters .is_encrypted .or(cached_filters.is_encrypted); list_filters.is_invite = list_filters.is_invite.or(cached_filters.is_invite); if list_filters.room_types.is_empty() { list_filters.room_types = cached_filters.room_types; } if list_filters.not_room_types.is_empty() { list_filters.not_room_types = cached_filters.not_room_types; } list_filters.room_name_like = list_filters .room_name_like .clone() .or(cached_filters.room_name_like); if list_filters.tags.is_empty() { list_filters.tags = cached_filters.tags; } if list_filters.not_tags.is_empty() { list_filters.not_tags = cached_filters.not_tags; } } (_, Some(cached_filters)) => { list.filters = Some(cached_filters); } (Some(list_filters), _) => { list.filters = Some(list_filters.clone()); } (..) => {} } if list.bump_event_types.is_empty() { list.bump_event_types .clone_from(&cached_list.bump_event_types); }; } cached.lists.insert(list_id.clone(), list.clone()); } cached.subscriptions.extend( request .room_subscriptions .iter() .map(|(k, v)| (k.clone(), v.clone())), ); request.room_subscriptions.extend( cached.subscriptions.iter().map(|(k, v)| (k.clone(), v.clone())), ); request.extensions.e2ee.enabled = request.extensions.e2ee.enabled.or(cached.extensions.e2ee.enabled); request.extensions.to_device.enabled = request .extensions .to_device .enabled .or(cached.extensions.to_device.enabled); request.extensions.account_data.enabled = request .extensions .account_data .enabled .or(cached.extensions.account_data.enabled); request.extensions.account_data.lists = request .extensions .account_data .lists .clone() .or(cached.extensions.account_data.lists.clone()); request.extensions.account_data.rooms = request .extensions .account_data .rooms .clone() .or(cached.extensions.account_data.rooms.clone()); cached.extensions = request.extensions.clone(); cached.known_rooms.clone() } pub(crate) fn update_sync_subscriptions( &self, user_id: OwnedUserId, device_id: OwnedDeviceId, conn_id: String, subscriptions: BTreeMap, ) { let mut cache = self.connections.lock().unwrap(); let cached = Arc::clone( cache.entry((user_id, device_id, conn_id)).or_insert_with(|| { Arc::new(Mutex::new(SlidingSyncCache { lists: BTreeMap::new(), subscriptions: BTreeMap::new(), known_rooms: BTreeMap::new(), extensions: ExtensionsConfig::default(), })) }), ); let cached = &mut cached.lock().unwrap(); drop(cache); cached.subscriptions = subscriptions; } pub(crate) fn update_sync_known_rooms( &self, user_id: OwnedUserId, device_id: OwnedDeviceId, conn_id: String, list_id: String, new_cached_rooms: BTreeSet, globalsince: u64, ) { let mut cache = self.connections.lock().unwrap(); let cached = Arc::clone( cache.entry((user_id, device_id, conn_id)).or_insert_with(|| { Arc::new(Mutex::new(SlidingSyncCache { lists: BTreeMap::new(), subscriptions: BTreeMap::new(), known_rooms: BTreeMap::new(), extensions: ExtensionsConfig::default(), })) }), ); let cached = &mut cached.lock().unwrap(); drop(cache); for (roomid, lastsince) in cached.known_rooms.entry(list_id.clone()).or_default().iter_mut() { if !new_cached_rooms.contains(roomid) { *lastsince = 0; } } let list = cached.known_rooms.entry(list_id).or_default(); for roomid in new_cached_rooms { list.insert(roomid, globalsince); } } /// Check if account is deactivated pub(crate) fn is_deactivated(&self, user_id: &UserId) -> Result { self.db.is_deactivated(user_id) } /// Check if a user is an admin // Allowed because this function uses `services()` #[allow(clippy::unused_self)] pub(crate) fn is_admin(&self, user_id: &UserId) -> Result { services().admin.get_admin_room()?.map_or(Ok(false), |admin_room_id| { services().rooms.state_cache.is_joined(user_id, &admin_room_id) }) } /// Create a new user account on this homeserver. pub(crate) fn create( &self, user_id: &UserId, password: Option<&str>, ) -> Result<()> { self.db.set_password(user_id, password)?; Ok(()) } /// Returns the number of users registered on this server. pub(crate) fn count(&self) -> Result { self.db.count() } /// Find out which user an access token belongs to. pub(crate) fn find_from_token( &self, token: &str, ) -> Result> { self.db.find_from_token(token) } /// Returns an iterator over all users on this homeserver. pub(crate) fn iter( &self, ) -> impl Iterator> + '_ { self.db.iter() } /// Returns a list of local users as list of usernames. /// /// A user account is considered `local` if the length of it's password is /// greater then zero. pub(crate) fn list_local_users(&self) -> Result> { self.db.list_local_users() } /// Returns the password hash for the given user. pub(crate) fn password_hash( &self, user_id: &UserId, ) -> Result> { self.db.password_hash(user_id) } /// Hash and set the user's password to the Argon2 hash pub(crate) fn set_password( &self, user_id: &UserId, password: Option<&str>, ) -> Result<()> { self.db.set_password(user_id, password) } /// Returns the displayname of a user on this homeserver. pub(crate) fn displayname( &self, user_id: &UserId, ) -> Result> { self.db.displayname(user_id) } /// Sets a new displayname or removes it if displayname is None. You still /// need to nofify all rooms of this change. pub(crate) fn set_displayname( &self, user_id: &UserId, displayname: Option, ) -> Result<()> { self.db.set_displayname(user_id, displayname) } /// Get the `avatar_url` of a user. pub(crate) fn avatar_url( &self, user_id: &UserId, ) -> Result> { self.db.avatar_url(user_id) } /// Sets a new `avatar_url` or removes it if `avatar_url` is `None`. pub(crate) fn set_avatar_url( &self, user_id: &UserId, avatar_url: Option, ) -> Result<()> { self.db.set_avatar_url(user_id, avatar_url) } /// Get the `blurhash` of a user. pub(crate) fn blurhash(&self, user_id: &UserId) -> Result> { self.db.blurhash(user_id) } /// Sets a new `avatar_url` or removes it if `avatar_url` is `None`. pub(crate) fn set_blurhash( &self, user_id: &UserId, blurhash: Option, ) -> Result<()> { self.db.set_blurhash(user_id, blurhash) } /// Adds a new device to a user. pub(crate) fn create_device( &self, user_id: &UserId, device_id: &DeviceId, token: &str, initial_device_display_name: Option, ) -> Result<()> { self.db.create_device( user_id, device_id, token, initial_device_display_name, ) } /// Removes a device from a user. pub(crate) fn remove_device( &self, user_id: &UserId, device_id: &DeviceId, ) -> Result<()> { self.db.remove_device(user_id, device_id) } /// Returns an iterator over all device ids of this user. pub(crate) fn all_device_ids<'a>( &'a self, user_id: &UserId, ) -> impl Iterator> + 'a { self.db.all_device_ids(user_id) } /// Replaces the access token of one device. pub(crate) fn set_token( &self, user_id: &UserId, device_id: &DeviceId, token: &str, ) -> Result<()> { self.db.set_token(user_id, device_id, token) } pub(crate) fn add_one_time_key( &self, user_id: &UserId, device_id: &DeviceId, one_time_key_key: &OwnedKeyId, one_time_key_value: &Raw, ) -> Result<()> { self.db.add_one_time_key( user_id, device_id, one_time_key_key, one_time_key_value, ) } pub(crate) fn take_one_time_key( &self, user_id: &UserId, device_id: &DeviceId, key_algorithm: &OneTimeKeyAlgorithm, ) -> Result)>> { self.db.take_one_time_key(user_id, device_id, key_algorithm) } pub(crate) fn count_one_time_keys( &self, user_id: &UserId, device_id: &DeviceId, ) -> Result> { self.db.count_one_time_keys(user_id, device_id) } pub(crate) fn add_device_keys( &self, user_id: &UserId, device_id: &DeviceId, device_keys: &Raw, ) -> Result<()> { self.db.add_device_keys(user_id, device_id, device_keys) } pub(crate) fn add_cross_signing_keys( &self, user_id: &UserId, master_key: &Raw, self_signing_key: &Option>, user_signing_key: &Option>, notify: bool, ) -> Result<()> { self.db.add_cross_signing_keys( user_id, master_key, self_signing_key, user_signing_key, notify, ) } pub(crate) fn sign_key( &self, target_id: &UserId, key_id: &str, signature: (String, String), sender_id: &UserId, ) -> Result<()> { self.db.sign_key(target_id, key_id, signature, sender_id) } pub(crate) fn keys_changed<'a>( &'a self, user_or_room_id: &str, from: u64, to: Option, ) -> impl Iterator> + 'a { self.db.keys_changed(user_or_room_id, from, to) } pub(crate) fn mark_device_key_update( &self, user_id: &UserId, ) -> Result<()> { self.db.mark_device_key_update(user_id) } pub(crate) fn get_device_keys( &self, user_id: &UserId, device_id: &DeviceId, ) -> Result>> { self.db.get_device_keys(user_id, device_id) } pub(crate) fn parse_master_key( &self, user_id: &UserId, master_key: &Raw, ) -> Result<(Vec, CrossSigningKey)> { self.db.parse_master_key(user_id, master_key) } pub(crate) fn get_key( &self, key: &[u8], sender_user: Option<&UserId>, user_id: &UserId, allowed_signatures: &dyn Fn(&UserId) -> bool, ) -> Result>> { self.db.get_key(key, sender_user, user_id, allowed_signatures) } pub(crate) fn get_master_key( &self, sender_user: Option<&UserId>, user_id: &UserId, allowed_signatures: &dyn Fn(&UserId) -> bool, ) -> Result>> { self.db.get_master_key(sender_user, user_id, allowed_signatures) } pub(crate) fn get_self_signing_key( &self, sender_user: Option<&UserId>, user_id: &UserId, allowed_signatures: &dyn Fn(&UserId) -> bool, ) -> Result>> { self.db.get_self_signing_key(sender_user, user_id, allowed_signatures) } pub(crate) fn get_user_signing_key( &self, user_id: &UserId, ) -> Result>> { self.db.get_user_signing_key(user_id) } pub(crate) fn add_to_device_event( &self, sender: &UserId, target_user_id: &UserId, target_device_id: &DeviceId, event_type: &str, content: serde_json::Value, ) -> Result<()> { self.db.add_to_device_event( sender, target_user_id, target_device_id, event_type, content, ) } pub(crate) fn get_to_device_events( &self, user_id: &UserId, device_id: &DeviceId, ) -> Result>> { self.db.get_to_device_events(user_id, device_id) } pub(crate) fn remove_to_device_events( &self, user_id: &UserId, device_id: &DeviceId, until: u64, ) -> Result<()> { self.db.remove_to_device_events(user_id, device_id, until) } pub(crate) fn update_device_metadata( &self, user_id: &UserId, device_id: &DeviceId, device: &Device, ) -> Result<()> { self.db.update_device_metadata(user_id, device_id, device) } /// Get device metadata. pub(crate) fn get_device_metadata( &self, user_id: &UserId, device_id: &DeviceId, ) -> Result> { self.db.get_device_metadata(user_id, device_id) } pub(crate) fn get_devicelist_version( &self, user_id: &UserId, ) -> Result> { self.db.get_devicelist_version(user_id) } pub(crate) fn all_devices_metadata<'a>( &'a self, user_id: &UserId, ) -> impl Iterator> + 'a { self.db.all_devices_metadata(user_id) } /// Deactivate account pub(crate) fn deactivate_account(&self, user_id: &UserId) -> Result<()> { // Remove all associated devices for device_id in self.all_device_ids(user_id) { self.remove_device(user_id, &device_id?)?; } // Set the password to "" to indicate a deactivated account. Hashes will // never result in an empty string, so the user will not be able // to log in again. Systems like changing the password without // logging in should check if the account is deactivated. self.db.set_password(user_id, None)?; // TODO: Unhook 3PID Ok(()) } /// Creates a new sync filter. Returns the filter id. pub(crate) fn create_filter( &self, user_id: &UserId, filter: &FilterDefinition, ) -> Result { self.db.create_filter(user_id, filter) } pub(crate) fn get_filter( &self, user_id: &UserId, filter_id: &str, ) -> Result> { self.db.get_filter(user_id, filter_id) } } /// Ensure that a user only sees signatures from themselves and the target user pub(crate) fn clean_signatures bool>( cross_signing_key: &mut serde_json::Value, sender_user: Option<&UserId>, user_id: &UserId, allowed_signatures: F, ) -> Result<(), Error> { if let Some(signatures) = cross_signing_key.get_mut("signatures").and_then(|v| v.as_object_mut()) { // Don't allocate for the full size of the current signatures, but // require at most one resize if nothing is dropped let new_capacity = signatures.len() / 2; for (user, signature) in mem::replace( signatures, serde_json::Map::with_capacity(new_capacity), ) { let sid = <&UserId>::try_from(user.as_str()).map_err(|_| { Error::bad_database("Invalid user ID in database.") })?; if sender_user == Some(user_id) || sid == user_id || allowed_signatures(sid) { signatures.insert(user, signature); } } } Ok(()) }