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}