refactor(sending): Stop passing byte buffers(?!) to EDU send methods

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;
 			}

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();
 	}

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<Server>,
@@ -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
 	}
 }

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<F>(
 		&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<S>(&self, servers: S, serialized: EduBuf) -> Result
+	#[tracing::instrument(skip(self, servers, edu), level = "debug")]
+	pub async fn send_edu_servers<S>(&self, servers: S, edu: &Edu) -> Result
 	where
 		S: Stream<Item = OwnedServerName> + 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::<Vec<_>>()
 			.await;