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

matrix_sdk_base/event_cache/store/
mod.rs
