matrix_sdk_ui/room_list_service/mod.rs
1// Copyright 2023 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 that specific language governing permissions and
13// limitations under the License.
14
15//! `RoomListService` API.
16//!
17//! The `RoomListService` is a UI API dedicated to present a list of Matrix
18//! rooms to the user. The syncing is handled by [`SlidingSync`]. The idea is to
19//! expose a simple API to handle most of the client app use cases, like:
20//! Showing and updating a list of rooms, filtering a list of rooms, handling
21//! particular updates of a range of rooms (the ones the client app is showing
22//! to the view, i.e. the rooms present in the viewport) etc.
23//!
24//! As such, the `RoomListService` works as an opinionated state machine. The
25//! states are defined by [`State`]. Actions are attached to the each state
26//! transition.
27//!
28//! The API is purposely small. Sliding Sync is versatile. `RoomListService` is
29//! _one_ specific usage of Sliding Sync.
30//!
31//! # Basic principle
32//!
33//! `RoomListService` works with 1 Sliding Sync List:
34//!
35//! * `all_rooms` (referred by the constant [`ALL_ROOMS_LIST_NAME`]) is the only
36//! list. Its goal is to load all the user' rooms. It starts with a
37//! [`SlidingSyncMode::Selective`] sync-mode with a small range (i.e. a small
38//! set of rooms) to load the first rooms quickly, and then updates to a
39//! [`SlidingSyncMode::Growing`] sync-mode to load the remaining rooms “in the
40//! background”: it will sync the existing rooms and will fetch new rooms, by
41//! a certain batch size.
42//!
43//! This behavior has proven to be empirically satisfying to provide a fast and
44//! fluid user experience for a Matrix client.
45//!
46//! [`RoomListService::all_rooms`] provides a way to get a [`RoomList`] for all
47//! the rooms. From that, calling [`RoomList::entries_with_dynamic_adapters`]
48//! provides a way to get a stream of rooms. This stream is sorted, can be
49//! filtered, and the filter can be changed over time.
50//!
51//! [`RoomListService::state`] provides a way to get a stream of the state
52//! machine's state, which can be pretty helpful for the client app.
53
54pub mod filters;
55mod room_list;
56pub mod sorters;
57mod state;
58
59use std::{sync::Arc, time::Duration};
60
61use async_stream::stream;
62use eyeball::Subscriber;
63use futures_util::{Stream, StreamExt, pin_mut};
64use matrix_sdk::{
65 Client, Error as SlidingSyncError, Room, SlidingSync, SlidingSyncList, SlidingSyncMode,
66 event_cache::EventCacheError, timeout::timeout,
67};
68pub use room_list::*;
69use ruma::{
70 OwnedRoomId, RoomId, UInt, api::client::sync::sync_events::v5 as http, assign,
71 events::StateEventType,
72};
73pub use state::*;
74use thiserror::Error;
75use tracing::{debug, error};
76
77/// The default `required_state` constant value for sliding sync lists and
78/// sliding sync room subscriptions.
79const DEFAULT_REQUIRED_STATE: &[(StateEventType, &str)] = &[
80 (StateEventType::RoomName, ""),
81 (StateEventType::RoomEncryption, ""),
82 (StateEventType::RoomMember, "$LAZY"),
83 (StateEventType::RoomMember, "$ME"),
84 (StateEventType::RoomTopic, ""),
85 // Temporary workaround for https://github.com/matrix-org/matrix-rust-sdk/issues/5285
86 (StateEventType::RoomAvatar, ""),
87 (StateEventType::RoomCanonicalAlias, ""),
88 (StateEventType::RoomPowerLevels, ""),
89 (StateEventType::CallMember, "*"),
90 (StateEventType::RoomJoinRules, ""),
91 (StateEventType::RoomTombstone, ""),
92 // Those two events are required to properly compute room previews.
93 // `StateEventType::RoomCreate` is also necessary to compute the room
94 // version, and thus handling the tombstoned room correctly.
95 (StateEventType::RoomCreate, ""),
96 (StateEventType::RoomHistoryVisibility, ""),
97 // Required to correctly calculate the room display name.
98 (StateEventType::MemberHints, ""),
99 (StateEventType::SpaceParent, "*"),
100 (StateEventType::SpaceChild, "*"),
101];
102
103/// The default `required_state` constant value for sliding sync room
104/// subscriptions that must be added to `DEFAULT_REQUIRED_STATE`.
105const DEFAULT_ROOM_SUBSCRIPTION_EXTRA_REQUIRED_STATE: &[(StateEventType, &str)] =
106 &[(StateEventType::RoomPinnedEvents, "")];
107
108/// The default `timeline_limit` value when used with room subscriptions.
109const DEFAULT_ROOM_SUBSCRIPTION_TIMELINE_LIMIT: u32 = 20;
110
111/// The [`RoomListService`] type. See the module's documentation to learn more.
112#[derive(Debug)]
113pub struct RoomListService {
114 /// Client that has created this [`RoomListService`].
115 client: Client,
116
117 /// The Sliding Sync instance.
118 sliding_sync: Arc<SlidingSync>,
119
120 /// The current state of the `RoomListService`.
121 ///
122 /// `RoomListService` is a simple state-machine.
123 state_machine: StateMachine,
124}
125
126impl RoomListService {
127 /// Create a new `RoomList`.
128 ///
129 /// A [`matrix_sdk::SlidingSync`] client will be created, with a cached list
130 /// already pre-configured.
131 ///
132 /// This won't start an encryption sync, and it's the user's responsibility
133 /// to create one in this case using
134 /// [`EncryptionSyncService`][crate::encryption_sync_service::EncryptionSyncService].
135 pub async fn new(client: Client) -> Result<Self, Error> {
136 Self::new_with_share_pos(client, true).await
137 }
138
139 /// Like [`RoomListService::new`] but with a flag to turn the
140 /// [`SlidingSyncBuilder::share_pos`] on and off.
141 ///
142 /// [`SlidingSyncBuilder::share_pos`]: matrix_sdk::sliding_sync::SlidingSyncBuilder::share_pos
143 pub async fn new_with_share_pos(client: Client, share_pos: bool) -> Result<Self, Error> {
144 let mut builder = client
145 .sliding_sync("room-list")
146 .map_err(Error::SlidingSync)?
147 .with_account_data_extension(
148 assign!(http::request::AccountData::default(), { enabled: Some(true) }),
149 )
150 .with_receipt_extension(assign!(http::request::Receipts::default(), {
151 enabled: Some(true),
152 rooms: Some(vec![http::request::ExtensionRoomConfig::AllSubscribed])
153 }))
154 .with_typing_extension(assign!(http::request::Typing::default(), {
155 enabled: Some(true),
156 }));
157
158 if share_pos {
159 // We don't deal with encryption device messages here so this is safe
160 builder = builder.share_pos();
161 }
162
163 let sliding_sync = builder
164 .add_cached_list(
165 SlidingSyncList::builder(ALL_ROOMS_LIST_NAME)
166 .sync_mode(
167 SlidingSyncMode::new_selective()
168 .add_range(ALL_ROOMS_DEFAULT_SELECTIVE_RANGE),
169 )
170 .timeline_limit(1)
171 .required_state(
172 DEFAULT_REQUIRED_STATE
173 .iter()
174 .map(|(state_event, value)| (state_event.clone(), (*value).to_owned()))
175 .collect(),
176 )
177 .filters(Some(assign!(http::request::ListFilters::default(), {
178 // As defined in the [SlidingSync MSC](https://github.com/matrix-org/matrix-spec-proposals/blob/9450ced7fb9cf5ea9077d029b3adf36aebfa8709/proposals/3575-sync.md?plain=1#L444)
179 // If unset, both invited and joined rooms are returned. If false, no invited rooms are
180 // returned. If true, only invited rooms are returned.
181 is_invite: None,
182 }))),
183 )
184 .await
185 .map_err(Error::SlidingSync)?
186 .build()
187 .await
188 .map(Arc::new)
189 .map_err(Error::SlidingSync)?;
190
191 // Eagerly subscribe the event cache to sync responses.
192 client.event_cache().subscribe()?;
193
194 let state_machine = StateMachine::new();
195
196 // If the sliding sync has successfully restored a sync position, skip the
197 // waiting for the initial sync, and set the state to `SettingUp`; this
198 // way, the first sync will move us to the steady state, and update the
199 // sliding sync list to use the growing sync mode.
200 if sliding_sync.has_pos().await {
201 state_machine.set(State::SettingUp);
202 }
203
204 Ok(Self { client, sliding_sync, state_machine })
205 }
206
207 /// Start to sync the room list.
208 ///
209 /// It's the main method of this entire API. Calling `sync` allows to
210 /// receive updates on the room list: new rooms, rooms updates etc. Those
211 /// updates can be read with [`RoomList::entries`] for example. This method
212 /// returns a [`Stream`] where produced items only hold an empty value
213 /// in case of a sync success, otherwise an error.
214 ///
215 /// The `RoomListService`' state machine is run by this method.
216 ///
217 /// Stopping the [`Stream`] (i.e. by calling [`Self::stop_sync`]), and
218 /// calling [`Self::sync`] again will resume from the previous state of
219 /// the state machine.
220 ///
221 /// This should be used only for testing. In practice, most users should be
222 /// using the [`SyncService`] instead.
223 #[doc(hidden)]
224 pub fn sync(&self) -> impl Stream<Item = Result<(), Error>> + '_ {
225 stream! {
226 let sync = self.sliding_sync.sync();
227 pin_mut!(sync);
228
229 // This is a state machine implementation.
230 // Things happen in this order:
231 //
232 // 1. The next state is calculated,
233 // 2. The actions associated to the next state are run,
234 // 3. A sync is done,
235 // 4. The next state is stored.
236 loop {
237 debug!("Run a sync iteration");
238
239 // Calculate the next state, and run the associated actions.
240 let next_state = self.state_machine.next(&self.sliding_sync).await?;
241
242 // Do the sync.
243 match sync.next().await {
244 // Got a successful result while syncing.
245 Some(Ok(_update_summary)) => {
246 debug!(state = ?next_state, "New state");
247
248 // Update the state.
249 self.state_machine.set(next_state);
250
251 yield Ok(());
252 }
253
254 // Got an error while syncing.
255 Some(Err(error)) => {
256 debug!(expected_state = ?next_state, "New state is an error");
257
258 let next_state = State::Error { from: Box::new(next_state) };
259 self.state_machine.set(next_state);
260
261 yield Err(Error::SlidingSync(error));
262
263 break;
264 }
265
266 // Sync loop has terminated.
267 None => {
268 debug!(expected_state = ?next_state, "New state is a termination");
269
270 let next_state = State::Terminated { from: Box::new(next_state) };
271 self.state_machine.set(next_state);
272
273 break;
274 }
275 }
276 }
277 }
278 }
279
280 /// Force to stop the sync of the `RoomListService` started by
281 /// [`Self::sync`].
282 ///
283 /// It's of utter importance to call this method rather than stop polling
284 /// the `Stream` returned by [`Self::sync`] because it will force the
285 /// cancellation and exit the sync loop, i.e. it will cancel any
286 /// in-flight HTTP requests, cancel any pending futures etc. and put the
287 /// service into a termination state.
288 ///
289 /// Ideally, one wants to consume the `Stream` returned by [`Self::sync`]
290 /// until it returns `None`, because of [`Self::stop_sync`], so that it
291 /// ensures the states are correctly placed.
292 ///
293 /// Stopping the sync of the room list via this method will put the
294 /// state-machine into the [`State::Terminated`] state.
295 ///
296 /// This should be used only for testing. In practice, most users should be
297 /// using the [`SyncService`] instead.
298 #[doc(hidden)]
299 pub fn stop_sync(&self) -> Result<(), Error> {
300 self.sliding_sync.stop_sync().map_err(Error::SlidingSync)
301 }
302
303 /// Force the sliding sync session to expire.
304 ///
305 /// This is used by [`SyncService`][crate::SyncService].
306 ///
307 /// **Warning**: This method **must not** be called while the sync loop is
308 /// running!
309 pub(crate) async fn expire_sync_session(&self) {
310 self.sliding_sync.expire_session().await;
311
312 // Usually, when the session expires, it leads the state to be `Error`,
313 // thus some actions (like refreshing the lists) are executed. However,
314 // if the sync loop has been stopped manually, the state is `Terminated`, and
315 // when the session is forced to expire, the state remains `Terminated`, thus
316 // the actions aren't executed as expected. Consequently, let's update the
317 // state.
318 if let State::Terminated { from } = self.state_machine.get() {
319 self.state_machine.set(State::Error { from });
320 }
321 }
322
323 /// Get a [`Stream`] of [`SyncIndicator`].
324 ///
325 /// Read the documentation of [`SyncIndicator`] to learn more about it.
326 pub fn sync_indicator(
327 &self,
328 delay_before_showing: Duration,
329 delay_before_hiding: Duration,
330 ) -> impl Stream<Item = SyncIndicator> + use<> {
331 let mut state = self.state();
332
333 stream! {
334 // Ensure the `SyncIndicator` is always hidden to start with.
335 yield SyncIndicator::Hide;
336
337 // Let's not wait for an update to happen. The `SyncIndicator` must be
338 // computed as fast as possible.
339 let mut current_state = state.next_now();
340
341 loop {
342 let (sync_indicator, yield_delay) = match current_state {
343 State::Init | State::SettingUp | State::Error { .. } => {
344 (SyncIndicator::Show, delay_before_showing)
345 }
346
347 State::Recovering | State::Running | State::Terminated { .. } => {
348 (SyncIndicator::Hide, delay_before_hiding)
349 }
350 };
351
352 // `state.next().await` has a maximum of `yield_delay` time to execute…
353 let next_state = match timeout(state.next(), yield_delay).await {
354 // A new state has been received before `yield_delay` time. The new
355 // `sync_indicator` value won't be yielded.
356 Ok(next_state) => next_state,
357
358 // No new state has been received before `yield_delay` time. The
359 // `sync_indicator` value can be yielded.
360 Err(_) => {
361 yield sync_indicator;
362
363 // Now that `sync_indicator` has been yielded, let's wait on
364 // the next state again.
365 state.next().await
366 }
367 };
368
369 if let Some(next_state) = next_state {
370 // Update the `current_state`.
371 current_state = next_state;
372 } else {
373 // Something is broken with the state. Let's stop this stream too.
374 break;
375 }
376 }
377 }
378 }
379
380 /// Get the [`Client`] that has been used to create [`Self`].
381 pub fn client(&self) -> &Client {
382 &self.client
383 }
384
385 /// Get a subscriber to the state.
386 pub fn state(&self) -> Subscriber<State> {
387 self.state_machine.subscribe()
388 }
389
390 async fn list_for(&self, sliding_sync_list_name: &str) -> Result<RoomList, Error> {
391 RoomList::new(&self.client, &self.sliding_sync, sliding_sync_list_name, self.state()).await
392 }
393
394 /// Get a [`RoomList`] for all rooms.
395 pub async fn all_rooms(&self) -> Result<RoomList, Error> {
396 self.list_for(ALL_ROOMS_LIST_NAME).await
397 }
398
399 /// Get a [`Room`] if it exists.
400 pub fn room(&self, room_id: &RoomId) -> Result<Room, Error> {
401 self.client.get_room(room_id).ok_or_else(|| Error::RoomNotFound(room_id.to_owned()))
402 }
403
404 /// Subscribe to rooms.
405 ///
406 /// It means that all events from these rooms will be received every time,
407 /// no matter how the `RoomList` is configured.
408 ///
409 /// [`LatestEvents::listen_to_room`][listen_to_room] will be called for each
410 /// room in `room_ids`, so that the [`LatestEventValue`] will automatically
411 /// be calculated and updated for these rooms, for free.
412 ///
413 /// [listen_to_room]: matrix_sdk::latest_events::LatestEvents::listen_to_room
414 /// [`LatestEventValue`]: matrix_sdk::latest_events::LatestEventValue
415 pub async fn subscribe_to_rooms(&self, room_ids: &[&RoomId]) {
416 // Calculate the settings for the room subscriptions.
417 let settings = assign!(http::request::RoomSubscription::default(), {
418 required_state: DEFAULT_REQUIRED_STATE.iter().map(|(state_event, value)| {
419 (state_event.clone(), (*value).to_owned())
420 })
421 .chain(
422 DEFAULT_ROOM_SUBSCRIPTION_EXTRA_REQUIRED_STATE.iter().map(|(state_event, value)| {
423 (state_event.clone(), (*value).to_owned())
424 })
425 )
426 .collect(),
427 timeline_limit: UInt::from(DEFAULT_ROOM_SUBSCRIPTION_TIMELINE_LIMIT),
428 });
429
430 // Decide whether the in-flight request (if any) should be cancelled if needed.
431 let cancel_in_flight_request = match self.state_machine.get() {
432 State::Init | State::Recovering | State::Error { .. } | State::Terminated { .. } => {
433 false
434 }
435 State::SettingUp | State::Running => true,
436 };
437
438 // Before subscribing, let's listen these rooms to calculate their latest
439 // events.
440 let latest_events = self.client.latest_events().await;
441
442 for room_id in room_ids {
443 if let Err(error) = latest_events.listen_to_room(room_id).await {
444 // Let's not fail the room subscription. Instead, emit a log because it's very
445 // unlikely to happen.
446 error!(?error, ?room_id, "Failed to listen to the latest event for this room");
447 }
448 }
449
450 // Subscribe to the rooms.
451 self.sliding_sync.subscribe_to_rooms(room_ids, Some(settings), cancel_in_flight_request)
452 }
453
454 #[cfg(test)]
455 pub fn sliding_sync(&self) -> &SlidingSync {
456 &self.sliding_sync
457 }
458}
459
460/// [`RoomList`]'s errors.
461#[derive(Debug, Error)]
462pub enum Error {
463 /// Error from [`matrix_sdk::SlidingSync`].
464 #[error(transparent)]
465 SlidingSync(SlidingSyncError),
466
467 /// An operation has been requested on an unknown list.
468 #[error("Unknown list `{0}`")]
469 UnknownList(String),
470
471 /// The requested room doesn't exist.
472 #[error("Room `{0}` not found")]
473 RoomNotFound(OwnedRoomId),
474
475 #[error(transparent)]
476 EventCache(#[from] EventCacheError),
477}
478
479/// An hint whether a _sync spinner/loader/toaster_ should be prompted to the
480/// user, indicating that the [`RoomListService`] is syncing.
481///
482/// This is entirely arbitrary and optinionated. Of course, once
483/// [`RoomListService::sync`] has been called, it's going to be constantly
484/// syncing, until [`RoomListService::stop_sync`] is called, or until an error
485/// happened. But in some cases, it's better for the user experience to prompt
486/// to the user that a sync is happening. It's usually the first sync, or the
487/// recovering sync. However, the sync indicator must be prompted if the
488/// aforementioned sync is “slow”, otherwise the indicator is likely to “blink”
489/// pretty fast, which can be very confusing. It's also common to indicate to
490/// the user that a syncing is happening in case of a network error, that
491/// something is catching up etc.
492#[derive(Debug, Eq, PartialEq)]
493pub enum SyncIndicator {
494 /// Show the sync indicator.
495 Show,
496
497 /// Hide the sync indicator.
498 Hide,
499}
500
501#[cfg(test)]
502mod tests {
503 use std::future::ready;
504
505 use futures_util::{StreamExt, pin_mut};
506 use matrix_sdk::{
507 Client, SlidingSyncMode, config::RequestConfig, test_utils::client::mock_matrix_session,
508 };
509 use matrix_sdk_test::async_test;
510 use ruma::api::MatrixVersion;
511 use serde_json::json;
512 use wiremock::{Match, Mock, MockServer, Request, ResponseTemplate, http::Method};
513
514 use super::{ALL_ROOMS_LIST_NAME, Error, RoomListService, State};
515
516 async fn new_client() -> (Client, MockServer) {
517 let session = mock_matrix_session();
518
519 let server = MockServer::start().await;
520 let client = Client::builder()
521 .homeserver_url(server.uri())
522 .server_versions([MatrixVersion::V1_0])
523 .request_config(RequestConfig::new().disable_retry())
524 .build()
525 .await
526 .unwrap();
527 client.restore_session(session).await.unwrap();
528
529 (client, server)
530 }
531
532 pub(super) async fn new_room_list() -> Result<RoomListService, Error> {
533 let (client, _) = new_client().await;
534
535 RoomListService::new(client).await
536 }
537
538 struct SlidingSyncMatcher;
539
540 impl Match for SlidingSyncMatcher {
541 fn matches(&self, request: &Request) -> bool {
542 request.url.path() == "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync"
543 && request.method == Method::POST
544 }
545 }
546
547 #[async_test]
548 async fn test_all_rooms_are_declared() -> Result<(), Error> {
549 let room_list = new_room_list().await?;
550 let sliding_sync = room_list.sliding_sync();
551
552 // List is present, in Selective mode.
553 assert_eq!(
554 sliding_sync
555 .on_list(ALL_ROOMS_LIST_NAME, |list| ready(matches!(
556 list.sync_mode(),
557 SlidingSyncMode::Selective { ranges } if ranges == vec![0..=19]
558 )))
559 .await,
560 Some(true)
561 );
562
563 Ok(())
564 }
565
566 #[async_test]
567 async fn test_expire_sliding_sync_session_manually() -> Result<(), Error> {
568 let (client, server) = new_client().await;
569
570 let room_list = RoomListService::new(client).await?;
571
572 let sync = room_list.sync();
573 pin_mut!(sync);
574
575 // Run a first sync.
576 {
577 let _mock_guard = Mock::given(SlidingSyncMatcher)
578 .respond_with(move |_request: &Request| {
579 ResponseTemplate::new(200).set_body_json(json!({
580 "pos": "0",
581 "lists": {
582 ALL_ROOMS_LIST_NAME: {
583 "count": 0,
584 "ops": [],
585 },
586 },
587 "rooms": {},
588 }))
589 })
590 .mount_as_scoped(&server)
591 .await;
592
593 let _ = sync.next().await;
594 }
595
596 assert_eq!(room_list.state().get(), State::SettingUp);
597
598 // Stop the sync.
599 room_list.stop_sync()?;
600
601 // Do another sync.
602 let _ = sync.next().await;
603
604 // State is `Terminated`, as expected!
605 assert_eq!(
606 room_list.state_machine.get(),
607 State::Terminated { from: Box::new(State::Running) }
608 );
609
610 // Now, let's make the sliding sync session to expire.
611 room_list.expire_sync_session().await;
612
613 // State is `Error`, as a regular session expiration would generate!
614 assert_eq!(room_list.state_machine.get(), State::Error { from: Box::new(State::Running) });
615
616 Ok(())
617 }
618}