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/sync/mod.rs

@@ -4,7 +4,7 @@ mod v5;
 use std::collections::VecDeque;
 
 use conduwuit::{
-	Event, PduCount, Result, err,
+	Event, PduCount, Result, debug_warn, err,
 	matrix::pdu::PduEvent,
 	ref_at, trace,
 	utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
@@ -73,12 +73,22 @@ async fn load_timeline(
 				.timeline
 				.pdus_rev(room_id, ending_count.map(|count| count.saturating_add(1)))
 				.ignore_err()
+				.ready_take_while(move |&(pducount, _)| pducount > starting_count)
 				.map(move |mut pdu| {
 					pdu.1.set_unsigned(Some(sender_user));
-					// TODO: bundled aggregations
 					pdu
 				})
-				.ready_take_while(move |&(pducount, _)| pducount > starting_count)
+				.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
+				})
 				.boxed()
 		},
 		| None => {
@@ -91,7 +101,17 @@ async fn load_timeline(
 				.ignore_err()
 				.map(move |mut pdu| {
 					pdu.1.set_unsigned(Some(sender_user));
-					// TODO: bundled aggregations
+					pdu
+				})
+				.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
 				})
 				.boxed()