fix(event cache): have redecryption update the event-focused caches too

crates/matrix-sdk/src/event_cache/redecryptor.rs

@@ -121,7 +121,7 @@ use std::{
 
 use as_variant::as_variant;
 use futures_core::Stream;
-use futures_util::{StreamExt, pin_mut};
+use futures_util::{StreamExt, future::join_all, pin_mut};
 #[cfg(doc)]
 use matrix_sdk_base::{BaseClient, crypto::OlmMachine};
 use matrix_sdk_base::{
@@ -381,6 +381,13 @@ impl EventCache {
             pinned_cache.replace_utds(&events).await?;
         }
 
+        // Consider all the live event-focused caches too.
+        // TODO: This ain't great for performance; there shouldn't be that many
+        // event-focused caches alive at the same time, but they could
+        // accumulate over time. Consider keeping track of which linked chunk contain
+        // which event id, to avoid doing the linear searches here.
+        join_all(state.event_focused_caches().map(|cache| cache.replace_utds(&events))).await;
+
         // Consider the room linked chunk.
         for (event_id, decrypted, actions) in events {
             // The event isn't in the cache, nothing to replace. Realistically this can't

crates/matrix-sdk/src/event_cache/room/event_focused.rs

@@ -29,10 +29,12 @@
 //! case where we'd want to persist these caches on disk (e.g., for permalinks
 //! to work across sessions).
 
-use std::sync::Arc;
+use std::{collections::BTreeSet, sync::Arc};
 
+#[cfg(feature = "e2e-encryption")]
+use matrix_sdk_base::linked_chunk::Position;
 use matrix_sdk_base::{
-    deserialized_responses::TimelineEvent,
+    deserialized_responses::{TimelineEvent, TimelineEventKind},
     event_cache::{Event, Gap},
     linked_chunk::OwnedLinkedChunkId,
 };
@@ -40,6 +42,8 @@ use matrix_sdk_common::{
     linked_chunk::{ChunkContent, ChunkIdentifier},
     serde_helpers::extract_thread_root,
 };
+#[cfg(feature = "e2e-encryption")]
+use ruma::EventId;
 use ruma::{OwnedEventId, UInt, api::Direction};
 use tokio::sync::{
     RwLock,
@@ -262,16 +266,15 @@ impl EventFocusedCacheInner {
     /// Propagate changes to the linked chunk update sender.
     fn propagate_changes(&mut self) {
         let updates = self.chunk.store_updates().take();
-        if updates.is_empty() {
-            return;
+        if !updates.is_empty() {
+            let _ = self.linked_chunk_update_sender.send(RoomEventCacheLinkedChunkUpdate {
+                updates,
+                linked_chunk_id: OwnedLinkedChunkId::EventFocused(
+                    self.room.room_id().to_owned(),
+                    self.focused_event_id.clone(),
+                ),
+            });
         }
-        let _ = self.linked_chunk_update_sender.send(RoomEventCacheLinkedChunkUpdate {
-            updates,
-            linked_chunk_id: OwnedLinkedChunkId::EventFocused(
-                self.room.room_id().to_owned(),
-                self.focused_event_id.clone(),
-            ),
-        });
     }
 
     /// Notify subscribers of timeline updates.
@@ -489,6 +492,25 @@ impl EventFocusedCacheInner {
 
         Ok((result.chunk, result.next_batch_token))
     }
+
+    /// Find an event in the linked chunk by its event ID, and return its
+    /// location.
+    ///
+    /// Note: the in-memory content is always the same as the one in the store,
+    /// since the store is updated synchronously with changes in the linked
+    /// chunk, so we can afford to only look for the event in the memory
+    /// linked chunk.
+    // TODO(bnjbvr): common out in EventLinkedChunk! use it both here and for the pinned event
+    // cache.
+    #[cfg(feature = "e2e-encryption")]
+    fn find_event(&self, event_id: &EventId) -> Option<(Position, Event)> {
+        for (position, event) in self.chunk.revents() {
+            if event.event_id().as_deref() == Some(event_id) {
+                return Some((position, event.clone()));
+            }
+        }
+        None
+    }
 }
 
 /// A cache for an event-focused timeline.
@@ -579,6 +601,54 @@ impl EventFocusedCache {
             _ => None,
         }
     }
+
+    /// Try to locate the events in the linked chunk corresponding to the given
+    /// list of decrypted events, and replace them, while alerting observers
+    /// about the update.
+    pub async fn replace_utds(&self, events: &[ResolvedUtd]) {
+        let mut guard = self.inner.write().await;
+
+        let event_set = guard
+            .chunk
+            .events()
+            .filter_map(|(_pos, ev)| ev.event_id())
+            .into_iter()
+            .collect::<BTreeSet<_>>();
+
+        let mut replaced_some = false;
+
+        for (event_id, decrypted, actions) in events {
+            // As a performance optimization, do a lookup in the current pinned events
+            // check, before looking for the event in the linked chunk.
+
+            if !event_set.contains(event_id) {
+                continue;
+            }
+
+            // The event should be in the linked chunk.
+            let Some((position, mut target_event)) = guard.find_event(event_id) else {
+                continue;
+            };
+
+            target_event.kind = TimelineEventKind::Decrypted(decrypted.clone());
+
+            if let Some(actions) = actions {
+                target_event.set_push_actions(actions.clone());
+            }
+
+            guard
+                .chunk
+                .replace_event_at(position, target_event.clone())
+                .expect("position should be valid");
+
+            replaced_some = true;
+        }
+
+        if replaced_some {
+            guard.propagate_changes();
+            guard.notify_subscribers(EventsOrigin::Cache);
+        }
+    }
 }
 
 #[cfg(not(tarpaulin_include))]

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

@@ -1184,6 +1184,12 @@ mod private {
             self.state.pinned_event_cache.get()
         }
 
+        /// Get a reference to all the live [`event_focused_caches`].
+        #[cfg(feature = "e2e-encryption")]
+        pub fn event_focused_caches(&self) -> impl Iterator<Item = &EventFocusedCache> {
+            self.state.event_focused_caches.values()
+        }
+
         /// Get the `waited_for_initial_prev_token` value.
         pub fn waited_for_initial_prev_token(&mut self) -> &mut bool {
             &mut self.state.waited_for_initial_prev_token

crates/matrix-sdk/src/room/mod.rs

@@ -615,6 +615,11 @@ impl Room {
         )
         .await;
 
+        // Save the loaded events into the event cache, if it's set up.
+        if let Ok((cache, _handles)) = self.event_cache().await {
+            cache.save_events(chunk.clone()).await;
+        }
+
         Ok(Messages {
             start: http_response.start,
             end: http_response.end,

testing/matrix-sdk-integration-testing/src/tests/timeline.rs

@@ -1271,9 +1271,6 @@ async fn test_pinned_events_are_decrypted_after_recovering_with_event_not_in_tim
 /// variant even if the focused UTD event isn't part of the main timeline and
 /// thus wasn't put into the event cache by the main timeline backpaginating.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-// FIXME: This test is ignored because R2D2 can't decrypt this event as
-// it's never put into the event cache.
-#[ignore]
 async fn test_permalink_timelines_redecrypt() -> TestResult {
     const RECOVERY_PASSPHRASE: &str = "I am error";