feat: Send private read receipts to appservices

src/api/client/read_marker.rs

@@ -103,7 +103,8 @@ pub(crate) async fn set_read_marker_route(
 		services
 			.rooms
 			.read_receipt
-			.private_read_set(&body.room_id, sender_user, count);
+			.private_read_set(&body.room_id, sender_user, event, count)
+			.await;
 	}
 
 	services.sync.wake(sender_user).await;
@@ -200,7 +201,8 @@ pub(crate) async fn create_receipt_route(
 			services
 				.rooms
 				.read_receipt
-				.private_read_set(&body.room_id, sender_user, count);
+				.private_read_set(&body.room_id, sender_user, &body.event_id, count)
+				.await;
 		},
 		| _ => {
 			return Err!(Request(InvalidParam(warn!(

src/service/rooms/read_receipt/mod.rs

@@ -13,7 +13,7 @@
 };
 use futures::{Stream, TryFutureExt, try_join};
 use ruma::{
-	OwnedEventId, OwnedUserId, RoomId, UserId,
+	EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedUserId, RoomId, UserId,
 	api::appservice::event::push_events::v1::EphemeralData,
 	events::{
 		AnySyncEphemeralRoomEvent, SyncEphemeralRoomEvent,
@@ -133,11 +133,44 @@ pub fn readreceipts_since<'a>(
 		self.db.readreceipts_since(room_id, since.unwrap_or(0))
 	}
 
-	/// Sets a private read marker at PDU `count`.
-	#[inline]
+	/// Sets a private read marker at PDU `count` and notifies interested
+	/// appservices matching the user's namespace.
 	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn private_read_set(&self, room_id: &RoomId, user_id: &UserId, count: u64) {
+	pub async fn private_read_set(
+		&self,
+		room_id: &RoomId,
+		user_id: &UserId,
+		event_id: &EventId,
+		count: u64,
+	) {
 		self.db.private_read_set(room_id, user_id, count);
+
+		// update appservices matching the user's namespace (MSC2409)
+		let receipt_content = [(
+			event_id.to_owned(),
+			BTreeMap::from_iter([(
+				ruma::events::receipt::ReceiptType::ReadPrivate,
+				BTreeMap::from_iter([(
+					user_id.to_owned(),
+					ruma::events::receipt::Receipt::new(MilliSecondsSinceUnixEpoch::now()),
+				)]),
+			)]),
+		)];
+		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)
+			})
+			.await
+			.log_err();
 	}
 
 	/// Returns the private read marker PDU count.

src/service/rooms/timeline/append.rs

@@ -423,7 +423,8 @@ pub async fn append_pdu<'a, Leaves>(
 
 		self.services
 			.read_receipt
-			.private_read_set(room_id, pdu.sender(), count1);
+			.private_read_set(room_id, pdu.sender(), pdu.event_id(), count1)
+			.await;
 
 		self.services
 			.user

src/service/sending/mod.rs

@@ -37,7 +37,7 @@
 };
 use crate::{
 	Dep, account_data,
-	appservice::NamespaceRegex,
+	appservice::{NamespaceRegex, RegistrationInfo},
 	client,
 	federation::{self, FederationPathBuilderInput},
 	globals, presence, pusher,
@@ -251,13 +251,27 @@ pub async fn send_edu_appservice_room(
 		room_id: &RoomId,
 		serialized: EduBuf,
 	) -> Result<()> {
+		self.send_edu_appservice_room_filtered(room_id, serialized, |_| true)
+			.await
+	}
+
+	#[tracing::instrument(skip(self, room_id, serialized, filter), level = "debug")]
+	pub async fn send_edu_appservice_room_filtered<F>(
+		&self,
+		room_id: &RoomId,
+		serialized: EduBuf,
+		filter: F,
+	) -> Result<()>
+	where
+		F: Fn(&RegistrationInfo) -> bool + Send,
+	{
 		let appservices: Vec<_> = self
 			.services
 			.appservice
 			.read()
 			.await
 			.values()
-			.filter(|appservice| appservice.registration.receive_ephemeral)
+			.filter(|appservice| appservice.registration.receive_ephemeral && filter(appservice))
 			.cloned()
 			.collect();