use std::collections::{BTreeMap, HashMap};
use matrix_sdk_base::{
serde_helpers::{extract_redaction_target, extract_relation, extract_thread_root},
sync::Timeline,
};
use ruma::{
OwnedEventId,
events::{
AnySyncEphemeralRoomEvent,
receipt::{ReceiptEventContent, ReceiptThread, Receipts},
relation::RelationType,
},
room_version_rules::RedactionRules,
};
use super::{
super::{Result, states::StateLockReadGuard},
read_receipts::MaybeReceiptEventContent,
room::RoomEventCacheState,
thread::ThreadEventCacheState,
};
pub fn aggregate_timeline_and_read_receipts_for_room(
timeline: &Timeline,
ephemeral: &[AnySyncEphemeralRoomEvent],
) -> (Timeline, MaybeReceiptEventContent) {
(
timeline.clone(),
filter_read_receipts_and_group_by(ephemeral, |receipt_thread| match receipt_thread {
ReceiptThread::Main | ReceiptThread::Unthreaded => Some(()),
_ => None,
})
.map(|(_receipt_thread, (event_id, event_receipts))| {
(event_id.clone(), event_receipts.clone())
})
.collect(),
)
}
pub async fn aggregate_timeline_and_read_receipts_for_threads<'sync, 'state>(
timeline: &'sync Timeline,
ephemeral: &'sync [AnySyncEphemeralRoomEvent],
existing_threads: StateLockReadGuard<'state, HashMap<OwnedEventId, ThreadEventCacheState>>,
maybe_room: Option<StateLockReadGuard<'state, RoomEventCacheState>>,
redaction_rules: &'sync RedactionRules,
) -> Result<HashMap<OwnedEventId, (Timeline, MaybeReceiptEventContent)>> {
let mut new_events_by_thread = HashMap::new();
let default_entry = || {
(
Timeline {
limited: timeline.limited,
prev_batch: timeline.prev_batch.clone(),
events: Vec::new(),
},
MaybeReceiptEventContent::none(),
)
};
for (nth, event) in timeline.events.iter().enumerate() {
match extract_relation(event.raw()) {
Some((relation_type, related_event_id)) => match relation_type {
RelationType::Thread => {
new_events_by_thread
.entry(related_event_id)
.or_insert_with(default_entry)
.0
.events
.push(event.clone());
}
RelationType::Annotation
| RelationType::Replacement
| RelationType::Reference
| _ => {
if let Some(thread_root) = match timeline.events[..nth]
.iter()
.rev()
.find(|event| event.event_id() == Some(&related_event_id))
{
Some(related_event) => extract_thread_root(related_event.raw()),
None => match &maybe_room {
Some(room) => room.find_event(&related_event_id).await?.and_then(
|(_location, related_event)| {
extract_thread_root(related_event.raw())
},
),
None => None,
},
} {
new_events_by_thread
.entry(thread_root)
.or_insert_with(default_entry)
.0
.events
.push(event.clone());
}
}
},
None => {
if let Some(event_id) = event.event_id()
&& existing_threads.contains_key(event_id)
{
new_events_by_thread
.entry(event_id.to_owned())
.or_insert_with(default_entry)
.0
.events
.push(event.clone());
}
else if let Some(redaction_target) =
extract_redaction_target(event.raw(), redaction_rules)
&& match &maybe_room {
Some(room) => room.find_event(&redaction_target).await?.is_some(),
None => false,
}
{
let mut associated_thread_root = None;
for thread in existing_threads.values() {
if thread.find_event(&redaction_target).await?.is_some() {
associated_thread_root = Some(thread.thread_id.clone());
break;
}
}
if let Some(thread_root) = associated_thread_root {
new_events_by_thread
.entry(thread_root)
.or_insert_with(default_entry)
.0
.events
.push(event.clone());
}
}
}
}
}
for (thread_root, (read_receipt_event_id, read_receipt_event)) in
filter_read_receipts_and_group_by(ephemeral, |receipt_thread| match receipt_thread {
ReceiptThread::Thread(thread_id) => Some(thread_id),
_ => None,
})
{
new_events_by_thread
.entry(thread_root.to_owned())
.or_insert_with(default_entry)
.1
.get_or_insert_with(|| ReceiptEventContent(BTreeMap::new()))
.insert(read_receipt_event_id.clone(), read_receipt_event.clone());
}
Ok(new_events_by_thread)
}
pub fn aggregate_timeline_for_pinned_events(
timeline: &Timeline,
pinned_event_ids: &[OwnedEventId],
redaction_rules: &RedactionRules,
) -> Timeline {
let mut new_timeline = Timeline {
limited: timeline.limited,
prev_batch: timeline.prev_batch.clone(),
events: Vec::new(),
};
if pinned_event_ids.is_empty() {
return new_timeline;
}
for event in &timeline.events {
match extract_relation(event.raw()) {
Some((relation_type, related_event_id)) => match relation_type {
RelationType::Thread => {}
RelationType::Annotation
| RelationType::Replacement
| RelationType::Reference
| _ => {
if pinned_event_ids.contains(&related_event_id) {
new_timeline.events.push(event.clone());
}
}
},
None => {
if let Some(redaction_target) =
extract_redaction_target(event.raw(), redaction_rules)
&& pinned_event_ids.contains(&redaction_target)
{
new_timeline.events.push(event.clone());
}
}
}
}
new_timeline
}
fn filter_read_receipts_and_group_by<'e, F, G>(
events: &'e [AnySyncEphemeralRoomEvent],
predicate: F,
) -> impl Iterator<Item = (G, (&'e OwnedEventId, &'e Receipts))>
where
F: Fn(&'e ReceiptThread) -> Option<G>,
{
events
.iter()
.filter_map(|ephemeral| match ephemeral {
AnySyncEphemeralRoomEvent::Receipt(receipt_event) => Some(receipt_event),
_ => None,
})
.flat_map(|receipt_event| receipt_event.content.iter())
.filter_map(move |(event_id, event_receipts)| {
Some((
predicate(&event_receipts.first_key_value()?.1.first_key_value()?.1.thread)?,
(event_id, event_receipts),
))
})
}