use crate::ids::{EventId, RoomId, TxnId, UserId};
use crate::room::relations::{aggregate_relation, recompute_bundle, RelatingEvent, RelationsBundle};
use crate::wire::events::{
Membership, RawEvent, RedactionContent, RelatesTo, RoomEncryptedContent, RoomMessageContent, TextLikeMessageContent,
};
use std::collections::{BTreeMap, VecDeque};
const PENDING_CAPACITY: usize = 5_000;
const EVENT_ROOM_MESSAGE: &str = "m.room.message";
const EVENT_ROOM_ENCRYPTED: &str = "m.room.encrypted";
const EVENT_ROOM_REDACTION: &str = "m.room.redaction";
const EVENT_ROOM_MEMBER: &str = "m.room.member";
const EVENT_ROOM_NAME: &str = "m.room.name";
const EVENT_ROOM_TOPIC: &str = "m.room.topic";
const EVENT_ROOM_POWER_LEVELS: &str = "m.room.power_levels";
const EVENT_ROOM_HISTORY_VISIBILITY: &str = "m.room.history_visibility";
const EVENT_ROOM_CREATE: &str = "m.room.create";
pub(crate) const KEY_FORWARDED: &str = "forwarded";
pub(crate) const KEY_FORWARDED_FROM: &str = "forwarded_from";
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SendState {
LocalEcho,
Sending,
Sent,
Failed {
reason: String,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum StateChangeSummary {
Member {
user_id: String,
membership: Membership,
displayname: Option<String>,
},
Name(String),
Topic(String),
}
#[derive(Clone, Debug, PartialEq)]
pub enum ItemContent {
Text(TextLikeMessageContent),
Notice(TextLikeMessageContent),
Emote(TextLikeMessageContent),
Encrypted {
session_id: String,
ciphertext_b64: String,
},
Undecryptable {
reason: String,
},
Redacted,
StateChange(StateChangeSummary),
Unknown,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Forwarded {
Hidden,
Channel {
room_id: RoomId,
room_name: String,
},
}
fn parse_forwarded(event_type: &str, content: &serde_json::Value) -> Option<Forwarded> {
if event_type != EVENT_ROOM_MESSAGE {
return None;
}
let channel = content.get(KEY_FORWARDED_FROM).and_then(|from| {
let room_id = RoomId::parse(from.get("room_id")?.as_str()?).ok()?;
let room_name = from.get("room_name")?.as_str()?.to_string();
Some(Forwarded::Channel { room_id, room_name })
});
channel.or_else(|| {
(content.get(KEY_FORWARDED).and_then(serde_json::Value::as_bool) == Some(true)).then_some(Forwarded::Hidden)
})
}
#[derive(Clone, Debug, PartialEq)]
struct DecryptedEvent {
event_type: String,
content: serde_json::Value,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum ForwardRefusal {
Missing,
Redacted,
Sealed,
NotForwardable {
event_type: String,
},
}
impl std::fmt::Display for ForwardRefusal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ForwardRefusal::Missing => f.write_str("no such event in that room"),
ForwardRefusal::Redacted => f.write_str("the event was redacted"),
ForwardRefusal::Sealed => f.write_str("the event is still sealed (undecryptable or not decrypted yet)"),
ForwardRefusal::NotForwardable { event_type } => write!(f, "{event_type} events cannot be forwarded"),
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct ForwardSource {
pub(crate) event_type: String,
pub(crate) content: serde_json::Value,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GapMarker {
pub prev_batch: Option<String>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct TimelineItem {
pub event_id: Option<EventId>,
pub txn_id: Option<TxnId>,
pub sender: UserId,
pub origin_server_ts: i64,
pub event_type: String,
pub state_key: Option<String>,
pub content: ItemContent,
pub relations: RelationsBundle,
pub send_state: SendState,
pub redacted: bool,
raw_content: serde_json::Value,
pub forwarded: Option<Forwarded>,
decrypted: Option<DecryptedEvent>,
}
pub(crate) fn interpret_content(event_type: &str, state_key: Option<&str>, content: &serde_json::Value) -> ItemContent {
if let Some(state_key) = state_key {
return interpret_state_change(event_type, state_key, content);
}
match event_type {
EVENT_ROOM_MESSAGE => match serde_json::from_value::<RoomMessageContent>(content.clone()) {
Ok(RoomMessageContent::Text(inner)) => ItemContent::Text(inner),
Ok(RoomMessageContent::Notice(inner)) => ItemContent::Notice(inner),
Ok(RoomMessageContent::Emote(inner)) => ItemContent::Emote(inner),
_ => ItemContent::Unknown,
},
EVENT_ROOM_ENCRYPTED => match serde_json::from_value::<RoomEncryptedContent>(content.clone()) {
Ok(RoomEncryptedContent::Megolm(inner)) => {
ItemContent::Encrypted { session_id: inner.session_id, ciphertext_b64: inner.ciphertext }
}
_ => ItemContent::Undecryptable {
reason: "m.room.encrypted event in a room timeline was not the megolm algorithm".to_string(),
},
},
_ => ItemContent::Unknown,
}
}
fn interpret_state_change(event_type: &str, state_key: &str, content: &serde_json::Value) -> ItemContent {
match event_type {
EVENT_ROOM_MEMBER => {
let membership = content
.get("membership")
.and_then(|value| serde_json::from_value::<Membership>(value.clone()).ok())
.unwrap_or(Membership::Unknown);
let displayname = content.get("displayname").and_then(serde_json::Value::as_str).map(str::to_string);
ItemContent::StateChange(StateChangeSummary::Member {
user_id: state_key.to_string(),
membership,
displayname,
})
}
EVENT_ROOM_NAME => {
let name = content.get("name").and_then(serde_json::Value::as_str).unwrap_or_default().to_string();
ItemContent::StateChange(StateChangeSummary::Name(name))
}
EVENT_ROOM_TOPIC => {
let topic = content.get("topic").and_then(serde_json::Value::as_str).unwrap_or_default().to_string();
ItemContent::StateChange(StateChangeSummary::Topic(topic))
}
_ => ItemContent::Unknown,
}
}
fn content_after_redaction(event_type: &str, state_key: Option<&str>, content: &serde_json::Value) -> ItemContent {
if event_type == EVENT_ROOM_MEMBER {
let membership = content
.get("membership")
.and_then(|value| serde_json::from_value::<Membership>(value.clone()).ok())
.unwrap_or(Membership::Unknown);
return ItemContent::StateChange(StateChangeSummary::Member {
user_id: state_key.unwrap_or_default().to_string(),
membership,
displayname: None,
});
}
match event_type {
EVENT_ROOM_POWER_LEVELS | EVENT_ROOM_HISTORY_VISIBILITY | EVENT_ROOM_CREATE => ItemContent::Unknown,
_ => ItemContent::Redacted,
}
}
fn build_own_relation_pointers(event: &RawEvent) -> RelationsBundle {
let mut bundle = RelationsBundle::default();
match event.relates_to() {
Some(RelatesTo::InReplyTo(reply)) => bundle.reply_to = Some(reply.event_id),
Some(RelatesTo::Thread { event_id, in_reply_to, .. }) => {
bundle.thread_root = Some(event_id);
if let Some(reply) = in_reply_to {
bundle.reply_to = Some(reply.event_id);
}
}
_ => {}
}
bundle
}
fn build_item(event: &RawEvent) -> TimelineItem {
let already_redacted = event.unsigned.redacted_because.is_some();
let content = if already_redacted {
content_after_redaction(&event.event_type, event.state_key.as_deref(), &event.content)
} else {
interpret_content(&event.event_type, event.state_key.as_deref(), &event.content)
};
let forwarded = if already_redacted || event.state_key.is_some() {
None
} else {
parse_forwarded(&event.event_type, &event.content)
};
TimelineItem {
event_id: Some(event.event_id.clone()),
txn_id: event.unsigned.transaction_id.clone().map(TxnId::from),
sender: event.sender.clone(),
origin_server_ts: event.origin_server_ts,
event_type: event.event_type.clone(),
state_key: event.state_key.clone(),
content,
relations: build_own_relation_pointers(event),
send_state: SendState::Sent,
redacted: already_redacted,
raw_content: event.content.clone(),
forwarded,
decrypted: None,
}
}
fn redact_item(item: &mut TimelineItem) {
item.content = content_after_redaction(&item.event_type, item.state_key.as_deref(), &item.raw_content);
item.redacted = true;
item.forwarded = None;
item.decrypted = None;
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Timeline {
items: Vec<TimelineItem>,
by_event_id: BTreeMap<EventId, usize>,
by_txn_id: BTreeMap<TxnId, usize>,
relation_contributions: BTreeMap<EventId, EventId>,
relating_events_by_target: BTreeMap<EventId, Vec<RelatingEvent>>,
pending_relations: BTreeMap<EventId, Vec<RelatingEvent>>,
pending_relation_order: VecDeque<EventId>,
pending_redactions: BTreeMap<EventId, EventId>,
pending_redaction_order: VecDeque<EventId>,
gap: Option<GapMarker>,
older_token: Option<String>,
}
impl Timeline {
pub fn new() -> Self {
Self::default()
}
pub fn items(&self) -> &[TimelineItem] {
&self.items
}
pub fn item_by_event_id(&self, event_id: &EventId) -> Option<&TimelineItem> {
self.by_event_id.get(event_id).map(|&idx| &self.items[idx])
}
pub fn relating_events_for(&self, target: &EventId) -> &[RelatingEvent] {
self.relating_events_by_target.get(target).map(Vec::as_slice).unwrap_or(&[])
}
pub fn gap(&self) -> Option<&GapMarker> {
self.gap.as_ref()
}
pub fn older_token(&self) -> Option<&str> {
self.older_token.as_deref()
}
pub fn apply_timeline_batch(&mut self, events: &[RawEvent], limited: bool, prev_batch: Option<String>) {
if limited {
self.gap = Some(GapMarker { prev_batch });
}
for event in events {
self.apply_one_event(event);
}
}
fn apply_one_event(&mut self, event: &RawEvent) {
if event.event_type == EVENT_ROOM_REDACTION {
if let Ok(redaction) = serde_json::from_value::<RedactionContent>(event.content.clone()) {
self.apply_or_queue_redaction(&event.event_id, &redaction.redacts);
}
return;
}
if event.state_key.is_none() && event.event_type != EVENT_ROOM_ENCRYPTED {
if let Some(relation) = event.relates_to() {
if matches!(relation, RelatesTo::Annotation { .. } | RelatesTo::Replace { .. }) {
self.apply_relation_contribution(event, relation);
return;
}
}
}
let item = build_item(event);
if let Some(txn_id) = item.txn_id.clone() {
if let Some(&idx) = self.by_txn_id.get(&txn_id) {
self.reconcile_local_echo(idx, item);
return;
}
}
if let Some(event_id) = item.event_id.clone() {
if let Some(&idx) = self.by_event_id.get(&event_id) {
self.items[idx] = item;
return;
}
}
self.insert_new_item(item);
}
fn apply_relation_contribution(&mut self, event: &RawEvent, relation: RelatesTo) {
let new_content =
if matches!(relation, RelatesTo::Replace { .. }) { event.content.get("m.new_content").cloned() } else { None };
let relating = RelatingEvent {
event_id: event.event_id.clone(),
sender: event.sender.clone(),
origin_server_ts: event.origin_server_ts,
relation,
new_content,
};
self.fold_or_queue_relation(relating);
}
fn fold_or_queue_relation(&mut self, relating: RelatingEvent) {
let target = match &relating.relation {
RelatesTo::Annotation { event_id, .. } | RelatesTo::Replace { event_id } => event_id.clone(),
_ => return,
};
let Some(&idx) = self.by_event_id.get(&target) else {
self.queue_pending_relation(target, relating);
return;
};
let original_sender = self.items[idx].sender.clone();
if aggregate_relation(&mut self.items[idx].relations, &original_sender, &relating) {
self.relation_contributions.insert(relating.event_id.clone(), target.clone());
self.relating_events_by_target.entry(target).or_default().push(relating);
}
}
fn queue_pending_relation(&mut self, target: EventId, relating: RelatingEvent) {
self.pending_relations.entry(target.clone()).or_default().push(relating);
self.pending_relation_order.push_back(target);
while self.pending_relation_order.len() > PENDING_CAPACITY {
let Some(oldest_target) = self.pending_relation_order.pop_front() else { break };
let mut now_empty = false;
if let Some(pending) = self.pending_relations.get_mut(&oldest_target) {
if !pending.is_empty() {
pending.remove(0);
}
now_empty = pending.is_empty();
}
if now_empty {
self.pending_relations.remove(&oldest_target);
}
}
}
fn queue_pending_redaction(&mut self, target: EventId, redaction_event_id: EventId) {
if self.pending_redactions.contains_key(&target) {
return;
}
self.pending_redactions.insert(target.clone(), redaction_event_id);
self.pending_redaction_order.push_back(target);
while self.pending_redaction_order.len() > PENDING_CAPACITY {
let Some(oldest) = self.pending_redaction_order.pop_front() else { break };
self.pending_redactions.remove(&oldest);
}
}
fn resolve_pending_for(&mut self, event_id: &EventId, idx: usize) {
if self.pending_redactions.remove(event_id).is_some() {
redact_item(&mut self.items[idx]);
}
if let Some(pending) = self.pending_relations.remove(event_id) {
let original_sender = self.items[idx].sender.clone();
for relating in pending {
if aggregate_relation(&mut self.items[idx].relations, &original_sender, &relating) {
self.relation_contributions.insert(relating.event_id.clone(), event_id.clone());
self.relating_events_by_target.entry(event_id.clone()).or_default().push(relating);
}
}
}
}
pub fn apply_redaction(&mut self, target: &EventId) {
self.apply_or_queue_redaction(target, target);
}
fn apply_or_queue_redaction(&mut self, redaction_event_id: &EventId, target: &EventId) {
if let Some(&idx) = self.by_event_id.get(target) {
redact_item(&mut self.items[idx]);
return;
}
if let Some(bundle_target) = self.relation_contributions.remove(target) {
if let Some(contributions) = self.relating_events_by_target.get_mut(&bundle_target) {
contributions.retain(|relating| &relating.event_id != target);
}
if let Some(&target_idx) = self.by_event_id.get(&bundle_target) {
let original_sender = self.items[target_idx].sender.clone();
let remaining = self.relating_events_by_target.get(&bundle_target).cloned().unwrap_or_default();
self.items[target_idx].relations = recompute_bundle(&original_sender, &remaining);
}
return;
}
self.queue_pending_redaction(target.clone(), redaction_event_id.clone());
}
fn insert_new_item(&mut self, item: TimelineItem) {
let idx = self.items.len();
let event_id = item.event_id.clone();
if let Some(event_id) = &event_id {
self.by_event_id.insert(event_id.clone(), idx);
}
if let Some(txn_id) = &item.txn_id {
self.by_txn_id.insert(txn_id.clone(), idx);
}
self.items.push(item);
if let Some(event_id) = event_id {
self.resolve_pending_for(&event_id, idx);
}
}
fn reconcile_local_echo(&mut self, idx: usize, confirmed: TimelineItem) {
let event_id = confirmed.event_id.clone();
if let Some(event_id) = &event_id {
self.by_event_id.insert(event_id.clone(), idx);
}
self.items[idx] = confirmed;
if let Some(event_id) = event_id {
self.resolve_pending_for(&event_id, idx);
}
}
pub fn push_local_echo(
&mut self,
txn_id: TxnId,
sender: UserId,
origin_server_ts: i64,
event_type: &str,
content: ItemContent,
wire_content: serde_json::Value,
) {
let item = TimelineItem {
event_id: None,
txn_id: Some(txn_id),
sender,
origin_server_ts,
event_type: event_type.to_string(),
state_key: None,
content,
relations: RelationsBundle::default(),
send_state: SendState::LocalEcho,
redacted: false,
forwarded: parse_forwarded(event_type, &wire_content),
raw_content: wire_content,
decrypted: None,
};
self.insert_new_item(item);
}
pub fn mark_send_state(&mut self, txn_id: &TxnId, send_state: SendState) {
if let Some(&idx) = self.by_txn_id.get(txn_id) {
self.items[idx].send_state = send_state;
}
}
pub fn prepend_back_page(&mut self, events: &[RawEvent], end_token: Option<String>) {
let mut new_items = Vec::with_capacity(events.len());
for event in events {
if self.by_event_id.contains_key(&event.event_id) {
continue;
}
if event.event_type == EVENT_ROOM_REDACTION {
if let Ok(redaction) = serde_json::from_value::<RedactionContent>(event.content.clone()) {
self.apply_or_queue_redaction(&event.event_id, &redaction.redacts);
}
continue;
}
if event.state_key.is_none() {
if let Some(relation) = event.relates_to() {
if matches!(relation, RelatesTo::Annotation { .. } | RelatesTo::Replace { .. }) {
self.apply_relation_contribution(event, relation);
continue;
}
}
}
new_items.push(build_item(event));
}
let inserted_count = new_items.len();
self.items.splice(0..0, new_items);
self.older_token = end_token;
self.reindex();
for idx in 0..inserted_count {
if let Some(event_id) = self.items[idx].event_id.clone() {
self.resolve_pending_for(&event_id, idx);
}
}
}
fn reindex(&mut self) {
self.by_event_id.clear();
self.by_txn_id.clear();
for (idx, item) in self.items.iter().enumerate() {
if let Some(event_id) = &item.event_id {
self.by_event_id.insert(event_id.clone(), idx);
}
if let Some(txn_id) = &item.txn_id {
self.by_txn_id.insert(txn_id.clone(), idx);
}
}
}
pub fn set_decrypted(&mut self, event_id: &EventId, content: ItemContent) {
if let Some(&idx) = self.by_event_id.get(event_id) {
self.items[idx].content = content;
}
}
pub(crate) fn set_decrypted_event(&mut self, event_id: &EventId, event_type: &str, content: &serde_json::Value) {
let Some(&idx) = self.by_event_id.get(event_id) else { return };
let item = &mut self.items[idx];
item.content = interpret_content(event_type, None, content);
item.forwarded = parse_forwarded(event_type, content);
item.decrypted = Some(DecryptedEvent { event_type: event_type.to_string(), content: content.clone() });
}
pub(crate) fn forward_source(&self, event_id: &EventId) -> Result<ForwardSource, ForwardRefusal> {
let item = self.item_by_event_id(event_id).ok_or(ForwardRefusal::Missing)?;
if item.redacted || item.content == ItemContent::Redacted {
return Err(ForwardRefusal::Redacted);
}
match &item.content {
ItemContent::Encrypted { .. } | ItemContent::Undecryptable { .. } => return Err(ForwardRefusal::Sealed),
_ => {}
}
let (event_type, original) = match &item.decrypted {
Some(decrypted) => (decrypted.event_type.as_str(), &decrypted.content),
None => (item.event_type.as_str(), &item.raw_content),
};
let not_forwardable = || ForwardRefusal::NotForwardable { event_type: event_type.to_string() };
if item.state_key.is_some() || event_type != EVENT_ROOM_MESSAGE {
return Err(not_forwardable());
}
let mut content = match &item.relations.latest_edit {
Some(edit) => serde_json::to_value(&edit.content).map_err(|_| not_forwardable())?,
None => original.clone(),
};
if let Some(replacement) = content.get("m.new_content").filter(|value| value.is_object()).cloned() {
content = replacement;
}
if !content.is_object() {
return Err(not_forwardable());
}
Ok(ForwardSource { event_type: event_type.to_string(), content })
}
pub(crate) fn apply_decrypted_relation(
&mut self,
event_id: &EventId,
sender: UserId,
origin_server_ts: i64,
relation: RelatesTo,
new_content: Option<serde_json::Value>,
) -> bool {
if !matches!(relation, RelatesTo::Annotation { .. } | RelatesTo::Replace { .. }) {
return false;
}
let Some(&idx) = self.by_event_id.get(event_id) else { return false };
self.items.remove(idx);
self.reindex();
let relating = RelatingEvent { event_id: event_id.clone(), sender, origin_server_ts, relation, new_content };
self.fold_or_queue_relation(relating);
true
}
pub fn set_echo_event_id(&mut self, txn_id: &TxnId, event_id: EventId) {
let Some(&idx) = self.by_txn_id.get(txn_id) else { return };
self.items[idx].event_id = Some(event_id.clone());
self.items[idx].send_state = SendState::Sent;
self.by_event_id.insert(event_id, idx);
}
}
pub fn highlight(user_id: &UserId, displayname: Option<&str>, item: &TimelineItem) -> bool {
let body = match &item.content {
ItemContent::Text(inner) | ItemContent::Notice(inner) | ItemContent::Emote(inner) => inner.body.as_str(),
_ => return false,
};
mentions_word(body, user_id.as_str()) || displayname.is_some_and(|name| !name.trim().is_empty() && mentions_word(body, name))
}
fn mentions_word(haystack: &str, needle: &str) -> bool {
if needle.is_empty() {
return false;
}
let haystack_lower = haystack.to_lowercase();
let needle_lower = needle.to_lowercase();
let mut search_from = 0;
while let Some(found_at) = haystack_lower[search_from..].find(&needle_lower) {
let start = search_from + found_at;
let end = start + needle_lower.len();
let before_is_boundary = haystack_lower[..start].chars().next_back().is_none_or(|c| !c.is_alphanumeric());
let after_is_boundary = haystack_lower[end..].chars().next().is_none_or(|c| !c.is_alphanumeric());
if before_is_boundary && after_is_boundary {
return true;
}
search_from = start + needle_lower.len().max(1);
}
false
}
pub fn last_preview(timeline: &Timeline) -> Option<&TimelineItem> {
timeline.items.iter().rev().find(|item| {
!item.redacted
&& !matches!(
item.content,
ItemContent::Encrypted { .. } | ItemContent::Undecryptable { .. } | ItemContent::StateChange(_)
)
})
}
#[cfg(test)]
mod tests {
use super::*;
fn raw_event(event_id: &str, event_type: &str, sender: &str, ts: i64, content: serde_json::Value) -> RawEvent {
serde_json::from_value(serde_json::json!({
"event_id": event_id,
"type": event_type,
"sender": sender,
"origin_server_ts": ts,
"content": content,
}))
.expect("valid raw event")
}
fn raw_event_with_txn(
event_id: &str,
event_type: &str,
sender: &str,
ts: i64,
content: serde_json::Value,
txn_id: &str,
) -> RawEvent {
serde_json::from_value(serde_json::json!({
"event_id": event_id,
"type": event_type,
"sender": sender,
"origin_server_ts": ts,
"content": content,
"unsigned": { "transaction_id": txn_id },
}))
.expect("valid raw event")
}
fn text_content(body: &str) -> serde_json::Value {
serde_json::json!({ "msgtype": "m.text", "body": body })
}
fn text_item(body: &str) -> TimelineItem {
TimelineItem {
event_id: Some(EventId::parse("$x:example.org").expect("valid event id")),
txn_id: None,
sender: UserId::parse("@someone:example.org").expect("valid user id"),
origin_server_ts: 1,
event_type: EVENT_ROOM_MESSAGE.to_string(),
state_key: None,
content: ItemContent::Text(TextLikeMessageContent { body: body.to_string(), format: None, formatted_body: None }),
relations: RelationsBundle::default(),
send_state: SendState::Sent,
redacted: false,
raw_content: serde_json::Value::Null,
forwarded: None,
decrypted: None,
}
}
#[test]
fn apply_timeline_batch_reconciles_local_echo_by_txn_id() {
let mut timeline = Timeline::new();
let alice = UserId::parse("@alice:example.org").expect("valid user id");
let txn_id = TxnId::new(0);
timeline.push_local_echo(
txn_id.clone(),
alice,
100,
EVENT_ROOM_MESSAGE,
ItemContent::Text(TextLikeMessageContent { body: "hi".to_string(), format: None, formatted_body: None }),
text_content("hi"),
);
assert_eq!(timeline.items().len(), 1);
assert_eq!(timeline.items()[0].send_state, SendState::LocalEcho);
assert!(timeline.items()[0].event_id.is_none());
let confirmed =
raw_event_with_txn("$confirmed:example.org", "m.room.message", "@alice:example.org", 105, text_content("hi"), txn_id.as_str());
timeline.apply_timeline_batch(&[confirmed], false, None);
assert_eq!(timeline.items().len(), 1, "the echo is replaced in place, not duplicated");
let item = &timeline.items()[0];
assert_eq!(item.event_id, Some(EventId::parse("$confirmed:example.org").expect("valid event id")));
assert_eq!(item.send_state, SendState::Sent);
assert_eq!(item.txn_id, Some(txn_id));
}
#[test]
fn apply_redaction_keeps_v11_allow_listed_fields() {
let mut timeline = Timeline::new();
let member_event: RawEvent = serde_json::from_value(serde_json::json!({
"event_id": "$member:example.org",
"type": "m.room.member",
"sender": "@alice:example.org",
"origin_server_ts": 1,
"state_key": "@bob:example.org",
"content": { "membership": "join", "displayname": "Bob" }
}))
.expect("valid member event");
let message_event = raw_event("$msg:example.org", "m.room.message", "@bob:example.org", 2, text_content("hello"));
timeline.apply_timeline_batch(&[member_event, message_event], false, None);
let member_id = EventId::parse("$member:example.org").expect("valid event id");
timeline.apply_redaction(&member_id);
let member_item = timeline.item_by_event_id(&member_id).expect("present");
assert!(member_item.redacted);
match &member_item.content {
ItemContent::StateChange(StateChangeSummary::Member { membership, displayname, .. }) => {
assert_eq!(*membership, Membership::Join, "membership survives the v11 allow-list");
assert_eq!(*displayname, None, "displayname is not on the v11 allow-list");
}
other => panic!("expected a Member state change, got {other:?}"),
}
let msg_id = EventId::parse("$msg:example.org").expect("valid event id");
timeline.apply_redaction(&msg_id);
let msg_item = timeline.item_by_event_id(&msg_id).expect("present");
assert!(msg_item.redacted);
assert_eq!(msg_item.content, ItemContent::Redacted, "m.room.message has nothing on the v11 allow-list");
}
#[test]
fn redacting_a_reaction_removes_it_from_the_bundle() {
let mut timeline = Timeline::new();
let target = raw_event("$target:example.org", "m.room.message", "@alice:example.org", 1, text_content("hi"));
let reaction = raw_event(
"$reaction:example.org",
"m.reaction",
"@bob:example.org",
2,
serde_json::json!({
"m.relates_to": { "rel_type": "m.annotation", "event_id": "$target:example.org", "key": "\u{1F44D}" }
}),
);
timeline.apply_timeline_batch(&[target, reaction], false, None);
let target_id = EventId::parse("$target:example.org").expect("valid event id");
let bundle_before = timeline.item_by_event_id(&target_id).expect("present").relations.clone();
assert_eq!(bundle_before.reactions.get("\u{1F44D}").expect("reaction present").len(), 1);
assert_eq!(timeline.items().len(), 1, "the reaction event never gets its own row");
timeline.apply_redaction(&EventId::parse("$reaction:example.org").expect("valid event id"));
let bundle_after = &timeline.item_by_event_id(&target_id).expect("present").relations;
assert!(
!bundle_after.reactions.contains_key("\u{1F44D}"),
"the retracted reaction is gone entirely, not just emptied"
);
}
#[test]
fn limited_batch_records_a_gap() {
let mut timeline = Timeline::new();
assert!(timeline.gap().is_none());
let event = raw_event("$1:example.org", "m.room.message", "@alice:example.org", 1, text_content("hi"));
timeline.apply_timeline_batch(&[event], true, Some("t123".to_string()));
let gap = timeline.gap().expect("limited batch records a gap");
assert_eq!(gap.prev_batch.as_deref(), Some("t123"));
assert_eq!(timeline.items().len(), 1, "the window's own events are still applied");
}
#[test]
fn last_preview_skips_redacted_and_state() {
let mut timeline = Timeline::new();
let text = raw_event("$text:example.org", "m.room.message", "@alice:example.org", 1, text_content("first"));
let name_change: RawEvent = serde_json::from_value(serde_json::json!({
"event_id": "$name:example.org", "type": "m.room.name", "sender": "@alice:example.org",
"origin_server_ts": 2, "state_key": "", "content": { "name": "New name" }
}))
.expect("valid name event");
let redacted: RawEvent = serde_json::from_value(serde_json::json!({
"event_id": "$redacted:example.org", "type": "m.room.message", "sender": "@alice:example.org",
"origin_server_ts": 3, "content": {},
"unsigned": { "redacted_because": {
"event_id": "$r:example.org", "type": "m.room.redaction", "sender": "@mod:example.org",
"origin_server_ts": 4, "content": { "redacts": "$redacted:example.org" }
}}
}))
.expect("valid redacted event");
timeline.apply_timeline_batch(&[text, name_change, redacted], false, None);
let preview = last_preview(&timeline).expect("an eligible item exists");
assert_eq!(preview.event_id, Some(EventId::parse("$text:example.org").expect("valid event id")));
}
#[test]
fn highlight_matches_mention_not_substring() {
let alice = UserId::parse("@alice:example.org").expect("valid user id");
let mentioned = text_item("hey @alice:example.org, are you around?");
assert!(highlight(&alice, Some("Alice"), &mentioned));
let by_displayname = text_item("Alice, can you review this?");
assert!(highlight(&alice, Some("Alice"), &by_displayname));
let substring_only = text_item("notalice123 says hi to nobody");
assert!(!highlight(&alice, Some("Alice"), &substring_only));
}
#[test]
fn reaction_seen_before_its_target_applies_when_target_back_paginates() {
let mut timeline = Timeline::new();
let reaction = raw_event(
"$reaction:example.org",
"m.reaction",
"@bob:example.org",
10,
serde_json::json!({
"m.relates_to": { "rel_type": "m.annotation", "event_id": "$target:example.org", "key": "\u{1F44D}" }
}),
);
timeline.apply_timeline_batch(&[reaction], false, None);
assert!(timeline.items().is_empty(), "the reaction never gets its own row, pending or not");
let target = raw_event("$target:example.org", "m.room.message", "@alice:example.org", 1, text_content("hi"));
timeline.prepend_back_page(&[target], Some("t1".to_string()));
let target_id = EventId::parse("$target:example.org").expect("valid event id");
let item = timeline.item_by_event_id(&target_id).expect("target now loaded");
assert_eq!(
item.relations.reactions.get("\u{1F44D}").map(std::collections::BTreeSet::len),
Some(1),
"the pending reaction applied once its target loaded"
);
}
#[test]
fn edit_seen_before_its_target_applies_later() {
let mut timeline = Timeline::new();
let edit = raw_event(
"$edit:example.org",
"m.room.message",
"@alice:example.org",
10,
serde_json::json!({
"msgtype": "m.text",
"body": "* edited",
"m.new_content": { "msgtype": "m.text", "body": "edited" },
"m.relates_to": { "rel_type": "m.replace", "event_id": "$target:example.org" }
}),
);
timeline.apply_timeline_batch(&[edit], false, None);
assert!(timeline.items().is_empty(), "an accepted edit never gets its own row, pending or not");
let target = raw_event("$target:example.org", "m.room.message", "@alice:example.org", 1, text_content("original"));
timeline.prepend_back_page(&[target], None);
let target_id = EventId::parse("$target:example.org").expect("valid event id");
let item = timeline.item_by_event_id(&target_id).expect("target now loaded");
let latest = item.relations.latest_edit.as_ref().expect("the pending edit applied once its target loaded");
assert_eq!(
latest.content,
RoomMessageContent::Text(TextLikeMessageContent {
body: "edited".to_string(),
format: None,
formatted_body: None
})
);
}
#[test]
fn redaction_seen_before_its_target_applies_later() {
let mut timeline = Timeline::new();
let redaction = raw_event(
"$redaction:example.org",
"m.room.redaction",
"@mod:example.org",
10,
serde_json::json!({ "redacts": "$target:example.org" }),
);
timeline.apply_timeline_batch(&[redaction], false, None);
assert!(timeline.items().is_empty(), "a redaction never gets its own row, pending or not");
let target = raw_event("$target:example.org", "m.room.message", "@alice:example.org", 1, text_content("hi"));
timeline.prepend_back_page(&[target], None);
let target_id = EventId::parse("$target:example.org").expect("valid event id");
let item = timeline.item_by_event_id(&target_id).expect("target now loaded");
assert!(item.redacted, "the pending redaction applied once its target loaded");
assert_eq!(item.content, ItemContent::Redacted);
}
fn encrypted_placeholder(event_id: &str, sender: &str, extra: serde_json::Value) -> RawEvent {
let mut content = serde_json::json!({
"algorithm": "m.megolm.v1.aes-sha2",
"ciphertext": "AAAA",
"session_id": "session-1",
});
if let (Some(target), Some(extra)) = (content.as_object_mut(), extra.as_object()) {
target.extend(extra.clone());
}
raw_event(event_id, "m.room.encrypted", sender, 1, content)
}
fn event_id(value: &str) -> EventId {
EventId::parse(value).expect("valid event id")
}
fn forwarded_of(timeline: &Timeline, id: &str) -> Option<Forwarded> {
timeline.item_by_event_id(&event_id(id)).expect("present").forwarded.clone()
}
#[test]
fn unrecognized_event_and_msgtype_stay_unknown() {
let mut timeline = Timeline::new();
let sticker = raw_event(
"$sticker:example.org",
"m.sticker",
"@alice:example.org",
1,
serde_json::json!({ "body": "Bull", "url": "mxc://example.org/sticker" }),
);
timeline.apply_timeline_batch(&[sticker], false, None);
let item = timeline.item_by_event_id(&event_id("$sticker:example.org")).expect("present");
assert_eq!(item.content, ItemContent::Unknown);
assert_eq!(item.raw_content.get("url").and_then(serde_json::Value::as_str), Some("mxc://example.org/sticker"));
let image = serde_json::json!({ "msgtype": "m.image", "body": "x.png", "url": "mxc://example.org/a" });
assert_eq!(interpret_content(EVENT_ROOM_MESSAGE, None, &image), ItemContent::Unknown);
let other = serde_json::json!({ "msgtype": "com.example.card", "body": "card", "symbol": "BTCUSDT" });
assert_eq!(interpret_content(EVENT_ROOM_MESSAGE, None, &other), ItemContent::Unknown);
}
#[test]
fn forwarded_markers_parse_to_hidden_and_channel() {
let mut timeline = Timeline::new();
let hidden = raw_event(
"$hidden:example.org",
"m.room.message",
"@alice:example.org",
1,
serde_json::json!({ "msgtype": "m.text", "body": "fwd", "forwarded": true }),
);
let channel = raw_event(
"$channel:example.org",
"m.room.message",
"@alice:example.org",
2,
serde_json::json!({
"msgtype": "m.text",
"body": "fwd",
"forwarded_from": { "room_id": "!chan:example.org", "room_name": "Announcements" },
}),
);
let sticker = raw_event(
"$sticker:example.org",
"m.sticker",
"@alice:example.org",
3,
serde_json::json!({
"body": "Bull",
"url": "mxc://example.org/bull",
"info": { "pack": "trader", "id": "bull" },
"forwarded": true,
}),
);
let plain = raw_event("$plain:example.org", "m.room.message", "@alice:example.org", 4, text_content("plain"));
let denied = raw_event(
"$denied:example.org",
"m.room.message",
"@alice:example.org",
5,
serde_json::json!({ "msgtype": "m.text", "body": "x", "forwarded": false }),
);
timeline.apply_timeline_batch(&[hidden, channel, sticker, plain, denied], false, None);
assert_eq!(forwarded_of(&timeline, "$hidden:example.org"), Some(Forwarded::Hidden));
assert_eq!(
forwarded_of(&timeline, "$channel:example.org"),
Some(Forwarded::Channel {
room_id: RoomId::parse("!chan:example.org").expect("valid room id"),
room_name: "Announcements".to_string(),
})
);
assert_eq!(forwarded_of(&timeline, "$sticker:example.org"), None, "an unrecognized event type is not a forward");
assert_eq!(forwarded_of(&timeline, "$plain:example.org"), None);
assert_eq!(forwarded_of(&timeline, "$denied:example.org"), None, "only `true` marks a forward");
let sealed = encrypted_placeholder(
"$sealed:example.org",
"@bob:example.org",
serde_json::json!({ "forwarded": true }),
);
timeline.apply_timeline_batch(&[sealed], false, None);
assert_eq!(forwarded_of(&timeline, "$sealed:example.org"), None);
timeline.set_decrypted_event(
&event_id("$sealed:example.org"),
"m.room.message",
&serde_json::json!({ "msgtype": "m.text", "body": "fwd", "forwarded": true }),
);
assert_eq!(forwarded_of(&timeline, "$sealed:example.org"), Some(Forwarded::Hidden));
timeline.push_local_echo(
TxnId::new(7),
UserId::parse("@alice:example.org").expect("valid user id"),
10,
EVENT_ROOM_MESSAGE,
ItemContent::Unknown,
serde_json::json!({ "msgtype": "m.text", "body": "fwd", "forwarded": true }),
);
let echo = timeline.items().last().expect("echo pushed");
assert_eq!(echo.forwarded, Some(Forwarded::Hidden));
timeline.apply_redaction(&event_id("$hidden:example.org"));
assert_eq!(forwarded_of(&timeline, "$hidden:example.org"), None);
timeline.apply_redaction(&event_id("$sealed:example.org"));
assert_eq!(forwarded_of(&timeline, "$sealed:example.org"), None);
}
#[test]
fn malformed_forwarded_from_is_ignored() {
let bad_markers = [
serde_json::json!("a string"),
serde_json::json!(42),
serde_json::json!({}),
serde_json::json!({ "room_id": "!chan:example.org" }),
serde_json::json!({ "room_name": "Announcements" }),
serde_json::json!({ "room_id": 5, "room_name": "Announcements" }),
serde_json::json!({ "room_id": "not a room id", "room_name": "Announcements" }),
serde_json::json!({ "room_id": "!chan:example.org", "room_name": null }),
];
let mut timeline = Timeline::new();
for (n, marker) in bad_markers.iter().enumerate() {
let id = format!("$bad{n}:example.org");
let event = raw_event(
&id,
"m.room.message",
"@alice:example.org",
n as i64,
serde_json::json!({ "msgtype": "m.text", "body": "still readable", "forwarded_from": marker }),
);
timeline.apply_timeline_batch(&[event], false, None);
let item = timeline.item_by_event_id(&event_id(&id)).expect("the event still decodes");
assert_eq!(item.forwarded, None, "malformed marker {marker} is ignored");
assert!(matches!(item.content, ItemContent::Text(_)), "the message itself is untouched by a bad marker");
}
let both = raw_event(
"$both:example.org",
"m.room.message",
"@alice:example.org",
99,
serde_json::json!({
"msgtype": "m.text",
"body": "x",
"forwarded": true,
"forwarded_from": "garbage",
}),
);
timeline.apply_timeline_batch(&[both], false, None);
assert_eq!(timeline.item_by_event_id(&event_id("$both:example.org")).expect("present").forwarded, Some(Forwarded::Hidden));
}
#[test]
fn pending_relations_are_bounded() {
let mut timeline = Timeline::new();
for n in 0..=PENDING_CAPACITY {
let target = format!("$target{n}:example.org");
let reaction = raw_event(
&format!("$reaction{n}:example.org"),
"m.reaction",
"@alice:example.org",
n as i64,
serde_json::json!({
"m.relates_to": { "rel_type": "m.annotation", "event_id": target, "key": "\u{1F44D}" }
}),
);
timeline.apply_timeline_batch(std::slice::from_ref(&reaction), false, None);
}
assert_eq!(timeline.pending_relation_order.len(), PENDING_CAPACITY, "bounded at capacity");
let oldest_target =
raw_event("$target0:example.org", "m.room.message", "@alice:example.org", 0, text_content("oldest"));
timeline.prepend_back_page(&[oldest_target], None);
let oldest_id = EventId::parse("$target0:example.org").expect("valid event id");
assert!(
timeline.item_by_event_id(&oldest_id).expect("present").relations.reactions.is_empty(),
"the oldest pending contribution was evicted, never applied"
);
let newest_target_id_str = format!("$target{PENDING_CAPACITY}:example.org");
let newest_target = raw_event(&newest_target_id_str, "m.room.message", "@alice:example.org", 0, text_content("newest"));
timeline.prepend_back_page(&[newest_target], None);
let newest_id = EventId::parse(&newest_target_id_str).expect("valid event id");
assert_eq!(
timeline
.item_by_event_id(&newest_id)
.expect("present")
.relations
.reactions
.get("\u{1F44D}")
.map(std::collections::BTreeSet::len),
Some(1),
"the most recently queued pending contribution survived the bound"
);
}
}