feat: Add bundled aggregations support

Add support for the m.replace and m.reference bundled
aggregations.
This should fix plenty of subtle client issues.
Threads are not included in the new code as they have
historically been written to the database. Replacing the
old system would result in issues when switching away from
continuwuity, so saved for later.
Some TODOs have been left re event visibility and ignored users.
These should be OK for now, though.

src/api/client/context.rs

@@ -93,7 +93,14 @@ pub(crate) async fn get_context_route(
 		.ignore_err()
 		.then(async |mut pdu| {
 			pdu.1.set_unsigned(Some(sender_user));
-			// TODO: bundled aggregations
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations: {e}");
+			}
 			pdu
 		})
 		.ready_filter_map(|item| event_filter(item, filter))
@@ -109,7 +116,14 @@ pub(crate) async fn get_context_route(
 		.ignore_err()
 		.then(async |mut pdu| {
 			pdu.1.set_unsigned(Some(sender_user));
-			// TODO: bundled aggregations
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations: {e}");
+			}
 			pdu
 		})
 		.ready_filter_map(|item| event_filter(item, filter))

src/api/client/message.rs

@@ -2,7 +2,7 @@ use core::panic;
 
 use axum::extract::State;
 use conduwuit::{
-	Err, Result, at,
+	Err, Result, at, debug_warn,
 	matrix::{
 		Event,
 		pdu::{PduCount, PduEvent},
@@ -134,7 +134,14 @@ pub(crate) async fn get_message_events_route(
 		.take(limit)
 		.then(async |mut pdu| {
 			pdu.1.set_unsigned(Some(sender_user));
-			// TODO: bundled aggregations
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations: {e}");
+			}
 			pdu
 		})
 		.collect()

src/api/client/relations.rs

@@ -1,6 +1,6 @@
 use axum::extract::State;
 use conduwuit::{
-	Result, at,
+	Result, at, debug_warn,
 	matrix::pdu::PduCount,
 	utils::{IterStream, ReadyExt, result::FlatOk, stream::WidebandExt},
 };
@@ -149,6 +149,17 @@ async fn paginate_relations_with_filter(
 		.ready_take_while(|(count, _)| Some(*count) != to)
 		.wide_filter_map(|item| visibility_filter(services, sender_user, item))
 		.take(limit)
+		.then(async |mut pdu| {
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations to relation: {e}");
+			}
+			pdu
+		})
 		.collect()
 		.await;
 

src/api/client/room/event.rs

@@ -1,5 +1,5 @@
 use axum::extract::State;
-use conduwuit::{Err, Event, Result, err};
+use conduwuit::{Err, Event, Result, debug_warn, err};
 use futures::{FutureExt, TryFutureExt, future::try_join};
 use ruma::api::client::room::get_room_event;
 
@@ -38,6 +38,15 @@ pub(crate) async fn get_room_event_route(
 		"Fetched PDU must match requested"
 	);
 
+	if let Err(e) = services
+		.rooms
+		.pdu_metadata
+		.add_bundled_aggregations_to_pdu(body.sender_user(), &mut event)
+		.await
+	{
+		debug_warn!("Failed to add bundled aggregations to event: {e}");
+	}
+
 	event.set_unsigned(body.sender_user.as_deref());
 
 	Ok(get_room_event::v3::Response { event: event.into_room_event() })

src/api/client/room/initial_sync.rs

@@ -1,6 +1,6 @@
 use axum::extract::State;
 use conduwuit::{
-	Err, PduEvent, Result, at,
+	Err, PduEvent, Result, at, debug_warn,
 	utils::{BoolExt, stream::TryTools},
 };
 use futures::TryStreamExt;
@@ -35,7 +35,16 @@ pub(crate) async fn room_initial_sync_route(
 		.try_take(limit)
 		.and_then(async |mut pdu| {
 			pdu.1.set_unsigned(body.sender_user.as_deref());
-			// TODO: bundled aggregations
+			if let Some(sender_user) = body.sender_user.as_deref() {
+				if let Err(e) = services
+					.rooms
+					.pdu_metadata
+					.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+					.await
+				{
+					debug_warn!("Failed to add bundled aggregations: {e}");
+				}
+			}
 			Ok(pdu)
 		})
 		.try_collect()

src/api/client/search.rs

@@ -2,7 +2,7 @@ use std::collections::BTreeMap;
 
 use axum::extract::State;
 use conduwuit::{
-	Err, Result, at, is_true,
+	Err, Result, at, debug_warn, is_true,
 	matrix::pdu::PduEvent,
 	result::FlatOk,
 	utils::{IterStream, stream::ReadyExt},
@@ -144,6 +144,17 @@ async fn category_room_events(
 		.map(at!(2))
 		.flatten()
 		.stream()
+		.then(|mut pdu| async {
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations to search result: {e}");
+			}
+			pdu
+		})
 		.map(PduEvent::into_room_event)
 		.map(|result| SearchResult {
 			rank: None,

src/api/client/sync/mod.rs

@@ -3,7 +3,7 @@ mod v4;
 mod v5;
 
 use conduwuit::{
-	Error, PduCount, Result,
+	Error, PduCount, Result, debug_warn,
 	matrix::pdu::PduEvent,
 	utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
 };
@@ -42,13 +42,23 @@ async fn load_timeline(
 		.timeline
 		.pdus_rev(room_id, None)
 		.ignore_err()
+		.ready_skip_while(|&(pducount, _)| pducount > next_batch.unwrap_or_else(PduCount::max))
+		.ready_take_while(|&(pducount, _)| pducount > roomsincecount)
 		.map(move |mut pdu| {
 			pdu.1.set_unsigned(Some(sender_user));
-			// TODO: bundled aggregations
 			pdu
 		})
-		.ready_skip_while(|&(pducount, _)| pducount > next_batch.unwrap_or_else(PduCount::max))
-		.ready_take_while(|&(pducount, _)| pducount > roomsincecount);
+		.then(async move |mut pdu| {
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(sender_user, &mut pdu.1)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations: {e}");
+			}
+			pdu
+		});
 
 	// Take the last events for the timeline
 	pin_mut!(non_timeline_pdus);

src/api/client/threads.rs

@@ -1,6 +1,6 @@
 use axum::extract::State;
 use conduwuit::{
-	Result, at,
+	Result, at, debug_warn,
 	matrix::pdu::{PduCount, PduEvent},
 };
 use futures::StreamExt;
@@ -28,7 +28,6 @@ pub(crate) async fn get_threads_route(
 		.transpose()?
 		.unwrap_or_else(PduCount::max);
 
-	// TODO: bundled aggregation
 	// TODO: user_can_see_event and set_unsigned should be at the same level /
 	// function, so unsigned is only set for seen events.
 	let threads: Vec<(PduCount, PduEvent)> = services
@@ -45,6 +44,17 @@ pub(crate) async fn get_threads_route(
 				.await
 				.then_some((count, pdu))
 		})
+		.then(|(count, mut pdu)| async move {
+			if let Err(e) = services
+				.rooms
+				.pdu_metadata
+				.add_bundled_aggregations_to_pdu(body.sender_user(), &mut pdu)
+				.await
+			{
+				debug_warn!("Failed to add bundled aggregations to thread: {e}");
+			}
+			(count, pdu)
+		})
 		.collect()
 		.await;