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}