From 5d55b413a265841aec92bbd9bd0729ac81b45dd0 Mon Sep 17 00:00:00 2001 From: Jade Ellis Date: Mon, 5 Oct 2026 15:12:54 +0100 Subject: [PATCH] feat: Send private read receipts to appservices --- src/api/client/read_marker.rs | 6 ++-- src/service/rooms/read_receipt/mod.rs | 41 ++++++++++++++++++++++++--- src/service/rooms/timeline/append.rs | 3 +- src/service/sending/mod.rs | 18 ++++++++++-- 4 files changed, 59 insertions(+), 9 deletions(-) diff --git a/src/api/client/read_marker.rs b/src/api/client/read_marker.rs index 3793cffe0..2f8b21a39 100644 --- a/src/api/client/read_marker.rs +++ b/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!( diff --git a/src/service/rooms/read_receipt/mod.rs b/src/service/rooms/read_receipt/mod.rs index 0ebc8ecd9..2815828fd 100644 --- a/src/service/rooms/read_receipt/mod.rs +++ b/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. diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 9639e6733..07389355c 100644 --- a/src/service/rooms/timeline/append.rs +++ b/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 diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index 171f4a2a3..0e25f4d3d 100644 --- a/src/service/sending/mod.rs +++ b/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( + &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();