matrix_sdk_base/
client.rs

1// Copyright 2020 Damir Jelić
2// Copyright 2020 The Matrix.org Foundation C.I.C.
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16#[cfg(feature = "e2e-encryption")]
17use std::sync::Arc;
18use std::{
19    collections::{BTreeMap, BTreeSet, HashMap},
20    fmt,
21    ops::Deref,
22};
23
24use eyeball::{SharedObservable, Subscriber};
25use eyeball_im::{Vector, VectorDiff};
26use futures_util::Stream;
27use matrix_sdk_common::timer;
28#[cfg(feature = "e2e-encryption")]
29use matrix_sdk_crypto::{
30    CollectStrategy, DecryptionSettings, EncryptionSettings, OlmError, OlmMachine,
31    TrustRequirement, store::DynCryptoStore, types::requests::ToDeviceRequest,
32};
33#[cfg(doc)]
34use ruma::DeviceId;
35#[cfg(feature = "e2e-encryption")]
36use ruma::events::room::{history_visibility::HistoryVisibility, member::MembershipState};
37use ruma::{
38    MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedUserId, RoomId, UserId,
39    api::client::{self as api, sync::sync_events::v5},
40    events::{
41        StateEvent, StateEventType,
42        ignored_user_list::IgnoredUserListEventContent,
43        push_rules::{PushRulesEvent, PushRulesEventContent},
44        room::member::SyncRoomMemberEvent,
45    },
46    push::Ruleset,
47    time::Instant,
48};
49use tokio::sync::{Mutex, broadcast};
50#[cfg(feature = "e2e-encryption")]
51use tokio::sync::{RwLock, RwLockReadGuard};
52use tracing::{Level, debug, enabled, info, instrument, warn};
53
54#[cfg(feature = "e2e-encryption")]
55use crate::RoomMemberships;
56use crate::{
57    InviteAcceptanceDetails, RoomStateFilter, SessionMeta,
58    deserialized_responses::DisplayName,
59    error::{Error, Result},
60    event_cache::store::EventCacheStoreLock,
61    media::store::MediaStoreLock,
62    response_processors::{self as processors, Context},
63    room::{
64        Room, RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons, RoomMembersUpdate, RoomState,
65    },
66    store::{
67        BaseStateStore, DynStateStore, MemoryStore, Result as StoreResult, RoomLoadSettings,
68        StateChanges, StateStoreDataKey, StateStoreDataValue, StateStoreExt, StoreConfig,
69        ambiguity_map::AmbiguityCache,
70    },
71    sync::{RoomUpdates, SyncResponse},
72};
73
74/// A no (network) IO client implementation.
75///
76/// This client is a state machine that receives responses and events and
77/// accordingly updates its state. It is not designed to be used directly, but
78/// rather through `matrix_sdk::Client`.
79///
80/// ```rust
81/// use matrix_sdk_base::{BaseClient, ThreadingSupport, store::StoreConfig};
82///
83/// let client = BaseClient::new(
84///     StoreConfig::new("cross-process-holder-name".to_owned()),
85///     ThreadingSupport::Disabled,
86/// );
87/// ```
88#[derive(Clone)]
89pub struct BaseClient {
90    /// The state store.
91    pub(crate) state_store: BaseStateStore,
92
93    /// The store used by the event cache.
94    event_cache_store: EventCacheStoreLock,
95
96    /// The store used by the media cache.
97    media_store: MediaStoreLock,
98
99    /// The store used for encryption.
100    ///
101    /// This field is only meant to be used for `OlmMachine` initialization.
102    /// All operations on it happen inside the `OlmMachine`.
103    #[cfg(feature = "e2e-encryption")]
104    crypto_store: Arc<DynCryptoStore>,
105
106    /// The olm-machine that is created once the
107    /// [`SessionMeta`][crate::session::SessionMeta] is set via
108    /// [`BaseClient::activate`]
109    #[cfg(feature = "e2e-encryption")]
110    olm_machine: Arc<RwLock<Option<OlmMachine>>>,
111
112    /// Observable of when a user is ignored/unignored.
113    pub(crate) ignore_user_list_changes: SharedObservable<Vec<String>>,
114
115    /// A sender that is used to communicate changes to room information. Each
116    /// tick contains the room ID and the reasons that have generated this tick.
117    pub(crate) room_info_notable_update_sender: broadcast::Sender<RoomInfoNotableUpdate>,
118
119    /// The strategy to use for picking recipient devices, when sending an
120    /// encrypted message.
121    #[cfg(feature = "e2e-encryption")]
122    pub room_key_recipient_strategy: CollectStrategy,
123
124    /// The settings to use for decrypting events.
125    #[cfg(feature = "e2e-encryption")]
126    pub decryption_settings: DecryptionSettings,
127
128    /// If the client should handle verification events received when syncing.
129    #[cfg(feature = "e2e-encryption")]
130    pub handle_verification_events: bool,
131
132    /// Whether the client supports threads or not.
133    pub threading_support: ThreadingSupport,
134}
135
136#[cfg(not(tarpaulin_include))]
137impl fmt::Debug for BaseClient {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        f.debug_struct("BaseClient")
140            .field("session_meta", &self.state_store.session_meta())
141            .field("sync_token", &self.state_store.sync_token)
142            .finish_non_exhaustive()
143    }
144}
145
146/// Whether this client instance supports threading or not. Currently used to
147/// determine how the client handles read receipts and unread count computations
148/// on the base SDK level.
149///
150/// Timelines on the other hand have a separate `TimelineFocus`
151/// `hide_threaded_events` associated value that can be used to hide threaded
152/// events but also to enable threaded read receipt sending. This is because
153/// certain timeline instances should ignore threading no matter what's defined
154/// at the client level. One such example are media filtered timelines which
155/// should contain all the room's media no matter what thread its in (unless
156/// explicitly opted into).
157#[derive(Clone, Copy, Debug)]
158pub enum ThreadingSupport {
159    /// Threading enabled.
160    Enabled {
161        /// Enable client-wide thread subscriptions support (MSC4306 / MSC4308).
162        ///
163        /// This may cause filtering out of thread subscriptions, and loading
164        /// the thread subscriptions via the sliding sync extension,
165        /// when the room list service is being used.
166        with_subscriptions: bool,
167    },
168    /// Threading disabled.
169    Disabled,
170}
171
172impl BaseClient {
173    /// Create a new client.
174    ///
175    /// # Arguments
176    ///
177    /// * `config` - the configuration for the stores (state store, event cache
178    ///   store and crypto store).
179    pub fn new(config: StoreConfig, threading_support: ThreadingSupport) -> Self {
180        let store = BaseStateStore::new(config.state_store);
181
182        // Create the channel to receive `RoomInfoNotableUpdate`.
183        //
184        // Let's consider the channel will receive 5 updates for 100 rooms maximum. This
185        // is unrealistic in practise, as the sync mechanism is pretty unlikely to
186        // trigger such amount of updates, it's a safe value.
187        //
188        // Also, note that it must not be
189        // zero, because (i) it will panic, (ii) a new user has no room, but can create
190        // rooms; remember that the channel's capacity is immutable.
191        let (room_info_notable_update_sender, _room_info_notable_update_receiver) =
192            broadcast::channel(500);
193
194        BaseClient {
195            state_store: store,
196            event_cache_store: config.event_cache_store,
197            media_store: config.media_store,
198            #[cfg(feature = "e2e-encryption")]
199            crypto_store: config.crypto_store,
200            #[cfg(feature = "e2e-encryption")]
201            olm_machine: Default::default(),
202            ignore_user_list_changes: Default::default(),
203            room_info_notable_update_sender,
204            #[cfg(feature = "e2e-encryption")]
205            room_key_recipient_strategy: Default::default(),
206            #[cfg(feature = "e2e-encryption")]
207            decryption_settings: DecryptionSettings {
208                sender_device_trust_requirement: TrustRequirement::Untrusted,
209            },
210            #[cfg(feature = "e2e-encryption")]
211            handle_verification_events: true,
212            threading_support,
213        }
214    }
215
216    /// Clones the current base client to use the same crypto store but a
217    /// different, in-memory store config, and resets transient state.
218    #[cfg(feature = "e2e-encryption")]
219    pub async fn clone_with_in_memory_state_store(
220        &self,
221        cross_process_store_locks_holder_name: &str,
222        handle_verification_events: bool,
223    ) -> Result<Self> {
224        let config = StoreConfig::new(cross_process_store_locks_holder_name.to_owned())
225            .state_store(MemoryStore::new());
226        let config = config.crypto_store(self.crypto_store.clone());
227
228        let copy = Self {
229            state_store: BaseStateStore::new(config.state_store),
230            event_cache_store: config.event_cache_store,
231            media_store: config.media_store,
232            // We copy the crypto store as well as the `OlmMachine` for two reasons:
233            // 1. The `self.crypto_store` is the same as the one used inside the `OlmMachine`.
234            // 2. We need to ensure that the parent and child use the same data and caches inside
235            //    the `OlmMachine` so the various ratchets and places where new randomness gets
236            //    introduced don't diverge, i.e. one-time keys that get generated by the Olm Account
237            //    or Olm sessions when they encrypt or decrypt messages.
238            crypto_store: self.crypto_store.clone(),
239            olm_machine: self.olm_machine.clone(),
240            ignore_user_list_changes: Default::default(),
241            room_info_notable_update_sender: self.room_info_notable_update_sender.clone(),
242            room_key_recipient_strategy: self.room_key_recipient_strategy.clone(),
243            decryption_settings: self.decryption_settings.clone(),
244            handle_verification_events,
245            threading_support: self.threading_support,
246        };
247
248        copy.state_store
249            .derive_from_other(&self.state_store, &copy.room_info_notable_update_sender)
250            .await?;
251
252        Ok(copy)
253    }
254
255    /// Clones the current base client to use the same crypto store but a
256    /// different, in-memory store config, and resets transient state.
257    #[cfg(not(feature = "e2e-encryption"))]
258    #[allow(clippy::unused_async)]
259    pub async fn clone_with_in_memory_state_store(
260        &self,
261        cross_process_store_locks_holder: &str,
262        _handle_verification_events: bool,
263    ) -> Result<Self> {
264        let config = StoreConfig::new(cross_process_store_locks_holder.to_owned())
265            .state_store(MemoryStore::new());
266        Ok(Self::new(config, ThreadingSupport::Disabled))
267    }
268
269    /// Get the session meta information.
270    ///
271    /// If the client is currently logged in, this will return a
272    /// [`SessionMeta`] object which contains the user ID and device ID.
273    /// Otherwise it returns `None`.
274    pub fn session_meta(&self) -> Option<&SessionMeta> {
275        self.state_store.session_meta()
276    }
277
278    /// Get all the rooms this client knows about.
279    pub fn rooms(&self) -> Vec<Room> {
280        self.state_store.rooms()
281    }
282
283    /// Get all the rooms this client knows about, filtered by room state.
284    pub fn rooms_filtered(&self, filter: RoomStateFilter) -> Vec<Room> {
285        self.state_store.rooms_filtered(filter)
286    }
287
288    /// Get a stream of all the rooms changes, in addition to the existing
289    /// rooms.
290    pub fn rooms_stream(
291        &self,
292    ) -> (Vector<Room>, impl Stream<Item = Vec<VectorDiff<Room>>> + use<>) {
293        self.state_store.rooms_stream()
294    }
295
296    /// Lookup the Room for the given RoomId, or create one, if it didn't exist
297    /// yet in the store
298    pub fn get_or_create_room(&self, room_id: &RoomId, room_state: RoomState) -> Room {
299        self.state_store.get_or_create_room(
300            room_id,
301            room_state,
302            self.room_info_notable_update_sender.clone(),
303        )
304    }
305
306    /// Get a reference to the state store.
307    pub fn state_store(&self) -> &DynStateStore {
308        self.state_store.deref()
309    }
310
311    /// Get a reference to the event cache store.
312    pub fn event_cache_store(&self) -> &EventCacheStoreLock {
313        &self.event_cache_store
314    }
315
316    /// Get a reference to the media store.
317    pub fn media_store(&self) -> &MediaStoreLock {
318        &self.media_store
319    }
320
321    /// Check whether the client has been activated.
322    ///
323    /// See [`BaseClient::activate`] to know what it means.
324    pub fn is_active(&self) -> bool {
325        self.state_store.session_meta().is_some()
326    }
327
328    /// Activate the client.
329    ///
330    /// A client is considered active when:
331    ///
332    /// 1. It has a `SessionMeta` (user ID, device ID and access token),
333    /// 2. Has loaded cached data from storage,
334    /// 3. If encryption is enabled, it also initialized or restored its
335    ///    `OlmMachine`.
336    ///
337    /// # Arguments
338    ///
339    /// * `session_meta` - The meta of a session that the user already has from
340    ///   a previous login call.
341    ///
342    /// * `custom_account` - A custom
343    ///   [`matrix_sdk_crypto::vodozemac::olm::Account`] to be used for the
344    ///   identity and one-time keys of this [`BaseClient`]. If no account is
345    ///   provided, a new default one or one from the store will be used. If an
346    ///   account is provided and one already exists in the store for this
347    ///   [`UserId`]/[`DeviceId`] combination, an error will be raised. This is
348    ///   useful if one wishes to create identity keys before knowing the
349    ///   user/device IDs, e.g., to use the identity key as the device ID.
350    ///
351    /// * `room_load_settings` — Specify how many rooms must be restored; use
352    ///   `::default()` if you don't know which value to pick.
353    ///
354    /// # Panics
355    ///
356    /// This method panics if it is called twice.
357    ///
358    /// [`UserId`]: ruma::UserId
359    pub async fn activate(
360        &self,
361        session_meta: SessionMeta,
362        room_load_settings: RoomLoadSettings,
363        #[cfg(feature = "e2e-encryption")] custom_account: Option<
364            crate::crypto::vodozemac::olm::Account,
365        >,
366    ) -> Result<()> {
367        debug!(user_id = ?session_meta.user_id, device_id = ?session_meta.device_id, "Activating the client");
368
369        self.state_store
370            .load_rooms(
371                &session_meta.user_id,
372                room_load_settings,
373                &self.room_info_notable_update_sender,
374            )
375            .await?;
376        self.state_store.load_sync_token().await?;
377        self.state_store.set_session_meta(session_meta);
378
379        #[cfg(feature = "e2e-encryption")]
380        self.regenerate_olm(custom_account).await?;
381
382        Ok(())
383    }
384
385    /// Recreate an `OlmMachine` from scratch.
386    ///
387    /// In particular, this will clear all its caches.
388    #[cfg(feature = "e2e-encryption")]
389    pub async fn regenerate_olm(
390        &self,
391        custom_account: Option<crate::crypto::vodozemac::olm::Account>,
392    ) -> Result<()> {
393        tracing::debug!("regenerating OlmMachine");
394        let session_meta = self.session_meta().ok_or(Error::OlmError(OlmError::MissingSession))?;
395
396        // Recreate the `OlmMachine` and wipe the in-memory cache in the store
397        // because we suspect it has stale data.
398        let olm_machine = OlmMachine::with_store(
399            &session_meta.user_id,
400            &session_meta.device_id,
401            self.crypto_store.clone(),
402            custom_account,
403        )
404        .await
405        .map_err(OlmError::from)?;
406
407        *self.olm_machine.write().await = Some(olm_machine);
408        Ok(())
409    }
410
411    /// Get the current, if any, sync token of the client.
412    /// This will be None if the client didn't sync at least once.
413    pub async fn sync_token(&self) -> Option<String> {
414        self.state_store.sync_token.read().await.clone()
415    }
416
417    /// User has knocked on a room.
418    ///
419    /// Update the internal and cached state accordingly. Return the final Room.
420    pub async fn room_knocked(&self, room_id: &RoomId) -> Result<Room> {
421        let room = self.state_store.get_or_create_room(
422            room_id,
423            RoomState::Knocked,
424            self.room_info_notable_update_sender.clone(),
425        );
426
427        if room.state() != RoomState::Knocked {
428            let _sync_lock = self.sync_lock().lock().await;
429
430            let mut room_info = room.clone_info();
431            room_info.mark_as_knocked();
432            room_info.mark_state_partially_synced();
433            room_info.mark_members_missing(); // the own member event changed
434            let mut changes = StateChanges::default();
435            changes.add_room(room_info.clone());
436            self.state_store.save_changes(&changes).await?; // Update the store
437            room.set_room_info(room_info, RoomInfoNotableUpdateReasons::MEMBERSHIP);
438        }
439
440        Ok(room)
441    }
442
443    /// The user has joined a room using this specific client.
444    ///
445    /// This method should be called if the user accepts an invite or if they
446    /// join a public room.
447    ///
448    /// The method will create a [`Room`] object if one does not exist yet and
449    /// set the state of the [`Room`] to [`RoomState::Joined`]. The [`Room`]
450    /// object will be persisted in the cache. Please note that the [`Room`]
451    /// will be a stub until a sync has been received with the full room
452    /// state using [`BaseClient::receive_sync_response`].
453    ///
454    /// Update the internal and cached state accordingly. Return the final Room.
455    ///
456    /// # Arguments
457    ///
458    /// * `room_id` - The unique ID identifying the joined room.
459    /// * `inviter` - When joining this room in response to an invitation, the
460    ///   inviter should be recorded before sending the join request to the
461    ///   server. Providing the inviter here ensures that the
462    ///   [`InviteAcceptanceDetails`] are stored for this room.
463    ///
464    /// # Examples
465    ///
466    /// ```rust
467    /// # use matrix_sdk_base::{BaseClient, store::StoreConfig, RoomState, ThreadingSupport};
468    /// # use ruma::{OwnedRoomId, OwnedUserId, RoomId};
469    /// # async {
470    /// # let client = BaseClient::new(StoreConfig::new("example".to_owned()), ThreadingSupport::Disabled);
471    /// # async fn send_join_request() -> anyhow::Result<OwnedRoomId> { todo!() }
472    /// # async fn maybe_get_inviter(room_id: &RoomId) -> anyhow::Result<Option<OwnedUserId>> { todo!() }
473    /// # let room_id: &RoomId = todo!();
474    /// let maybe_inviter = maybe_get_inviter(room_id).await?;
475    /// let room_id = send_join_request().await?;
476    /// let room = client.room_joined(&room_id, maybe_inviter).await?;
477    ///
478    /// assert_eq!(room.state(), RoomState::Joined);
479    /// # matrix_sdk_test::TestResult::Ok(()) };
480    /// ```
481    pub async fn room_joined(
482        &self,
483        room_id: &RoomId,
484        inviter: Option<OwnedUserId>,
485    ) -> Result<Room> {
486        let room = self.state_store.get_or_create_room(
487            room_id,
488            RoomState::Joined,
489            self.room_info_notable_update_sender.clone(),
490        );
491
492        // If the state isn't `RoomState::Joined` then this means that we knew about
493        // this room before. Let's modify the existing state now.
494        if room.state() != RoomState::Joined {
495            let _sync_lock = self.sync_lock().lock().await;
496
497            let mut room_info = room.clone_info();
498            let previous_state = room.state();
499
500            room_info.mark_as_joined();
501            room_info.mark_state_partially_synced();
502            room_info.mark_members_missing(); // the own member event changed
503
504            // If our previous state was an invite and we're now in the joined state, this
505            // means that the user has explicitly accepted an invite. Let's
506            // remember some details about the invite.
507            //
508            // This is somewhat of a workaround for our lack of cryptographic membership.
509            // Later on we will decide if historic room keys should be accepted
510            // based on this info. If a user has accepted an invite and we receive a room
511            // key bundle shortly after, we might accept it. If we don't do
512            // this, the homeserver could trick us into accepting any historic room key
513            // bundle.
514            if previous_state == RoomState::Invited
515                && let Some(inviter) = inviter
516            {
517                let details = InviteAcceptanceDetails {
518                    invite_accepted_at: MilliSecondsSinceUnixEpoch::now(),
519                    inviter,
520                };
521                room_info.set_invite_acceptance_details(details);
522            }
523
524            let mut changes = StateChanges::default();
525            changes.add_room(room_info.clone());
526
527            self.state_store.save_changes(&changes).await?; // Update the store
528
529            room.set_room_info(room_info, RoomInfoNotableUpdateReasons::MEMBERSHIP);
530        }
531
532        Ok(room)
533    }
534
535    /// User has left a room.
536    ///
537    /// Update the internal and cached state accordingly.
538    pub async fn room_left(&self, room_id: &RoomId) -> Result<()> {
539        let room = self.state_store.get_or_create_room(
540            room_id,
541            RoomState::Left,
542            self.room_info_notable_update_sender.clone(),
543        );
544
545        if room.state() != RoomState::Left {
546            let _sync_lock = self.sync_lock().lock().await;
547
548            let mut room_info = room.clone_info();
549            room_info.mark_as_left();
550            room_info.mark_state_partially_synced();
551            room_info.mark_members_missing(); // the own member event changed
552            let mut changes = StateChanges::default();
553            changes.add_room(room_info.clone());
554            self.state_store.save_changes(&changes).await?; // Update the store
555            room.set_room_info(room_info, RoomInfoNotableUpdateReasons::MEMBERSHIP);
556        }
557
558        Ok(())
559    }
560
561    /// Get access to the store's sync lock.
562    pub fn sync_lock(&self) -> &Mutex<()> {
563        self.state_store.sync_lock()
564    }
565
566    /// Receive a response from a sync call.
567    ///
568    /// # Arguments
569    ///
570    /// * `response` - The response that we received after a successful sync.
571    #[instrument(skip_all)]
572    pub async fn receive_sync_response(
573        &self,
574        response: api::sync::sync_events::v3::Response,
575    ) -> Result<SyncResponse> {
576        self.receive_sync_response_with_requested_required_states(
577            response,
578            &RequestedRequiredStates::default(),
579        )
580        .await
581    }
582
583    /// Receive a response from a sync call, with the requested required state
584    /// events.
585    ///
586    /// # Arguments
587    ///
588    /// * `response` - The response that we received after a successful sync.
589    /// * `requested_required_states` - The requested required state events.
590    pub async fn receive_sync_response_with_requested_required_states(
591        &self,
592        response: api::sync::sync_events::v3::Response,
593        requested_required_states: &RequestedRequiredStates,
594    ) -> Result<SyncResponse> {
595        // The server might respond multiple times with the same sync token, in
596        // that case we already received this response and there's nothing to
597        // do.
598        if self.state_store.sync_token.read().await.as_ref() == Some(&response.next_batch) {
599            info!("Got the same sync response twice");
600            return Ok(SyncResponse::default());
601        }
602
603        let now = if enabled!(Level::INFO) { Some(Instant::now()) } else { None };
604
605        #[cfg(feature = "e2e-encryption")]
606        let olm_machine = self.olm_machine().await;
607
608        let mut context = Context::new(StateChanges::new(response.next_batch.clone()));
609
610        #[cfg(feature = "e2e-encryption")]
611        let to_device = {
612            let processors::e2ee::to_device::Output {
613                processed_to_device_events: to_device,
614                room_key_updates,
615            } = processors::e2ee::to_device::from_sync_v2(
616                &response,
617                olm_machine.as_ref(),
618                &self.decryption_settings,
619            )
620            .await?;
621
622            processors::latest_event::decrypt_from_rooms(
623                &mut context,
624                room_key_updates
625                    .into_iter()
626                    .flatten()
627                    .filter_map(|room_key_info| self.get_room(&room_key_info.room_id))
628                    .collect(),
629                processors::e2ee::E2EE::new(
630                    olm_machine.as_ref(),
631                    &self.decryption_settings,
632                    self.handle_verification_events,
633                ),
634            )
635            .await?;
636
637            to_device
638        };
639
640        #[cfg(not(feature = "e2e-encryption"))]
641        let to_device = response
642            .to_device
643            .events
644            .into_iter()
645            .map(|raw| {
646                use matrix_sdk_common::deserialized_responses::{
647                    ProcessedToDeviceEvent, ToDeviceUnableToDecryptInfo,
648                    ToDeviceUnableToDecryptReason,
649                };
650
651                if let Ok(Some(event_type)) = raw.get_field::<String>("type") {
652                    if event_type == "m.room.encrypted" {
653                        ProcessedToDeviceEvent::UnableToDecrypt {
654                            encrypted_event: raw,
655                            utd_info: ToDeviceUnableToDecryptInfo {
656                                reason: ToDeviceUnableToDecryptReason::EncryptionIsDisabled,
657                            },
658                        }
659                    } else {
660                        ProcessedToDeviceEvent::PlainText(raw)
661                    }
662                } else {
663                    // Exclude events with no type
664                    ProcessedToDeviceEvent::Invalid(raw)
665                }
666            })
667            .collect();
668
669        let mut ambiguity_cache = AmbiguityCache::new(self.state_store.inner.clone());
670
671        let global_account_data_processor =
672            processors::account_data::global(&response.account_data.events);
673
674        let push_rules = self.get_push_rules(&global_account_data_processor).await?;
675
676        let mut room_updates = RoomUpdates::default();
677        let mut notifications = Default::default();
678
679        let mut updated_members_in_room: BTreeMap<OwnedRoomId, BTreeSet<OwnedUserId>> =
680            BTreeMap::new();
681
682        for (room_id, joined_room) in response.rooms.join {
683            let joined_room_update = processors::room::sync_v2::update_joined_room(
684                &mut context,
685                processors::room::RoomCreationData::new(
686                    &room_id,
687                    self.room_info_notable_update_sender.clone(),
688                    requested_required_states,
689                    &mut ambiguity_cache,
690                ),
691                joined_room,
692                &mut updated_members_in_room,
693                processors::notification::Notification::new(
694                    &push_rules,
695                    &mut notifications,
696                    &self.state_store,
697                ),
698                #[cfg(feature = "e2e-encryption")]
699                processors::e2ee::E2EE::new(
700                    olm_machine.as_ref(),
701                    &self.decryption_settings,
702                    self.handle_verification_events,
703                ),
704            )
705            .await?;
706
707            room_updates.joined.insert(room_id, joined_room_update);
708        }
709
710        for (room_id, left_room) in response.rooms.leave {
711            let left_room_update = processors::room::sync_v2::update_left_room(
712                &mut context,
713                processors::room::RoomCreationData::new(
714                    &room_id,
715                    self.room_info_notable_update_sender.clone(),
716                    requested_required_states,
717                    &mut ambiguity_cache,
718                ),
719                left_room,
720                processors::notification::Notification::new(
721                    &push_rules,
722                    &mut notifications,
723                    &self.state_store,
724                ),
725                #[cfg(feature = "e2e-encryption")]
726                processors::e2ee::E2EE::new(
727                    olm_machine.as_ref(),
728                    &self.decryption_settings,
729                    self.handle_verification_events,
730                ),
731            )
732            .await?;
733
734            room_updates.left.insert(room_id, left_room_update);
735        }
736
737        for (room_id, invited_room) in response.rooms.invite {
738            let invited_room_update = processors::room::sync_v2::update_invited_room(
739                &mut context,
740                &room_id,
741                invited_room,
742                self.room_info_notable_update_sender.clone(),
743                processors::notification::Notification::new(
744                    &push_rules,
745                    &mut notifications,
746                    &self.state_store,
747                ),
748            )
749            .await?;
750
751            room_updates.invited.insert(room_id, invited_room_update);
752        }
753
754        for (room_id, knocked_room) in response.rooms.knock {
755            let knocked_room_update = processors::room::sync_v2::update_knocked_room(
756                &mut context,
757                &room_id,
758                knocked_room,
759                self.room_info_notable_update_sender.clone(),
760                processors::notification::Notification::new(
761                    &push_rules,
762                    &mut notifications,
763                    &self.state_store,
764                ),
765            )
766            .await?;
767
768            room_updates.knocked.insert(room_id, knocked_room_update);
769        }
770
771        global_account_data_processor.apply(&mut context, &self.state_store).await;
772
773        context.state_changes.presence = response
774            .presence
775            .events
776            .iter()
777            .filter_map(|e| {
778                let event = e.deserialize().ok()?;
779                Some((event.sender, e.clone()))
780            })
781            .collect();
782
783        context.state_changes.ambiguity_maps = ambiguity_cache.cache;
784
785        {
786            let _sync_lock = self.sync_lock().lock().await;
787
788            processors::changes::save_and_apply(
789                context,
790                &self.state_store,
791                &self.ignore_user_list_changes,
792                Some(response.next_batch.clone()),
793            )
794            .await?;
795        }
796
797        let mut context = Context::default();
798
799        // Now that all the rooms information have been saved, update the display name
800        // of the updated rooms (which relies on information stored in the database).
801        processors::room::display_name::update_for_rooms(
802            &mut context,
803            &room_updates,
804            &self.state_store,
805        )
806        .await;
807
808        // Save the new display name updates if any.
809        processors::changes::save_only(context, &self.state_store).await?;
810
811        for (room_id, member_ids) in updated_members_in_room {
812            if let Some(room) = self.get_room(&room_id) {
813                let _ =
814                    room.room_member_updates_sender.send(RoomMembersUpdate::Partial(member_ids));
815            }
816        }
817
818        if enabled!(Level::INFO) {
819            info!("Processed a sync response in {:?}", now.map(|now| now.elapsed()));
820        }
821
822        let response = SyncResponse {
823            rooms: room_updates,
824            presence: response.presence.events,
825            account_data: response.account_data.events,
826            to_device,
827            notifications,
828        };
829
830        Ok(response)
831    }
832
833    /// Receive a get member events response and convert it to a deserialized
834    /// `MembersResponse`
835    ///
836    /// This client-server request must be made without filters to make sure all
837    /// members are received. Otherwise, an error is returned.
838    ///
839    /// # Arguments
840    ///
841    /// * `room_id` - The room id this response belongs to.
842    ///
843    /// * `response` - The raw response that was received from the server.
844    #[instrument(skip_all, fields(?room_id))]
845    pub async fn receive_all_members(
846        &self,
847        room_id: &RoomId,
848        request: &api::membership::get_member_events::v3::Request,
849        response: &api::membership::get_member_events::v3::Response,
850    ) -> Result<()> {
851        if request.membership.is_some() || request.not_membership.is_some() || request.at.is_some()
852        {
853            // This function assumes all members are loaded at once to optimise how display
854            // name disambiguation works. Using it with partial member list results
855            // would produce incorrect disambiguated display name entries
856            return Err(Error::InvalidReceiveMembersParameters);
857        }
858
859        let Some(room) = self.state_store.room(room_id) else {
860            // The room is unknown to us: leave early.
861            return Ok(());
862        };
863
864        let mut chunk = Vec::with_capacity(response.chunk.len());
865        let mut context = Context::default();
866
867        #[cfg(feature = "e2e-encryption")]
868        let mut user_ids = BTreeSet::new();
869
870        let mut ambiguity_map: HashMap<DisplayName, BTreeSet<OwnedUserId>> = Default::default();
871
872        for raw_event in &response.chunk {
873            let member = match raw_event.deserialize() {
874                Ok(ev) => ev,
875                Err(e) => {
876                    let event_id: Option<String> = raw_event.get_field("event_id").ok().flatten();
877                    debug!(event_id, "Failed to deserialize member event: {e}");
878                    continue;
879                }
880            };
881
882            // TODO: All the actions in this loop used to be done only when the membership
883            // event was not in the store before. This was changed with the new room API,
884            // because e.g. leaving a room makes members events outdated and they need to be
885            // fetched by `members`. Therefore, they need to be overwritten here, even
886            // if they exist.
887            // However, this makes a new problem occur where setting the member events here
888            // potentially races with the sync.
889            // See <https://github.com/matrix-org/matrix-rust-sdk/issues/1205>.
890
891            #[cfg(feature = "e2e-encryption")]
892            match member.membership() {
893                MembershipState::Join | MembershipState::Invite => {
894                    user_ids.insert(member.state_key().to_owned());
895                }
896                _ => (),
897            }
898
899            if let StateEvent::Original(e) = &member
900                && let Some(d) = &e.content.displayname
901            {
902                let display_name = DisplayName::new(d);
903                ambiguity_map.entry(display_name).or_default().insert(member.state_key().clone());
904            }
905
906            let sync_member: SyncRoomMemberEvent = member.clone().into();
907            processors::profiles::upsert_or_delete(&mut context, room_id, &sync_member);
908
909            context
910                .state_changes
911                .state
912                .entry(room_id.to_owned())
913                .or_default()
914                .entry(member.event_type())
915                .or_default()
916                .insert(member.state_key().to_string(), raw_event.clone().cast());
917            chunk.push(member);
918        }
919
920        #[cfg(feature = "e2e-encryption")]
921        processors::e2ee::tracked_users::update(
922            self.olm_machine().await.as_ref(),
923            room.encryption_state(),
924            &user_ids,
925        )
926        .await?;
927
928        context.state_changes.ambiguity_maps.insert(room_id.to_owned(), ambiguity_map);
929
930        let _sync_lock = self.sync_lock().lock().await;
931        let mut room_info = room.clone_info();
932        room_info.mark_members_synced();
933        context.state_changes.add_room(room_info);
934
935        processors::changes::save_and_apply(
936            context,
937            &self.state_store,
938            &self.ignore_user_list_changes,
939            None,
940        )
941        .await?;
942
943        let _ = room.room_member_updates_sender.send(RoomMembersUpdate::FullReload);
944
945        Ok(())
946    }
947
948    /// Receive a successful filter upload response, the filter id will be
949    /// stored under the given name in the store.
950    ///
951    /// The filter id can later be retrieved with the [`get_filter`] method.
952    ///
953    ///
954    /// # Arguments
955    ///
956    /// * `filter_name` - The name that should be used to persist the filter id
957    ///   in the store.
958    ///
959    /// * `response` - The successful filter upload response containing the
960    ///   filter id.
961    ///
962    /// [`get_filter`]: #method.get_filter
963    pub async fn receive_filter_upload(
964        &self,
965        filter_name: &str,
966        response: &api::filter::create_filter::v3::Response,
967    ) -> Result<()> {
968        Ok(self
969            .state_store
970            .set_kv_data(
971                StateStoreDataKey::Filter(filter_name),
972                StateStoreDataValue::Filter(response.filter_id.clone()),
973            )
974            .await?)
975    }
976
977    /// Get the filter id of a previously uploaded filter.
978    ///
979    /// *Note*: A filter will first need to be uploaded and persisted using
980    /// [`receive_filter_upload`].
981    ///
982    /// # Arguments
983    ///
984    /// * `filter_name` - The name of the filter that was previously used to
985    ///   persist the filter.
986    ///
987    /// [`receive_filter_upload`]: #method.receive_filter_upload
988    pub async fn get_filter(&self, filter_name: &str) -> StoreResult<Option<String>> {
989        let filter = self
990            .state_store
991            .get_kv_data(StateStoreDataKey::Filter(filter_name))
992            .await?
993            .map(|d| d.into_filter().expect("State store data not a filter"));
994
995        Ok(filter)
996    }
997
998    /// Get a to-device request that will share a room key with users in a room.
999    #[cfg(feature = "e2e-encryption")]
1000    pub async fn share_room_key(&self, room_id: &RoomId) -> Result<Vec<Arc<ToDeviceRequest>>> {
1001        match self.olm_machine().await.as_ref() {
1002            Some(o) => {
1003                let Some(room) = self.get_room(room_id) else {
1004                    return Err(Error::InsufficientData);
1005                };
1006
1007                let history_visibility = room.history_visibility_or_default();
1008                let Some(room_encryption_event) = room.encryption_settings() else {
1009                    return Err(Error::EncryptionNotEnabled);
1010                };
1011
1012                // Don't share the group session with members that are invited
1013                // if the history visibility is set to `Joined`
1014                let filter = if history_visibility == HistoryVisibility::Joined {
1015                    RoomMemberships::JOIN
1016                } else {
1017                    RoomMemberships::ACTIVE
1018                };
1019
1020                let members = self.state_store.get_user_ids(room_id, filter).await?;
1021
1022                let settings = EncryptionSettings::new(
1023                    room_encryption_event,
1024                    history_visibility,
1025                    self.room_key_recipient_strategy.clone(),
1026                );
1027
1028                Ok(o.share_room_key(room_id, members.iter().map(Deref::deref), settings).await?)
1029            }
1030            None => panic!("Olm machine wasn't started"),
1031        }
1032    }
1033
1034    /// Get the room with the given room id.
1035    ///
1036    /// # Arguments
1037    ///
1038    /// * `room_id` - The id of the room that should be fetched.
1039    pub fn get_room(&self, room_id: &RoomId) -> Option<Room> {
1040        self.state_store.room(room_id)
1041    }
1042
1043    /// Forget the room with the given room ID.
1044    ///
1045    /// The room will be dropped from the room list and the store.
1046    ///
1047    /// # Arguments
1048    ///
1049    /// * `room_id` - The id of the room that should be forgotten.
1050    pub async fn forget_room(&self, room_id: &RoomId) -> Result<()> {
1051        // Forget the room in the state store.
1052        self.state_store.forget_room(room_id).await?;
1053
1054        // Remove the room in the event cache store too.
1055        self.event_cache_store().lock().await?.remove_room(room_id).await?;
1056
1057        Ok(())
1058    }
1059
1060    /// Get the olm machine.
1061    #[cfg(feature = "e2e-encryption")]
1062    pub async fn olm_machine(&self) -> RwLockReadGuard<'_, Option<OlmMachine>> {
1063        self.olm_machine.read().await
1064    }
1065
1066    /// Get the push rules.
1067    ///
1068    /// Gets the push rules previously processed, otherwise get them from the
1069    /// store. As a fallback, uses [`Ruleset::server_default`] if the user
1070    /// is logged in.
1071    pub(crate) async fn get_push_rules(
1072        &self,
1073        global_account_data_processor: &processors::account_data::Global,
1074    ) -> Result<Ruleset> {
1075        let _timer = timer!(Level::TRACE, "get_push_rules");
1076        if let Some(event) = global_account_data_processor
1077            .push_rules()
1078            .and_then(|ev| ev.deserialize_as_unchecked::<PushRulesEvent>().ok())
1079        {
1080            Ok(event.content.global)
1081        } else if let Some(event) = self
1082            .state_store
1083            .get_account_data_event_static::<PushRulesEventContent>()
1084            .await?
1085            .and_then(|ev| ev.deserialize().ok())
1086        {
1087            Ok(event.content.global)
1088        } else if let Some(session_meta) = self.state_store.session_meta() {
1089            Ok(Ruleset::server_default(&session_meta.user_id))
1090        } else {
1091            Ok(Ruleset::new())
1092        }
1093    }
1094
1095    /// Returns a subscriber that publishes an event every time the ignore user
1096    /// list changes
1097    pub fn subscribe_to_ignore_user_list_changes(&self) -> Subscriber<Vec<String>> {
1098        self.ignore_user_list_changes.subscribe()
1099    }
1100
1101    /// Returns a new receiver that gets future room info notable updates.
1102    ///
1103    /// Learn more by reading the [`RoomInfoNotableUpdate`] type.
1104    pub fn room_info_notable_update_receiver(&self) -> broadcast::Receiver<RoomInfoNotableUpdate> {
1105        self.room_info_notable_update_sender.subscribe()
1106    }
1107
1108    /// Checks whether the provided `user_id` belongs to an ignored user.
1109    pub async fn is_user_ignored(&self, user_id: &UserId) -> bool {
1110        match self.state_store.get_account_data_event_static::<IgnoredUserListEventContent>().await
1111        {
1112            Ok(Some(raw_ignored_user_list)) => match raw_ignored_user_list.deserialize() {
1113                Ok(current_ignored_user_list) => {
1114                    current_ignored_user_list.content.ignored_users.contains_key(user_id)
1115                }
1116                Err(error) => {
1117                    warn!(?error, "Failed to deserialize the ignored user list event");
1118                    false
1119                }
1120            },
1121            Ok(None) => false,
1122            Err(error) => {
1123                warn!(?error, "Could not get the ignored user list from the state store");
1124                false
1125            }
1126        }
1127    }
1128}
1129
1130/// Represent the `required_state` values sent by a sync request.
1131///
1132/// This is useful to track what state events have been requested when handling
1133/// a response.
1134///
1135/// For example, if a sync requests the `m.room.encryption` state event, and the
1136/// server replies with nothing, if means the room **is not** encrypted. Without
1137/// knowing which state event was required by the sync, it is impossible to
1138/// interpret the absence of state event from the server as _the room's
1139/// encryption state is **not encrypted**_ or _the room's encryption state is
1140/// **unknown**_.
1141#[derive(Debug, Default)]
1142pub struct RequestedRequiredStates {
1143    default: Vec<(StateEventType, String)>,
1144    for_rooms: HashMap<OwnedRoomId, Vec<(StateEventType, String)>>,
1145}
1146
1147impl RequestedRequiredStates {
1148    /// Create a new `RequestedRequiredStates`.
1149    ///
1150    /// `default` represents the `required_state` value for all rooms.
1151    /// `for_rooms` is the `required_state` per room.
1152    pub fn new(
1153        default: Vec<(StateEventType, String)>,
1154        for_rooms: HashMap<OwnedRoomId, Vec<(StateEventType, String)>>,
1155    ) -> Self {
1156        Self { default, for_rooms }
1157    }
1158
1159    /// Get the `required_state` value for a specific room.
1160    pub fn for_room(&self, room_id: &RoomId) -> &[(StateEventType, String)] {
1161        self.for_rooms.get(room_id).unwrap_or(&self.default)
1162    }
1163}
1164
1165impl From<&v5::Request> for RequestedRequiredStates {
1166    fn from(request: &v5::Request) -> Self {
1167        // The following information is missing in the MSC4186 at the time of writing
1168        // (2025-03-12) but: the `required_state`s from all lists and from all room
1169        // subscriptions are combined by doing an union.
1170        //
1171        // Thus, we can do the same here, put the union in `default` and keep
1172        // `for_rooms` empty. The `Self::for_room` will automatically do the fallback.
1173        let mut default = BTreeSet::new();
1174
1175        for list in request.lists.values() {
1176            default.extend(BTreeSet::from_iter(list.room_details.required_state.iter().cloned()));
1177        }
1178
1179        for room_subscription in request.room_subscriptions.values() {
1180            default.extend(BTreeSet::from_iter(room_subscription.required_state.iter().cloned()));
1181        }
1182
1183        Self { default: default.into_iter().collect(), for_rooms: HashMap::new() }
1184    }
1185}
1186
1187#[cfg(test)]
1188mod tests {
1189    use std::collections::HashMap;
1190
1191    use assert_matches2::{assert_let, assert_matches};
1192    use futures_util::FutureExt as _;
1193    use matrix_sdk_test::{
1194        BOB, InvitedRoomBuilder, LeftRoomBuilder, StateTestEvent, StrippedStateTestEvent,
1195        SyncResponseBuilder, async_test, event_factory::EventFactory, ruma_response_from_json,
1196    };
1197    use ruma::{
1198        api::client::{self as api, sync::sync_events::v5},
1199        event_id,
1200        events::{StateEventType, room::member::MembershipState},
1201        room_id,
1202        serde::Raw,
1203        user_id,
1204    };
1205    use serde_json::{json, value::to_raw_value};
1206
1207    use super::{BaseClient, RequestedRequiredStates};
1208    use crate::{
1209        RoomDisplayName, RoomState, SessionMeta,
1210        client::ThreadingSupport,
1211        store::{RoomLoadSettings, StateStoreExt, StoreConfig},
1212        test_utils::logged_in_base_client,
1213    };
1214
1215    #[test]
1216    fn test_requested_required_states() {
1217        let room_id_0 = room_id!("!r0");
1218        let room_id_1 = room_id!("!r1");
1219
1220        let requested_required_states = RequestedRequiredStates::new(
1221            vec![(StateEventType::RoomAvatar, "".to_owned())],
1222            HashMap::from([(
1223                room_id_0.to_owned(),
1224                vec![
1225                    (StateEventType::RoomMember, "foo".to_owned()),
1226                    (StateEventType::RoomEncryption, "".to_owned()),
1227                ],
1228            )]),
1229        );
1230
1231        // A special set of state events exists for `room_id_0`.
1232        assert_eq!(
1233            requested_required_states.for_room(room_id_0),
1234            &[
1235                (StateEventType::RoomMember, "foo".to_owned()),
1236                (StateEventType::RoomEncryption, "".to_owned()),
1237            ]
1238        );
1239
1240        // No special list for `room_id_1`, it should return the defaults.
1241        assert_eq!(
1242            requested_required_states.for_room(room_id_1),
1243            &[(StateEventType::RoomAvatar, "".to_owned()),]
1244        );
1245    }
1246
1247    #[test]
1248    fn test_requested_required_states_from_sync_v5_request() {
1249        let room_id_0 = room_id!("!r0");
1250        let room_id_1 = room_id!("!r1");
1251
1252        // Empty request.
1253        let mut request = v5::Request::new();
1254
1255        {
1256            let requested_required_states = RequestedRequiredStates::from(&request);
1257
1258            assert!(requested_required_states.default.is_empty());
1259            assert!(requested_required_states.for_rooms.is_empty());
1260        }
1261
1262        // One list.
1263        request.lists.insert("foo".to_owned(), {
1264            let mut list = v5::request::List::default();
1265            list.room_details.required_state = vec![
1266                (StateEventType::RoomAvatar, "".to_owned()),
1267                (StateEventType::RoomEncryption, "".to_owned()),
1268            ];
1269
1270            list
1271        });
1272
1273        {
1274            let requested_required_states = RequestedRequiredStates::from(&request);
1275
1276            assert_eq!(
1277                requested_required_states.default,
1278                &[
1279                    (StateEventType::RoomAvatar, "".to_owned()),
1280                    (StateEventType::RoomEncryption, "".to_owned())
1281                ]
1282            );
1283            assert!(requested_required_states.for_rooms.is_empty());
1284        }
1285
1286        // Two lists.
1287        request.lists.insert("bar".to_owned(), {
1288            let mut list = v5::request::List::default();
1289            list.room_details.required_state = vec![
1290                (StateEventType::RoomEncryption, "".to_owned()),
1291                (StateEventType::RoomName, "".to_owned()),
1292            ];
1293
1294            list
1295        });
1296
1297        {
1298            let requested_required_states = RequestedRequiredStates::from(&request);
1299
1300            // Union of the state events.
1301            assert_eq!(
1302                requested_required_states.default,
1303                &[
1304                    (StateEventType::RoomAvatar, "".to_owned()),
1305                    (StateEventType::RoomEncryption, "".to_owned()),
1306                    (StateEventType::RoomName, "".to_owned()),
1307                ]
1308            );
1309            assert!(requested_required_states.for_rooms.is_empty());
1310        }
1311
1312        // One room subscription.
1313        request.room_subscriptions.insert(room_id_0.to_owned(), {
1314            let mut room_subscription = v5::request::RoomSubscription::default();
1315
1316            room_subscription.required_state = vec![
1317                (StateEventType::RoomJoinRules, "".to_owned()),
1318                (StateEventType::RoomEncryption, "".to_owned()),
1319            ];
1320
1321            room_subscription
1322        });
1323
1324        {
1325            let requested_required_states = RequestedRequiredStates::from(&request);
1326
1327            // Union of state events, all in `default`, still nothing in `for_rooms`.
1328            assert_eq!(
1329                requested_required_states.default,
1330                &[
1331                    (StateEventType::RoomAvatar, "".to_owned()),
1332                    (StateEventType::RoomEncryption, "".to_owned()),
1333                    (StateEventType::RoomJoinRules, "".to_owned()),
1334                    (StateEventType::RoomName, "".to_owned()),
1335                ]
1336            );
1337            assert!(requested_required_states.for_rooms.is_empty());
1338        }
1339
1340        // Two room subscriptions.
1341        request.room_subscriptions.insert(room_id_1.to_owned(), {
1342            let mut room_subscription = v5::request::RoomSubscription::default();
1343
1344            room_subscription.required_state = vec![
1345                (StateEventType::RoomName, "".to_owned()),
1346                (StateEventType::RoomTopic, "".to_owned()),
1347            ];
1348
1349            room_subscription
1350        });
1351
1352        {
1353            let requested_required_states = RequestedRequiredStates::from(&request);
1354
1355            // Union of state events, all in `default`, still nothing in `for_rooms`.
1356            assert_eq!(
1357                requested_required_states.default,
1358                &[
1359                    (StateEventType::RoomAvatar, "".to_owned()),
1360                    (StateEventType::RoomEncryption, "".to_owned()),
1361                    (StateEventType::RoomJoinRules, "".to_owned()),
1362                    (StateEventType::RoomName, "".to_owned()),
1363                    (StateEventType::RoomTopic, "".to_owned()),
1364                ]
1365            );
1366        }
1367    }
1368
1369    #[async_test]
1370    async fn test_invite_after_leaving() {
1371        let user_id = user_id!("@alice:example.org");
1372        let room_id = room_id!("!test:example.org");
1373
1374        let client = logged_in_base_client(Some(user_id)).await;
1375
1376        let mut sync_builder = SyncResponseBuilder::new();
1377
1378        let response = sync_builder
1379            .add_left_room(
1380                LeftRoomBuilder::new(room_id).add_timeline_event(
1381                    EventFactory::new()
1382                        .member(user_id)
1383                        .membership(MembershipState::Leave)
1384                        .display_name("Alice")
1385                        .event_id(event_id!("$994173582443PhrSn:example.org")),
1386                ),
1387            )
1388            .build_sync_response();
1389        client.receive_sync_response(response).await.unwrap();
1390        assert_eq!(client.get_room(room_id).unwrap().state(), RoomState::Left);
1391
1392        let response = sync_builder
1393            .add_invited_room(InvitedRoomBuilder::new(room_id).add_state_event(
1394                StrippedStateTestEvent::Custom(json!({
1395                    "content": {
1396                        "displayname": "Alice",
1397                        "membership": "invite",
1398                    },
1399                    "event_id": "$143273582443PhrSn:example.org",
1400                    "origin_server_ts": 1432735824653u64,
1401                    "sender": "@example:example.org",
1402                    "state_key": user_id,
1403                    "type": "m.room.member",
1404                })),
1405            ))
1406            .build_sync_response();
1407        client.receive_sync_response(response).await.unwrap();
1408        assert_eq!(client.get_room(room_id).unwrap().state(), RoomState::Invited);
1409    }
1410
1411    #[async_test]
1412    async fn test_invite_displayname() {
1413        let user_id = user_id!("@alice:example.org");
1414        let room_id = room_id!("!ithpyNKDtmhneaTQja:example.org");
1415
1416        let client = logged_in_base_client(Some(user_id)).await;
1417
1418        let response = ruma_response_from_json(&json!({
1419            "next_batch": "asdkl;fjasdkl;fj;asdkl;f",
1420            "device_one_time_keys_count": {
1421                "signed_curve25519": 50u64
1422            },
1423            "device_unused_fallback_key_types": [
1424                "signed_curve25519"
1425            ],
1426            "rooms": {
1427                "invite": {
1428                    "!ithpyNKDtmhneaTQja:example.org": {
1429                        "invite_state": {
1430                            "events": [
1431                                {
1432                                    "content": {
1433                                        "creator": "@test:example.org",
1434                                        "room_version": "9"
1435                                    },
1436                                    "sender": "@test:example.org",
1437                                    "state_key": "",
1438                                    "type": "m.room.create"
1439                                },
1440                                {
1441                                    "content": {
1442                                        "join_rule": "invite"
1443                                    },
1444                                    "sender": "@test:example.org",
1445                                    "state_key": "",
1446                                    "type": "m.room.join_rules"
1447                                },
1448                                {
1449                                    "content": {
1450                                        "algorithm": "m.megolm.v1.aes-sha2"
1451                                    },
1452                                    "sender": "@test:example.org",
1453                                    "state_key": "",
1454                                    "type": "m.room.encryption"
1455                                },
1456                                {
1457                                    "content": {
1458                                        "avatar_url": "mxc://example.org/dcBBDwuWEUrjfrOchvkirUST",
1459                                        "displayname": "Kyra",
1460                                        "membership": "join"
1461                                    },
1462                                    "sender": "@test:example.org",
1463                                    "state_key": "@test:example.org",
1464                                    "type": "m.room.member"
1465                                },
1466                                {
1467                                    "content": {
1468                                        "avatar_url": "mxc://example.org/ABFEXSDrESxovWwEnCYdNcHT",
1469                                        "displayname": "alice",
1470                                        "is_direct": true,
1471                                        "membership": "invite"
1472                                    },
1473                                    "origin_server_ts": 1650878657984u64,
1474                                    "sender": "@test:example.org",
1475                                    "state_key": "@alice:example.org",
1476                                    "type": "m.room.member",
1477                                    "unsigned": {
1478                                        "age": 14u64
1479                                    },
1480                                    "event_id": "$fLDqltg9Puj-kWItLSFVHPGN4YkgpYQf2qImPzdmgrE"
1481                                }
1482                            ]
1483                        }
1484                    }
1485                }
1486            }
1487        }));
1488
1489        client.receive_sync_response(response).await.unwrap();
1490
1491        let room = client.get_room(room_id).expect("Room not found");
1492        assert_eq!(room.state(), RoomState::Invited);
1493        assert_eq!(
1494            room.compute_display_name().await.expect("fetching display name failed").into_inner(),
1495            RoomDisplayName::Calculated("Kyra".to_owned())
1496        );
1497    }
1498
1499    #[async_test]
1500    async fn test_deserialization_failure() {
1501        let user_id = user_id!("@alice:example.org");
1502        let room_id = room_id!("!ithpyNKDtmhneaTQja:example.org");
1503
1504        let client = BaseClient::new(
1505            StoreConfig::new("cross-process-store-locks-holder-name".to_owned()),
1506            ThreadingSupport::Disabled,
1507        );
1508        client
1509            .activate(
1510                SessionMeta { user_id: user_id.to_owned(), device_id: "FOOBAR".into() },
1511                RoomLoadSettings::default(),
1512                #[cfg(feature = "e2e-encryption")]
1513                None,
1514            )
1515            .await
1516            .unwrap();
1517
1518        let response = ruma_response_from_json(&json!({
1519            "next_batch": "asdkl;fjasdkl;fj;asdkl;f",
1520            "rooms": {
1521                "join": {
1522                    "!ithpyNKDtmhneaTQja:example.org": {
1523                        "state": {
1524                            "events": [
1525                                {
1526                                    "invalid": "invalid",
1527                                },
1528                                {
1529                                    "content": {
1530                                        "name": "The room name"
1531                                    },
1532                                    "event_id": "$143273582443PhrSn:example.org",
1533                                    "origin_server_ts": 1432735824653u64,
1534                                    "room_id": "!jEsUZKDJdhlrceRyVU:example.org",
1535                                    "sender": "@example:example.org",
1536                                    "state_key": "",
1537                                    "type": "m.room.name",
1538                                    "unsigned": {
1539                                        "age": 1234
1540                                    }
1541                                },
1542                            ]
1543                        }
1544                    }
1545                }
1546            }
1547        }));
1548
1549        client.receive_sync_response(response).await.unwrap();
1550        client
1551            .state_store()
1552            .get_state_event_static::<ruma::events::room::name::RoomNameEventContent>(room_id)
1553            .await
1554            .expect("Failed to fetch state event")
1555            .expect("State event not found")
1556            .deserialize()
1557            .expect("Failed to deserialize state event");
1558    }
1559
1560    #[async_test]
1561    async fn test_invited_members_arent_ignored() {
1562        let user_id = user_id!("@alice:example.org");
1563        let inviter_user_id = user_id!("@bob:example.org");
1564        let room_id = room_id!("!ithpyNKDtmhneaTQja:example.org");
1565
1566        let client = BaseClient::new(
1567            StoreConfig::new("cross-process-store-locks-holder-name".to_owned()),
1568            ThreadingSupport::Disabled,
1569        );
1570        client
1571            .activate(
1572                SessionMeta { user_id: user_id.to_owned(), device_id: "FOOBAR".into() },
1573                RoomLoadSettings::default(),
1574                #[cfg(feature = "e2e-encryption")]
1575                None,
1576            )
1577            .await
1578            .unwrap();
1579
1580        // Preamble: let the SDK know about the room.
1581        let mut sync_builder = SyncResponseBuilder::new();
1582        let response = sync_builder
1583            .add_joined_room(matrix_sdk_test::JoinedRoomBuilder::new(room_id))
1584            .build_sync_response();
1585        client.receive_sync_response(response).await.unwrap();
1586
1587        // When I process the result of a /members request that only contains an invited
1588        // member,
1589        let request = api::membership::get_member_events::v3::Request::new(room_id.to_owned());
1590
1591        let raw_member_event = json!({
1592            "content": {
1593                "avatar_url": "mxc://localhost/fewjilfewjil42",
1594                "displayname": "Invited Alice",
1595                "membership": "invite"
1596            },
1597            "event_id": "$151800140517rfvjc:localhost",
1598            "origin_server_ts": 151800140,
1599            "room_id": room_id,
1600            "sender": inviter_user_id,
1601            "state_key": user_id,
1602            "type": "m.room.member",
1603            "unsigned": {
1604                "age": 13374242,
1605            }
1606        });
1607        let response = api::membership::get_member_events::v3::Response::new(vec![Raw::from_json(
1608            to_raw_value(&raw_member_event).unwrap(),
1609        )]);
1610
1611        // It's correctly processed,
1612        client.receive_all_members(room_id, &request, &response).await.unwrap();
1613
1614        let room = client.get_room(room_id).unwrap();
1615
1616        // And I can get the invited member display name and avatar.
1617        let member = room.get_member(user_id).await.expect("ok").expect("exists");
1618
1619        assert_eq!(member.user_id(), user_id);
1620        assert_eq!(member.display_name().unwrap(), "Invited Alice");
1621        assert_eq!(member.avatar_url().unwrap().to_string(), "mxc://localhost/fewjilfewjil42");
1622    }
1623
1624    #[async_test]
1625    async fn test_reinvited_members_get_a_display_name() {
1626        let user_id = user_id!("@alice:example.org");
1627        let inviter_user_id = user_id!("@bob:example.org");
1628        let room_id = room_id!("!ithpyNKDtmhneaTQja:example.org");
1629
1630        let client = BaseClient::new(
1631            StoreConfig::new("cross-process-store-locks-holder-name".to_owned()),
1632            ThreadingSupport::Disabled,
1633        );
1634        client
1635            .activate(
1636                SessionMeta { user_id: user_id.to_owned(), device_id: "FOOBAR".into() },
1637                RoomLoadSettings::default(),
1638                #[cfg(feature = "e2e-encryption")]
1639                None,
1640            )
1641            .await
1642            .unwrap();
1643
1644        // Preamble: let the SDK know about the room, and that the invited user left it.
1645        let mut sync_builder = SyncResponseBuilder::new();
1646        let response = sync_builder
1647            .add_joined_room(matrix_sdk_test::JoinedRoomBuilder::new(room_id).add_state_event(
1648                StateTestEvent::Custom(json!({
1649                    "content": {
1650                        "avatar_url": null,
1651                        "displayname": null,
1652                        "membership": "leave"
1653                    },
1654                    "event_id": "$151803140217rkvjc:localhost",
1655                    "origin_server_ts": 151800139,
1656                    "room_id": room_id,
1657                    "sender": user_id,
1658                    "state_key": user_id,
1659                    "type": "m.room.member",
1660                })),
1661            ))
1662            .build_sync_response();
1663        client.receive_sync_response(response).await.unwrap();
1664
1665        // Now, say that the user has been re-invited.
1666        let request = api::membership::get_member_events::v3::Request::new(room_id.to_owned());
1667
1668        let raw_member_event = json!({
1669            "content": {
1670                "avatar_url": "mxc://localhost/fewjilfewjil42",
1671                "displayname": "Invited Alice",
1672                "membership": "invite"
1673            },
1674            "event_id": "$151800140517rfvjc:localhost",
1675            "origin_server_ts": 151800140,
1676            "room_id": room_id,
1677            "sender": inviter_user_id,
1678            "state_key": user_id,
1679            "type": "m.room.member",
1680            "unsigned": {
1681                "age": 13374242,
1682            }
1683        });
1684        let response = api::membership::get_member_events::v3::Response::new(vec![Raw::from_json(
1685            to_raw_value(&raw_member_event).unwrap(),
1686        )]);
1687
1688        // It's correctly processed,
1689        client.receive_all_members(room_id, &request, &response).await.unwrap();
1690
1691        let room = client.get_room(room_id).unwrap();
1692
1693        // And I can get the invited member display name and avatar.
1694        let member = room.get_member(user_id).await.expect("ok").expect("exists");
1695
1696        assert_eq!(member.user_id(), user_id);
1697        assert_eq!(member.display_name().unwrap(), "Invited Alice");
1698        assert_eq!(member.avatar_url().unwrap().to_string(), "mxc://localhost/fewjilfewjil42");
1699    }
1700
1701    #[async_test]
1702    async fn test_ignored_user_list_changes() {
1703        let user_id = user_id!("@alice:example.org");
1704        let client = BaseClient::new(
1705            StoreConfig::new("cross-process-store-locks-holder-name".to_owned()),
1706            ThreadingSupport::Disabled,
1707        );
1708
1709        client
1710            .activate(
1711                SessionMeta { user_id: user_id.to_owned(), device_id: "FOOBAR".into() },
1712                RoomLoadSettings::default(),
1713                #[cfg(feature = "e2e-encryption")]
1714                None,
1715            )
1716            .await
1717            .unwrap();
1718
1719        let mut subscriber = client.subscribe_to_ignore_user_list_changes();
1720        assert!(subscriber.next().now_or_never().is_none());
1721
1722        let f = EventFactory::new();
1723        let mut sync_builder = SyncResponseBuilder::new();
1724        let response = sync_builder
1725            .add_global_account_data(f.ignored_user_list([(*BOB).into()]))
1726            .build_sync_response();
1727        client.receive_sync_response(response).await.unwrap();
1728
1729        assert_let!(Some(ignored) = subscriber.next().await);
1730        assert_eq!(ignored, [BOB.to_string()]);
1731
1732        // Receive the same response.
1733        let response = sync_builder
1734            .add_global_account_data(f.ignored_user_list([(*BOB).into()]))
1735            .build_sync_response();
1736        client.receive_sync_response(response).await.unwrap();
1737
1738        // No changes in the ignored list.
1739        assert!(subscriber.next().now_or_never().is_none());
1740
1741        // Now remove Bob from the ignored list.
1742        let response =
1743            sync_builder.add_global_account_data(f.ignored_user_list([])).build_sync_response();
1744        client.receive_sync_response(response).await.unwrap();
1745
1746        assert_let!(Some(ignored) = subscriber.next().await);
1747        assert!(ignored.is_empty());
1748    }
1749
1750    #[async_test]
1751    async fn test_is_user_ignored() {
1752        let ignored_user_id = user_id!("@alice:example.org");
1753        let client = logged_in_base_client(None).await;
1754
1755        let mut sync_builder = SyncResponseBuilder::new();
1756        let f = EventFactory::new();
1757        let response = sync_builder
1758            .add_global_account_data(f.ignored_user_list([ignored_user_id.to_owned()]))
1759            .build_sync_response();
1760        client.receive_sync_response(response).await.unwrap();
1761
1762        assert!(client.is_user_ignored(ignored_user_id).await);
1763    }
1764
1765    #[async_test]
1766    async fn test_invite_details_are_set() {
1767        let user_id = user_id!("@alice:localhost");
1768        let client = logged_in_base_client(Some(user_id)).await;
1769        let invited_room_id = room_id!("!invited:localhost");
1770        let unknown_room_id = room_id!("!unknown:localhost");
1771
1772        let mut sync_builder = SyncResponseBuilder::new();
1773        let response = sync_builder
1774            .add_invited_room(InvitedRoomBuilder::new(invited_room_id))
1775            .build_sync_response();
1776        client.receive_sync_response(response).await.unwrap();
1777
1778        // Let us first check the initial state, we should have a room in the invite
1779        // state.
1780        let invited_room = client
1781            .get_room(invited_room_id)
1782            .expect("The sync should have created a room in the invited state");
1783
1784        assert_eq!(invited_room.state(), RoomState::Invited);
1785        assert!(invited_room.invite_acceptance_details().is_none());
1786
1787        // Now we join the room.
1788        let joined_room = client
1789            .room_joined(invited_room_id, Some(user_id.to_owned()))
1790            .await
1791            .expect("We should be able to mark a room as joined");
1792
1793        // Yup, we now have some invite details.
1794        assert_eq!(joined_room.state(), RoomState::Joined);
1795        assert_matches!(joined_room.invite_acceptance_details(), Some(details));
1796        assert_eq!(details.inviter, user_id);
1797
1798        // If we didn't know about the room before the join, we assume that there wasn't
1799        // an invite and we don't record the timestamp.
1800        assert!(client.get_room(unknown_room_id).is_none());
1801        let unknown_room = client
1802            .room_joined(unknown_room_id, Some(user_id.to_owned()))
1803            .await
1804            .expect("We should be able to mark a room as joined");
1805
1806        assert_eq!(unknown_room.state(), RoomState::Joined);
1807        assert!(unknown_room.invite_acceptance_details().is_none());
1808
1809        sync_builder.clear();
1810        let response =
1811            sync_builder.add_left_room(LeftRoomBuilder::new(invited_room_id)).build_sync_response();
1812        client.receive_sync_response(response).await.unwrap();
1813
1814        // Now that we left the room, we shouldn't have any details anymore.
1815        let left_room = client
1816            .get_room(invited_room_id)
1817            .expect("The sync should have created a room in the invited state");
1818
1819        assert_eq!(left_room.state(), RoomState::Left);
1820        assert!(left_room.invite_acceptance_details().is_none());
1821    }
1822}