use std::{
collections::HashSet,
ops::{ControlFlow, Deref, DerefMut, Not},
};
use matrix_sdk_base::{
read_receipts::{LatestReadReceipt, ReadReceipts},
serde_helpers::extract_relation,
store::DynStateStore,
};
use matrix_sdk_common::{
deserialized_responses::TimelineEvent, ring_buffer::RingBuffer,
serde_helpers::extract_thread_root,
};
use ruma::{
EventId, OwnedEventId, OwnedUserId, RoomId, UserId,
events::{
AnySyncTimelineEvent, MessageLikeEventType,
receipt::{Receipt, ReceiptEventContent, ReceiptThread, ReceiptType, Receipts},
relation::RelationType,
},
serde::Raw,
};
use tracing::{debug, instrument, trace, warn};
use super::{
super::back_pagination_queue::{
BATCH_SIZE, BackPaginationQueue, BackPaginationRequest, Priority,
},
event_linked_chunk::EventLinkedChunk,
};
use crate::event_cache::caches::pagination::BackPaginationOutcome;
const READ_RECEIPT_MAX_BATCHES: usize = 20;
fn paginate_for_read_receipt(
queue: &BackPaginationQueue,
room_id: &RoomId,
targets: HashSet<OwnedEventId>,
) {
debug!(%room_id, "started backfill request for read receipts");
let request = BackPaginationRequest {
room_id: room_id.to_owned(),
priority: Priority::Normal,
stop: Box::new(stop_on_event_ids(targets)),
batch_size: BATCH_SIZE,
max_batches: Some(READ_RECEIPT_MAX_BATCHES),
};
match queue.enqueue(request) {
Ok(handle) => handle.detach(),
Err(err) => warn!(%room_id, "couldn't enqueue a read-receipt backfill request: {err}"),
}
}
fn stop_on_event_ids(
targets: HashSet<OwnedEventId>,
) -> impl FnMut(&BackPaginationOutcome) -> ControlFlow<()> + Send + 'static {
move |outcome| {
let found = outcome
.events
.iter()
.any(|event| event.event_id().is_some_and(|id| targets.contains(id)));
if found { ControlFlow::Break(()) } else { ControlFlow::Continue(()) }
}
}
trait ReadReceiptsExt {
fn process_event(&mut self, event: &TimelineEvent, user_id: &UserId);
fn reset(&mut self);
fn find_and_process_events<'a>(
&mut self,
receipt_event_id: &EventId,
user_id: &UserId,
events: impl Iterator<Item = &'a TimelineEvent>,
) -> bool;
}
impl ReadReceiptsExt for ReadReceipts {
#[inline(always)]
fn process_event(&mut self, event: &TimelineEvent, user_id: &UserId) {
if marks_as_unread(event.raw(), user_id) {
self.num_unread += 1;
}
let mut has_notify = false;
let mut has_mention = false;
let Some(actions) = event.push_actions() else {
return;
};
for action in actions.iter() {
if !has_notify && action.should_notify() {
self.num_notifications += 1;
has_notify = true;
}
if !has_mention && action.is_highlight() {
self.num_mentions += 1;
has_mention = true;
}
}
}
#[inline(always)]
fn reset(&mut self) {
self.num_unread = 0;
self.num_notifications = 0;
self.num_mentions = 0;
}
#[instrument(skip_all)]
fn find_and_process_events<'a>(
&mut self,
receipt_event_id: &EventId,
user_id: &UserId,
events: impl Iterator<Item = &'a TimelineEvent>,
) -> bool {
let mut counting_receipts = false;
for event in events {
if event.event_id() == Some(receipt_event_id) {
trace!("Found the event the receipt was referring to! Starting to count.");
self.reset();
counting_receipts = true;
continue;
}
if counting_receipts {
self.process_event(event, user_id);
}
}
counting_receipts
}
}
pub trait EventFilter {
fn room_id(&self) -> &RoomId;
fn filter(&self, event: &TimelineEvent) -> bool;
fn receipt_thread_matches(&self, receipt_thread: &ReceiptThread) -> bool;
async fn stored_receipt_event_for_user(
&self,
user_id: &UserId,
receipt_type: ReceiptType,
) -> Option<(OwnedEventId, Receipt)>;
}
pub struct RoomReadReceiptEventFilter<'cache> {
room_id: &'cache RoomId,
with_threading_support: bool,
state_store: &'cache DynStateStore,
}
impl<'cache> RoomReadReceiptEventFilter<'cache> {
pub fn new(
room_event_cache_state: &'cache super::room::RoomEventCacheState,
state_store: &'cache DynStateStore,
) -> Self {
Self {
room_id: &room_event_cache_state.room_id,
with_threading_support: room_event_cache_state.enabled_thread_support,
state_store,
}
}
}
impl<'cache> EventFilter for RoomReadReceiptEventFilter<'cache> {
fn room_id(&self) -> &RoomId {
self.room_id
}
fn filter(&self, event: &TimelineEvent) -> bool {
(self.with_threading_support && extract_thread_root(event.raw()).is_some()).not()
}
fn receipt_thread_matches(&self, receipt_thread: &ReceiptThread) -> bool {
matches!(receipt_thread, ReceiptThread::Unthreaded | ReceiptThread::Main)
}
async fn stored_receipt_event_for_user(
&self,
user_id: &UserId,
receipt_type: ReceiptType,
) -> Option<(OwnedEventId, Receipt)> {
for receipt_thread in [ReceiptThread::Unthreaded, ReceiptThread::Main] {
let receipt_event = self
.state_store
.get_user_room_receipt_event(
self.room_id,
receipt_type.clone(),
&receipt_thread,
user_id,
)
.await
.ok()
.flatten();
if receipt_event.is_some() {
return receipt_event;
}
}
None
}
}
pub struct ThreadReadReceiptEventFilter<'cache> {
room_id: &'cache RoomId,
thread_id: &'cache EventId,
state_store: &'cache DynStateStore,
}
impl<'cache> ThreadReadReceiptEventFilter<'cache> {
pub fn new(
thread_event_cache_state: &'cache super::thread::ThreadEventCacheState,
state_store: &'cache DynStateStore,
) -> Self {
Self {
room_id: &thread_event_cache_state.room_id,
thread_id: &thread_event_cache_state.thread_id,
state_store,
}
}
}
impl<'cache> EventFilter for ThreadReadReceiptEventFilter<'cache> {
fn room_id(&self) -> &RoomId {
self.room_id
}
fn filter(&self, _event: &TimelineEvent) -> bool {
true
}
fn receipt_thread_matches(&self, receipt_thread: &ReceiptThread) -> bool {
matches!(
receipt_thread,
ReceiptThread::Thread(thread_id) if self.thread_id == thread_id
)
}
async fn stored_receipt_event_for_user(
&self,
user_id: &UserId,
receipt_type: ReceiptType,
) -> Option<(OwnedEventId, Receipt)> {
self.state_store
.get_user_room_receipt_event(
self.room_id,
receipt_type,
&ReceiptThread::Thread(self.thread_id.to_owned()),
user_id,
)
.await
.ok()
.flatten()
}
}
const ALL_RECEIPT_TYPES: [ReceiptType; 2] = [ReceiptType::ReadPrivate, ReceiptType::Read];
fn select_best_receipt<T>(
user_id: &UserId,
linked_chunk: &EventLinkedChunk,
event_filter: &T,
pending_receipts: &mut RingBuffer<OwnedEventId>,
new_receipt_event: Option<&ReceiptEventContent>,
latest_active: Option<&EventId>,
) -> Option<OwnedEventId>
where
T: EventFilter,
{
if let Some(receipt_event) = new_receipt_event {
for (event_id, receipts) in &receipt_event.0 {
for ty in ALL_RECEIPT_TYPES {
if let Some(receipts) = receipts.get(&ty)
&& let Some(receipt) = receipts.get(user_id)
&& event_filter.receipt_thread_matches(&receipt.thread)
{
trace!(%event_id, "found new receipt (added to pending)");
pending_receipts.push(event_id.clone());
}
}
}
}
let mut receipt = None;
for (event, event_id) in linked_chunk.revents().filter_map(|(_pos, event)| {
event_filter.filter(event).then_some((event, event.event_id()?))
}) {
if receipt.is_none() {
if latest_active == Some(event_id) {
trace!(active = %event_id, "the latest active receipt is still the most recent; stopping search");
receipt = Some(event_id.to_owned());
}
else if event.sender().as_deref() == Some(user_id) {
trace!(implicit = %event_id, "found an implicit receipt; stopping search");
receipt = Some(event_id.to_owned());
}
}
if receipt.is_some() && pending_receipts.is_empty() {
trace!("exiting loop; found a better receipt, and no more pending receipt to match");
break;
}
pending_receipts.retain(|pending| {
if *pending == event_id {
if receipt.is_none() {
trace!(pending = %event_id, "found a pending receipt; stopping search");
receipt = Some(event_id.to_owned());
} else {
trace!(%event_id, "discarding a pending receipt that wasn't selected");
}
false
} else {
true
}
});
}
receipt
}
async fn try_find_stored_receipts<T>(
user_id: &UserId,
event_filter: &T,
read_receipts: &mut ReadReceipts,
) where
T: EventFilter,
{
for receipt_type in ALL_RECEIPT_TYPES {
if let Some((event_id, _receipt)) =
event_filter.stored_receipt_event_for_user(user_id, receipt_type).await
{
trace!(%event_id, "Found a dormant receipt in the store");
if read_receipts.latest_active.is_none() {
read_receipts.latest_active = Some(LatestReadReceipt { event_id });
} else {
read_receipts.pending.push(event_id);
}
}
}
}
#[instrument(skip_all, fields(room_id = %event_filter.room_id()))]
pub(crate) async fn compute_unread_counts<T>(
user_id: &UserId,
receipt_event: Option<&ReceiptEventContent>,
linked_chunk: &EventLinkedChunk,
event_filter: &T,
read_receipts: &mut ReadReceipts,
back_pagination_queue: Option<&BackPaginationQueue>,
) where
T: EventFilter,
{
debug!(?read_receipts, "Starting");
if read_receipts.latest_active.is_none() {
try_find_stored_receipts(user_id, event_filter, read_receipts).await;
}
let better_receipt = select_best_receipt(
user_id,
linked_chunk,
event_filter,
&mut read_receipts.pending,
receipt_event,
read_receipts.latest_active.as_ref().map(|latest_active| latest_active.event_id.as_ref()),
);
if let Some(event_id) = better_receipt {
trace!(%event_id, "Saving a new active read receipt");
read_receipts.latest_active = Some(LatestReadReceipt { event_id: event_id.clone() });
read_receipts.find_and_process_events(
&event_id,
user_id,
linked_chunk
.events()
.filter_map(|(_pos, event)| event_filter.filter(event).then_some(event)),
);
debug!(?read_receipts, "after finding a better receipt");
return;
}
if let Some(back_pagination_queue) = back_pagination_queue {
let targets: HashSet<OwnedEventId> = read_receipts
.pending
.iter()
.cloned()
.chain(read_receipts.latest_active.as_ref().map(|receipt| receipt.event_id.clone()))
.collect();
paginate_for_read_receipt(back_pagination_queue, event_filter.room_id(), targets);
}
read_receipts.reset();
for event in linked_chunk
.events()
.filter_map(|(_pos, event)| event_filter.filter(event).then_some(event))
{
read_receipts.process_event(event, user_id);
}
debug!(?read_receipts, "no better receipt");
}
fn marks_as_unread(event: &Raw<AnySyncTimelineEvent>, user_id: &UserId) -> bool {
if event.get_field::<OwnedUserId>("sender").ok().flatten().as_deref() == Some(user_id) {
tracing::trace!("not interesting because sent by the current user");
return false;
}
let Some(event_type) = event.get_field::<MessageLikeEventType>("type").ok().flatten() else {
tracing::trace!(
"failed to parse event type for event with id {:?}, skipping it",
event.get_field::<OwnedEventId>("event_id").ok().flatten()
);
return false;
};
match event_type {
MessageLikeEventType::Message
| MessageLikeEventType::PollStart
| MessageLikeEventType::UnstablePollStart
| MessageLikeEventType::PollEnd
| MessageLikeEventType::UnstablePollEnd
| MessageLikeEventType::RoomEncrypted
| MessageLikeEventType::RoomMessage
| MessageLikeEventType::Sticker => {}
_ => {
tracing::trace!("not interesting because not an interesting message-like");
return false;
}
}
if let Some((RelationType::Replacement, _)) = extract_relation(event) {
tracing::trace!("not interesting because edited");
return false;
}
#[derive(serde::Deserialize)]
struct UnsignedContent {
redacted_because: Option<Raw<AnySyncTimelineEvent>>,
}
if let Ok(Some(UnsignedContent { redacted_because: Some(_redaction) })) =
event.get_field::<UnsignedContent>("unsigned")
{
tracing::trace!("not interesting because redacted");
return false;
}
true
}
pub struct MaybeReceiptEventContent(Option<ReceiptEventContent>);
impl MaybeReceiptEventContent {
pub fn none() -> Self {
Self(None)
}
pub fn into_inner(self) -> Option<ReceiptEventContent> {
self.0
}
}
impl Deref for MaybeReceiptEventContent {
type Target = Option<ReceiptEventContent>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl DerefMut for MaybeReceiptEventContent {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
impl FromIterator<(OwnedEventId, Receipts)> for MaybeReceiptEventContent {
fn from_iter<T>(iterator: T) -> Self
where
T: IntoIterator<Item = (OwnedEventId, Receipts)>,
{
let mut iterator = iterator.into_iter().peekable();
Self(if iterator.peek().is_some() { Some(iterator.collect()) } else { None })
}
}
#[cfg(test)]
mod tests {
use std::{num::NonZeroUsize, ops::Not as _};
use matrix_sdk_base::{read_receipts::ReadReceipts, store::MemoryStore};
use matrix_sdk_common::{deserialized_responses::TimelineEvent, ring_buffer::RingBuffer};
use matrix_sdk_test::{ALICE, event_factory::EventFactory};
use ruma::{
EventId, MilliSecondsSinceUnixEpoch, RoomId, UserId, event_id,
events::{
receipt::{Receipt, ReceiptThread, ReceiptType, UserReceipts},
room::{member::MembershipState, message::MessageType},
},
owned_event_id,
push::{Action, HighlightTweakValue, Tweak},
room_id, user_id,
};
use super::{
EventFilter, MaybeReceiptEventContent, ReadReceiptsExt as _, Receipts,
RoomReadReceiptEventFilter, marks_as_unread, select_best_receipt, stop_on_event_ids,
};
use crate::event_cache::caches::{
event_linked_chunk::EventLinkedChunk, pagination::BackPaginationOutcome,
};
#[test]
fn test_stop_on_event_ids() {
use std::collections::HashSet;
use matrix_sdk_test::BOB;
let room = room_id!("!omelette:fromage.fr");
let f = EventFactory::new().room(room).sender(*BOB);
let outcome = BackPaginationOutcome {
reached_start: false,
events: vec![
f.text_msg("a").event_id(event_id!("$1")).into_event(),
f.text_msg("b").event_id(event_id!("$2")).into_event(),
],
};
assert!(stop_on_event_ids(HashSet::from([owned_event_id!("$2")]))(&outcome).is_break());
assert!(stop_on_event_ids(HashSet::from([owned_event_id!("$3")]))(&outcome).is_continue());
assert!(stop_on_event_ids(HashSet::new())(&outcome).is_continue());
}
#[test]
fn test_room_message_marks_as_unread() {
let user_id = user_id!("@alice:example.org");
let other_user_id = user_id!("@bob:example.org");
let f = EventFactory::new();
let ev = f.text_msg("A").event_id(event_id!("$ida")).sender(other_user_id).into_raw_sync();
assert!(marks_as_unread(&ev, user_id));
let ev = f.text_msg("A").event_id(event_id!("$ida")).sender(user_id).into_raw_sync();
assert!(marks_as_unread(&ev, user_id).not());
}
#[test]
fn test_room_edit_does_not_mark_as_unread() {
let user_id = user_id!("@alice:example.org");
let other_user_id = user_id!("@bob:example.org");
let ev = EventFactory::new()
.text_msg("* edited message")
.edit(
event_id!("$someeventid:localhost"),
MessageType::text_plain("edited message").into(),
)
.event_id(event_id!("$ida"))
.sender(other_user_id)
.into_raw_sync();
assert!(marks_as_unread(&ev, user_id).not());
}
#[test]
fn test_redaction_does_not_mark_room_as_unread() {
let user_id = user_id!("@alice:example.org");
let other_user_id = user_id!("@bob:example.org");
let ev = EventFactory::new()
.redaction(event_id!("$151957878228ssqrj:localhost"))
.sender(other_user_id)
.event_id(event_id!("$151957878228ssqrJ:localhost"))
.into_raw_sync();
assert!(marks_as_unread(&ev, user_id).not());
}
#[test]
fn test_reaction_does_not_mark_room_as_unread() {
let user_id = user_id!("@alice:example.org");
let other_user_id = user_id!("@bob:example.org");
let ev = EventFactory::new()
.reaction(event_id!("$15275047031IXQRj:localhost"), "👍")
.sender(other_user_id)
.event_id(event_id!("$15275047031IXQRi:localhost"))
.into_raw_sync();
assert!(marks_as_unread(&ev, user_id).not());
}
#[test]
fn test_state_event_does_not_mark_as_unread() {
let user_id = user_id!("@alice:example.org");
let event_id = event_id!("$1");
let ev = EventFactory::new()
.member(user_id)
.membership(MembershipState::Join)
.display_name("Alice")
.event_id(event_id)
.into_raw_sync();
assert!(marks_as_unread(&ev, user_id).not());
let other_user_id = user_id!("@bob:example.org");
assert!(marks_as_unread(&ev, other_user_id).not());
}
#[test]
fn test_count_unread_and_mentions() {
fn make_event(user_id: &UserId, push_actions: Vec<Action>) -> TimelineEvent {
let mut ev = EventFactory::new()
.text_msg("A")
.sender(user_id)
.event_id(event_id!("$ida"))
.into_event();
ev.set_push_actions(push_actions);
ev
}
let user_id = user_id!("@alice:example.org");
let event = make_event(user_id, Vec::new());
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 0);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 0);
let event = make_event(user_id!("@bob:example.org"), Vec::new());
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 0);
let event = make_event(user_id!("@bob:example.org"), vec![Action::Notify]);
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 1);
let event = make_event(
user_id!("@bob:example.org"),
vec![Action::SetTweak(Tweak::Highlight(HighlightTweakValue::Yes))],
);
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 1);
assert_eq!(receipts.num_notifications, 0);
let event = make_event(
user_id!("@bob:example.org"),
vec![Action::SetTweak(Tweak::Highlight(HighlightTweakValue::Yes)), Action::Notify],
);
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 1);
assert_eq!(receipts.num_notifications, 1);
let event = make_event(user_id!("@bob:example.org"), vec![Action::Notify, Action::Notify]);
let mut receipts = ReadReceipts::default();
receipts.process_event(&event, user_id);
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 1);
}
#[test]
fn test_find_and_process_events() {
let ev0 = event_id!("$0");
let user_id = user_id!("@alice:example.org");
let mut receipts = ReadReceipts::default();
assert!(receipts.find_and_process_events(ev0, user_id, [].iter()).not());
assert_eq!(receipts.num_unread, 0);
assert_eq!(receipts.num_notifications, 0);
assert_eq!(receipts.num_mentions, 0);
fn make_event(event_id: &EventId) -> TimelineEvent {
EventFactory::new()
.text_msg("A")
.sender(user_id!("@bob:example.org"))
.event_id(event_id)
.into()
}
let mut receipts = ReadReceipts {
num_unread: 42,
num_notifications: 13,
num_mentions: 37,
..Default::default()
};
assert!(
receipts
.find_and_process_events(ev0, user_id, [make_event(event_id!("$1"))].iter())
.not()
);
assert_eq!(receipts.num_unread, 42);
assert_eq!(receipts.num_notifications, 13);
assert_eq!(receipts.num_mentions, 37);
let mut receipts = ReadReceipts {
num_unread: 42,
num_notifications: 13,
num_mentions: 37,
..Default::default()
};
assert!(receipts.find_and_process_events(ev0, user_id, [make_event(ev0)].iter()));
assert_eq!(receipts.num_unread, 0);
assert_eq!(receipts.num_notifications, 0);
assert_eq!(receipts.num_mentions, 0);
let mut receipts = ReadReceipts {
num_unread: 42,
num_notifications: 13,
num_mentions: 37,
..Default::default()
};
assert!(
receipts
.find_and_process_events(
ev0,
user_id,
[
make_event(event_id!("$1")),
make_event(event_id!("$2")),
make_event(event_id!("$3"))
]
.iter(),
)
.not()
);
assert_eq!(receipts.num_unread, 42);
assert_eq!(receipts.num_notifications, 13);
assert_eq!(receipts.num_mentions, 37);
let mut receipts = ReadReceipts {
num_unread: 42,
num_notifications: 13,
num_mentions: 37,
..Default::default()
};
assert!(
receipts.find_and_process_events(
ev0,
user_id,
[
make_event(event_id!("$1")),
make_event(ev0),
make_event(event_id!("$2")),
make_event(event_id!("$3"))
]
.iter(),
)
);
assert_eq!(receipts.num_unread, 2);
assert_eq!(receipts.num_notifications, 0);
assert_eq!(receipts.num_mentions, 0);
let mut receipts = ReadReceipts {
num_unread: 42,
num_notifications: 13,
num_mentions: 37,
..Default::default()
};
assert!(
receipts.find_and_process_events(
ev0,
user_id,
[
make_event(ev0),
make_event(event_id!("$1")),
make_event(ev0),
make_event(event_id!("$2")),
make_event(event_id!("$3"))
]
.iter(),
)
);
assert_eq!(receipts.num_unread, 2);
assert_eq!(receipts.num_notifications, 0);
assert_eq!(receipts.num_mentions, 0);
}
#[test]
fn test_compute_unread_counts_with_threading_enabled() {
fn make_in_thread_event(
user_id: &UserId,
room_id: &RoomId,
thread_root: &EventId,
) -> TimelineEvent {
EventFactory::new()
.room(room_id)
.text_msg("A")
.sender(user_id)
.event_id(event_id!("$ida"))
.in_thread(thread_root, event_id!("$latest_event"))
.into_event()
}
let mut receipts = ReadReceipts::default();
let state_store = MemoryStore::new();
let room_id = room_id!("!r");
let own_alice = user_id!("@alice:example.org");
let bob = user_id!("@bob:example.org");
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: true,
state_store: &state_store,
};
for event in [
make_in_thread_event(own_alice, room_id, event_id!("$some_thread_root")),
make_in_thread_event(own_alice, room_id, event_id!("$some_other_thread_root")),
make_in_thread_event(bob, room_id, event_id!("$some_thread_root")),
make_in_thread_event(bob, room_id, event_id!("$some_other_thread_root")),
]
.into_iter()
.filter(|event| event_filter.filter(event))
{
receipts.process_event(&event, own_alice);
}
assert_eq!(receipts.num_unread, 0);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 0);
for event in [EventFactory::new()
.room(room_id)
.text_msg("A")
.sender(bob)
.event_id(event_id!("$ida"))
.into_event()]
.into_iter()
.filter(|event| event_filter.filter(event))
{
receipts.process_event(&event, own_alice);
}
assert_eq!(receipts.num_unread, 1);
assert_eq!(receipts.num_mentions, 0);
assert_eq!(receipts.num_notifications, 0);
}
#[test]
fn test_select_best_receipt_noop() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let state_store = MemoryStore::new();
let own_user_id = user_id!("@not_alice:example.org");
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
let new_receipt_event = None;
let active_receipt = None;
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event,
active_receipt,
);
assert!(receipt.is_none());
assert!(pending_receipts.is_empty());
}
#[test]
fn test_select_best_receipt_implicit() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let own_user_id = user_id!("@not_alice:example.org");
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).sender(own_user_id).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
let new_receipt_event = None;
let active_receipt = None;
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event,
active_receipt,
);
assert_eq!(receipt.unwrap(), "$2");
assert!(pending_receipts.is_empty());
}
#[test]
fn test_select_best_receipt_active_receipt() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let own_user_id = user_id!("@not_alice:example.org");
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
let new_receipt_event = None;
let active_receipt = Some(event_id!("$2"));
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event,
active_receipt,
);
assert_eq!(receipt.unwrap(), "$2");
assert!(pending_receipts.is_empty());
}
#[test]
fn test_select_best_receipt_new_receipt_event() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let own_user_id = user_id!("@not_alice:example.org");
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
let new_receipt_event = Some(
f.read_receipts()
.add(event_id!("$2"), own_user_id, ReceiptType::Read, ReceiptThread::Unthreaded)
.into_content(),
);
let active_receipt = None;
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event.as_ref(),
active_receipt,
);
assert_eq!(receipt.unwrap(), "$2");
assert!(pending_receipts.is_empty());
}
#[test]
fn test_select_best_receipt_stashes_pending_receipts() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let own_user_id = user_id!("@not_alice:example.org");
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
let new_receipt_event = Some(
f.read_receipts()
.add(event_id!("$4"), own_user_id, ReceiptType::Read, ReceiptThread::Unthreaded)
.into_content(),
);
let active_receipt = None;
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event.as_ref(),
active_receipt,
);
assert!(receipt.is_none());
assert_eq!(pending_receipts.len(), 1);
assert_eq!(pending_receipts.get(0).unwrap(), "$4");
}
#[test]
fn test_select_best_receipt_matched_pending_receipt() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let own_user_id = user_id!("@not_alice:example.org");
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).into_event(),
]);
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
pending_receipts.push(owned_event_id!("$2"));
let new_receipt_event = None;
let active_receipt = None;
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event.as_ref(),
active_receipt,
);
assert_eq!(receipt.unwrap(), "$2");
assert!(pending_receipts.is_empty());
}
#[test]
fn test_select_best_receipt_mixed() {
let room_id = room_id!("!roomid:example.org");
let f = EventFactory::new().room(room_id).sender(*ALICE);
let own_user_id = user_id!("@not_alice:example.org");
let mut linked_chunk = EventLinkedChunk::new();
linked_chunk.push_events(vec![
f.text_msg("Event 1").event_id(event_id!("$1")).into_event(),
f.text_msg("Event 2").event_id(event_id!("$2")).into_event(),
f.text_msg("Event 3").event_id(event_id!("$3")).sender(own_user_id).into_event(),
f.text_msg("Event 4").event_id(event_id!("$4")).into_event(),
f.text_msg("Event 5").event_id(event_id!("$5")).into_event(),
]);
let mut pending_receipts = RingBuffer::new(NonZeroUsize::new(16).unwrap());
pending_receipts.push(owned_event_id!("$2"));
pending_receipts.push(owned_event_id!("$6"));
let new_receipt_event = Some(
f.read_receipts()
.add(event_id!("$4"), own_user_id, ReceiptType::Read, ReceiptThread::Unthreaded)
.add(event_id!("$7"), own_user_id, ReceiptType::ReadPrivate, ReceiptThread::Main)
.into_content(),
);
let active_receipt = Some(event_id!("$1"));
let state_store = MemoryStore::new();
let event_filter = RoomReadReceiptEventFilter {
room_id,
with_threading_support: false,
state_store: &state_store,
};
let receipt = select_best_receipt(
own_user_id,
&linked_chunk,
&event_filter,
&mut pending_receipts,
new_receipt_event.as_ref(),
active_receipt,
);
assert_eq!(receipt.unwrap(), "$4");
assert_eq!(pending_receipts.len(), 2);
assert!(pending_receipts.iter().any(|ev| ev == event_id!("$6")));
assert!(pending_receipts.iter().any(|ev| ev == event_id!("$7")));
}
#[test]
fn test_maybe_receipt_event_content_from_empty_iterator() {
let maybe: MaybeReceiptEventContent = std::iter::empty().collect();
assert!(maybe.is_none());
}
#[test]
fn test_maybe_receipt_event_content_from_iterator() {
let maybe: MaybeReceiptEventContent = vec![(
event_id!("$ev").to_owned(),
Receipts::from([(
ReceiptType::Read,
UserReceipts::from([(
user_id!("@ali:ce").to_owned(),
Receipt::new(MilliSecondsSinceUnixEpoch::now()),
)]),
)]),
)]
.into_iter()
.collect();
assert!(maybe.is_some());
let receipt_event_content = maybe.into_inner().unwrap();
assert_eq!(receipt_event_content.len(), 1);
assert!(receipt_event_content.contains_key(event_id!("$ev")));
}
}