feat(sdk): Add creation of indexes and indexing of messages.

Author: dragonfly1033

Cargo.lock

@@ -193,7 +193,7 @@ mod tests {
     use std::{collections::HashSet, error::Error};
 
     use matrix_sdk_test::event_factory::EventFactory;
-    use ruma::{event_id, owned_event_id, room_id, user_id};
+    use ruma::{event_id, room_id, user_id};
 
     use crate::index::RoomIndex;
 
@@ -227,10 +227,14 @@ mod tests {
         let mut index =
             RoomIndex::new_in_ram(room_id).expect("failed to make index in ram: {index:?}");
 
+        let event_id_1 = event_id!("$event_id_1:localhost");
+        let event_id_2 = event_id!("$event_id_2:localhost");
+        let event_id_3 = event_id!("$event_id_3:localhost");
+
         index.add_event(
             EventFactory::new()
                 .text_msg("This is a sentence")
-                .event_id(event_id!("$event_id_1:localhost"))
+                .event_id(event_id_1)
                 .room(room_id)
                 .sender(user_id!("@user_id:localhost"))
                 .into_any_message_like_event(),
@@ -239,7 +243,7 @@ mod tests {
         index.add_event(
             EventFactory::new()
                 .text_msg("All new words")
-                .event_id(event_id!("$event_id_2:localhost"))
+                .event_id(event_id_2)
                 .room(room_id)
                 .sender(user_id!("@user_id:localhost"))
                 .into_any_message_like_event(),
@@ -248,7 +252,7 @@ mod tests {
         index.add_event(
             EventFactory::new()
                 .text_msg("A similar sentence")
-                .event_id(event_id!("$event_id_3:localhost"))
+                .event_id(event_id_3)
                 .room(room_id)
                 .sender(user_id!("@user_id:localhost"))
                 .into_any_message_like_event(),
@@ -259,8 +263,7 @@ mod tests {
         let result = index.search("sentence", 10).expect("search failed with: {result:?}");
         let result: HashSet<_> = result.iter().collect();
 
-        let true_value =
-            [owned_event_id!("$event_id_1:localhost"), owned_event_id!("$event_id_3:localhost")];
+        let true_value = [event_id_1.to_owned(), event_id_3.to_owned()];
         let true_value: HashSet<_> = true_value.iter().collect();
 
         assert_eq!(result, true_value, "search result not correct: {result:?}");

crates/matrix-sdk-search/src/index.rs

@@ -62,6 +62,8 @@ docsrs = ["e2e-encryption", "sqlite", "indexeddb", "sso-login", "qrcode"]
 # Add support for inline media galleries via msgtypes
 unstable-msc4274 = ["ruma/unstable-msc4274", "matrix-sdk-base/unstable-msc4274"]
 
+experimental-search = ["matrix-sdk-search"]
+
 [dependencies]
 anyhow = { workspace = true, optional = true }
 anymap2 = "0.13.0"
@@ -92,6 +94,7 @@ matrix-sdk-ffi-macros = { workspace = true, optional = true }
 matrix-sdk-indexeddb = { workspace = true, optional = true }
 matrix-sdk-sqlite = { workspace = true, optional = true }
 matrix-sdk-test = { workspace = true, optional = true }
+matrix-sdk-search = { workspace = true, optional = true }
 mime.workspace = true
 mime2ext = "0.1.53"
 oauth2.workspace = true

crates/matrix-sdk/Cargo.toml

@@ -15,16 +15,24 @@
 
 mod homeserver_config;
 
+#[cfg(feature = "experimental-search")]
+use std::collections::HashMap;
 #[cfg(feature = "sqlite")]
 use std::path::Path;
+#[cfg(any(feature = "experimental-search", feature = "sqlite"))]
+use std::path::PathBuf;
 use std::{collections::BTreeSet, fmt, sync::Arc};
 
 use homeserver_config::*;
 #[cfg(feature = "e2e-encryption")]
 use matrix_sdk_base::crypto::DecryptionSettings;
 use matrix_sdk_base::{store::StoreConfig, BaseClient, ThreadingSupport};
+#[cfg(feature = "experimental-search")]
+use matrix_sdk_search::index::RoomIndex;
 #[cfg(feature = "sqlite")]
 use matrix_sdk_sqlite::SqliteStoreConfig;
+#[cfg(feature = "experimental-search")]
+use ruma::OwnedRoomId;
 use ruma::{
     api::{error::FromHttpResponseError, MatrixVersion, SupportedVersions},
     OwnedServerName, ServerName,
@@ -114,6 +122,8 @@ pub struct ClientBuilder {
     enable_share_history_on_invite: bool,
     cross_process_store_locks_holder_name: String,
     threading_support: ThreadingSupport,
+    #[cfg(feature = "experimental-search")]
+    index_base_dir: IndexBaseDir,
 }
 
 impl ClientBuilder {
@@ -145,6 +155,8 @@ impl ClientBuilder {
             cross_process_store_locks_holder_name:
                 Self::DEFAULT_CROSS_PROCESS_STORE_LOCKS_HOLDER_NAME.to_owned(),
             threading_support: ThreadingSupport::Disabled,
+            #[cfg(feature = "experimental-search")]
+            index_base_dir: IndexBaseDir::Ram,
         }
     }
 
@@ -489,6 +501,13 @@ impl ClientBuilder {
         self
     }
 
+    /// The base directory in which each room's index directory will be stored.
+    #[cfg(feature = "experimental-search")]
+    pub fn index_base_directory(mut self, path: IndexBaseDir) -> Self {
+        self.index_base_dir = path;
+        self
+    }
+
     /// Create a [`Client`] with the options set on this builder.
     ///
     /// # Errors
@@ -590,6 +609,10 @@ impl ClientBuilder {
         let event_cache = OnceCell::new();
         let latest_events = OnceCell::new();
 
+        #[cfg(feature = "experimental-search")]
+        let room_indexes: Arc<Mutex<HashMap<OwnedRoomId, RoomIndex>>> =
+            Arc::new(Mutex::new(HashMap::new()));
+
         let inner = ClientInner::new(
             auth_ctx,
             server,
@@ -607,6 +630,10 @@ impl ClientBuilder {
             #[cfg(feature = "e2e-encryption")]
             self.enable_share_history_on_invite,
             self.cross_process_store_locks_holder_name,
+            #[cfg(feature = "experimental-search")]
+            room_indexes,
+            #[cfg(feature = "experimental-search")]
+            self.index_base_dir,
         )
         .await;
 
@@ -717,6 +744,14 @@ async fn build_indexeddb_store_config(
     panic!("the IndexedDB is only available on the 'wasm32' arch")
 }
 
+#[cfg(feature = "experimental-search")]
+#[allow(dead_code)]
+#[derive(Clone, Debug)]
+pub enum IndexBaseDir {
+    Directory(PathBuf),
+    Ram,
+}
+
 #[derive(Clone, Debug)]
 enum HttpConfig {
     #[cfg(not(target_family = "wasm"))]
@@ -755,7 +790,7 @@ enum BuilderStoreConfig {
     #[cfg(feature = "sqlite")]
     Sqlite {
         config: SqliteStoreConfig,
-        cache_path: Option<std::path::PathBuf>,
+        cache_path: Option<PathBuf>,
     },
     #[cfg(feature = "indexeddb")]
     IndexedDb {

crates/matrix-sdk/src/client/builder/mod.rs

@@ -14,6 +14,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(feature = "experimental-search")]
+use std::collections::HashMap;
 use std::{
     collections::{btree_map, BTreeMap, BTreeSet},
     fmt::{self, Debug},
@@ -38,6 +40,12 @@ use matrix_sdk_base::{
     StateStoreDataKey, StateStoreDataValue, SyncOutsideWasm, ThreadingSupport,
 };
 use matrix_sdk_common::ttl_cache::TtlCache;
+#[cfg(feature = "experimental-search")]
+use matrix_sdk_search::error::IndexError;
+#[cfg(feature = "experimental-search")]
+use matrix_sdk_search::index::RoomIndex;
+#[cfg(feature = "experimental-search")]
+use ruma::events::AnyMessageLikeEvent;
 #[cfg(feature = "e2e-encryption")]
 use ruma::events::{room::encryption::RoomEncryptionEventContent, InitialStateEvent};
 use ruma::{
@@ -72,11 +80,15 @@ use ruma::{
     RoomAliasId, RoomId, RoomOrAliasId, ServerName, UInt, UserId,
 };
 use serde::de::DeserializeOwned;
+#[cfg(feature = "experimental-search")]
+use tokio::sync::MutexGuard;
 use tokio::sync::{broadcast, Mutex, OnceCell, RwLock, RwLockReadGuard};
 use tracing::{debug, error, instrument, trace, warn, Instrument, Span};
 use url::Url;
 
 use self::futures::SendRequest;
+#[cfg(feature = "experimental-search")]
+use crate::client::builder::IndexBaseDir;
 use crate::{
     authentication::{
         matrix::MatrixAuth, oauth::OAuth, AuthCtx, AuthData, ReloadSessionCallback,
@@ -350,6 +362,14 @@ pub(crate) struct ClientInner {
     ///
     /// [`LatestEvent`]: crate::latest_event::LatestEvent
     latest_events: OnceCell<LatestEvents>,
+
+    /// HashMap that links each joined room to its RoomIndex
+    #[cfg(feature = "experimental-search")]
+    room_indexes: Arc<Mutex<HashMap<OwnedRoomId, RoomIndex>>>,
+
+    /// Base directory that stores the directories for each RoomIndex
+    #[cfg(feature = "experimental-search")]
+    index_base_dir: IndexBaseDir,
 }
 
 impl ClientInner {
@@ -374,6 +394,10 @@ impl ClientInner {
         #[cfg(feature = "e2e-encryption")] encryption_settings: EncryptionSettings,
         #[cfg(feature = "e2e-encryption")] enable_share_history_on_invite: bool,
         cross_process_store_locks_holder_name: String,
+        #[cfg(feature = "experimental-search")] room_indexes: Arc<
+            Mutex<HashMap<OwnedRoomId, RoomIndex>>,
+        >,
+        #[cfg(feature = "experimental-search")] index_base_dir: IndexBaseDir,
     ) -> Arc<Self> {
         let caches = ClientCaches {
             server_info: server_info.into(),
@@ -409,6 +433,10 @@ impl ClientInner {
             #[cfg(feature = "e2e-encryption")]
             enable_share_history_on_invite,
             server_max_upload_size: Mutex::new(OnceCell::new()),
+            #[cfg(feature = "experimental-search")]
+            room_indexes,
+            #[cfg(feature = "experimental-search")]
+            index_base_dir,
         };
 
         #[allow(clippy::let_and_return)]
@@ -2713,6 +2741,10 @@ impl Client {
                 #[cfg(feature = "e2e-encryption")]
                 self.inner.enable_share_history_on_invite,
                 cross_process_store_locks_holder_name,
+                #[cfg(feature = "experimental-search")]
+                self.inner.room_indexes.clone(),
+                #[cfg(feature = "experimental-search")]
+                self.inner.index_base_dir.clone(),
             )
             .await,
         };
@@ -2810,6 +2842,85 @@ impl Client {
         &self.base_client().decryption_settings
     }
 
+    /// Add [`AnyMessageLikeEvent`] to [`RoomIndex`] of given [`RoomId`]
+    #[cfg(feature = "experimental-search")]
+    pub async fn index_event(
+        &self,
+        event: AnyMessageLikeEvent,
+        room_id: &RoomId,
+    ) -> Result<(), IndexError> {
+        let mut hash_map = self.inner.room_indexes.lock().await;
+
+        let result = if let Some(index) = hash_map.get_mut(room_id) {
+            index.add_event(event)
+        } else {
+            self.add_index_impl(room_id, &mut hash_map)?;
+            let index = hash_map.get_mut(room_id).expect("key just added");
+            index.add_event(event)
+        };
+
+        match result {
+            Ok(_) => {}
+            Err(IndexError::CannotIndexRedactedMessage)
+            | Err(IndexError::EmptyMessage)
+            | Err(IndexError::MessageTypeNotSupported) => {
+                debug!("failed to parse event for indexing: {result:?}")
+            }
+            Err(IndexError::TantivyError(err)) => {
+                error!("failed to add/commit event to index: {err:?}")
+            }
+            Err(_) => error!("unexpected error during indexing: {result:?}"),
+        };
+        Ok(())
+    }
+
+    /// Add [`RoomIndex`] for given [`RoomId`] to room_indexes
+    #[cfg(feature = "experimental-search")]
+    pub async fn add_index(&self, room_id: &RoomId) -> Result<(), IndexError> {
+        let mut hash_map = self.inner.room_indexes.lock().await;
+        self.add_index_impl(room_id, &mut hash_map)?;
+        Ok(())
+    }
+
+    #[cfg(feature = "experimental-search")]
+    fn add_index_impl(
+        &self,
+        room_id: &RoomId,
+        hash_map: &mut MutexGuard<'_, HashMap<OwnedRoomId, RoomIndex>>,
+    ) -> Result<(), IndexError> {
+        if !hash_map.contains_key(room_id) {
+            let index = match &self.inner.index_base_dir {
+                IndexBaseDir::Directory(path) => RoomIndex::open_or_create(path, room_id)?,
+                IndexBaseDir::Ram => RoomIndex::new_in_ram(room_id)?,
+            };
+            hash_map.insert(room_id.to_owned(), index);
+        }
+        Ok(())
+    }
+
+    /// Search a [`Room`]'s index for the query and return at most
+    /// max_number_of_results results.
+    #[cfg(feature = "experimental-search")]
+    pub async fn search_index(
+        &self,
+        query: &str,
+        max_number_of_results: usize,
+        room_id: &RoomId,
+    ) -> Option<Vec<OwnedEventId>> {
+        let hash_map = self.inner.room_indexes.lock().await;
+        if let Some(index) = hash_map.get(room_id) {
+            index
+                .search(query, max_number_of_results)
+                .inspect_err(|err| {
+                    error!("error occured while searching index: {err:?}");
+                })
+                .ok()
+        } else {
+            warn!("Tried to search in a room with no index");
+            None
+        }
+    }
+
     /// Whether the client is configured to take thread subscriptions (MSC4306
     /// and MSC4308) into account.
     ///

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

@@ -45,6 +45,8 @@ use matrix_sdk_base::{
     timer,
 };
 use matrix_sdk_common::executor::{spawn, JoinHandle};
+#[cfg(feature = "experimental-search")]
+use matrix_sdk_search::error::IndexError;
 use room::RoomEventCacheState;
 use ruma::{events::AnySyncEphemeralRoomEvent, serde::Raw, OwnedEventId, OwnedRoomId, RoomId};
 use tokio::sync::{
@@ -116,6 +118,11 @@ pub enum EventCacheError {
         /// A string containing details about the error.
         details: String,
     },
+
+    /// An error occurred higher up in the SDK.
+    #[cfg(feature = "experimental-search")]
+    #[error(transparent)]
+    IndexError(#[from] IndexError),
 }
 
 /// A result using the [`EventCacheError`].