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. Clears the current results, restarts
136    /// pagination from scratch and loads the first page. Call
137    /// [`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::{pin::pin, time::Duration};
224
225    use eyeball_im::VectorDiff;
226    use matrix_sdk::test_utils::mocks::MatrixMockServer;
227    use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
228    use ruma::{event_id, room_id, user_id};
229    use strass::assert_let;
230    use stream_assert::{assert_next_matches, assert_pending};
231    use tokio::time::sleep;
232
233    use super::{PaginationState, ResultType, SearchService};
234
235    #[async_test]
236    async fn test_search_pagination() {
237        let server = MatrixMockServer::new().await;
238        let client = server.client_builder().build().await;
239
240        let room_id = room_id!("!room:localhost");
241        let event_id = event_id!("$event:localhost");
242        let f = EventFactory::new().sender(user_id!("@user:localhost"));
243
244        server
245            .mock_sync()
246            .ok_and_run(&client, |builder| {
247                builder.add_joined_room(JoinedRoomBuilder::new(room_id).add_timeline_event(
248                    f.text_msg("hello world").room(room_id).event_id(event_id),
249                ));
250            })
251            .await;
252
253        // Let the indexer process the event.
254        sleep(Duration::from_millis(300)).await;
255
256        let search = SearchService::new(client);
257
258        // Starts idle and empty.
259        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false });
260        assert!(search.results().await.is_empty());
261
262        // Setting the query loads the first page automatically.
263        search.set_query("world".to_owned()).await.unwrap();
264
265        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: false });
266        let results = search.results().await;
267        assert_eq!(results.len(), 1);
268        assert_let!(ResultType::Message(message) = &results[0]);
269        assert_eq!(message.event_id, event_id);
270
271        // Subscribing now yields the loaded results as the current state.
272        let (initial, results_stream) = search.subscribe_to_results().await;
273        assert_eq!(initial.len(), 1);
274        let mut results_stream = pin!(results_stream);
275        assert_pending!(results_stream);
276
277        // The next page is empty, so the end is reached and nothing more is
278        // emitted.
279        search.paginate().await.unwrap();
280
281        assert_pending!(results_stream);
282        assert_eq!(search.pagination_state(), PaginationState::Idle { end_reached: true });
283        assert_eq!(search.results().await.len(), 1);
284    }
285
286    #[async_test]
287    async fn test_search_resets_on_query_change() {
288        let server = MatrixMockServer::new().await;
289        let client = server.client_builder().build().await;
290
291        let room_id = room_id!("!room:localhost");
292        let apple_event = event_id!("$apple:localhost");
293        let banana_event = event_id!("$banana:localhost");
294        let f = EventFactory::new().sender(user_id!("@user:localhost"));
295
296        server
297            .mock_sync()
298            .ok_and_run(&client, |builder| {
299                builder.add_joined_room(
300                    JoinedRoomBuilder::new(room_id)
301                        .add_timeline_event(
302                            f.text_msg("apple pie").room(room_id).event_id(apple_event),
303                        )
304                        .add_timeline_event(
305                            f.text_msg("banana split").room(room_id).event_id(banana_event),
306                        ),
307                );
308            })
309            .await;
310
311        sleep(Duration::from_millis(300)).await;
312
313        let search = SearchService::new(client);
314
315        // The first query loads the apple result automatically.
316        search.set_query("apple".to_owned()).await.unwrap();
317
318        let (initial, results_stream) = search.subscribe_to_results().await;
319        assert_eq!(initial.len(), 1);
320        assert_let!(ResultType::Message(message) = &initial[0]);
321        assert_eq!(message.event_id, apple_event);
322        let mut results_stream = pin!(results_stream);
323        assert_pending!(results_stream);
324
325        // Changing the query clears the previous results and loads the new
326        // ones.
327        search.set_query("banana".to_owned()).await.unwrap();
328
329        // The subscriber observes the clear followed by the new page in one
330        // batch.
331        assert_next_matches!(results_stream, diffs => {
332            assert_let!([VectorDiff::Clear, VectorDiff::Append { values }] = diffs.as_slice());
333            assert_eq!(values.len(), 1);
334            assert_let!(ResultType::Message(message) = &values[0]);
335            assert_eq!(message.event_id, banana_event);
336        });
337
338        let results = search.results().await;
339        assert_eq!(results.len(), 1);
340        assert_let!(ResultType::Message(message) = &results[0]);
341        assert_eq!(message.event_id, banana_event);
342    }
343}
matrix_sdk_ui/search_service.rs

matrix_sdk_ui/
search_service.rs
