diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 0a384717f..20da3572c 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -581,28 +581,20 @@ async fn handle_edu_receipt_room_user( } let data = &user_updates.data; - user_updates - .event_ids - .into_iter() - .stream() - .for_each_concurrent(automatic_width(), |event_id| async move { - let user_data = [(user_id.to_owned(), data.clone())]; - let receipts = [(ReceiptType::Read, BTreeMap::from(user_data))]; - let content = [(event_id.clone(), BTreeMap::from(receipts))]; - services - .rooms - .read_receipt - .readreceipt_update( - user_id, - room_id, - ReceiptEvent::new( - room_id.to_owned(), - ReceiptEventContent::from_iter(content), - ), - ) - .await; - }) - .await; + for event_id in user_updates.event_ids { + let user_data = [(user_id.to_owned(), data.clone())]; + let receipts = [(ReceiptType::Read, BTreeMap::from(user_data))]; + let content = [(event_id, BTreeMap::from(receipts))]; + services + .rooms + .read_receipt + .readreceipt_update( + user_id, + room_id, + ReceiptEvent::new(room_id.to_owned(), ReceiptEventContent::from_iter(content)), + ) + .await; + } } async fn handle_edu_typing(services: &Services, origin: &ServerName, typing: TypingContent) { diff --git a/src/service/rooms/read_receipt/data.rs b/src/service/rooms/read_receipt/data.rs index bcffd708c..3464d4c33 100644 --- a/src/service/rooms/read_receipt/data.rs +++ b/src/service/rooms/read_receipt/data.rs @@ -45,20 +45,67 @@ impl Data { user_id: &UserId, room_id: &RoomId, event: &ReceiptEvent, - ) { - // Remove old entry + ) -> bool { + let new_target = event.content.0.iter().find_map(|(ev_id, types)| { + types + .iter() + .find_map(|(rtype, users)| users.get(user_id).map(|r| (ev_id, rtype, &r.thread))) + }); + + // Find old entries for this user in this room let last_possible_key = (room_id, u64::MAX); + let mut keys = Vec::new(); self.readreceiptid_readreceipt .rev_keys_from_raw(&last_possible_key) .ignore_err() .ready_take_while(|key| key.starts_with(room_id.as_bytes())) - .ready_filter_map(|key| key.ends_with(user_id.as_bytes()).then_some(key)) - .ready_for_each(|key| self.readreceiptid_readreceipt.del(key)) + .ready_filter_map(|key| key.ends_with(user_id.as_bytes()).then(|| key.to_vec())) + .ready_for_each(|key| keys.push(key)) .await; + let mut to_del = Vec::new(); + let mut is_dup = false; + + for key in keys { + if let Ok(val) = self.readreceiptid_readreceipt.get(&key).await { + if let Ok(existing) = serde_json::from_slice::(&val) { + let existing_target = existing.content.0.iter().find_map(|(ev_id, types)| { + types.iter().find_map(|(rtype, users)| { + users.get(user_id).map(|r| (ev_id, rtype, &r.thread)) + }) + }); + + if let (Some(new_t), Some(old_t)) = (new_target, existing_target) { + if new_t.1 == old_t.1 && new_t.2 == old_t.2 { + if new_t.0 == old_t.0 { + is_dup = true; + } else { + to_del.push(key); + } + } + } else { + to_del.push(key); + } + } else { + to_del.push(key); + } + } else { + to_del.push(key); + } + } + + if is_dup { + return false; + } + + for key in to_del { + self.readreceiptid_readreceipt.del(&key); + } + let count = self.services.globals.next_count().unwrap(); let latest_id = (room_id, count, user_id); self.readreceiptid_readreceipt.put(latest_id, Json(event)); + true } pub(super) fn readreceipts_since<'a>( @@ -86,12 +133,28 @@ impl Data { .ignore_err() } - pub(super) fn private_read_set(&self, room_id: &RoomId, user_id: &UserId, pdu_count: u64) { + pub(super) async fn private_read_set( + &self, + room_id: &RoomId, + user_id: &UserId, + pdu_count: u64, + ) -> bool { let key = (room_id, user_id); - let next_count = self.services.globals.next_count().unwrap(); + if let Ok(current) = self + .roomuserid_privateread + .qry(&key) + .await + .deserialized::() + { + if current == pdu_count { + return false; + } + } + let next_count = self.services.globals.next_count().unwrap(); self.roomuserid_privateread.put(key, pdu_count); self.roomuserid_lastprivatereadupdate.put(key, next_count); + true } pub(super) async fn private_read_get_count( diff --git a/src/service/rooms/read_receipt/mod.rs b/src/service/rooms/read_receipt/mod.rs index 097c22b22..c463e264e 100644 --- a/src/service/rooms/read_receipt/mod.rs +++ b/src/service/rooms/read_receipt/mod.rs @@ -65,7 +65,10 @@ impl Service { room_id: &RoomId, event: ReceiptEvent, ) { - self.db.readreceipt_update(user_id, room_id, &event).await; + if !self.db.readreceipt_update(user_id, room_id, &event).await { + return; + } + self.services.sync.wake_all_joined(room_id).await; // update appservices let edu = EphemeralData::Receipt(event); @@ -142,7 +145,9 @@ impl Service { event_id: &EventId, count: u64, ) { - self.db.private_read_set(room_id, user_id, count); + if !self.db.private_read_set(room_id, user_id, count).await { + return; + } // update appservices matching the user's namespace (MSC2409) let receipt_content = [(