Abstract Pdu filter matching into trait Event.

Abstract Pdu unsigned accessors into trait Event.

Abstract Pdu relation related into trait Event.

Abstract PDU content into trait Event.

Move event_id utils from pdu to event.

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

src/api/client/account.rs

@@ -24,7 +24,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	Err, Error, Result, debug_info, err, error, info, is_equal_to,
-	matrix::pdu::PduBuilder,
+	matrix::{Event, pdu::PduBuilder},
 	utils,
 	utils::{ReadyExt, stream::BroadbandExt},
 	warn,
@@ -934,7 +934,7 @@ pub async fn full_user_deactivate(
 				.state_accessor
 				.room_state_get(room_id, &StateEventType::RoomCreate, "")
 				.await
-				.is_ok_and(|event| event.sender == user_id);
+				.is_ok_and(|event| event.sender() == user_id);
 
 		if user_can_demote_self {
 			let mut power_levels_content = room_power_levels.unwrap_or_default();

src/api/client/directory.rs

@@ -28,6 +28,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	Err, Result, err, info,
+	matrix::Event,
 	utils::{
 		TryFutureExtExt,
 		math::Expected,
@@ -389,7 +390,7 @@ async fn user_can_publish_room(
 		.room_state_get(room_id, &StateEventType::RoomPowerLevels, "")
 		.await
 	{
-		| Ok(event) => serde_json::from_str(event.content.get())
+		| Ok(event) => serde_json::from_str(event.content().get())
 			.map_err(|_| err!(Database("Invalid event content for m.room.power_levels")))
 			.map(|content: RoomPowerLevelsEventContent| {
 				RoomPowerLevels::from(content)
@@ -402,7 +403,7 @@ async fn user_can_publish_room(
 				.room_state_get(room_id, &StateEventType::RoomCreate, "")
 				.await
 			{
-				| Ok(event) => Ok(event.sender == user_id),
+				| Ok(event) => Ok(event.sender() == user_id),
 				| _ => Err!(Request(Forbidden("User is not allowed to publish this room"))),
 			}
 		},

src/api/client/membership/invite.rs

@@ -8,7 +8,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	Err, Result, debug_error, err, info,
-	matrix::pdu::{PduBuilder, gen_event_id_canonical_json},
+	matrix::{event::gen_event_id_canonical_json, pdu::PduBuilder},
 };
 use tuwunel_service::Services;
 

src/api/client/membership/join.rs

@@ -26,7 +26,8 @@ use tuwunel_core::{
 	Err, Result, debug, debug_info, debug_warn, err, error, info,
 	matrix::{
 		StateKey,
-		pdu::{PduBuilder, PduEvent, gen_event_id, gen_event_id_canonical_json},
+		event::{gen_event_id, gen_event_id_canonical_json},
+		pdu::{PduBuilder, PduEvent},
 		state_res,
 	},
 	result::FlatOk,

src/api/client/membership/knock.rs

@@ -18,7 +18,10 @@ use ruma::{
 };
 use tuwunel_core::{
 	Err, Result, debug, debug_info, debug_warn, err, info,
-	matrix::pdu::{PduBuilder, PduEvent, gen_event_id},
+	matrix::{
+		event::{Event, gen_event_id},
+		pdu::{PduBuilder, PduEvent},
+	},
 	result::FlatOk,
 	trace,
 	utils::{self, shuffle, stream::IterStream},

src/api/client/membership/leave.rs

@@ -15,7 +15,7 @@ use ruma::{
 };
 use tuwunel_core::{
 	Err, Result, debug_info, debug_warn, err,
-	matrix::pdu::{PduBuilder, gen_event_id},
+	matrix::{event::gen_event_id, pdu::PduBuilder},
 	utils::{self, FutureBoolExt, future::ReadyEqExt},
 	warn,
 };

src/api/client/membership/members.rs

@@ -1,5 +1,5 @@
 use axum::extract::State;
-use futures::{StreamExt, future::join};
+use futures::{FutureExt, StreamExt, future::join};
 use ruma::{
 	api::client::membership::{
 		get_member_events::{self, v3::MembershipEventFilter},
@@ -11,8 +11,8 @@ use ruma::{
 	},
 };
 use tuwunel_core::{
-	Err, Event, Result, at,
-	matrix::pdu::PduEvent,
+	Err, Result, at,
+	matrix::Event,
 	utils::{
 		future::TryExtExt,
 		stream::{BroadbandExt, ReadyExt},
@@ -55,6 +55,7 @@ pub(crate) async fn get_member_events_route(
 			.ready_filter_map(|pdu| membership_filter(pdu, membership, not_membership))
 			.map(Event::into_format)
 			.collect()
+			.boxed()
 			.await,
 	})
 }
@@ -98,11 +99,11 @@ pub(crate) async fn joined_members_route(
 	})
 }
 
-fn membership_filter(
-	pdu: PduEvent,
+fn membership_filter<Pdu: Event>(
+	pdu: Pdu,
 	for_membership: Option<&MembershipEventFilter>,
 	not_membership: Option<&MembershipEventFilter>,
-) -> Option<PduEvent> {
+) -> Option<impl Event> {
 	let membership_state_filter = match for_membership {
 		| Some(MembershipEventFilter::Ban) => MembershipState::Ban,
 		| Some(MembershipEventFilter::Invite) => MembershipState::Invite,

src/api/client/message.rs

@@ -12,9 +12,10 @@ use ruma::{
 use tuwunel_core::{
 	Err, Result, at,
 	matrix::{
-		Event,
-		pdu::{PduCount, PduEvent},
+		event::{Event, Matches},
+		pdu::PduCount,
 	},
+	ref_at,
 	utils::{
 		IterStream, ReadyExt,
 		result::{FlatOk, LogErr},
@@ -202,7 +203,9 @@ where
 	pin_mut!(receipts);
 	let witness: Witness = events
 		.stream()
-		.map(|(_, pdu)| pdu.sender.clone())
+		.map(ref_at!(1))
+		.map(Event::sender)
+		.map(ToOwned::to_owned)
 		.chain(
 			receipts
 				.ready_take_while(|(_, c, _)| *c <= newest.into_unsigned())
@@ -247,31 +250,34 @@ pub(crate) async fn ignored_filter(
 }
 
 #[inline]
-pub(crate) async fn is_ignored_pdu(
+pub(crate) async fn is_ignored_pdu<Pdu>(
 	services: &Services,
-	pdu: &PduEvent,
+	event: &Pdu,
 	user_id: &UserId,
-) -> bool {
+) -> bool
+where
+	Pdu: Event + Send + Sync,
+{
 	// exclude Synapse's dummy events from bloating up response bodies. clients
 	// don't need to see this.
-	if pdu.kind.to_cow_str() == "org.matrix.dummy_event" {
+	if event.kind().to_cow_str() == "org.matrix.dummy_event" {
 		return true;
 	}
 
 	let ignored_type = IGNORED_MESSAGE_TYPES
-		.binary_search(&pdu.kind)
+		.binary_search(event.kind())
 		.is_ok();
 
 	let ignored_server = services
 		.config
 		.forbidden_remote_server_names
-		.is_match(pdu.sender().server_name().host());
+		.is_match(event.sender().server_name().host());
 
 	if ignored_type
 		&& (ignored_server
 			|| services
 				.users
-				.user_is_ignored(&pdu.sender, user_id)
+				.user_is_ignored(event.sender(), user_id)
 				.await)
 	{
 		return true;
@@ -291,7 +297,7 @@ pub(crate) async fn visibility_filter(
 	services
 		.rooms
 		.state_accessor
-		.user_can_see_event(user_id, &pdu.room_id, &pdu.event_id)
+		.user_can_see_event(user_id, pdu.room_id(), pdu.event_id())
 		.await
 		.then_some(item)
 }
@@ -299,7 +305,7 @@ pub(crate) async fn visibility_filter(
 #[inline]
 pub(crate) fn event_filter(item: PdusIterItem, filter: &RoomEventFilter) -> Option<PdusIterItem> {
 	let (_, pdu) = &item;
-	pdu.matches(filter).then_some(item)
+	filter.matches(pdu).then_some(item)
 }
 
 #[cfg_attr(debug_assertions, tuwunel_core::ctor)]

src/api/client/relations.rs

@@ -13,10 +13,13 @@ use ruma::{
 };
 use tuwunel_core::{
 	Result, at,
-	matrix::{Event, pdu::PduCount},
+	matrix::{
+		event::{Event, RelationTypeEqual},
+		pdu::PduCount,
+	},
 	utils::{IterStream, ReadyExt, result::FlatOk, stream::WidebandExt},
 };
-use tuwunel_service::{Services, rooms::timeline::PdusIterItem};
+use tuwunel_service::Services;
 
 use crate::Ruma;
 
@@ -129,7 +132,7 @@ async fn paginate_relations_with_filter(
 	// Spec (v1.10) recommends depth of at least 3
 	let depth: u8 = if recurse { 3 } else { 1 };
 
-	let events: Vec<PdusIterItem> = services
+	let events: Vec<_> = services
 		.rooms
 		.pdu_metadata
 		.get_relations(sender_user, room_id, target, start, limit, depth, dir)
@@ -138,12 +141,12 @@ async fn paginate_relations_with_filter(
 		.filter(|(_, pdu)| {
 			filter_event_type
 				.as_ref()
-				.is_none_or(|kind| *kind == pdu.kind)
+				.is_none_or(|kind| kind == pdu.kind())
 		})
 		.filter(|(_, pdu)| {
 			filter_rel_type
 				.as_ref()
-				.is_none_or(|rel_type| pdu.relation_type_equal(rel_type))
+				.is_none_or(|rel_type| rel_type.relation_type_equal(pdu))
 		})
 		.stream()
 		.ready_take_while(|(count, _)| Some(*count) != to)
@@ -172,17 +175,17 @@ async fn paginate_relations_with_filter(
 	})
 }
 
-async fn visibility_filter(
+async fn visibility_filter<Pdu: Event + Send + Sync>(
 	services: &Services,
 	sender_user: &UserId,
-	item: PdusIterItem,
-) -> Option<PdusIterItem> {
+	item: (PduCount, Pdu),
+) -> Option<(PduCount, Pdu)> {
 	let (_, pdu) = &item;
 
 	services
 		.rooms
 		.state_accessor
-		.user_can_see_event(sender_user, &pdu.room_id, &pdu.event_id)
+		.user_can_see_event(sender_user, pdu.room_id(), pdu.event_id())
 		.await
 		.then_some(item)
 }

src/api/client/room/initial_sync.rs

@@ -53,7 +53,9 @@ pub(crate) async fn room_initial_sync_route(
 		.try_collect::<Vec<_>>();
 
 	let (membership, visibility, state, events) =
-		try_join4(membership, visibility, state, events).await?;
+		try_join4(membership, visibility, state, events)
+			.boxed()
+			.await?;
 
 	let messages = PaginationChunk {
 		start: events

src/api/client/room/upgrade.rs

@@ -18,7 +18,7 @@ use ruma::{
 use serde_json::{json, value::to_raw_value};
 use tuwunel_core::{
 	Error, Result, err, info,
-	matrix::{StateKey, pdu::PduBuilder},
+	matrix::{Event, StateKey, pdu::PduBuilder},
 };
 
 use crate::Ruma;
@@ -226,7 +226,7 @@ pub(crate) async fn upgrade_room_route(
 			.room_state_get(&body.room_id, event_type, "")
 			.await
 		{
-			| Ok(v) => v.content.clone(),
+			| Ok(v) => v.content().to_owned(),
 			| Err(_) => continue, // Skipping missing events.
 		};