Mirrors/continuwuity
0
0 Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2026-11-13 04:06:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

style: Document & refactor rooms/state_accessor service

Browse Source
This commit is contained in:
timedout committed 2026-06-27 18:36:44 +01:00
1 parent 4844c6ae3e
commit 4d30ee5d33
5 files changed
+673 -666
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/service/rooms/state_accessor/mod.rs
+14 -3
View File
@@ -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<String> {
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<RoomAvatarEventContent> {
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<String> {
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<RoomType> {
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 {
src/service/rooms/state_accessor/room_state.rs
+86 -95
View File
@@ -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<T>(
&self,
room_id: &RoomId,
event_type: &StateEventType,
state_key: &str,
) -> Result<T>
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<T>(
&self,
room_id: &RoomId,
event_type: &StateEventType,
state_key: &str,
) -> Result<T>
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<Item = Result<((StateEventType, StateKey), impl Event)>> + 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<Item = Result<((StateEventType, StateKey), impl Event)>> + 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<Item = Result<impl Event>> + 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<Item = Result<impl Event>> + 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<Id>(
&self,
room_id: &RoomId,
event_type: &StateEventType,
state_key: &str,
) -> Result<Id>
where
Id: for<'de> Deserialize<'de> + Sized + ToOwned,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<Id>(
&self,
room_id: &RoomId,
event_type: &StateEventType,
state_key: &str,
) -> Result<Id>
where
Id: for<'de> Deserialize<'de> + Sized + ToOwned,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<Pdu> {
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<Pdu> {
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<Vec<String>> {
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<Vec<String>> {
let shortstatehash = self.services.state.get_room_shortstatehash(room_id).await?;
let state_keys: Vec<String> = self
.state_keys(shortstatehash, event_type)
.map(|state_key| state_key.to_string())
.collect()
.await;
let state_keys: Vec<String> = self
.state_keys(shortstatehash, event_type)
.map(|state_key| state_key.to_string())
.collect()
.await;
Ok(state_keys)
Ok(state_keys)
}
}
src/service/rooms/state_accessor/server_can.rs
+46 -44
View File
@@ -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,
}
}
}
src/service/rooms/state_accessor/state.rs
+403 -399
View File
@@ -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<T>(
&self,
shortstatehash: ShortStateHash,
event_type: &StateEventType,
state_key: &str,
) -> Result<T>
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<Pdu> {
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<Id>(
&self,
shortstatehash: ShortStateHash,
event_type: &StateEventType,
state_key: &str,
) -> Result<Id>
where
Id: for<'de> Deserialize<'de> + Sized + ToOwned,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<ShortEventId> {
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<Item = (StateKey, Id)> + Send + 'a
where
Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
let state_keys_with_short_ids = self
.state_keys_with_shortids(shortstatehash, event_type)
.unzip()
.map(|(ssks, sids): (Vec<StateKey>, Vec<u64>)| (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<Item = (StateKey, ShortEventId)> + Send + 'a {
let short_ids = self
.state_full_shortids(shortstatehash)
.ignore_err()
.unzip()
.map(|(ssks, sids): (Vec<u64>, Vec<u64>)| (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<Item = StateKey> + 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<Vec<(ShortStateKey, ShortEventId)>> {
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<Vec<(ShortStateKey, ShortEventId)>> {
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<Item = ((StateEventType, StateKey), impl Event)> + 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<Item = impl Event> + 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<Item = (ShortStateKey, Id)> + Send + 'a
where
Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<Item = Result<(ShortStateKey, ShortEventId)>> + 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<Arc<CompressedState>> {
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<ShortStateHash> {
const BUFSIZE: usize = size_of::<ShortEventId>();
self.services
.short
.get_shorteventid(event_id)
.and_then(|shorteventid| {
self.db
.shorteventid_shortstatehash
.aqry::<BUFSIZE, _>(&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<T>(
&self,
shortstatehash: ShortStateHash,
event_type: &StateEventType,
state_key: &str,
) -> Result<T>
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<Pdu> {
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<Id>(
&self,
shortstatehash: ShortStateHash,
event_type: &StateEventType,
state_key: &str,
) -> Result<Id>
where
Id: for<'de> Deserialize<'de> + Sized + ToOwned,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<ShortEventId> {
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<Item = (StateKey, Id)> + Send + 'a
where
Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
let state_keys_with_short_ids = self
.state_keys_with_shortids(shortstatehash, event_type)
.unzip()
.map(|(ssks, sids): (Vec<StateKey>, Vec<u64>)| (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<Item = (StateKey, ShortEventId)> + Send + 'a {
let short_ids = self
.state_full_shortids(shortstatehash)
.ignore_err()
.unzip()
.map(|(ssks, sids): (Vec<u64>, Vec<u64>)| (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<Item = StateKey> + 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<Vec<(ShortStateKey, ShortEventId)>> {
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<Vec<(ShortStateKey, ShortEventId)>> {
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<Item = ((StateEventType, StateKey), impl Event)> + 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<Item = impl Event> + 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<Item = (ShortStateKey, Id)> + Send + 'a
where
Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
<Id as ToOwned>::Owned: Borrow<EventId>,
{
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<Item = Result<(ShortStateKey, ShortEventId)>> + 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<Arc<CompressedState>> {
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<ShortStateHash> {
const BUFSIZE: usize = size_of::<ShortEventId>();
self.services
.short
.get_shorteventid(event_id)
.and_then(|shorteventid| {
self.db
.shorteventid_shortstatehash
.aqry::<BUFSIZE, _>(&shorteventid)
})
.await
.deserialized()
}
}
src/service/rooms/state_accessor/user_can.rs
+124 -125
View File
@@ -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<bool> {
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<bool> {
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()
}