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