Search messages in a room

src/home/room_screen.rs

@@ -25,8 +25,8 @@ use crate::{
         user_profile::{AvatarState, ShowUserProfileAction, UserProfile, UserProfileAndRoomId, UserProfilePaneInfo, UserProfileSlidingPaneRef, UserProfileSlidingPaneWidgetExt},
         user_profile_cache,
     }, shared::{
-        avatar::AvatarWidgetRefExt, callout_tooltip::TooltipAction, html_or_plaintext::{HtmlOrPlaintextRef, HtmlOrPlaintextWidgetRefExt}, jump_to_bottom_button::{JumpToBottomButtonWidgetExt, UnreadMessageCount}, popup_list::enqueue_popup_notification, text_or_image::{TextOrImageRef, TextOrImageWidgetRefExt}, typing_animation::TypingAnimationWidgetExt
-    }, sliding_sync::{self, get_client, submit_async_request, take_timeline_endpoints, BackwardsPaginateUntilEventRequest, MatrixRequest, PaginationDirection, TimelineRequestSender, UserPowerLevels}, utils::{self, unix_time_millis_to_datetime, ImageFormat, MediaFormatConst, MEDIA_THUMBNAIL_FORMAT}
+        avatar::AvatarWidgetRefExt, callout_tooltip::TooltipAction, html_or_plaintext::{HtmlOrPlaintextRef, HtmlOrPlaintextWidgetRefExt}, jump_to_bottom_button::{JumpToBottomButtonWidgetExt, UnreadMessageCount}, popup_list::enqueue_popup_notification, text_or_image::{TextOrImageRef, TextOrImageWidgetRefExt}, typing_animation::TypingAnimationWidgetExt, search::SearchUpdate
+    }, sliding_sync::{self, get_client, submit_async_request, take_timeline_endpoints, BackwardsPaginateUntilEventRequest, MatrixRequest, PaginationDirection, TimelineRequestSender, UserPowerLevels, SEARCH_RESULTS_RECEIVER}, utils::{self, unix_time_millis_to_datetime, ImageFormat, MediaFormatConst, MEDIA_THUMBNAIL_FORMAT}
 };
 use crate::home::event_reaction_list::ReactionListWidgetRefExt;
 use crate::home::room_read_receipt::AvatarRowWidgetRefExt;
@@ -936,7 +936,12 @@ pub struct RoomScreen {
     #[rust] room_name: String,
     /// The persistent UI-relevant states for the room that this widget is currently displaying.
     #[rust] tl_state: Option<TimelineUiState>,
+    /// The current search state for the room.
+    #[rust] search_state: Option<SearchUiState>,
+    /// The last search term used in the room.
+    #[rust] last_search_term: String,
 }
