From 4ec9124ce1344e8a1b445ea25c21f3d74d327220 Mon Sep 17 00:00:00 2001 From: mgoldenberg Date: Tue, 10 Mar 2026 09:55:02 -0400 Subject: [PATCH] Allow storing the same `Event` in multiple `LinkedChunk`s of the same `Room` (#6200) # Overview There are scenarios in which it is sensible to have an event exist in the same room more than once. Notably, this is true in the context of a thread, where an event exists in the main timeline of a room, as well as in a thread of that same room. Support for this behavior has been implemented in the `SQLiteEventCacheStore` in #6065; however, this was never implemented for the `IndexeddbEventCacheStore` or the `MemoryStore`. This pull request extends this behavior to both of those stores. # Changes ## Integration Tests First, `test_event_chunks_allows_same_event_in_room_and_thread` was moved from `matrix_sdk_sqlite::event_cache_store` to `matrix_sdk_base::event_cache::store::integration_tests`. Then, a few additional integration tests were added to ensure that behavior is consistent across implementations of `EventCacheStore`. ## `IndexeddbEventCacheStore` In order to accommodate the behavioral changes specified by the integration tests, it was necessary to modify the schema in the IndexedDB implementation of `EventCacheStore`. Namely, the events object store was cleared and removed and then replaced with a nearly identical one, the only difference being the removal of a uniqueness constraint on one of the indices. The remaining changes mostly involved updating the behavior of top-level `EventCacheStore` functions - e.g., filtering out events where they were duplicated or removing positioning information where it was not relevant. ## `MemoryStore` The changes to `MemoryStore` mostly involved updating the behavior of top-level `EventCacheStore` function - e.g., filtering out events where they were duplicated or removing positioning information where it was not relevant. That being said, it also involved some breaking changes to `RelationalLinkedChunk`. 1. `RelationalLinkedChunk::items` - this function returned an `Iterator` that did not contain information about the `LinkedChunkId`, so this information was added to the items in the `Iterator`. 2. `RelationalLinkedChunk::save_item` - this function did not update the item in all linked chunks of the provided `Room`. It now does this, but requires that the provided `Item` be `Clone`. (1) could probably have been a new function, but I thought a nicer interface was worth the breaking change. (2) could probably be prevented by re-organizing `RelationalLinkedChunk`'s internal data structures to remove the `Clone` requirement, but that seemed like it could turn into a large refactoring project, so I opted for something simpler albeit somewhat crude. In both cases, I'm open to suggestions and would be happy to revisit if something else is preferred. --- Closes #6094. - [x] I've documented the public API Changes in the appropriate `CHANGELOG.md` files. - [x] I've read [the `CONTRIBUTING.md` file](https://github.com/matrix-org/matrix-rust-sdk/blob/main/CONTRIBUTING.md), notably the sections about Pull requests, Commit message format, and AI policy. - [ ] This PR was made with the help of AI. Signed-off-by: Michael Goldenberg --------- Signed-off-by: Michael Goldenberg --- crates/matrix-sdk-base/CHANGELOG.md | 5 + .../event_cache/store/integration_tests.rs | 422 ++++++++++++++++++ .../src/event_cache/store/memory_store.rs | 60 ++- crates/matrix-sdk-common/CHANGELOG.md | 10 + .../src/linked_chunk/relational.rs | 42 +- crates/matrix-sdk-indexeddb/CHANGELOG.md | 7 + .../src/event_cache_store/error.rs | 5 +- .../src/event_cache_store/migrations.rs | 75 +++- .../src/event_cache_store/mod.rs | 81 +++- .../serializer/indexed_types.rs | 4 +- .../src/event_cache_store/transaction.rs | 11 + .../src/event_cache_store/types.rs | 15 + .../src/event_cache_store.rs | 83 +--- 13 files changed, 689 insertions(+), 131 deletions(-) diff --git a/crates/matrix-sdk-base/CHANGELOG.md b/crates/matrix-sdk-base/CHANGELOG.md index fe2235275..2a736d523 100644 --- a/crates/matrix-sdk-base/CHANGELOG.md +++ b/crates/matrix-sdk-base/CHANGELOG.md @@ -25,6 +25,11 @@ All notable changes to this project will be documented in this file. ### Features +- Add support in the `MemoryStore`'s implementation of `EventCacheStore` for + having duplicate events in a room, where each duplicate is in a different + `LinkedChunk`. This is useful, e.g., when an event is in a room and a + thread in that room. + (#[6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200)) - Add `StateStore::upsert_thread_subscriptions()` method for bulk upserts. ([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848)) - The `LatestEventValue::LocalHasBeenSent` variant gains a new `event_id: diff --git a/crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs b/crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs index bd137bf72..10d92b8fd 100644 --- a/crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs +++ b/crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs @@ -141,6 +141,10 @@ pub trait EventCacheStoreIntegrationTests { /// already exist in the store. async fn test_linked_chunk_exists_before_referenced(&self); + /// Test that the same event can exist in a room's linked chunk and a + /// thread's linked chunk simultaneously. + async fn test_linked_chunk_allows_same_event_in_room_and_thread(&self); + /// Test loading the last chunk in a linked chunk from the store. async fn test_load_last_chunk(&self); @@ -202,18 +206,34 @@ pub trait EventCacheStoreIntegrationTests { /// Test that an event can be found or not. async fn test_find_event(&self); + /// Test that an event can be found when it exists in both a room and a + /// thread in that room. + async fn test_find_event_when_event_in_room_and_thread(&self); + /// Test that finding event relations works as expected. async fn test_find_event_relations(&self); + /// Test that find event relations works as expected when an event is both a + /// room and a thread in that room. + async fn test_find_event_relations_when_event_in_room_and_thread(&self); + /// Test that getting all events in a room works as expected. async fn test_get_room_events(&self); /// Test that getting events in a room of a certain type works as expected. async fn test_get_room_events_filtered(&self); + /// Test that getting all events in a room works as expected when the event + /// is in both a room and thread in that room. + async fn test_get_room_events_with_event_in_room_and_thread(&self); + /// Test that saving an event works as expected. async fn test_save_event(&self); + /// Test that saving an existing event updates it's contents in both room + /// and thread linked chunks. + async fn test_save_event_updates_event_in_room_and_thread(&self); + /// Test multiple things related to distinguishing a thread linked chunk /// from a room linked chunk. async fn test_thread_vs_room_linked_chunk(&self); @@ -355,6 +375,62 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore { .unwrap_err(); } + async fn test_linked_chunk_allows_same_event_in_room_and_thread(&self) { + // This test verifies that the same event can appear in both a room's linked + // chunk and a thread's linked chunk. This is the real-world use case: + // a thread reply appears in both the main room timeline and the thread. + + let room_id = *DEFAULT_TEST_ROOM_ID; + let thread_root = event_id!("$thread_root"); + + // Create an event that will be inserted into both the room and thread linked + // chunks. + let event_id = event_id!("$thread_reply"); + let event = make_test_event_with_event_id(room_id, "thread reply", Some(event_id)); + + let room_linked_chunk_id = LinkedChunkId::Room(room_id); + let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); + + // Insert the event into the room's linked chunk. + self.handle_linked_chunk_updates( + room_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] }, + ], + ) + .await + .unwrap(); + + // Insert the same event into the thread's linked chunk. + self.handle_linked_chunk_updates( + thread_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event] }, + ], + ) + .await + .unwrap(); + + // Verify both entries exist by loading chunks from both linked chunk IDs. + let room_chunks = self.load_all_chunks(room_linked_chunk_id).await.unwrap(); + let thread_chunks = self.load_all_chunks(thread_linked_chunk_id).await.unwrap(); + + assert_eq!(room_chunks.len(), 1); + assert_eq!(thread_chunks.len(), 1); + + // Verify the event is in both. + assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => { + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_id().as_deref(), Some(event_id)); + }); + assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => { + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_id().as_deref(), Some(event_id)); + }); + } + async fn test_load_all_chunks_metadata(&self) { let room_id = room_id!("!r0:matrix.org"); let linked_chunk_id = LinkedChunkId::Room(room_id); @@ -1460,6 +1536,75 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore { ); } + async fn test_find_event_when_event_in_room_and_thread(&self) { + let room_id = *DEFAULT_TEST_ROOM_ID; + let thread_root = event_id!("$thread_root"); + + // Create an event that will be only be inserted into the room + let room_event_id = event_id!("$room_event"); + let room_event = make_test_event_with_event_id(room_id, "room event", Some(room_event_id)); + + // Create an event that will only be inserted into the thread + let thread_event_id = event_id!("$thread_event"); + let thread_event = + make_test_event_with_event_id(room_id, "thread event", Some(thread_event_id)); + + // Create an event that will be inserted into both the room and thread linked + // chunks. + let room_and_thread_event_id = event_id!("$room_and_thread"); + let room_and_thread_event = make_test_event_with_event_id( + room_id, + "room and thread", + Some(room_and_thread_event_id), + ); + + let room_linked_chunk_id = LinkedChunkId::Room(room_id); + let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); + + // Insert the relevant events into the room's linked chunk. + self.handle_linked_chunk_updates( + room_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![room_event, room_and_thread_event.clone()], + }, + ], + ) + .await + .unwrap(); + + // Insert the relevant events into the thread's linked chunk. + self.handle_linked_chunk_updates( + thread_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![thread_event, room_and_thread_event], + }, + ], + ) + .await + .unwrap(); + + // Verify that event that is only in the room can be retrieved + assert_matches!(self.find_event(room_id, room_event_id).await, Ok(Some(event)) => { + assert_eq!(event.event_id().unwrap(), room_event_id) + }); + + // Verify that the event that is only in the thread can be retrieved + assert_matches!(self.find_event(room_id, thread_event_id).await, Ok(Some(event)) => { + assert_eq!(event.event_id().unwrap(), thread_event_id) + }); + + // Verify that event that is in both room and thread can be retrieved + assert_matches!(self.find_event(room_id, room_and_thread_event_id).await, Ok(Some(event)) => { + assert_eq!(event.event_id().unwrap(), room_and_thread_event_id); + }); + } + async fn test_find_event_relations(&self) { let room_id = room_id!("!r0:matrix.org"); let another_room_id = room_id!("!r1:matrix.org"); @@ -1574,6 +1719,121 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore { ); } + async fn test_find_event_relations_when_event_in_room_and_thread(&self) { + let room_id = *DEFAULT_TEST_ROOM_ID; + let thread_root = event_id!("$thread_root"); + + // Create an event that will inserted into both the room and thread linked + // chunks. + let event_id = event_id!("$event"); + let event = make_test_event_with_event_id(room_id, "event", Some(event_id)); + + // Create an event that will only be inserted into the thread in order to help + // distinguish between the room and thread linked chunks. + let extra_thread_event_id = event_id!("$extra_thread_event"); + let extra_thread_event = make_test_event_with_event_id( + room_id, + "extra thread event", + Some(extra_thread_event_id), + ); + + // Create a reaction that will only be inserted into the room + let room_reaction_id = event_id!("$room_reaction"); + let room_reaction = EventFactory::new() + .room(room_id) + .sender(*ALICE) + .reaction(event_id, "room") + .event_id(room_reaction_id) + .into_event(); + + // Create a reaction that will only be inserted into the thread + let thread_reaction_id = event_id!("$thread_reaction"); + let thread_reaction = EventFactory::new() + .room(room_id) + .sender(*ALICE) + .reaction(event_id, "thread") + .event_id(thread_reaction_id) + .into_event(); + + // Create a reaction that will be inserted into both the room and thread linked + // chunks. + let room_and_thread_reaction_id = event_id!("$room_and_thread_reaction"); + let room_and_thread_reaction = EventFactory::new() + .room(room_id) + .sender(*ALICE) + .reaction(event_id, "room and thread") + .event_id(room_and_thread_reaction_id) + .into_event(); + + let room_linked_chunk_id = LinkedChunkId::Room(room_id); + let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); + + // Insert the relevant events into the room's linked chunk. + self.handle_linked_chunk_updates( + room_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![event.clone(), room_reaction, room_and_thread_reaction.clone()], + }, + ], + ) + .await + .unwrap(); + + // Insert the relevant events into the thread's linked chunk. + self.handle_linked_chunk_updates( + thread_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![ + event.clone(), + extra_thread_event, + thread_reaction, + room_and_thread_reaction, + ], + }, + ], + ) + .await + .unwrap(); + + // Verify that only related events from the room are returned + assert_matches!(self.find_event_relations(room_id, event_id, None).await, Ok(relations) => { + assert_eq!(relations.len(), 3); + // Verify that room reaction is in the list and associated with its + // position in the room linked chunk. + let room_relation = relations + .iter() + .find(|relation| relation.0.event_id().unwrap() == room_reaction_id) + .unwrap(); + assert_matches!(room_relation, (_, Some(position)) => { + assert_eq!(*position, Position::new(CId::new(1), 1)); + }); + + // Verify that thread reaction is in the list and not associated with a + // position, as all positions are provided for the room linked chunk. + let thread_relation = relations + .iter() + .find(|relation| relation.0.event_id().unwrap() == thread_reaction_id) + .unwrap(); + assert_matches!(thread_relation, (_, None)); + + // Verify that room and thread reaction is in the list and associated + // with its position in the room linked chunk, not the thread linked chunk. + let room_and_thread_relation = relations + .iter() + .find(|relation| relation.0.event_id().unwrap() == room_and_thread_reaction_id) + .unwrap(); + assert_matches!(room_and_thread_relation, (_, Some(position)) => { + assert_eq!(*position, Position::new(CId::new(1), 2)); + }); + }); + } + async fn test_get_room_events(&self) { let room_id = room_id!("!r0:matrix.org"); let another_room_id = room_id!("!r1:matrix.org"); @@ -1702,6 +1962,73 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore { assert_expected_events!(events, [first_event]); } + async fn test_get_room_events_with_event_in_room_and_thread(&self) { + let room_id = *DEFAULT_TEST_ROOM_ID; + let thread_root = event_id!("$thread_root"); + + // Create an event that will be only be inserted into the room + let room_event_id = event_id!("$room_event"); + let room_event = make_test_event_with_event_id(room_id, "room event", Some(room_event_id)); + + // Create an event that will only be inserted into the thread. This may not be a + // sensible operation in practice, as threads seem to always exist in a + // room, but let's test it anyway. + let thread_event_id = event_id!("$thread_event"); + let thread_event = + make_test_event_with_event_id(room_id, "thread event", Some(thread_event_id)); + + // Create an event that will be inserted into both the room and thread linked + // chunks. + let room_and_thread_event_id = event_id!("$room_and_thread"); + let room_and_thread_event = make_test_event_with_event_id( + room_id, + "room and thread", + Some(room_and_thread_event_id), + ); + + let room_linked_chunk_id = LinkedChunkId::Room(room_id); + let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); + + // Insert the relevant events into the room's linked chunk. + self.handle_linked_chunk_updates( + room_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![room_event, room_and_thread_event.clone()], + }, + ], + ) + .await + .unwrap(); + + // Insert the relevant events into the thread's linked chunk. + self.handle_linked_chunk_updates( + thread_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { + at: Position::new(CId::new(1), 0), + items: vec![thread_event, room_and_thread_event], + }, + ], + ) + .await + .unwrap(); + + // Verify that all events can be retrieved and none are duplicated in the + // returned list. + let expected_event_ids = + BTreeSet::from([room_event_id, thread_event_id, room_and_thread_event_id]); + assert_matches!(self.get_room_events(room_id, None, None).await, Ok(events) => { + assert_eq!(events.len(), 3); + assert!(events.iter().all(|event| { + expected_event_ids.contains(&event.event_id().unwrap().as_ref()) + })); + }); + } + async fn test_save_event(&self) { let room_id = room_id!("!r0:matrix.org"); let another_room_id = room_id!("!r1:matrix.org"); @@ -1746,6 +2073,66 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore { ); } + async fn test_save_event_updates_event_in_room_and_thread(&self) { + let room_id = *DEFAULT_TEST_ROOM_ID; + let thread_root = event_id!("$thread_root"); + + // Create an event that will be inserted into both the room and thread linked + // chunks. + let event_id = event_id!("$event"); + let event = make_test_event_with_event_id(room_id, "event", Some(event_id)); + + let room_linked_chunk_id = LinkedChunkId::Room(room_id); + let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); + + // Insert the relevant events into the room's linked chunk. + self.handle_linked_chunk_updates( + room_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] }, + ], + ) + .await + .unwrap(); + + // Insert the relevant events into the thread's linked chunk. + self.handle_linked_chunk_updates( + thread_linked_chunk_id, + vec![ + Update::NewItemsChunk { previous: None, new: CId::new(1), next: None }, + Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] }, + ], + ) + .await + .unwrap(); + + // Save updated version of original event, which should replace the content of + // the existing event + let updated_content = "updated content"; + let updated = make_test_event_with_event_id(room_id, updated_content, Some(event_id)); + self.save_event(room_id, updated).await.unwrap(); + + // Load all chunks from both room and thread + let room_chunks = self.load_all_chunks(room_linked_chunk_id).await.unwrap(); + let thread_chunks = self.load_all_chunks(thread_linked_chunk_id).await.unwrap(); + + assert_eq!(room_chunks.len(), 1); + assert_eq!(thread_chunks.len(), 1); + + // Verify the event has been updated in both room and thread + assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => { + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_id().as_deref(), Some(event_id)); + check_test_event(&events[0], updated_content); + }); + assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => { + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_id().as_deref(), Some(event_id)); + check_test_event(&events[0], updated_content); + }); + } + async fn test_thread_vs_room_linked_chunk(&self) { let room_id = room_id!("!r0:matrix.org"); @@ -1930,6 +2317,13 @@ macro_rules! event_cache_store_integration_tests { event_cache_store.test_linked_chunk_exists_before_referenced().await; } + #[async_test] + async fn test_linked_chunk_allow_same_event_in_room_and_thread() { + let event_cache_store = + get_event_cache_store().await.unwrap().into_event_cache_store(); + event_cache_store.test_linked_chunk_allows_same_event_in_room_and_thread().await; + } + #[async_test] async fn test_load_last_chunk() { let event_cache_store = @@ -2063,6 +2457,13 @@ macro_rules! event_cache_store_integration_tests { event_cache_store.test_find_event().await; } + #[async_test] + async fn test_find_event_when_event_in_room_and_thread() { + let event_cache_store = + get_event_cache_store().await.unwrap().into_event_cache_store(); + event_cache_store.test_find_event_when_event_in_room_and_thread().await; + } + #[async_test] async fn test_find_event_relations() { let event_cache_store = @@ -2070,6 +2471,13 @@ macro_rules! event_cache_store_integration_tests { event_cache_store.test_find_event_relations().await; } + #[async_test] + async fn test_find_event_relations_when_event_in_room_and_thread() { + let event_cache_store = + get_event_cache_store().await.unwrap().into_event_cache_store(); + event_cache_store.test_find_event_relations_when_event_in_room_and_thread().await; + } + #[async_test] async fn test_get_room_events() { let event_cache_store = @@ -2084,6 +2492,13 @@ macro_rules! event_cache_store_integration_tests { event_cache_store.test_get_room_events_filtered().await; } + #[async_test] + async fn test_get_room_events_with_event_in_room_and_thread() { + let event_cache_store = + get_event_cache_store().await.unwrap().into_event_cache_store(); + event_cache_store.test_get_room_events_with_event_in_room_and_thread().await; + } + #[async_test] async fn test_save_event() { let event_cache_store = @@ -2091,6 +2506,13 @@ macro_rules! event_cache_store_integration_tests { event_cache_store.test_save_event().await; } + #[async_test] + async fn test_save_event_updates_event_in_room_and_thread() { + let event_cache_store = + get_event_cache_store().await.unwrap().into_event_cache_store(); + event_cache_store.test_save_event_updates_event_in_room_and_thread().await; + } + #[async_test] async fn test_thread_vs_room_linked_chunk() { let event_cache_store = diff --git a/crates/matrix-sdk-base/src/event_cache/store/memory_store.rs b/crates/matrix-sdk-base/src/event_cache/store/memory_store.rs index 283e8a630..eca1b317b 100644 --- a/crates/matrix-sdk-base/src/event_cache/store/memory_store.rs +++ b/crates/matrix-sdk-base/src/event_cache/store/memory_store.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, sync::{Arc, RwLock as StdRwLock}, }; @@ -188,10 +188,9 @@ impl EventCacheStore for MemoryStore { ) -> Result, Self::Error> { let inner = self.inner.read().unwrap(); - let event = inner - .events - .items(room_id) - .find_map(|(event, _pos)| (event.event_id()? == event_id).then_some(event.clone())); + let event = inner.events.items(room_id).find_map(|(_, (event, _pos))| { + (event.event_id()? == event_id).then_some(event.clone()) + }); Ok(event) } @@ -204,10 +203,10 @@ impl EventCacheStore for MemoryStore { ) -> Result)>, Self::Error> { let inner = self.inner.read().unwrap(); - let related_events = inner + let related_events: Vec<_> = inner .events .items(room_id) - .filter_map(|(event, pos)| { + .filter_map(|(linked_chunk_id, (event, pos))| { // Must have a relation. let (related_to, rel_type) = extract_event_relation(event.raw())?; let rel_type = RelationType::from(rel_type.as_str()); @@ -219,14 +218,35 @@ impl EventCacheStore for MemoryStore { // Must not be filtered out. if let Some(filters) = &filters { - filters.contains(&rel_type).then_some((event.clone(), pos)) + filters.contains(&rel_type).then_some((linked_chunk_id, (event.clone(), pos))) } else { - Some((event.clone(), pos)) + Some((linked_chunk_id, (event.clone(), pos))) } }) .collect(); - Ok(related_events) + // Remove any duplicate events which may exist in both a room and thread + // linked chunk. Additionally, remove any position information from non-room + // linked chunks. + let mut deduplicated = HashMap::new(); + for (linked_chunk_id, (event, position)) in related_events { + let event_id = event + .event_id() + .ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?; + match linked_chunk_id.as_ref() { + LinkedChunkId::Room(_) => { + // Prioritize events that come from a room linked chunk + deduplicated.insert(event_id, (event, position)); + } + _ => { + // Remove position information from events that come + // from any other type of linked chunk + deduplicated.entry(event_id).or_insert_with(|| (event, None)); + } + } + } + + Ok(deduplicated.into_values().collect()) } async fn get_room_events( @@ -237,17 +257,29 @@ impl EventCacheStore for MemoryStore { ) -> Result, Self::Error> { let inner = self.inner.read().unwrap(); - let event: Vec<_> = inner + let (_, event): (_, Vec<_>) = inner .events .items(room_id) - .map(|(event, _pos)| event.clone()) + .map(|(_, (event, _pos))| event.clone()) .filter(|e| { event_type .is_none_or(|event_type| Some(event_type) == e.kind.event_type().as_deref()) }) .filter(|e| session_id.is_none_or(|s| Some(s) == e.kind.session_id())) - .collect(); - + .map(|e| { + e.event_id() + .map(|id| (id, e)) + .ok_or(Self::Error::InvalidData { details: String::from("missing event id") }) + }) + .collect::>>()? + .into_iter() + .fold((HashSet::new(), Vec::new()), |(mut ids, mut es), (id, e)| { + if !ids.contains(&id) { + ids.insert(id); + es.push(e); + } + (ids, es) + }); Ok(event) } diff --git a/crates/matrix-sdk-common/CHANGELOG.md b/crates/matrix-sdk-common/CHANGELOG.md index f0e487b39..2c7b1b711 100644 --- a/crates/matrix-sdk-common/CHANGELOG.md +++ b/crates/matrix-sdk-common/CHANGELOG.md @@ -7,7 +7,17 @@ All notable changes to this project will be documented in this file. ## [Unreleased] - ReleaseDate ### Features +- Add support in the `MemoryStore`'s implementation of `EventCacheStore` for + having duplicate events in a room, where each duplicate is in a different + `LinkedChunk`. This is useful, e.g., when an event is in a room and a + thread in that room. + +- [**breaking**] In order to support having duplicate events in the same room (in different `LinkedChunk`'s) a few + functions were changed in `RelationalLinkedChunk`. The items in the `Iterator` returned by `RelationalLinkedChunk::items` + now also include the `LinkedChunkId` in which the `Item` was found. Additionally, `RelationalLinkedChunk::save_item` + now requires the `Item` to be `Clone` as it may be stored in multiple `LinkedChunk`s. + (#[6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200)) - [**breaking**] Added `CrossProcessLockConfig`, which can be used to configure the behavior of the cross-process lock. `CrossProcessLock` now takes a `CrossProcessLockConfig` as an argument to its constructor instead of a `lock_holder` value. ([#6160](https://github.com/matrix-org/matrix-rust-sdk/pull/6160)) diff --git a/crates/matrix-sdk-common/src/linked_chunk/relational.rs b/crates/matrix-sdk-common/src/linked_chunk/relational.rs index ed2a3fa23..c5cfbaf39 100644 --- a/crates/matrix-sdk-common/src/linked_chunk/relational.rs +++ b/crates/matrix-sdk-common/src/linked_chunk/relational.rs @@ -16,7 +16,7 @@ //! [`RelationalLinkedChunk`]. use std::{ - collections::{BTreeMap, HashMap}, + collections::{BTreeMap, HashMap, HashSet}, hash::Hash, }; @@ -395,7 +395,8 @@ where } /// Return an iterator over all items of all linked chunks of a room, along - /// with their positions, if available. + /// with the linked chunk they are in and the position in that linked chunk, + /// if available. /// /// The only items which will NOT have a position are those saved with /// [`Self::save_item`]. @@ -404,26 +405,41 @@ where pub fn items<'a>( &'a self, room_id: &'a RoomId, - ) -> impl Iterator)> { + ) -> impl Iterator))> { self.items .iter() - .filter_map(move |(linked_chunk_id, items)| { - (linked_chunk_id.room_id() == room_id).then_some(items) + .filter(move |(linked_chunk_id, _)| linked_chunk_id.room_id() == room_id) + .flat_map(|(linked_chunk_id, items)| { + items.values().map(move |(item, pos)| (linked_chunk_id, (item, *pos))) }) - .flat_map(|items| items.values().map(|(item, pos)| (item, *pos))) } +} +impl RelationalLinkedChunk +where + Item: IndexableItem + Clone, + ItemId: Hash + PartialEq + Eq + Clone + Ord, +{ /// Save a single item "out-of-band" in the relational linked chunk. pub fn save_item(&mut self, room_id: OwnedRoomId, item: Item) { let id = item.id(); - let linked_chunk_id = OwnedLinkedChunkId::Room(room_id); - let map = self.items.entry(linked_chunk_id).or_default(); - if let Some(prev_value) = map.get_mut(&id) { - // If the item already exists, we keep the position. - prev_value.0 = item; - } else { - map.insert(id, (item, None)); + let mut linked_chunk_ids = self + .items + .keys() + .filter(|linked_chunk_id| linked_chunk_id.room_id() == room_id) + .cloned() + .collect::>(); + linked_chunk_ids.insert(OwnedLinkedChunkId::Room(room_id)); + + for linked_chunk_id in linked_chunk_ids { + let map = self.items.entry(linked_chunk_id).or_default(); + if let Some(prev_value) = map.get_mut(&id) { + // If the item already exists, we keep the position. + prev_value.0 = item.clone(); + } else { + map.insert(id.clone(), (item.clone(), None)); + } } } } diff --git a/crates/matrix-sdk-indexeddb/CHANGELOG.md b/crates/matrix-sdk-indexeddb/CHANGELOG.md index abf734ad5..be71c34ee 100644 --- a/crates/matrix-sdk-indexeddb/CHANGELOG.md +++ b/crates/matrix-sdk-indexeddb/CHANGELOG.md @@ -8,6 +8,13 @@ All notable changes to this project will be documented in this file. ### Features +- Add support in the implementation of `EventCacheStore` for + having duplicate events in a room, where each duplicate is in a different + `LinkedChunk`. This is useful, e.g., when an event is in a room and a + thread in that room. The change involves a database migration where + the `EVENTS` object store is cleared and then modified so that the + `ROOM` index no longer requires keys to be unique. + ([#6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200)) - Implement `CryptoStore::get_pending_key_bundle_details_for_room` and `CryptoStore::get_all_rooms_pending_key_bundle`, and process `rooms_pending_key_bundle` field in `Changes`. diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/error.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/error.rs index c70e70ddd..8051e7176 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/error.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/error.rs @@ -32,6 +32,8 @@ pub enum IndexeddbEventCacheStoreError { UnableToLoadChunk, #[error("no max chunk id")] NoMaxChunkId, + #[error("event without id")] + EventWithoutId, #[error("transaction: {0}")] Transaction(#[from] TransactionError), } @@ -68,7 +70,8 @@ impl From for EventCacheStoreError { | ChunksContainCycle | ChunksContainDisjointLists | NoMaxChunkId - | UnableToLoadChunk => Self::InvalidData { details: value.to_string() }, + | UnableToLoadChunk + | EventWithoutId => Self::InvalidData { details: value.to_string() }, Transaction(inner) => inner.into(), } } diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/migrations.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/migrations.rs index 91140762c..a0f75d139 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/migrations.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/migrations.rs @@ -21,10 +21,10 @@ use thiserror::Error; /// The current version and keys used in the database. pub mod current { - use super::{Version, v2}; + use super::{Version, v3}; - pub const VERSION: Version = Version::V2; - pub use v2::keys; + pub const VERSION: Version = Version::V3; + pub use v3::keys; } /// Opens a connection to the IndexedDB database and takes care of upgrading it @@ -56,6 +56,8 @@ pub enum Version { V1 = 1, /// Version 2 of the database, for details see [`v2`]. V2 = 2, + /// Version 3 of the database, for details see [`v3`]. + V3 = 3, } impl Version { @@ -64,7 +66,8 @@ impl Version { match self { Self::V0 => v0::upgrade(transaction).map(Some), Self::V1 => v1::upgrade(transaction).map(Some), - Self::V2 => Ok(None), + Self::V2 => v2::upgrade(transaction).map(Some), + Self::V3 => Ok(None), } } } @@ -81,6 +84,7 @@ impl TryFrom for Version { 0 => Ok(Version::V0), 1 => Ok(Version::V1), 2 => Ok(Version::V2), + 3 => Ok(Version::V3), v => Err(UnknownVersionError(v)), } } @@ -225,4 +229,67 @@ mod v2 { Ok(()) } + + /// Upgrade database from `v2` to `v3` + pub fn upgrade(transaction: &Transaction<'_>) -> Result { + v3::update_events_object_store(transaction)?; + Ok(Version::V3) + } +} + +mod v3 { + use indexed_db_futures::Build; + + // Re-use all the same keys from `v2`. + pub use super::v2::keys; + use super::*; + + /// Update the events object store, so that the `room` index is no longer + /// unique. This allows an event to be stored in a room twice - e.g., once + /// in the main thread and once in a side thread. + /// + /// Note that this operation removes the existing events object store and + /// all of its contents. + pub fn update_events_object_store(transaction: &Transaction<'_>) -> Result<(), Error> { + remove_events_object_store(transaction)?; + create_events_object_store(transaction.db())?; + Ok(()) + } + + /// Remove events object store + pub fn remove_events_object_store(transaction: &Transaction<'_>) -> Result<(), Error> { + let object_store = transaction.object_store(keys::EVENTS)?; + // It is faster to clear all events first, then delete the object store rather + // than immediately deleting. + // + // For details, see https://www.artificialworlds.net/blog/2024/02/02/deleting-an-indexed-db-store-can-be-incredibly-slow-on-firefox/ + object_store.clear()?; + transaction.db().delete_object_store(keys::EVENTS)?; + Ok(()) + } + + /// Create an object store for tracking information about events. + /// + /// * Primary Key - `id` + /// * Index - `room` - tracks whether an event is in a given room + /// * Index (unique) - `position` - tracks position of an event in linked + /// chunks + /// * Index - `relation` - tracks any event to which the given event is + /// related + pub fn create_events_object_store(db: &Database) -> Result<(), Error> { + let events = db + .create_object_store(keys::EVENTS) + .with_key_path(keys::EVENTS_KEY_PATH.into()) + .build()?; + let _ = + events.create_index(keys::EVENTS_ROOM, keys::EVENTS_ROOM_KEY_PATH.into()).build()?; + let _ = events + .create_index(keys::EVENTS_POSITION, keys::EVENTS_POSITION_KEY_PATH.into()) + .with_unique(true) + .build()?; + let _ = events + .create_index(keys::EVENTS_RELATION, keys::EVENTS_RELATION_KEY_PATH.into()) + .build()?; + Ok(()) + } } diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs index 108812cab..6056566cf 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs @@ -14,7 +14,7 @@ #![cfg_attr(not(test), allow(unused))] -use std::{rc::Rc, time::Duration}; +use std::{collections::HashMap, rc::Rc, time::Duration}; use indexed_db_futures::{Build, database::Database}; #[cfg(target_family = "wasm")] @@ -464,9 +464,9 @@ impl EventCacheStore for IndexeddbEventCacheStore { let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?; transaction - .get_event_by_room(room_id, event_id) + .get_events_by_room(room_id, event_id) .await - .map(|ok| ok.map(Into::into)) + .map(|mut events| events.pop().map(Into::into)) .map_err(Into::into) } @@ -481,26 +481,60 @@ impl EventCacheStore for IndexeddbEventCacheStore { let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?; - let mut related_events = Vec::new(); + let mut related_events = HashMap::::new(); match filters { Some(relation_types) if !relation_types.is_empty() => { for relation_type in relation_types { let relation = (event_id, relation_type); let events = transaction.get_events_by_relation(room_id, relation).await?; for event in events { - let position = event.position().map(Into::into); - related_events.push((event.into(), position)); + let Some(event_id) = event.event_id() else { + return Err(IndexeddbEventCacheStoreError::EventWithoutId); + }; + match event.linked_chunk_id() { + LinkedChunkId::Room(_) => { + // Prioritize events that come from a room linked chunk + related_events.insert(event_id, event); + } + _ => { + // Remove position information from events that come + // from any other type of linked chunk + related_events + .entry(event_id) + .or_insert_with(|| event.into_out_of_band_event()); + } + } } } } _ => { for event in transaction.get_events_by_related_event(room_id, event_id).await? { - let position = event.position().map(Into::into); - related_events.push((event.into(), position)); + let Some(event_id) = event.event_id() else { + return Err(IndexeddbEventCacheStoreError::EventWithoutId); + }; + match event.linked_chunk_id() { + LinkedChunkId::Room(_) => { + // Prioritize events that come from a room linked chunk + related_events.insert(event_id, event); + } + _ => { + // Remove position information from events that come + // from any other type of linked chunk + related_events + .entry(event_id) + .or_insert_with(|| event.into_out_of_band_event()); + } + } } } } - Ok(related_events) + Ok(related_events + .into_values() + .map(|event| { + let position = event.position().map(Into::into); + (event.into(), position) + }) + .collect()) } #[instrument(skip(self))] @@ -519,7 +553,14 @@ impl EventCacheStore for IndexeddbEventCacheStore { transaction .get_room_events(room_id) .await - .map(|vec| { + .map(|mut vec| { + vec.dedup_by(|a, b| { + if let (Some(a), Some(b)) = (a.event_id(), b.event_id()) { + a == b + } else { + false + } + }); vec.into_iter() .map(Event::from) .filter(|e| { @@ -546,15 +587,23 @@ impl EventCacheStore for IndexeddbEventCacheStore { return Ok(()); }; let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readwrite)?; - let event = match transaction.get_event_by_room(room_id, &event_id).await? { - Some(inner) => inner.with_content(event), - None => types::Event::OutOfBand(OutOfBandEvent { + + let mut events = transaction + .get_events_by_room(room_id, &event_id) + .await? + .into_iter() + .map(|e| e.with_content(event.clone())) + .collect::>(); + if events.is_empty() { + events.push(types::Event::OutOfBand(OutOfBandEvent { linked_chunk_id: LinkedChunkId::Room(room_id).to_owned(), content: event, position: (), - }), - }; - transaction.put_event(&event).await?; + })); + } + for event in events { + transaction.put_event(&event).await?; + } transaction.commit().await?; Ok(()) } diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/serializer/indexed_types.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/serializer/indexed_types.rs index f8455b0ea..204e3ded0 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/serializer/indexed_types.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/serializer/indexed_types.rs @@ -420,13 +420,13 @@ impl IndexedPrefixKeyBounds> for IndexedEventIdKey { } } -/// The value associated with the [primary key](IndexedEvent::id) of the +/// The value associated with the [`room`](IndexedEvent::room) index of the /// [`EVENTS`][1] object store, which is constructed from: /// /// - The (possibly) hashed Room ID /// - The (possibly) hashed Event ID. /// -/// [1]: crate::event_cache_store::migrations::v1::create_events_object_store +/// [1]: crate::event_cache_store::migrations::v3::create_events_object_store #[derive(Debug, Serialize, Deserialize)] pub struct IndexedEventRoomKey(IndexedRoomId, IndexedEventId); diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs index 681b443f5..5240a9857 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs @@ -350,6 +350,17 @@ impl<'a> IndexeddbEventCacheStoreTransaction<'a> { self.get_item_by_key::(key).await } + /// Query IndexedDB for events that match the given event id in the given + /// room. + pub async fn get_events_by_room( + &self, + room_id: &RoomId, + event_id: &EventId, + ) -> Result, TransactionError> { + let key: IndexedEventRoomKey = self.serializer().encode_key((room_id, event_id)); + self.get_items_by_key::(key).await + } + /// Query IndexedDB for events that are in the given /// room. pub async fn get_room_events(&self, room_id: &RoomId) -> Result, TransactionError> { diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs index 42058c404..94b7d0326 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs @@ -136,6 +136,15 @@ impl Event { } self } + + /// Ensures that the underlying [`GenericEvent`] is an [`OutOfBandEvent`]. + /// If it is not an [`OutOfBandEvent`], then it is converted into one. + pub fn into_out_of_band_event(self) -> Self { + match self { + Event::InBand(i) => Self::OutOfBand(i.into()), + o @ Event::OutOfBand(_) => o, + } + } } /// A generic representation of an @@ -182,6 +191,12 @@ pub type InBandEvent = GenericEvent; /// events which are not part of a chunk and therefore have no position. pub type OutOfBandEvent = GenericEvent<()>; +impl From for OutOfBandEvent { + fn from(value: InBandEvent) -> Self { + Self { linked_chunk_id: value.linked_chunk_id, content: value.content, position: () } + } +} + /// A representation of [`Position`](matrix_sdk_base::linked_chunk::Position) /// which can be stored in IndexedDB. #[derive(Debug, Default, Copy, Clone, Serialize, Deserialize)] diff --git a/crates/matrix-sdk-sqlite/src/event_cache_store.rs b/crates/matrix-sdk-sqlite/src/event_cache_store.rs index e88a6121d..794f9323e 100644 --- a/crates/matrix-sdk-sqlite/src/event_cache_store.rs +++ b/crates/matrix-sdk-sqlite/src/event_cache_store.rs @@ -1657,13 +1657,12 @@ mod tests { use matrix_sdk_base::{ event_cache::store::{ EventCacheStore, EventCacheStoreError, IntoEventCacheStore, - integration_tests::{EventCacheStoreIntegrationTests, make_test_event_with_event_id}, + integration_tests::EventCacheStoreIntegrationTests, }, event_cache_store_integration_tests, event_cache_store_integration_tests_time, - linked_chunk::{ChunkContent, ChunkIdentifier, LinkedChunkId, Position, Update}, + linked_chunk::{ChunkIdentifier, LinkedChunkId, Update}, }; use matrix_sdk_test::{DEFAULT_TEST_ROOM_ID, async_test}; - use ruma::event_id; use tempfile::{TempDir, tempdir}; use super::SqliteEventCacheStore; @@ -1827,84 +1826,6 @@ mod tests { let chunks = store.load_all_chunks(linked_chunk_id).await.unwrap(); assert!(chunks.is_empty()); } - - #[async_test] - async fn test_event_chunks_allows_same_event_in_room_and_thread() { - // This test verifies that the same event can appear in both a room's linked - // chunk and a thread's linked chunk. This is the real-world use case: - // a thread reply appears in both the main room timeline and the thread. - - let store = get_event_cache_store().await.expect("creating cache store failed"); - - let room_id = *DEFAULT_TEST_ROOM_ID; - let thread_root = event_id!("$thread_root"); - - // Create an event that will be inserted into both the room and thread linked - // chunks. - let event = make_test_event_with_event_id( - room_id, - "thread reply", - Some(event_id!("$thread_reply")), - ); - - let room_linked_chunk_id = LinkedChunkId::Room(room_id); - let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root); - - // Insert the event into the room's linked chunk. - store - .handle_linked_chunk_updates( - room_linked_chunk_id, - vec![ - Update::NewItemsChunk { - previous: None, - new: ChunkIdentifier::new(1), - next: None, - }, - Update::PushItems { - at: Position::new(ChunkIdentifier::new(1), 0), - items: vec![event.clone()], - }, - ], - ) - .await - .unwrap(); - - // Insert the same event into the thread's linked chunk. - store - .handle_linked_chunk_updates( - thread_linked_chunk_id, - vec![ - Update::NewItemsChunk { - previous: None, - new: ChunkIdentifier::new(1), - next: None, - }, - Update::PushItems { - at: Position::new(ChunkIdentifier::new(1), 0), - items: vec![event], - }, - ], - ) - .await - .unwrap(); - - // Verify both entries exist by loading chunks from both linked chunk IDs. - let room_chunks = store.load_all_chunks(room_linked_chunk_id).await.unwrap(); - let thread_chunks = store.load_all_chunks(thread_linked_chunk_id).await.unwrap(); - - assert_eq!(room_chunks.len(), 1); - assert_eq!(thread_chunks.len(), 1); - - // Verify the event is in both. - assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => { - assert_eq!(events.len(), 1); - assert_eq!(events[0].event_id().as_deref(), Some(event_id!("$thread_reply"))); - }); - assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => { - assert_eq!(events.len(), 1); - assert_eq!(events[0].event_id().as_deref(), Some(event_id!("$thread_reply"))); - }); - } } #[cfg(test)]