From 4b4918d677b2cb9a0df6da67154723710cafefd6 Mon Sep 17 00:00:00 2001 From: Ginger Date: Wed, 4 Nov 2026 14:56:22 -0500 Subject: [PATCH] refactor(sending): Stop passing byte buffers(?!) to EDU send methods --- src/api/client/to_device.rs | 13 ++------ src/service/rooms/read_receipt/mod.rs | 28 ++++++---------- src/service/rooms/typing/mod.rs | 15 +++------ src/service/sending/mod.rs | 48 +++++++++++++++------------ 4 files changed, 44 insertions(+), 60 deletions(-) diff --git a/src/api/client/to_device.rs b/src/api/client/to_device.rs index 1659d8c00..560e00d21 100644 --- a/src/api/client/to_device.rs +++ b/src/api/client/to_device.rs @@ -2,7 +2,6 @@ use std::collections::BTreeMap; use axum::extract::State; use conduwuit::{Err, Result}; -use conduwuit_service::sending::EduBuf; use futures::StreamExt; use ruma::{ api::{ @@ -44,9 +43,8 @@ pub(crate) async fn send_event_to_device_route( messages.insert(target_user_id.clone(), map); let count = services.globals.next_count()?; - let mut buf = EduBuf::new(); - serde_json::to_writer( - &mut buf, + services.sending.send_edu_server( + target_user_id.server_name(), &Edu::DirectToDevice(assign!( DirectDeviceContent::new( sender_user.to_owned(), @@ -55,12 +53,7 @@ pub(crate) async fn send_event_to_device_route( ), { messages } )), - ) - .expect("DirectToDevice EDU can be serialized"); - - services - .sending - .send_edu_server(target_user_id.server_name(), buf)?; + )?; continue; } diff --git a/src/service/rooms/read_receipt/mod.rs b/src/service/rooms/read_receipt/mod.rs index 2815828fd..ddc4d9d65 100644 --- a/src/service/rooms/read_receipt/mod.rs +++ b/src/service/rooms/read_receipt/mod.rs @@ -23,11 +23,7 @@ use ruma::{ }; use self::data::{Data, ReceiptItem}; -use crate::{ - Dep, rooms, - sending::{self, EduBuf}, - sync, -}; +use crate::{Dep, rooms, sending, sync}; pub struct Service { services: Services, @@ -69,12 +65,10 @@ impl Service { self.services.sync.wake_all_joined(room_id).await; // update appservices let edu = EphemeralData::Receipt(event); - let mut buf = EduBuf::new(); - serde_json::to_writer(&mut buf, &edu).expect("Serialized EphemeralData::Receipt"); _ = self .services .sending - .send_edu_appservice_room(room_id, buf) + .send_edu_appservice_room(room_id, &edu) .await .log_err(); } @@ -156,19 +150,17 @@ impl Service { )]), )]), )]; - let event = ReceiptEvent::new( - room_id.to_owned(), - ReceiptEventContent::from_iter(receipt_content), - ); - let edu = EphemeralData::Receipt(event); - let mut buf = EduBuf::new(); - serde_json::to_writer(&mut buf, &edu).expect("Serialized EphemeralData::Receipt"); _ = self .services .sending - .send_edu_appservice_room_filtered(room_id, buf, |appservice| { - appservice.is_user_match(user_id) - }) + .send_edu_appservice_room_filtered( + room_id, + &EphemeralData::Receipt(ReceiptEvent::new( + room_id.to_owned(), + ReceiptEventContent::from_iter(receipt_content), + )), + |appservice| appservice.is_user_match(user_id), + ) .await .log_err(); } diff --git a/src/service/rooms/typing/mod.rs b/src/service/rooms/typing/mod.rs index 0cd6385fa..0f5e5e08e 100644 --- a/src/service/rooms/typing/mod.rs +++ b/src/service/rooms/typing/mod.rs @@ -15,7 +15,7 @@ use ruma::{ events::{EphemeralRoomEvent, SyncEphemeralRoomEvent, typing::TypingEventContent}, }; -use crate::{Dep, globals, sending, sending::EduBuf, sync, users}; +use crate::{Dep, globals, sending, sync, users}; pub struct Service { server: Arc, @@ -244,13 +244,9 @@ impl Service { return Ok(()); } - let content = TypingContent::new(room_id.to_owned(), user_id.to_owned(), typing); - let edu = Edu::Typing(content); + let edu = Edu::Typing(TypingContent::new(room_id.to_owned(), user_id.to_owned(), typing)); - let mut buf = EduBuf::new(); - serde_json::to_writer(&mut buf, &edu).expect("Serialized Edu::Typing"); - - self.services.sending.send_edu_room(room_id, buf).await?; + self.services.sending.send_edu_room(room_id, &edu).await?; Ok(()) } @@ -261,12 +257,9 @@ impl Service { self.typings_content(room_id), )); - let mut buf = EduBuf::new(); - serde_json::to_writer(&mut buf, &edu).expect("Serialized Edu::Typing"); - self.services .sending - .send_edu_appservice_room(room_id, buf) + .send_edu_appservice_room(room_id, &edu) .await } } diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index cc518f5a5..e7257103d 100644 --- a/src/service/sending/mod.rs +++ b/src/service/sending/mod.rs @@ -24,8 +24,9 @@ use ruma::{ OwnedServerName, RoomId, ServerName, UserId, api::{ OutgoingRequest, + appservice::event::push_events::v1::EphemeralData, auth_scheme::{NoAccessToken, NoAuthentication, SendAccessToken}, - federation::authentication::ServerSignatures, + federation::{authentication::ServerSignatures, transactions::edu::Edu}, path_builder::PathBuilder, }, }; @@ -220,10 +221,12 @@ impl Service { Ok(()) } - #[tracing::instrument(skip(self, server, serialized), level = "debug")] - pub fn send_edu_server(&self, server: &ServerName, serialized: EduBuf) -> Result { + #[tracing::instrument(skip(self, server, edu), level = "debug")] + pub fn send_edu_server(&self, server: &ServerName, edu: &Edu) -> Result { let dest = Destination::Federation(server.to_owned()); - let event = SendingEvent::Edu(serialized); + let mut buf = EduBuf::new(); + serde_json::to_writer(&mut buf, edu).expect("should be able to serialize edu"); + let event = SendingEvent::Edu(buf); let _cork = self.db.db.cork(); let keys = self.db.queue_requests(once((&event, &dest))); self.dispatch(Msg { @@ -233,10 +236,12 @@ impl Service { }) } - #[tracing::instrument(skip(self, serialized), level = "debug")] - pub fn send_edu_appservice(&self, appservice_id: &str, serialized: EduBuf) -> Result { + #[tracing::instrument(skip(self, edu), level = "debug")] + pub fn send_edu_appservice(&self, appservice_id: &str, edu: &EphemeralData) -> Result { let dest = Destination::Appservice(appservice_id.to_owned()); - let event = SendingEvent::Edu(serialized); + let mut buf = EduBuf::new(); + serde_json::to_writer(&mut buf, edu).expect("should be able to serialize edu"); + let event = SendingEvent::Edu(buf); let _cork = self.db.db.cork(); let keys = self.db.queue_requests(once((&event, &dest))); self.dispatch(Msg { @@ -246,21 +251,21 @@ impl Service { }) } - #[tracing::instrument(skip(self, room_id, serialized), level = "debug")] + #[tracing::instrument(skip(self, room_id, edu), level = "debug")] pub async fn send_edu_appservice_room( &self, room_id: &RoomId, - serialized: EduBuf, + edu: &EphemeralData, ) -> Result<()> { - self.send_edu_appservice_room_filtered(room_id, serialized, |_| true) + self.send_edu_appservice_room_filtered(room_id, edu, |_| true) .await } - #[tracing::instrument(skip(self, room_id, serialized, filter), level = "debug")] + #[tracing::instrument(skip(self, room_id, edu, filter), level = "debug")] pub async fn send_edu_appservice_room_filtered( &self, room_id: &RoomId, - serialized: EduBuf, + edu: &EphemeralData, filter: F, ) -> Result<()> where @@ -292,7 +297,7 @@ impl Service { || matching_aliases(appservice.aliases.clone()).await { _ = self - .send_edu_appservice(&appservice.registration.id, serialized.clone()) + .send_edu_appservice(&appservice.registration.id, edu) .inspect_err(|e| { warn!( "failed to send EDU to appservice {}: {e:?}", @@ -304,26 +309,27 @@ impl Service { Ok(()) } - #[tracing::instrument(skip(self, room_id, serialized), level = "debug")] - pub async fn send_edu_room(&self, room_id: &RoomId, serialized: EduBuf) -> Result { + #[tracing::instrument(skip(self, room_id, edu), level = "debug")] + pub async fn send_edu_room(&self, room_id: &RoomId, edu: &Edu) -> Result { let servers = self .services .state_cache .room_servers(room_id) .ready_filter(|server_name| !self.services.globals.server_is_ours(server_name)); - self.send_edu_servers(servers, serialized).await + self.send_edu_servers(servers, edu).await } - #[tracing::instrument(skip(self, servers, serialized), level = "debug")] - pub async fn send_edu_servers(&self, servers: S, serialized: EduBuf) -> Result + #[tracing::instrument(skip(self, servers, edu), level = "debug")] + pub async fn send_edu_servers(&self, servers: S, edu: &Edu) -> Result where S: Stream + Send, { + let mut buf = EduBuf::new(); + serde_json::to_writer(&mut buf, edu).expect("should be able to serialize edu"); + let requests = servers - .map(|server| { - (Destination::Federation(server), SendingEvent::Edu(serialized.clone())) - }) + .map(|server| (Destination::Federation(server), SendingEvent::Edu(buf.clone()))) .collect::>() .await;