+
 impl Drop for RoomScreen {
     fn drop(&mut self) {
         // This ensures that the `TimelineUiState` instance owned by this room is *always* returned
@@ -948,6 +953,66 @@ impl Drop for RoomScreen {
     }
 }
 
+impl RoomScreen {
+
+    fn start_message_search(&mut self, room_id: OwnedRoomId, search_term: String) {
+        let (update_sender, update_receiver) = crossbeam_channel::unbounded();
+        let search_state = SearchUiState {
+            room_id: room_id.clone(),
+            fully_paginated: false,
+            items: Vector::new(),
+            next_batch: None,
+            update_receiver,
+        };
+        self.search_state = Some(search_state);
+
+        submit_async_request(MatrixRequest::SearchRoomMessages {
+            room_id,
+            search_term,
+            next_batch: None,
+        });
+    }
+
+    fn process_search_updates(&mut self, cx: &mut Cx) {
+        let Some(search_state) = self.search_state.as_mut() else { return };
+
+        let Some(receiver) = SEARCH_RESULTS_RECEIVER.get() else { return };
+
+        while let Ok(update) = receiver.try_recv() {
+            match update {
+                SearchUpdate::NewResults { room_id, results, next_batch } => {
+                    if search_state.room_id == room_id {
+                        search_state.items.extend(results);
+                        search_state.next_batch = next_batch.clone();
+
+                        // Stop pagination if no more results
+                        if next_batch.is_none() {
+                            search_state.fully_paginated = true;
+                        }
+                    }
+                }
+            }
+        }
+        self.redraw(cx);
+    }    
+
+
+    fn paginate_search_results(&mut self) {
+        let Some(search_state) = self.search_state.as_mut() else { return };
+        if search_state.fully_paginated {
+            return;
+        }
+
+        if let Some(next_batch) = search_state.next_batch.clone() {
+            submit_async_request(MatrixRequest::SearchRoomMessages {
+                room_id: search_state.room_id.clone(),
+                search_term: self.last_search_term.clone(),
+                next_batch: Some(next_batch),
+            });
+        }
+    }
+}
+
 impl Widget for RoomScreen {
     // Handle events and actions for the RoomScreen widget and its inner Timeline view.
     fn handle_event(&mut self, cx: &mut Cx, event: &Event, scope: &mut Scope) {
@@ -2727,19 +2792,27 @@ struct TimelineUiState {
     /// This index is saved before the timeline undergoes any jumps, e.g.,
     /// receiving new items, major scroll changes, or other timeline view jumps.
     prev_first_index: Option<usize>,
+
+/// Whether the user has scrolled past their latest read marker.
+///
+/// This is used to determine whether we should send a fully-read receipt
+/// after the user scrolls past their "read marker", i.e., their latest fully-read receipt.
+/// Its value is determined by comparing the fully-read event's timestamp with the
+/// first and last timestamp of displayed events in the timeline.
+/// When scrolling down, if the value is true, we send a fully-read receipt
+/// for the last visible event in the timeline.
+///
+/// When new message come in, this value is reset to `false`.
+scrolled_past_read_marker: bool,
+latest_own_user_receipt: Option<Receipt>,
+}
 
-    /// Whether the user has scrolled past their latest read marker.
-    ///
-    /// This is used to determine whether we should send a fully-read receipt
-    /// after the user scrolls past their "read marker", i.e., their latest fully-read receipt.
-    /// Its value is determined by comparing the fully-read event's timestamp with the
-    /// first and last timestamp of displayed events in the timeline.
-    /// When scrolling down, if the value is true, we send a fully-read receipt
-    /// for the last visible event in the timeline.
-    ///
-    /// When new message come in, this value is reset to `false`.
-    scrolled_past_read_marker: bool,
-    latest_own_user_receipt: Option<Receipt>,
+struct SearchUiState {
+    room_id: OwnedRoomId,
+    fully_paginated: bool,
+    items: Vector<Arc<TimelineItem>>,
+    next_batch: Option<String>,
+    update_receiver: crossbeam_channel::Receiver<SearchUpdate>,
 }
 
 #[derive(Default, Debug)]

src/shared/mod.rs

@@ -12,6 +12,7 @@ pub mod typing_animation;
 pub mod popup_list;
 pub mod verification_badge;
 pub mod callout_tooltip;
+pub mod search;
 
 pub fn live_design(cx: &mut Cx) {
     // Order matters here, as some widget definitions depend on others.

src/shared/search.rs

@@ -0,0 +1,13 @@
+use matrix_sdk::ruma::OwnedRoomId;
+use std::sync::Arc;
+use matrix_sdk_ui::timeline::TimelineItem;
+
+/// Search result updates sent from async worker to UI.
+pub enum SearchUpdate {
+    // A new batch of search results has been received.
+    NewResults {
+        room_id: OwnedRoomId,
+        results: Vec<Arc<TimelineItem>>, // List of matching search results
+        next_batch: Option<String>,      // Token for pagination
+    },
+}

src/sliding_sync.rs

@@ -8,10 +8,12 @@ use imbl::Vector;
 use makepad_widgets::{error, log, warning, Cx, SignalToUI};
 use matrix_sdk::{
     config::RequestConfig, event_handler::EventHandlerDropGuard, media::MediaRequest, room::{edit::EditedContent, RoomMember}, ruma::{
-        api::client::receipt::create_receipt::v3::ReceiptType, events::{
+        api::client::{
+            receipt::create_receipt::v3::ReceiptType, search::search_events::v3::{Criteria, Categories, Request}
+        }, events::{
             receipt::ReceiptThread, room::{
                 message::{ForwardThread, RoomMessageEventContent}, power_levels::RoomPowerLevels, MediaSource
-            }, FullStateEventContent, MessageLikeEventType, StateEventType
+            }, FullStateEventContent, MessageLikeEventType, StateEventType, AnyTimelineEvent
         }, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedMxcUri, OwnedRoomAliasId, OwnedRoomId, OwnedUserId, UserId
     }, sliding_sync::VersionBuilder, Client, ClientBuildError, Error, Room, RoomMemberships
 };
@@ -33,7 +35,7 @@ use crate::{
     }, login::login_screen::LoginAction, media_cache::MediaCacheEntry, persistent_state::{self, ClientSessionPersisted}, profile::{
         user_profile::{AvatarState, UserProfile},
         user_profile_cache::{enqueue_user_profile_update, UserProfileUpdate},
-    }, shared::{jump_to_bottom_button::UnreadMessageCount, popup_list::enqueue_popup_notification}, utils::{self, AVATAR_THUMBNAIL_FORMAT}, verification::add_verification_event_handlers_and_sync_client
+    }, shared::{search::SearchUpdate, jump_to_bottom_button::UnreadMessageCount, popup_list::enqueue_popup_notification}, utils::{self, AVATAR_THUMBNAIL_FORMAT}, verification::add_verification_event_handlers_and_sync_client
 };
 
 #[derive(Parser, Debug, Default)]
@@ -357,8 +359,19 @@ pub enum MatrixRequest {
         timeline_event_id: TimelineEventItemId,
         reason: Option<String>,
     },
+
+    /// Request to search for messages in a room.
+    SearchRoomMessages {
+        room_id: OwnedRoomId,
+        search_term: String,
+        next_batch: Option<String>,
+    },
 }
 
+// Create a global sender/receiver pair for search results
+pub static SEARCH_RESULTS_SENDER: OnceLock<crossbeam_channel::Sender<SearchUpdate>> = OnceLock::new();
+pub static SEARCH_RESULTS_RECEIVER: OnceLock<crossbeam_channel::Receiver<SearchUpdate>> = OnceLock::new();
+
 /// Submits a request to the worker thread to be executed asynchronously.
 pub fn submit_async_request(req: MatrixRequest) {
     REQUEST_SENDER.get()
@@ -974,6 +987,64 @@ async fn async_worker(
                     }
                 });
             },
