diff --git a/src/service/rooms/state_accessor/mod.rs b/src/service/rooms/state_accessor/mod.rs index ae331fa28..8d2918302 100644 --- a/src/service/rooms/state_accessor/mod.rs +++ b/src/service/rooms/state_accessor/mod.rs @@ -70,12 +70,14 @@ impl crate::Service for Service { } impl Service { + /// Gets the name of a room, if any. pub async fn get_name(&self, room_id: &RoomId) -> Result { self.room_state_get_content(room_id, &StateEventType::RoomName, "") .await .map(|c: RoomNameEventContent| c.name) } + /// Gets the avatar event of a room, if any. pub async fn get_avatar(&self, room_id: &RoomId) -> JsOption { let content = self .room_state_get_content(room_id, &StateEventType::RoomAvatar, "") @@ -85,6 +87,7 @@ impl Service { JsOption::from_option(content) } + /// Gets a member's event content, if one exists. pub async fn get_member( &self, room_id: &RoomId, @@ -124,7 +127,13 @@ impl Service { pub async fn get_room_topic(&self, room_id: &RoomId) -> Result { self.room_state_get_content(room_id, &StateEventType::RoomTopic, "") .await - .map(|c: RoomTopicEventContent| c.topic) + .map(|c: RoomTopicEventContent| { + c.topic_block + .text + .find_plain() + .map(ToOwned::to_owned) + .unwrap_or(c.topic) + }) } /// Returns the join rules for a given room (`JoinRule` type). Will default @@ -135,6 +144,7 @@ impl Service { .map_or(JoinRule::Invite, |c: RoomJoinRulesEventContent| c.join_rule) } + /// Gets the room type of a room, typically a space. pub async fn get_room_type(&self, room_id: &RoomId) -> Result { self.room_state_get_content(room_id, &StateEventType::RoomCreate, "") .await @@ -156,6 +166,7 @@ impl Service { .map(|content: RoomEncryptionEventContent| content.algorithm) } + /// Checks if the given room is one with encryption enabled. pub async fn is_encrypted_room(&self, room_id: &RoomId) -> bool { self.room_state_get(room_id, &StateEventType::RoomEncryption, "") .await @@ -166,7 +177,7 @@ impl Service { pub async fn get_room_create_event(&self, room_id: &RoomId) -> Pdu { self.room_state_get(room_id, &StateEventType::RoomCreate, "") .await - .expect("room should have a create event") + .expect("room must have a create event") } /// Get a set of the room's creators. This will always contain a single user @@ -202,7 +213,7 @@ impl Service { creators } - /// Get the room's power levels. This will never fail -- if the room has no + /// Get the room's power levels. This will never fail - if the room has no /// power level state event, the default power levels for the room's /// version will be returned. pub async fn get_room_power_levels(&self, room_id: &RoomId) -> RoomPowerLevels { diff --git a/src/service/rooms/state_accessor/room_state.rs b/src/service/rooms/state_accessor/room_state.rs index fbc7ed5d4..d1dfbbacc 100644 --- a/src/service/rooms/state_accessor/room_state.rs +++ b/src/service/rooms/state_accessor/room_state.rs @@ -1,112 +1,103 @@ use std::borrow::Borrow; use conduwuit::{ - Pdu, Result, err, implement, + Pdu, Result, err, matrix::{Event, StateKey}, }; use futures::{Stream, StreamExt, TryFutureExt}; use ruma::{EventId, RoomId, events::StateEventType}; use serde::Deserialize; -/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`). -#[implement(super::Service)] -pub async fn room_state_get_content( - &self, - room_id: &RoomId, - event_type: &StateEventType, - state_key: &str, -) -> Result -where - T: for<'de> Deserialize<'de>, -{ - self.room_state_get(room_id, event_type, state_key) - .await - .and_then(|event| event.get_content()) -} +impl super::Service { + /// Returns a single PDU from `room_id` with key (`event_type`,`state_key`). + pub async fn room_state_get_content( + &self, + room_id: &RoomId, + event_type: &StateEventType, + state_key: &str, + ) -> Result + where + T: for<'de> Deserialize<'de>, + { + self.room_state_get(room_id, event_type, state_key) + .await + .and_then(|event| event.get_content()) + } -/// Returns the full room state. -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_state_full<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream> + Send + 'a { - self.services - .state - .get_room_shortstatehash(room_id) - .map_ok(|shortstatehash| self.state_full(shortstatehash).map(Ok).boxed()) - .map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}"))) - .try_flatten_stream() -} + /// Returns the full current room state. + pub fn room_state_full<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream> + Send + 'a { + self.services + .state + .get_room_shortstatehash(room_id) + .map_ok(|shortstatehash| self.state_full(shortstatehash).map(Ok).boxed()) + .map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}"))) + .try_flatten_stream() + } -/// Returns the full room state pdus -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub fn room_state_full_pdus<'a>( - &'a self, - room_id: &'a RoomId, -) -> impl Stream> + Send + 'a { - self.services - .state - .get_room_shortstatehash(room_id) - .map_ok(|shortstatehash| self.state_full_pdus(shortstatehash).map(Ok).boxed()) - .map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}"))) - .try_flatten_stream() -} + /// Returns the full room state pdus + pub fn room_state_full_pdus<'a>( + &'a self, + room_id: &'a RoomId, + ) -> impl Stream> + Send + 'a { + self.services + .state + .get_room_shortstatehash(room_id) + .map_ok(|shortstatehash| self.state_full_pdus(shortstatehash).map(Ok).boxed()) + .map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}"))) + .try_flatten_stream() + } -/// Returns a single EventId from `room_id` with key (`event_type`, -/// `state_key`). -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn room_state_get_id( - &self, - room_id: &RoomId, - event_type: &StateEventType, - state_key: &str, -) -> Result -where - Id: for<'de> Deserialize<'de> + Sized + ToOwned, - ::Owned: Borrow, -{ - self.services - .state - .get_room_shortstatehash(room_id) - .and_then(|shortstatehash| self.state_get_id(shortstatehash, event_type, state_key)) - .await -} + /// Returns a single EventId from `room_id` with key (`event_type`, + /// `state_key`). + pub async fn room_state_get_id( + &self, + room_id: &RoomId, + event_type: &StateEventType, + state_key: &str, + ) -> Result + where + Id: for<'de> Deserialize<'de> + Sized + ToOwned, + ::Owned: Borrow, + { + self.services + .state + .get_room_shortstatehash(room_id) + .and_then(|shortstatehash| self.state_get_id(shortstatehash, event_type, state_key)) + .await + } -/// Returns a single PDU from `room_id` with key (`event_type`, -/// `state_key`). -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn room_state_get( - &self, - room_id: &RoomId, - event_type: &StateEventType, - state_key: &str, -) -> Result { - self.services - .state - .get_room_shortstatehash(room_id) - .and_then(|shortstatehash| self.state_get(shortstatehash, event_type, state_key)) - .await -} + /// Returns a single PDU from `room_id` with key (`event_type`, + /// `state_key`). + pub async fn room_state_get( + &self, + room_id: &RoomId, + event_type: &StateEventType, + state_key: &str, + ) -> Result { + self.services + .state + .get_room_shortstatehash(room_id) + .and_then(|shortstatehash| self.state_get(shortstatehash, event_type, state_key)) + .await + } -/// Returns all state keys for the given `room_id` and `event_type`. -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn room_state_keys( - &self, - room_id: &RoomId, - event_type: &StateEventType, -) -> Result> { - let shortstatehash = self.services.state.get_room_shortstatehash(room_id).await?; + /// Returns all state keys for the given `room_id` and `event_type`. + pub async fn room_state_keys( + &self, + room_id: &RoomId, + event_type: &StateEventType, + ) -> Result> { + let shortstatehash = self.services.state.get_room_shortstatehash(room_id).await?; - let state_keys: Vec = self - .state_keys(shortstatehash, event_type) - .map(|state_key| state_key.to_string()) - .collect() - .await; + let state_keys: Vec = self + .state_keys(shortstatehash, event_type) + .map(|state_key| state_key.to_string()) + .collect() + .await; - Ok(state_keys) + Ok(state_keys) + } } diff --git a/src/service/rooms/state_accessor/server_can.rs b/src/service/rooms/state_accessor/server_can.rs index b992007aa..bbc47529b 100644 --- a/src/service/rooms/state_accessor/server_can.rs +++ b/src/service/rooms/state_accessor/server_can.rs @@ -1,4 +1,4 @@ -use conduwuit::{implement, utils::stream::ReadyExt, warn}; +use conduwuit::{debug_warn, utils::stream::ReadyExt}; use futures::StreamExt; use ruma::{ EventId, RoomId, ServerName, @@ -8,51 +8,53 @@ use ruma::{ }, }; -/// Whether a server is allowed to see an event through federation, based on -/// the room's history_visibility at that event's state. -#[implement(super::Service)] -#[tracing::instrument(skip_all, level = "trace")] -pub async fn server_can_see_event( - &self, - origin: &ServerName, - room_id: &RoomId, - event_id: &EventId, -) -> bool { - let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else { - warn!( - "Unable to visibility check event {} in room {} for server {}: shortstatehash not \ - found", - event_id, room_id, origin - ); - return false; - }; +impl super::Service { + /// Whether a server is allowed to see an event through federation, based on + /// the room's history_visibility at that event's state. + pub async fn server_can_see_event( + &self, + origin: &ServerName, + room_id: &RoomId, + event_id: &EventId, + ) -> bool { + let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else { + debug_warn!( + "Unable to visibility check event {} in room {} for server {}: shortstatehash \ + not found", + event_id, + room_id, + origin + ); + return false; + }; - let history_visibility = self - .state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "") - .await - .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { - c.history_visibility - }); + let history_visibility = self + .state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "") + .await + .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { + c.history_visibility + }); - let current_server_members = self - .services - .state_cache - .room_members(room_id) - .ready_filter(|member| member.server_name() == origin); + let current_server_members = self + .services + .state_cache + .room_members(room_id) + .ready_filter(|member| member.server_name() == origin); - match history_visibility { - | HistoryVisibility::Invited => { - // Allow if any member on requesting server was AT LEAST invited, else deny - current_server_members - .any(async |member| self.user_was_invited(shortstatehash, &member).await) - .await - }, - | HistoryVisibility::Joined => { - // Allow if any member on requested server was joined, else deny - current_server_members - .any(async |member| self.user_was_joined(shortstatehash, &member).await) - .await - }, - | HistoryVisibility::WorldReadable | HistoryVisibility::Shared | _ => true, + match history_visibility { + | HistoryVisibility::Invited => { + // Allow if any member on requesting server was AT LEAST invited, else deny + current_server_members + .any(async |member| self.user_was_invited(shortstatehash, &member).await) + .await + }, + | HistoryVisibility::Joined => { + // Allow if any member on requested server was joined, else deny + current_server_members + .any(async |member| self.user_was_joined(shortstatehash, &member).await) + .await + }, + | HistoryVisibility::WorldReadable | HistoryVisibility::Shared | _ => true, + } } } diff --git a/src/service/rooms/state_accessor/state.rs b/src/service/rooms/state_accessor/state.rs index 8aabbfcf2..2ca26b3ce 100644 --- a/src/service/rooms/state_accessor/state.rs +++ b/src/service/rooms/state_accessor/state.rs @@ -1,7 +1,7 @@ use std::{borrow::Borrow, ops::Deref, sync::Arc}; use conduwuit::{ - Pdu, Result, at, err, implement, + Pdu, Result, at, err, matrix::{Event, StateKey}, pair_of, utils::{ @@ -25,402 +25,406 @@ use crate::rooms::{ state_compressor::{CompressedState, compress_state_event, parse_compressed_state_event}, }; -/// The user was a joined member at this state (potentially in the past) -#[implement(super::Service)] -#[inline] -pub async fn user_was_joined(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool { - self.user_membership(shortstatehash, user_id).await == MembershipState::Join -} - -/// The user was an invited or joined room member at this state (potentially -/// in the past) -#[implement(super::Service)] -#[inline] -pub async fn user_was_invited(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool { - let s = self.user_membership(shortstatehash, user_id).await; - s == MembershipState::Join || s == MembershipState::Invite -} - -/// Get membership for given user in state -#[implement(super::Service)] -pub async fn user_membership( - &self, - shortstatehash: ShortStateHash, - user_id: &UserId, -) -> MembershipState { - self.state_get_content(shortstatehash, &StateEventType::RoomMember, user_id.as_str()) - .await - .map_or(MembershipState::Leave, |c: RoomMemberEventContent| c.membership) -} - -/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`). -#[implement(super::Service)] -pub async fn state_get_content( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, - state_key: &str, -) -> Result -where - T: for<'de> Deserialize<'de>, -{ - self.state_get(shortstatehash, event_type, state_key) - .await - .and_then(|event| event.get_content()) -} - -#[implement(super::Service)] -pub async fn state_contains( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, - state_key: &str, -) -> bool { - let Ok(shortstatekey) = self - .services - .short - .get_shortstatekey(event_type, state_key) - .await - else { - return false; - }; - - self.state_contains_shortstatekey(shortstatehash, shortstatekey) - .await -} - -#[implement(super::Service)] -pub async fn state_contains_type( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, -) -> bool { - let state_keys = self.state_keys(shortstatehash, event_type); - - pin_mut!(state_keys); - state_keys.next().await.is_some() -} - -#[implement(super::Service)] -pub async fn state_contains_shortstatekey( - &self, - shortstatehash: ShortStateHash, - shortstatekey: ShortStateKey, -) -> bool { - let start = compress_state_event(shortstatekey, 0); - let end = compress_state_event(shortstatekey, u64::MAX); - - self.load_full_state(shortstatehash) - .map_ok(|full_state| full_state.range(start..=end).next().copied()) - .await - .flat_ok() - .is_some() -} - -/// Returns a single PDU from `room_id` with key (`event_type`, -/// `state_key`). -#[implement(super::Service)] -pub async fn state_get( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, - state_key: &str, -) -> Result { - self.state_get_id(shortstatehash, event_type, state_key) - .and_then(async |event_id: OwnedEventId| self.services.timeline.get_pdu(&event_id).await) - .await -} - -/// Returns a single EventId from `room_id` with key (`event_type`, -/// `state_key`). -#[implement(super::Service)] -pub async fn state_get_id( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, - state_key: &str, -) -> Result -where - Id: for<'de> Deserialize<'de> + Sized + ToOwned, - ::Owned: Borrow, -{ - let shorteventid = self - .state_get_shortid(shortstatehash, event_type, state_key) - .await?; - - self.services - .short - .get_eventid_from_short(shorteventid) - .await -} - -/// Returns a single EventId from `room_id` with key (`event_type`, -/// `state_key`). -#[implement(super::Service)] -pub async fn state_get_shortid( - &self, - shortstatehash: ShortStateHash, - event_type: &StateEventType, - state_key: &str, -) -> Result { - let shortstatekey = self - .services - .short - .get_shortstatekey(event_type, state_key) - .await?; - - let start = compress_state_event(shortstatekey, 0); - let end = compress_state_event(shortstatekey, u64::MAX); - self.load_full_state(shortstatehash) - .map_ok(|full_state| { - full_state - .range(start..=end) - .next() - .copied() - .map(parse_compressed_state_event) - .map(at!(1)) - .ok_or(err!(Request(NotFound("Not found in room state")))) - }) - .await? -} - -/// Iterates the state_keys for an event_type in the state; current state -/// event_id included. -#[implement(super::Service)] -pub fn state_keys_with_ids<'a, Id>( - &'a self, - shortstatehash: ShortStateHash, - event_type: &'a StateEventType, -) -> impl Stream + Send + 'a -where - Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a, - ::Owned: Borrow, -{ - let state_keys_with_short_ids = self - .state_keys_with_shortids(shortstatehash, event_type) - .unzip() - .map(|(ssks, sids): (Vec, Vec)| (ssks, sids)) - .shared(); - - let state_keys = state_keys_with_short_ids - .clone() - .map(at!(0)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - let shorteventids = state_keys_with_short_ids - .map(at!(1)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - self.services - .short - .multi_get_eventid_from_short(shorteventids) - .zip(state_keys) - .ready_filter_map(|(eid, sk)| eid.map(move |eid| (sk, eid)).ok()) -} - -/// Iterates the state_keys for an event_type in the state; current state -/// event_id included. -#[implement(super::Service)] -pub fn state_keys_with_shortids<'a>( - &'a self, - shortstatehash: ShortStateHash, - event_type: &'a StateEventType, -) -> impl Stream + Send + 'a { - let short_ids = self - .state_full_shortids(shortstatehash) - .ignore_err() - .unzip() - .map(|(ssks, sids): (Vec, Vec)| (ssks, sids)) - .boxed() - .shared(); - - let shortstatekeys = short_ids - .clone() - .map(at!(0)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - let shorteventids = short_ids - .map(at!(1)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - self.services - .short - .multi_get_statekey_from_short(shortstatekeys) - .zip(shorteventids) - .ready_filter_map(|(res, id)| res.map(|res| (res, id)).ok()) - .ready_filter_map(move |((event_type_, state_key), event_id)| { - event_type_.eq(event_type).then_some((state_key, event_id)) - }) -} - -/// Iterates the state_keys for an event_type in the state -#[implement(super::Service)] -pub fn state_keys<'a>( - &'a self, - shortstatehash: ShortStateHash, - event_type: &'a StateEventType, -) -> impl Stream + Send + 'a { - let short_ids = self - .state_full_shortids(shortstatehash) - .ignore_err() - .map(at!(0)); - - self.services - .short - .multi_get_statekey_from_short(short_ids) - .ready_filter_map(Result::ok) - .ready_filter_map(move |(event_type_, state_key)| { - event_type_.eq(event_type).then_some(state_key) - }) -} - -/// Returns the state events removed between the interval (present in .0 but -/// not in .1) -#[implement(super::Service)] -#[inline] -pub async fn state_removed( - &self, - shortstatehash: pair_of!(ShortStateHash), -) -> Result> { - self.state_added((shortstatehash.1, shortstatehash.0)).await -} - -/// Returns the state events added between the interval (present in .1 but -/// not in .0) -#[implement(super::Service)] -pub async fn state_added( - &self, - shortstatehash: pair_of!(ShortStateHash), -) -> Result> { - let full_state_a = self.load_full_state(shortstatehash.0).await?; - let full_state_b = self.load_full_state(shortstatehash.1).await?; - - Ok(full_state_b - .difference(&full_state_a) - .copied() - .map(parse_compressed_state_event) - .collect()) -} - -#[implement(super::Service)] -pub fn state_full( - &self, - shortstatehash: ShortStateHash, -) -> impl Stream + Send + '_ { - self.state_full_pdus(shortstatehash) - .ready_filter_map(|pdu| { - Some(((pdu.kind().to_string().into(), pdu.state_key()?.into()), pdu)) - }) -} - -#[implement(super::Service)] -pub fn state_full_pdus( - &self, - shortstatehash: ShortStateHash, -) -> impl Stream + Send + '_ { - let short_ids = self - .state_full_shortids(shortstatehash) - .ignore_err() - .map(at!(1)); - - self.services - .short - .multi_get_eventid_from_short(short_ids) - .ready_filter_map(Result::ok) - .broad_filter_map(move |event_id: OwnedEventId| async move { - self.services.timeline.get_pdu(&event_id).await.ok() - }) -} - -/// Builds a StateMap by iterating over all keys that start -/// with state_hash, this gives the full state for the given state_hash. -#[implement(super::Service)] -pub fn state_full_ids<'a, Id>( - &'a self, - shortstatehash: ShortStateHash, -) -> impl Stream + Send + 'a -where - Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a, - ::Owned: Borrow, -{ - let shortids = self - .state_full_shortids(shortstatehash) - .ignore_err() - .unzip() - .shared(); - - let shortstatekeys = shortids - .clone() - .map(at!(0)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - let shorteventids = shortids - .map(at!(1)) - .map(Vec::into_iter) - .map(IterStream::stream) - .flatten_stream(); - - self.services - .short - .multi_get_eventid_from_short(shorteventids) - .zip(shortstatekeys) - .ready_filter_map(|(event_id, shortstatekey)| Some((shortstatekey, event_id.ok()?))) -} - -#[implement(super::Service)] -pub fn state_full_shortids( - &self, - shortstatehash: ShortStateHash, -) -> impl Stream> + Send + '_ { - self.load_full_state(shortstatehash) - .map_ok(|full_state| { - full_state - .deref() - .iter() - .copied() - .map(parse_compressed_state_event) - .collect() - }) - .map_ok(Vec::into_iter) - .map_ok(IterStream::try_stream) - .try_flatten_stream() - .boxed() -} - -#[implement(super::Service)] -#[tracing::instrument(name = "load", level = "debug", skip(self))] -async fn load_full_state(&self, shortstatehash: ShortStateHash) -> Result> { - self.services - .state_compressor - .load_shortstatehash_info(shortstatehash) - .map_err(|e| err!(Database("Missing state IDs: {e}"))) - .map_ok(|vec| vec.last().expect("at least one layer").full_state.clone()) - .await -} - -/// Returns the state hash for this pdu. -#[implement(super::Service)] -pub async fn pdu_shortstatehash(&self, event_id: &EventId) -> Result { - const BUFSIZE: usize = size_of::(); - - self.services - .short - .get_shorteventid(event_id) - .and_then(|shorteventid| { - self.db - .shorteventid_shortstatehash - .aqry::(&shorteventid) - }) - .await - .deserialized() +impl super::Service { + /// The user was a joined member at this state (potentially in the past) + pub async fn user_was_joined( + &self, + shortstatehash: ShortStateHash, + user_id: &UserId, + ) -> bool { + self.user_membership(shortstatehash, user_id).await == MembershipState::Join + } + + /// The user was an invited or joined room member at this state (potentially + /// in the past) + pub async fn user_was_invited( + &self, + shortstatehash: ShortStateHash, + user_id: &UserId, + ) -> bool { + matches!( + self.user_membership(shortstatehash, user_id).await, + MembershipState::Invite | MembershipState::Join + ) + } + + /// Get membership for given user in state. If the user has no membership, + /// leave is returned. + pub async fn user_membership( + &self, + shortstatehash: ShortStateHash, + user_id: &UserId, + ) -> MembershipState { + self.state_get_content(shortstatehash, &StateEventType::RoomMember, user_id.as_str()) + .await + .map_or(MembershipState::Leave, |c: RoomMemberEventContent| c.membership) + } + + /// Returns a single PDU from `room_id` with key (`event_type`,`state_key`). + pub async fn state_get_content( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + state_key: &str, + ) -> Result + where + T: for<'de> Deserialize<'de>, + { + self.state_get(shortstatehash, event_type, state_key) + .await + .and_then(|event| event.get_content()) + } + + /// Returns true if the state at the given snapshot contains a specific + /// state key. Returns false if the shortstatehash is not found. + pub async fn state_contains( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + state_key: &str, + ) -> bool { + let Ok(shortstatekey) = self + .services + .short + .get_shortstatekey(event_type, state_key) + .await + else { + return false; + }; + + self.state_contains_shortstatekey(shortstatehash, shortstatekey) + .await + } + + /// Returns true if the state at the given snapshot contains a specific + /// event type, regardless of state key. Returns false if the shortstatehash + /// is not found. + pub async fn state_contains_type( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + ) -> bool { + let state_keys = self.state_keys(shortstatehash, event_type); + + pin_mut!(state_keys); + state_keys.next().await.is_some() + } + + /// Returns true if the given snapshot contains a compressed state ID. + /// Returns false if the snapshot cannot be found. + pub async fn state_contains_shortstatekey( + &self, + shortstatehash: ShortStateHash, + shortstatekey: ShortStateKey, + ) -> bool { + let start = compress_state_event(shortstatekey, 0); + let end = compress_state_event(shortstatekey, u64::MAX); + + self.load_full_state(shortstatehash) + .map_ok(|full_state| full_state.range(start..=end).next().copied()) + .await + .flat_ok() + .is_some() + } + + /// Returns a single PDU from `room_id` with key (`event_type`, + /// `state_key`). + pub async fn state_get( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + state_key: &str, + ) -> Result { + self.state_get_id(shortstatehash, event_type, state_key) + .and_then(async |event_id: OwnedEventId| { + self.services.timeline.get_pdu(&event_id).await + }) + .await + } + + /// Returns a single EventId from `room_id` with key (`event_type`, + /// `state_key`). + pub async fn state_get_id( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + state_key: &str, + ) -> Result + where + Id: for<'de> Deserialize<'de> + Sized + ToOwned, + ::Owned: Borrow, + { + let shorteventid = self + .state_get_shortid(shortstatehash, event_type, state_key) + .await?; + + self.services + .short + .get_eventid_from_short(shorteventid) + .await + } + + /// Returns a single EventId from `room_id` with key (`event_type`, + /// `state_key`). + pub async fn state_get_shortid( + &self, + shortstatehash: ShortStateHash, + event_type: &StateEventType, + state_key: &str, + ) -> Result { + let shortstatekey = self + .services + .short + .get_shortstatekey(event_type, state_key) + .await?; + + let start = compress_state_event(shortstatekey, 0); + let end = compress_state_event(shortstatekey, u64::MAX); + self.load_full_state(shortstatehash) + .map_ok(|full_state| { + full_state + .range(start..=end) + .next() + .copied() + .map(parse_compressed_state_event) + .map(at!(1)) + .ok_or(err!(Request(NotFound("Not found in room state")))) + }) + .await? + } + + /// Iterates the state_keys for an event_type in the state; current state + /// event_id included. + pub fn state_keys_with_ids<'a, Id>( + &'a self, + shortstatehash: ShortStateHash, + event_type: &'a StateEventType, + ) -> impl Stream + Send + 'a + where + Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a, + ::Owned: Borrow, + { + let state_keys_with_short_ids = self + .state_keys_with_shortids(shortstatehash, event_type) + .unzip() + .map(|(ssks, sids): (Vec, Vec)| (ssks, sids)) + .shared(); + + let state_keys = state_keys_with_short_ids + .clone() + .map(at!(0)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + let shorteventids = state_keys_with_short_ids + .map(at!(1)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + self.services + .short + .multi_get_eventid_from_short(shorteventids) + .zip(state_keys) + .ready_filter_map(|(eid, sk)| eid.map(move |eid| (sk, eid)).ok()) + } + + /// Iterates the state_keys for an event_type in the state; current state + /// event_id included. + pub fn state_keys_with_shortids<'a>( + &'a self, + shortstatehash: ShortStateHash, + event_type: &'a StateEventType, + ) -> impl Stream + Send + 'a { + let short_ids = self + .state_full_shortids(shortstatehash) + .ignore_err() + .unzip() + .map(|(ssks, sids): (Vec, Vec)| (ssks, sids)) + .boxed() + .shared(); + + let shortstatekeys = short_ids + .clone() + .map(at!(0)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + let shorteventids = short_ids + .map(at!(1)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + self.services + .short + .multi_get_statekey_from_short(shortstatekeys) + .zip(shorteventids) + .ready_filter_map(|(res, id)| res.map(|res| (res, id)).ok()) + .ready_filter_map(move |((event_type_, state_key), event_id)| { + event_type_.eq(event_type).then_some((state_key, event_id)) + }) + } + + /// Iterates the state_keys for an event_type in the state + pub fn state_keys<'a>( + &'a self, + shortstatehash: ShortStateHash, + event_type: &'a StateEventType, + ) -> impl Stream + Send + 'a { + let short_ids = self + .state_full_shortids(shortstatehash) + .ignore_err() + .map(at!(0)); + + self.services + .short + .multi_get_statekey_from_short(short_ids) + .ready_filter_map(Result::ok) + .ready_filter_map(move |(event_type_, state_key)| { + event_type_.eq(event_type).then_some(state_key) + }) + } + + /// Returns the state events removed between the interval (present in .0 but + /// not in .1) + pub async fn state_removed( + &self, + shortstatehash: pair_of!(ShortStateHash), + ) -> Result> { + self.state_added((shortstatehash.1, shortstatehash.0)).await + } + + /// Returns the state events added between the interval (present in .1 but + /// not in .0) + pub async fn state_added( + &self, + shortstatehash: pair_of!(ShortStateHash), + ) -> Result> { + let full_state_a = self.load_full_state(shortstatehash.0).await?; + let full_state_b = self.load_full_state(shortstatehash.1).await?; + + Ok(full_state_b + .difference(&full_state_a) + .copied() + .map(parse_compressed_state_event) + .collect()) + } + + /// Returns the full set of state keys in the state snapshot. + pub fn state_full( + &self, + shortstatehash: ShortStateHash, + ) -> impl Stream + Send + '_ { + self.state_full_pdus(shortstatehash) + .ready_filter_map(|pdu| { + Some(((pdu.kind().to_string().into(), pdu.state_key()?.into()), pdu)) + }) + } + + /// Fetches all PDUs belonging to the given state snapshot. + pub fn state_full_pdus( + &self, + shortstatehash: ShortStateHash, + ) -> impl Stream + Send + '_ { + let short_ids = self + .state_full_shortids(shortstatehash) + .ignore_err() + .map(at!(1)); + + self.services + .short + .multi_get_eventid_from_short(short_ids) + .ready_filter_map(Result::ok) + .broad_filter_map(move |event_id: OwnedEventId| async move { + self.services.timeline.get_pdu(&event_id).await.ok() + }) + } + + /// Builds a StateMap by iterating over all keys that start + /// with state_hash, this gives the full state for the given state_hash. + pub fn state_full_ids<'a, Id>( + &'a self, + shortstatehash: ShortStateHash, + ) -> impl Stream + Send + 'a + where + Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a, + ::Owned: Borrow, + { + let shortids = self + .state_full_shortids(shortstatehash) + .ignore_err() + .unzip() + .shared(); + + let shortstatekeys = shortids + .clone() + .map(at!(0)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + let shorteventids = shortids + .map(at!(1)) + .map(Vec::into_iter) + .map(IterStream::stream) + .flatten_stream(); + + self.services + .short + .multi_get_eventid_from_short(shorteventids) + .zip(shortstatekeys) + .ready_filter_map(|(event_id, shortstatekey)| Some((shortstatekey, event_id.ok()?))) + } + + /// Fetches all compressed state IDs that make up the given state snapshot. + pub fn state_full_shortids( + &self, + shortstatehash: ShortStateHash, + ) -> impl Stream> + Send + '_ { + self.load_full_state(shortstatehash) + .map_ok(|full_state| { + full_state + .deref() + .iter() + .copied() + .map(parse_compressed_state_event) + .collect() + }) + .map_ok(Vec::into_iter) + .map_ok(IterStream::try_stream) + .try_flatten_stream() + .boxed() + } + + /// Loads the full state for a state snapshot + async fn load_full_state( + &self, + shortstatehash: ShortStateHash, + ) -> Result> { + self.services + .state_compressor + .load_shortstatehash_info(shortstatehash) + .map_err(|e| err!(Database("Missing state IDs: {e}"))) + .map_ok(|vec| vec.last().expect("at least one layer").full_state.clone()) + .await + } + + /// Returns the state hash for this pdu. + pub async fn pdu_shortstatehash(&self, event_id: &EventId) -> Result { + const BUFSIZE: usize = size_of::(); + + self.services + .short + .get_shorteventid(event_id) + .and_then(|shorteventid| { + self.db + .shorteventid_shortstatehash + .aqry::(&shorteventid) + }) + .await + .deserialized() + } } diff --git a/src/service/rooms/state_accessor/user_can.rs b/src/service/rooms/state_accessor/user_can.rs index fb74b27ad..b27a57304 100644 --- a/src/service/rooms/state_accessor/user_can.rs +++ b/src/service/rooms/state_accessor/user_can.rs @@ -1,4 +1,4 @@ -use conduwuit::{Err, Result, implement, matrix::Event, pdu::PartialPdu}; +use conduwuit::{Err, Result, matrix::Event, pdu::PartialPdu}; use ruma::{ EventId, RoomId, UserId, events::{ @@ -12,140 +12,139 @@ use ruma::{ use crate::rooms::state::RoomMutexGuard; -/// Checks if a given user can redact a given event -/// -/// If federation is true, it allows redaction events from any user of the -/// same server as the original event sender -#[implement(super::Service)] -pub async fn user_can_redact( - &self, - redacts: &EventId, - sender: &UserId, - room_id: &RoomId, - federation: bool, -) -> Result { - let redacting_event = self.services.timeline.get_pdu(redacts).await; +impl super::Service { + /// Checks if a given user can redact a given event + /// + /// If federation is true, it allows redaction events from any user of the + /// same server as the original event sender + pub async fn user_can_redact( + &self, + redacts: &EventId, + sender: &UserId, + room_id: &RoomId, + federation: bool, + ) -> Result { + let redacting_event = self.services.timeline.get_pdu(redacts).await; - if redacting_event - .as_ref() - .is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomCreate) - { - return Err!(Request(Forbidden("Redacting m.room.create is not safe, forbidding."))); + if redacting_event + .as_ref() + .is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomCreate) + { + return Err!(Request(Forbidden("Redacting m.room.create is not safe, forbidding."))); + } + + if redacting_event + .as_ref() + .is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomServerAcl) + { + return Err!(Request(Forbidden( + "Redacting m.room.server_acl will result in the room being inaccessible for \ + everyone (empty allow key), forbidding." + ))); + } + + let power_levels = self.get_room_power_levels(room_id).await; + + if power_levels.user_can_redact_event_of_other(sender) { + return Ok(true); + } + + if power_levels.user_can_redact_own_event(sender) { + let is_own_event = match redacting_event { + | Ok(redacting_event) => + if federation { + redacting_event.sender().server_name() == sender.server_name() + } else { + redacting_event.sender() == sender + }, + | _ => false, + }; + + return Ok(is_own_event); + } + + Ok(false) } - if redacting_event - .as_ref() - .is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomServerAcl) - { - return Err!(Request(Forbidden( - "Redacting m.room.server_acl will result in the room being inaccessible for \ - everyone (empty allow key), forbidding." - ))); - } - - let power_levels = self.get_room_power_levels(room_id).await; - - if power_levels.user_can_redact_event_of_other(sender) { - return Ok(true); - } - - if power_levels.user_can_redact_own_event(sender) { - let is_own_event = match redacting_event { - | Ok(redacting_event) => - if federation { - redacting_event.sender().server_name() == sender.server_name() - } else { - redacting_event.sender() == sender - }, - | _ => false, + /// Whether a user is allowed to see an event, based on + /// the room's history_visibility at that event's state. + pub async fn user_can_see_event( + &self, + user_id: &UserId, + room_id: &RoomId, + event_id: &EventId, + ) -> bool { + let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else { + return true; }; - return Ok(is_own_event); + let currently_member = self.services.state_cache.is_joined(user_id, room_id).await; + + let history_visibility = self + .state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "") + .await + .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { + c.history_visibility + }); + + match history_visibility { + | HistoryVisibility::Invited => { + // Allow if any member on requesting server was AT LEAST invited, else deny + self.user_was_invited(shortstatehash, user_id).await + }, + | HistoryVisibility::Joined => { + // Allow if any member on requested server was joined, else deny + self.user_was_joined(shortstatehash, user_id).await + }, + | HistoryVisibility::WorldReadable => true, + | HistoryVisibility::Shared | _ => currently_member, + } } - Ok(false) -} + /// Whether a user is allowed to see an event, based on + /// the room's history_visibility at that event's state. + pub async fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> bool { + if self.services.state_cache.is_joined(user_id, room_id).await { + return true; + } -/// Whether a user is allowed to see an event, based on -/// the room's history_visibility at that event's state. -#[implement(super::Service)] -#[tracing::instrument(skip_all, level = "trace")] -pub async fn user_can_see_event( - &self, - user_id: &UserId, - room_id: &RoomId, - event_id: &EventId, -) -> bool { - let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else { - return true; - }; + let history_visibility = self + .room_state_get_content(room_id, &StateEventType::RoomHistoryVisibility, "") + .await + .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { + c.history_visibility + }); - let currently_member = self.services.state_cache.is_joined(user_id, room_id).await; - - let history_visibility = self - .state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "") - .await - .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { - c.history_visibility - }); - - match history_visibility { - | HistoryVisibility::Invited => { - // Allow if any member on requesting server was AT LEAST invited, else deny - self.user_was_invited(shortstatehash, user_id).await - }, - | HistoryVisibility::Joined => { - // Allow if any member on requested server was joined, else deny - self.user_was_joined(shortstatehash, user_id).await - }, - | HistoryVisibility::WorldReadable => true, - | HistoryVisibility::Shared | _ => currently_member, - } -} - -/// Whether a user is allowed to see an event, based on -/// the room's history_visibility at that event's state. -#[implement(super::Service)] -#[tracing::instrument(skip_all, level = "trace")] -pub async fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> bool { - if self.services.state_cache.is_joined(user_id, room_id).await { - return true; + match history_visibility { + | HistoryVisibility::Invited => + self.services.state_cache.is_invited(user_id, room_id).await, + | HistoryVisibility::WorldReadable => true, + | _ => false, + } } - let history_visibility = self - .room_state_get_content(room_id, &StateEventType::RoomHistoryVisibility, "") - .await - .map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| { - c.history_visibility - }); - - match history_visibility { - | HistoryVisibility::Invited => - self.services.state_cache.is_invited(user_id, room_id).await, - | HistoryVisibility::WorldReadable => true, - | _ => false, + /// Checks if the target user can send a specific invite by creating and + /// discarding a new PDU. If creating the PDU fails, the user is not + /// permitted to issue the invite. + pub async fn user_can_invite( + &self, + room_id: &RoomId, + sender: &UserId, + target_user: &UserId, + state_lock: &RoomMutexGuard, + ) -> bool { + self.services + .timeline + .create_event( + PartialPdu::state( + target_user.as_str(), + &RoomMemberEventContent::new(MembershipState::Invite), + ), + sender, + Some(room_id), + state_lock, + ) + .await + .is_ok() } } - -#[implement(super::Service)] -pub async fn user_can_invite( - &self, - room_id: &RoomId, - sender: &UserId, - target_user: &UserId, - state_lock: &RoomMutexGuard, -) -> bool { - self.services - .timeline - .create_hash_and_sign_event( - PartialPdu::state( - target_user.as_str(), - &RoomMemberEventContent::new(MembershipState::Invite), - ), - sender, - Some(room_id), - state_lock, - ) - .await - .is_ok() -}