git-mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2025-09-11 17:53:01 +02:00
Code Issues Projects Releases Packages Wiki Activity

Post-formatting aesthetic and spacing corrections

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2025-04-27 02:39:28 +00:00 • committed by Jade Ellis
commit 364293608d
No known key found for this signature in database
GPG key ID: 8705A2A3EBF77BD2
72 changed files with 704 additions and 528 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/service/rooms/alias/mod.rs
View file

@ -3,7 +3,7 @@ mod remote;
use std::sync::Arc;
use conduwuit::{
Err, Result, Server, err,
Err, Event, Result, Server, err,
utils::{ReadyExt, stream::TryIgnore},
};
use database::{Deserialized, Ignore, Interfix, Map};
@ -241,7 +241,7 @@ impl Service {
.room_state_get(&room_id, &StateEventType::RoomCreate, "")
.await
{
return Ok(event.sender == user_id);
return Ok(event.sender() == user_id);
}
Err!(Database("Room has no m.room.create event"))

35
src/service/rooms/event_handler/fetch_and_handle_outliers.rs
View file

@ -4,11 +4,13 @@ use std::{
};
use conduwuit::{
PduEvent, debug, debug_error, debug_warn, implement, pdu, trace,
utils::continue_exponential_backoff_secs, warn,
Event, PduEvent, debug, debug_error, debug_warn, implement,
matrix::event::gen_event_id_canonical_json, trace, utils::continue_exponential_backoff_secs,
warn,
};
use ruma::{
CanonicalJsonValue, OwnedEventId, RoomId, ServerName, api::federation::event::get_event,
CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
api::federation::event::get_event,
};
use super::get_room_version_id;
@ -23,13 +25,17 @@ use super::get_room_version_id;
/// c. Ask origin server over federation
/// d. TODO: Ask other servers over federation?
#[implement(super::Service)]
pub(super) async fn fetch_and_handle_outliers<'a>(
pub(super) async fn fetch_and_handle_outliers<'a, Pdu, Events>(
&self,
origin: &'a ServerName,
events: &'a [OwnedEventId],
create_event: &'a PduEvent,
events: Events,
create_event: &'a Pdu,
room_id: &'a RoomId,
) -> Vec<(PduEvent, Option<BTreeMap<String, CanonicalJsonValue>>)> {
) -> Vec<(PduEvent, Option<BTreeMap<String, CanonicalJsonValue>>)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + Clone + Send,
{
let back_off = |id| match self
.services
.globals
@ -46,22 +52,23 @@ pub(super) async fn fetch_and_handle_outliers<'a>(
},
};
let mut events_with_auth_events = Vec::with_capacity(events.len());
let mut events_with_auth_events = Vec::with_capacity(events.clone().count());
for id in events {
// a. Look in the main timeline (pduid_pdu tree)
// b. Look at outlier pdu tree
// (get_pdu_json checks both)
if let Ok(local_pdu) = self.services.timeline.get_pdu(id).await {
trace!("Found {id} in db");
events_with_auth_events.push((id, Some(local_pdu), vec![]));
events_with_auth_events.push((id.to_owned(), Some(local_pdu), vec![]));
continue;
}
// c. Ask origin server over federation
// We also handle its auth chain here so we don't get a stack overflow in
// handle_outlier_pdu.
let mut todo_auth_events: VecDeque<_> = [id.clone()].into();
let mut todo_auth_events: VecDeque<_> = [id.to_owned()].into();
let mut events_in_reverse_order = Vec::with_capacity(todo_auth_events.len());
let mut events_all = HashSet::with_capacity(todo_auth_events.len());
while let Some(next_id) = todo_auth_events.pop_front() {
if let Some((time, tries)) = self
@ -117,7 +124,7 @@ pub(super) async fn fetch_and_handle_outliers<'a>(
};
let Ok((calculated_event_id, value)) =
pdu::gen_event_id_canonical_json(&res.pdu, &room_version_id)
gen_event_id_canonical_json(&res.pdu, &room_version_id)
else {
back_off((*next_id).to_owned());
continue;
@ -160,7 +167,8 @@ pub(super) async fn fetch_and_handle_outliers<'a>(
},
}
}
events_with_auth_events.push((id, None, events_in_reverse_order));
events_with_auth_events.push((id.to_owned(), None, events_in_reverse_order));
}
let mut pdus = Vec::with_capacity(events_with_auth_events.len());
@ -217,5 +225,6 @@ pub(super) async fn fetch_and_handle_outliers<'a>(
}
}
}
pdus
}

49
src/service/rooms/event_handler/fetch_prev.rs
View file

