From 199f6eb67b29c3af5ce2f0d61cf0f4420422c1ca Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 03:06:03 +0100 Subject: [PATCH] style: Document & refactor rooms/user and rooms/timeline services --- src/service/rooms/membership/mod.rs | 3 +- src/service/rooms/timeline/append.rs | 701 ++++++++++++++----------- src/service/rooms/timeline/backfill.rs | 508 +++++++++--------- src/service/rooms/timeline/build.rs | 466 ++++++++-------- src/service/rooms/timeline/create.rs | 528 ++++++++++--------- src/service/rooms/timeline/redact.rs | 83 +-- src/service/rooms/user/mod.rs | 85 +-- 7 files changed, 1241 insertions(+), 1133 deletions(-) diff --git a/src/service/rooms/membership/mod.rs b/src/service/rooms/membership/mod.rs index ff69e095e..0cbad3509 100644 --- a/src/service/rooms/membership/mod.rs +++ b/src/service/rooms/membership/mod.rs @@ -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")] diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 8f999a737..76e19f10f 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -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, - soft_fail: bool, - state_lock: &'a RoomMutexGuard, - room_id: &'a ruma::RoomId, -) -> Result> -where - Leaves: Iterator + 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, + soft_fail: bool, + state_lock: &'a RoomMutexGuard, + room_id: &'a ruma::RoomId, + ) -> Result> + where + Leaves: Iterator + 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 -where - Leaves: Iterator + 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 + where + Leaves: Iterator + 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::() { - 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::() { - 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::() { + 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::() { + 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::>(); + 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; + } } diff --git a/src/service/rooms/timeline/backfill.rs b/src/service/rooms/timeline/backfill.rs index 9930cf466..26c36bf69 100644 --- a/src/service/rooms/timeline/backfill.rs +++ b/src/service/rooms/timeline/backfill.rs @@ -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 { - 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::(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) -> 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 { - 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 { + 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::(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) -> 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 { + 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 + } } diff --git a/src/service/rooms/timeline/build.rs b/src/service/rooms/timeline/build.rs index 85f1f88d9..8ff04b024 100644 --- a/src/service/rooms/timeline/build.rs +++ b/src/service/rooms/timeline/build.rs @@ -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 { - 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 { + 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 = 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(&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 = 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(&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(()) + } } diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs index d86b9c442..9cdd3ea34 100644 --- a/src/service/rooms/timeline/create.rs +++ b/src/service/rooms/timeline/create.rs @@ -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 = 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 = 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)) + } } diff --git a/src/service/rooms/timeline/redact.rs b/src/service/rooms/timeline/redact.rs index b3aff9eca..5509feeb5 100644 --- a/src/service/rooms/timeline/redact.rs +++ b/src/service/rooms/timeline/redact.rs @@ -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( - &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( + &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::() { + 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::() { - 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 } diff --git a/src/service/rooms/user/mod.rs b/src/service/rooms/user/mod.rs index c6a9f9cca..9705e2e3f 100644 --- a/src/service/rooms/user/mod.rs +++ b/src/service/rooms/user/mod.rs @@ -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> { 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) + } }