Toward abstracting Pdu into trait Event.

Signed-off-by: Jason Volk <jason@zemos.net>

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -127,7 +127,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(

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -4,7 +4,7 @@ use futures::{FutureExt, StreamExt, future::ready};
 use ruma::{CanonicalJsonValue, RoomId, ServerName, events::StateEventType};
 use tuwunel_core::{
 	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,
@@ -118,7 +118,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(

src/service/rooms/search/mod.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use futures::{Stream, StreamExt};
 use ruma::{RoomId, UserId, api::client::search::search_events::v3::Criteria};
 use tuwunel_core::{
-	PduCount, PduEvent, Result,
+	Event, PduCount, PduEvent, Result,
 	arrayvec::ArrayVec,
 	implement,
 	utils::{

src/service/rooms/spaces/mod.rs

@@ -25,7 +25,7 @@ use ruma::{
 };
 use tokio::sync::{Mutex, MutexGuard};
 use tuwunel_core::{
-	Err, Error, PduEvent, Result, implement,
+	Err, Error, Event, PduEvent, Result, implement,
 	utils::{
 		IterStream,
 		future::{BoolExt, TryExtExt},
@@ -155,7 +155,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;
 
@@ -567,7 +567,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,

src/service/rooms/state/mod.rs

@@ -13,7 +13,7 @@ use ruma::{
 	serde::Raw,
 };
 use tuwunel_core::{
-	PduEvent, Result, err,
+	Event, PduEvent, Result, err,
 	result::FlatOk,
 	state_res::{self, StateMap},
 	utils::{
@@ -332,30 +332,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()
 	}
 

src/service/rooms/threads/mod.rs

@@ -7,7 +7,7 @@ use ruma::{
 };
 use serde_json::json;
 use tuwunel_core::{
-	Result, err,
+	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)
 	}

src/service/rooms/timeline/mod.rs

@@ -379,8 +379,6 @@ impl Service {
 			.await
 			.unwrap_or_default();
 
-		let sync_pdu = pdu.to_sync_room_event();
-
 		let mut push_target: HashSet<_> = self
 			.services
 			.state_cache
@@ -410,6 +408,7 @@ impl Service {
 			}
 		}
 
+		let serialized = pdu.to_format();
 		for user in &push_target {
 			let rules_for_user = self
 				.services
@@ -427,7 +426,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 {
@@ -783,7 +782,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(