dandri/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity/ synced 2026-08-28 11:54:22 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

style: Document & refactor rooms/user and rooms/timeline services

Browse Source
This commit is contained in:
timedout
2026-06-27 18:36:44 +01:00
parent aa2f506996
commit 199f6eb67b
7 changed files with 1241 additions and 1133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/rooms/membership/mod.rs
+1 -2
View File
@@ -282,8 +282,7 @@ async fn join_local_room(
remote_servers = %servers.len(),
"Could not join room locally, attempting remote join",
);
self.join_remote_room(sender_user, room_id, reason, servers, state_lock)
.await
Box::pin(self.join_remote_room(sender_user, room_id, reason, servers, state_lock)).await
}
#[tracing::instrument(skip_all, fields(%sender_user, %room_id), name = "join_remote_room", level = "info")]
src/service/rooms/timeline/append.rs
+393 -308
View File
@@ -3,16 +3,22 @@
sync::Arc,
};
use conduwuit::trace;
use conduwuit::{
debug_warn,
pdu::{Count, ShortRoomId},
trace,
utils::{IterStream, TryFutureExtExt, stream::BroadbandExt},
warn,
};
use conduwuit_core::{
Result, err, error, implement,
Result, err, error,
matrix::{
event::Event,
pdu::{PduCount, PduEvent, PduId, RawPduId},
},
utils::{self, ReadyExt},
};
use futures::StreamExt;
use futures::{StreamExt, TryFutureExt};
use ruma::{
CanonicalJsonObject, CanonicalJsonValue, EventId, RoomVersionId, UserId,
events::{
@@ -24,206 +30,245 @@
};
use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard};
use crate::{appservice::NamespaceRegex, rooms::state_compressor::CompressedState};
use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState};
/// Append the incoming event setting the state snapshot to the state from
/// the server that sent the event.
#[implement(super::Service)]
#[tracing::instrument(level = "debug", skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn append_incoming_pdu<'a, Leaves>(
&'a self,
pdu: &'a PduEvent,
pdu_json: CanonicalJsonObject,
new_room_leaves: Leaves,
state_ids_compressed: Arc<CompressedState>,
soft_fail: bool,
state_lock: &'a RoomMutexGuard,
room_id: &'a ruma::RoomId,
) -> Result<Option<RawPduId>>
where
Leaves: Iterator<Item = &'a EventId> + Send + 'a,
{
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
self.services
.state
.set_event_state(&pdu.event_id, room_id, state_ids_compressed)
.await?;
impl super::Service {
/// Append the incoming event setting the state snapshot to the state from
/// the server that sent the event.
#[allow(clippy::too_many_arguments)]
pub async fn append_incoming_pdu<'a, Leaves>(
&'a self,
pdu: &'a PduEvent,
pdu_json: CanonicalJsonObject,
new_room_leaves: Leaves,
state_ids_compressed: Arc<CompressedState>,
soft_fail: bool,
state_lock: &'a RoomMutexGuard,
room_id: &'a ruma::RoomId,
) -> Result<Option<RawPduId>>
where
Leaves: Iterator<Item = &'a EventId> + Send + 'a,
{
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
self.services
.state
.set_event_state(&pdu.event_id, room_id, state_ids_compressed)
.await?;
if soft_fail {
// Nothing else to do with a soft-failed event.
return Ok(None);
}
let pdu_id = self
.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
.await?;
// Process admin commands for federation events
if *pdu.kind() == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
if let Some(source) = self
.services
.admin
.is_admin_command(pdu, &body, false)
.await
{
self.services.admin.command_with_sender(
body,
Some(pdu.event_id().into()),
source,
pdu.sender.clone(),
)?;
}
if soft_fail {
// Nothing else to do with a soft-failed event.
return Ok(None);
}
}
Ok(Some(pdu_id))
}
let pdu_id = self
.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
.await?;
/// Creates a new persisted data unit and adds it to a room.
///
/// By this point the incoming event should be fully authenticated, no auth
/// happens in `append_pdu`.
///
/// Returns pdu id
#[implement(super::Service)]
#[tracing::instrument(level = "debug", skip_all)]
pub async fn append_pdu<'a, Leaves>(
&'a self,
pdu: &'a PduEvent,
mut pdu_json: CanonicalJsonObject,
leaves: Leaves,
state_lock: &'a RoomMutexGuard,
room_id: &'a ruma::RoomId,
) -> Result<RawPduId>
where
Leaves: Iterator<Item = &'a EventId> + Send + 'a,
{
// Coalesce database writes for the remainder of this scope.
let _cork = self.db.db.cork_and_flush();
let shortroomid = self
.services
.short
.get_shortroomid(room_id)
.await
.map_err(|_| err!(Database("Room does not exist")))?;
// Make unsigned fields correct. This is not properly documented in the spec,
// but state events need to have previous content in the unsigned field, so
// clients can easily interpret things like membership changes
if let Some(state_key) = pdu.state_key() {
if let CanonicalJsonValue::Object(unsigned) = pdu_json
.entry("unsigned".to_owned())
.or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::default()))
{
if let Ok(shortstatehash) = self
.services
.state_accessor
.pdu_shortstatehash(pdu.event_id())
.await
{
if let Ok(prev_state) = self
// Process admin commands for federation events
if *pdu.kind() == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
if let Some(source) = self
.services
.state_accessor
.state_get(shortstatehash, &pdu.kind().to_string().into(), state_key)
.admin
.is_admin_command(pdu, &body, false)
.await
{
unsigned.insert(
"prev_content".to_owned(),
CanonicalJsonValue::Object(
utils::to_canonical_object(prev_state.get_content_as_value())
.map_err(|e| {
err!(Database(error!(
"Failed to convert prev_state to canonical JSON: {e}",
)))
})?,
),
);
unsigned.insert(
String::from("prev_sender"),
CanonicalJsonValue::String(prev_state.sender().to_string()),
);
unsigned.insert(
String::from("replaces_state"),
CanonicalJsonValue::String(prev_state.event_id().to_string()),
);
self.services.admin.command_with_sender(
body,
Some(pdu.event_id().into()),
source,
pdu.sender.clone(),
)?;
}
}
} else {
error!("Invalid unsigned type in pdu.");
}
Ok(Some(pdu_id))
}
// We must keep track of all events that have been referenced.
self.services
.pdu_metadata
.mark_as_referenced(room_id, pdu.prev_events().map(AsRef::as_ref));
/// Populates the unsigned data of a PDU
async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) {
let Some(state_key) = pdu.state_key() else {
return; // Non-state events can't have replaced state
};
trace!("setting forward extremities");
self.services
.state
.set_forward_extremities(room_id, leaves, state_lock)
.await;
let CanonicalJsonValue::Object(unsigned) = pdu_json
.entry("unsigned".into())
.or_insert_with(|| CanonicalJsonValue::Object(BTreeMap::new()))
else {
return; // This shouldn't be reachable, really.
};
let insert_lock = self.mutex_insert.lock(room_id).await;
let Ok(shortstatehash) = self
.services
.state_accessor
.pdu_shortstatehash(pdu.event_id())
.await
else {
return;
};
let Ok(prev_state) = self
.services
.state_accessor
.state_get(shortstatehash, &pdu.kind().to_string().into(), state_key)
.await
else {
return;
};
unsigned.insert(
"prev_content".to_owned(),
CanonicalJsonValue::Object(
utils::to_canonical_object(prev_state.get_content_as_value())
.expect("Failed to convert prev_content into canonical JSON object"),
),
);
unsigned.insert(
String::from("prev_sender"),
CanonicalJsonValue::String(prev_state.sender().to_string()),
);
unsigned.insert(
String::from("replaces_state"),
CanonicalJsonValue::String(prev_state.event_id().to_string()),
);
}
let count1 = self.services.globals.next_count().unwrap();
/// Creates a new persisted data unit and adds it to a room.
///
/// By this point the incoming event should be fully authenticated, no auth
/// happens in `append_pdu`.
///
/// Returns pdu id
pub async fn append_pdu<'a, Leaves>(
&'a self,
pdu: &'a PduEvent,
mut pdu_json: CanonicalJsonObject,
leaves: Leaves,
state_lock: &'a RoomMutexGuard,
room_id: &'a ruma::RoomId,
) -> Result<RawPduId>
where
Leaves: Iterator<Item = &'a EventId> + Send + 'a,
{
// Coalesce database writes for the remainder of this scope.
let _cork = self.db.db.cork_and_flush();
// Mark as read first so the sending client doesn't get a notification even if
// appending fails
self.services
.read_receipt
.private_read_set(room_id, pdu.sender(), count1);
let shortroomid = self
.services
.short
.get_shortroomid(room_id)
.await
.map_err(|_| err!(Database("Room does not exist")))?;
self.services
.user
.reset_notification_counts(pdu.sender(), room_id);
// Make unsigned fields correct. This is not properly documented in the spec,
// but state events need to have previous content in the unsigned field, so
// clients can easily interpret things like membership changes
let count2 = PduCount::Normal(self.services.globals.next_count().unwrap());
let pdu_id: RawPduId = PduId { shortroomid, shorteventid: count2 }.into();
// TODO: This needs to be refactored to add this information on the fly.
// Because the prev content becomes part of the unsigned object in the PDU, we
// unintentionally leak redacted or hidden content to local users.
// See: https://forgejo.ellis.link/continuwuation/continuwuity/issues/1103
self.populate_unsigned(pdu, &mut pdu_json).await;
// Insert pdu
self.db.append_pdu(&pdu_id, pdu, &pdu_json, count2).await;
// We must keep track of all events that have been referenced.
self.services
.pdu_metadata
.mark_as_referenced(room_id, pdu.prev_events().map(AsRef::as_ref));
drop(insert_lock);
trace!("setting forward extremities");
self.services
.state
.set_forward_extremities(room_id, leaves, state_lock)
.await;
// See if the event matches any known pushers via power level
if *pdu.kind() != TimelineEventType::RoomCreate {
let insert_lock = self.mutex_insert.lock(room_id).await;
let count1 = self.services.globals.next_count().unwrap();
// Mark as read first so the sending client doesn't get a notification even if
// appending fails
// TODO: Is this necessary? appending doesn't seem that fallible, and if it is,
// there's bigger issues than ghost notifications.
self.services
.read_receipt
.private_read_set(room_id, pdu.sender(), count1);
self.services
.user
.reset_notification_counts(pdu.sender(), room_id);
let count2 = PduCount::Normal(self.services.globals.next_count().unwrap());
let pdu_id: RawPduId = PduId { shortroomid, shorteventid: count2 }.into();
// Insert pdu
self.db.append_pdu(&pdu_id, pdu, &pdu_json, count2).await;
drop(insert_lock);
// See if the event matches any known pushers via power level
if *pdu.kind() != TimelineEventType::RoomCreate {
tokio::join!(
self.notify_local_users(pdu, &pdu_id),
self.handle_pdu_effects(pdu, &pdu_id, shortroomid)
.inspect_err(|e| {
error!(
"failed to handle PDU effects of incoming PDU {}: {e:?}",
pdu.event_id()
);
})
.ok(),
self.aggregate_relations(pdu, count2),
);
}
self.send_to_interested_appservices(pdu, &pdu_id).await;
Ok(pdu_id)
}
/// Notifies local users of the incoming event with a power levels context.
async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
let power_levels = self
.services
.state_accessor
.get_room_power_levels(room_id)
.get_room_power_levels(pdu.room_id().unwrap())
.await;
let mut push_targets: HashSet<_> = self
.services
.state_cache
.active_local_users_in_room(pdu.room_id().unwrap())
// Don't notify the sender of their own events, and don't send from ignored users
.ready_filter(|user| *user != pdu.sender())
.filter_map(|recipient_user| async move {
(!self.services.users.user_is_ignored(pdu.sender(), &recipient_user).await).then_some(recipient_user)
})
.collect()
.await;
let mut push_target: HashSet<_> = self
.services
.state_cache
.active_local_users_in_room(room_id)
// Don't notify the sender of their own events, and dont send from ignored users
.ready_filter(|user| *user != pdu.sender())
.filter_map(|recipient_user| async move { (!self.services.users.user_is_ignored(pdu.sender(), &recipient_user).await).then_some(recipient_user) })
.collect()
.await;
let mut notifies = Vec::with_capacity(push_target.len().saturating_add(1));
let mut highlights = Vec::with_capacity(push_target.len().saturating_add(1));
let mut notifies = Vec::with_capacity(push_targets.len().saturating_add(1));
let mut highlights = Vec::with_capacity(push_targets.len().saturating_add(1));
if *pdu.kind() == TimelineEventType::RoomMember {
if let Some(state_key) = pdu.state_key() {
let target_user_id = UserId::parse(state_key)?;
if self.services.users.is_active_local(&target_user_id).await {
push_target.insert(target_user_id.clone());
match UserId::parse(state_key) {
| Ok(target_user_id) => {
if self.services.users.is_active_local(&target_user_id).await {
push_targets.insert(target_user_id.clone());
}
},
| Err(e) => debug_warn!(user_id=?state_key, ?e, "failed to parse user ID"),
}
}
}
if push_targets.is_empty() {
return;
}
let serialized = pdu.to_format();
for user in &push_target {
for user in &push_targets {
let rules_for_user = self
.services
.account_data
@@ -240,7 +285,13 @@ pub async fn append_pdu<'a, Leaves>(
for action in self
.services
.pusher
.get_actions(user, &rules_for_user, power_levels.clone(), &serialized, room_id)
.get_actions(
user,
&rules_for_user,
power_levels.clone(),
&serialized,
pdu.room_id().unwrap(),
)
.await
{
match action {
@@ -273,163 +324,197 @@ pub async fn append_pdu<'a, Leaves>(
.ready_for_each(|push_key| {
self.services
.sending
.send_pdu_push(&pdu_id, user, push_key.to_owned())
.send_pdu_push(pdu_id, user, push_key.to_owned())
.expect("TODO: replace with future");
})
.await;
}
self.db
.increment_notification_counts(room_id, notifies, highlights);
.increment_notification_counts(pdu.room_id().unwrap(), notifies, highlights);
}
match *pdu.kind() {
| TimelineEventType::RoomRedaction => {
use RoomVersionId::*;
/// Handles PDU effects based on the type of incoming event.
/// For redaction events, handles redacting. Memberships update the
/// membership cache. Et cetera.
async fn handle_pdu_effects(
&self,
pdu: &PduEvent,
pdu_id: &RawPduId,
short_room_id: ShortRoomId,
) -> Result {
let room_id = pdu.room_id().unwrap();
match *pdu.kind() {
| TimelineEventType::RoomRedaction => {
use RoomVersionId::*;
let room_version_id = self.services.state.get_room_version(room_id).await?;
match room_version_id {
| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
if let Some(redact_id) = pdu.redacts() {
if self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, shortroomid).await?;
// TODO: support delayed redaction (MSC2815)
let room_version_id = self.services.state.get_room_version(room_id).await?;
match room_version_id {
| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
if let Some(redact_id) = pdu.redacts() {
if self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, short_room_id).await?;
}
}
}
},
| _ => {
let content: RoomRedactionEventContent = pdu.get_content()?;
if let Some(redact_id) = &content.redacts {
if self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, shortroomid).await?;
},
| _ => {
let content: RoomRedactionEventContent = pdu.get_content()?;
if let Some(redact_id) = &content.redacts {
if self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), room_id, false)
.await?
{
self.redact_pdu(redact_id, pdu, short_room_id).await?;
}
}
}
},
}
},
| TimelineEventType::RoomMember => {
if let Some(state_key) = pdu.state_key() {
// if the state_key fails
let target_user_id =
UserId::parse(state_key).expect("This state_key was previously validated");
// Update our membership info, we do this here incase a user is invited or
// knocked and immediately leaves we need the DB to record the invite or
// knock event for auth
self.services
.state_cache
.update_membership(room_id, &target_user_id, pdu, true)
.await?;
}
},
| TimelineEventType::RoomMessage => {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
self.services.search.index_pdu(shortroomid, &pdu_id, &body);
}
},
| _ => {},
}
// CONCERN: If we receive events with a relation out-of-order, we never write
// their relation / thread. We need some kind of way to trigger when we receive
// this event, and potentially a way to rebuild the table entirely.
if let Ok(content) = pdu.get_content::<ExtractRelatesToEventId>() {
if let Ok(related_pducount) = self.get_pdu_count(&content.relates_to.event_id).await {
self.services
.pdu_metadata
.add_relation(count2, related_pducount);
}
}
if let Ok(content) = pdu.get_content::<ExtractRelatesTo>() {
match content.relates_to {
| Relation::Reply(in_reply_to) => {
// We need to do it again here, because replies don't have
// event_id as a top level field
if let Ok(related_pducount) =
self.get_pdu_count(&in_reply_to.in_reply_to.event_id).await
{
self.services
.pdu_metadata
.add_relation(count2, related_pducount);
},
}
},
| Relation::Thread(thread) => {
self.services
.threads
.add_to_thread(&thread.event_id, pdu)
.await?;
| TimelineEventType::RoomMember => {
if let Some(state_key) = pdu.state_key() {
// if the state_key fails
let target_user_id = UserId::parse(state_key)
.expect("This state_key was previously validated");
// Update our membership info, we do this here incase a user is invited or
// knocked and immediately leaves we need the DB to record the invite or
// knock event for auth
self.services
.state_cache
.update_membership(room_id, &target_user_id, pdu, true)
.await?;
}
},
| _ => {}, // TODO: Aggregate other types
| TimelineEventType::RoomMessage => {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
self.services.search.index_pdu(short_room_id, pdu_id, &body);
}
},
| _ => {},
}
Ok(())
}
/// Adds relation data to the incoming event and events it relates to.
async fn aggregate_relations(&self, pdu: &PduEvent, count2: Count) {
// CONCERN: If we receive events with a relation out-of-order, we never write
// their relation / thread. We need some kind of way to trigger when we receive
// this event, and potentially a way to rebuild the table entirely.
if let Ok(content) = pdu.get_content::<ExtractRelatesToEventId>() {
if let Ok(related_pducount) = self.get_pdu_count(&content.relates_to.event_id).await {
self.services
.pdu_metadata
.add_relation(count2, related_pducount);
}
}
if let Ok(content) = pdu.get_content::<ExtractRelatesTo>() {
match content.relates_to {
| Relation::Reply(in_reply_to) => {
// We need to do it again here, because replies don't have
// event_id as a top level field
if let Ok(related_pducount) =
self.get_pdu_count(&in_reply_to.in_reply_to.event_id).await
{
self.services
.pdu_metadata
.add_relation(count2, related_pducount);
}
},
| Relation::Thread(thread) => {
self.services
.threads
.add_to_thread(&thread.event_id, pdu)
.await
.inspect_err(|e| {
warn!(
"Failed to add incoming event {} to thread {}: {e:?}",
pdu.event_id(),
thread.event_id
);
})
.ok();
},
| _ => {}, // TODO: Aggregate other types
}
}
}
for appservice in self.services.appservice.read().await.values() {
if self
.services
.state_cache
.appservice_in_room(room_id, appservice)
/// Determines if an appservice is interested in a particular event.
async fn is_appservice_interested(
&self,
appservice: &RegistrationInfo,
pdu: &PduEvent,
) -> bool {
let room_id = pdu.room_id().unwrap();
let target = if *pdu.kind() == TimelineEventType::RoomMember {
pdu.state_key().and_then(|sk| UserId::parse(sk).ok())
} else {
None
};
let (target_matches_sender, target_matches_namespace) = match target {
| Some(target) => (
appservice.registration.sender_localpart.as_str() == target.localpart(),
appservice.users.is_match(target.as_str()),
),
| _ => (false, false),
};
let sender_matches_namespace = appservice.users.is_match(pdu.sender().as_str());
if target_matches_sender || target_matches_namespace || sender_matches_namespace {
return true;
}
let aliases = appservice.aliases.clone();
self.services
.alias
.local_aliases_for_room(room_id)
.ready_any(move |room_alias| aliases.is_match(room_alias.as_str()))
.await
{
self.services
.sending
.send_pdu_appservice(appservice.registration.id.clone(), pdu_id)?;
continue;
}
// If the RoomMember event has a non-empty state_key, it is targeted at someone.
// If it is our appservice user, we send this PDU to it.
if *pdu.kind() == TimelineEventType::RoomMember {
if let Some(state_key_uid) = &pdu
.state_key
.as_ref()
.and_then(|state_key| UserId::parse(state_key.as_str()).ok())
{
let appservice_uid = appservice.registration.sender_localpart.as_str();
if state_key_uid == appservice_uid {
self.services
.sending
.send_pdu_appservice(appservice.registration.id.clone(), pdu_id)?;
continue;
}
}
}
let matching_users = |users: &NamespaceRegex| {
appservice.users.is_match(pdu.sender().as_str())
|| *pdu.kind() == TimelineEventType::RoomMember
&& pdu
.state_key
.as_ref()
.is_some_and(|state_key| users.is_match(state_key))
};
let matching_aliases = |aliases: NamespaceRegex| {
self.services
.alias
.local_aliases_for_room(room_id)
.ready_any(move |room_alias| aliases.is_match(room_alias.as_str()))
};
if matching_aliases(appservice.aliases.clone()).await
|| appservice.rooms.is_match(room_id.as_str())
|| matching_users(&appservice.users)
{
self.services
.sending
.send_pdu_appservice(appservice.registration.id.clone(), pdu_id)?;
}
}
Ok(pdu_id)
/// Notifies interested appservices of a new PDU.
async fn send_to_interested_appservices(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
let interested_appservices = self
.services
.appservice
.read()
.await
.values()
.map(ToOwned::to_owned) // TODO: is this to_owned expensive?
.collect::<Vec<_>>();
interested_appservices
.stream()
.broad_filter_map(|appservice| async move {
self.is_appservice_interested(&appservice, pdu)
.await
.then_some(appservice)
})
.for_each_concurrent(None, |appservice| async move {
self.services
.sending
.send_pdu_appservice(appservice.registration.id.clone(), *pdu_id)
.inspect_err(|e| {
warn!(
"failed to send PDU {} to appservice {}: {e:?}",
pdu.event_id(),
appservice.registration.id
);
})
.ok();
})
.await;
}
}
src/service/rooms/timeline/backfill.rs
+257 -251
View File
@@ -2,7 +2,7 @@
use conduwuit::{Err, PduEvent};
use conduwuit_core::{
Result, debug, debug_warn, err, implement, info,
Result, debug, debug_warn, err, info,
matrix::{
event::Event,
pdu::{PduCount, PduId, RawPduId},
@@ -18,261 +18,267 @@
use super::ExtractBody;
#[implement(super::Service)]
#[tracing::instrument(name = "backfill", level = "trace", skip(self))]
pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Result<()> {
if self
.services
.state_cache
.room_joined_count(room_id)
.await
.is_ok_and(|count| count <= 1)
&& !self
.services
.state_accessor
.is_world_readable(room_id)
.await
{
// Room is empty (1 user or none), there is no one that can backfill
debug_warn!("Room {room_id} is empty, skipping backfill");
return Ok(());
}
let first_pdu = self
.first_item_in_room(room_id)
.await
.expect("Room is not empty");
if first_pdu.0 < from {
// No backfill required, there are still events between them
debug!("No backfill required in room {room_id}, {:?} < {from}", first_pdu.0);
return Ok(());
}
let servers = self.candidate_backfill_servers(room_id).await;
let mut federated_room = false;
for backfill_server in servers {
if !self.services.globals.server_is_ours(&backfill_server) {
federated_room = true;
}
info!("Asking {backfill_server} for backfill in {room_id}");
let response = self
.services
.sending
.send_federation_request(
&backfill_server,
federation::backfill::get_backfill::v1::Request::new(
room_id.to_owned(),
vec![first_pdu.1.event_id().to_owned()],
uint!(100),
),
)
.await;
match response {
| Ok(response) => {
for pdu in response.pdus {
if let Err(e) = self.backfill_pdu(&backfill_server, pdu).boxed().await {
debug_warn!("Failed to add backfilled pdu in room {room_id}: {e}");
}
}
return Ok(());
},
| Err(e) => {
warn!("{backfill_server} failed to provide backfill for room {room_id}: {e}");
},
}
}
if federated_room {
warn!("No servers could backfill, but backfill was needed in room {room_id}");
}
Ok(())
}
#[implement(super::Service)]
#[tracing::instrument(name = "get_remote_pdu", level = "debug", skip(self))]
pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Result<PduEvent> {
let local = self.get_pdu(event_id).await;
if local.is_ok() {
// We already have this PDU, no need to backfill
debug!("We already have {event_id} in {room_id}, no need to backfill.");
return local;
}
debug!("Preparing to fetch event {event_id} in room {room_id} from remote servers.");
// Similar to backfill_if_required, but only for a single PDU
// Fetch a list of servers to try
if self
.services
.state_cache
.room_joined_count(room_id)
.await
.is_ok_and(|count| count <= 1)
&& !self
.services
.state_accessor
.is_world_readable(room_id)
.await
{
// Room is empty (1 user or none), there is no one that can backfill
return Err!(Request(NotFound("No one can backfill this PDU, room is empty.")));
}
let servers = self.candidate_backfill_servers(room_id).await;
for backfill_server in servers {
info!("Asking {backfill_server} for event {}", event_id);
let value = self
.services
.sending
.send_federation_request(
&backfill_server,
federation::event::get_event::v1::Request::new(event_id.to_owned()),
)
.await
.and_then(|response| {
serde_json::from_str::<CanonicalJsonObject>(response.pdu.get()).map_err(|e| {
err!(BadServerResponse(debug_warn!(
"Error parsing incoming event {e:?} from {backfill_server}"
)))
})
});
let pdu = match value {
| Ok(value) => {
self.services
.event_handler
.handle_incoming_pdu(&backfill_server, room_id, event_id, value, false)
.boxed()
.await?;
debug!("Successfully backfilled {event_id} from {backfill_server}");
Some(self.get_pdu(event_id).await)
},
| Err(e) => {
warn!("{backfill_server} failed to provide backfill for room {room_id}: {e}");
None
},
};
if let Some(pdu) = pdu {
debug!("Fetched {event_id} from {backfill_server}");
return pdu;
}
}
Err!("No servers could be used to fetch {} in {}.", room_id, event_id)
}
#[implement(super::Service)]
#[tracing::instrument(skip(self, pdu), level = "debug")]
pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) -> Result<()> {
let (room_id, event_id, value) = self.services.event_handler.parse_incoming_pdu(&pdu).await?;
// Lock so we cannot backfill the same pdu twice at the same time
let mutex_lock = self
.services
.event_handler
.mutex_federation
.lock(room_id.as_str())
.await;
// Skip the PDU if we already have it as a timeline event
if let Ok(pdu_id) = self.get_pdu_id(&event_id).await {
debug!("We already know {event_id} at {pdu_id:?}");
return Ok(());
}
self.services
.event_handler
.handle_incoming_pdu(origin, &room_id, &event_id, value, false)
.boxed()
.await?;
let value = self.get_pdu_json(&event_id).await?;
let pdu = self.get_pdu(&event_id).await?;
let shortroomid = self.services.short.get_shortroomid(&room_id).await?;
let insert_lock = self.mutex_insert.lock(room_id.as_str()).await;
let count: i64 = self.services.globals.next_count().unwrap().try_into()?;
let pdu_id: RawPduId = PduId {
shortroomid,
shorteventid: PduCount::Backfilled(validated!(0 - count)),
}
.into();
// Insert pdu
self.db.prepend_backfill_pdu(&pdu_id, &event_id, &value);
drop(insert_lock);
if pdu.kind == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
self.services.search.index_pdu(shortroomid, &pdu_id, &body);
}
}
drop(mutex_lock);
debug!("Prepended backfill pdu");
Ok(())
}
#[implement(super::Service)]
async fn candidate_backfill_servers(&self, room_id: &RoomId) -> HashSet<OwnedServerName> {
let mut candidate_backfill_servers = HashSet::new();
let power_levels = self
.services
.state_accessor
.get_room_power_levels(room_id)
.await;
// Insert servers of room creators
if let Some(creators) = &power_levels.rules.privileged_creators {
for creator in creators {
candidate_backfill_servers.insert(creator.server_name().to_owned());
}
}
// Insert servers of remote users with higher-than-default PL
for (user_id, level) in &power_levels.users {
if !self.services.globals.user_is_local(user_id) && *level > power_levels.users_default {
candidate_backfill_servers.insert(user_id.server_name().to_owned());
}
}
// Insert the canonical room alias server
if let Ok(canonical_alias) = self
.services
.state_accessor
.get_canonical_alias(room_id)
.await
{
candidate_backfill_servers.insert(canonical_alias.server_name().to_owned());
}
// Insert all trusted servers in the config
candidate_backfill_servers
.extend(self.services.server.config.trusted_servers.iter().cloned());
// Remove our own name, we can't request backfill from ourselves
candidate_backfill_servers.remove(self.services.globals.server_name());
// Remove all servers that aren't in the room
for server in candidate_backfill_servers.clone() {
if !self
impl super::Service {
/// Performs backfill, if it is required.
#[tracing::instrument(name = "backfill", level = "trace", skip(self))]
pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Result<()> {
if self
.services
.state_cache
.server_in_room(&server, room_id)
.room_joined_count(room_id)
.await
.is_ok_and(|count| count <= 1)
&& !self
.services
.state_accessor
.is_world_readable(room_id)
.await
{
candidate_backfill_servers.remove(&server);
// Room is empty (1 user or none), there is no one that can backfill
debug_warn!("Room {room_id} is empty, skipping backfill");
return Ok(());
}
let first_pdu = self
.first_item_in_room(room_id)
.await
.expect("Room is not empty");
if first_pdu.0 < from {
// No backfill required, there are still events between them
debug!("No backfill required in room {room_id}, {:?} < {from}", first_pdu.0);
return Ok(());
}
let servers = self.candidate_backfill_servers(room_id).await;
let mut federated_room = false;
for backfill_server in servers {
if !self.services.globals.server_is_ours(&backfill_server) {
federated_room = true;
}
info!("Asking {backfill_server} for backfill in {room_id}");
let response = self
.services
.sending
.send_federation_request(
&backfill_server,
federation::backfill::get_backfill::v1::Request::new(
room_id.to_owned(),
vec![first_pdu.1.event_id().to_owned()],
uint!(100),
),
)
.await;
match response {
| Ok(response) => {
for pdu in response.pdus {
if let Err(e) = self.backfill_pdu(&backfill_server, pdu).boxed().await {
debug_warn!("Failed to add backfilled pdu in room {room_id}: {e}");
}
}
return Ok(());
},
| Err(e) => {
warn!("{backfill_server} failed to provide backfill for room {room_id}: {e}");
},
}
}
if federated_room {
warn!("No servers could backfill, but backfill was needed in room {room_id}");
}
Ok(())
}
debug!(?candidate_backfill_servers, "Found candidate servers for backfill");
candidate_backfill_servers
/// Fetches a single PDU from a remote server, without persisting it.
#[tracing::instrument(name = "get_remote_pdu", level = "debug", skip(self))]
pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Result<PduEvent> {
let local = self.get_pdu(event_id).await;
if local.is_ok() {
// We already have this PDU, no need to backfill
debug!("We already have {event_id} in {room_id}, no need to backfill.");
return local;
}
debug!("Preparing to fetch event {event_id} in room {room_id} from remote servers.");
// Similar to backfill_if_required, but only for a single PDU
// Fetch a list of servers to try
if self
.services
.state_cache
.room_joined_count(room_id)
.await
.is_ok_and(|count| count <= 1)
&& !self
.services
.state_accessor
.is_world_readable(room_id)
.await
{
// Room is empty (1 user or none), there is no one that can backfill
return Err!(Request(NotFound("No one can backfill this PDU, room is empty.")));
}
let servers = self.candidate_backfill_servers(room_id).await;
for backfill_server in servers {
info!("Asking {backfill_server} for event {}", event_id);
let value = self
.services
.sending
.send_federation_request(
&backfill_server,
federation::event::get_event::v1::Request::new(event_id.to_owned()),
)
.await
.and_then(|response| {
serde_json::from_str::<CanonicalJsonObject>(response.pdu.get()).map_err(|e| {
err!(BadServerResponse(debug_warn!(
"Error parsing incoming event {e:?} from {backfill_server}"
)))
})
});
let pdu = match value {
| Ok(value) => {
self.services
.event_handler
.handle_incoming_pdu(&backfill_server, room_id, event_id, value, false)
.boxed()
.await?;
debug!("Successfully backfilled {event_id} from {backfill_server}");
Some(self.get_pdu(event_id).await)
},
| Err(e) => {
warn!("{backfill_server} failed to provide backfill for room {room_id}: {e}");
None
},
};
if let Some(pdu) = pdu {
debug!("Fetched {event_id} from {backfill_server}");
return pdu;
}
}
Err!("No servers could be used to fetch {} in {}.", room_id, event_id)
}
/// Backfills a single PDU.
#[tracing::instrument(skip(self, pdu), level = "debug")]
pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) -> Result<()> {
let (room_id, event_id, value) =
self.services.event_handler.parse_incoming_pdu(&pdu).await?;
// Lock so we cannot backfill the same pdu twice at the same time
let mutex_lock = self
.services
.event_handler
.mutex_federation
.lock(room_id.as_str())
.await;
// Skip the PDU if we already have it as a timeline event
if let Ok(pdu_id) = self.get_pdu_id(&event_id).await {
debug!("We already know {event_id} at {pdu_id:?}");
return Ok(());
}
self.services
.event_handler
.handle_incoming_pdu(origin, &room_id, &event_id, value, false)
.boxed()
.await?;
let value = self.get_pdu_json(&event_id).await?;
let pdu = self.get_pdu(&event_id).await?;
let shortroomid = self.services.short.get_shortroomid(&room_id).await?;
let insert_lock = self.mutex_insert.lock(room_id.as_str()).await;
let count: i64 = self.services.globals.next_count().unwrap().try_into()?;
let pdu_id: RawPduId = PduId {
shortroomid,
shorteventid: PduCount::Backfilled(validated!(0 - count)),
}
.into();
// Insert pdu
self.db.prepend_backfill_pdu(&pdu_id, &event_id, &value);
drop(insert_lock);
if pdu.kind == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
self.services.search.index_pdu(shortroomid, &pdu_id, &body);
}
}
drop(mutex_lock);
debug!("Prepended backfill pdu");
Ok(())
}
/// Determines which servers are trusted enough to provide backfill in a
/// room.
async fn candidate_backfill_servers(&self, room_id: &RoomId) -> HashSet<OwnedServerName> {
let mut candidate_backfill_servers = HashSet::new();
let power_levels = self
.services
.state_accessor
.get_room_power_levels(room_id)
.await;
// Insert servers of room creators
if let Some(creators) = &power_levels.rules.privileged_creators {
for creator in creators {
candidate_backfill_servers.insert(creator.server_name().to_owned());
}
}
// Insert servers of remote users with higher-than-default PL
for (user_id, level) in &power_levels.users {
if !self.services.globals.user_is_local(user_id)
&& *level > power_levels.users_default
{
candidate_backfill_servers.insert(user_id.server_name().to_owned());
}
}
// Insert the canonical room alias server
if let Ok(canonical_alias) = self
.services
.state_accessor
.get_canonical_alias(room_id)
.await
{
candidate_backfill_servers.insert(canonical_alias.server_name().to_owned());
}
// Insert all trusted servers in the config
candidate_backfill_servers
.extend(self.services.server.config.trusted_servers.iter().cloned());
// Remove our own name, we can't request backfill from ourselves
candidate_backfill_servers.remove(self.services.globals.server_name());
// Remove all servers that aren't in the room
for server in candidate_backfill_servers.clone() {
if !self
.services
.state_cache
.server_in_room(&server, room_id)
.await
{
candidate_backfill_servers.remove(&server);
}
}
debug!(?candidate_backfill_servers, "Found candidate servers for backfill");
candidate_backfill_servers
}
}
src/service/rooms/timeline/build.rs
+234 -232
View File
@@ -2,13 +2,13 @@
use conduwuit::trace;
use conduwuit_core::{
Err, Result, implement,
Err, Result,
matrix::{event::Event, pdu::PartialPdu},
utils::{IterStream, ReadyExt},
};
use futures::{FutureExt, StreamExt};
use ruma::{
OwnedEventId, OwnedServerName, RoomId, RoomVersionId, UserId,
OwnedEventId, OwnedServerName, RoomId, UserId,
events::{
TimelineEventType,
room::{
@@ -20,246 +20,248 @@
use super::{ExtractBody, RoomMutexGuard};
/// Creates a new persisted data unit and adds it to a room. This function
/// takes a roomid_mutex_state, meaning that only this function is able to
/// mutate the room state.
#[implement(super::Service)]
#[tracing::instrument(skip(self, state_lock, partial_pdu), level = "trace")]
pub async fn build_and_append_pdu(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
state_lock: &RoomMutexGuard,
) -> Result<OwnedEventId> {
let (pdu, pdu_json) = self
.create_hash_and_sign_event(partial_pdu, sender, room_id, state_lock)
.await?;
impl super::Service {
/// Creates a new persisted data unit and adds it to a room. This function
/// takes a roomid_mutex_state, meaning that only this function is able to
/// mutate the room state.
#[tracing::instrument(skip(self, state_lock, partial_pdu), level = "trace")]
pub async fn build_and_append_pdu(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
state_lock: &RoomMutexGuard,
) -> Result<OwnedEventId> {
let (pdu, pdu_json) = self
.create_hash_and_sign_event(partial_pdu, sender, room_id, state_lock)
.await?;
let room_id = pdu.room_id_or_hash();
if self.services.admin.is_admin_room(&room_id).await {
self.check_pdu_for_admin_room(&pdu, sender).boxed().await?;
}
let room_id = pdu.room_id_or_hash();
if self.services.admin.is_admin_room(&room_id).await {
self.check_pdu_for_admin_room(&pdu, sender).boxed().await?;
}
// If redaction event is not authorized, do not append it to the timeline
if *pdu.kind() == TimelineEventType::RoomRedaction {
use RoomVersionId::*;
trace!("Running redaction checks for room {room_id}");
match self.services.state.get_room_version(&room_id).await? {
| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
if let Some(redact_id) = pdu.redacts() {
if !self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), &room_id, false)
.await?
{
return Err!(Request(Forbidden("User cannot redact this event.")));
// If redaction event is not authorized, do not append it to the timeline
if *pdu.kind() == TimelineEventType::RoomRedaction {
use ruma::RoomVersionId::*;
trace!("Running redaction checks for room {room_id}");
match self.services.state.get_room_version(&room_id).await? {
| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
if let Some(redact_id) = pdu.redacts() {
if !self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), &room_id, false)
.await?
{
return Err!(Request(Forbidden("User cannot redact this event.")));
}
}
}
},
| _ => {
let content: RoomRedactionEventContent = pdu.get_content()?;
if let Some(redact_id) = &content.redacts {
if !self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), &room_id, false)
.await?
{
return Err!(Request(Forbidden("User cannot redact this event.")));
},
| _ => {
let content: RoomRedactionEventContent = pdu.get_content()?;
if let Some(redact_id) = &content.redacts {
if !self
.services
.state_accessor
.user_can_redact(redact_id, pdu.sender(), &room_id, false)
.await?
{
return Err!(Request(Forbidden("User cannot redact this event.")));
}
}
}
},
}
}
if *pdu.kind() == TimelineEventType::RoomMember {
trace!("Running room member checks for room {room_id}");
let content: RoomMemberEventContent = pdu.get_content()?;
if content.join_authorized_via_users_server.is_some()
&& content.membership != MembershipState::Join
{
return Err!(Request(BadJson(
"join_authorised_via_users_server is only for member joins"
)));
}
if content
.join_authorized_via_users_server
.as_ref()
.is_some_and(|authorising_user| {
!self.services.globals.user_is_local(authorising_user)
}) {
return Err!(Request(InvalidParam(
"Authorising user does not belong to this homeserver"
)));
}
}
if *pdu.kind() == TimelineEventType::RoomCreate {
trace!("Creating shortroomid for {room_id}");
self.services
.short
.get_or_create_shortroomid(&room_id)
.await;
}
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
trace!("Appending {} state for room {room_id}", pdu.event_id());
let statehashid = self.services.state.append_to_state(&pdu, &room_id).await?;
trace!("State hash ID for {room_id}: {statehashid:?}");
trace!("Generating raw ID for PDU {}", pdu.event_id());
let pdu_id = self
.append_pdu(
&pdu,
pdu_json,
// Since this PDU references all pdu_leaves we can update the leaves
// of the room
once(pdu.event_id()),
state_lock,
&room_id,
)
.boxed()
.await?;
// Process admin commands for locally sent events
if *pdu.kind() == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
if let Some(source) = self
.services
.admin
.is_admin_command(&pdu, &body, true)
.await
{
self.services.admin.command_with_sender(
body,
Some(pdu.event_id().into()),
source,
pdu.sender.clone(),
)?;
},
}
}
}
// We set the room state after inserting the pdu, so that we never have a moment
// in time where events in the current room state do not exist
trace!("Setting room state for room {room_id}");
self.services
.state
.set_room_state(&room_id, statehashid, state_lock);
let mut servers: HashSet<OwnedServerName> = self
.services
.state_cache
.room_servers(&room_id)
.collect()
.await;
// In case we are kicking or banning a user, we need to inform their server of
// the change
if *pdu.kind() == TimelineEventType::RoomMember {
if let Some(state_key_uid) = &pdu
.state_key
.as_ref()
.and_then(|state_key| UserId::parse(state_key.as_str()).ok())
{
servers.insert(state_key_uid.server_name().to_owned());
}
}
// Remove our server from the server list since it will be added to it by
// room_servers() and/or the if statement above
servers.remove(self.services.globals.server_name());
trace!("Sending PDU {} to {} servers", pdu.event_id(), servers.len());
self.services
.sending
.send_pdu_servers(servers.stream(), &pdu_id)
.await?;
trace!("Event {} in room {:?} has been appended", pdu.event_id(), room_id);
Ok(pdu.event_id().to_owned())
}
/// Assert invariants about the admin room, to prevent (for example) all admins
/// from leaving or being banned from the room
#[implement(super::Service)]
#[tracing::instrument(skip_all, level = "debug")]
async fn check_pdu_for_admin_room<Pdu>(&self, pdu: &Pdu, sender: &UserId) -> Result
where
Pdu: Event + Send + Sync,
{
match pdu.kind() {
| TimelineEventType::RoomEncryption => {
return Err!(Request(Forbidden(error!("Encryption not supported in admins room."))));
},
| TimelineEventType::RoomMember => {
let target = pdu
.state_key()
.filter(|v| v.starts_with('@'))
.unwrap_or(sender.as_str());
let server_user = &self.services.globals.server_user.to_string();
if *pdu.kind() == TimelineEventType::RoomMember {
trace!("Running room member checks for room {room_id}");
let content: RoomMemberEventContent = pdu.get_content()?;
match content.membership {
| MembershipState::Leave => {
if target == server_user {
return Err!(Request(Forbidden(error!(
"Server user cannot leave the admins room."
))));
}
let count = self
.services
.state_cache
.room_members(&pdu.room_id_or_hash())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()
.count()
.await;
if count < 2 {
return Err!(Request(Forbidden(error!(
"Last admin cannot leave the admins room."
))));
}
},
| MembershipState::Ban if pdu.state_key().is_some() => {
if target == server_user {
return Err!(Request(Forbidden(error!(
"Server cannot be banned from admins room."
))));
}
let count = self
.services
.state_cache
.room_members(&pdu.room_id_or_hash())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()
.count()
.await;
if count < 2 {
return Err!(Request(Forbidden(error!(
"Last admin cannot be banned from admins room."
))));
}
},
| _ => {},
if content.join_authorized_via_users_server.is_some()
&& content.membership != MembershipState::Join
{
return Err!(Request(BadJson(
"join_authorised_via_users_server is only for member joins"
)));
}
},
| _ => {},
if content
.join_authorized_via_users_server
.as_ref()
.is_some_and(|authorising_user| {
!self.services.globals.user_is_local(authorising_user)
}) {
return Err!(Request(InvalidParam(
"Authorising user does not belong to this homeserver"
)));
}
}
if *pdu.kind() == TimelineEventType::RoomCreate {
trace!("Creating shortroomid for {room_id}");
self.services
.short
.get_or_create_shortroomid(&room_id)
.await;
}
// We append to state before appending the pdu, so we don't have a moment in
// time with the pdu without it's state. This is okay because append_pdu can't
// fail.
trace!("Appending {} state for room {room_id}", pdu.event_id());
let statehashid = self.services.state.append_to_state(&pdu, &room_id).await?;
trace!("State hash ID for {room_id}: {statehashid:?}");
trace!("Generating raw ID for PDU {}", pdu.event_id());
let pdu_id = self
.append_pdu(
&pdu,
pdu_json,
// Since this PDU references all pdu_leaves we can update the leaves
// of the room
once(pdu.event_id()),
state_lock,
&room_id,
)
.boxed()
.await?;
// Process admin commands for locally sent events
if *pdu.kind() == TimelineEventType::RoomMessage {
let content: ExtractBody = pdu.get_content()?;
if let Some(body) = content.body {
if let Some(source) = self
.services
.admin
.is_admin_command(&pdu, &body, true)
.await
{
self.services.admin.command_with_sender(
body,
Some(pdu.event_id().into()),
source,
pdu.sender.clone(),
)?;
}
}
}
// We set the room state after inserting the pdu, so that we never have a moment
// in time where events in the current room state do not exist
trace!("Setting room state for room {room_id}");
self.services
.state
.set_room_state(&room_id, statehashid, state_lock);
let mut servers: HashSet<OwnedServerName> = self
.services
.state_cache
.room_servers(&room_id)
.collect()
.await;
// In case we are kicking or banning a user, we need to inform their server of
// the change
if *pdu.kind() == TimelineEventType::RoomMember {
if let Some(state_key_uid) = &pdu
.state_key
.as_ref()
.and_then(|state_key| UserId::parse(state_key.as_str()).ok())
{
servers.insert(state_key_uid.server_name().to_owned());
}
}
// Remove our server from the server list since it will be added to it by
// room_servers() and/or the if statement above
servers.remove(self.services.globals.server_name());
trace!("Sending PDU {} to {} servers", pdu.event_id(), servers.len());
self.services
.sending
.send_pdu_servers(servers.stream(), &pdu_id)
.await?;
trace!("Event {} in room {:?} has been appended", pdu.event_id(), room_id);
Ok(pdu.event_id().to_owned())
}
Ok(())
/// Assert invariants about the admin room, to prevent (for example) all
/// admins from leaving or being banned from the room
#[tracing::instrument(skip_all, level = "debug")]
async fn check_pdu_for_admin_room<Pdu>(&self, pdu: &Pdu, sender: &UserId) -> Result
where
Pdu: Event + Send + Sync,
{
match pdu.kind() {
| TimelineEventType::RoomEncryption => {
return Err!(Request(Forbidden(error!(
"Encryption not supported in admins room."
))));
},
| TimelineEventType::RoomMember => {
let target = pdu
.state_key()
.filter(|v| v.starts_with('@'))
.unwrap_or(sender.as_str());
let server_user = &self.services.globals.server_user.to_string();
let content: RoomMemberEventContent = pdu.get_content()?;
match content.membership {
| MembershipState::Leave => {
if target == server_user {
return Err!(Request(Forbidden(error!(
"Server user cannot leave the admins room."
))));
}
let count = self
.services
.state_cache
.room_members(&pdu.room_id_or_hash())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()
.count()
.await;
if count < 2 {
return Err!(Request(Forbidden(error!(
"Last admin cannot leave the admins room."
))));
}
},
| MembershipState::Ban if pdu.state_key().is_some() => {
if target == server_user {
return Err!(Request(Forbidden(error!(
"Server cannot be banned from admins room."
))));
}
let count = self
.services
.state_cache
.room_members(&pdu.room_id_or_hash())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()
.count()
.await;
if count < 2 {
return Err!(Request(Forbidden(error!(
"Last admin cannot be banned from admins room."
))));
}
},
| _ => {},
}
},
| _ => {},
}
Ok(())
}
}
src/service/rooms/timeline/create.rs
+270 -258
View File
@@ -2,7 +2,7 @@
use conduwuit::{smallstr::SmallString, trace};
use conduwuit_core::{
Err, Error, Result, err, implement,
Err, Error, Result, err,
matrix::{
event::{Event, gen_event_id},
pdu::{EventHash, PartialPdu, PduEvent},
@@ -73,277 +73,289 @@ fn room_version_from_event(
}
}
// Creates an event, but does not hash or sign it.
#[implement(super::Service)]
pub async fn create_event(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
_mutex_lock: &RoomMutexGuard,
) -> Result<(PduEvent, RoomVersionRules)> {
let PartialPdu {
event_type,
content,
unsigned,
state_key,
redacts,
timestamp,
} = partial_pdu;
impl super::Service {
/// Creates a new PDU and runs an auth check before returning it. Does not
/// hash or sign the PDU, meaning it can be mutated after.
pub async fn create_event(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
_mutex_lock: &RoomMutexGuard,
) -> Result<(PduEvent, RoomVersionRules)> {
let PartialPdu {
event_type,
content,
unsigned,
state_key,
redacts,
timestamp,
} = partial_pdu;
trace!(
"Creating event of type {} in room {}",
event_type,
room_id.as_ref().map_or("None", |id| id.as_str())
);
let room_version = match room_id {
| Some(room_id) => {
trace!(%room_id, "Looking up existing room ID");
self.services
.state
.get_room_version(room_id)
trace!(
"Creating event of type {} in room {}",
event_type,
room_id.as_ref().map_or("None", |id| id.as_str())
);
let room_version = match room_id {
| Some(room_id) => {
trace!(%room_id, "Looking up existing room ID");
self.services
.state
.get_room_version(room_id)
.await
.or_else(|_| {
room_version_from_event(
room_id.to_owned(),
&event_type.clone(),
&content.clone(),
)
})?
},
| None => {
trace!("No room ID, assuming room creation");
room_version_from_event(
RoomId::new_v1(self.services.globals.server_name()),
&event_type.clone(),
&content.clone(),
)?
},
};
let Some(room_version_rules) = room_version.rules() else {
return Err!(Request(UnsupportedRoomVersion("Unsupported room version")));
};
let prev_events: Vec<OwnedEventId> = match room_id {
| Some(room_id) =>
self.services
.state
.get_forward_extremities(room_id)
.take(20)
.map(Into::into)
.collect()
.await,
| None => Vec::new(),
};
let auth_events: HashMap<(StateEventType, SmallString<[u8; 48]>), PduEvent> =
match room_id {
| Some(room_id) =>
self.services
.state
.get_auth_events(
room_id,
&event_type,
sender,
state_key.as_deref(),
&content,
&room_version_rules,
)
.await?,
| None => HashMap::new(),
};
// Our depth is the maximum depth of prev_events + 1
let depth = match room_id {
| Some(_) => prev_events
.iter()
.stream()
.map(Ok)
.and_then(|event_id| self.get_pdu(event_id))
.and_then(|pdu| future::ok(pdu.depth))
.ignore_err()
.ready_fold(uint!(0), cmp::max)
.await
.or_else(|_| {
room_version_from_event(
room_id.to_owned(),
&event_type.clone(),
&content.clone(),
)
})?
},
| None => {
trace!("No room ID, assuming room creation");
room_version_from_event(
RoomId::new_v1(self.services.globals.server_name()),
&event_type.clone(),
&content.clone(),
)?
},
};
.saturating_add(uint!(1)),
| None => uint!(1),
};
let Some(room_version_rules) = room_version.rules() else {
return Err!(Request(UnsupportedRoomVersion("Unsupported room version")));
};
let mut unsigned = unsigned.unwrap_or_default();
let prev_events: Vec<OwnedEventId> = match room_id {
| Some(room_id) =>
self.services
.state
.get_forward_extremities(room_id)
.take(20)
.map(Into::into)
.collect()
.await,
| None => Vec::new(),
};
let auth_events: HashMap<(StateEventType, SmallString<[u8; 48]>), PduEvent> = match room_id {
| Some(room_id) =>
self.services
.state
.get_auth_events(
room_id,
&event_type,
sender,
state_key.as_deref(),
&content,
&room_version_rules,
)
.await?,
| None => HashMap::new(),
};
// Our depth is the maximum depth of prev_events + 1
let depth = match room_id {
| Some(_) => prev_events
.iter()
.stream()
.map(Ok)
.and_then(|event_id| self.get_pdu(event_id))
.and_then(|pdu| future::ok(pdu.depth))
.ignore_err()
.ready_fold(uint!(0), cmp::max)
.await
.saturating_add(uint!(1)),
| None => uint!(1),
};
let mut unsigned = unsigned.unwrap_or_default();
if let Some(room_id) = room_id {
if let Some(state_key) = &state_key {
if let Ok(prev_pdu) = self
.services
.state_accessor
.room_state_get(room_id, &event_type.clone().to_string().into(), state_key)
.await
{
unsigned.insert("prev_content".to_owned(), prev_pdu.get_content_as_value());
unsigned
.insert("prev_sender".to_owned(), serde_json::to_value(prev_pdu.sender())?);
unsigned.insert(
"replaces_state".to_owned(),
serde_json::to_value(prev_pdu.event_id())?,
);
if let Some(room_id) = room_id {
if let Some(state_key) = &state_key {
if let Ok(prev_pdu) = self
.services
.state_accessor
.room_state_get(room_id, &event_type.clone().to_string().into(), state_key)
.await
{
unsigned.insert("prev_content".to_owned(), prev_pdu.get_content_as_value());
unsigned.insert(
"prev_sender".to_owned(),
serde_json::to_value(prev_pdu.sender())?,
);
unsigned.insert(
"replaces_state".to_owned(),
serde_json::to_value(prev_pdu.event_id())?,
);
}
}
}
}
let pdu = PduEvent {
event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
room_id: room_id.map(ToOwned::to_owned),
sender: sender.to_owned(),
origin: None,
origin_server_ts: timestamp.map_or_else(
|| {
utils::millis_since_unix_epoch()
.try_into()
.expect("u64 fits into UInt")
let pdu = PduEvent {
event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
room_id: room_id.map(ToOwned::to_owned),
sender: sender.to_owned(),
origin: None,
origin_server_ts: timestamp.map_or_else(
|| {
utils::millis_since_unix_epoch()
.try_into()
.expect("u64 fits into UInt")
},
|ts| ts.get(),
),
kind: event_type,
content,
state_key,
prev_events,
depth,
auth_events: auth_events
.values()
.map(|pdu| pdu.event_id.clone())
.collect(),
redacts,
unsigned: if unsigned.is_empty() {
None
} else {
Some(to_raw_value(&unsigned)?)
},
|ts| ts.get(),
),
kind: event_type,
content,
state_key,
prev_events,
depth,
auth_events: auth_events
.values()
.map(|pdu| pdu.event_id.clone())
.collect(),
redacts,
unsigned: if unsigned.is_empty() {
None
} else {
Some(to_raw_value(&unsigned)?)
},
hashes: EventHash { sha256: String::new() },
signatures: None,
};
let auth_fetch = |k: &StateEventType, s: &str| {
let key = (k.clone(), s.into());
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
let room_id_or_hash = pdu.room_id_or_hash();
let create_pdu = match &pdu.kind {
| TimelineEventType::RoomCreate => None,
| _ => Some(
self.services
.state_accessor
.room_state_get(&room_id_or_hash, &StateEventType::RoomCreate, "")
.await
.map_err(|e| {
err!(Request(Forbidden(warn!("Failed to fetch room create event: {e}"))))
})?,
),
};
let create_event = match &pdu.kind {
| TimelineEventType::RoomCreate => &pdu,
| _ => create_pdu.as_ref().unwrap().as_pdu(),
};
let auth_check = state_res::auth_check(
&room_version_rules,
&pdu,
None, // TODO: third_party_invite
auth_fetch,
create_event,
)
.await
.map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?;
if !auth_check {
return Err!(Request(Forbidden("Event is not authorized.")));
}
trace!(
"Event {} in room {} is authorized",
pdu.event_id,
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
Ok((pdu, room_version_rules))
}
#[implement(super::Service)]
pub async fn create_hash_and_sign_event(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
mutex_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room
* state mutex */
) -> Result<(PduEvent, CanonicalJsonObject)> {
if !self.services.globals.user_is_local(sender) {
return Err!(Request(Forbidden("Sender must be a local user")));
}
let (mut pdu, room_version_rules) = self
.create_event(partial_pdu, sender, room_id, mutex_lock)
.await?;
// Hash and sign
let mut pdu_json = utils::to_canonical_object(&pdu).map_err(|e| {
err!(Request(BadJson(warn!("Failed to convert PDU to canonical JSON: {e}"))))
})?;
pdu_json.remove("event_id");
trace!("hashing and signing event {}", pdu.event_id);
if let Err(e) = self
.services
.server_keys
.hash_and_sign_event(&mut pdu_json, &room_version_rules)
{
return match e {
| Error::SignatureJson(ruma::signatures::JsonError::PduTooLarge) => {
Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")))
},
| _ => Err!(Request(Unknown(warn!("Signing event failed: {e}")))),
hashes: EventHash { sha256: String::new() },
signatures: None,
};
}
// Generate event id
pdu.event_id = gen_event_id(&pdu_json, &room_version_rules)?;
pdu_json.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
// Verify that the *full* PDU isn't over 64KiB.
if !pdu_fits(&mut pdu_json.clone()) {
// feckin huge PDU mate
return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
}
// Check with the policy server
if room_id.is_some() {
let auth_fetch = |k: &StateEventType, s: &str| {
let key = (k.clone(), s.into());
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
let room_id_or_hash = pdu.room_id_or_hash();
let create_pdu = match &pdu.kind {
| TimelineEventType::RoomCreate => None,
| _ => Some(
self.services
.state_accessor
.room_state_get(&room_id_or_hash, &StateEventType::RoomCreate, "")
.await
.map_err(|e| {
err!(Request(Forbidden(warn!("Failed to fetch room create event: {e}"))))
})?,
),
};
let create_event = match &pdu.kind {
| TimelineEventType::RoomCreate => &pdu,
| _ => create_pdu.as_ref().unwrap().as_pdu(),
};
let auth_check = state_res::auth_check(
&room_version_rules,
&pdu,
None, // TODO: third_party_invite
auth_fetch,
create_event,
)
.await
.map_err(|e| err!(Request(Forbidden(warn!("Auth check failed: {e:?}")))))?;
if !auth_check {
return Err!(Request(Forbidden("Event is not authorized.")));
}
trace!(
"Checking event in room {} with policy server",
"Event {} in room {} is authorized",
pdu.event_id,
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
// We need to remove the event ID before getting a PS signature on the event.
// Note that we seemingly pointlessly add it above just to remove it here, but
// it's important to make sure the event ID isn't the field that makes the
// difference between an illegally-large event and one that is okay.
pdu_json.remove("event_id");
self.services
.event_handler
.policy_server_allows_event(
&pdu,
&mut pdu_json,
pdu.room_id().expect("has room ID"),
&room_version_rules,
false,
)
.await?;
pdu_json
.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
Ok((pdu, room_version_rules))
}
// Generate short event id
trace!(
"Generating short event ID for {} in room {}",
pdu.event_id,
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
let _shorteventid = self
.services
.short
.get_or_create_shorteventid(&pdu.event_id)
.await;
/// Creates, checks, hashes, and signs a new PDU. The resulting PDU is
/// immutable. Since the PDU is immutable, the `event_id` field is
/// populated.
///
/// TODO: The `event_id` field should be a separate return option, not
/// embedded.
pub async fn create_hash_and_sign_event(
&self,
partial_pdu: PartialPdu,
sender: &UserId,
room_id: Option<&RoomId>,
mutex_lock: &RoomMutexGuard,
) -> Result<(PduEvent, CanonicalJsonObject)> {
if !self.services.globals.user_is_local(sender) {
return Err!(Request(Forbidden("Sender must be a local user")));
}
let (mut pdu, room_version_rules) = self
.create_event(partial_pdu, sender, room_id, mutex_lock)
.await?;
// Hash and sign
let mut pdu_json = utils::to_canonical_object(&pdu).map_err(|e| {
err!(Request(BadJson(warn!("Failed to convert PDU to canonical JSON: {e}"))))
})?;
pdu_json.remove("event_id");
trace!("New PDU created: {pdu:?}");
Ok((pdu, pdu_json))
trace!("hashing and signing event {}", pdu.event_id);
if let Err(e) = self
.services
.server_keys
.hash_and_sign_event(&mut pdu_json, &room_version_rules)
{
return match e {
| Error::SignatureJson(ruma::signatures::JsonError::PduTooLarge) => {
Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")))
},
| _ => Err!(Request(Unknown(warn!("Signing event failed: {e}")))),
};
}
// Generate event id
pdu.event_id = gen_event_id(&pdu_json, &room_version_rules)?;
pdu_json
.insert("event_id".into(), CanonicalJsonValue::String(pdu.event_id.clone().into()));
// Verify that the *full* PDU isn't over 64KiB.
if !pdu_fits(&mut pdu_json.clone()) {
// feckin huge PDU mate
return Err!(Request(TooLarge("Message/PDU is too long (exceeds 65535 bytes)")));
}
// Check with the policy server
if room_id.is_some() {
trace!(
"Checking event in room {} with policy server",
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
// We need to remove the event ID before getting a PS signature on the event.
// Note that we seemingly pointlessly add it above just to remove it here, but
// it's important to make sure the event ID isn't the field that makes the
// difference between an illegally-large event and one that is okay.
pdu_json.remove("event_id");
self.services
.event_handler
.policy_server_allows_event(
&pdu,
&mut pdu_json,
pdu.room_id().expect("has room ID"),
&room_version_rules,
false,
)
.await?;
pdu_json.insert(
"event_id".into(),
CanonicalJsonValue::String(pdu.event_id.clone().into()),
);
}
// Generate short event id
trace!(
"Generating short event ID for {} in room {}",
pdu.event_id,
pdu.room_id.as_ref().map_or("None", |id| id.as_str())
);
let _shorteventid = self
.services
.short
.get_or_create_shorteventid(&pdu.event_id)
.await;
trace!("New PDU created: {pdu:?}");
Ok((pdu, pdu_json))
}
}
src/service/rooms/timeline/redact.rs
+42 -41
View File
@@ -1,5 +1,5 @@
use conduwuit_core::{
Result, err, implement,
Result, err,
matrix::event::Event,
utils::{self},
};
@@ -8,48 +8,49 @@
use super::ExtractBody;
use crate::rooms::short::ShortRoomId;
/// Replace a PDU with the redacted form.
#[implement(super::Service)]
#[tracing::instrument(name = "redact", level = "debug", skip(self))]
pub async fn redact_pdu<Pdu: Event + Send + Sync>(
&self,
event_id: &EventId,
reason: &Pdu,
shortroomid: ShortRoomId,
) -> Result {
// TODO: Don't reserialize, keep original json
let Ok(pdu_id) = self.get_pdu_id(event_id).await else {
// If event does not exist, just noop
return Ok(());
};
impl super::Service {
/// Replace a PDU with the redacted form.
#[tracing::instrument(name = "redact", level = "debug", skip(self))]
pub async fn redact_pdu<Pdu: Event + Send + Sync>(
&self,
event_id: &EventId,
reason: &Pdu,
shortroomid: ShortRoomId,
) -> Result {
// TODO: Don't reserialize, keep original json
let Ok(pdu_id) = self.get_pdu_id(event_id).await else {
// If event does not exist, just noop
return Ok(());
};
let mut pdu = self
.get_pdu_from_id(&pdu_id)
.await
.map(Event::into_pdu)
.map_err(|e| {
err!(Database(error!(?pdu_id, %event_id, ?e, "PDU ID points to invalid PDU.")))
let mut pdu = self
.get_pdu_from_id(&pdu_id)
.await
.map(Event::into_pdu)
.map_err(|e| {
err!(Database(error!(?pdu_id, %event_id, ?e, "PDU ID points to invalid PDU.")))
})?;
if let Ok(content) = pdu.get_content::<ExtractBody>() {
if let Some(body) = content.body {
self.services
.search
.deindex_pdu(shortroomid, &pdu_id, &body);
}
}
let room_version_id = self
.services
.state
.get_room_version(&pdu.room_id_or_hash())
.await?;
pdu.redact(&room_version_id, reason.to_value())?;
let obj = utils::to_canonical_object(&pdu).map_err(|e| {
err!(Database(error!(%event_id, ?e, "Failed to convert PDU to canonical JSON")))
})?;
if let Ok(content) = pdu.get_content::<ExtractBody>() {
if let Some(body) = content.body {
self.services
.search
.deindex_pdu(shortroomid, &pdu_id, &body);
}
self.replace_pdu(&pdu_id, &obj).await
}
let room_version_id = self
.services
.state
.get_room_version(&pdu.room_id_or_hash())
.await?;
pdu.redact(&room_version_id, reason.to_value())?;
let obj = utils::to_canonical_object(&pdu).map_err(|e| {
err!(Database(error!(%event_id, ?e, "Failed to convert PDU to canonical JSON")))
})?;
self.replace_pdu(&pdu_id, &obj).await
}
src/service/rooms/user/mod.rs
+44 -41
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use database::{Deserialized, Map};
use ruma::{RoomId, UserId};
@@ -38,48 +38,51 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
}
#[implement(Service)]
pub fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) {
let userroom_id = (user_id, room_id);
self.db.userroomid_highlightcount.put(userroom_id, 0_u64);
self.db.userroomid_notificationcount.put(userroom_id, 0_u64);
impl Service {
/// Resets the notification counts for a room the user is in.
pub fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) {
let userroom_id = (user_id, room_id);
self.db.userroomid_highlightcount.put(userroom_id, 0_u64);
self.db.userroomid_notificationcount.put(userroom_id, 0_u64);
let roomuser_id = (room_id, user_id);
let count = self.services.globals.next_count().unwrap();
self.db
.roomuserid_lastnotificationread
.put(roomuser_id, count);
}
let roomuser_id = (room_id, user_id);
let count = self.services.globals.next_count().unwrap();
self.db
.roomuserid_lastnotificationread
.put(roomuser_id, count);
}
#[implement(Service)]
pub async fn notification_count(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (user_id, room_id);
self.db
.userroomid_notificationcount
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
}
/// Gets the notification count for a room the user is in.
pub async fn notification_count(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (user_id, room_id);
self.db
.userroomid_notificationcount
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
}
#[implement(Service)]
pub async fn highlight_count(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (user_id, room_id);
self.db
.userroomid_highlightcount
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
}
/// Gets the number of events that highlighted the user in a given room.
/// These aren't necessarily notifications.
pub async fn highlight_count(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (user_id, room_id);
self.db
.userroomid_highlightcount
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
}
#[implement(Service)]
pub async fn last_notification_read(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (room_id, user_id);
self.db
.roomuserid_lastnotificationread
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
/// Returns the last notification the user read in the room, or 0 if none.
pub async fn last_notification_read(&self, user_id: &UserId, room_id: &RoomId) -> u64 {
let key = (room_id, user_id);
self.db
.roomuserid_lastnotificationread
.qry(&key)
.await
.deserialized()
.unwrap_or(0)
}
}