feat(sdk): Annotations and redactions events are stored in their associated threads.

This patch stores all events with a relation to an in-thread event,
plus the `m.room.redaction` events targeting in-thread events, in their
respective thread cache.

This patch also introduces a new `extract_redaction_target` helper, that
is used in the `maybe_apply_new_redaction` methods of the room and the
thread cache.

The aggregator for the threads is updated to aggregate the
`m.room.redaction` event, and annotation relation type. A refactoring
is done to use a `match` over `RelationType`, instead of using multiple
`extract_*` helper, which clarifies the code I believe. A limitation has
been found and a `TODO` has been added to keep the commits “small”.

crates/matrix-sdk-common/src/serde_helpers.rs

@@ -19,8 +19,10 @@ use ruma::{
     MilliSecondsSinceUnixEpoch, OwnedEventId,
     events::{
         AnyMessageLikeEventContent, AnySyncMessageLikeEvent, AnySyncTimelineEvent,
+        MessageLikeEventType,
         relation::{BundledThread, RelationType},
     },
+    room_version_rules::RedactionRules,
     serde::Raw,
 };
 use serde::Deserialize;
@@ -70,23 +72,6 @@ pub fn extract_thread_root(event: &Raw<AnySyncTimelineEvent>) -> Option<OwnedEve
     extract_thread_root_from_content(event.get_field("content").ok().flatten()?)
 }
 
