refactor(sdk): Move index handling to EventCache and use linked_chunk_update_sender.

Remove previous code that got updates in `RoomEventCacheState::post_process_new_events`.
Add new task in `EventCache` that subscribes to `linked_chunk_update_sender` and forwards new events to `client::search::SearchIndex` same as before.

Signed-off-by: Shrey Patel [email protected]

Author: dragonfly1033

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

@@ -50,9 +50,9 @@ 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;
+#[cfg(feature = "experimental-search")]
+use ruma::events::AnySyncMessageLikeEvent;
 use ruma::{
     events::AnySyncEphemeralRoomEvent, serde::Raw, OwnedEventId, OwnedRoomId, OwnedTransactionId,
     RoomId,
@@ -133,11 +133,6 @@ pub enum EventCacheError {
         /// A string containing details about the error.
         details: String,
     },
-
-    /// An error occurred in the index.
-    #[cfg(feature = "experimental-search")]
-    #[error(transparent)]
-    IndexError(#[from] IndexError),
 }
 
 /// A result using the [`EventCacheError`].
@@ -199,6 +194,12 @@ impl EventCache {
             thread_subscriber_sender,
         )));
 
+        #[cfg(feature = "experimental-search")]
+        let search_indexing_task = AbortOnDrop::new(spawn(Self::search_indexing_task(
+            client.clone(),
+            linked_chunk_update_sender.clone(),
+        )));
+
         Self {
             inner: Arc::new(EventCacheInner {
                 client,
@@ -210,6 +211,8 @@ impl EventCache {
                 generic_update_sender,
                 linked_chunk_update_sender,
                 _thread_subscriber_task: thread_subscriber_task,
+                #[cfg(feature = "experimental-search")]
+                _search_indexing_task: search_indexing_task,
                 thread_subscriber_receiver,
             }),
         }
@@ -689,6 +692,82 @@ impl EventCache {
             }
         }
     }
