matrix_sdk_base/event_cache/store/
mod.rs

1// Copyright 2024 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 event cache stores holds events when the cache was
16//! activated to save bandwidth at the cost of increased storage space usage.
17//!
18//! Implementing the `EventCacheStore` trait, you can plug any storage backend
19//! into the event cache for the actual storage. By default this brings an
20//! in-memory store.
21
22use std::{fmt, ops::Deref, str::Utf8Error, sync::Arc};
23
24#[cfg(any(test, feature = "testing"))]
25#[macro_use]
26pub mod integration_tests;
27mod memory_store;
28mod traits;
29
30use matrix_sdk_common::cross_process_lock::{
31    CrossProcessLock, CrossProcessLockConfig, CrossProcessLockError, CrossProcessLockGeneration,
32    CrossProcessLockGuard, MappedCrossProcessLockState, TryLock,
33};
34pub use matrix_sdk_store_encryption::Error as StoreEncryptionError;
35use ruma::{OwnedEventId, events::AnySyncTimelineEvent, serde::Raw};
36use tracing::trace;
37
38#[cfg(any(test, feature = "testing"))]
39pub use self::integration_tests::EventCacheStoreIntegrationTests;
40pub use self::{
41    memory_store::MemoryStore,
42    traits::{DEFAULT_CHUNK_CAPACITY, DynEventCacheStore, EventCacheStore, IntoEventCacheStore},
43};
44
45/// The high-level public type to represent an `EventCacheStore` lock.
46#[derive(Clone)]
47pub struct EventCacheStoreLock {
48    /// The inner cross process lock that is used to lock the `EventCacheStore`.
49    cross_process_lock: Arc<CrossProcessLock<LockableEventCacheStore>>,
50
51    /// The store itself.
52    ///
53    /// That's the only place where the store exists.
54    store: Arc<DynEventCacheStore>,
55}
56
57#[cfg(not(tarpaulin_include))]
58impl fmt::Debug for EventCacheStoreLock {
59    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
60        formatter.debug_struct("EventCacheStoreLock").finish_non_exhaustive()
61    }
62}
63
64impl EventCacheStoreLock {
65    /// Create a new lock around the [`EventCacheStore`].
66    ///
67    /// The `cross_process_lock_config` argument controls whether we need to
68    /// hold the cross process lock or not.
69    pub fn new<S>(store: S, cross_process_lock_config: CrossProcessLockConfig) -> Self
70    where
71        S: IntoEventCacheStore,
72    {
73        let store = store.into_event_cache_store();
74
75        let cross_process_lock = Arc::new(CrossProcessLock::new(
76            LockableEventCacheStore(store.clone()),
77            "default".to_owned(),
78            cross_process_lock_config,
79        ));
80        Self { cross_process_lock, store }
81    }
82
83    /// Acquire a spin lock (see [`CrossProcessLock::spin_lock`]).
84    pub async fn lock(&self) -> Result<EventCacheStoreLockState, CrossProcessLockError> {
85        Ok(self.cross_process_lock.spin_lock(None).await??.map(|cross_process_lock_guard| {
86            EventCacheStoreLockGuard { cross_process_lock_guard, store: self.store.clone() }
87        }))
88    }
89}
90
91/// The equivalent of [`CrossProcessLockState`] but for the [`EventCacheStore`].
92///
93/// [`CrossProcessLockState`]: matrix_sdk_common::cross_process_lock::CrossProcessLockState
94pub type EventCacheStoreLockState = MappedCrossProcessLockState<EventCacheStoreLockGuard>;
95
96/// An RAII implementation of a “scoped lock” of an [`EventCacheStoreLock`].
97/// When this structure is dropped (falls out of scope), the lock will be
98/// unlocked.
99#[derive(Clone)]
100pub struct EventCacheStoreLockGuard {
101    /// The cross process lock guard.
102    #[allow(unused)]
103    cross_process_lock_guard: CrossProcessLockGuard,
104
105    /// A reference to the store.
106    store: Arc<DynEventCacheStore>,
107}
108
109impl EventCacheStoreLockGuard {
110    /// Forward to [`CrossProcessLockGuard::clear_dirty`].
111    ///
112    /// This is an associated method to avoid colliding with the [`Deref`]
113    /// implementation.
114    pub fn clear_dirty(this: &Self) {
115        this.cross_process_lock_guard.clear_dirty();
116    }
117
118    /// Force to [`CrossProcessLockGuard::is_dirty`].
119    pub fn is_dirty(this: &Self) -> bool {
120        this.cross_process_lock_guard.is_dirty()
121    }
122}
123
124#[cfg(not(tarpaulin_include))]
125impl fmt::Debug for EventCacheStoreLockGuard {
126    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
127        formatter.debug_struct("EventCacheStoreLockGuard").finish_non_exhaustive()
128    }
129}
130
131impl Deref for EventCacheStoreLockGuard {
132    type Target = DynEventCacheStore;
133
134    fn deref(&self) -> &Self::Target {
135        self.store.as_ref()
136    }
137}
138
139/// Event cache store specific error type.
140#[derive(Debug, thiserror::Error)]
141pub enum EventCacheStoreError {
142    /// An error happened in the underlying database backend.
143    #[error(transparent)]
144    Backend(Box<dyn std::error::Error + Send + Sync>),
145
146    /// The store is locked with a passphrase and an incorrect passphrase
147    /// was given.
148    #[error("The event cache store failed to be unlocked")]
149    Locked,
150
151    /// An unencrypted store was tried to be unlocked with a passphrase.
152    #[error("The event cache store is not encrypted but tried to be opened with a passphrase")]
153    Unencrypted,
154
155    /// The store failed to encrypt or decrypt some data.
156    #[error("Error encrypting or decrypting data from the event cache store: {0}")]
157    Encryption(#[from] StoreEncryptionError),
158
159    /// The store failed to encode or decode some data.
160    #[error("Error encoding or decoding data from the event cache store: {0}")]
161    Codec(#[from] Utf8Error),
162
163    /// The store failed to serialize or deserialize some data.
164    #[error("Error serializing or deserializing data from the event cache store: {0}")]
165    Serialization(#[from] serde_json::Error),
166
167    /// The database format has changed in a backwards incompatible way.
168    #[error(
169        "The database format of the event cache store changed in an incompatible way, \
170         current version: {0}, latest version: {1}"
171    )]
172    UnsupportedDatabaseVersion(usize, usize),
173
174    /// The store contains invalid data.
175    #[error("The store contains invalid data: {details}")]
176    InvalidData {
177        /// Details why the data contained in the store was invalid.
178        details: String,
179    },
180}
181
182impl EventCacheStoreError {
183    /// Create a new [`Backend`][Self::Backend] error.
184    ///
185    /// Shorthand for `EventCacheStoreError::Backend(Box::new(error))`.
186    #[inline]
187    pub fn backend<E>(error: E) -> Self
188    where
189        E: std::error::Error + Send + Sync + 'static,
190    {
191        Self::Backend(Box::new(error))
192    }
193}
194
195impl From<EventCacheStoreError> for CrossProcessLockError {
196    fn from(value: EventCacheStoreError) -> Self {
197        Self::TryLock(Box::new(value))
198    }
199}
200
201/// An `EventCacheStore` specific result type.
202pub type Result<T, E = EventCacheStoreError> = std::result::Result<T, E>;
203
204/// A type that wraps the [`EventCacheStore`] but implements [`TryLock`] to
205/// make it usable inside the cross process lock.
206#[derive(Clone, Debug)]
207struct LockableEventCacheStore(Arc<DynEventCacheStore>);
208
209impl TryLock for LockableEventCacheStore {
210    type LockError = EventCacheStoreError;
211
212    async fn try_lock(
213        &self,
214        lease_duration_ms: u32,
215        key: &str,
216        holder: &str,
217    ) -> std::result::Result<Option<CrossProcessLockGeneration>, Self::LockError> {
218        self.0.try_take_leased_lock(lease_duration_ms, key, holder).await
219    }
220}
221
222/// Helper to extract the relation information from an event.
223///
224/// If the event isn't in relation to another event, then this will return
225/// `None`. Otherwise, returns both the event id this event relates to, and the
226/// kind of relation as a string (e.g. `m.replace`).
227pub fn extract_event_relation(event: &Raw<AnySyncTimelineEvent>) -> Option<(OwnedEventId, String)> {
228    #[derive(serde::Deserialize)]
229    struct RelatesTo {
230        event_id: OwnedEventId,
231        rel_type: String,
232    }
233
234    #[derive(serde::Deserialize)]
235    struct EventContent {
236        #[serde(rename = "m.relates_to")]
237        rel: Option<RelatesTo>,
238    }
239
240    match event.get_field::<EventContent>("content") {
241        Ok(event_content) => {
242            event_content.and_then(|c| c.rel).map(|rel| (rel.event_id, rel.rel_type))
243        }
244        Err(err) => {
245            trace!("when extracting relation data from an event: {err}");
246            None
247        }
248    }
249}
matrix_sdk_base/event_cache/store/mod.rs

matrix_sdk_base/event_cache/store/
mod.rs
