feat: Allow events to be fetched by event type

Author: poljar

crates/matrix-sdk-base/CHANGELOG.md

@@ -13,6 +13,11 @@ All notable changes to this project will be documented in this file.
 
 ### Features
 
+- [**breaking**] The `EventCacheStore::get_room_events()` method has received
+  two new arguments. This allows users to load only events of a certain event
+  type and events that were encrypted using a certain room key identified by its
+  session ID.
+  ([#5817](https://github.com/matrix-org/matrix-rust-sdk/pull/5817))
 - `ComposerDraft` can now store attachments alongside text messages.
   ([#5794](https://github.com/matrix-org/matrix-rust-sdk/pull/5794))
 

crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs

@@ -14,14 +14,17 @@
 
 //! Trait and macro of integration tests for `EventCacheStore` implementations.
 
-use std::{collections::BTreeMap, sync::Arc};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    sync::Arc,
+};
 
 use assert_matches::assert_matches;
 use assert_matches2::assert_let;
 use matrix_sdk_common::{
     deserialized_responses::{
         AlgorithmInfo, DecryptedRoomEvent, EncryptionInfo, TimelineEvent, TimelineEventKind,
-        VerificationState,
+        UnableToDecryptInfo, UnableToDecryptReason, VerificationState,
     },
     linked_chunk::{
         ChunkContent, ChunkIdentifier as CId, LinkedChunkId, Position, Update, lazy_loader,
@@ -49,6 +52,24 @@ pub fn make_test_event(room_id: &RoomId, content: &str) -> TimelineEvent {
     make_test_event_with_event_id(room_id, content, None)
 }
 
+/// Create a `m.room.encrypted` test event with all data filled, for testing
+/// that linked chunk correctly stores event data for encrypted events.
+pub fn make_encrypted_test_event(room_id: &RoomId, session_id: &str) -> TimelineEvent {
+    let device_id = "DEVICEID";
+    let builder = EventFactory::new()
+        .encrypted("", "curve_key", device_id, session_id)
+        .room(room_id)
+        .sender(*ALICE);
+
+    let event = builder.into_raw();
+    let utd_info = UnableToDecryptInfo {
+        session_id: Some(session_id.to_owned()),
+        reason: UnableToDecryptReason::MissingMegolmSession { withheld_code: None },
+    };
+
+    TimelineEvent::from_utd(event, utd_info)
+}
+
 /// Same as [`make_test_event`], with an extra event id.
 pub fn make_test_event_with_event_id(
     room_id: &RoomId,
@@ -144,6 +165,9 @@ pub trait EventCacheStoreIntegrationTests {
     /// 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 saving an event works as expected.
     async fn test_save_event(&self);
 
@@ -962,7 +986,10 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
         .unwrap();
 
         // Now let's find the events.
-        let events = self.get_room_events(room_id).await.expect("failed to query for room events");
+        let events = self
+            .get_room_events(room_id, None, None)
+            .await
+            .expect("failed to query for room events");
 
         assert_eq!(events.len(), 2);
 
@@ -977,6 +1004,76 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
         }
     }
 
+    async fn test_get_room_events_filtered(&self) {
+        macro_rules! assert_expected_events {
+            ($events:expr, [$($item:expr),* $(,)?]) => {{
+                let got_ids: BTreeSet<_> = $events.into_iter().map(|ev| ev.event_id().unwrap()).collect();
+                let expected_ids = BTreeSet::from([$($item.event_id().unwrap()),*]);
+
+                assert_eq!(got_ids, expected_ids);
+            }};
+        }
+
+        let room_id = room_id!("!r0:matrix.org");
+        let linked_chunk_id = LinkedChunkId::Room(room_id);
+        let another_room_id = room_id!("!r1:matrix.org");
+        let another_linked_chunk_id = LinkedChunkId::Room(another_room_id);
+
+        let event = |session_id: &str| make_encrypted_test_event(room_id, session_id);
+
+        let first_event = event("session_1");
+        let second_event = event("session_2");
+        let third_event = event("session_3");
+        let fourth_event = make_test_event(room_id, "It's a secret to everybody");
+
+        // Add one event in one room.
+        self.handle_linked_chunk_updates(
+            linked_chunk_id,
+            vec![
+                Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
+                Update::PushItems {
+                    at: Position::new(CId::new(0), 0),
+                    items: vec![first_event.clone(), second_event.clone(), fourth_event.clone()],
+                },
+            ],
+        )
+        .await
+        .unwrap();
+
+        // Add an event in a different room.
+        self.handle_linked_chunk_updates(
+            another_linked_chunk_id,
+            vec![
+                Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
+                Update::PushItems {
+                    at: Position::new(CId::new(0), 0),
+                    items: vec![third_event.clone()],
+                },
+            ],
+        )
+        .await
+        .unwrap();
+
+        // Now let's find all the encrypted events of the first room.
+        let events = self
+            .get_room_events(room_id, Some("m.room.encrypted"), None)
+            .await
+            .expect("failed to query for room events");
+
+        assert_eq!(events.len(), 2);
+        assert_expected_events!(events, [first_event, second_event]);
+
+        // Now let's find all the encrypted events which were encrypted using the first
+        // session ID.
+        let events = self
+            .get_room_events(room_id, Some("m.room.encrypted"), Some("session_1"))
+            .await
+            .expect("failed to query for room events");
+
+        assert_eq!(events.len(), 1);
+        assert_expected_events!(events, [first_event]);
+    }
+
     async fn test_save_event(&self) {
         let room_id = room_id!("!r0:matrix.org");
         let another_room_id = room_id!("!r1:matrix.org");
@@ -1264,6 +1361,13 @@ macro_rules! event_cache_store_integration_tests {
                 event_cache_store.test_get_room_events().await;
             }
 
+            #[async_test]
+            async fn test_get_room_events_filtered() {
+                let event_cache_store =
+                    get_event_cache_store().await.unwrap().into_event_cache_store();
+                event_cache_store.test_get_room_events_filtered().await;
+            }
+
             #[async_test]
             async fn test_save_event() {
                 let event_cache_store =

crates/matrix-sdk-base/src/event_cache/store/memory_store.rs

@@ -217,11 +217,24 @@ impl EventCacheStore for MemoryStore {
         Ok(related_events)
     }
 
-    async fn get_room_events(&self, room_id: &RoomId) -> Result<Vec<Event>, Self::Error> {
+    async fn get_room_events(
+        &self,
+        room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
+    ) -> Result<Vec<Event>, Self::Error> {
         let inner = self.inner.read().unwrap();
 
-        let event: Vec<_> =
-            inner.events.items(room_id).map(|(event, _pos)| event.clone()).collect();
+        let event: Vec<_> = inner
+            .events
+            .items(room_id)
+            .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();
 
         Ok(event)
     }

crates/matrix-sdk-base/src/event_cache/store/traits.rs

@@ -158,7 +158,12 @@ pub trait EventCacheStore: AsyncTraitDeps {
     ///
     /// This method must return events saved either in any linked chunks, *or*
     /// events saved "out-of-band" with the [`Self::save_event`] method.
-    async fn get_room_events(&self, room_id: &RoomId) -> Result<Vec<Event>, Self::Error>;
+    async fn get_room_events(
+        &self,
+        room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
+    ) -> Result<Vec<Event>, Self::Error>;
 
     /// Save an event, that might or might not be part of an existing linked
     /// chunk.
@@ -264,8 +269,13 @@ impl<T: EventCacheStore> EventCacheStore for EraseEventCacheStoreError<T> {
         self.0.find_event_relations(room_id, event_id, filter).await.map_err(Into::into)
     }
 
-    async fn get_room_events(&self, room_id: &RoomId) -> Result<Vec<Event>, Self::Error> {
-        self.0.get_room_events(room_id).await.map_err(Into::into)
+    async fn get_room_events(
+        &self,
+        room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
+    ) -> Result<Vec<Event>, Self::Error> {
+        self.0.get_room_events(room_id, event_type, session_id).await.map_err(Into::into)
     }
 
     async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {

crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs

@@ -479,14 +479,29 @@ impl EventCacheStore for IndexeddbEventCacheStore {
     async fn get_room_events(
         &self,
         room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
     ) -> Result<Vec<Event>, IndexeddbEventCacheStoreError> {
         let _timer = timer!("method");
 
+        // TODO: Make this more efficient so we don't load all events and filter them
+        // here. We should instead only load the relevant events.
+
         let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
         transaction
             .get_room_events(room_id)
             .await
-            .map(|vec| vec.into_iter().map(Into::into).collect())
+            .map(|vec| {
+                vec.into_iter()
+                    .map(Event::from)
+                    .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_err(Into::into)
     }
 

crates/matrix-sdk-sqlite/src/event_cache_store.rs

@@ -34,7 +34,9 @@ use matrix_sdk_store_encryption::StoreCipher;
 use ruma::{
     events::relation::RelationType, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId,
 };
-use rusqlite::{params_from_iter, OptionalExtension, ToSql, Transaction, TransactionBehavior};
+use rusqlite::{
+    params, params_from_iter, OptionalExtension, ToSql, Transaction, TransactionBehavior,
+};
 use tokio::{
     fs,
     sync::{Mutex, OwnedMutexGuard},
@@ -1329,19 +1331,47 @@ impl EventCacheStore for SqliteEventCacheStore {
     }
 
     #[instrument(skip(self))]
-    async fn get_room_events(&self, room_id: &RoomId) -> Result<Vec<Event>, Self::Error> {
+    async fn get_room_events(
+        &self,
+        room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
+    ) -> Result<Vec<Event>, Self::Error> {
         let _timer = timer!("method");
 
         let this = self.clone();
 
         let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, room_id);
+        let hashed_event_type = event_type.map(|e| self.encode_key(keys::EVENTS, e));
+        let hashed_session_id = session_id.map(|s| self.encode_key(keys::EVENTS, s));
 
         self.read()
             .await?
             .with_transaction(move |txn| -> Result<_> {
-                let mut statement = txn.prepare("SELECT content FROM events WHERE room_id = ?")?;
-                let maybe_events =
-                    statement.query_map((hashed_room_id,), |row| row.get::<_, Vec<u8>>(0))?;
+                // I'm not sure why clippy claims that the clones aren't required. The compiler
+                // tells us that the lifetimes aren't long enough if we remove them. Doesn't matter
+                // much so let's silence things.
+                #[allow(clippy::redundant_clone)]
+                let (query, keys) = match (hashed_event_type, hashed_session_id) {
+                    (None, None) => {
+                        ("SELECT content FROM events WHERE room_id = ?", params![hashed_room_id])
+                    }
+                    (None, Some(session_id)) => (
+                        "SELECT content FROM events WHERE room_id = ?1 AND session_id = ?2",
+                        params![hashed_room_id, session_id.to_owned()],
+                    ),
+                    (Some(event_type), None) => (
+                        "SELECT content FROM events WHERE room_id = ? AND event_type = ?",
+                        params![hashed_room_id, event_type.to_owned()]
+                    ),
+                    (Some(event_type), Some(session_id)) => (
+                        "SELECT content FROM events WHERE room_id = ?1 AND event_type = ?2 AND session_id = ?3",
+                        params![hashed_room_id, event_type.to_owned(), session_id.to_owned()],
+                    ),
+                };
+
+                let mut statement = txn.prepare(query)?;
+                let maybe_events = statement.query_map(keys, |row| row.get::<_, Vec<u8>>(0))?;
 
                 let mut events = Vec::new();
                 for ev in maybe_events {

labs/multiverse/src/main.rs

@@ -394,7 +394,7 @@ impl App {
 
             let redaction_rules = room.clone_info().room_version_rules_or_default().redaction;
 
-            let maybe_timeline_events = store.get_room_events(room_id).await;
+            let maybe_timeline_events = store.get_room_events(room_id, None, None).await;
             let Ok(timeline_events) = maybe_timeline_events else {
                 warn!("Failed to get room's events: {maybe_timeline_events:?}");
                 continue;