1use std::{cmp::Ordering, collections::HashMap};
16
17use futures_core::Stream;
18use indexmap::IndexMap;
19use ruma::{
20 EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedUserId, UserId,
21 events::receipt::{Receipt, ReceiptEventContent, ReceiptThread, ReceiptType},
22};
23use tokio::sync::watch;
24use tokio_stream::wrappers::WatchStream;
25use tracing::{debug, error, instrument, trace, warn};
26
27use super::{
28 AllRemoteEvents, ObservableItemsTransaction, RelativePosition, RoomDataProvider,
29 TimelineMetadata, TimelineState, rfind_event_by_id,
30};
31use crate::timeline::{TimelineItem, controller::TimelineStateTransaction};
32
33#[derive(Clone, Debug, Default)]
35pub(super) struct ReadReceipts {
36 by_event: HashMap<OwnedEventId, IndexMap<OwnedUserId, Receipt>>,
40
41 latest_by_user: HashMap<OwnedUserId, HashMap<ReceiptType, (OwnedEventId, Receipt)>>,
46
47 own_user_read_receipts_changed_sender: watch::Sender<()>,
49}
50
51impl ReadReceipts {
52 pub(super) fn clear(&mut self) {
54 self.by_event.clear();
55 self.latest_by_user.clear();
56 }
57
58 pub(super) fn subscribe_own_user_read_receipts_changed(
60 &self,
61 ) -> impl Stream<Item = ()> + use<> {
62 let subscriber = self.own_user_read_receipts_changed_sender.subscribe();
63 WatchStream::from_changes(subscriber)
64 }
65
66 pub(crate) fn get_latest(
69 &self,
70 user_id: &UserId,
71 receipt_type: &ReceiptType,
72 ) -> Option<&(OwnedEventId, Receipt)> {
73 self.latest_by_user.get(user_id).and_then(|map| map.get(receipt_type))
74 }
75
76 fn upsert_latest(
79 &mut self,
80 user_id: OwnedUserId,
81 receipt_type: ReceiptType,
82 read_receipt: (OwnedEventId, Receipt),
83 ) {
84 self.latest_by_user.entry(user_id).or_default().insert(receipt_type, read_receipt);
85 }
86
87 #[instrument(skip_all, fields(user_id = %new_receipt.user_id, event_id = %new_receipt.event_id))]
97 fn maybe_update_read_receipt(
98 &mut self,
99 new_receipt: FullReceipt<'_>,
100 is_own_user_id: bool,
101 timeline_items: &mut ObservableItemsTransaction<'_>,
102 ) {
103 let all_events = timeline_items.all_remote_events();
104
105 let old_receipt = self.get_latest(new_receipt.user_id, &new_receipt.receipt_type);
107
108 if old_receipt
109 .as_ref()
110 .is_some_and(|(old_receipt_event_id, _)| old_receipt_event_id == new_receipt.event_id)
111 {
112 if !is_own_user_id {
114 trace!("receipt hasn't changed, nothing to do");
115 }
116 return;
117 }
118
119 let old_event_id = old_receipt.map(|(event_id, _)| event_id);
120
121 let mut old_receipt_pos = None;
123 let mut old_item_pos = None;
124 let mut old_item_event_id = None;
125 let mut new_receipt_pos = None;
126 let mut new_item_pos = None;
127 let mut new_item_event_id = None;
128
129 for (pos, event) in all_events.iter().rev().enumerate() {
130 if old_receipt_pos.is_none() && old_event_id == Some(&event.event_id) {
131 old_receipt_pos = Some(pos);
132 }
133
134 if old_receipt_pos.is_some()
137 && old_item_event_id.is_none()
138 && event.visible
139 && event.can_show_read_receipts
140 {
141 old_item_pos = event.timeline_item_index;
142 old_item_event_id = Some(event.event_id.clone());
143 }
144
145 if new_receipt_pos.is_none() && new_receipt.event_id == event.event_id {
146 new_receipt_pos = Some(pos);
147 }
148
149 if new_receipt_pos.is_some()
152 && new_item_event_id.is_none()
153 && event.visible
154 && event.can_show_read_receipts
155 {
156 new_item_pos = event.timeline_item_index;
157 new_item_event_id = Some(event.event_id.clone());
158 }
159
160 if old_item_event_id.is_some() && new_item_event_id.is_some() {
161 break;
163 }
164 }
165
166 if let Some(old_receipt_pos) = old_receipt_pos {
168 let Some(new_receipt_pos) = new_receipt_pos else {
169 if !is_own_user_id {
172 trace!(
173 "we had a previous read receipt, but couldn't find the event \
174 targeted by the new read receipt in the timeline, exiting"
175 );
176 }
177 return;
178 };
179
180 if old_receipt_pos < new_receipt_pos {
181 if !is_own_user_id {
183 trace!("the previous read receipt is more recent than the new one, exiting");
184 }
185 return;
186 }
187 }
188
189 if !is_own_user_id {
199 trace!(
200 from_event = ?old_event_id,
201 from_visible_event = ?old_item_event_id,
202 to_event = ?new_receipt.event_id,
203 to_visible_event = ?new_item_event_id,
204 ?old_item_pos,
205 ?new_item_pos,
206 "moving read receipt",
207 );
208
209 if let Some(old_event_id) = old_event_id.cloned() {
211 self.remove_event_receipt_for_user(&old_event_id, new_receipt.user_id);
212 }
213
214 self.add_event_receipt_for_user(
216 new_receipt.event_id.to_owned(),
217 new_receipt.user_id.to_owned(),
218 new_receipt.receipt.clone(),
219 );
220 }
221
222 self.upsert_latest(
224 new_receipt.user_id.to_owned(),
225 new_receipt.receipt_type,
226 (new_receipt.event_id.to_owned(), new_receipt.receipt.clone()),
227 );
228
229 if is_own_user_id {
230 self.own_user_read_receipts_changed_sender.send_replace(());
231 return;
233 }
234
235 if new_item_event_id == old_item_event_id {
236 return;
238 }
239
240 let timeline_update = ReadReceiptTimelineUpdate {
241 old_item_pos,
242 old_event_id: old_item_event_id,
243 new_item_pos,
244 new_event_id: new_item_event_id,
245 };
246
247 timeline_update.apply(
248 timeline_items,
249 new_receipt.user_id.to_owned(),
250 new_receipt.receipt.clone(),
251 );
252 }
253
254 fn get_event_receipts(&self, event_id: &EventId) -> Option<&IndexMap<OwnedUserId, Receipt>> {
256 self.by_event.get(event_id)
257 }
258
259 fn add_event_receipt_for_user(
261 &mut self,
262 event_id: OwnedEventId,
263 user_id: OwnedUserId,
264 receipt: Receipt,
265 ) {
266 self.by_event.entry(event_id).or_default().insert(user_id, receipt);
267 }
268
269 fn remove_event_receipt_for_user(&mut self, event_id: &EventId, user_id: &UserId) {
271 if let Some(map) = self.by_event.get_mut(event_id) {
272 map.swap_remove(user_id);
273 if map.is_empty() {
275 self.by_event.remove(event_id);
276 }
277 }
278 }
279
280 #[instrument(skip(self, timeline_items, at_end))]
285 pub(super) fn compute_event_receipts(
286 &self,
287 event_id: &EventId,
288 timeline_items: &mut ObservableItemsTransaction<'_>,
289 at_end: bool,
290 ) -> IndexMap<OwnedUserId, Receipt> {
291 let mut all_receipts = self.get_event_receipts(event_id).cloned().unwrap_or_default();
292
293 if at_end {
294 trace!(
296 "early return because @end, retrieved receipts: {}",
297 all_receipts.iter().map(|(u, _)| u.as_str()).collect::<Vec<_>>().join(", ")
298 );
299 return all_receipts;
300 }
301
302 trace!(
303 "loaded receipts: {}",
304 all_receipts.iter().map(|(u, _)| u.as_str()).collect::<Vec<_>>().join(", ")
305 );
306
307 let mut events_iter = timeline_items.all_remote_events().iter();
318 let mut prev_event_and_item_index = None;
319
320 for meta in events_iter.by_ref() {
321 if meta.event_id == event_id {
322 break;
323 }
324 if let Some(item_index) = meta.timeline_item_index {
325 prev_event_and_item_index = Some((meta.event_id.clone(), item_index));
326 }
327 }
328
329 let mut hidden = Vec::new();
332 for hidden_receipt_event_meta in
333 events_iter.take_while(|meta| !meta.visible || !meta.can_show_read_receipts)
334 {
335 if let Some(event_receipts) =
336 self.get_event_receipts(&hidden_receipt_event_meta.event_id)
337 {
338 trace!(%hidden_receipt_event_meta.event_id, "found receipts on hidden event");
339 hidden.extend(event_receipts.clone());
340 }
341 }
342
343 if let Some((prev_event_id, prev_item_index)) = prev_event_and_item_index {
345 let prev_item = &timeline_items[prev_item_index];
346 if let Some(remote_prev_item) = prev_item.as_event() {
350 let prev_receipts = remote_prev_item.read_receipts().clone();
351 for (user_id, _) in &hidden {
352 if !prev_receipts.contains_key(user_id) {
353 continue;
354 }
355 let mut up = ReadReceiptTimelineUpdate {
356 old_item_pos: Some(prev_item_index),
357 old_event_id: Some(prev_event_id.clone()),
358 new_item_pos: None,
359 new_event_id: None,
360 };
361 up.remove_old_receipt(timeline_items, user_id);
362 }
363 }
364 }
365
366 all_receipts.extend(hidden);
367 trace!(
368 "computed receipts: {}",
369 all_receipts.iter().map(|(u, _)| u.as_str()).collect::<Vec<_>>().join(", ")
370 );
371 all_receipts
372 }
373}
374
375struct FullReceipt<'a> {
376 event_id: &'a EventId,
377 user_id: &'a UserId,
378 receipt_type: ReceiptType,
379 receipt: &'a Receipt,
380}
381
382#[derive(Clone, Debug, Default)]
384struct ReadReceiptTimelineUpdate {
385 old_item_pos: Option<usize>,
388 old_event_id: Option<OwnedEventId>,
390 new_item_pos: Option<usize>,
393 new_event_id: Option<OwnedEventId>,
395}
396
397impl ReadReceiptTimelineUpdate {
398 #[instrument(skip_all)]
400 fn remove_old_receipt(&mut self, items: &mut ObservableItemsTransaction<'_>, user_id: &UserId) {
401 let Some(event_id) = &self.old_event_id else {
402 return;
404 };
405
406 let item_pos = self.old_item_pos.or_else(|| {
407 items
408 .iter_remotes_region()
409 .rev()
410 .filter_map(|(nth, item)| Some((nth, item.as_event()?)))
411 .find_map(|(nth, event_item)| {
412 (event_item.event_id() == Some(event_id)).then_some(nth)
413 })
414 });
415
416 let Some(item_pos) = item_pos else {
417 debug!(%event_id, %user_id, "inconsistent state: old event item for read receipt was not found");
418 return;
419 };
420
421 self.old_item_pos = Some(item_pos);
422
423 let event_item = &items[item_pos];
424 let event_item_id = event_item.unique_id().to_owned();
425
426 let Some(mut event_item) = event_item.as_event().cloned() else {
427 warn!("received a read receipt for a virtual item, this should not be possible");
428 return;
429 };
430
431 if let Some(remote_event_item) = event_item.as_remote_mut() {
432 if remote_event_item.read_receipts.swap_remove(user_id).is_none() {
433 debug!(
434 %event_id, %user_id,
435 "inconsistent state: old event item for user's read \
436 receipt doesn't have a receipt for the user"
437 );
438 }
439 trace!(%user_id, %event_id, "removed read receipt from event item");
440 items.replace(item_pos, TimelineItem::new(event_item, event_item_id));
441 } else {
442 warn!("received a read receipt for a local item, this should not be possible");
443 }
444 }
445
446 #[instrument(skip_all)]
448 fn add_new_receipt(
449 self,
450 items: &mut ObservableItemsTransaction<'_>,
451 user_id: OwnedUserId,
452 receipt: Receipt,
453 ) {
454 let Some(event_id) = self.new_event_id else {
455 return;
457 };
458
459 let old_item_pos = self.old_item_pos.unwrap_or(0);
460
461 let item_pos = self.new_item_pos.or_else(|| {
462 items
463 .iter_remotes_region()
464 .skip_while(|(nth, _)| *nth < old_item_pos)
467 .find_map(|(nth, item)| {
468 if let Some(event_item) = item.as_event() {
469 (event_item.event_id() == Some(&event_id)).then_some(nth)
470 } else {
471 None
472 }
473 })
474 });
475
476 let Some(item_pos) = item_pos else {
477 debug!(
478 %event_id, %user_id,
479 "inconsistent state: new event item for read receipt was not found",
480 );
481 return;
482 };
483
484 debug_assert!(
485 item_pos >= self.old_item_pos.unwrap_or(0),
486 "The new receipt must be added on a timeline item that is _after_ the timeline item \
487 that was holding the old receipt"
488 );
489
490 let event_item = &items[item_pos];
491 let event_item_id = event_item.unique_id().to_owned();
492
493 let Some(mut event_item) = event_item.as_event().cloned() else {
494 warn!("received a read receipt for a virtual item, this should not be possible");
495 return;
496 };
497
498 if let Some(remote_event_item) = event_item.as_remote_mut() {
499 trace!(%user_id, %event_id, "added read receipt to event item");
500 remote_event_item.read_receipts.insert(user_id, receipt);
501 items.replace(item_pos, TimelineItem::new(event_item, event_item_id));
502 } else {
503 warn!("received a read receipt for a local item, this should not be possible");
504 }
505 }
506
507 fn apply(
509 mut self,
510 items: &mut ObservableItemsTransaction<'_>,
511 user_id: OwnedUserId,
512 receipt: Receipt,
513 ) {
514 self.remove_old_receipt(items, &user_id);
515 self.add_new_receipt(items, user_id, receipt);
516 }
517}
518
519impl<P: RoomDataProvider> TimelineStateTransaction<'_, P> {
520 pub(super) fn handle_explicit_read_receipts(
521 &mut self,
522 receipt_event_content: ReceiptEventContent,
523 own_user_id: &UserId,
524 ) {
525 trace!("handling explicit read receipts");
526 let own_receipt_thread = self.focus.receipt_thread();
527
528 for (event_id, receipt_types) in receipt_event_content.0 {
529 for (receipt_type, receipts) in receipt_types {
530 if !matches!(receipt_type, ReceiptType::Read | ReceiptType::ReadPrivate) {
532 continue;
533 }
534
535 for (user_id, receipt) in receipts {
536 if matches!(own_receipt_thread, ReceiptThread::Unthreaded | ReceiptThread::Main)
537 {
538 if !matches!(
542 receipt.thread,
543 ReceiptThread::Unthreaded | ReceiptThread::Main
544 ) {
545 continue;
546 }
547 } else if own_receipt_thread != receipt.thread {
548 continue;
550 }
551
552 let is_own_user_id = user_id == own_user_id;
553 let full_receipt = FullReceipt {
554 event_id: &event_id,
555 user_id: &user_id,
556 receipt_type: receipt_type.clone(),
557 receipt: &receipt,
558 };
559
560 self.meta.read_receipts.maybe_update_read_receipt(
561 full_receipt,
562 is_own_user_id,
563 &mut self.items,
564 );
565 }
566 }
567 }
568 }
569
570 pub(super) async fn load_read_receipts_for_event(
574 &mut self,
575 event_id: &EventId,
576 room_data_provider: &P,
577 ) {
578 trace!(%event_id, "loading initial receipts for an event");
579
580 let receipt_thread = self.focus.receipt_thread();
581
582 let receipts = if matches!(receipt_thread, ReceiptThread::Unthreaded | ReceiptThread::Main)
583 {
584 let mut main_receipts =
590 room_data_provider.load_event_receipts(event_id, ReceiptThread::Main).await;
591
592 let unthreaded_receipts =
594 room_data_provider.load_event_receipts(event_id, ReceiptThread::Unthreaded).await;
595
596 main_receipts.extend(unthreaded_receipts);
600 main_receipts
601 } else {
602 room_data_provider.load_event_receipts(event_id, receipt_thread.clone()).await
605 };
606
607 let own_user_id = room_data_provider.own_user_id();
608
609 for (user_id, receipt) in receipts {
612 let full_receipt = FullReceipt {
613 event_id,
614 user_id: &user_id,
615 receipt_type: ReceiptType::Read,
616 receipt: &receipt,
617 };
618
619 self.meta.read_receipts.maybe_update_read_receipt(
620 full_receipt,
621 user_id == own_user_id,
622 &mut self.items,
623 );
624 }
625 }
626
627 pub(super) fn maybe_add_implicit_read_receipt(
636 &mut self,
637 event_id: &EventId,
638 sender: Option<&UserId>,
639 timestamp: Option<MilliSecondsSinceUnixEpoch>,
640 ) {
641 let (Some(user_id), Some(timestamp)) = (sender, timestamp) else {
642 return;
644 };
645
646 trace!(%user_id, %event_id, "adding implicit read receipt");
647
648 let mut receipt = Receipt::new(timestamp);
649 receipt.thread = self.focus.receipt_thread();
650
651 let full_receipt =
652 FullReceipt { event_id, user_id, receipt_type: ReceiptType::Read, receipt: &receipt };
653
654 let is_own_event = sender.is_some_and(|sender| sender == self.meta.own_user_id);
655
656 self.meta.read_receipts.maybe_update_read_receipt(
657 full_receipt,
658 is_own_event,
659 &mut self.items,
660 );
661 }
662
663 #[instrument(skip(self))]
666 pub(super) fn maybe_update_read_receipts_of_prev_event(&mut self, event_id: &EventId) {
667 let Some(prev_event_meta) = self
669 .items
670 .all_remote_events()
671 .iter()
672 .rev()
673 .skip_while(|meta| meta.event_id != event_id)
675 .skip(1)
677 .find(|meta| meta.visible && meta.can_show_read_receipts)
679 else {
680 trace!("Couldn't find any previous visible event, exiting");
681 return;
682 };
683
684 let Some((prev_item_pos, prev_event_item)) =
685 rfind_event_by_id(&self.items, &prev_event_meta.event_id)
686 else {
687 error!("inconsistent state: timeline item of visible event was not found");
688 return;
689 };
690
691 let prev_event_item_id = prev_event_item.internal_id.to_owned();
692 let mut prev_event_item = prev_event_item.clone();
693
694 let Some(remote_prev_event_item) = prev_event_item.as_remote_mut() else {
695 warn!("loading read receipts for a local item, this should not be possible");
696 return;
697 };
698
699 let read_receipts = self.meta.read_receipts.compute_event_receipts(
700 &remote_prev_event_item.event_id,
701 &mut self.items,
702 false,
703 );
704
705 if read_receipts.len() == remote_prev_event_item.read_receipts.len() {
707 trace!("same count of read receipts, not doing anything");
708 return;
709 }
710
711 trace!("replacing read receipts with the new ones");
712 remote_prev_event_item.read_receipts = read_receipts;
713 self.items.replace(prev_item_pos, TimelineItem::new(prev_event_item, prev_event_item_id));
714 }
715}
716
717impl<P: RoomDataProvider> TimelineState<P> {
718 pub(super) async fn populate_initial_user_receipt(
721 &mut self,
722 room_data_provider: &P,
723 receipt_type: ReceiptType,
724 ) {
725 let own_user_id = room_data_provider.own_user_id().to_owned();
726
727 let receipt_thread = self.focus.receipt_thread();
728 let wants_unthreaded_receipts = receipt_thread == ReceiptThread::Unthreaded;
729
730 let mut read_receipt = room_data_provider
731 .load_user_receipt(receipt_type.clone(), receipt_thread, &own_user_id)
732 .await;
733
734 if wants_unthreaded_receipts && read_receipt.is_none() {
735 read_receipt = room_data_provider
737 .load_user_receipt(receipt_type.clone(), ReceiptThread::Main, &own_user_id)
738 .await;
739 }
740
741 if let Some(read_receipt) = read_receipt {
742 self.meta.read_receipts.upsert_latest(own_user_id, receipt_type, read_receipt);
743 }
744 }
745
746 pub(super) async fn latest_user_read_receipt(
750 &self,
751 user_id: &UserId,
752 receipt_thread: ReceiptThread,
753 room_data_provider: &P,
754 ) -> Option<(OwnedEventId, Receipt)> {
755 let all_remote_events = self.items.all_remote_events();
756
757 let public_read_receipt = self
758 .meta
759 .user_receipt(
760 user_id,
761 ReceiptType::Read,
762 receipt_thread.clone(),
763 room_data_provider,
764 all_remote_events,
765 )
766 .await;
767
768 let private_read_receipt = self
769 .meta
770 .user_receipt(
771 user_id,
772 ReceiptType::ReadPrivate,
773 receipt_thread,
774 room_data_provider,
775 all_remote_events,
776 )
777 .await;
778
779 match TimelineMetadata::compare_optional_receipts(
783 public_read_receipt.as_ref(),
784 private_read_receipt.as_ref(),
785 all_remote_events,
786 ) {
787 Ordering::Greater => public_read_receipt,
788 Ordering::Less => private_read_receipt,
789 _ => unreachable!(),
790 }
791 }
792
793 pub(super) fn latest_user_read_receipt_timeline_event_id(
796 &self,
797 user_id: &UserId,
798 ) -> Option<OwnedEventId> {
799 let public_read_receipt = self.meta.read_receipts.get_latest(user_id, &ReceiptType::Read);
802 let private_read_receipt =
803 self.meta.read_receipts.get_latest(user_id, &ReceiptType::ReadPrivate);
804
805 let (latest_receipt_id, _) = match TimelineMetadata::compare_optional_receipts(
809 public_read_receipt,
810 private_read_receipt,
811 self.items.all_remote_events(),
812 ) {
813 Ordering::Greater => public_read_receipt?,
814 Ordering::Less => private_read_receipt?,
815 _ => unreachable!(),
816 };
817
818 self.items
820 .all_remote_events()
821 .iter()
822 .rev()
823 .skip_while(|ev| ev.event_id != *latest_receipt_id)
824 .find(|ev| ev.visible && ev.can_show_read_receipts)
825 .map(|ev| ev.event_id.clone())
826 }
827}
828
829impl TimelineMetadata {
830 pub(super) async fn user_receipt<P: RoomDataProvider>(
841 &self,
842 user_id: &UserId,
843 receipt_type: ReceiptType,
844 receipt_thread: ReceiptThread,
845 room_data_provider: &P,
846 all_remote_events: &AllRemoteEvents,
847 ) -> Option<(OwnedEventId, Receipt)> {
848 if let Some(receipt) = self.read_receipts.get_latest(user_id, &receipt_type) {
849 return Some(receipt.clone());
851 }
852
853 if receipt_thread == ReceiptThread::Unthreaded {
854 let unthreaded_read_receipt = room_data_provider
857 .load_user_receipt(receipt_type.clone(), ReceiptThread::Unthreaded, user_id)
858 .await;
859
860 let main_thread_read_receipt = room_data_provider
861 .load_user_receipt(receipt_type.clone(), ReceiptThread::Main, user_id)
862 .await;
863
864 match Self::compare_optional_receipts(
867 main_thread_read_receipt.as_ref(),
868 unthreaded_read_receipt.as_ref(),
869 all_remote_events,
870 ) {
871 Ordering::Greater => main_thread_read_receipt,
872 Ordering::Less => unthreaded_read_receipt,
873 _ => unreachable!(),
874 }
875 } else {
876 room_data_provider
880 .load_user_receipt(receipt_type.clone(), receipt_thread, user_id)
881 .await
882 }
883 }
884
885 fn compare_optional_receipts(
892 lhs: Option<&(OwnedEventId, Receipt)>,
893 rhs_or_default: Option<&(OwnedEventId, Receipt)>,
894 all_remote_events: &AllRemoteEvents,
895 ) -> Ordering {
896 let Some((lhs_event_id, lhs_receipt)) = lhs else {
898 return Ordering::Less;
899 };
900 let Some((rhs_event_id, rhs_receipt)) = rhs_or_default else {
901 return Ordering::Greater;
902 };
903
904 if let Some(relative_pos) =
906 Self::compare_events_positions(lhs_event_id, rhs_event_id, all_remote_events)
907 {
908 if relative_pos == RelativePosition::Before {
909 return Ordering::Greater;
910 }
911
912 return Ordering::Less;
913 }
914
915 if let Some((lhs_ts, rhs_ts)) = lhs_receipt.ts.zip(rhs_receipt.ts) {
917 if lhs_ts > rhs_ts {
918 return Ordering::Greater;
919 }
920
921 return Ordering::Less;
922 }
923
924 Ordering::Less
925 }
926}