-/// Try to extract the target of an edit event, from a raw timeline event, if
-/// provided.
-///
-/// The target event is the field located at
-/// `content`.`m.relates_to`.`event_id`, if the field at
-/// `content`.`m.relates_to`.`rel_type` is `m.replace`.
-///
-/// Returns `None` if we couldn't find it, or if there was an issue
-/// during deserialization.
-pub fn extract_edit_target(event: &Raw<AnySyncTimelineEvent>) -> Option<OwnedEventId> {
-    let relates_to = event.get_field::<SimplifiedContent>("content").ok().flatten()?.relates_to?;
-    match relates_to.rel_type {
-        RelationType::Replacement => relates_to.event_id,
-        _ => None,
-    }
-}
-
 /// Try to extract the type and target of a relation, from a raw timeline event,
 /// if provided.
 pub fn extract_relation(event: &Raw<AnySyncTimelineEvent>) -> Option<(RelationType, OwnedEventId)> {
@@ -101,6 +86,32 @@ struct Relations {
     thread: Option<Box<BundledThread>>,
 }
 
+/// Try to extract the event ID of the event targeted by `event` if it is of
+/// type `m.room.redaction`.
+pub fn extract_redaction_target(
+    event: &Raw<AnySyncTimelineEvent>,
+    redaction_rules: &RedactionRules,
+) -> Option<OwnedEventId> {
+    // Check if it's a `m.room.redaction`.
+    let Ok(Some(MessageLikeEventType::RoomRedaction)) =
+        event.get_field::<MessageLikeEventType>("type")
+    else {
+        // Not the expected event. Early return.
+        return None;
+    };
+
+    // It is a `m.room.redaction`! We can deserialize it entirely.
+
+    let Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomRedaction(redaction))) =
+        event.deserialize()
+    else {
+        // Failed to deserialized. Early return.
+        return None;
+    };
+
+    redaction.redacts(redaction_rules).map(ToOwned::to_owned)
+}
+
 #[allow(missing_debug_implementations)]
 #[derive(Deserialize)]
 struct Unsigned {

crates/matrix-sdk/src/event_cache/caches/aggregator.rs

@@ -15,10 +15,10 @@
 use std::collections::HashMap;
 
 use matrix_sdk_base::{
-    serde_helpers::{extract_edit_target, extract_thread_root},
+    serde_helpers::{extract_redaction_target, extract_relation, extract_thread_root},
     sync::Timeline,
 };
-use ruma::OwnedEventId;
+use ruma::{OwnedEventId, events::relation::RelationType, room_version_rules::RedactionRules};
 
 use super::{super::Result, room::RoomEventCacheStateLockReadGuard, thread::ThreadEventCache};
 
@@ -30,6 +30,7 @@ pub async fn aggregate_timeline_for_threads(
     timeline: &Timeline,
     existing_threads: &HashMap<OwnedEventId, ThreadEventCache>,
     room_event_cache: RoomEventCacheStateLockReadGuard<'_>,
+    redaction_rules: &RedactionRules,
 ) -> Result<HashMap<OwnedEventId, Timeline>> {
     let mut new_events_by_thread = HashMap::new();
 
@@ -39,38 +40,89 @@ pub async fn aggregate_timeline_for_threads(
         events: Vec::new(),
     };
 
+    // Look for in-thread events, i.e. events that are part of threads.
     for event in &timeline.events {
-        // This event is part of a thread.
-        if let Some(thread_root) = extract_thread_root(event.raw()) {
-            new_events_by_thread
-                .entry(thread_root)
-                .or_insert_with(default_timeline)
-                .events
-                .push(event.clone());
-        }
-        // This event is the root of a thread.
-        else if let Some(event_id) = event.event_id()
-            && existing_threads.contains_key(&event_id)
-        {
-            new_events_by_thread
-                .entry(event_id)
-                .or_insert_with(default_timeline)
-                .events
-                .push(event.clone());
-        }
+        match extract_relation(event.raw()) {
+            // Ohh, this event relates to another event!
+            Some((relation_type, related_event_id)) => match relation_type {
+                // `related_event` represents a thread root.
+                RelationType::Thread => {
+                    new_events_by_thread
+                        .entry(related_event_id)
+                        .or_insert_with(default_timeline)
+                        .events
+                        .push(event.clone());
+                }
 
-        // This event is an edit that may apply to a thread.
-        if let Some(edit_target) = extract_edit_target(event.raw()) {
-            // This event is known and part of a thread.
-            if let Some((_location, edited_event)) =
-                room_event_cache.find_event(&edit_target).await?
-                && let Some(thread_root) = extract_thread_root(edited_event.raw())
-            {
-                new_events_by_thread
-                    .entry(thread_root)
-                    .or_insert_with(default_timeline)
-                    .events
-                    .push(event.clone());
+                // `event` represents an annotation (e.g. reactions), a replacement (an edit), a
+                // reference or something custom. Let's see if the `related_event_id` is an
+                // in-thread event.
+                RelationType::Annotation
+                | RelationType::Replacement
+                | RelationType::Reference
+                | _ => {
+                    // TODO(@hywan): It's good to look for the related event in the memory and
+                    // store, but we MUST also look for it in `timeline`!
+                    if let Some((_location, related_event)) =
+                        room_event_cache.find_event(&related_event_id).await?
+                        && let Some(thread_root) = extract_thread_root(related_event.raw())
+                    {
+                        new_events_by_thread
+                            .entry(thread_root)
+                            .or_insert_with(default_timeline)
+                            .events
+                            .push(event.clone());
+                    }
+                }
+            },
+
+            // No explicit relation, okay, but it can still be related to a thread!
+            None => {
+                // We previously found events that are part of a thread, but we didn't see the
+                // thread root yet. And guess what? This might be this event!
+                if let Some(event_id) = event.event_id()
+                    && existing_threads.contains_key(&event_id)
+                {
+                    new_events_by_thread
+                        .entry(event_id)
+                        .or_insert_with(default_timeline)
+                        .events
+                        .push(event.clone());
+                }
+                // Otherwise, this event might be a redaction that applies to a thread.
+                else if let Some(redaction_target) =
+                    extract_redaction_target(event.raw(), redaction_rules)
+                    // TODO(@hywan): It's good to look for the redacted event in the memory and
+                    // store, but we MUST also look for it in `timeline`!
+                    && room_event_cache.find_event(&redaction_target).await?.is_some()
+                {
+                    // The redacted event exists (in the room, because it contains _all_ the
+                    // events) **but** the event has been redacted (in
+                    // the room). It's no more possible to extract its
+                    // thread root (because this information has been removed).
+                    //
+                    // But we need to know if the event is part of a thread to apply the
+                    // redaction in the thread too. No other choice than
+                    // doing a full search…
+
+                    let mut associated_thread_root = None;
+
+                    for (thread_root, thread) in existing_threads {
+                        if thread.find_event(&redaction_target).await?.is_some() {
+                            associated_thread_root = Some(thread_root.clone());
+                            break;
+                        }
+                    }
+
+                    // We've found the thread owning the event being redacted!
+                    if let Some(thread_root) = associated_thread_root {
+                        new_events_by_thread
+                            .entry(thread_root)
+                            .or_insert_with(default_timeline)
+                            .events
+                            .push(event.clone());
+                    }
+                }
             }
         }
     }

crates/matrix-sdk/src/event_cache/caches/mod.rs

@@ -185,7 +185,7 @@ impl Caches {
 
     /// Update all the event caches with a [`JoinedRoomUpdate`].
     pub(super) async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> {
-        let Self { room, threads, internals: _ } = &self;
+        let Self { room, threads, internals } = &self;
 
         // Room.
         {
@@ -205,6 +205,7 @@ impl Caches {
                 &updates.timeline,
                 threads.read().await.deref(),
                 room.state().read().await?,
+                &internals.room_version_rules.redaction,
             )
             .await?;
 
@@ -228,7 +229,7 @@ impl Caches {
 
     /// Update all the event caches with a [`LeftRoomUpdate`].
     pub(super) async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> {
-        let Self { room, threads, internals: _ } = &self;
+        let Self { room, threads, internals } = &self;
 
         // Room.
         {
@@ -248,6 +249,7 @@ impl Caches {
                 &updates.timeline,
                 threads.read().await.deref(),
                 room.state().read().await?,
+                &internals.room_version_rules.redaction,
             )
             .await?;
 

crates/matrix-sdk/src/event_cache/caches/room/state.rs

@@ -32,14 +32,14 @@ use matrix_sdk_base::{
     linked_chunk::{
         ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader,
     },
+    serde_helpers::extract_redaction_target,
     sync::Timeline,
 };
 use matrix_sdk_common::executor::spawn;
 use ruma::{
     EventId, OwnedEventId, OwnedRoomId, OwnedUserId,
     events::{
-        AnySyncEphemeralRoomEvent, AnySyncMessageLikeEvent, AnySyncTimelineEvent,
-        MessageLikeEventType,
+        AnySyncEphemeralRoomEvent,
         receipt::{ReceiptEventContent, SyncReceiptEvent},
         relation::RelationType,
         room::redaction::SyncRoomRedactionEvent,
@@ -857,32 +857,15 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
     /// redacted form.
     #[instrument(skip_all)]
     async fn maybe_apply_new_redaction(&mut self, event: &Event) -> Result<(), EventCacheError> {
-        let raw_event = event.raw();
-
-        // Do not deserialise the entire event if we aren't certain it's a
-        // `m.room.redaction`. It saves a non-negligible amount of computations.
-        let Ok(Some(MessageLikeEventType::RoomRedaction)) =
-            raw_event.get_field::<MessageLikeEventType>("type")
+        let Some(target_event_id) =
+            extract_redaction_target(event.raw(), &self.room_version_rules.redaction)
         else {
-            return Ok(());
-        };
-
-        // It is a `m.room.redaction`! We can deserialize it entirely.
-
-        let Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomRedaction(
-            redaction,
-        ))) = raw_event.deserialize()
-        else {
-            return Ok(());
-        };
-
-        let Some(target_event_id) = redaction.redacts(&self.room_version_rules.redaction) else {
             warn!("missing target event id from the redaction event");
             return Ok(());
         };
 
         // Replace the redacted event by a redacted form, if we knew about it.
-        let Some((location, mut target_event)) = self.find_event(target_event_id).await? else {
+        let Some((location, mut target_event)) = self.find_event(&target_event_id).await? else {
             trace!("redacted event is missing from the linked chunk");
             return Ok(());
         };

crates/matrix-sdk/src/event_cache/caches/thread/mod.rs

@@ -218,8 +218,7 @@ impl ThreadEventCache {
     ///
     /// It starts by looking into loaded events in `EventLinkedChunk` before
     /// looking inside the storage.
-    #[cfg(test)]
-    async fn find_event(
+    pub(super) async fn find_event(
         &self,
         event_id: &EventId,
     ) -> Result<Option<(super::EventLocation, Event)>> {

crates/matrix-sdk/src/event_cache/caches/thread/state.rs

@@ -23,20 +23,17 @@ use matrix_sdk_base::{
     linked_chunk::{
         ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader,
     },
-    serde_helpers::extract_edit_target,
+    serde_helpers::extract_redaction_target,
     sync::Timeline,
 };
 use matrix_sdk_common::executor::spawn;
 use ruma::{
     EventId, OwnedEventId, OwnedRoomId, OwnedUserId,
-    events::{
-        AnySyncMessageLikeEvent, AnySyncTimelineEvent, MessageLikeEventType,
-        relation::RelationType, room::redaction::SyncRoomRedactionEvent,
-    },
+    events::{relation::RelationType, room::redaction::SyncRoomRedactionEvent},
     room_version_rules::RoomVersionRules,
 };
 use tokio::sync::broadcast::Sender;
-use tracing::{debug, error, instrument, trace, warn};
+use tracing::{debug, error, instrument, trace};
 
 use super::{
     super::{
@@ -523,32 +520,14 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> {
     /// redacted form.
     #[instrument(skip_all)]
     async fn maybe_apply_new_redaction(&mut self, event: &Event) -> Result<()> {
-        let raw_event = event.raw();
-
-        // Do not deserialise the entire event if we aren't certain it's a
-        // `m.room.redaction`. It saves a non-negligible amount of computations.
-        let Ok(Some(MessageLikeEventType::RoomRedaction)) =
-            raw_event.get_field::<MessageLikeEventType>("type")
+        let Some(event_id) =
+            extract_redaction_target(event.raw(), &self.room_version_rules.redaction)
         else {
             return Ok(());
         };
 
-        // It is a `m.room.redaction`! We can deserialize it entirely.
-
-        let Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomRedaction(
-            redaction,
-        ))) = raw_event.deserialize()
-        else {
-            return Ok(());
-        };
-
-        let Some(event_id) = redaction.redacts(&self.room_version_rules.redaction) else {
-            warn!("missing target event id from the redaction event");
-            return Ok(());
-        };
-
         // Replace the redacted event by a redacted form, if we knew about it.
-        let Some((location, mut target_event)) = self.find_event(event_id).await? else {
+        let Some((location, mut target_event)) = self.find_event(&event_id).await? else {
             trace!("redacted event is missing from the linked chunk");
             return Ok(());
         };