@ -1,13 +1,16 @@
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::{
collections::{BTreeMap, HashMap, HashSet, VecDeque},
iter::once,
};
use conduwuit::{
PduEvent, Result, debug_warn, err, implement,
Event, PduEvent, Result, debug_warn, err, implement,
state_res::{self},
};
use futures::{FutureExt, future};
use ruma::{
CanonicalJsonValue, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName, UInt, int,
uint,
CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName,
int, uint,
};
use super::check_room_id;
@ -19,20 +22,26 @@ use super::check_room_id;
fields(%origin),
)]
#[allow(clippy::type_complexity)]
pub(super) async fn fetch_prev(
pub(super) async fn fetch_prev<'a, Pdu, Events>(
&self,
origin: &ServerName,
create_event: &PduEvent,
create_event: &Pdu,
room_id: &RoomId,
first_ts_in_room: UInt,
initial_set: Vec<OwnedEventId>,
first_ts_in_room: MilliSecondsSinceUnixEpoch,
initial_set: Events,
) -> Result<(
Vec<OwnedEventId>,
HashMap<OwnedEventId, (PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
)> {
let mut graph: HashMap<OwnedEventId, _> = HashMap::with_capacity(initial_set.len());
)>
where
Pdu: Event + Send + Sync,
Events: Iterator<Item = &'a EventId> + Clone + Send,
{
let num_ids = initial_set.clone().count();
let mut eventid_info = HashMap::new();
let mut todo_outlier_stack: VecDeque<OwnedEventId> = initial_set.into();
let mut graph: HashMap<OwnedEventId, _> = HashMap::with_capacity(num_ids);
let mut todo_outlier_stack: VecDeque<OwnedEventId> =
initial_set.map(ToOwned::to_owned).collect();
let mut amount = 0;
@ -40,7 +49,12 @@ pub(super) async fn fetch_prev(
self.services.server.check_running()?;
match self
.fetch_and_handle_outliers(origin, &[prev_event_id.clone()], create_event, room_id)
.fetch_and_handle_outliers(
origin,
once(prev_event_id.as_ref()),
create_event,
room_id,
)
.boxed()
.await
.pop()
@ -65,17 +79,17 @@ pub(super) async fn fetch_prev(
}
if let Some(json) = json_opt {
if pdu.origin_server_ts > first_ts_in_room {
if pdu.origin_server_ts() > first_ts_in_room {
amount = amount.saturating_add(1);
for prev_prev in &pdu.prev_events {
for prev_prev in pdu.prev_events() {
if !graph.contains_key(prev_prev) {
todo_outlier_stack.push_back(prev_prev.clone());
todo_outlier_stack.push_back(prev_prev.to_owned());
}
}
graph.insert(
prev_event_id.clone(),
pdu.prev_events.iter().cloned().collect(),
pdu.prev_events().map(ToOwned::to_owned).collect(),
);
} else {
// Time based check failed
@ -98,8 +112,7 @@ pub(super) async fn fetch_prev(
let event_fetch = |event_id| {
let origin_server_ts = eventid_info
.get(&event_id)
.cloned()
.map_or_else(|| uint!(0), |info| info.0.origin_server_ts);
.map_or_else(|| uint!(0), |info| info.0.origin_server_ts().get());
// This return value is the key used for sorting events,
// events are then sorted by power level, time,

25
src/service/rooms/event_handler/fetch_state.rs
View file

@ -1,6 +1,6 @@
use std::collections::{HashMap, hash_map};
use conduwuit::{Err, Error, PduEvent, Result, debug, debug_warn, implement};
use conduwuit::{Err, Event, Result, debug, debug_warn, err, implement};
use futures::FutureExt;
use ruma::{
EventId, OwnedEventId, RoomId, ServerName, api::federation::event::get_room_state_ids,
@ -18,13 +18,16 @@ use crate::rooms::short::ShortStateKey;
skip_all,
fields(%origin),
)]
pub(super) async fn fetch_state(
pub(super) async fn fetch_state<Pdu>(
&self,
origin: &ServerName,
create_event: &PduEvent,
create_event: &Pdu,
room_id: &RoomId,
event_id: &EventId,
) -> Result<Option<HashMap<u64, OwnedEventId>>> {
) -> Result<Option<HashMap<u64, OwnedEventId>>>
where
Pdu: Event + Send + Sync,
{
let res = self
.services
.sending
@ -36,27 +39,27 @@ pub(super) async fn fetch_state(
.inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?;
debug!("Fetching state events");
let state_ids = res.pdu_ids.iter().map(AsRef::as_ref);
let state_vec = self
.fetch_and_handle_outliers(origin, &res.pdu_ids, create_event, room_id)
.fetch_and_handle_outliers(origin, state_ids, create_event, room_id)
.boxed()
.await;
let mut state: HashMap<ShortStateKey, OwnedEventId> = HashMap::with_capacity(state_vec.len());
for (pdu, _) in state_vec {
let state_key = pdu
.state_key
.clone()
.ok_or_else(|| Error::bad_database("Found non-state pdu in state events."))?;
.state_key()
.ok_or_else(|| err!(Database("Found non-state pdu in state events.")))?;
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(&pdu.kind.to_string().into(), &state_key)
.get_or_create_shortstatekey(&pdu.kind().to_string().into(), state_key)
.await;
match state.entry(shortstatekey) {
| hash_map::Entry::Vacant(v) => {
v.insert(pdu.event_id.clone());
v.insert(pdu.event_id().to_owned());
},
| hash_map::Entry::Occupied(_) => {
return Err!(Database(
@ -73,7 +76,7 @@ pub(super) async fn fetch_state(
.get_shortstatekey(&StateEventType::RoomCreate, "")
.await?;
if state.get(&create_shortstatekey) != Some(&create_event.event_id) {
if state.get(&create_shortstatekey).map(AsRef::as_ref) != Some(create_event.event_id()) {
return Err!(Database("Incoming event refers to wrong create event."));
}

15
src/service/rooms/event_handler/handle_incoming_pdu.rs
View file

@ -4,7 +4,7 @@ use std::{
};
use conduwuit::{
Err, Result, debug, debug::INFO_SPAN_LEVEL, defer, err, implement, utils::stream::IterStream,
Err, Event, Result, debug::INFO_SPAN_LEVEL, defer, err, implement, utils::stream::IterStream,
warn,
};
use futures::{
@ -12,6 +12,7 @@ use futures::{
future::{OptionFuture, try_join5},
};
use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UserId, events::StateEventType};
use tracing::debug;
use crate::rooms::timeline::RawPduId;
@ -121,22 +122,16 @@ pub async fn handle_incoming_pdu<'a>(
.timeline
.first_pdu_in_room(room_id)
.await?
.origin_server_ts;
.origin_server_ts();
if incoming_pdu.origin_server_ts < first_ts_in_room {
if incoming_pdu.origin_server_ts() < first_ts_in_room {
return Ok(None);
}
// 9. Fetch any missing prev events doing all checks listed here starting at 1.
// These are timeline events
let (sorted_prev_events, mut eventid_info) = self
.fetch_prev(
origin,
create_event,
room_id,
first_ts_in_room,
incoming_pdu.prev_events.clone(),
)
.fetch_prev(origin, create_event, room_id, first_ts_in_room, incoming_pdu.prev_events())
.await?;
debug!(

35
src/service/rooms/event_handler/handle_outlier_pdu.rs
View file

@ -1,7 +1,7 @@
use std::collections::{BTreeMap, HashMap, hash_map};
use conduwuit::{
Err, PduEvent, Result, debug, debug_info, err, implement, state_res, trace, warn,
Err, Event, PduEvent, Result, debug, debug_info, err, implement, state_res, trace, warn,
};
use futures::future::ready;
use ruma::{
@ -12,15 +12,18 @@ use super::{check_room_id, get_room_version_id, to_room_version};
#[implement(super::Service)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn handle_outlier_pdu<'a>(
pub(super) async fn handle_outlier_pdu<'a, Pdu>(
&self,
origin: &'a ServerName,
create_event: &'a PduEvent,
create_event: &'a Pdu,
event_id: &'a EventId,
room_id: &'a RoomId,
mut value: CanonicalJsonObject,
auth_events_known: bool,
) -> Result<(PduEvent, BTreeMap<String, CanonicalJsonValue>)> {
) -> Result<(PduEvent, BTreeMap<String, CanonicalJsonValue>)>
where
Pdu: Event + Send + Sync,
{
// 1. Remove unsigned field
value.remove("unsigned");
@ -29,7 +32,7 @@ pub(super) async fn handle_outlier_pdu<'a>(
// 2. Check signatures, otherwise drop
// 3. check content hash, redact if doesn't match
let room_version_id = get_room_version_id(create_event)?;
let mut val = match self
let mut incoming_pdu = match self
.services
.server_keys
.verify_event(&value, Some(&room_version_id))
@ -61,13 +64,15 @@ pub(super) async fn handle_outlier_pdu<'a>(
// Now that we have checked the signature and hashes we can add the eventID and
// convert to our PduEvent type
val.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()));
let incoming_pdu = serde_json::from_value::<PduEvent>(
serde_json::to_value(&val).expect("CanonicalJsonObj is a valid JsonValue"),
incoming_pdu
.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()));
let pdu_event = serde_json::from_value::<PduEvent>(
serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"),
)
.map_err(|e| err!(Request(BadJson(debug_warn!("Event is not a valid PDU: {e}")))))?;
check_room_id(room_id, &incoming_pdu)?;
check_room_id(room_id, &pdu_event)?;
if !auth_events_known {
// 4. fetch any missing auth events doing all checks listed here starting at 1.
@ -78,7 +83,7 @@ pub(super) async fn handle_outlier_pdu<'a>(
debug!("Fetching auth events");
Box::pin(self.fetch_and_handle_outliers(
origin,
&incoming_pdu.auth_events,
pdu_event.auth_events(),
create_event,
room_id,
))
@ -89,8 +94,8 @@ pub(super) async fn handle_outlier_pdu<'a>(
// auth events
debug!("Checking based on auth events");
// Build map of auth events
let mut auth_events = HashMap::with_capacity(incoming_pdu.auth_events.len());
for id in &incoming_pdu.auth_events {
let mut auth_events = HashMap::with_capacity(pdu_event.auth_events().count());
for id in pdu_event.auth_events() {
let Ok(auth_event) = self.services.timeline.get_pdu(id).await else {
warn!("Could not find auth event {id}");
continue;
@ -131,7 +136,7 @@ pub(super) async fn handle_outlier_pdu<'a>(
let auth_check = state_res::event_auth::auth_check(
&to_room_version(&room_version_id),
&incoming_pdu,
&pdu_event,
None, // TODO: third party invite
state_fetch,
)
@ -147,9 +152,9 @@ pub(super) async fn handle_outlier_pdu<'a>(
// 7. Persist the event as an outlier.
self.services
.outlier
.add_pdu_outlier(&incoming_pdu.event_id, &val);
.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
trace!("Added pdu as outlier.");
Ok((incoming_pdu, val))
Ok((pdu_event, incoming_pdu))
}

18
src/service/rooms/event_handler/handle_prev_pdu.rs
View file

@ -1,10 +1,11 @@
use std::{collections::BTreeMap, time::Instant};
use conduwuit::{
Err, PduEvent, Result, debug, debug::INFO_SPAN_LEVEL, defer, implement,
Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, defer, implement,
utils::continue_exponential_backoff_secs,
};
use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UInt};
use ruma::{CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName};
use tracing::debug;
#[implement(super::Service)]
#[allow(clippy::type_complexity)]
@ -15,16 +16,19 @@ use ruma::{CanonicalJsonValue, EventId, RoomId, ServerName, UInt};
skip_all,
fields(%prev_id),
)]
pub(super) async fn handle_prev_pdu<'a>(
pub(super) async fn handle_prev_pdu<'a, Pdu>(
&self,
origin: &'a ServerName,
event_id: &'a EventId,
room_id: &'a RoomId,
eventid_info: Option<(PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
create_event: &'a PduEvent,
first_ts_in_room: UInt,
create_event: &'a Pdu,
first_ts_in_room: MilliSecondsSinceUnixEpoch,
prev_id: &'a EventId,
) -> Result {
) -> Result
where
Pdu: Event + Send + Sync,
{
// Check for disabled again because it might have changed
if self.services.metadata.is_disabled(room_id).await {
return Err!(Request(Forbidden(debug_warn!(
@ -59,7 +63,7 @@ pub(super) async fn handle_prev_pdu<'a>(
};
// Skip old events
if pdu.origin_server_ts < first_ts_in_room {
if pdu.origin_server_ts() < first_ts_in_room {
return Ok(());
}

12
src/service/rooms/event_handler/mod.rs
View file

@ -18,7 +18,7 @@ use std::{
};
use async_trait::async_trait;
use conduwuit::{Err, PduEvent, Result, RoomVersion, Server, utils::MutexMap};
use conduwuit::{Err, Event, PduEvent, Result, RoomVersion, Server, utils::MutexMap};
use ruma::{
OwnedEventId, OwnedRoomId, RoomId, RoomVersionId,
events::room::create::RoomCreateEventContent,
@ -104,11 +104,11 @@ impl Service {
}
}
fn check_room_id(room_id: &RoomId, pdu: &PduEvent) -> Result {
if pdu.room_id != room_id {
fn check_room_id<Pdu: Event>(room_id: &RoomId, pdu: &Pdu) -> Result {
if pdu.room_id() != room_id {
return Err!(Request(InvalidParam(error!(
pdu_event_id = ?pdu.event_id,
pdu_room_id = ?pdu.room_id,
pdu_event_id = ?pdu.event_id(),
pdu_room_id = ?pdu.room_id(),
?room_id,
"Found event from room in room",
))));
@ -117,7 +117,7 @@ fn check_room_id(room_id: &RoomId, pdu: &PduEvent) -> Result {
Ok(())
}
fn get_room_version_id(create_event: &PduEvent) -> Result<RoomVersionId> {
fn get_room_version_id<Pdu: Event>(create_event: &Pdu) -> Result<RoomVersionId> {
let content: RoomCreateEventContent = create_event.get_content()?;
let room_version = content.room_version;

4
src/service/rooms/event_handler/parse_incoming_pdu.rs
View file

@ -1,4 +1,6 @@
use conduwuit::{Result, err, implement, pdu::gen_event_id_canonical_json, result::FlatOk};
use conduwuit::{
Result, err, implement, matrix::event::gen_event_id_canonical_json, result::FlatOk,
};
use ruma::{CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId};
use serde_json::value::RawValue as RawJsonValue;

48
src/service/rooms/event_handler/state_at_incoming.rs
View file

@ -6,7 +6,7 @@ use std::{
use conduwuit::{
Result, debug, err, implement,
matrix::{PduEvent, StateMap},
matrix::{Event, StateMap},
trace,
utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt},
};
@ -19,11 +19,18 @@ use crate::rooms::short::ShortStateHash;
#[implement(super::Service)]
// request and build the state from a known point and resolve if > 1 prev_event
#[tracing::instrument(name = "state", level = "debug", skip_all)]
pub(super) async fn state_at_incoming_degree_one(
pub(super) async fn state_at_incoming_degree_one<Pdu>(
&self,
incoming_pdu: &PduEvent,
) -> Result<Option<HashMap<u64, OwnedEventId>>> {
let prev_event = &incoming_pdu.prev_events[0];
incoming_pdu: &Pdu,
) -> Result<Option<HashMap<u64, OwnedEventId>>>
where
Pdu: Event + Send + Sync,
{
let prev_event = incoming_pdu
.prev_events()
.next()
.expect("at least one prev_event");
let Ok(prev_event_sstatehash) = self
.services
.state_accessor
@ -55,7 +62,7 @@ pub(super) async fn state_at_incoming_degree_one(
.get_or_create_shortstatekey(&prev_pdu.kind.to_string().into(), state_key)
.await;
state.insert(shortstatekey, prev_event.clone());
state.insert(shortstatekey, prev_event.to_owned());
// Now it's the state after the pdu
}
@ -66,16 +73,18 @@ pub(super) async fn state_at_incoming_degree_one(
#[implement(super::Service)]
#[tracing::instrument(name = "state", level = "debug", skip_all)]
pub(super) async fn state_at_incoming_resolved(
pub(super) async fn state_at_incoming_resolved<Pdu>(
&self,
incoming_pdu: &PduEvent,
incoming_pdu: &Pdu,
room_id: &RoomId,
room_version_id: &RoomVersionId,
) -> Result<Option<HashMap<u64, OwnedEventId>>> {
) -> Result<Option<HashMap<u64, OwnedEventId>>>
where
Pdu: Event + Send + Sync,
{
trace!("Calculating extremity statehashes...");
let Ok(extremity_sstatehashes) = incoming_pdu
.prev_events
.iter()
.prev_events()
.try_stream()
.broad_and_then(|prev_eventid| {
self.services
@ -133,12 +142,15 @@ pub(super) async fn state_at_incoming_resolved(
}
#[implement(super::Service)]
async fn state_at_incoming_fork(
async fn state_at_incoming_fork<Pdu>(
&self,
room_id: &RoomId,
sstatehash: ShortStateHash,
prev_event: PduEvent,
) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)> {
prev_event: Pdu,
) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)>
where
Pdu: Event,
{
let mut leaf_state: HashMap<_, _> = self
.services
.state_accessor
@ -146,15 +158,15 @@ async fn state_at_incoming_fork(
.collect()
.await;
if let Some(state_key) = &prev_event.state_key {
if let Some(state_key) = prev_event.state_key() {
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(&prev_event.kind.to_string().into(), state_key)
.get_or_create_shortstatekey(&prev_event.kind().to_string().into(), state_key)
.await;
let event_id = &prev_event.event_id;
leaf_state.insert(shortstatekey, event_id.clone());
let event_id = prev_event.event_id();
leaf_state.insert(shortstatekey, event_id.to_owned());
// Now it's the state after the pdu
}

50
src/service/rooms/event_handler/upgrade_outlier_pdu.rs
View file

@ -1,7 +1,7 @@
use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
use conduwuit::{
Err, Result, debug, debug_info, err, implement,
Err, Result, debug, debug_info, err, implement, is_equal_to,
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
trace,
utils::stream::{BroadbandExt, ReadyExt},
@ -17,19 +17,22 @@ use crate::rooms::{
};
#[implement(super::Service)]
pub(super) async fn upgrade_outlier_to_timeline_pdu(
pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
&self,
incoming_pdu: PduEvent,
val: BTreeMap<String, CanonicalJsonValue>,
create_event: &PduEvent,
create_event: &Pdu,
origin: &ServerName,
room_id: &RoomId,
) -> Result<Option<RawPduId>> {
) -> Result<Option<RawPduId>>
where
Pdu: Event + Send + Sync,
{
// Skip the PDU if we already have it as a timeline event
if let Ok(pduid) = self
.services
.timeline
.get_pdu_id(&incoming_pdu.event_id)
.get_pdu_id(incoming_pdu.event_id())
.await
{
return Ok(Some(pduid));
@ -38,7 +41,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
if self
.services
.pdu_metadata
.is_event_soft_failed(&incoming_pdu.event_id)
.is_event_soft_failed(incoming_pdu.event_id())
.await
{
return Err!(Request(InvalidParam("Event has been soft failed")));
@ -53,7 +56,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
// These are not timeline events.
debug!("Resolving state at event");
let mut state_at_incoming_event = if incoming_pdu.prev_events.len() == 1 {
let mut state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
self.state_at_incoming_degree_one(&incoming_pdu).await?
} else {
self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_id)
@ -62,12 +65,13 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
if state_at_incoming_event.is_none() {
state_at_incoming_event = self
.fetch_state(origin, create_event, room_id, &incoming_pdu.event_id)
.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
.await?;
}
let state_at_incoming_event =
state_at_incoming_event.expect("we always set this to some above");
let room_version = to_room_version(&room_version_id);
debug!("Performing auth check");
@ -99,10 +103,10 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.state
.get_auth_events(
room_id,
&incoming_pdu.kind,
&incoming_pdu.sender,
incoming_pdu.state_key.as_deref(),
&incoming_pdu.content,
incoming_pdu.kind(),
incoming_pdu.sender(),
incoming_pdu.state_key(),
incoming_pdu.content(),
)
.await?;
@ -129,7 +133,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
!self
.services
.state_accessor
.user_can_redact(&redact_id, &incoming_pdu.sender, &incoming_pdu.room_id, true)
.user_can_redact(&redact_id, incoming_pdu.sender(), incoming_pdu.room_id(), true)
.await?,
};
@ -149,7 +153,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.map(ToOwned::to_owned)
.ready_filter(|event_id| {
// Remove any that are referenced by this incoming event's prev_events
!incoming_pdu.prev_events.contains(event_id)
!incoming_pdu.prev_events().any(is_equal_to!(event_id))
})
.broad_filter_map(|event_id| async move {
// Only keep those extremities were not referenced yet
@ -166,7 +170,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
debug!(
"Retained {} extremities checked against {} prev_events",
extremities.len(),
incoming_pdu.prev_events.len()
incoming_pdu.prev_events().count()
);
let state_ids_compressed: Arc<CompressedState> = self
@ -181,20 +185,20 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
.map(Arc::new)
.await;
if incoming_pdu.state_key.is_some() {
if incoming_pdu.state_key().is_some() {
debug!("Event is a state-event. Deriving new room state");
// We also add state after incoming event to the fork states
let mut state_after = state_at_incoming_event.clone();
if let Some(state_key) = &incoming_pdu.state_key {
if let Some(state_key) = incoming_pdu.state_key() {
let shortstatekey = self
.services
.short
.get_or_create_shortstatekey(&incoming_pdu.kind.to_string().into(), state_key)
.get_or_create_shortstatekey(&incoming_pdu.kind().to_string().into(), state_key)
.await;
let event_id = &incoming_pdu.event_id;
state_after.insert(shortstatekey, event_id.clone());
let event_id = incoming_pdu.event_id();
state_after.insert(shortstatekey, event_id.to_owned());
}
let new_room_state = self
@ -236,9 +240,9 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
// Soft fail, we keep the event as an outlier but don't add it to the timeline
self.services
.pdu_metadata
.mark_event_soft_failed(&incoming_pdu.event_id);
.mark_event_soft_failed(incoming_pdu.event_id());
warn!("Event was soft failed: {incoming_pdu:?}");
warn!("Event was soft failed: {:?}", incoming_pdu.event_id());
return Err!(Request(InvalidParam("Event has been soft failed")));
}
@ -249,7 +253,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
let extremities = extremities
.iter()
.map(Borrow::borrow)
.chain(once(incoming_pdu.event_id.borrow()));
.chain(once(incoming_pdu.event_id()));
let pdu_id = self
.services

4
src/service/rooms/outlier/mod.rs
View file

@ -1,7 +1,7 @@
use std::sync::Arc;
use conduwuit::{Result, implement, matrix::pdu::PduEvent};
use conduwuit_database::{Deserialized, Json, Map};
use conduwuit::{Result, implement, matrix::PduEvent};
use database::{Deserialized, Json, Map};
use ruma::{CanonicalJsonObject, EventId};
pub struct Service {

10
src/service/rooms/pdu_metadata/data.rs
View file

@ -1,8 +1,8 @@
use std::{mem::size_of, sync::Arc};
use conduwuit::{
PduCount, PduEvent,
arrayvec::ArrayVec,
matrix::{Event, PduCount},
result::LogErr,
utils::{
ReadyExt,
@ -33,8 +33,6 @@ struct Services {
timeline: Dep<rooms::timeline::Service>,
}
pub(super) type PdusIterItem = (PduCount, PduEvent);
impl Data {
pub(super) fn new(args: &crate::Args<'_>) -> Self {
let db = &args.db;
@ -62,7 +60,7 @@ impl Data {
target: ShortEventId,
from: PduCount,
dir: Direction,
) -> impl Stream<Item = PdusIterItem> + Send + '_ {
) -> impl Stream<Item = (PduCount, impl Event)> + Send + '_ {
let mut current = ArrayVec::<u8, 16>::new();
current.extend(target.to_be_bytes());
current.extend(from.saturating_inc(dir).into_unsigned().to_be_bytes());
@ -80,8 +78,8 @@ impl Data {
let mut pdu = self.services.timeline.get_pdu_from_id(&pdu_id).await.ok()?;
if pdu.sender != user_id {
pdu.remove_transaction_id().log_err().ok();
if pdu.sender() != user_id {
pdu.as_mut_pdu().remove_transaction_id().log_err().ok();
}
Some((shorteventid, pdu))

19
src/service/rooms/pdu_metadata/mod.rs
View file

@ -1,11 +1,14 @@
mod data;
use std::sync::Arc;
use conduwuit::{PduCount, Result};
use conduwuit::{
Result,
matrix::{Event, PduCount},
};
use futures::{StreamExt, future::try_join};
use ruma::{EventId, RoomId, UserId, api::Direction};
use self::data::{Data, PdusIterItem};
use self::data::Data;
use crate::{Dep, rooms};
pub struct Service {
@ -44,16 +47,16 @@ impl Service {
}
#[allow(clippy::too_many_arguments)]
pub async fn get_relations(
&self,
user_id: &UserId,
room_id: &RoomId,
target: &EventId,
pub async fn get_relations<'a>(
&'a self,
user_id: &'a UserId,
room_id: &'a RoomId,
target: &'a EventId,
from: PduCount,
limit: usize,
max_depth: u8,
dir: Direction,
) -> Vec<PdusIterItem> {
) -> Vec<(PduCount, impl Event)> {
let room_id = self.services.short.get_shortroomid(room_id);
let target = self.services.timeline.get_pdu_count(target);

10
src/service/rooms/read_receipt/mod.rs
View file

@ -4,7 +4,10 @@ use std::{collections::BTreeMap, sync::Arc};
use conduwuit::{
Result, debug, err,
matrix::pdu::{PduCount, PduId, RawPduId},
matrix::{
Event,
pdu::{PduCount, PduId, RawPduId},
},
warn,
};
use futures::{Stream, TryFutureExt, try_join};
@ -74,14 +77,13 @@ impl Service {
let shortroomid = self.services.short.get_shortroomid(room_id).map_err(|e| {
err!(Database(warn!("Short room ID does not exist in database for {room_id}: {e}")))
});
let (pdu_count, shortroomid) = try_join!(pdu_count, shortroomid)?;
let (pdu_count, shortroomid) = try_join!(pdu_count, shortroomid)?;
let shorteventid = PduCount::Normal(pdu_count);
let pdu_id: RawPduId = PduId { shortroomid, shorteventid }.into();
let pdu = self.services.timeline.get_pdu_from_id(&pdu_id).await?;
let event_id: OwnedEventId = pdu.event_id;
let event_id: OwnedEventId = pdu.event_id().to_owned();
let user_id: OwnedUserId = user_id.to_owned();
let content: BTreeMap<OwnedEventId, Receipts> = BTreeMap::from_iter([(
event_id,

12
src/service/rooms/search/mod.rs
View file

@ -1,9 +1,10 @@
use std::sync::Arc;
use conduwuit_core::{
Event, PduCount, PduEvent, Result,
use conduwuit::{
PduCount, Result,
arrayvec::ArrayVec,
implement,
matrix::event::{Event, Matches},
utils::{
ArrayVecExt, IterStream, ReadyExt, set,
stream::{TryIgnore, WidebandExt},
@ -103,9 +104,10 @@ pub fn deindex_pdu(&self, shortroomid: ShortRoomId, pdu_id: &RawPduId, message_b
pub async fn search_pdus<'a>(
&'a self,
query: &'a RoomQuery<'a>,
) -> Result<(usize, impl Stream<Item = PduEvent> + Send + 'a)> {
) -> Result<(usize, impl Stream<Item = impl Event + use<>> + Send + '_)> {
let pdu_ids: Vec<_> = self.search_pdu_ids(query).await?.collect().await;
let filter = &query.criteria.filter;
let count = pdu_ids.len();
let pdus = pdu_ids
.into_iter()
@ -118,11 +120,11 @@ pub async fn search_pdus<'a>(
.ok()
})
.ready_filter(|pdu| !pdu.is_redacted())
.ready_filter(|pdu| pdu.matches(&query.criteria.filter))
.ready_filter(move |pdu| filter.matches(pdu))
.wide_filter_map(move |pdu| async move {
self.services
.state_accessor
.user_can_see_event(query.user_id?, &pdu.room_id, &pdu.event_id)
.user_can_see_event(query.user_id?, pdu.room_id(), pdu.event_id())
.await
.then_some(pdu)
})

4
src/service/rooms/state/mod.rs
View file

@ -356,8 +356,8 @@ impl Service {
&self,
room_id: &RoomId,
shortstatehash: u64,
_mutex_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room
* state mutex */
// Take mutex guard to make sure users get the room state mutex
_mutex_lock: &RoomMutexGuard,
) {
const BUFSIZE: usize = size_of::<u64>();

8
src/service/rooms/state_accessor/room_state.rs
View file

@ -2,7 +2,7 @@ use std::borrow::Borrow;
use conduwuit::{
Result, err, implement,
matrix::{PduEvent, StateKey},
matrix::{Event, StateKey},
};
use futures::{Stream, StreamExt, TryFutureExt};
use ruma::{EventId, RoomId, events::StateEventType};
@ -30,7 +30,7 @@ where
pub fn room_state_full<'a>(
&'a self,
room_id: &'a RoomId,
) -> impl Stream<Item = Result<((StateEventType, StateKey), PduEvent)>> + Send + 'a {
) -> impl Stream<Item = Result<((StateEventType, StateKey), impl Event)>> + Send + 'a {
self.services
.state
.get_room_shortstatehash(room_id)
@ -45,7 +45,7 @@ pub fn room_state_full<'a>(
pub fn room_state_full_pdus<'a>(
&'a self,
room_id: &'a RoomId,
) -> impl Stream<Item = Result<PduEvent>> + Send + 'a {
) -> impl Stream<Item = Result<impl Event>> + Send + 'a {
self.services
.state
.get_room_shortstatehash(room_id)
@ -84,7 +84,7 @@ pub async fn room_state_get(
room_id: &RoomId,
event_type: &StateEventType,
state_key: &str,
) -> Result<PduEvent> {
) -> Result<impl Event> {
self.services
.state
.get_room_shortstatehash(room_id)

18
src/service/rooms/state_accessor/state.rs
View file

@ -2,14 +2,14 @@ use std::{borrow::Borrow, ops::Deref, sync::Arc};
use conduwuit::{
Result, at, err, implement,
matrix::{PduEvent, StateKey},
matrix::{Event, StateKey},
pair_of,
utils::{
result::FlatOk,
stream::{BroadbandExt, IterStream, ReadyExt, TryIgnore},
},
};
use conduwuit_database::Deserialized;
use database::Deserialized;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, future::try_join, pin_mut};
use ruma::{
EventId, OwnedEventId, UserId,
@ -125,11 +125,9 @@ pub async fn state_get(
shortstatehash: ShortStateHash,
event_type: &StateEventType,
state_key: &str,
) -> Result<PduEvent> {
) -> Result<impl Event> {
self.state_get_id(shortstatehash, event_type, state_key)
.and_then(|event_id: OwnedEventId| async move {
self.services.timeline.get_pdu(&event_id).await
})
.and_then(async |event_id: OwnedEventId| self.services.timeline.get_pdu(&event_id).await)
.await
}
@ -316,18 +314,16 @@ pub fn state_added(
pub fn state_full(
&self,
shortstatehash: ShortStateHash,
) -> impl Stream<Item = ((StateEventType, StateKey), PduEvent)> + Send + '_ {
) -> impl Stream<Item = ((StateEventType, StateKey), impl Event)> + Send + '_ {
self.state_full_pdus(shortstatehash)
.ready_filter_map(|pdu| {
Some(((pdu.kind.to_string().into(), pdu.state_key.clone()?), pdu))
})
.ready_filter_map(|pdu| Some(((pdu.kind().clone().into(), pdu.state_key()?.into()), pdu)))
}
#[implement(super::Service)]
pub fn state_full_pdus(
&self,
shortstatehash: ShortStateHash,
) -> impl Stream<Item = PduEvent> + Send + '_ {
) -> impl Stream<Item = impl Event> + Send + '_ {
let short_ids = self
.state_full_shortids(shortstatehash)
.ignore_err()

14
src/service/rooms/state_accessor/user_can.rs
View file

@ -1,4 +1,4 @@
use conduwuit::{Err, Result, implement, pdu::PduBuilder};
use conduwuit::{Err, Result, implement, matrix::Event, pdu::PduBuilder};
use ruma::{
EventId, RoomId, UserId,
events::{
@ -29,14 +29,14 @@ pub async fn user_can_redact(
if redacting_event
.as_ref()
.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomCreate)
.is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomCreate)
{
return Err!(Request(Forbidden("Redacting m.room.create is not safe, forbidding.")));
}
if redacting_event
.as_ref()
.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomServerAcl)
.is_ok_and(|pdu| *pdu.kind() == TimelineEventType::RoomServerAcl)
{
return Err!(Request(Forbidden(
"Redacting m.room.server_acl will result in the room being inaccessible for \
@ -59,9 +59,9 @@ pub async fn user_can_redact(
&& match redacting_event {
| Ok(redacting_event) =>
if federation {
redacting_event.sender.server_name() == sender.server_name()
redacting_event.sender().server_name() == sender.server_name()
} else {
redacting_event.sender == sender
redacting_event.sender() == sender
},
| _ => false,
})
@ -72,10 +72,10 @@ pub async fn user_can_redact(
.room_state_get(room_id, &StateEventType::RoomCreate, "")
.await
{
| Ok(room_create) => Ok(room_create.sender == sender
| Ok(room_create) => Ok(room_create.sender() == sender
|| redacting_event
.as_ref()
.is_ok_and(|redacting_event| redacting_event.sender == sender)),
.is_ok_and(|redacting_event| redacting_event.sender() == sender)),
| _ => Err!(Database(
"No m.room.power_levels or m.room.create events in database for room"
)),

13
src/service/rooms/threads/mod.rs
View file

@ -49,10 +49,9 @@ impl crate::Service for Service {
}
impl Service {
pub async fn add_to_thread<'a, E>(&self, root_event_id: &EventId, event: &'a E) -> Result
pub async fn add_to_thread<E>(&self, root_event_id: &EventId, event: &E) -> Result
where
E: Event + Send + Sync,
&'a E: Event + Send,
{
let root_id = self
.services
@ -120,7 +119,7 @@ impl Service {
self.services
.timeline
.replace_pdu(&root_id, &root_pdu_json, &root_pdu)
.replace_pdu(&root_id, &root_pdu_json)
.await?;
}
@ -130,7 +129,7 @@ impl Service {
users.extend_from_slice(&userids);
},
| _ => {
users.push(root_pdu.sender);
users.push(root_pdu.sender().to_owned());
},
}
users.push(event.sender().to_owned());
@ -162,10 +161,10 @@ impl Service {
.ready_take_while(move |pdu_id| pdu_id.shortroomid() == shortroomid.to_be_bytes())
.wide_filter_map(move |pdu_id| async move {
let mut pdu = self.services.timeline.get_pdu_from_id(&pdu_id).await.ok()?;
let pdu_id: PduId = pdu_id.into();
if pdu.sender != user_id {
pdu.remove_transaction_id().ok();
let pdu_id: PduId = pdu_id.into();
if pdu.sender() != user_id {
pdu.as_mut_pdu().remove_transaction_id().ok();
}
Some((pdu_id.shorteventid, pdu))

1
src/service/rooms/timeline/data.rs
View file

@ -207,7 +207,6 @@ impl Data {
&self,
pdu_id: &RawPduId,
pdu_json: &CanonicalJsonObject,
_pdu: &PduEvent,
) -> Result {
if self.pduid_pdu.get(pdu_id).await.is_not_found() {
return Err!(Request(NotFound("PDU does not exist.")));

89
src/service/rooms/timeline/mod.rs
View file

@ -14,8 +14,8 @@ pub use conduwuit::matrix::pdu::{PduId, RawPduId};
use conduwuit::{
Err, Error, Result, Server, at, debug, debug_warn, err, error, implement, info,
matrix::{
Event,
pdu::{EventHash, PduBuilder, PduCount, PduEvent, gen_event_id},
event::{Event, gen_event_id},
pdu::{EventHash, PduBuilder, PduCount, PduEvent},
state_res::{self, RoomVersion},
},
utils::{
@ -159,12 +159,12 @@ impl crate::Service for Service {
impl Service {
#[tracing::instrument(skip(self), level = "debug")]
pub async fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<PduEvent> {
pub async fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<impl Event> {
self.first_item_in_room(room_id).await.map(at!(1))
}
#[tracing::instrument(skip(self), level = "debug")]
pub async fn first_item_in_room(&self, room_id: &RoomId) -> Result<(PduCount, PduEvent)> {
pub async fn first_item_in_room(&self, room_id: &RoomId) -> Result<(PduCount, impl Event)> {
let pdus = self.pdus(None, room_id, None);
pin_mut!(pdus);
@ -174,7 +174,7 @@ impl Service {
}
#[tracing::instrument(skip(self), level = "debug")]
pub async fn latest_pdu_in_room(&self, room_id: &RoomId) -> Result<PduEvent> {
pub async fn latest_pdu_in_room(&self, room_id: &RoomId) -> Result<impl Event> {
self.db.latest_pdu_in_room(None, room_id).await
}
@ -216,13 +216,14 @@ impl Service {
///
/// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
#[inline]
pub async fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<PduEvent> {
pub async fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<impl Event> {
self.db.get_non_outlier_pdu(event_id).await
}
/// Returns the pdu.
///
/// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
#[inline]
pub async fn get_pdu(&self, event_id: &EventId) -> Result<PduEvent> {
self.db.get_pdu(event_id).await
}
@ -230,11 +231,13 @@ impl Service {
/// Returns the pdu.
///
/// This does __NOT__ check the outliers `Tree`.
#[inline]
pub async fn get_pdu_from_id(&self, pdu_id: &RawPduId) -> Result<PduEvent> {
self.db.get_pdu_from_id(pdu_id).await
}
/// Returns the pdu as a `BTreeMap<String, CanonicalJsonValue>`.
#[inline]
pub async fn get_pdu_json_from_id(&self, pdu_id: &RawPduId) -> Result<CanonicalJsonObject> {
self.db.get_pdu_json_from_id(pdu_id).await
}
@ -242,6 +245,7 @@ impl Service {
/// Checks if pdu exists
///
/// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
#[inline]
pub fn pdu_exists<'a>(
&'a self,
event_id: &'a EventId,
@ -251,13 +255,8 @@ impl Service {
/// Removes a pdu and creates a new one with the same id.
#[tracing::instrument(skip(self), level = "debug")]
pub async fn replace_pdu(
&self,
pdu_id: &RawPduId,
pdu_json: &CanonicalJsonObject,
pdu: &PduEvent,
) -> Result<()> {
self.db.replace_pdu(pdu_id, pdu_json, pdu).await
pub async fn replace_pdu(&self, pdu_id: &RawPduId, pdu_json: &CanonicalJsonObject) -> Result {
self.db.replace_pdu(pdu_id, pdu_json).await
}
/// Creates a new persisted data unit and adds it to a room.
@ -310,25 +309,21 @@ impl Service {
unsigned.insert(
"prev_content".to_owned(),
CanonicalJsonValue::Object(
utils::to_canonical_object(prev_state.content.clone()).map_err(
|e| {
error!(
"Failed to convert prev_state to canonical JSON: {e}"
);
Error::bad_database(
"Failed to convert prev_state to canonical JSON.",
)
},
)?,
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()),
CanonicalJsonValue::String(prev_state.sender().to_string()),
);
unsigned.insert(
String::from("replaces_state"),
CanonicalJsonValue::String(prev_state.event_id.to_string()),
CanonicalJsonValue::String(prev_state.event_id().to_string()),
);
}
}
@ -709,14 +704,11 @@ impl Service {
.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)
.expect("UserId::to_value always works"),
);
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).expect("EventId is valid json"),
serde_json::to_value(prev_pdu.event_id())?,
);
}
}
@ -759,7 +751,7 @@ impl Service {
unsigned: if unsigned.is_empty() {
None
} else {
Some(to_raw_value(&unsigned).expect("to_raw_value always works"))
Some(to_raw_value(&unsigned)?)
},
hashes: EventHash { sha256: "aaa".to_owned() },
signatures: None,
@ -1041,10 +1033,10 @@ impl Service {
/// Replace a PDU with the redacted form.
#[tracing::instrument(name = "redact", level = "debug", skip(self))]
pub async fn redact_pdu(
pub async fn redact_pdu<Pdu: Event + Send + Sync>(
&self,
event_id: &EventId,
reason: &PduEvent,
reason: &Pdu,
shortroomid: ShortRoomId,
) -> Result {
// TODO: Don't reserialize, keep original json
@ -1053,9 +1045,13 @@ impl Service {
return Ok(());
};
let mut pdu = self.get_pdu_from_id(&pdu_id).await.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 {
@ -1065,15 +1061,15 @@ impl Service {
}
}
let room_version_id = self.services.state.get_room_version(&pdu.room_id).await?;
let room_version_id = self.services.state.get_room_version(pdu.room_id()).await?;
pdu.redact(&room_version_id, reason)?;
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, &pdu).await
self.replace_pdu(&pdu_id, &obj).await
}
#[tracing::instrument(name = "backfill", level = "debug", skip(self))]
@ -1163,7 +1159,7 @@ impl Service {
backfill_server,
federation::backfill::get_backfill::v1::Request {
room_id: room_id.to_owned(),
v: vec![first_pdu.1.event_id.clone()],
v: vec![first_pdu.1.event_id().to_owned()],
limit: uint!(100),
},
)
@ -1248,8 +1244,11 @@ impl Service {
#[implement(Service)]
#[tracing::instrument(skip_all, level = "debug")]
async fn check_pdu_for_admin_room(&self, pdu: &PduEvent, sender: &UserId) -> Result<()> {
match &pdu.kind {
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."))));
},
@ -1273,7 +1272,7 @@ async fn check_pdu_for_admin_room(&self, pdu: &PduEvent, sender: &UserId) -> Res
let count = self
.services
.state_cache
.room_members(&pdu.room_id)
.room_members(pdu.room_id())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()
@ -1297,7 +1296,7 @@ async fn check_pdu_for_admin_room(&self, pdu: &PduEvent, sender: &UserId) -> Res
let count = self
.services
.state_cache
.room_members(&pdu.room_id)
.room_members(pdu.room_id())
.ready_filter(|user| self.services.globals.user_is_local(user))
.ready_filter(|user| *user != target)
.boxed()