+            MatrixRequest::SearchRoomMessages { room_id, search_term, next_batch } => {
+                let mut categories = Categories::new();
+                categories.room_events = Some(Criteria::new(search_term));
+
+                let mut search_request = Request::new(categories);
+                search_request.next_batch = next_batch;
+
+                let client = get_client().unwrap();
+
+                match client.send(search_request, None).await {
+                    Ok(response) => {
+                        let room_events = response.search_categories.room_events;
+
+                        let event_ids: Vec<OwnedEventId> = room_events
+                            .results
+                            .into_iter()
+                            .filter_map(|search_result| {
+                                search_result.result
+                                    .and_then(|raw_event| raw_event.deserialize().ok()) 
+                                    .and_then(|event| match event {
+                                        AnyTimelineEvent::MessageLike(msg) => Some(msg.event_id().to_owned()),
+                                        AnyTimelineEvent::State(state) => Some(state.event_id().to_owned()),
+                                        _ => None,
+                                    })
+                            })
+                            .collect();
+
+                        if event_ids.is_empty() {
+                            log!("Search found no events.");
+                            return Ok(());
+                        }
+
+                        // Now that we have event IDs, request pagination of those events.
+                        let timeline = {
+                            let mut all_room_info = ALL_ROOM_INFO.lock().unwrap();
+                            let Some(room_info) = all_room_info.get_mut(&room_id) else {
+                                log!("Skipping search request for not-yet-known room {room_id}");
+                                return Ok(());
+                            };
+                            room_info.timeline.clone()
+                        };
+
+                        // Fetch events by requesting timeline pagination for each found event.
+                        let _fetch_events_task = Handle::current().spawn(async move {
+                            log!("Fetching {} events for search results in room {}", event_ids.len(), room_id);
+
+                            for event_id in event_ids {
+                                let _ = timeline.fetch_details_for_event(&event_id).await;
+                            }
+
+                            log!("Finished fetching search results.");
+                        });
+                    }
+                    Err(e) => {
+                        error!("Search request failed: {:?}", e);
+                    }
+                }
+            },
         }
     }