matrix_sdk_base/store/
mod.rs

1// Copyright 2021 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 the specific language governing permissions and
13// limitations under the License.
14
15//! The state store holds the overall state for rooms, users and their
16//! profiles and their timelines. It is an overall cache for faster access
17//! and convenience- accessible through `Store`.
18//!
19//! Implementing the `StateStore` trait, you can plug any storage backend
20//! into the store for the actual storage. By default this brings an in-memory
21//! store.
22
23use std::{
24    borrow::Borrow,
25    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
26    fmt,
27    ops::Deref,
28    result::Result as StdResult,
29    str::{FromStr, Utf8Error},
30    sync::{Arc, OnceLock, RwLock as StdRwLock},
31};
32
33use eyeball_im::{Vector, VectorDiff};
34use futures_util::Stream;
35use matrix_sdk_common::ROOM_VERSION_RULES_FALLBACK;
36
37#[cfg(any(test, feature = "testing"))]
38#[macro_use]
39pub mod integration_tests;
40mod observable_map;
41mod traits;
42
43use matrix_sdk_common::{cross_process_lock::CrossProcessLockConfig, locks::Mutex as SyncMutex};
44#[cfg(feature = "e2e-encryption")]
45use matrix_sdk_crypto::store::{DynCryptoStore, IntoCryptoStore};
46pub use matrix_sdk_store_encryption::Error as StoreEncryptionError;
47use observable_map::ObservableMap;
48use ruma::{
49    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
50    events::{
51        AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnyStrippedStateEvent,
52        AnySyncStateEvent, EmptyStateKey, GlobalAccountDataEventType, RedactContent,
53        RedactedStateEventContent, RoomAccountDataEventType, StateEventType, StaticEventContent,
54        StaticStateEventContent, StrippedStateEvent, SyncStateEvent,
55        presence::PresenceEvent,
56        receipt::ReceiptEventContent,
57        room::{
58            create::RoomCreateEventContent,
59            member::{RoomMemberEventContent, StrippedRoomMemberEvent},
60            power_levels::{RoomPowerLevels, RoomPowerLevelsEventContent},
61            redaction::SyncRoomRedactionEvent,
62        },
63    },
64    profile::UserProfileUpdate,
65    serde::Raw,
66};
67use serde::de::DeserializeOwned;
68use tokio::sync::{Mutex, RwLock, broadcast};
69use tracing::warn;
70pub use traits::compare_thread_subscription_bump_stamps;
71
72use crate::{
73    MinimalRoomMemberEvent, Room, RoomCreateWithCreatorEventContent, RoomStateFilter, SessionMeta,
74    deserialized_responses::DisplayName,
75    event_cache::store as event_cache_store,
76    media::store as media_store,
77    room::{RoomInfo, RoomInfoNotableUpdate, RoomState},
78};
79
80pub(crate) mod ambiguity_map;
81mod avatar_cache;
82mod memory_store;
83pub mod migration_helpers;
84mod send_queue;
85
86pub use avatar_cache::AvatarCache;
87
88#[cfg(any(test, feature = "testing"))]
89pub use self::integration_tests::StateStoreIntegrationTests;
90#[cfg(feature = "unstable-msc4274")]
91pub use self::send_queue::{AccumulatedSentMediaInfo, FinishGalleryItemInfo};
92pub use self::{
93    memory_store::MemoryStore,
94    send_queue::{
95        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind,
96        FinishUploadThumbnailInfo, QueueWedgeError, QueuedRequest, QueuedRequestKind,
97        SentMediaInfo, SentRequestKey, SerializableEventContent,
98    },
99    traits::{
100        ComposerDraft, ComposerDraftType, DraftAttachment, DraftAttachmentContent, DraftThumbnail,
101        DynStateStore, IncorrectMutexGuardError, IntoStateStore, SaveLockedStateStore, StateStore,
102        StateStoreDataKey, StateStoreDataValue, StateStoreExt, SupportedVersionsResponse,
103        ThreadSubscriptionCatchupToken, WellKnownResponse,
104    },
105};
106
107/// State store specific error type.
108#[derive(Debug, thiserror::Error)]
109pub enum StoreError {
110    /// An error happened in the underlying database backend.
111    #[error(transparent)]
112    Backend(Box<dyn std::error::Error + Send + Sync>),
113
114    /// An error happened while serializing or deserializing some data.
115    #[error(transparent)]
116    Json(#[from] serde_json::Error),
117
118    /// An error happened while deserializing a Matrix identifier, e.g. an user
119    /// id.
120    #[error(transparent)]
121    Identifier(#[from] ruma::IdParseError),
122
123    /// The store is locked with a passphrase and an incorrect passphrase was
124    /// given.
125    #[error("The store failed to be unlocked")]
126    StoreLocked,
127
128    /// An unencrypted store was tried to be unlocked with a passphrase.
129    #[error("The store is not encrypted but was tried to be opened with a passphrase")]
130    UnencryptedStore,
131
132    /// The store failed to encrypt or decrypt some data.
133    #[error("Error encrypting or decrypting data from the store: {0}")]
134    Encryption(#[from] StoreEncryptionError),
135
136    /// The store failed to encode or decode some data.
137    #[error("Error encoding or decoding data from the store: {0}")]
138    Codec(#[from] Utf8Error),
139
140    /// The database format has changed in a backwards incompatible way.
141    #[error(
142        "The database format changed in an incompatible way, current \
143        version: {0}, latest version: {1}"
144    )]
145    UnsupportedDatabaseVersion(usize, usize),
146
147    /// Redacting an event in the store has failed.
148    ///
149    /// This should never happen.
150    #[error("Redaction failed: {0}")]
151    Redaction(#[source] ruma::canonical_json::CanonicalJsonFieldError),
152
153    /// The store contains invalid data.
154    #[error("The store contains invalid data: {details}")]
155    InvalidData {
156        /// Details about which data is invalid, and how.
157        details: String,
158    },
159}
160
161impl StoreError {
162    /// Create a new [`Backend`][Self::Backend] error.
163    ///
164    /// Shorthand for `StoreError::Backend(Box::new(error))`.
165    #[inline]
166    pub fn backend<E>(error: E) -> Self
167    where
168        E: std::error::Error + Send + Sync + 'static,
169    {
170        Self::Backend(Box::new(error))
171    }
172}
173
174/// A `StateStore` specific result type.
175pub type Result<T, E = StoreError> = std::result::Result<T, E>;
176
177/// A state store wrapper for the SDK.
178///
179/// This adds additional higher level store functionality on top of a
180/// `StateStore` implementation.
181#[derive(Clone)]
182pub(crate) struct BaseStateStore {
183    pub(super) inner: SaveLockedStateStore,
184    session_meta: Arc<OnceLock<SessionMeta>>,
185    room_load_settings: Arc<RwLock<RoomLoadSettings>>,
186
187    /// A sender that is used to communicate changes to room information. Each
188    /// tick contains the room ID and the reasons that have generated this tick.
189    pub(crate) room_info_notable_update_sender: broadcast::Sender<RoomInfoNotableUpdate>,
190
191    /// The current sync token that should be used for the next sync call.
192    pub(super) sync_token: Arc<RwLock<Option<String>>>,
193
194    /// All rooms the store knows about.
195    rooms: Arc<StdRwLock<ObservableMap<OwnedRoomId, Room>>>,
196
197    /// Which rooms have already logged a log line about missing room info, in
198    /// the context of response processors?
199    pub(crate) already_logged_missing_room: Arc<SyncMutex<HashSet<OwnedRoomId>>>,
200}
201
202impl BaseStateStore {
203    /// Create a new store, wrapping the given `StateStore`
204    pub fn new(inner: Arc<DynStateStore>) -> Self {
205        // Create the channel to receive `RoomInfoNotableUpdate`.
206        //
207        // Let's consider the channel will receive 5 updates for 100 rooms
208        // maximum. This is unrealistic in practise, as the sync mechanism is
209        // pretty unlikely to trigger such amount of updates, it's a safe value.
210        //
211        // Also, note that it must not be zero, because (i) it will panic, (ii)
212        // a new user has no room, but can create rooms; remember that the
213        // channel's capacity is immutable.
214        let (room_info_notable_update_sender, _room_info_notable_update_receiver) =
215            broadcast::channel(500);
216
217        Self {
218            inner: SaveLockedStateStore::new(inner),
219            session_meta: Default::default(),
220            room_load_settings: Default::default(),
221            room_info_notable_update_sender,
222            sync_token: Default::default(),
223            rooms: Arc::new(StdRwLock::new(ObservableMap::new())),
224            already_logged_missing_room: Default::default(),
225        }
226    }
227
228    /// Get access to the syncing lock.
229    pub fn lock(&self) -> &Mutex<()> {
230        self.inner.lock()
231    }
232
233    /// Set the [`SessionMeta`] into [`BaseStateStore::session_meta`].
234    ///
235    /// # Panics
236    ///
237    /// Panics if called twice.
238    pub(crate) fn set_session_meta(&self, session_meta: SessionMeta) {
239        self.session_meta.set(session_meta).expect("`SessionMeta` was already set");
240    }
241
242    /// Loads rooms from the given [`DynStateStore`] (in
243    /// [`BaseStateStore::new`]) into [`BaseStateStore::rooms`].
244    pub(crate) async fn load_rooms(
245        &self,
246        user_id: &UserId,
247        room_load_settings: RoomLoadSettings,
248    ) -> Result<()> {
249        *self.room_load_settings.write().await = room_load_settings.clone();
250
251        let room_infos = self.load_and_migrate_room_infos(room_load_settings).await?;
252
253        let mut rooms = self.rooms.write().unwrap();
254
255        for room_info in room_infos {
256            let new_room = Room::restore(
257                user_id,
258                self.inner.clone(),
259                room_info,
260                self.room_info_notable_update_sender.clone(),
261            );
262            let new_room_id = new_room.room_id().to_owned();
263
264            rooms.insert(new_room_id, new_room);
265        }
266
267        Ok(())
268    }
269
270    /// Load room infos from the [`StateStore`] and applies migrations onto
271    /// them.
272    async fn load_and_migrate_room_infos(
273        &self,
274        room_load_settings: RoomLoadSettings,
275    ) -> Result<Vec<RoomInfo>> {
276        let mut room_infos = self.inner.get_room_infos(&room_load_settings).await?;
277        let mut migrated_room_infos = Vec::with_capacity(room_infos.len());
278
279        for room_info in room_infos.iter_mut() {
280            if room_info.apply_migrations(self.inner.clone()).await {
281                migrated_room_infos.push(room_info.clone());
282            }
283        }
284
285        if !migrated_room_infos.is_empty() {
286            let changes = StateChanges {
287                room_infos: migrated_room_infos
288                    .into_iter()
289                    .map(|room_info| (room_info.room_id.clone(), room_info))
290                    .collect(),
291                ..Default::default()
292            };
293
294            if let Err(error) = self.inner.save_changes(&changes).await {
295                warn!("Failed to save migrated room infos: {error}");
296            }
297        }
298
299        Ok(room_infos)
300    }
301
302    /// Load sync token from the [`StateStore`], and put it in
303    /// [`BaseStateStore::sync_token`].
304    pub(crate) async fn load_sync_token(&self) -> Result<()> {
305        let token =
306            self.get_kv_data(StateStoreDataKey::SyncToken).await?.and_then(|s| s.into_sync_token());
307        *self.sync_token.write().await = token;
308
309        Ok(())
310    }
311
312    /// Restore the session meta, sync token and rooms from an existing
313    /// [`BaseStateStore`].
314    #[cfg(any(feature = "e2e-encryption", test))]
315    pub(crate) async fn derive_from_other(&self, other: &Self) -> Result<()> {
316        let Some(session_meta) = other.session_meta.get() else {
317            return Ok(());
318        };
319
320        let room_load_settings = other.room_load_settings.read().await.clone();
321
322        self.load_rooms(&session_meta.user_id, room_load_settings).await?;
323        self.load_sync_token().await?;
324        self.set_session_meta(session_meta.clone());
325
326        Ok(())
327    }
328
329    /// The current [`SessionMeta`] containing our user ID and device ID.
330    pub fn session_meta(&self) -> Option<&SessionMeta> {
331        self.session_meta.get()
332    }
333
334    /// Get all the rooms this store knows about.
335    pub fn rooms(&self) -> Vec<Room> {
336        self.rooms.read().unwrap().iter().cloned().collect()
337    }
338
339    /// Get all the rooms this store knows about, filtered by state.
340    pub fn rooms_filtered(&self, filter: RoomStateFilter) -> Vec<Room> {
341        self.rooms
342            .read()
343            .unwrap()
344            .iter()
345            .filter(|room| filter.matches(room.state()))
346            .cloned()
347            .collect()
348    }
349
350    /// Get a stream of all the rooms changes, in addition to the existing
351    /// rooms.
352    pub fn rooms_stream(
353        &self,
354    ) -> (Vector<Room>, impl Stream<Item = Vec<VectorDiff<Room>>> + use<>) {
355        self.rooms.read().unwrap().stream()
356    }
357
358    /// Get the room with the given room id.
359    pub fn room(&self, room_id: &RoomId) -> Option<Room> {
360        self.rooms.read().unwrap().get(room_id).cloned()
361    }
362
363    /// Check if a room exists.
364    pub(crate) fn room_exists(&self, room_id: &RoomId) -> bool {
365        self.rooms.read().unwrap().get(room_id).is_some()
366    }
367
368    /// Lookup the `Room` for the given `RoomId`, or create one, if it didn't
369    /// exist yet in the store
370    pub fn get_or_create_room(&self, room_id: &RoomId, room_state: RoomState) -> Room {
371        let user_id =
372            &self.session_meta.get().expect("Creating room while not being logged in").user_id;
373
374        self.rooms
375            .write()
376            .unwrap()
377            .get_or_create(room_id, || {
378                Room::new(
379                    user_id,
380                    self.inner.clone(),
381                    room_id,
382                    room_state,
383                    self.room_info_notable_update_sender.clone(),
384                )
385            })
386            .clone()
387    }
388
389    /// Forget the room with the given room ID.
390    ///
391    /// # Arguments
392    ///
393    /// * `room_id` - The id of the room that should be forgotten.
394    pub(crate) async fn forget_room(&self, room_id: &RoomId) -> Result<()> {
395        self.inner.remove_room(room_id).await?;
396        self.rooms.write().unwrap().remove(room_id);
397        Ok(())
398    }
399}
400
401#[cfg(not(tarpaulin_include))]
402impl fmt::Debug for BaseStateStore {
403    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
404        f.debug_struct("Store")
405            .field("inner", &self.inner)
406            .field("session_meta", &self.session_meta)
407            .field("sync_token", &self.sync_token)
408            .field("rooms", &self.rooms)
409            .finish_non_exhaustive()
410    }
411}
412
413impl Deref for BaseStateStore {
414    type Target = SaveLockedStateStore;
415
416    fn deref(&self) -> &Self::Target {
417        &self.inner
418    }
419}
420
421/// Configure how many rooms will be restored when restoring the session with
422/// `BaseStateStore::load_rooms`.
423///
424/// <div class="warning">
425///
426/// # ⚠️ be careful!
427///
428/// When loading a single room with [`RoomLoadSettings::One`], the in-memory
429/// state may not reflect the store state (in the databases). Thus, when one
430/// will get a room that exists in the store state but _not_ in the in-memory
431/// state, it will be created from scratch and, when saved, will override the
432/// data in the store state (in the databases). This can lead to weird
433/// behaviours.
434///
435/// This option is expected to be used as follows:
436///
437/// 1. Create a `BaseStateStore` with a [`StateStore`] based on SQLite for
438///    example,
439/// 2. Restore a session and load one room from the [`StateStore`] (in the case
440///    of dealing with a notification for example),
441/// 3. Derive the `BaseStateStore`, with `BaseStateStore::derive_from_other`,
442///    into another one with an in-memory [`StateStore`], such as
443///    [`MemoryStore`],
444/// 4. Work on this derived `BaseStateStore`.
445///
446/// Now, all operations happen in the [`MemoryStore`], not on the original store
447/// (SQLite in this example), thus protecting original data.
448///
449/// From a higher-level point of view, this is what
450/// [`BaseClient::clone_with_in_memory_state_store`] does.
451///
452/// </div>
453///
454/// [`BaseClient::clone_with_in_memory_state_store`]: crate::BaseClient::clone_with_in_memory_state_store
455#[derive(Clone, Debug, Default)]
456pub enum RoomLoadSettings {
457    /// Load all rooms from the [`StateStore`] into the in-memory state store
458    /// `BaseStateStore`.
459    ///
460    /// This is the default variant.
461    #[default]
462    All,
463
464    /// Load a single room from the [`StateStore`] into the in-memory state
465    /// store `BaseStateStore`.
466    ///
467    /// Please, be careful with this option. Read the documentation of
468    /// [`RoomLoadSettings`].
469    One(OwnedRoomId),
470}
471
472/// The subscription status of a thread.
473///
474/// We keep unsubscriptions in the database, because we need the bumpstamp
475/// information (in `ThreadSubscription`) to be around to order subscriptions
476/// and unsubscriptions.
477#[derive(Clone, Copy, Debug, PartialEq, Eq)]
478pub enum ThreadSubscriptionStatus {
479    /// The user is subscribed to the related thread.
480    Subscribed {
481        /// Whether the subscription was made automatically by a client, not by
482        /// manual user choice.
483        automatic: bool,
484    },
485
486    /// The user has been unsubscribed to the related thread.
487    Unsubscribed,
488}
489
490impl FromStr for ThreadSubscriptionStatus {
491    type Err = ();
492
493    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
494        match s {
495            "automatic" => Ok(ThreadSubscriptionStatus::Subscribed { automatic: true }),
496            "manual" => Ok(ThreadSubscriptionStatus::Subscribed { automatic: false }),
497            "unsubscribed" => Ok(ThreadSubscriptionStatus::Unsubscribed),
498            _ => Err(()),
499        }
500    }
501}
502
503impl ThreadSubscriptionStatus {
504    /// Represent the status as a static string ref, for it to be stored into a
505    /// persistent format.
506    ///
507    /// Note: this is serialized in some databases implementations, so make sure
508    /// to not change it lightly, and keep it in sync with [`Self::from_str`].
509    pub fn as_str(&self) -> &'static str {
510        match self {
511            ThreadSubscriptionStatus::Subscribed { automatic } => {
512                if *automatic {
513                    "automatic"
514                } else {
515                    "manual"
516                }
517            }
518            ThreadSubscriptionStatus::Unsubscribed => "unsubscribed",
519        }
520    }
521}
522
523/// A thread subscription, as saved in the state store.
524#[derive(Clone, Copy, Debug, PartialEq, Eq)]
525pub struct StoredThreadSubscription {
526    /// Current status of the subscription.
527    pub status: ThreadSubscriptionStatus,
528
529    /// An optional bump stamp, as defined in the MSC; the higher the value, the
530    /// most recent the thread subscription information is, and should be
531    /// remembered.
532    ///
533    /// If not set, this means it's a user-provided thread subscription, for
534    /// which we're waiting validation from a server (e.g. through a remote echo
535    /// via sync).
536    pub bump_stamp: Option<u64>,
537}
538
539/// Store state changes and pass them to the StateStore.
540#[derive(Clone, Debug, Default)]
541pub struct StateChanges {
542    /// The sync token that relates to this update.
543    pub sync_token: Option<String>,
544    /// A mapping of event type string to `AnyBasicEvent`.
545    pub account_data: BTreeMap<GlobalAccountDataEventType, Raw<AnyGlobalAccountDataEvent>>,
546    /// A mapping of `UserId` to `PresenceEvent`.
547    pub presence: BTreeMap<OwnedUserId, Raw<PresenceEvent>>,
548
549    /// A mapping of `RoomId` to a map of users and their
550    /// `MinimalRoomMemberEvent`.
551    pub profiles: BTreeMap<OwnedRoomId, BTreeMap<OwnedUserId, MinimalRoomMemberEvent>>,
552
553    /// A mapping of room profiles to delete.
554    ///
555    /// These are deleted *before* other room profiles are inserted.
556    pub profiles_to_delete: BTreeMap<OwnedRoomId, Vec<OwnedUserId>>,
557
558    /// A mapping of `RoomId` to a map of event type string to a state key and
559    /// `AnySyncStateEvent`.
560    pub state:
561        BTreeMap<OwnedRoomId, BTreeMap<StateEventType, BTreeMap<String, Raw<AnySyncStateEvent>>>>,
562    /// A mapping of `RoomId` to a map of event type string to `AnyBasicEvent`.
563    pub room_account_data:
564        BTreeMap<OwnedRoomId, BTreeMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>>,
565
566    /// A map of `OwnedRoomId` to `RoomInfo`.
567    pub room_infos: BTreeMap<OwnedRoomId, RoomInfo>,
568
569    /// A map of `RoomId` to `ReceiptEventContent`.
570    pub receipts: BTreeMap<OwnedRoomId, ReceiptEventContent>,
571
572    /// A map of `RoomId` to maps of `OwnedEventId` to be redacted by
573    /// `SyncRoomRedactionEvent`.
574    pub redactions: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, Raw<SyncRoomRedactionEvent>>>,
575
576    /// A mapping of `RoomId` to a map of event type to a map of state key to
577    /// `StrippedState`.
578    pub stripped_state: BTreeMap<
579        OwnedRoomId,
580        BTreeMap<StateEventType, BTreeMap<String, Raw<AnyStrippedStateEvent>>>,
581    >,
582
583    /// A map from room id to a map of a display name and a set of user ids that
584    /// share that display name in the given room.
585    pub ambiguity_maps: BTreeMap<OwnedRoomId, HashMap<DisplayName, BTreeSet<OwnedUserId>>>,
586
587    /// A mapping of `UserId` to global `UserProfile` updates, as defined in
588    /// MSC4262.
589    ///
590    /// These follow the MSC4262 update pattern: fields with an explicit `null`
591    /// value are removed, while fields that aren't present are left unchanged.
592    pub global_profiles: BTreeMap<OwnedUserId, UserProfileUpdate>,
593
594    /// Users whose global profiles should be marked as unavailable, mapped to
595    /// the time their profile was last queried.
596    ///
597    /// Saving a global profile for one of these users from
598    /// [`StateChanges::global_profiles`] removes this mark.
599    pub unavailable_global_profiles: BTreeMap<OwnedUserId, MilliSecondsSinceUnixEpoch>,
600}
601
602impl StateChanges {
603    /// Create a new `StateChanges` struct with the given sync_token.
604    pub fn new(sync_token: String) -> Self {
605        Self { sync_token: Some(sync_token), ..Default::default() }
606    }
607
608    /// Update the `StateChanges` struct with the given `PresenceEvent`.
609    pub fn add_presence_event(&mut self, event: PresenceEvent, raw_event: Raw<PresenceEvent>) {
610        self.presence.insert(event.sender, raw_event);
611    }
612
613    /// Update the `StateChanges` struct with the given `RoomInfo`.
614    pub fn add_room(&mut self, room: RoomInfo) {
615        self.room_infos.insert(room.room_id.clone(), room);
616    }
617
618    /// Update the `StateChanges` struct with the given room with a new
619    /// `AnyBasicEvent`.
620    pub fn add_room_account_data(
621        &mut self,
622        room_id: &RoomId,
623        event: AnyRoomAccountDataEvent,
624        raw_event: Raw<AnyRoomAccountDataEvent>,
625    ) {
626        self.room_account_data
627            .entry(room_id.to_owned())
628            .or_default()
629            .insert(event.event_type(), raw_event);
630    }
631
632    /// Update the `StateChanges` struct with the given room with a new
633    /// `StrippedMemberEvent`.
634    pub fn add_stripped_member(
635        &mut self,
636        room_id: &RoomId,
637        user_id: &UserId,
638        event: Raw<StrippedRoomMemberEvent>,
639    ) {
640        self.stripped_state
641            .entry(room_id.to_owned())
642            .or_default()
643            .entry(StateEventType::RoomMember)
644            .or_default()
645            .insert(user_id.into(), event.cast());
646    }
647
648    /// Update the `StateChanges` struct with the given room with a new
649    /// `AnySyncStateEvent`.
650    pub fn add_state_event(
651        &mut self,
652        room_id: &RoomId,
653        event: AnySyncStateEvent,
654        raw_event: Raw<AnySyncStateEvent>,
655    ) {
656        self.state
657            .entry(room_id.to_owned())
658            .or_default()
659            .entry(event.event_type())
660            .or_default()
661            .insert(event.state_key().to_owned(), raw_event);
662    }
663
664    /// Redact an event in the room
665    pub fn add_redaction(
666        &mut self,
667        room_id: &RoomId,
668        redacted_event_id: &EventId,
669        redaction: Raw<SyncRoomRedactionEvent>,
670    ) {
671        self.redactions
672            .entry(room_id.to_owned())
673            .or_default()
674            .insert(redacted_event_id.to_owned(), redaction);
675    }
676
677    /// Update the `StateChanges` struct with the given room with a new
678    /// `Receipts`.
679    pub fn add_receipts(&mut self, room_id: &RoomId, event: ReceiptEventContent) {
680        self.receipts.insert(room_id.to_owned(), event);
681    }
682
683    /// Get a specific state event of statically-known type with the given state
684    /// key in the given room, if it is present in the `state` map of these
685    /// `StateChanges`.
686    pub(crate) fn state_static_for_key<C, K>(
687        &self,
688        room_id: &RoomId,
689        state_key: &K,
690    ) -> Option<&Raw<SyncStateEvent<C>>>
691    where
692        C: StaticEventContent<IsPrefix = ruma::events::False>
693            + StaticStateEventContent
694            + RedactContent,
695        C::Redacted: RedactedStateEventContent,
696        C::StateKey: Borrow<K>,
697        K: AsRef<str> + ?Sized,
698    {
699        self.state
700            .get(room_id)?
701            .get(&C::TYPE.into())?
702            .get(state_key.as_ref())
703            .map(Raw::cast_ref_unchecked)
704    }
705
706    /// Get a specific stripped state event of statically-known type with the
707    /// given state key in the given room, if it is present in the
708    /// `stripped_state` map of these `StateChanges`.
709    pub(crate) fn stripped_state_static_for_key<C, K>(
710        &self,
711        room_id: &RoomId,
712        state_key: &K,
713    ) -> Option<&Raw<StrippedStateEvent<C::PossiblyRedacted>>>
714    where
715        C: StaticEventContent<IsPrefix = ruma::events::False> + StaticStateEventContent,
716        C::StateKey: Borrow<K>,
717        K: AsRef<str> + ?Sized,
718    {
719        self.stripped_state
720            .get(room_id)?
721            .get(&C::TYPE.into())?
722            .get(state_key.as_ref())
723            .map(Raw::cast_ref_unchecked)
724    }
725
726    /// Get a specific state event of statically-known type with the given state
727    /// key in the given room, if it is present in the `state` or
728    /// `stripped_state` map of these `StateChanges` and it deserializes
729    /// successfully.
730    pub(crate) fn any_state_static_for_key<C, K>(
731        &self,
732        room_id: &RoomId,
733        state_key: &K,
734    ) -> Option<StrippedStateEvent<C::PossiblyRedacted>>
735    where
736        C: StaticEventContent<IsPrefix = ruma::events::False>
737            + StaticStateEventContent
738            + RedactContent,
739        C::Redacted: RedactedStateEventContent,
740        C::PossiblyRedacted: StaticEventContent + DeserializeOwned,
741        C::StateKey: Borrow<K>,
742        K: AsRef<str> + ?Sized,
743    {
744        self.state_static_for_key::<C, K>(room_id, state_key)
745            .map(Raw::cast_ref)
746            .or_else(|| self.stripped_state_static_for_key::<C, K>(room_id, state_key))?
747            .deserialize()
748            .ok()
749    }
750
751    /// Get the member for the given user in the given room from an event
752    /// contained in these `StateChanges`, if any.
753    pub(crate) fn member(
754        &self,
755        room_id: &RoomId,
756        user_id: &UserId,
757    ) -> Option<StrippedRoomMemberEvent> {
758        self.any_state_static_for_key::<RoomMemberEventContent, _>(room_id, user_id)
759    }
760
761    /// Get the create event for the given room from an event contained in these
762    /// `StateChanges`, if any.
763    pub(crate) fn create(&self, room_id: &RoomId) -> Option<RoomCreateWithCreatorEventContent> {
764        self.any_state_static_for_key::<RoomCreateEventContent, _>(room_id, &EmptyStateKey)
765            .map(|event| {
766                RoomCreateWithCreatorEventContent::from_event_content(event.content, event.sender)
767            })
768            // Fallback to the content in the room info.
769            .or_else(|| self.room_infos.get(room_id)?.create().cloned())
770    }
771
772    /// Get the power levels for the given room from an event contained in these
773    /// `StateChanges`, if any.
774    pub(crate) fn power_levels(&self, room_id: &RoomId) -> Option<RoomPowerLevels> {
775        let power_levels_content = self
776            .any_state_static_for_key::<RoomPowerLevelsEventContent, _>(room_id, &EmptyStateKey)?;
777
778        let create_content = self.create(room_id)?;
779        let rules = create_content.room_version.rules().unwrap_or(ROOM_VERSION_RULES_FALLBACK);
780        let creators = create_content.creators();
781
782        Some(power_levels_content.power_levels(&rules.authorization, creators))
783    }
784}
785
786/// Configuration for the various stores.
787///
788/// By default, this always includes a state store and an event cache store.
789/// When the `e2e-encryption` feature is enabled, this also includes a crypto
790/// store.
791///
792/// # Examples
793///
794/// ```
795/// # use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
796/// # use matrix_sdk_base::store::StoreConfig;
797/// #
798/// let store_config = StoreConfig::new(CrossProcessLockConfig::MultiProcess {
799///     holder_name: "cross-process-store-locks-holder-name".to_owned(),
800/// });
801/// ```
802#[derive(Clone)]
803pub struct StoreConfig {
804    #[cfg(feature = "e2e-encryption")]
805    pub(crate) crypto_store: Arc<DynCryptoStore>,
806    pub(crate) state_store: Arc<DynStateStore>,
807    pub(crate) event_cache_store: event_cache_store::EventCacheStoreLock,
808    pub(crate) media_store: media_store::MediaStoreLock,
809    cross_process_lock_config: CrossProcessLockConfig,
810}
811
812#[cfg(not(tarpaulin_include))]
813impl fmt::Debug for StoreConfig {
814    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> StdResult<(), fmt::Error> {
815        fmt.debug_struct("StoreConfig").finish()
816    }
817}
818
819impl StoreConfig {
820    /// Create a new default `StoreConfig`.
821    ///
822    /// To learn more about `cross_process_lock_config`, please read
823    /// [`CrossProcessLock::new`](matrix_sdk_common::cross_process_lock::CrossProcessLock::new).
824    #[must_use]
825    pub fn new(cross_process_lock_config: CrossProcessLockConfig) -> Self {
826        Self {
827            #[cfg(feature = "e2e-encryption")]
828            crypto_store: matrix_sdk_crypto::store::MemoryStore::new().into_crypto_store(),
829            state_store: Arc::new(MemoryStore::new()),
830            event_cache_store: event_cache_store::EventCacheStoreLock::new(
831                event_cache_store::MemoryStore::new(),
832                cross_process_lock_config.clone(),
833            ),
834            media_store: media_store::MediaStoreLock::new(
835                media_store::MemoryMediaStore::new(),
836                cross_process_lock_config.clone(),
837            ),
838            cross_process_lock_config,
839        }
840    }
841
842    /// Set a custom implementation of a `CryptoStore`.
843    ///
844    /// The crypto store must be opened before being set.
845    #[cfg(feature = "e2e-encryption")]
846    pub fn crypto_store(mut self, store: impl IntoCryptoStore) -> Self {
847        self.crypto_store = store.into_crypto_store();
848        self
849    }
850
851    /// Set a custom implementation of a `StateStore`.
852    pub fn state_store(mut self, store: impl IntoStateStore) -> Self {
853        self.state_store = store.into_state_store();
854        self
855    }
856
857    /// Set a custom implementation of an `EventCacheStore`.
858    pub fn event_cache_store<S>(mut self, event_cache_store: S) -> Self
859    where
860        S: event_cache_store::IntoEventCacheStore,
861    {
862        self.event_cache_store = event_cache_store::EventCacheStoreLock::new(
863            event_cache_store,
864            self.cross_process_lock_config.clone(),
865        );
866        self
867    }
868
869    /// Set a custom implementation of an `MediaStore`.
870    pub fn media_store<S>(mut self, media_store: S) -> Self
871    where
872        S: media_store::IntoMediaStore,
873    {
874        self.media_store =
875            media_store::MediaStoreLock::new(media_store, self.cross_process_lock_config.clone());
876        self
877    }
878}
879
880#[cfg(test)]
881mod tests {
882    use std::{ops::Not, sync::Arc};
883
884    use assert_matches::assert_matches;
885    use matrix_sdk_test::async_test;
886    use ruma::{owned_device_id, owned_user_id, room_id, user_id};
887
888    use super::{BaseStateStore, MemoryStore, RoomLoadSettings};
889    use crate::{RoomInfo, RoomState, SessionMeta, StateChanges, StateStore};
890
891    #[async_test]
892    async fn test_set_session_meta() {
893        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
894
895        let session_meta = SessionMeta {
896            user_id: owned_user_id!("@mnt_io:matrix.org"),
897            device_id: owned_device_id!("HELLOYOU"),
898        };
899
900        assert!(store.session_meta.get().is_none());
901
902        store.set_session_meta(session_meta.clone());
903
904        assert_eq!(store.session_meta.get(), Some(&session_meta));
905    }
906
907    #[async_test]
908    #[should_panic]
909    async fn test_set_session_meta_twice() {
910        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
911
912        let session_meta = SessionMeta {
913            user_id: owned_user_id!("@mnt_io:matrix.org"),
914            device_id: owned_device_id!("HELLOYOU"),
915        };
916
917        store.set_session_meta(session_meta.clone());
918        // Kaboom.
919        store.set_session_meta(session_meta);
920    }
921
922    #[async_test]
923    async fn test_derive_from_other() {
924        // The first store.
925        let other = BaseStateStore::new(Arc::new(MemoryStore::new()));
926
927        let session_meta = SessionMeta {
928            user_id: owned_user_id!("@mnt_io:matrix.org"),
929            device_id: owned_device_id!("HELLOYOU"),
930        };
931        let room_id_0 = room_id!("!r0");
932
933        other
934            .load_rooms(&session_meta.user_id, RoomLoadSettings::One(room_id_0.to_owned()))
935            .await
936            .unwrap();
937        other.set_session_meta(session_meta.clone());
938
939        // Derive another store.
940        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
941        store.derive_from_other(&other).await.unwrap();
942
943        // `SessionMeta` is derived.
944        assert_eq!(store.session_meta.get(), Some(&session_meta));
945        // `RoomLoadSettings` is derived.
946        assert_matches!(*store.room_load_settings.read().await, RoomLoadSettings::One(ref room_id) => {
947            assert_eq!(room_id, room_id_0);
948        });
949
950        // The `RoomInfoNotableUpdate` is not derived. Every one has its own
951        // channel.
952        assert!(
953            store
954                .room_info_notable_update_sender
955                .same_channel(&other.room_info_notable_update_sender)
956                .not()
957        );
958    }
959
960    #[test]
961    fn test_room_load_settings_default() {
962        assert_matches!(RoomLoadSettings::default(), RoomLoadSettings::All);
963    }
964
965    #[async_test]
966    async fn test_load_all_rooms() {
967        let room_id_0 = room_id!("!r0");
968        let room_id_1 = room_id!("!r1");
969        let user_id = user_id!("@mnt_io:matrix.org");
970
971        let memory_state_store = Arc::new(MemoryStore::new());
972
973        // Initial state.
974        {
975            let store = BaseStateStore::new(memory_state_store.clone());
976            let mut changes = StateChanges::default();
977            changes.add_room(RoomInfo::new(room_id_0, RoomState::Joined));
978            changes.add_room(RoomInfo::new(room_id_1, RoomState::Joined));
979
980            store.inner.save_changes(&changes).await.unwrap();
981        }
982
983        // Check a `BaseStateStore` is able to load all rooms.
984        {
985            let store = BaseStateStore::new(memory_state_store.clone());
986
987            // Default value.
988            assert_matches!(*store.room_load_settings.read().await, RoomLoadSettings::All);
989
990            // Load rooms.
991            store.load_rooms(user_id, RoomLoadSettings::All).await.unwrap();
992
993            // Check the last room load settings.
994            assert_matches!(*store.room_load_settings.read().await, RoomLoadSettings::All);
995
996            // Check the loaded rooms.
997            let mut rooms = store.rooms();
998            rooms.sort_by(|a, b| a.room_id().cmp(b.room_id()));
999
1000            assert_eq!(rooms.len(), 2);
1001
1002            assert_eq!(rooms[0].room_id(), room_id_0);
1003            assert_eq!(rooms[0].own_user_id(), user_id);
1004
1005            assert_eq!(rooms[1].room_id(), room_id_1);
1006            assert_eq!(rooms[1].own_user_id(), user_id);
1007        }
1008    }
1009
1010    #[async_test]
1011    async fn test_load_one_room() {
1012        let room_id_0 = room_id!("!r0");
1013        let room_id_1 = room_id!("!r1");
1014        let user_id = user_id!("@mnt_io:matrix.org");
1015
1016        let memory_state_store = Arc::new(MemoryStore::new());
1017
1018        // Initial state.
1019        {
1020            let store = BaseStateStore::new(memory_state_store.clone());
1021            let mut changes = StateChanges::default();
1022            changes.add_room(RoomInfo::new(room_id_0, RoomState::Joined));
1023            changes.add_room(RoomInfo::new(room_id_1, RoomState::Joined));
1024
1025            store.inner.save_changes(&changes).await.unwrap();
1026        }
1027
1028        // Check a `BaseStateStore` is able to load one room.
1029        {
1030            let store = BaseStateStore::new(memory_state_store.clone());
1031
1032            // Default value.
1033            assert_matches!(*store.room_load_settings.read().await, RoomLoadSettings::All);
1034
1035            // Load rooms.
1036            store.load_rooms(user_id, RoomLoadSettings::One(room_id_1.to_owned())).await.unwrap();
1037
1038            // Check the last room load settings.
1039            assert_matches!(
1040                *store.room_load_settings.read().await,
1041                RoomLoadSettings::One(ref room_id) => {
1042                    assert_eq!(room_id, room_id_1);
1043                }
1044            );
1045
1046            // Check the loaded rooms.
1047            let rooms = store.rooms();
1048            assert_eq!(rooms.len(), 1);
1049
1050            assert_eq!(rooms[0].room_id(), room_id_1);
1051            assert_eq!(rooms[0].own_user_id(), user_id);
1052        }
1053    }
1054}
matrix_sdk_base/store/mod.rs

matrix_sdk_base/store/
mod.rs
