From 4844c6ae3e63d06f45761cf37d905c50ccb0ec4c Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 03:54:10 +0100 Subject: [PATCH] style: Document & refactor rooms/state_cache service --- src/service/rooms/state_cache/mod.rs | 775 +++++++++++------------- src/service/rooms/state_cache/update.rs | 623 ++++++++++--------- src/service/rooms/state_cache/via.rs | 139 +++-- 3 files changed, 745 insertions(+), 792 deletions(-) diff --git a/src/service/rooms/state_cache/mod.rs b/src/service/rooms/state_cache/mod.rs index c54970638..46ef9a769 100644 --- a/src/service/rooms/state_cache/mod.rs +++ b/src/service/rooms/state_cache/mod.rs @@ -4,10 +4,9 @@ use std::{collections::HashMap, sync::Arc}; use conduwuit::{ - Pdu, Result, SyncRwLock, implement, + Pdu, Result, SyncRwLock, result::LogErr, utils::{ReadyExt, stream::TryIgnore}, - warn, }; use database::{Deserialized, Ignore, Interfix, Map}; use futures::{Stream, StreamExt, future::join5, pin_mut}; @@ -93,448 +92,414 @@ fn build(args: crate::Args<'_>) -> Result> { fn name(&self) -> &str { crate::service::make_name(std::module_path!()) } } -#[implement(Service)] -#[tracing::instrument(level = "trace", skip_all)] -pub async fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> bool { - if let Some(cached) = self - .appservice_in_room_cache - .read() - .get(room_id) - .and_then(|map| map.get(&appservice.registration.id)) - .copied() - { - return cached; +impl Service { + /// Checks if an appservice (or any user in its claimed namespace) is in the + /// target room. + pub async fn appservice_in_room( + &self, + room_id: &RoomId, + appservice: &RegistrationInfo, + ) -> bool { + if let Some(cached) = self + .appservice_in_room_cache + .read() + .get(room_id) + .and_then(|map| map.get(&appservice.registration.id)) + .copied() + { + return cached; + } + + let appservice_user_id = UserId::parse_with_server_name( + appservice.registration.sender_localpart.as_str(), + self.services.globals.server_name(), + ); + + let Ok(bridge_user_id) = appservice_user_id.log_err() else { + return false; + }; + + let (as_user_joined, any_as_namespace_user_joined) = tokio::join!( + self.is_joined(&bridge_user_id, room_id), + self.room_members(room_id) + .ready_any(|user_id| appservice.users.is_match(user_id.as_str())) + ); + let in_room = as_user_joined || any_as_namespace_user_joined; + + self.appservice_in_room_cache + .write() + .entry(room_id.into()) + .or_default() + .insert(appservice.registration.id.clone(), in_room); + + in_room } - let bridge_user_id = UserId::parse_with_server_name( - appservice.registration.sender_localpart.as_str(), - self.services.globals.server_name(), - ); + /// Gets the size of the AS joined room cache and its capacity. + pub fn get_appservice_in_room_cache_usage(&self) -> (usize, usize) { + let cache = self.appservice_in_room_cache.read(); - let Ok(bridge_user_id) = bridge_user_id.log_err() else { - return false; - }; + (cache.len(), cache.capacity()) + } - let in_room = self.is_joined(&bridge_user_id, room_id).await - || self - .room_members(room_id) - .ready_any(|user_id| appservice.users.is_match(user_id.as_str())) - .await; + /// Clears the internal cache of appservice joined rooms. + pub fn clear_appservice_in_room_cache(&self) { + self.appservice_in_room_cache.write().clear(); + } - self.appservice_in_room_cache - .write() - .entry(room_id.into()) - .or_default() - .insert(appservice.registration.id.clone(), in_room); + /// Returns an iterator of all servers participating in this room. + pub fn room_servers<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + let prefix = (room_id, Interfix); + self.db + .roomserverids + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, server): (Ignore, OwnedServerName)| server) + } - in_room -} + /// Determines if a server name has at least one joined member in a room. + pub async fn server_in_room<'a>( + &'a self, + server: &'a ServerName, + room_id: &'a RoomId, + ) -> bool { + let key = (server, room_id); + self.db.serverroomids.qry(&key).await.is_ok() + } -#[implement(Service)] -pub fn get_appservice_in_room_cache_usage(&self) -> (usize, usize) { - let cache = self.appservice_in_room_cache.read(); + /// Returns an iterator of all rooms a server participates in (as far as we + /// know). + pub fn server_rooms<'a>( + &'a self, + server: &'a ServerName, + ) -> impl Stream + Send + 'a { + let prefix = (server, Interfix); + self.db + .serverroomids + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, room_id): (Ignore, OwnedRoomId)| room_id) + } - (cache.len(), cache.capacity()) -} + /// Returns true if server can see user by sharing at least one room. + pub async fn server_sees_user(&self, server: &ServerName, user_id: &UserId) -> bool { + self.server_rooms(server) + .any(async |room_id| self.is_joined(user_id, &room_id).await) + .await + } -#[implement(Service)] -#[tracing::instrument(level = "debug", skip_all)] -pub fn clear_appservice_in_room_cache(&self) { self.appservice_in_room_cache.write().clear(); } + /// Returns true if user_a and user_b share at least one room. + pub async fn user_sees_user(&self, user_a: &UserId, user_b: &UserId) -> bool { + let get_shared_rooms = self.get_shared_rooms(user_a, user_b); -/// Returns an iterator of all servers participating in this room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_servers<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - let prefix = (room_id, Interfix); - self.db - .roomserverids - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, server): (Ignore, OwnedServerName)| server) -} + pin_mut!(get_shared_rooms); + get_shared_rooms.next().await.is_some() + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn server_in_room<'a>(&'a self, server: &'a ServerName, room_id: &'a RoomId) -> bool { - let key = (server, room_id); - self.db.serverroomids.qry(&key).await.is_ok() -} + /// Lists the rooms common between two users. + pub fn get_shared_rooms<'a>( + &'a self, + user_a: &'a UserId, + user_b: &'a UserId, + ) -> impl Stream + Send + 'a { + use conduwuit::utils::set; -/// Returns an iterator of all rooms a server participates in (as far as we -/// know). -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn server_rooms<'a>( - &'a self, - server: &'a ServerName, -) -> impl Stream + Send + 'a { - let prefix = (server, Interfix); - self.db - .serverroomids - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, room_id): (Ignore, OwnedRoomId)| room_id) -} + let a = self.rooms_joined(user_a); + let b = self.rooms_joined(user_b); + set::intersection_sorted_stream2(a, b) + } -/// Returns true if server can see user by sharing at least one room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn server_sees_user(&self, server: &ServerName, user_id: &UserId) -> bool { - self.server_rooms(server) - .any(async |room_id| self.is_joined(user_id, &room_id).await) - .await -} + /// Returns an iterator of all joined members of a room. + pub fn room_members<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + let prefix = (room_id, Interfix); + self.db + .roomuserid_joined + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) + } -/// Returns true if user_a and user_b share at least one room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn user_sees_user(&self, user_a: &UserId, user_b: &UserId) -> bool { - let get_shared_rooms = self.get_shared_rooms(user_a, user_b); + /// Returns the number of users which are currently in a room. + pub async fn room_joined_count(&self, room_id: &RoomId) -> Result { + self.db.roomid_joinedcount.get(room_id).await.deserialized() + } - pin_mut!(get_shared_rooms); - get_shared_rooms.next().await.is_some() -} + /// Returns an iterator of all our local users in the room, even if they're + /// deactivated. + pub fn local_users_in_room<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + self.room_members(room_id) + .ready_filter(|user| self.services.globals.user_is_local(user)) + } -/// List the rooms common between two users -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn get_shared_rooms<'a>( - &'a self, - user_a: &'a UserId, - user_b: &'a UserId, -) -> impl Stream + Send + 'a { - use conduwuit::utils::set; + /// Returns an iterator of all our local joined users in a room who are + /// active (not deactivated) + pub fn active_local_users_in_room<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + self.local_users_in_room(room_id) + .filter_map(async |user_id| { + if self.services.users.is_active(&user_id).await { + Some(user_id) + } else { + None + } + }) + } - let a = self.rooms_joined(user_a); - let b = self.rooms_joined(user_b); - set::intersection_sorted_stream2(a, b) -} + /// Returns the number of users which are currently invited to a room + pub async fn room_invited_count(&self, room_id: &RoomId) -> Result { + self.db + .roomid_invitedcount + .get(room_id) + .await + .deserialized() + } -/// Returns an iterator of all joined members of a room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_members<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - let prefix = (room_id, Interfix); - self.db - .roomuserid_joined - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) -} + /// Returns an iterator over all User IDs who ever joined a room, even if + /// they are no longer joined. + pub fn room_useroncejoined<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + let prefix = (room_id, Interfix); + self.db + .roomuseroncejoinedids + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) + } -/// Returns the number of users which are currently in a room -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn room_joined_count(&self, room_id: &RoomId) -> Result { - self.db.roomid_joinedcount.get(room_id).await.deserialized() -} + /// Returns an iterator over all invited members of a room. + pub fn room_members_invited<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + let prefix = (room_id, Interfix); + self.db + .roomuserid_invitecount + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -/// Returns an iterator of all our local users in the room, even if they're -/// deactivated -pub fn local_users_in_room<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - self.room_members(room_id) - .ready_filter(|user| self.services.globals.user_is_local(user)) -} + /// Returns an iterator over all knocking members of a room. + pub fn room_members_knocked<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + let prefix = (room_id, Interfix); + self.db + .roomuserid_knockedcount + .keys_prefix(&prefix) + .ignore_err() + .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) + } -/// Returns an iterator of all our local joined users in a room who are -/// active (not deactivated) -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub fn active_local_users_in_room<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - self.local_users_in_room(room_id) - .filter_map(async |user_id| { - if self.services.users.is_active(&user_id).await { - Some(user_id) - } else { - None - } - }) -} + /// Gets the invite count value. + pub async fn get_invite_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { + let key = (room_id, user_id); + self.db + .roomuserid_invitecount + .qry(&key) + .await + .deserialized() + } -/// Returns the number of users which are currently invited to a room -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn room_invited_count(&self, room_id: &RoomId) -> Result { - self.db - .roomid_invitedcount - .get(room_id) - .await - .deserialized() -} + /// Gets the knock count value. + pub async fn get_knock_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { + let key = (room_id, user_id); + self.db + .roomuserid_knockedcount + .qry(&key) + .await + .deserialized() + } -/// Returns an iterator over all User IDs who ever joined a room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_useroncejoined<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - let prefix = (room_id, Interfix); - self.db - .roomuseroncejoinedids - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) -} + /// Gets the left count value. + pub async fn get_left_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { + let key = (room_id, user_id); + self.db.roomuserid_leftcount.qry(&key).await.deserialized() + } -/// Returns an iterator over all invited members of a room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_members_invited<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - let prefix = (room_id, Interfix); - self.db - .roomuserid_invitecount - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) -} + /// Returns an iterator over all rooms this user joined. + pub fn rooms_joined<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + Send + 'a { + self.db + .userroomid_joined + .keys_raw_prefix(user_id) + .ignore_err() + .map(|(_, room_id): (Ignore, OwnedRoomId)| room_id) + } -/// Returns an iterator over all knocked members of a room. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_members_knocked<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - let prefix = (room_id, Interfix); - self.db - .roomuserid_knockedcount - .keys_prefix(&prefix) - .ignore_err() - .map(|(_, user_id): (Ignore, OwnedUserId)| user_id) -} + /// Returns an iterator over all rooms a user was invited to. + pub fn rooms_invited<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + Send + 'a { + type KeyVal = (Key, Raw>); + type Key = (OwnedUserId, OwnedRoomId); -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn get_invite_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { - let key = (room_id, user_id); - self.db - .roomuserid_invitecount - .qry(&key) - .await - .deserialized() -} + let prefix = (user_id, Interfix); + self.db + .userroomid_invitestate + .stream_prefix(&prefix) + .ignore_err() + .map(|((_, room_id), state): KeyVal| (room_id, state)) + .map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?))) + .ignore_err() + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn get_knock_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { - let key = (room_id, user_id); - self.db - .roomuserid_knockedcount - .qry(&key) - .await - .deserialized() -} + /// Returns an iterator over all rooms a user is currently knocking on. + pub fn rooms_knocked<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream + Send + 'a { + type KeyVal = (Key, Raw>); + type Key = (OwnedUserId, OwnedRoomId); -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn get_left_count(&self, room_id: &RoomId, user_id: &UserId) -> Result { - let key = (room_id, user_id); - self.db.roomuserid_leftcount.qry(&key).await.deserialized() -} + let prefix = (user_id, Interfix); + self.db + .userroomid_knockedstate + .stream_prefix(&prefix) + .ignore_err() + .map(|((_, room_id), state): KeyVal| (room_id, state)) + .map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?))) + .ignore_err() + } -/// Returns an iterator over all rooms this user joined. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn rooms_joined<'a>( - &'a self, - user_id: &'a UserId, -) -> impl Stream + Send + 'a { - self.db - .userroomid_joined - .keys_raw_prefix(user_id) - .ignore_err() - .map(|(_, room_id): (Ignore, OwnedRoomId)| room_id) -} + /// Returns the invite state for a user in a room, if they are invited. + pub async fn invite_state( + &self, + user_id: &UserId, + room_id: &RoomId, + ) -> Result>> { + let key = (user_id, room_id); + self.db + .userroomid_invitestate + .qry(&key) + .await + .deserialized() + .and_then(|val: Raw>| { + val.deserialize_as_unchecked().map_err(Into::into) + }) + } -/// Returns an iterator over all rooms a user was invited to. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn rooms_invited<'a>( - &'a self, - user_id: &'a UserId, -) -> impl Stream + Send + 'a { - type KeyVal = (Key, Raw>); - type Key = (OwnedUserId, OwnedRoomId); + /// Returns the knock state for a user in a room, if they are knocking. + pub async fn knock_state( + &self, + user_id: &UserId, + room_id: &RoomId, + ) -> Result>> { + let key = (user_id, room_id); + self.db + .userroomid_knockedstate + .qry(&key) + .await + .deserialized() + .and_then(|val: Raw>| { + val.deserialize_as_unchecked().map_err(Into::into) + }) + } - let prefix = (user_id, Interfix); - self.db - .userroomid_invitestate - .stream_prefix(&prefix) - .ignore_err() - .map(|((_, room_id), state): KeyVal| (room_id, state)) - .map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?))) - .ignore_err() -} + /// Returns the left state for a user in a room, if they have left. + pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Result> { + let key = (user_id, room_id); + self.db.userroomid_leftstate.qry(&key).await.deserialized() + } -/// Returns an iterator over all rooms a user is currently knocking. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub fn rooms_knocked<'a>( - &'a self, - user_id: &'a UserId, -) -> impl Stream + Send + 'a { - type KeyVal = (Key, Raw>); - type Key = (OwnedUserId, OwnedRoomId); + /// Returns an iterator over all rooms a user left. + pub fn rooms_left<'a>( + &'a self, + user_id: &'a UserId, + ) -> impl Stream)> + Send + 'a { + type KeyVal = (Key, Raw>); + type Key = (OwnedUserId, OwnedRoomId); - let prefix = (user_id, Interfix); - self.db - .userroomid_knockedstate - .stream_prefix(&prefix) - .ignore_err() - .map(|((_, room_id), state): KeyVal| (room_id, state)) - .map(|(room_id, state)| Ok((room_id, state.deserialize_as_unchecked()?))) - .ignore_err() -} + let prefix = (user_id, Interfix); + self.db + .userroomid_leftstate + .stream_prefix(&prefix) + .ignore_err() + .map(|((_, room_id), state): KeyVal| (room_id, state)) + .map(|(room_id, state)| Ok((room_id, state.deserialize()?))) + .ignore_err() + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn invite_state( - &self, - user_id: &UserId, - room_id: &RoomId, -) -> Result>> { - let key = (user_id, room_id); - self.db - .userroomid_invitestate - .qry(&key) - .await - .deserialized() - .and_then(|val: Raw>| { - val.deserialize_as_unchecked().map_err(Into::into) - }) -} + /// Determines a user's membership in a target room. If the user has never + /// been a member of the target room at any point in time, None is returned. + pub async fn user_membership( + &self, + user_id: &UserId, + room_id: &RoomId, + ) -> Option { + let states = join5( + self.is_joined(user_id, room_id), + self.is_left(user_id, room_id), + self.is_knocked(user_id, room_id), + self.is_invited(user_id, room_id), + self.once_joined(user_id, room_id), + ) + .await; -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn knock_state( - &self, - user_id: &UserId, - room_id: &RoomId, -) -> Result>> { - let key = (user_id, room_id); - self.db - .userroomid_knockedstate - .qry(&key) - .await - .deserialized() - .and_then(|val: Raw>| { - val.deserialize_as_unchecked().map_err(Into::into) - }) -} + match states { + | (true, ..) => Some(MembershipState::Join), + | (_, true, ..) => Some(MembershipState::Leave), + | (_, _, true, ..) => Some(MembershipState::Knock), + | (_, _, _, true, ..) => Some(MembershipState::Invite), + | (false, false, false, false, true) => Some(MembershipState::Ban), + | _ => None, + } + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Result> { - let key = (user_id, room_id); - self.db.userroomid_leftstate.qry(&key).await.deserialized() -} + /// Checks if a user has joined at least once before. + pub async fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> bool { + let key = (user_id, room_id); + self.db.roomuseroncejoinedids.qry(&key).await.is_ok() + } -/// Returns an iterator over all rooms a user left. -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn rooms_left<'a>( - &'a self, - user_id: &'a UserId, -) -> impl Stream)> + Send + 'a { - type KeyVal = (Key, Raw>); - type Key = (OwnedUserId, OwnedRoomId); + /// Checks if a user is currently joined to a room. + pub async fn is_joined<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool { + let key = (user_id, room_id); + self.db.userroomid_joined.qry(&key).await.is_ok() + } - let prefix = (user_id, Interfix); - self.db - .userroomid_leftstate - .stream_prefix(&prefix) - .ignore_err() - .map(|((_, room_id), state): KeyVal| (room_id, state)) - .map(|(room_id, state)| Ok((room_id, state.deserialize()?))) - .ignore_err() -} + /// Checks if a user is currently knocking on a room. + pub async fn is_knocked<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool { + let key = (user_id, room_id); + self.db.userroomid_knockedstate.qry(&key).await.is_ok() + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn user_membership( - &self, - user_id: &UserId, - room_id: &RoomId, -) -> Option { - let states = join5( - self.is_joined(user_id, room_id), - self.is_left(user_id, room_id), - self.is_knocked(user_id, room_id), - self.is_invited(user_id, room_id), - self.once_joined(user_id, room_id), - ) - .await; + /// Checks if a user is currently invited to a room. + pub async fn is_invited(&self, user_id: &UserId, room_id: &RoomId) -> bool { + let key = (user_id, room_id); + self.db.userroomid_invitestate.qry(&key).await.is_ok() + } - match states { - | (true, ..) => Some(MembershipState::Join), - | (_, true, ..) => Some(MembershipState::Leave), - | (_, _, true, ..) => Some(MembershipState::Knock), - | (_, _, _, true, ..) => Some(MembershipState::Invite), - | (false, false, false, false, true) => Some(MembershipState::Ban), - | _ => None, + /// Checks if a user has left the room. + pub async fn is_left(&self, user_id: &UserId, room_id: &RoomId) -> bool { + let key = (user_id, room_id); + self.db.userroomid_leftstate.qry(&key).await.is_ok() + } + + /// Fetches the sender who invited a target user. + pub async fn invite_sender(&self, user_id: &UserId, room_id: &RoomId) -> Result { + let key = (user_id, room_id); + self.db + .userroomid_invitesender + .qry(&key) + .await + .deserialized() } } - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> bool { - let key = (user_id, room_id); - self.db.roomuseroncejoinedids.qry(&key).await.is_ok() -} - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn is_joined<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool { - let key = (user_id, room_id); - self.db.userroomid_joined.qry(&key).await.is_ok() -} - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn is_knocked<'a>(&'a self, user_id: &'a UserId, room_id: &'a RoomId) -> bool { - let key = (user_id, room_id); - self.db.userroomid_knockedstate.qry(&key).await.is_ok() -} - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn is_invited(&self, user_id: &UserId, room_id: &RoomId) -> bool { - let key = (user_id, room_id); - self.db.userroomid_invitestate.qry(&key).await.is_ok() -} - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn is_left(&self, user_id: &UserId, room_id: &RoomId) -> bool { - let key = (user_id, room_id); - self.db.userroomid_leftstate.qry(&key).await.is_ok() -} - -#[implement(Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn invite_sender(&self, user_id: &UserId, room_id: &RoomId) -> Result { - let key = (user_id, room_id); - self.db - .userroomid_invitesender - .qry(&key) - .await - .deserialized() -} diff --git a/src/service/rooms/state_cache/update.rs b/src/service/rooms/state_cache/update.rs index 0103713f4..20c6a700f 100644 --- a/src/service/rooms/state_cache/update.rs +++ b/src/service/rooms/state_cache/update.rs @@ -1,6 +1,6 @@ use std::collections::HashSet; -use conduwuit::{Err, Event, Pdu, Result, implement, is_not_empty, utils::ReadyExt, warn}; +use conduwuit::{Err, Event, Pdu, Result, is_not_empty, utils::ReadyExt}; use database::{Json, serialize_key}; use futures::StreamExt; use ruma::{ @@ -19,369 +19,360 @@ }; use ruminuwuity::invite_permission_config::FilterLevel; -/// Update current membership data. -#[implement(super::Service)] -#[tracing::instrument( - level = "debug", - skip_all, - fields( - %room_id, - %user_id, - ?pdu, - ), - )] -#[allow(clippy::too_many_arguments)] -pub async fn update_membership( - &self, - room_id: &RoomId, - user_id: &UserId, - pdu: &Pdu, - update_joined_count: bool, -) -> Result { - let membership = pdu.get_content::()?; +impl super::Service { + /// Update current membership data. + pub async fn update_membership( + &self, + room_id: &RoomId, + user_id: &UserId, + pdu: &Pdu, + update_joined_count: bool, + ) -> Result { + let membership = pdu.get_content::()?; - // Keep track what remote users exist by adding them as "deactivated" users - // - // TODO: use futures to update remote profiles without blocking the membership - // update - #[allow(clippy::collapsible_if)] - if !self.services.globals.user_is_local(user_id) { - if !self.services.users.exists(user_id).await { + // Keep track what remote users exist by adding them as "deactivated" users + // + // TODO: use futures to update remote profiles without blocking the membership + // update + #[allow(clippy::collapsible_if)] + if !self.services.globals.user_is_local(user_id) + && !self.services.users.exists(user_id).await + { self.services.users.create(user_id, None).await?; } - } - match &membership.membership { - | MembershipState::Join => { - // Check if the user never joined this room - if !self.once_joined(user_id, room_id).await { - // Add the user ID to the join list then - self.mark_as_once_joined(user_id, room_id); + match &membership.membership { + | MembershipState::Join => { + // Check if the user never joined this room + if !self.once_joined(user_id, room_id).await { + // Add the user ID to the join list then + self.mark_as_once_joined(user_id, room_id); - // Check if the room has a predecessor - if let Ok(Some(predecessor)) = self - .services - .state_accessor - .room_state_get_content(room_id, &StateEventType::RoomCreate, "") - .await - .map(|content: RoomCreateEventContent| content.predecessor) - { - // Copy old tags to new room - if let Ok(tag_event) = self + // Check if the room has a predecessor + if let Ok(Some(predecessor)) = self .services - .account_data - .get_room(&predecessor.room_id, user_id, RoomAccountDataEventType::Tag) + .state_accessor + .room_state_get_content(room_id, &StateEventType::RoomCreate, "") .await + .map(|content: RoomCreateEventContent| content.predecessor) { - self.services + // Copy old tags to new room + if let Ok(tag_event) = self + .services .account_data - .update( - Some(room_id), + .get_room( + &predecessor.room_id, user_id, RoomAccountDataEventType::Tag, - &tag_event, ) .await - .ok(); - } - - // Copy direct chat flag - if let Ok(mut direct_event) = self - .services - .account_data - .get_global::(user_id, GlobalAccountDataEventType::Direct) - .await - { - let mut room_ids_updated = false; - for room_ids in direct_event.content.0.values_mut() { - if room_ids.iter().any(|r| r == &predecessor.room_id) { - room_ids.push(room_id.to_owned()); - room_ids_updated = true; - } - } - - if room_ids_updated { + { self.services .account_data .update( - None, + Some(room_id), user_id, - GlobalAccountDataEventType::Direct.to_string().into(), - &serde_json::to_value(&direct_event) - .expect("to json always works"), + RoomAccountDataEventType::Tag, + &tag_event, ) - .await?; + .await + .ok(); + } + + // Copy direct chat flag + if let Ok(mut direct_event) = self + .services + .account_data + .get_global::( + user_id, + GlobalAccountDataEventType::Direct, + ) + .await + { + let mut room_ids_updated = false; + for room_ids in direct_event.content.0.values_mut() { + if room_ids.iter().any(|r| r == &predecessor.room_id) { + room_ids.push(room_id.to_owned()); + room_ids_updated = true; + } + } + + if room_ids_updated { + self.services + .account_data + .update( + None, + user_id, + GlobalAccountDataEventType::Direct.to_string().into(), + &serde_json::to_value(&direct_event) + .expect("to json always works"), + ) + .await?; + } } } } - } - self.mark_as_joined(user_id, room_id); - }, - | MembershipState::Invite => { - let last_state = self - .services - .state - .summary_stripped(pdu, room_id, user_id) - .await; + self.mark_as_joined(user_id, room_id); + }, + | MembershipState::Invite => { + let last_state = self + .services + .state + .summary_stripped(pdu, room_id, user_id) + .await; - self.mark_as_invited(user_id, room_id, pdu.sender(), last_state, None) - .await?; - }, - | MembershipState::Leave | MembershipState::Ban => { - self.mark_as_left(user_id, room_id, Some(pdu.clone())).await; - }, - | _ => {}, + self.mark_as_invited(user_id, room_id, pdu.sender(), last_state, None) + .await?; + }, + | MembershipState::Leave | MembershipState::Ban => { + self.mark_as_left(user_id, room_id, Some(pdu.clone())).await; + }, + | _ => {}, + } + + if update_joined_count { + self.update_joined_count(room_id).await; + } + + Ok(()) } - if update_joined_count { - self.update_joined_count(room_id).await; + /// Updates membership counts for the target room. + pub async fn update_joined_count(&self, room_id: &RoomId) { + let mut joinedcount = 0_u64; + let mut invitedcount = 0_u64; + let mut knockedcount = 0_u64; + let mut joined_servers = HashSet::new(); + + self.room_members(room_id) + .ready_for_each(|joined| { + joined_servers.insert(joined.server_name().to_owned()); + joinedcount = joinedcount.saturating_add(1); + }) + .await; + + invitedcount = invitedcount.saturating_add( + self.room_members_invited(room_id) + .count() + .await + .try_into() + .unwrap_or(0), + ); + + knockedcount = knockedcount.saturating_add( + self.room_members_knocked(room_id) + .count() + .await + .try_into() + .unwrap_or(0), + ); + + self.db.roomid_joinedcount.raw_put(room_id, joinedcount); + self.db.roomid_invitedcount.raw_put(room_id, invitedcount); + self.db + .roomuserid_knockedcount + .raw_put(room_id, knockedcount); + + self.room_servers(room_id) + .ready_for_each(|old_joined_server| { + if joined_servers.remove(&old_joined_server) { + return; + } + + // Server not in room anymore + let roomserver_id = (room_id, old_joined_server.clone()); + let serverroom_id = (old_joined_server, room_id); + + self.db.roomserverids.del(roomserver_id); + self.db.serverroomids.del(serverroom_id); + }) + .await; + + // Now only new servers are in joined_servers anymore + for server in &joined_servers { + let roomserver_id = (room_id, server); + let serverroom_id = (server, room_id); + + self.db.roomserverids.put_raw(roomserver_id, []); + self.db.serverroomids.put_raw(serverroom_id, []); + } + + self.appservice_in_room_cache.write().remove(room_id); } - Ok(()) -} + /// Direct DB function to directly mark a user as joined. It is not + /// recommended to use this directly. You most likely should use + /// `update_membership` instead + pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) { + let userroom_id = (user_id, room_id); + let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); -#[implement(super::Service)] -#[tracing::instrument(level = "debug", skip(self))] -pub async fn update_joined_count(&self, room_id: &RoomId) { - let mut joinedcount = 0_u64; - let mut invitedcount = 0_u64; - let mut knockedcount = 0_u64; - let mut joined_servers = HashSet::new(); + let roomuser_id = (room_id, user_id); + let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); - self.room_members(room_id) - .ready_for_each(|joined| { - joined_servers.insert(joined.server_name().to_owned()); - joinedcount = joinedcount.saturating_add(1); - }) - .await; + self.db.userroomid_joined.insert(&userroom_id, []); + self.db.roomuserid_joined.insert(&roomuser_id, []); - invitedcount = invitedcount.saturating_add( - self.room_members_invited(room_id) - .count() - .await - .try_into() - .unwrap_or(0), - ); + self.db.userroomid_invitestate.remove(&userroom_id); + self.db.roomuserid_invitecount.remove(&roomuser_id); + self.db.userroomid_invitesender.remove(&userroom_id); - knockedcount = knockedcount.saturating_add( - self.room_members_knocked(room_id) - .count() - .await - .try_into() - .unwrap_or(0), - ); + self.db.userroomid_leftstate.remove(&userroom_id); + self.db.roomuserid_leftcount.remove(&roomuser_id); - self.db.roomid_joinedcount.raw_put(room_id, joinedcount); - self.db.roomid_invitedcount.raw_put(room_id, invitedcount); - self.db - .roomuserid_knockedcount - .raw_put(room_id, knockedcount); + self.db.userroomid_knockedstate.remove(&userroom_id); + self.db.roomuserid_knockedcount.remove(&roomuser_id); - self.room_servers(room_id) - .ready_for_each(|old_joined_server| { - if joined_servers.remove(&old_joined_server) { - return; - } - - // Server not in room anymore - let roomserver_id = (room_id, old_joined_server.clone()); - let serverroom_id = (old_joined_server, room_id); - - self.db.roomserverids.del(roomserver_id); - self.db.serverroomids.del(serverroom_id); - }) - .await; - - // Now only new servers are in joined_servers anymore - for server in &joined_servers { - let roomserver_id = (room_id, server); - let serverroom_id = (server, room_id); - - self.db.roomserverids.put_raw(roomserver_id, []); - self.db.serverroomids.put_raw(serverroom_id, []); + self.db.roomid_inviteviaservers.remove(room_id); } - self.appservice_in_room_cache.write().remove(room_id); -} + /// Mark a user as having left a room. + /// + /// `leave_pdu` represents the m.room.member event which the user sent to + /// leave the room. If this is None, no event was actually sent, but we + /// must still behave as if the user is no longer in the room. This may + /// occur, for example, if the room being left has been server-banned by an + /// administrator. + pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option) { + let userroom_id = (user_id, room_id); + let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); -/// Direct DB function to directly mark a user as joined. It is not -/// recommended to use this directly. You most likely should use -/// `update_membership` instead -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) { - let userroom_id = (user_id, room_id); - let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); + let roomuser_id = (room_id, user_id); + let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); - let roomuser_id = (room_id, user_id); - let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); + self.db + .userroomid_leftstate + .raw_put(&userroom_id, Json(leave_pdu)); + self.db + .roomuserid_leftcount + .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); - self.db.userroomid_joined.insert(&userroom_id, []); - self.db.roomuserid_joined.insert(&roomuser_id, []); + self.db.userroomid_joined.remove(&userroom_id); + self.db.roomuserid_joined.remove(&roomuser_id); - self.db.userroomid_invitestate.remove(&userroom_id); - self.db.roomuserid_invitecount.remove(&roomuser_id); - self.db.userroomid_invitesender.remove(&userroom_id); + self.db.userroomid_invitestate.remove(&userroom_id); + self.db.roomuserid_invitecount.remove(&roomuser_id); + self.db.userroomid_invitesender.remove(&userroom_id); - self.db.userroomid_leftstate.remove(&userroom_id); - self.db.roomuserid_leftcount.remove(&roomuser_id); + self.db.userroomid_knockedstate.remove(&userroom_id); + self.db.roomuserid_knockedcount.remove(&roomuser_id); - self.db.userroomid_knockedstate.remove(&userroom_id); - self.db.roomuserid_knockedcount.remove(&roomuser_id); + self.db.roomid_inviteviaservers.remove(room_id); - self.db.roomid_inviteviaservers.remove(room_id); -} - -/// Mark a user as having left a room. -/// -/// `leave_pdu` represents the m.room.member event which the user sent to leave -/// the room. If this is None, no event was actually sent, but we must still -/// behave as if the user is no longer in the room. This may occur, for example, -/// if the room being left has been server-banned by an administrator. -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option) { - let userroom_id = (user_id, room_id); - let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); - - let roomuser_id = (room_id, user_id); - let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); - - self.db - .userroomid_leftstate - .raw_put(&userroom_id, Json(leave_pdu)); - self.db - .roomuserid_leftcount - .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); - - self.db.userroomid_joined.remove(&userroom_id); - self.db.roomuserid_joined.remove(&roomuser_id); - - self.db.userroomid_invitestate.remove(&userroom_id); - self.db.roomuserid_invitecount.remove(&roomuser_id); - self.db.userroomid_invitesender.remove(&userroom_id); - - self.db.userroomid_knockedstate.remove(&userroom_id); - self.db.roomuserid_knockedcount.remove(&roomuser_id); - - self.db.roomid_inviteviaservers.remove(room_id); - - if self.services.globals.user_is_local(user_id) - && (self.services.config.forget_forced_upon_leave - || self.services.metadata.is_banned(room_id).await - || self.services.metadata.is_disabled(room_id).await) - { - self.forget(room_id, user_id); + if self.services.globals.user_is_local(user_id) + && (self.services.config.forget_forced_upon_leave + || self.services.metadata.is_banned(room_id).await + || self.services.metadata.is_disabled(room_id).await) + { + self.forget(room_id, user_id); + } } -} -/// Direct DB function to directly mark a user as knocked. It is not -/// recommended to use this directly. You most likely should use -/// `update_membership` instead -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn mark_as_knocked( - &self, - user_id: &UserId, - room_id: &RoomId, - knocked_state: Option>>, -) { - let userroom_id = (user_id, room_id); - let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); - - let roomuser_id = (room_id, user_id); - let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); - - self.db - .userroomid_knockedstate - .raw_put(&userroom_id, Json(knocked_state.unwrap_or_default())); - self.db - .roomuserid_knockedcount - .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); - - self.db.userroomid_joined.remove(&userroom_id); - self.db.roomuserid_joined.remove(&roomuser_id); - - self.db.userroomid_invitestate.remove(&userroom_id); - self.db.roomuserid_invitecount.remove(&roomuser_id); - self.db.userroomid_invitesender.remove(&userroom_id); - - self.db.userroomid_leftstate.remove(&userroom_id); - self.db.roomuserid_leftcount.remove(&roomuser_id); - - self.db.roomid_inviteviaservers.remove(room_id); -} - -/// Makes a user forget a room. -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn forget(&self, room_id: &RoomId, user_id: &UserId) { - let userroom_id = (user_id, room_id); - let roomuser_id = (room_id, user_id); - - self.db.userroomid_leftstate.del(userroom_id); - self.db.roomuserid_leftcount.del(roomuser_id); -} - -#[implement(super::Service)] -#[tracing::instrument(level = "debug", skip(self))] -fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) { - let key = (user_id, room_id); - self.db.roomuseroncejoinedids.put_raw(key, []); -} - -#[implement(super::Service)] -#[tracing::instrument(level = "debug", skip(self, last_state, invite_via))] -pub async fn mark_as_invited( - &self, - user_id: &UserId, - room_id: &RoomId, - sender_user: &UserId, - last_state: Vec, - invite_via: Option>, -) -> Result<()> { - // return an error for blocked invites. ignored invites aren't handled here - // since the recipient's membership should still be changed to `invite`. - // they're filtered out in the individual /sync handlers - if matches!( - self.services - .users - .invite_filter_level(sender_user, user_id) - .await, - FilterLevel::Block + /// Direct DB function to directly mark a user as knocked. It is not + /// recommended to use this directly. You most likely should use + /// `update_membership` instead + pub fn mark_as_knocked( + &self, + user_id: &UserId, + room_id: &RoomId, + knocked_state: Option>>, ) { - return Err!(Request(InviteBlocked("{user_id} has blocked invites from {sender_user}."))); + let userroom_id = (user_id, room_id); + let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); + + let roomuser_id = (room_id, user_id); + let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); + + self.db + .userroomid_knockedstate + .raw_put(&userroom_id, Json(knocked_state.unwrap_or_default())); + self.db + .roomuserid_knockedcount + .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); + + self.db.userroomid_joined.remove(&userroom_id); + self.db.roomuserid_joined.remove(&roomuser_id); + + self.db.userroomid_invitestate.remove(&userroom_id); + self.db.roomuserid_invitecount.remove(&roomuser_id); + self.db.userroomid_invitesender.remove(&userroom_id); + + self.db.userroomid_leftstate.remove(&userroom_id); + self.db.roomuserid_leftcount.remove(&roomuser_id); + + self.db.roomid_inviteviaservers.remove(room_id); } - let roomuser_id = (room_id, user_id); - let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); + /// Makes a user forget a room. + pub fn forget(&self, room_id: &RoomId, user_id: &UserId) { + let userroom_id = (user_id, room_id); + let roomuser_id = (room_id, user_id); - let userroom_id = (user_id, room_id); - let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); - - self.db - .userroomid_invitestate - .raw_put(&userroom_id, Json(last_state)); - self.db - .roomuserid_invitecount - .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); - self.db - .userroomid_invitesender - .insert(&userroom_id, sender_user); - - self.db.userroomid_joined.remove(&userroom_id); - self.db.roomuserid_joined.remove(&roomuser_id); - - self.db.userroomid_leftstate.remove(&userroom_id); - self.db.roomuserid_leftcount.remove(&roomuser_id); - - self.db.userroomid_knockedstate.remove(&userroom_id); - self.db.roomuserid_knockedcount.remove(&roomuser_id); - - if let Some(servers) = invite_via.filter(is_not_empty!()) { - self.add_servers_invite_via(room_id, servers).await; + self.db.userroomid_leftstate.del(userroom_id); + self.db.roomuserid_leftcount.del(roomuser_id); } - Ok(()) + /// Immutably a user as having joined a room. + fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) { + let key = (user_id, room_id); + self.db.roomuseroncejoinedids.put_raw(key, []); + } + + /// Marks a user as having been invited to a room, saving the relevant + /// invite state. + pub async fn mark_as_invited( + &self, + user_id: &UserId, + room_id: &RoomId, + sender_user: &UserId, + last_state: Vec, + invite_via: Option>, + ) -> Result<()> { + // return an error for blocked invites. ignored invites aren't handled here + // since the recipient's membership should still be changed to `invite`. + // they're filtered out in the individual /sync handlers + if matches!( + self.services + .users + .invite_filter_level(sender_user, user_id) + .await, + FilterLevel::Block + ) { + return Err!(Request(InviteBlocked( + "{user_id} has blocked invites from {sender_user}." + ))); + } + + let roomuser_id = (room_id, user_id); + let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id"); + + let userroom_id = (user_id, room_id); + let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id"); + + self.db + .userroomid_invitestate + .raw_put(&userroom_id, Json(last_state)); + self.db + .roomuserid_invitecount + .raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap()); + self.db + .userroomid_invitesender + .insert(&userroom_id, sender_user); + + self.db.userroomid_joined.remove(&userroom_id); + self.db.roomuserid_joined.remove(&roomuser_id); + + self.db.userroomid_leftstate.remove(&userroom_id); + self.db.roomuserid_leftcount.remove(&roomuser_id); + + self.db.userroomid_knockedstate.remove(&userroom_id); + self.db.roomuserid_knockedcount.remove(&roomuser_id); + + if let Some(servers) = invite_via.filter(is_not_empty!()) { + self.add_servers_invite_via(room_id, servers).await; + } + + Ok(()) + } } diff --git a/src/service/rooms/state_cache/via.rs b/src/service/rooms/state_cache/via.rs index 1ffc4770a..7494a9123 100644 --- a/src/service/rooms/state_cache/via.rs +++ b/src/service/rooms/state_cache/via.rs @@ -1,7 +1,6 @@ use conduwuit::{ - Result, implement, + Result, utils::{StreamTools, stream::TryIgnore}, - warn, }; use database::Ignore; use futures::{Stream, StreamExt, stream::iter}; @@ -12,80 +11,78 @@ int, }; -#[implement(super::Service)] -#[tracing::instrument(level = "debug", skip(self, servers))] -pub async fn add_servers_invite_via(&self, room_id: &RoomId, servers: Vec) { - let mut servers: Vec<_> = self - .servers_invite_via(room_id) - .chain(iter(servers.into_iter())) - .collect() - .await; +impl super::Service { + /// Adds N servers that are likely to be in the room in the future. + pub async fn add_servers_invite_via(&self, room_id: &RoomId, servers: Vec) { + let mut servers: Vec<_> = self + .servers_invite_via(room_id) + .chain(iter(servers)) + .collect() + .await; - servers.sort_unstable(); - servers.dedup(); + servers.sort_unstable(); + servers.dedup(); - let servers = servers - .iter() - .map(|server| server.as_bytes()) - .collect_vec() - .join(&[0xFF][..]); + let servers = servers + .iter() + .map(|server| server.as_bytes()) + .collect_vec() + .join(&[0xFF][..]); - self.db - .roomid_inviteviaservers - .insert(room_id.as_bytes(), &servers); -} - -/// Gets up to five servers that are likely to be in the room in the -/// distant future. -/// -/// See -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "trace")] -pub async fn servers_route_via(&self, room_id: &RoomId) -> Result> { - let most_powerful_user_server = self - .services - .state_accessor - .room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "") - .await - .map(|content: RoomPowerLevelsEventContent| { - content - .users - .iter() - .max_by_key(|(_, power)| *power) - .and_then(|x| (x.1 >= &int!(50)).then_some(x)) - .map(|(user, _power)| user.server_name().to_owned()) - }); - - let mut servers: Vec = self - .room_members(room_id) - .counts_by(|user| user.server_name().to_owned()) - .await - .into_iter() - .sorted_by_key(|(_, users)| *users) - .map(|(server, _)| server) - .rev() - .take(5) - .collect(); - - if let Ok(Some(server)) = most_powerful_user_server { - servers.insert(0, server); - servers.truncate(5); + self.db + .roomid_inviteviaservers + .insert(room_id.as_bytes(), &servers); } - Ok(servers) -} + /// Gets up to five servers that are likely to be in the room in the + /// distant future. + /// + /// See + pub async fn servers_route_via(&self, room_id: &RoomId) -> Result> { + let most_powerful_user_server = self + .services + .state_accessor + .room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "") + .await + .map(|content: RoomPowerLevelsEventContent| { + content + .users + .iter() + .max_by_key(|(_, power)| *power) + .and_then(|x| (x.1 >= &int!(50)).then_some(x)) + .map(|(user, _power)| user.server_name().to_owned()) + }); -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn servers_invite_via<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream + Send + 'a { - type KeyVal = (Ignore, Vec); + let mut servers: Vec = self + .room_members(room_id) + .counts_by(|user| user.server_name().to_owned()) + .await + .into_iter() + .sorted_by_key(|(_, users)| *users) + .map(|(server, _)| server) + .rev() + .take(5) + .collect(); - self.db - .roomid_inviteviaservers - .stream_raw_prefix(room_id) - .ignore_err() - .map(|(_, mut servers): KeyVal| servers.pop().expect("at least one server")) + if let Ok(Some(server)) = most_powerful_user_server { + servers.insert(0, server); + servers.truncate(5); + } + + Ok(servers) + } + + /// Fetches a list of servers that may be in the room as of an invite. + pub fn servers_invite_via<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream + Send + 'a { + type KeyVal = (Ignore, Vec); + + self.db + .roomid_inviteviaservers + .stream_raw_prefix(room_id) + .ignore_err() + .map(|(_, mut servers): KeyVal| servers.pop().expect("at least one server")) + } }