diff --git a/bindings/matrix-sdk-ffi/Cargo.toml b/bindings/matrix-sdk-ffi/Cargo.toml index f414ad03074..905c6438c2b 100644 --- a/bindings/matrix-sdk-ffi/Cargo.toml +++ b/bindings/matrix-sdk-ffi/Cargo.toml @@ -25,7 +25,7 @@ crate-type = [ [features] default = ["bundled-sqlite", "unstable-msc4274", "experimental-element-recent-emojis", "experimental-push-secrets", "experimental-search"] -experimental-search = ["matrix-sdk/experimental-search"] +experimental-search = ["matrix-sdk/experimental-search", "matrix-sdk-ui/experimental-search"] # Use SQLite for the session storage. sqlite = ["matrix-sdk/sqlite"] # Use an embedded version of SQLite. diff --git a/bindings/matrix-sdk-ffi/changelog.d/6695.changed.md b/bindings/matrix-sdk-ffi/changelog.d/6695.changed.md new file mode 100644 index 00000000000..627b29a0b46 --- /dev/null +++ b/bindings/matrix-sdk-ffi/changelog.d/6695.changed.md @@ -0,0 +1 @@ +[**breaking**] The message search FFI is now reactive. `Client::search_messages` (and its `GlobalSearchIterator`) and `Room::search_messages` (and its `RoomSearchIterator`) are removed, replaced by `Client::search_service(query, filter)` which returns a `SearchService` object. Call `SearchService::subscribe_to_results` with a `SearchServiceResultsListener` to receive `SearchServiceResultsUpdate`s (`VectorDiff`-style) over a single list of typed `SearchResult`s. diff --git a/bindings/matrix-sdk-ffi/src/lib.rs b/bindings/matrix-sdk-ffi/src/lib.rs index 22bf101a61a..536cba5ff9a 100644 --- a/bindings/matrix-sdk-ffi/src/lib.rs +++ b/bindings/matrix-sdk-ffi/src/lib.rs @@ -26,7 +26,7 @@ mod room_preview; mod ruma; mod runtime; #[cfg(feature = "experimental-search")] -mod search; +mod search_service; mod session_verification; mod spaces; mod store; diff --git a/bindings/matrix-sdk-ffi/src/search.rs b/bindings/matrix-sdk-ffi/src/search.rs deleted file mode 100644 index 9becd3d68c6..00000000000 --- a/bindings/matrix-sdk-ffi/src/search.rs +++ /dev/null @@ -1,203 +0,0 @@ -// Copyright 2026 The Matrix.org Foundation C.I.C. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for that specific language governing permissions and -// limitations under the License. - -use std::pin::Pin; - -use futures_util::{Stream, StreamExt as _}; -use matrix_sdk::{ - deserialized_responses::TimelineEvent, message_search::SearchError as SdkSearchError, -}; -use matrix_sdk_ui::timeline::TimelineDetails; -use ruma::OwnedRoomId; -use tokio::sync::Mutex; - -use crate::{ - client::Client, - error::ClientError, - room::Room, - timeline::{ProfileDetails, TimelineItemContent}, - utils::Timestamp, -}; - -#[derive(uniffi::Error, thiserror::Error, Debug)] -pub enum SearchError { - #[error("Failed to search through the index: {0}")] - IndexError(String), - #[error("Failed to load event content for search result: {0}")] - EventLoadError(String), -} - -impl From for SearchError { - fn from(err: SdkSearchError) -> Self { - match err { - SdkSearchError::IndexError(err) => SearchError::IndexError(err.to_string()), - SdkSearchError::EventLoadError(err) => SearchError::EventLoadError(err.to_string()), - } - } -} - -/// A boxed stream of pages of the [`TimelineEvent`]s matching a single-room -/// search. -type RoomEventStream = - Pin, SdkSearchError>> + Send>>; - -/// A boxed stream of pages of the `(room_id, event)`s matching a global search. -type GlobalEventStream = - Pin, SdkSearchError>> + Send>>; - -#[matrix_sdk_ffi_macros::export] -impl Room { - /// Search for messages in this room matching the given query, returning an - /// iterator that yields one page of results at a time. - pub fn search_messages(&self, query: String) -> RoomSearchIterator { - RoomSearchIterator { - sdk_room: (*self.inner).clone(), - stream: Mutex::new(Box::pin(self.inner.search_messages_events(query))), - } - } -} - -#[derive(uniffi::Object)] -pub struct RoomSearchIterator { - sdk_room: matrix_sdk::room::Room, - stream: Mutex, -} - -#[matrix_sdk_ffi_macros::export] -impl RoomSearchIterator { - /// Return the next page of search results, or `None` if there are no more - /// results. - pub async fn next_events(&self) -> Result>, SearchError> { - let Some(events) = self.stream.lock().await.next().await else { - return Ok(None); - }; - - let events = events?; - let mut results = Vec::with_capacity(events.len()); - for event in events { - if let Some(result) = RoomSearchResult::from(&self.sdk_room, event).await { - results.push(result); - } - } - - results.shrink_to_fit(); - - Ok(Some(results)) - } -} - -#[derive(Clone, uniffi::Record)] -pub struct RoomSearchResult { - event_id: String, - sender: String, - sender_profile: ProfileDetails, - content: TimelineItemContent, - timestamp: Timestamp, -} - -impl RoomSearchResult { - async fn from(room: &matrix_sdk::room::Room, event: TimelineEvent) -> Option { - let sender = event.sender()?; - - let event_id = event.event_id().unwrap().to_string(); - let timestamp = - event.timestamp().unwrap_or_else(ruma::MilliSecondsSinceUnixEpoch::now).into(); - - let content = matrix_sdk_ui::timeline::TimelineItemContent::from_event(room, event).await?; - let profile = TimelineDetails::from_initial_value( - matrix_sdk_ui::timeline::Profile::load(room, &sender).await, - ); - - Some(Self { - event_id, - sender: sender.to_string(), - sender_profile: ProfileDetails::from(profile), - content: TimelineItemContent::from(content), - timestamp, - }) - } -} - -#[derive(Clone, uniffi::Enum)] -pub enum SearchRoomFilter { - /// All the joined rooms (= DMs + non-DMs). - Rooms, - /// Only joined DM rooms. - Dms, - /// Only joined non-DM (group) rooms. - NonDms, -} - -#[matrix_sdk_ffi_macros::export] -impl Client { - /// Search across all all rooms for the given query, returning an iterator - /// over the results. - pub async fn search_messages( - &self, - query: String, - filter: SearchRoomFilter, - ) -> Result { - let sdk_client = (*self.inner).clone(); - let mut builder = sdk_client.search_messages(query); - - match filter { - SearchRoomFilter::Rooms => {} - SearchRoomFilter::Dms => builder = builder.only_dm_rooms().await?, - SearchRoomFilter::NonDms => builder = builder.no_dms().await?, - } - - Ok(GlobalSearchIterator { - sdk_client, - stream: Mutex::new(Box::pin(builder.build_events())), - }) - } -} - -#[derive(uniffi::Record)] -pub struct GlobalSearchResult { - room_id: String, - result: RoomSearchResult, -} - -#[derive(uniffi::Object)] -pub struct GlobalSearchIterator { - sdk_client: matrix_sdk::Client, - stream: Mutex, -} - -#[matrix_sdk_ffi_macros::export] -impl GlobalSearchIterator { - /// Return the next page of search results, or `None` if there are no more - /// results. - pub async fn next_events(&self) -> Result>, SearchError> { - let Some(batch) = self.stream.lock().await.next().await else { - return Ok(None); - }; - - let batch = batch?; - let mut results = Vec::with_capacity(batch.len()); - for (room_id, event) in batch { - let Some(room) = self.sdk_client.get_room(&room_id) else { - continue; - }; - if let Some(result) = RoomSearchResult::from(&room, event).await { - results.push(GlobalSearchResult { room_id: room_id.to_string(), result }); - } - } - - results.shrink_to_fit(); - - Ok(Some(results)) - } -} diff --git a/bindings/matrix-sdk-ffi/src/search_service.rs b/bindings/matrix-sdk-ffi/src/search_service.rs new file mode 100644 index 00000000000..7d7adf5f947 --- /dev/null +++ b/bindings/matrix-sdk-ffi/src/search_service.rs @@ -0,0 +1,197 @@ +// Copyright 2026 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for that specific language governing permissions and +// limitations under the License. + +use std::{fmt::Debug, sync::Arc}; + +use eyeball_im::VectorDiff; +use futures_util::{StreamExt as _, pin_mut}; +use matrix_sdk_common::{SendOutsideWasm, SyncOutsideWasm}; +use matrix_sdk_ui::search_service::{ + MessageResult as UIMessageResult, PaginationState as SearchServicePaginationState, + ResultType as UIResultType, SearchService as UISearchService, +}; + +use crate::{ + TaskHandle, + client::Client, + error::ClientError, + runtime::get_runtime_handle, + timeline::{ProfileDetails, TimelineItemContent}, + utils::Timestamp, +}; + +#[matrix_sdk_ffi_macros::export] +impl Client { + /// Create a search service. + /// + /// The search service aggregates results of different kinds (currently only + /// messages) into a single reactive, paginated list of typed + /// [`SearchResult`]s. Call [`SearchService::set_query`] to start or update + /// the search, then [`SearchService::paginate`] to load more results. + pub fn search_service(&self) -> Arc { + Arc::new(SearchService { inner: UISearchService::new((*self.inner).clone()) }) + } +} + +/// A reactive, paginated search across all the user's data. +#[derive(uniffi::Object)] +pub struct SearchService { + inner: UISearchService, +} + +#[matrix_sdk_ffi_macros::export] +impl SearchService { + /// Set (or update) the search query. + /// Clears the current results, restarts pagination from scratch and loads + /// the first page. Call [`Self::paginate`] to load any further pages. + pub async fn set_query(&self, query: String) -> Result<(), ClientError> { + self.inner.set_query(query).await.map_err(|err| ClientError::from(anyhow::Error::from(err))) + } + + /// Load the next page of results if a page isn't already loading and the + /// end hasn't been reached. Otherwise it no-ops. + pub async fn paginate(&self) -> Result<(), ClientError> { + self.inner.paginate().await.map_err(|err| ClientError::from(anyhow::Error::from(err))) + } + + /// Returns the current pagination state. + pub fn pagination_state(&self) -> SearchServicePaginationState { + self.inner.pagination_state() + } + + /// Subscribe to pagination state updates. + pub fn subscribe_to_pagination_state_updates( + &self, + listener: Box, + ) -> Arc { + let pagination_state = self.inner.subscribe_to_pagination_state_updates(); + + Arc::new(TaskHandle::new(get_runtime_handle().spawn(async move { + pin_mut!(pagination_state); + + while let Some(state) = pagination_state.next().await { + listener.on_update(state); + } + }))) + } + + /// Subscribe to the search results. + pub async fn subscribe_to_results( + &self, + listener: Box, + ) -> Arc { + let (initial_values, mut stream) = self.inner.subscribe_to_results().await; + + listener.on_update(vec![SearchServiceResultsUpdate::Reset { + values: initial_values.into_iter().map(Into::into).collect(), + }]); + + Arc::new(TaskHandle::new(get_runtime_handle().spawn(async move { + while let Some(diffs) = stream.next().await { + listener.on_update(diffs.into_iter().map(Into::into).collect()); + } + }))) + } +} + +#[matrix_sdk_ffi_macros::export(callback_interface)] +pub trait SearchServicePaginationStateListener: SendOutsideWasm + SyncOutsideWasm + Debug { + fn on_update(&self, pagination_state: SearchServicePaginationState); +} + +#[matrix_sdk_ffi_macros::export(callback_interface)] +pub trait SearchServiceResultsListener: SendOutsideWasm + SyncOutsideWasm + Debug { + fn on_update(&self, updates: Vec); +} + +/// A single search result, tagged by the kind of entity it represents. +#[derive(uniffi::Enum)] +pub enum SearchServiceResult { + /// A message (room timeline event) matching the query. + Message { room_id: String, result: MessageSearchResult }, +} + +impl From for SearchServiceResult { + fn from(result: UIResultType) -> Self { + match result { + UIResultType::Message(message) => { + let room_id = message.room_id.to_string(); + Self::Message { room_id, result: message.into() } + } + } + } +} + +/// A message matching a search query, with its content and sender resolved. +#[derive(Clone, uniffi::Record)] +pub struct MessageSearchResult { + event_id: String, + sender: String, + sender_profile: ProfileDetails, + content: TimelineItemContent, + timestamp: Timestamp, +} + +impl From for MessageSearchResult { + fn from(result: UIMessageResult) -> Self { + Self { + event_id: result.event_id.to_string(), + sender: result.sender.to_string(), + sender_profile: ProfileDetails::from(result.sender_profile), + content: TimelineItemContent::from(result.content), + timestamp: result.timestamp.into(), + } + } +} + +#[derive(uniffi::Enum)] +pub enum SearchServiceResultsUpdate { + Append { values: Vec }, + Clear, + PushFront { value: SearchServiceResult }, + PushBack { value: SearchServiceResult }, + PopFront, + PopBack, + Insert { index: u32, value: SearchServiceResult }, + Set { index: u32, value: SearchServiceResult }, + Remove { index: u32 }, + Truncate { length: u32 }, + Reset { values: Vec }, +} + +impl From> for SearchServiceResultsUpdate { + fn from(diff: VectorDiff) -> Self { + match diff { + VectorDiff::Append { values } => { + Self::Append { values: values.into_iter().map(Into::into).collect() } + } + VectorDiff::Clear => Self::Clear, + VectorDiff::PushFront { value } => Self::PushFront { value: value.into() }, + VectorDiff::PushBack { value } => Self::PushBack { value: value.into() }, + VectorDiff::PopFront => Self::PopFront, + VectorDiff::PopBack => Self::PopBack, + VectorDiff::Insert { index, value } => { + Self::Insert { index: index as u32, value: value.into() } + } + VectorDiff::Set { index, value } => { + Self::Set { index: index as u32, value: value.into() } + } + VectorDiff::Remove { index } => Self::Remove { index: index as u32 }, + VectorDiff::Truncate { length } => Self::Truncate { length: length as u32 }, + VectorDiff::Reset { values } => { + Self::Reset { values: values.into_iter().map(Into::into).collect() } + } + } + } +} diff --git a/crates/matrix-sdk-ui/Cargo.toml b/crates/matrix-sdk-ui/Cargo.toml index f88538c9b89..605dabd89ea 100644 --- a/crates/matrix-sdk-ui/Cargo.toml +++ b/crates/matrix-sdk-ui/Cargo.toml @@ -21,6 +21,10 @@ unstable-msc3956 = ["ruma/unstable-msc3956"] # Add support for inline media galleries via msgtypes unstable-msc4274 = ["matrix-sdk/unstable-msc4274"] +# Enable the global, reactive search service built on top of the SDK's +# message search. +experimental-search = ["matrix-sdk/experimental-search"] + # Enable experimental support for encrypting state events; see # https://github.com/matrix-org/matrix-rust-sdk/issues/5397. experimental-encrypted-state-events = [ diff --git a/crates/matrix-sdk-ui/changelog.d/6695.added.md b/crates/matrix-sdk-ui/changelog.d/6695.added.md new file mode 100644 index 00000000000..b8d639405d9 --- /dev/null +++ b/crates/matrix-sdk-ui/changelog.d/6695.added.md @@ -0,0 +1 @@ +Add a reactive `search::SearchService` that aggregates results of different kinds (currently only messages) into a single list of typed `SearchResult`s, observed as a stream of `VectorDiff`s. Gated behind the new `experimental-search` feature. diff --git a/crates/matrix-sdk-ui/src/lib.rs b/crates/matrix-sdk-ui/src/lib.rs index 96d184a0ead..0336496b053 100644 --- a/crates/matrix-sdk-ui/src/lib.rs +++ b/crates/matrix-sdk-ui/src/lib.rs @@ -21,6 +21,8 @@ use ruma::html::HtmlSanitizerMode; pub mod encryption_sync_service; pub mod notification_client; pub mod room_list_service; +#[cfg(feature = "experimental-search")] +pub mod search_service; pub mod spaces; pub mod sync_service; pub mod timeline; diff --git a/crates/matrix-sdk-ui/src/search_service.rs b/crates/matrix-sdk-ui/src/search_service.rs new file mode 100644 index 00000000000..90eb2cb34f7 --- /dev/null +++ b/crates/matrix-sdk-ui/src/search_service.rs @@ -0,0 +1,347 @@ +// Copyright 2026 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! A reactive search service. +//! +//! [`SearchService`] aggregates results of different kinds into a single +//! reactive, paginated list of typed [`ResultType`]s. Call +//! [`SearchService::set_query`] to start (or restart) a search, then drive it +//! page by page with [`SearchService::paginate`], observing the results and the +//! [`PaginationState`] as they change. +//! +//! Today the only source is the SDK's per-room message search; people, rooms +//! and other kinds are expected to be added as further [`ResultType`] +//! variants, and a `matrix-sdk` source can be swapped for a server-side one +//! without changing this interface. + +use std::pin::Pin; + +use eyeball::{ObservableWriteGuard, SharedObservable, Subscriber}; +use eyeball_im::{ObservableVector, Vector, VectorSubscriberBatchedStream}; +use futures_util::{Stream, StreamExt as _}; +use matrix_sdk::{ + Client, deserialized_responses::TimelineEvent, message_search::SearchError, room::Room, +}; +use ruma::{MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId}; +use tokio::sync::Mutex as AsyncMutex; + +use crate::timeline::{Profile, TimelineDetails, TimelineItemContent}; + +/// A boxed stream of pages of resolved search hits, as produced by the SDK's +/// global message search. +type ResultsStream = + Pin, SearchError>> + Send>>; + +/// Whether the search service is currently loading a page of results. +#[cfg_attr(feature = "uniffi", derive(uniffi::Enum), uniffi(name = "SearchServicePaginationState"))] +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum PaginationState { + /// Not currently paginating. `end_reached` is `true` once every source has + /// been exhausted for the current query. + Idle { end_reached: bool }, + /// A page of results is currently being loaded. + Loading, +} + +/// A single search result, tagged by the kind of entity it represents. +/// +/// New result kinds will be added as additional variants. +#[derive(Debug, Clone)] +pub enum ResultType { + /// A message (room timeline event) matching the query. + Message(MessageResult), +} + +/// A message matching a search query, with its content and sender resolved. +#[derive(Debug, Clone)] +pub struct MessageResult { + /// The room the message belongs to. + pub room_id: OwnedRoomId, + /// The event ID of the message. + pub event_id: OwnedEventId, + /// The sender of the message. + pub sender: OwnedUserId, + /// The sender's profile, as far as it could be resolved. + pub sender_profile: TimelineDetails, + /// The rendered content of the message. + pub content: TimelineItemContent, + /// The origin server timestamp of the message. + pub timestamp: MilliSecondsSinceUnixEpoch, +} + +impl MessageResult { + /// Resolve a search hit into a full result by loading its content and + /// sender profile, returning `None` if the event can't be rendered. + async fn from_event(room: &Room, event: TimelineEvent) -> Option { + let sender = event.sender()?; + let event_id = event.event_id()?.to_owned(); + let timestamp = event.timestamp().unwrap_or_else(MilliSecondsSinceUnixEpoch::now); + + let content = TimelineItemContent::from_event(room, event).await?; + let sender_profile = + TimelineDetails::from_initial_value(Profile::load(room, &sender).await); + + Some(Self { + room_id: room.room_id().to_owned(), + event_id, + sender, + sender_profile, + content, + timestamp, + }) + } +} + +/// A reactive, paginated search across all the user's data. +pub struct SearchService { + /// The client used to run searches and resolve their results. + client: Client, + + /// The current query's result stream, set by [`Self::set_query`] and pulled + /// one page at a time by [`Self::paginate`]. `None` until a query is set. + stream: AsyncMutex>, + + /// The current pagination state, observable via + /// [`Self::subscribe_to_pagination_state_updates`]. + pagination_state: SharedObservable, + + /// The accumulated results across the pages loaded so far, observable via + /// [`Self::subscribe_to_results`]. + results: AsyncMutex>, +} + +impl SearchService { + /// Create a new [`SearchService`] for the given client. + pub fn new(client: Client) -> Self { + Self { + client, + stream: AsyncMutex::new(None), + pagination_state: SharedObservable::new(PaginationState::Idle { end_reached: false }), + results: AsyncMutex::new(ObservableVector::new()), + } + } + + /// Set (or update) the search query. + /// Clears the current results, restarts pagination from scratch and loads + /// the first page. Call [`Self::paginate`] to load any further pages. + pub async fn set_query(&self, query: String) -> Result<(), SearchError> { + let stream = self.client.search_messages(query).build_events(); + *self.stream.lock().await = Some(Box::pin(stream)); + self.results.lock().await.clear(); + self.pagination_state.set(PaginationState::Idle { end_reached: false }); + + self.paginate().await + } + + /// Returns the current pagination state. + pub fn pagination_state(&self) -> PaginationState { + self.pagination_state.get() + } + + /// Subscribe to pagination state updates. + pub fn subscribe_to_pagination_state_updates(&self) -> Subscriber { + self.pagination_state.subscribe() + } + + /// Return the current list of results. + pub async fn results(&self) -> Vec { + self.results.lock().await.iter().cloned().collect() + } + + /// Subscribe to result list updates. + pub async fn subscribe_to_results( + &self, + ) -> (Vector, VectorSubscriberBatchedStream) { + self.results.lock().await.subscribe().into_values_and_batched_stream() + } + + /// Load the next page of results, appending them to the list. + pub async fn paginate(&self) -> Result<(), SearchError> { + { + let mut pagination_state = self.pagination_state.write(); + + match *pagination_state { + PaginationState::Idle { end_reached } if end_reached => return Ok(()), + PaginationState::Loading => return Ok(()), + _ => {} + } + + ObservableWriteGuard::set(&mut pagination_state, PaginationState::Loading); + } + + let mut stream = self.stream.lock().await; + let Some(stream) = stream.as_mut() else { + self.pagination_state.set(PaginationState::Idle { end_reached: true }); + return Ok(()); + }; + + match stream.next().await { + None => { + self.pagination_state.set(PaginationState::Idle { end_reached: true }); + } + Some(Err(err)) => { + self.pagination_state.set(PaginationState::Idle { end_reached: false }); + return Err(err); + } + Some(Ok(page)) => { + let mut resolved = Vector::new(); + for (room_id, event) in page { + let Some(room) = self.client.get_room(&room_id) else { + continue; + }; + let Some(result) = MessageResult::from_event(&room, event).await else { + continue; + }; + resolved.push_back(ResultType::Message(result)); + } + + if !resolved.is_empty() { + self.results.lock().await.append(resolved); + } + + self.pagination_state.set(PaginationState::Idle { end_reached: false }); + } + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use assert_matches2::assert_let; + use eyeball_im::VectorDiff; + use futures_util::pin_mut; + use matrix_sdk::test_utils::mocks::MatrixMockServer; + use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory}; + use ruma::{event_id, room_id, user_id}; + use stream_assert::{assert_next_matches, assert_pending}; + use tokio::time::sleep; + + use super::{PaginationState, ResultType, SearchService}; + + #[async_test] + async fn test_search_pagination() { + let server = MatrixMockServer::new().await; + let client = server.client_builder().build().await; + + let event_cache = client.event_cache(); + event_cache.subscribe().unwrap(); + + let room_id = room_id!("!room:localhost"); + let event_id = event_id!("$event:localhost"); + let f = EventFactory::new().sender(user_id!("@user:localhost")); + + server + .mock_sync() + .ok_and_run(&client, |builder| { + builder.add_joined_room(JoinedRoomBuilder::new(room_id).add_timeline_event( + f.text_msg("hello world").room(room_id).event_id(event_id), + )); + }) + .await; + + // Let the indexer process the event. + sleep(Duration::from_millis(300)).await; + + let search = SearchService::new(client); + + // Starts idle and empty. + assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false }); + assert!(search.results().await.is_empty()); + + // Setting the query loads the first page automatically. + search.set_query("world".to_owned()).await.unwrap(); + + assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false }); + let results = search.results().await; + assert_eq!(results.len(), 1); + assert_let!(ResultType::Message(message) = &results[0]); + assert_eq!(message.event_id, event_id); + + // Subscribing now yields the loaded results as the current state. + let (initial, results_stream) = search.subscribe_to_results().await; + assert_eq!(initial.len(), 1); + pin_mut!(results_stream); + assert_pending!(results_stream); + + // The next page is empty, so the end is reached and nothing more is emitted. + search.paginate().await.unwrap(); + + assert_pending!(results_stream); + assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: true }); + assert_eq!(search.results().await.len(), 1); + } + + #[async_test] + async fn test_search_resets_on_query_change() { + let server = MatrixMockServer::new().await; + let client = server.client_builder().build().await; + + let event_cache = client.event_cache(); + event_cache.subscribe().unwrap(); + + let room_id = room_id!("!room:localhost"); + let apple_event = event_id!("$apple:localhost"); + let banana_event = event_id!("$banana:localhost"); + let f = EventFactory::new().sender(user_id!("@user:localhost")); + + server + .mock_sync() + .ok_and_run(&client, |builder| { + builder.add_joined_room( + JoinedRoomBuilder::new(room_id) + .add_timeline_event( + f.text_msg("apple pie").room(room_id).event_id(apple_event), + ) + .add_timeline_event( + f.text_msg("banana split").room(room_id).event_id(banana_event), + ), + ); + }) + .await; + + sleep(Duration::from_millis(300)).await; + + let search = SearchService::new(client); + + // The first query loads the apple result automatically. + search.set_query("apple".to_owned()).await.unwrap(); + + let (initial, results_stream) = search.subscribe_to_results().await; + assert_eq!(initial.len(), 1); + assert_let!(ResultType::Message(message) = &initial[0]); + assert_eq!(message.event_id, apple_event); + pin_mut!(results_stream); + assert_pending!(results_stream); + + // Changing the query clears the previous results and loads the new ones. + search.set_query("banana".to_owned()).await.unwrap(); + + // The subscriber observes the clear followed by the new page in one batch. + assert_next_matches!(results_stream, diffs => { + assert_let!([VectorDiff::Clear, VectorDiff::Append { values }] = diffs.as_slice()); + assert_eq!(values.len(), 1); + assert_let!(ResultType::Message(message) = &values[0]); + assert_eq!(message.event_id, banana_event); + }); + + let results = search.results().await; + assert_eq!(results.len(), 1); + assert_let!(ResultType::Message(message) = &results[0]); + assert_eq!(message.event_id, banana_event); + } +}