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    collections::{BTreeMap, BTreeSet, HashMap},
25    fmt,
26    ops::Deref,
27    result::Result as StdResult,
28    str::Utf8Error,
29    sync::{Arc, RwLock as StdRwLock},
30};
31
32use eyeball_im::{Vector, VectorDiff};
33use futures_util::Stream;
34use once_cell::sync::OnceCell;
35
36#[cfg(any(test, feature = "testing"))]
37#[macro_use]
38pub mod integration_tests;
39mod observable_map;
40mod traits;
41
42#[cfg(feature = "e2e-encryption")]
43use matrix_sdk_crypto::store::{DynCryptoStore, IntoCryptoStore};
44pub use matrix_sdk_store_encryption::Error as StoreEncryptionError;
45use observable_map::ObservableMap;
46use ruma::{
47    events::{
48        presence::PresenceEvent,
49        receipt::ReceiptEventContent,
50        room::{member::StrippedRoomMemberEvent, redaction::SyncRoomRedactionEvent},
51        AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnyStrippedStateEvent,
52        AnySyncStateEvent, GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType,
53    },
54    serde::Raw,
55    EventId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
56};
57use tokio::sync::{broadcast, Mutex, RwLock};
58use tracing::warn;
59
60use crate::{
61    deserialized_responses::DisplayName,
62    event_cache::store as event_cache_store,
63    rooms::{normal::RoomInfoNotableUpdate, RoomInfo, RoomState},
64    MinimalRoomMemberEvent, Room, RoomStateFilter, SessionMeta,
65};
66
67pub(crate) mod ambiguity_map;
68mod memory_store;
69pub mod migration_helpers;
70mod send_queue;
71
72#[cfg(any(test, feature = "testing"))]
73pub use self::integration_tests::StateStoreIntegrationTests;
74pub use self::{
75    memory_store::MemoryStore,
76    send_queue::{
77        ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind,
78        FinishUploadThumbnailInfo, QueueWedgeError, QueuedRequest, QueuedRequestKind,
79        SentMediaInfo, SentRequestKey, SerializableEventContent,
80    },
81    traits::{
82        ComposerDraft, ComposerDraftType, DynStateStore, IntoStateStore, ServerCapabilities,
83        StateStore, StateStoreDataKey, StateStoreDataValue, StateStoreExt,
84    },
85};
86
87/// State store specific error type.
88#[derive(Debug, thiserror::Error)]
89pub enum StoreError {
90    /// An error happened in the underlying database backend.
91    #[error(transparent)]
92    Backend(Box<dyn std::error::Error + Send + Sync>),
93    /// An error happened while serializing or deserializing some data.
94    #[error(transparent)]
95    Json(#[from] serde_json::Error),
96    /// An error happened while deserializing a Matrix identifier, e.g. an user
97    /// id.
98    #[error(transparent)]
99    Identifier(#[from] ruma::IdParseError),
100    /// The store is locked with a passphrase and an incorrect passphrase was
101    /// given.
102    #[error("The store failed to be unlocked")]
103    StoreLocked,
104    /// An unencrypted store was tried to be unlocked with a passphrase.
105    #[error("The store is not encrypted but was tried to be opened with a passphrase")]
106    UnencryptedStore,
107    /// The store failed to encrypt or decrypt some data.
108    #[error("Error encrypting or decrypting data from the store: {0}")]
109    Encryption(#[from] StoreEncryptionError),
110
111    /// The store failed to encode or decode some data.
112    #[error("Error encoding or decoding data from the store: {0}")]
113    Codec(#[from] Utf8Error),
114
115    /// The database format has changed in a backwards incompatible way.
116    #[error(
117        "The database format changed in an incompatible way, current \
118        version: {0}, latest version: {1}"
119    )]
120    UnsupportedDatabaseVersion(usize, usize),
121    /// Redacting an event in the store has failed.
122    ///
123    /// This should never happen.
124    #[error("Redaction failed: {0}")]
125    Redaction(#[source] ruma::canonical_json::RedactionError),
126}
127
128impl StoreError {
129    /// Create a new [`Backend`][Self::Backend] error.
130    ///
131    /// Shorthand for `StoreError::Backend(Box::new(error))`.
132    #[inline]
133    pub fn backend<E>(error: E) -> Self
134    where
135        E: std::error::Error + Send + Sync + 'static,
136    {
137        Self::Backend(Box::new(error))
138    }
139}
140
141/// A `StateStore` specific result type.
142pub type Result<T, E = StoreError> = std::result::Result<T, E>;
143
144/// A state store wrapper for the SDK.
145///
146/// This adds additional higher level store functionality on top of a
147/// `StateStore` implementation.
148#[derive(Clone)]
149pub(crate) struct BaseStateStore {
150    pub(super) inner: Arc<DynStateStore>,
151    session_meta: Arc<OnceCell<SessionMeta>>,
152    /// The current sync token that should be used for the next sync call.
153    pub(super) sync_token: Arc<RwLock<Option<String>>>,
154    /// All rooms the store knows about.
155    rooms: Arc<StdRwLock<ObservableMap<OwnedRoomId, Room>>>,
156    /// A lock to synchronize access to the store, such that data by the sync is
157    /// never overwritten.
158    sync_lock: Arc<Mutex<()>>,
159}
160
161impl BaseStateStore {
162    /// Create a new store, wrapping the given `StateStore`
163    pub fn new(inner: Arc<DynStateStore>) -> Self {
164        Self {
165            inner,
166            session_meta: Default::default(),
167            sync_token: Default::default(),
168            rooms: Arc::new(StdRwLock::new(ObservableMap::new())),
169            sync_lock: Default::default(),
170        }
171    }
172
173    /// Get access to the syncing lock.
174    pub fn sync_lock(&self) -> &Mutex<()> {
175        &self.sync_lock
176    }
177
178    /// Set the `SessionMeta` into [`BaseStateStore::session_meta`].
179    ///
180    /// # Panics
181    ///
182    /// Panics if called twice.
183    pub(crate) fn set_session_meta(&self, session_meta: SessionMeta) {
184        self.session_meta.set(session_meta).expect("`SessionMeta` was already set");
185    }
186
187    /// Loads rooms from the `StateStore` into [`BaseStateStore::rooms`].
188    pub(crate) async fn load_rooms(
189        &self,
190        user_id: &UserId,
191        room_info_notable_update_sender: &broadcast::Sender<RoomInfoNotableUpdate>,
192    ) -> Result<()> {
193        let room_infos = self.load_and_migrate_room_infos().await?;
194
195        let mut rooms = self.rooms.write().unwrap();
196
197        for room_info in room_infos {
198            let new_room = Room::restore(
199                user_id,
200                self.inner.clone(),
201                room_info,
202                room_info_notable_update_sender.clone(),
203            );
204            let new_room_id = new_room.room_id().to_owned();
205
206            rooms.insert(new_room_id, new_room);
207        }
208
209        Ok(())
210    }
211
212    /// Load room infos from the [`StateStore`] and applies migrations onto
213    /// them.
214    async fn load_and_migrate_room_infos(&self) -> Result<Vec<RoomInfo>> {
215        let mut room_infos = self.inner.get_room_infos().await?;
216        let mut migrated_room_infos = Vec::with_capacity(room_infos.len());
217
218        for room_info in room_infos.iter_mut() {
219            if room_info.apply_migrations(self.inner.clone()).await {
220                migrated_room_infos.push(room_info.clone());
221            }
222        }
223
224        if !migrated_room_infos.is_empty() {
225            let changes = StateChanges {
226                room_infos: migrated_room_infos
227                    .into_iter()
228                    .map(|room_info| (room_info.room_id.clone(), room_info))
229                    .collect(),
230                ..Default::default()
231            };
232
233            if let Err(error) = self.inner.save_changes(&changes).await {
234                warn!("Failed to save migrated room infos: {error}");
235            }
236        }
237
238        Ok(room_infos)
239    }
240
241    /// Load sync token from the [`StateStore`], and put it in
242    /// [`BaseStateStore::sync_token`].
243    pub(crate) async fn load_sync_token(&self) -> Result<()> {
244        let token =
245            self.get_kv_data(StateStoreDataKey::SyncToken).await?.and_then(|s| s.into_sync_token());
246        *self.sync_token.write().await = token;
247
248        Ok(())
249    }
250
251    /// Restore the session meta, sync token and rooms from an existing
252    /// [`BaseStateStore`].
253    #[cfg(any(feature = "e2e-encryption", test))]
254    pub(crate) async fn derive_from_other(
255        &self,
256        other: &Self,
257        room_info_notable_update_sender: &broadcast::Sender<RoomInfoNotableUpdate>,
258    ) -> Result<()> {
259        let Some(session_meta) = other.session_meta.get() else {
260            return Ok(());
261        };
262
263        self.load_rooms(&session_meta.user_id, room_info_notable_update_sender).await?;
264        self.load_sync_token().await?;
265        self.set_session_meta(session_meta.clone());
266
267        Ok(())
268    }
269
270    /// The current [`SessionMeta`] containing our user ID and device ID.
271    pub fn session_meta(&self) -> Option<&SessionMeta> {
272        self.session_meta.get()
273    }
274
275    /// Get all the rooms this store knows about.
276    pub fn rooms(&self) -> Vec<Room> {
277        self.rooms.read().unwrap().iter().cloned().collect()
278    }
279
280    /// Get all the rooms this store knows about, filtered by state.
281    pub fn rooms_filtered(&self, filter: RoomStateFilter) -> Vec<Room> {
282        self.rooms
283            .read()
284            .unwrap()
285            .iter()
286            .filter(|room| filter.matches(room.state()))
287            .cloned()
288            .collect()
289    }
290
291    /// Get a stream of all the rooms changes, in addition to the existing
292    /// rooms.
293    pub fn rooms_stream(&self) -> (Vector<Room>, impl Stream<Item = Vec<VectorDiff<Room>>>) {
294        self.rooms.read().unwrap().stream()
295    }
296
297    /// Get the room with the given room id.
298    pub fn room(&self, room_id: &RoomId) -> Option<Room> {
299        self.rooms.read().unwrap().get(room_id).cloned()
300    }
301
302    /// Check if a room exists.
303    pub(crate) fn room_exists(&self, room_id: &RoomId) -> bool {
304        self.rooms.read().unwrap().get(room_id).is_some()
305    }
306
307    /// Lookup the `Room` for the given `RoomId`, or create one, if it didn't
308    /// exist yet in the store
309    pub fn get_or_create_room(
310        &self,
311        room_id: &RoomId,
312        room_type: RoomState,
313        room_info_notable_update_sender: broadcast::Sender<RoomInfoNotableUpdate>,
314    ) -> Room {
315        let user_id =
316            &self.session_meta.get().expect("Creating room while not being logged in").user_id;
317
318        self.rooms
319            .write()
320            .unwrap()
321            .get_or_create(room_id, || {
322                Room::new(
323                    user_id,
324                    self.inner.clone(),
325                    room_id,
326                    room_type,
327                    room_info_notable_update_sender,
328                )
329            })
330            .clone()
331    }
332
333    /// Forget the room with the given room ID.
334    ///
335    /// # Arguments
336    ///
337    /// * `room_id` - The id of the room that should be forgotten.
338    pub(crate) async fn forget_room(&self, room_id: &RoomId) -> Result<()> {
339        self.inner.remove_room(room_id).await?;
340        self.rooms.write().unwrap().remove(room_id);
341        Ok(())
342    }
343}
344
345#[cfg(not(tarpaulin_include))]
346impl fmt::Debug for BaseStateStore {
347    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
348        f.debug_struct("Store")
349            .field("inner", &self.inner)
350            .field("session_meta", &self.session_meta)
351            .field("sync_token", &self.sync_token)
352            .field("rooms", &self.rooms)
353            .finish_non_exhaustive()
354    }
355}
356
357impl Deref for BaseStateStore {
358    type Target = DynStateStore;
359
360    fn deref(&self) -> &Self::Target {
361        self.inner.deref()
362    }
363}
364
365/// Store state changes and pass them to the StateStore.
366#[derive(Clone, Debug, Default)]
367pub struct StateChanges {
368    /// The sync token that relates to this update.
369    pub sync_token: Option<String>,
370    /// A mapping of event type string to `AnyBasicEvent`.
371    pub account_data: BTreeMap<GlobalAccountDataEventType, Raw<AnyGlobalAccountDataEvent>>,
372    /// A mapping of `UserId` to `PresenceEvent`.
373    pub presence: BTreeMap<OwnedUserId, Raw<PresenceEvent>>,
374
375    /// A mapping of `RoomId` to a map of users and their
376    /// `MinimalRoomMemberEvent`.
377    pub profiles: BTreeMap<OwnedRoomId, BTreeMap<OwnedUserId, MinimalRoomMemberEvent>>,
378
379    /// A mapping of room profiles to delete.
380    ///
381    /// These are deleted *before* other room profiles are inserted.
382    pub profiles_to_delete: BTreeMap<OwnedRoomId, Vec<OwnedUserId>>,
383
384    /// A mapping of `RoomId` to a map of event type string to a state key and
385    /// `AnySyncStateEvent`.
386    pub state:
387        BTreeMap<OwnedRoomId, BTreeMap<StateEventType, BTreeMap<String, Raw<AnySyncStateEvent>>>>,
388    /// A mapping of `RoomId` to a map of event type string to `AnyBasicEvent`.
389    pub room_account_data:
390        BTreeMap<OwnedRoomId, BTreeMap<RoomAccountDataEventType, Raw<AnyRoomAccountDataEvent>>>,
391
392    /// A map of `OwnedRoomId` to `RoomInfo`.
393    pub room_infos: BTreeMap<OwnedRoomId, RoomInfo>,
394
395    /// A map of `RoomId` to `ReceiptEventContent`.
396    pub receipts: BTreeMap<OwnedRoomId, ReceiptEventContent>,
397
398    /// A map of `RoomId` to maps of `OwnedEventId` to be redacted by
399    /// `SyncRoomRedactionEvent`.
400    pub redactions: BTreeMap<OwnedRoomId, BTreeMap<OwnedEventId, Raw<SyncRoomRedactionEvent>>>,
401
402    /// A mapping of `RoomId` to a map of event type to a map of state key to
403    /// `AnyStrippedStateEvent`.
404    pub stripped_state: BTreeMap<
405        OwnedRoomId,
406        BTreeMap<StateEventType, BTreeMap<String, Raw<AnyStrippedStateEvent>>>,
407    >,
408
409    /// A map from room id to a map of a display name and a set of user ids that
410    /// share that display name in the given room.
411    pub ambiguity_maps: BTreeMap<OwnedRoomId, HashMap<DisplayName, BTreeSet<OwnedUserId>>>,
412}
413
414impl StateChanges {
415    /// Create a new `StateChanges` struct with the given sync_token.
416    pub fn new(sync_token: String) -> Self {
417        Self { sync_token: Some(sync_token), ..Default::default() }
418    }
419
420    /// Update the `StateChanges` struct with the given `PresenceEvent`.
421    pub fn add_presence_event(&mut self, event: PresenceEvent, raw_event: Raw<PresenceEvent>) {
422        self.presence.insert(event.sender, raw_event);
423    }
424
425    /// Update the `StateChanges` struct with the given `RoomInfo`.
426    pub fn add_room(&mut self, room: RoomInfo) {
427        self.room_infos.insert(room.room_id.clone(), room);
428    }
429
430    /// Update the `StateChanges` struct with the given room with a new
431    /// `AnyBasicEvent`.
432    pub fn add_room_account_data(
433        &mut self,
434        room_id: &RoomId,
435        event: AnyRoomAccountDataEvent,
436        raw_event: Raw<AnyRoomAccountDataEvent>,
437    ) {
438        self.room_account_data
439            .entry(room_id.to_owned())
440            .or_default()
441            .insert(event.event_type(), raw_event);
442    }
443
444    /// Update the `StateChanges` struct with the given room with a new
445    /// `StrippedMemberEvent`.
446    pub fn add_stripped_member(
447        &mut self,
448        room_id: &RoomId,
449        user_id: &UserId,
450        event: Raw<StrippedRoomMemberEvent>,
451    ) {
452        self.stripped_state
453            .entry(room_id.to_owned())
454            .or_default()
455            .entry(StateEventType::RoomMember)
456            .or_default()
457            .insert(user_id.into(), event.cast());
458    }
459
460    /// Update the `StateChanges` struct with the given room with a new
461    /// `AnySyncStateEvent`.
462    pub fn add_state_event(
463        &mut self,
464        room_id: &RoomId,
465        event: AnySyncStateEvent,
466        raw_event: Raw<AnySyncStateEvent>,
467    ) {
468        self.state
469            .entry(room_id.to_owned())
470            .or_default()
471            .entry(event.event_type())
472            .or_default()
473            .insert(event.state_key().to_owned(), raw_event);
474    }
475
476    /// Redact an event in the room
477    pub fn add_redaction(
478        &mut self,
479        room_id: &RoomId,
480        redacted_event_id: &EventId,
481        redaction: Raw<SyncRoomRedactionEvent>,
482    ) {
483        self.redactions
484            .entry(room_id.to_owned())
485            .or_default()
486            .insert(redacted_event_id.to_owned(), redaction);
487    }
488
489    /// Update the `StateChanges` struct with the given room with a new
490    /// `Receipts`.
491    pub fn add_receipts(&mut self, room_id: &RoomId, event: ReceiptEventContent) {
492        self.receipts.insert(room_id.to_owned(), event);
493    }
494}
495
496/// Configuration for the various stores.
497///
498/// By default, this always includes a state store and an event cache store.
499/// When the `e2e-encryption` feature is enabled, this also includes a crypto
500/// store.
501///
502/// # Examples
503///
504/// ```
505/// # use matrix_sdk_base::store::StoreConfig;
506/// #
507/// let store_config =
508///     StoreConfig::new("cross-process-store-locks-holder-name".to_owned());
509/// ```
510#[derive(Clone)]
511pub struct StoreConfig {
512    #[cfg(feature = "e2e-encryption")]
513    pub(crate) crypto_store: Arc<DynCryptoStore>,
514    pub(crate) state_store: Arc<DynStateStore>,
515    pub(crate) event_cache_store: event_cache_store::EventCacheStoreLock,
516    cross_process_store_locks_holder_name: String,
517}
518
519#[cfg(not(tarpaulin_include))]
520impl fmt::Debug for StoreConfig {
521    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> StdResult<(), fmt::Error> {
522        fmt.debug_struct("StoreConfig").finish()
523    }
524}
525
526impl StoreConfig {
527    /// Create a new default `StoreConfig`.
528    ///
529    /// To learn more about `cross_process_store_locks_holder_name`, please read
530    /// [`CrossProcessStoreLock::new`](matrix_sdk_common::store_locks::CrossProcessStoreLock::new).
531    #[must_use]
532    pub fn new(cross_process_store_locks_holder_name: String) -> Self {
533        Self {
534            #[cfg(feature = "e2e-encryption")]
535            crypto_store: matrix_sdk_crypto::store::MemoryStore::new().into_crypto_store(),
536            state_store: Arc::new(MemoryStore::new()),
537            event_cache_store: event_cache_store::EventCacheStoreLock::new(
538                event_cache_store::MemoryStore::new(),
539                cross_process_store_locks_holder_name.clone(),
540            ),
541            cross_process_store_locks_holder_name,
542        }
543    }
544
545    /// Set a custom implementation of a `CryptoStore`.
546    ///
547    /// The crypto store must be opened before being set.
548    #[cfg(feature = "e2e-encryption")]
549    pub fn crypto_store(mut self, store: impl IntoCryptoStore) -> Self {
550        self.crypto_store = store.into_crypto_store();
551        self
552    }
553
554    /// Set a custom implementation of a `StateStore`.
555    pub fn state_store(mut self, store: impl IntoStateStore) -> Self {
556        self.state_store = store.into_state_store();
557        self
558    }
559
560    /// Set a custom implementation of an `EventCacheStore`.
561    pub fn event_cache_store<S>(mut self, event_cache_store: S) -> Self
562    where
563        S: event_cache_store::IntoEventCacheStore,
564    {
565        self.event_cache_store = event_cache_store::EventCacheStoreLock::new(
566            event_cache_store,
567            self.cross_process_store_locks_holder_name.clone(),
568        );
569        self
570    }
571}
572
573#[cfg(test)]
574mod tests {
575    use std::sync::Arc;
576
577    use matrix_sdk_test::async_test;
578    use ruma::{owned_device_id, owned_user_id};
579    use tokio::sync::broadcast;
580
581    use super::{BaseStateStore, MemoryStore};
582    use crate::SessionMeta;
583
584    #[async_test]
585    async fn test_set_session_meta() {
586        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
587
588        let session_meta = SessionMeta {
589            user_id: owned_user_id!("@mnt_io:matrix.org"),
590            device_id: owned_device_id!("HELLOYOU"),
591        };
592
593        assert!(store.session_meta.get().is_none());
594
595        store.set_session_meta(session_meta.clone());
596
597        assert_eq!(store.session_meta.get(), Some(&session_meta));
598    }
599
600    #[async_test]
601    #[should_panic]
602    async fn test_set_session_meta_twice() {
603        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
604
605        let session_meta = SessionMeta {
606            user_id: owned_user_id!("@mnt_io:matrix.org"),
607            device_id: owned_device_id!("HELLOYOU"),
608        };
609
610        store.set_session_meta(session_meta.clone());
611        // Kaboom.
612        store.set_session_meta(session_meta);
613    }
614
615    #[async_test]
616    async fn test_derive_from_other() {
617        // The first store.
618        let other = BaseStateStore::new(Arc::new(MemoryStore::new()));
619
620        let session_meta = SessionMeta {
621            user_id: owned_user_id!("@mnt_io:matrix.org"),
622            device_id: owned_device_id!("HELLOYOU"),
623        };
624        let (room_info_notable_update_sender, _) = broadcast::channel(1);
625
626        other.set_session_meta(session_meta.clone());
627
628        // Derive another store.
629        let store = BaseStateStore::new(Arc::new(MemoryStore::new()));
630        store.derive_from_other(&other, &room_info_notable_update_sender).await.unwrap();
631
632        assert_eq!(store.session_meta.get(), Some(&session_meta));
633    }
634}
matrix_sdk_base/store/mod.rs

matrix_sdk_base/store/
mod.rs
