SSS: implement receipts

src/api/client_server/sync/msc4186.rs

@@ -19,13 +19,15 @@ use ruma::{
     },
     events::{
         direct::DirectEventContent,
+        receipt::ReceiptEventContent,
         room::{
             create::RoomCreateEventContent,
             encryption::PossiblyRedactedRoomEncryptionEventContent,
             member::{MembershipState, RoomMemberEventContent},
         },
-        AnyStrippedStateEvent, PossiblyRedactedStateEventContent,
-        StateEventType, StrippedStateEvent, TimelineEventType,
+        AnyStrippedStateEvent, AnySyncEphemeralRoomEvent,
+        PossiblyRedactedStateEventContent, StateEventType, StrippedStateEvent,
+        SyncEphemeralRoomEvent, TimelineEventType,
     },
     room::RoomType,
     serde::Raw,
@@ -510,6 +512,60 @@ pub(crate) async fn sync_events_v5_route(
         None
     };
 
+    #[allow(clippy::if_then_some_else_none)]
+    let receipts = if body.extensions.receipts.enabled == Some(true) {
+        let mut receipts = BTreeMap::new();
+        for (room_id, todo_room) in &todo_rooms {
+            let mut event_content: BTreeMap<_, BTreeMap<_, BTreeMap<_, _>>> =
+                BTreeMap::new();
+            for x in services()
+                .rooms
+                .edus
+                .read_receipt
+                .readreceipts_since(room_id, todo_room.roomsince)
+            {
+                let Ok((_user_id, _, edu)) = x else {
+                    // invalid DB entry
+                    continue;
+                };
+                let Ok(edu) = edu.deserialize() else {
+                    // invalid EDU JSON
+                    continue;
+                };
+                let AnySyncEphemeralRoomEvent::Receipt(edu) = edu else {
+                    // wrong EDU type
+                    continue;
+                };
+
+                // merge all receipt EDUs into one
+                for (event_id, receipts) in edu.content.0 {
+                    let entry = event_content.entry(event_id).or_default();
+                    for (typ, receipts) in receipts {
+                        let entry = entry.entry(typ).or_default();
+                        for (user, receipt) in receipts {
+                            entry.insert(user, receipt);
+                        }
+                    }
+                }
+            }
+
+            if !event_content.is_empty() {
+                let Ok(event) = Raw::new(&SyncEphemeralRoomEvent {
+                    content: ReceiptEventContent(event_content),
+                }) else {
+                    continue;
+                };
+                receipts.insert(room_id.clone(), event);
+            }
+        }
+
+        Some(sync_events::v5::response::Receipts {
+            rooms: receipts,
+        })
+    } else {
+        None
+    };
+
     let mut rooms = BTreeMap::new();
     for (room_id, todo_room) in todo_rooms {
         if let Some(room) = process_room(&sender_user, &room_id, &todo_room)? {
@@ -521,9 +577,7 @@ pub(crate) async fn sync_events_v5_route(
         to_device,
         e2ee: e2ee.unwrap_or_default(),
         account_data: account_data.unwrap_or_default(),
-        receipts: sync_events::v5::response::Receipts {
-            rooms: BTreeMap::new(),
-        },
+        receipts: receipts.unwrap_or_default(),
         typing: sync_events::v5::response::Typing {
             rooms: BTreeMap::new(),
         },