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

Toward abstracting Pdu into trait Event.

Browse source 
Co-authored-by: Jade Ellis <jade@ellis.link>
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2025-04-26 08:24:47 +00:00 • committed by Jade Ellis
commit 116f85360f
No known key found for this signature in database
GPG key ID: 8705A2A3EBF77BD2
41 changed files with 842 additions and 886 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/service/admin/mod.rs
View file

@ -9,8 +9,8 @@ use std::{
};
use async_trait::async_trait;
use conduwuit::{
Error, PduEvent, Result, Server, debug, err, error, error::default_log, pdu::PduBuilder,
use conduwuit_core::{
Error, Event, Result, Server, debug, err, error, error::default_log, pdu::PduBuilder,
};
pub use create::create_admin_room;
use futures::{Future, FutureExt, TryFutureExt};
@ -361,7 +361,10 @@ impl Service {
Ok(())
}
pub async fn is_admin_command(&self, pdu: &PduEvent, body: &str) -> bool {
pub async fn is_admin_command<E>(&self, event: &E, body: &str) -> bool
where
E: Event + Send + Sync,
{
// Server-side command-escape with public echo
let is_escape = body.starts_with('\\');
let is_public_escape = is_escape && body.trim_start_matches('\\').starts_with("!admin");
@ -376,8 +379,10 @@ impl Service {
return false;
}
let user_is_local = self.services.globals.user_is_local(event.sender());
// only allow public escaped commands by local admins
if is_public_escape && !self.services.globals.user_is_local(&pdu.sender) {
if is_public_escape && !user_is_local {
return false;
}
@ -387,20 +392,20 @@ impl Service {
}
// Prevent unescaped !admin from being used outside of the admin room
if is_public_prefix && !self.is_admin_room(&pdu.room_id).await {
if is_public_prefix && !self.is_admin_room(event.room_id()).await {
return false;
}
// Only senders who are admin can proceed
if !self.user_is_admin(&pdu.sender).await {
if !self.user_is_admin(event.sender()).await {
return false;
}
// This will evaluate to false if the emergency password is set up so that
// the administrator can execute commands as the server user
let emergency_password_set = self.services.server.config.emergency_password.is_some();
let from_server = pdu.sender == *server_user && !emergency_password_set;
if from_server && self.is_admin_room(&pdu.room_id).await {
let from_server = event.sender() == server_user && !emergency_password_set;
if from_server && self.is_admin_room(event.room_id()).await {
return false;
}

56
src/service/pusher/mod.rs
View file

@ -1,12 +1,12 @@
use std::{fmt::Debug, mem, sync::Arc};
use bytes::BytesMut;
use conduwuit::{
Err, PduEvent, Result, debug_warn, err, trace,
use conduwuit_core::{
Err, Event, Result, debug_warn, err, trace,
utils::{stream::TryIgnore, string_from_bytes},
warn,
};
use database::{Deserialized, Ignore, Interfix, Json, Map};
use conduwuit_database::{Deserialized, Ignore, Interfix, Json, Map};
use futures::{Stream, StreamExt};
use ipaddress::IPAddress;
use ruma::{
@ -272,22 +272,26 @@ impl Service {
}
}
#[tracing::instrument(skip(self, user, unread, pusher, ruleset, pdu))]
pub async fn send_push_notice(
#[tracing::instrument(skip(self, user, unread, pusher, ruleset, event))]
pub async fn send_push_notice<E>(
&self,
user: &UserId,
unread: UInt,
pusher: &Pusher,
ruleset: Ruleset,
pdu: &PduEvent,
) -> Result<()> {
event: &E,
) -> Result
where
E: Event + Send + Sync,
for<'a> &'a E: Event + Send,
{
let mut notify = None;
let mut tweaks = Vec::new();
let power_levels: RoomPowerLevelsEventContent = self
.services
.state_accessor
.room_state_get(&pdu.room_id, &StateEventType::RoomPowerLevels, "")
.room_state_get(event.room_id(), &StateEventType::RoomPowerLevels, "")
.await
.and_then(|ev| {
serde_json::from_str(ev.content.get()).map_err(|e| {
@ -296,8 +300,9 @@ impl Service {
})
.unwrap_or_default();
let serialized = event.to_format();
for action in self
.get_actions(user, &ruleset, &power_levels, &pdu.to_sync_room_event(), &pdu.room_id)
.get_actions(user, &ruleset, &power_levels, &serialized, event.room_id())
.await
{
let n = match action {
@ -319,7 +324,7 @@ impl Service {
}
if notify == Some(true) {
self.send_notice(unread, pusher, tweaks, pdu).await?;
self.send_notice(unread, pusher, tweaks, event).await?;
}
// Else the event triggered no actions
@ -369,13 +374,16 @@ impl Service {
}
#[tracing::instrument(skip(self, unread, pusher, tweaks, event))]
async fn send_notice(
async fn send_notice<E>(
&self,
unread: UInt,
pusher: &Pusher,
tweaks: Vec<Tweak>,
event: &PduEvent,
) -> Result {
event: &E,
) -> Result
where
E: Event + Send + Sync,
{
// TODO: email
match &pusher.kind {
| PusherKind::Http(http) => {
@ -421,8 +429,8 @@ impl Service {
let d = vec![device];
let mut notifi = Notification::new(d);
notifi.event_id = Some((*event.event_id).to_owned());
notifi.room_id = Some((*event.room_id).to_owned());
notifi.event_id = Some(event.event_id().to_owned());
notifi.room_id = Some(event.room_id().to_owned());
if http
.data
.get("org.matrix.msc4076.disable_badge_count")
@ -442,7 +450,7 @@ impl Service {
)
.await?;
} else {
if event.kind == TimelineEventType::RoomEncrypted
if *event.kind() == TimelineEventType::RoomEncrypted
|| tweaks
.iter()
.any(|t| matches!(t, Tweak::Highlight(true) | Tweak::Sound(_)))
@ -451,29 +459,29 @@ impl Service {
} else {
notifi.prio = NotificationPriority::Low;
}
notifi.sender = Some(event.sender.clone());
notifi.event_type = Some(event.kind.clone());
notifi.content = serde_json::value::to_raw_value(&event.content).ok();
notifi.sender = Some(event.sender().to_owned());
notifi.event_type = Some(event.kind().to_owned());
notifi.content = serde_json::value::to_raw_value(event.content()).ok();
if event.kind == TimelineEventType::RoomMember {
if *event.kind() == TimelineEventType::RoomMember {
notifi.user_is_target =
event.state_key.as_deref() == Some(event.sender.as_str());
event.state_key() == Some(event.sender().as_str());
}
notifi.sender_display_name =
self.services.users.displayname(&event.sender).await.ok();
self.services.users.displayname(event.sender()).await.ok();
notifi.room_name = self
.services
.state_accessor
.get_name(&event.room_id)
.get_name(event.room_id())
.await
.ok();
notifi.room_alias = self
.services
.state_accessor
.get_canonical_alias(&event.room_id)
.get_canonical_alias(event.room_id())
.await
.ok();

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

@ -126,7 +126,7 @@ pub(super) async fn handle_outlier_pdu<'a>(
let state_fetch = |ty: &StateEventType, sk: &str| {
let key = (ty.to_owned(), sk.into());
ready(auth_events.get(&key))
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
let auth_check = state_res::event_auth::auth_check(

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

@ -2,7 +2,7 @@ use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::In
use conduwuit::{
Err, Result, debug, debug_info, err, implement,
matrix::{EventTypeExt, PduEvent, StateKey, state_res},
matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
trace,
utils::stream::{BroadbandExt, ReadyExt},
warn,
@ -108,7 +108,7 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
let state_fetch = |k: &StateEventType, s: &str| {
let key = k.with_state_key(s);
ready(auth_events.get(&key).cloned())
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
let auth_check = state_res::event_auth::auth_check(

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

@ -1,7 +1,7 @@
use std::sync::Arc;
use conduwuit::{
PduCount, PduEvent, Result,
use conduwuit_core::{
Event, PduCount, PduEvent, Result,
arrayvec::ArrayVec,
implement,
utils::{

8
src/service/rooms/spaces/mod.rs
View file

@ -5,8 +5,8 @@ mod tests;
use std::{fmt::Write, sync::Arc};
use async_trait::async_trait;
use conduwuit::{
Err, Error, PduEvent, Result, implement,
use conduwuit_core::{
Err, Error, Event, PduEvent, Result, implement,
utils::{
IterStream,
future::{BoolExt, TryExtExt},
@ -142,7 +142,7 @@ pub async fn get_summary_and_children_local(
let children_pdus: Vec<_> = self
.get_space_child_events(current_room)
.map(PduEvent::into_stripped_spacechild_state_event)
.map(Event::into_format)
.collect()
.await;
@ -511,7 +511,7 @@ async fn cache_insert(
room_id: room_id.clone(),
children_state: self
.get_space_child_events(&room_id)
.map(PduEvent::into_stripped_spacechild_state_event)
.map(Event::into_format)
.collect()
.await,
encryption,

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

@ -1,8 +1,8 @@
use std::{collections::HashMap, fmt::Write, iter::once, sync::Arc};
use async_trait::async_trait;
use conduwuit::{
PduEvent, Result, err,
use conduwuit_core::{
Event, PduEvent, Result, err,
result::FlatOk,
state_res::{self, StateMap},
utils::{
@ -11,7 +11,7 @@ use conduwuit::{
},
warn,
};
use database::{Deserialized, Ignore, Interfix, Map};
use conduwuit_database::{Deserialized, Ignore, Interfix, Map};
use futures::{
FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future::join_all, pin_mut,
};
@ -319,30 +319,34 @@ impl Service {
}
#[tracing::instrument(skip_all, level = "debug")]
pub async fn summary_stripped(&self, event: &PduEvent) -> Vec<Raw<AnyStrippedStateEvent>> {
pub async fn summary_stripped<'a, E>(&self, event: &'a E) -> Vec<Raw<AnyStrippedStateEvent>>
where
E: Event + Send + Sync,
&'a E: Event + Send,
{
let cells = [
(&StateEventType::RoomCreate, ""),
(&StateEventType::RoomJoinRules, ""),
(&StateEventType::RoomCanonicalAlias, ""),
(&StateEventType::RoomName, ""),
(&StateEventType::RoomAvatar, ""),
(&StateEventType::RoomMember, event.sender.as_str()), // Add recommended events
(&StateEventType::RoomMember, event.sender().as_str()), // Add recommended events
(&StateEventType::RoomEncryption, ""),
(&StateEventType::RoomTopic, ""),
];
let fetches = cells.iter().map(|(event_type, state_key)| {
let fetches = cells.into_iter().map(|(event_type, state_key)| {
self.services
.state_accessor
.room_state_get(&event.room_id, event_type, state_key)
.room_state_get(event.room_id(), event_type, state_key)
});
join_all(fetches)
.await
.into_iter()
.filter_map(Result::ok)
.map(PduEvent::into_stripped_state_event)
.chain(once(event.to_stripped_state_event()))
.map(Event::into_format)
.chain(once(event.to_format()))
.collect()
}

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

@ -1,7 +1,7 @@
use std::{collections::BTreeMap, sync::Arc};
use conduwuit::{
Result, err,
use conduwuit_core::{
Event, Result, err,
matrix::pdu::{PduCount, PduEvent, PduId, RawPduId},
utils::{
ReadyExt,
@ -49,7 +49,11 @@ impl crate::Service for Service {
}
impl Service {
pub async fn add_to_thread(&self, root_event_id: &EventId, pdu: &PduEvent) -> Result<()> {
pub async fn add_to_thread<'a, E>(&self, root_event_id: &EventId, event: &'a E) -> Result
where
E: Event + Send + Sync,
&'a E: Event + Send,
{
let root_id = self
.services
.timeline
@ -86,7 +90,7 @@ impl Service {
}) {
// Thread already existed
relations.count = relations.count.saturating_add(uint!(1));
relations.latest_event = pdu.to_message_like_event();
relations.latest_event = event.to_format();
let content = serde_json::to_value(relations).expect("to_value always works");
@ -99,7 +103,7 @@ impl Service {
} else {
// New thread
let relations = BundledThread {
latest_event: pdu.to_message_like_event(),
latest_event: event.to_format(),
count: uint!(1),
current_user_participated: true,
};
@ -129,7 +133,7 @@ impl Service {
users.push(root_pdu.sender);
},
}
users.push(pdu.sender.clone());
users.push(event.sender().to_owned());
self.update_participants(&root_id, &users)
}

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

@ -375,8 +375,6 @@ impl Service {
.await
.unwrap_or_default();
let sync_pdu = pdu.to_sync_room_event();
let mut push_target: HashSet<_> = self
.services
.state_cache
@ -401,6 +399,7 @@ impl Service {
}
}
let serialized = pdu.to_format();
for user in &push_target {
let rules_for_user = self
.services
@ -418,7 +417,7 @@ impl Service {
for action in self
.services
.pusher
.get_actions(user, &rules_for_user, &power_levels, &sync_pdu, &pdu.room_id)
.get_actions(user, &rules_for_user, &power_levels, &serialized, &pdu.room_id)
.await
{
match action {
@ -768,7 +767,7 @@ impl Service {
let auth_fetch = |k: &StateEventType, s: &str| {
let key = (k.clone(), s.into());
ready(auth_events.get(&key))
ready(auth_events.get(&key).map(ToOwned::to_owned))
};
let auth_check = state_res::auth_check(

6
src/service/sending/sender.rs
View file

@ -9,8 +9,8 @@ use std::{
};
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use conduwuit::{
Error, Result, debug, err, error,
use conduwuit_core::{
Error, Event, Result, debug, err, error,
result::LogErr,
trace,
utils::{
@ -697,7 +697,7 @@ impl Service {
match event {
| SendingEvent::Pdu(pdu_id) => {
if let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await {
pdu_jsons.push(pdu.into_room_event());
pdu_jsons.push(pdu.to_format());
}
},
| SendingEvent::Edu(edu) =>