diff --git a/crates/matrix-sdk/src/event_cache/redecryptor.rs b/crates/matrix-sdk/src/event_cache/redecryptor.rs index 76e5266bc..6deaf0467 100644 --- a/crates/matrix-sdk/src/event_cache/redecryptor.rs +++ b/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 diff --git a/crates/matrix-sdk/src/event_cache/room/event_focused.rs b/crates/matrix-sdk/src/event_cache/room/event_focused.rs index 028f8b49a..e0b44f4f9 100644 --- a/crates/matrix-sdk/src/event_cache/room/event_focused.rs +++ b/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::>(); + + 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))] diff --git a/crates/matrix-sdk/src/event_cache/room/mod.rs b/crates/matrix-sdk/src/event_cache/room/mod.rs index 76671be77..35499b404 100644 --- a/crates/matrix-sdk/src/event_cache/room/mod.rs +++ b/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 { + 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 diff --git a/crates/matrix-sdk/src/room/mod.rs b/crates/matrix-sdk/src/room/mod.rs index 88fa702d6..53e247cd9 100644 --- a/crates/matrix-sdk/src/room/mod.rs +++ b/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, diff --git a/testing/matrix-sdk-integration-testing/src/tests/timeline.rs b/testing/matrix-sdk-integration-testing/src/tests/timeline.rs index 647ff7a51..1ab822e7f 100644 --- a/testing/matrix-sdk-integration-testing/src/tests/timeline.rs +++ b/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";