matrix_sdk_ui/
search_service.rs

1// Copyright 2026 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! A reactive search service.
16//!
17//! [`SearchService`] aggregates results of different kinds into a single
18//! reactive, paginated list of typed [`ResultType`]s. Call
19//! [`SearchService::set_query`] to start (or restart) a search, then drive it
20//! page by page with [`SearchService::paginate`], observing the results and the
21//! [`PaginationState`] as they change.
22//!
23//! Today the only source is the SDK's per-room message search; people, rooms
24//! and other kinds are expected to be added as further [`ResultType`]
25//! variants, and a `matrix-sdk` source can be swapped for a server-side one
26//! without changing this interface.
27
28use std::pin::Pin;
29
30use eyeball::{ObservableWriteGuard, SharedObservable, Subscriber};
31use eyeball_im::{ObservableVector, Vector, VectorSubscriberBatchedStream};
32use futures_util::{Stream, StreamExt as _};
33use matrix_sdk::{
34    Client, deserialized_responses::TimelineEvent, message_search::SearchError, room::Room,
35};
36use ruma::{MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId};
37use tokio::sync::Mutex as AsyncMutex;
38
39use crate::timeline::{Profile, TimelineDetails, TimelineItemContent};
40
41/// A boxed stream of pages of resolved search hits, as produced by the SDK's
42/// global message search.
43type ResultsStream =
44    Pin<Box<dyn Stream<Item = Result<Vec<(OwnedRoomId, TimelineEvent)>, SearchError>> + Send>>;
45
46/// Whether the search service is currently loading a page of results.
47#[cfg_attr(feature = "uniffi", derive(uniffi::Enum), uniffi(name = "SearchServicePaginationState"))]
48#[derive(Clone, Debug, Eq, PartialEq)]
49pub enum PaginationState {
50    /// Not currently paginating. `end_reached` is `true` once every source has
51    /// been exhausted for the current query.
52    Idle { end_reached: bool },
53    /// A page of results is currently being loaded.
54    Loading,
55}
56
57/// A single search result, tagged by the kind of entity it represents.
58///
59/// New result kinds will be added as additional variants.
60#[derive(Debug, Clone)]
61pub enum ResultType {
62    /// A message (room timeline event) matching the query.
63    Message(MessageResult),
64}
65
66/// A message matching a search query, with its content and sender resolved.
67#[derive(Debug, Clone)]
68pub struct MessageResult {
69    /// The room the message belongs to.
70    pub room_id: OwnedRoomId,
71    /// The event ID of the message.
72    pub event_id: OwnedEventId,
73    /// The sender of the message.
74    pub sender: OwnedUserId,
75    /// The sender's profile, as far as it could be resolved.
76    pub sender_profile: TimelineDetails<Profile>,
77    /// The rendered content of the message.
78    pub content: TimelineItemContent,
79    /// The origin server timestamp of the message.
80    pub timestamp: MilliSecondsSinceUnixEpoch,
81}
82
83impl MessageResult {
84    /// Resolve a search hit into a full result by loading its content and
85    /// sender profile, returning `None` if the event can't be rendered.
86    async fn from_event(room: &Room, event: TimelineEvent) -> Option<Self> {
87        let sender = event.sender()?;
88        let event_id = event.event_id()?.to_owned();
89        let timestamp = event.timestamp().unwrap_or_else(MilliSecondsSinceUnixEpoch::now);
90
91        let content = TimelineItemContent::from_event(room, event).await?;
92        let sender_profile =
93            TimelineDetails::from_initial_value(Profile::load(room, &sender).await);
94
95        Some(Self {
96            room_id: room.room_id().to_owned(),
97            event_id,
98            sender,
99            sender_profile,
100            content,
101            timestamp,
102        })
103    }
104}
105
106/// A reactive, paginated search across all the user's data.
107pub struct SearchService {
108    /// The client used to run searches and resolve their results.
109    client: Client,
110
111    /// The current query's result stream, set by [`Self::set_query`] and pulled
112    /// one page at a time by [`Self::paginate`]. `None` until a query is set.
113    stream: AsyncMutex<Option<ResultsStream>>,
114
115    /// The current pagination state, observable via
116    /// [`Self::subscribe_to_pagination_state_updates`].
117    pagination_state: SharedObservable<PaginationState>,
118
119    /// The accumulated results across the pages loaded so far, observable via
120    /// [`Self::subscribe_to_results`].
121    results: AsyncMutex<ObservableVector<ResultType>>,
122}
123
124impl SearchService {
125    /// Create a new [`SearchService`] for the given client.
126    pub fn new(client: Client) -> Self {
127        Self {
128            client,
129            stream: AsyncMutex::new(None),
130            pagination_state: SharedObservable::new(PaginationState::Idle { end_reached: false }),
131            results: AsyncMutex::new(ObservableVector::new()),
132        }
133    }
134
135    /// Set (or update) the search query.
136    /// Clears the current results, restarts pagination from scratch and loads
137    /// the first page. Call [`Self::paginate`] to load any further pages.
138    pub async fn set_query(&self, query: String) -> Result<(), SearchError> {
139        let stream = self.client.search_messages(query).build_events();
140        *self.stream.lock().await = Some(Box::pin(stream));
141        self.results.lock().await.clear();
142        self.pagination_state.set(PaginationState::Idle { end_reached: false });
143
144        self.paginate().await
145    }
146
147    /// Returns the current pagination state.
148    pub fn pagination_state(&self) -> PaginationState {
149        self.pagination_state.get()
150    }
151
152    /// Subscribe to pagination state updates.
153    pub fn subscribe_to_pagination_state_updates(&self) -> Subscriber<PaginationState> {
154        self.pagination_state.subscribe()
155    }
156
157    /// Return the current list of results.
158    pub async fn results(&self) -> Vec<ResultType> {
159        self.results.lock().await.iter().cloned().collect()
160    }
161
162    /// Subscribe to result list updates.
163    pub async fn subscribe_to_results(
164        &self,
165    ) -> (Vector<ResultType>, VectorSubscriberBatchedStream<ResultType>) {
166        self.results.lock().await.subscribe().into_values_and_batched_stream()
167    }
168
169    /// Load the next page of results, appending them to the list.
170    pub async fn paginate(&self) -> Result<(), SearchError> {
171        {
172            let mut pagination_state = self.pagination_state.write();
173
174            match *pagination_state {
175                PaginationState::Idle { end_reached } if end_reached => return Ok(()),
176                PaginationState::Loading => return Ok(()),
177                _ => {}
178            }
179
180            ObservableWriteGuard::set(&mut pagination_state, PaginationState::Loading);
181        }
182
183        let mut stream = self.stream.lock().await;
184        let Some(stream) = stream.as_mut() else {
185            self.pagination_state.set(PaginationState::Idle { end_reached: true });
186            return Ok(());
187        };
188
189        match stream.next().await {
190            None => {
191                self.pagination_state.set(PaginationState::Idle { end_reached: true });
192            }
193            Some(Err(err)) => {
194                self.pagination_state.set(PaginationState::Idle { end_reached: false });
195                return Err(err);
196            }
197            Some(Ok(page)) => {
198                let mut resolved = Vector::new();
199                for (room_id, event) in page {
200                    let Some(room) = self.client.get_room(&room_id) else {
201                        continue;
202                    };
203                    let Some(result) = MessageResult::from_event(&room, event).await else {
204                        continue;
205                    };
206                    resolved.push_back(ResultType::Message(result));
207                }
208
209                if !resolved.is_empty() {
210                    self.results.lock().await.append(resolved);
211                }
212
213                self.pagination_state.set(PaginationState::Idle { end_reached: false });
214            }
215        }
216
217        Ok(())
218    }
219}
220
221#[cfg(test)]
222mod tests {
223    use std::time::Duration;
224
225    use assert_matches2::assert_let;
226    use eyeball_im::VectorDiff;
227    use futures_util::pin_mut;
228    use matrix_sdk::test_utils::mocks::MatrixMockServer;
229    use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
230    use ruma::{event_id, room_id, user_id};
231    use stream_assert::{assert_next_matches, assert_pending};
232    use tokio::time::sleep;
233
234    use super::{PaginationState, ResultType, SearchService};
235
236    #[async_test]
237    async fn test_search_pagination() {
238        let server = MatrixMockServer::new().await;
239        let client = server.client_builder().build().await;
240
241        let event_cache = client.event_cache();
242        event_cache.subscribe().unwrap();
243
244        let room_id = room_id!("!room:localhost");
245        let event_id = event_id!("$event:localhost");
246        let f = EventFactory::new().sender(user_id!("@user:localhost"));
247
248        server
249            .mock_sync()
250            .ok_and_run(&client, |builder| {
251                builder.add_joined_room(JoinedRoomBuilder::new(room_id).add_timeline_event(
252                    f.text_msg("hello world").room(room_id).event_id(event_id),
253                ));
254            })
255            .await;
256
257        // Let the indexer process the event.
258        sleep(Duration::from_millis(300)).await;
259
260        let search = SearchService::new(client);
261
262        // Starts idle and empty.
263        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false });
264        assert!(search.results().await.is_empty());
265
266        // Setting the query loads the first page automatically.
267        search.set_query("world".to_owned()).await.unwrap();
268
269        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false });
270        let results = search.results().await;
271        assert_eq!(results.len(), 1);
272        assert_let!(ResultType::Message(message) = &results[0]);
273        assert_eq!(message.event_id, event_id);
274
275        // Subscribing now yields the loaded results as the current state.
276        let (initial, results_stream) = search.subscribe_to_results().await;
277        assert_eq!(initial.len(), 1);
278        pin_mut!(results_stream);
279        assert_pending!(results_stream);
280
281        // The next page is empty, so the end is reached and nothing more is emitted.
282        search.paginate().await.unwrap();
283
284        assert_pending!(results_stream);
285        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: true });
286        assert_eq!(search.results().await.len(), 1);
287    }
288
289    #[async_test]
290    async fn test_search_resets_on_query_change() {
291        let server = MatrixMockServer::new().await;
292        let client = server.client_builder().build().await;
293
294        let event_cache = client.event_cache();
295        event_cache.subscribe().unwrap();
296
297        let room_id = room_id!("!room:localhost");
298        let apple_event = event_id!("$apple:localhost");
299        let banana_event = event_id!("$banana:localhost");
300        let f = EventFactory::new().sender(user_id!("@user:localhost"));
301
302        server
303            .mock_sync()
304            .ok_and_run(&client, |builder| {
305                builder.add_joined_room(
306                    JoinedRoomBuilder::new(room_id)
307                        .add_timeline_event(
308                            f.text_msg("apple pie").room(room_id).event_id(apple_event),
309                        )
310                        .add_timeline_event(
311                            f.text_msg("banana split").room(room_id).event_id(banana_event),
312                        ),
313                );
314            })
315            .await;
316
317        sleep(Duration::from_millis(300)).await;
318
319        let search = SearchService::new(client);
320
321        // The first query loads the apple result automatically.
322        search.set_query("apple".to_owned()).await.unwrap();
323
324        let (initial, results_stream) = search.subscribe_to_results().await;
325        assert_eq!(initial.len(), 1);
326        assert_let!(ResultType::Message(message) = &initial[0]);
327        assert_eq!(message.event_id, apple_event);
328        pin_mut!(results_stream);
329        assert_pending!(results_stream);
330
331        // Changing the query clears the previous results and loads the new ones.
332        search.set_query("banana".to_owned()).await.unwrap();
333
334        // The subscriber observes the clear followed by the new page in one batch.
335        assert_next_matches!(results_stream, diffs => {
336            assert_let!([VectorDiff::Clear, VectorDiff::Append { values }] = diffs.as_slice());
337            assert_eq!(values.len(), 1);
338            assert_let!(ResultType::Message(message) = &values[0]);
339            assert_eq!(message.event_id, banana_event);
340        });
341
342        let results = search.results().await;
343        assert_eq!(results.len(), 1);
344        assert_let!(ResultType::Message(message) = &results[0]);
345        assert_eq!(message.event_id, banana_event);
346    }
347}
matrix_sdk_ui/search_service.rs

matrix_sdk_ui/
search_service.rs