+
+    /// Takes a [`TimelineEvent`] and passes it to the [`RoomIndex`] of the
+    /// given room which will add/remove/edit an event in the index based on
+    /// the event type.
+    #[cfg(feature = "experimental-search")]
+    #[instrument(skip_all)]
+    async fn search_indexing_task(
+        client: WeakClient,
+        linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
+    ) {
+        let mut linked_chunk_update_receiver = linked_chunk_update_sender.subscribe();
+
+        loop {
+            match linked_chunk_update_receiver.recv().await {
+                Ok(room_ec_lc_update) => {
+                    let OwnedLinkedChunkId::Room(room_id) =
+                        room_ec_lc_update.linked_chunk_id.clone()
+                    else {
+                        trace!("Received non-room updates, ignoring.");
+                        continue;
+                    };
+
+                    let mut timeline_events = room_ec_lc_update.events().peekable();
+
+                    if timeline_events.peek().is_none() {
+                        continue;
+                    }
+
+                    let Some(client) = client.get() else {
+                        trace!("Client is shutting down, not spawning thread subscriber task");
+                        return;
+                    };
+
+                    let mut search_index_guard = client.search_index().lock().await;
+
+                    for event in timeline_events {
+                        if let Some(message_event) = parse_timeline_event_for_search_index(&event) {
+                            if let Err(err) =
+                                search_index_guard.handle_event(message_event, &room_id)
+                            {
+                                warn!("Failed to handle event for indexing: {err}")
+                            }
+                        }
+                    }
+                }
+                Err(RecvError::Closed) => {
+                    debug!("Linked chunk update channel has been closed, exiting thread subscriber task");
+                    break;
+                }
+                Err(RecvError::Lagged(num_skipped)) => {
+                    warn!(num_skipped, "Lagged behind linked chunk updates");
+                }
+            }
+        }
+    }
+}
+
+#[cfg(feature = "experimental-search")]
+fn parse_timeline_event_for_search_index(event: &TimelineEvent) -> Option<AnySyncMessageLikeEvent> {
+    use ruma::events::AnySyncTimelineEvent;
+
+    if event.kind.is_utd() {
+        return None;
+    }
+
+    match event.raw().deserialize() {
+        Ok(event) => match event {
+            AnySyncTimelineEvent::MessageLike(event) => Some(event),
+            AnySyncTimelineEvent::State(_) => None,
+        },
+
+        Err(e) => {
+            warn!("failed to parse event: {e:?}");
+            None
+        }
+    }
 }
 
 struct EventCacheInner {
@@ -744,6 +823,15 @@ struct EventCacheInner {
     /// [`EventCache`], so it does listen to *all* rooms at the same time.
     _thread_subscriber_task: AbortOnDrop<()>,
 
+    /// A background task listening to room updates, and
+    /// automatically handling search index operations add/remove/edit
+    /// depending on the event type.
+    ///
+    /// One important constraint is that there is only one such task per
+    /// [`EventCache`], so it does listen to *all* rooms at the same time.
+    #[cfg(feature = "experimental-search")]
+    _search_indexing_task: AbortOnDrop<()>,
+
     /// A test helper receiver that will be emitted every time the thread
     /// subscriber task subscribed to a new thread.
     ///

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

@@ -260,34 +260,32 @@ impl RoomPagination {
         batch_size: u16,
         prev_token: Option<String>,
     ) -> Result<Option<BackPaginationOutcome>> {
-        let Some(room) = self.inner.weak_room.get() else {
-            // The client is shutting down, return an empty default response.
-            return Ok(Some(BackPaginationOutcome {
-                reached_start: false,
-                events: Default::default(),
-            }));
+        let (events, new_token) = {
+            let Some(room) = self.inner.weak_room.get() else {
+                // The client is shutting down, return an empty default response.
+                return Ok(Some(BackPaginationOutcome {
+                    reached_start: false,
+                    events: Default::default(),
+                }));
+            };
+
+            let mut options = MessagesOptions::new(Direction::Backward).from(prev_token.as_deref());
+            options.limit = batch_size.into();
+
+            let response = room
+                .messages(options)
+                .await
+                .map_err(|err| EventCacheError::BackpaginationError(Box::new(err)))?;
+
+            (response.chunk, response.end)
         };
 
-        let mut options = MessagesOptions::new(Direction::Backward).from(prev_token.as_deref());
-        options.limit = batch_size.into();
-
-        let response = room
-            .messages(options)
-            .await
-            .map_err(|err| EventCacheError::BackpaginationError(Box::new(err)))?;
-
         if let Some((outcome, timeline_event_diffs)) = self
             .inner
             .state
             .write()
             .await
-            .handle_backpagination(
-                response.chunk,
-                response.end,
-                prev_token,
-                #[cfg(feature = "experimental-search")]
-                &room,
-            )
+            .handle_backpagination(events, new_token, prev_token)
             .await?
         {
             if !timeline_event_diffs.is_empty() {

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

@@ -542,22 +542,8 @@ impl RoomEventCacheInner {
         // Add all the events to the backend.
         trace!("adding new events");
 
-        #[cfg(feature = "experimental-search")]
-        let Some(room) = self.weak_room.get() else {
-            trace!("Couldn't get room while handling timeline");
-            return Ok(());
-        };
-
-        let (stored_prev_batch_token, timeline_event_diffs) = self
-            .state
-            .write()
-            .await
-            .handle_sync(
-                timeline,
-                #[cfg(feature = "experimental-search")]
-                &room,
-            )
-            .await?;
+        let (stored_prev_batch_token, timeline_event_diffs) =
+            self.state.write().await.handle_sync(timeline).await?;
 
         // Now that all events have been added, we can trigger the
         // `pagination_token_notifier`.
@@ -622,8 +608,6 @@ mod private {
 
     use eyeball::SharedObservable;
     use eyeball_im::VectorDiff;
-    #[cfg(feature = "experimental-search")]
-    use matrix_sdk_base::deserialized_responses::TimelineEvent;
     use matrix_sdk_base::{
         apply_redaction,
         deserialized_responses::{ThreadSummary, ThreadSummaryStatus, TimelineEventKind},
@@ -662,8 +646,6 @@ mod private {
         BackPaginationOutcome, RoomEventCacheLinkedChunkUpdate, RoomPaginationStatus,
         ThreadEventCacheUpdate,
     };
-    #[cfg(feature = "experimental-search")]
-    use crate::Room;
 
     /// State for a single room's event cache.
     ///
@@ -1460,41 +1442,6 @@ mod private {
             Ok(Some((target, related)))
         }
 
-        #[cfg(feature = "experimental-search")]
-        fn parse_timeline_event(&self, event: &TimelineEvent) -> Option<AnySyncMessageLikeEvent> {
-            if event.kind.is_utd() {
-                return None;
-            }
-
-            match event.raw().deserialize() {
-                Ok(event) => match event {
-                    AnySyncTimelineEvent::MessageLike(event) => Some(event),
-                    AnySyncTimelineEvent::State(_) => None,
-                },
-
-                Err(e) => {
-                    warn!("failed to parse event: {e:?}");
-                    None
-                }
-            }
-        }
-
-        /// Takes a [`TimelineEvent`] and passes it to the [`RoomIndex`] of the
-        /// given room which will add/remove/edit an event in the index based on
-        /// the event type.
-        #[cfg(feature = "experimental-search")]
-        async fn index_event(
-            &self,
-            event: &TimelineEvent,
-            room: &Room,
-        ) -> Result<(), EventCacheError> {
-            if let Some(message_event) = self.parse_timeline_event(event) {
-                room.index_event(message_event).await.map_err(EventCacheError::from)
-            } else {
-                Ok(())
-            }
-        }
-
         /// Post-process new events, after they have been added to the in-memory
         /// linked chunk.
         ///
@@ -1503,21 +1450,14 @@ mod private {
             &mut self,
             events: Vec<Event>,
             is_sync: bool,
-            #[cfg(feature = "experimental-search")] room: &Room,
         ) -> Result<(), EventCacheError> {
             // Update the store before doing the post-processing.
             self.propagate_changes().await?;
 
             let mut new_events_by_thread: BTreeMap<_, Vec<_>> = BTreeMap::new();
 
             for event in events {
-                self.maybe_apply_new_redaction(&event).await?; // TODO: Handle redaction for search index
-
-                // We can also add the event to the index.
-                #[cfg(feature = "experimental-search")]
-                if let Err(err) = self.index_event(&event, room).await {
-                    warn!("error while trying to index event: {err:?}");
-                }
+                self.maybe_apply_new_redaction(&event).await?;
 
                 if let Some(thread_root) = extract_thread_root(event.raw()) {
                     new_events_by_thread.entry(thread_root).or_default().push(event.clone());
@@ -1760,7 +1700,6 @@ mod private {
         pub async fn handle_sync(
             &mut self,
             mut timeline: Timeline,
-            #[cfg(feature = "experimental-search")] room: &Room,
         ) -> Result<(bool, Vec<VectorDiff<Event>>), EventCacheError> {
             let mut prev_batch = timeline.prev_batch.take();
 
@@ -1853,13 +1792,7 @@ mod private {
             self.room_linked_chunk
                 .push_live_events(prev_batch.map(|prev_token| Gap { prev_token }), &events);
 
-            self.post_process_new_events(
-                events,
-                true,
-                #[cfg(feature = "experimental-search")]
-                room,
-            )
-            .await?;
+            self.post_process_new_events(events, true).await?;
 
             if timeline.limited && has_new_gap {
                 // If there was a previous batch token for a limited timeline, unload the chunks
@@ -1889,7 +1822,6 @@ mod private {
             events: Vec<Event>,
             mut new_token: Option<String>,
             prev_token: Option<String>,
-            #[cfg(feature = "experimental-search")] room: &Room,
         ) -> Result<Option<(BackPaginationOutcome, Vec<VectorDiff<Event>>)>, EventCacheError>
         {
             // Check that the previous token still exists; otherwise it's a sign that the
@@ -1964,13 +1896,7 @@ mod private {
             );
 
             // Note: this flushes updates to the store.
-            self.post_process_new_events(
-                topo_ordered_events,
-                false,
-                #[cfg(feature = "experimental-search")]
-                room,
-            )
-            .await?;
+            self.post_process_new_events(topo_ordered_events, false).await?;
 
             let event_diffs = self.room_linked_chunk.updates_as_vector_diffs();
 

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

@@ -55,8 +55,6 @@ use matrix_sdk_common::{
     timeout::timeout,
 };
 #[cfg(feature = "experimental-search")]
-use matrix_sdk_search::error::IndexError;
-#[cfg(feature = "experimental-search")]
 #[cfg(doc)]
 use matrix_sdk_search::index::RoomIndex;
 use mime::Mime;
@@ -3752,18 +3750,6 @@ impl Room {
         opts.send(self, event_id).await
     }
 
-    /// Handle an [`AnySyncMessageLikeEvent`] in this room's [`RoomIndex`].
-    ///
-    /// This which will add/remove/edit an event in the index based on the
-    /// event type.
-    #[cfg(feature = "experimental-search")]
-    pub(crate) async fn index_event(
-        &self,
-        event: AnySyncMessageLikeEvent,
-    ) -> Result<(), IndexError> {
-        self.client.search_index().lock().await.handle_event(event, self.room_id())
-    }
-
     /// Search this room's [`RoomIndex`] for query and return at most
     /// max_number_of_results results.
     #[cfg(feature = "experimental-search")]