matrix_sdk_base/event_cache/store/
mod.rs1use 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#[derive(Clone)]
47pub struct EventCacheStoreLock {
48 cross_process_lock: Arc<CrossProcessLock<LockableEventCacheStore>>,
50
51 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 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 pub async fn close(&self) -> Result<(), EventCacheStoreError> {
85 self.store.close().await
86 }
87
88 pub async fn reopen(&self) -> Result<(), EventCacheStoreError> {
90 self.store.reopen().await
91 }
92
93 pub async fn lock(&self) -> Result<EventCacheStoreLockState, CrossProcessLockError> {
95 Ok(self.cross_process_lock.spin_lock(None).await??.map(|cross_process_lock_guard| {
96 EventCacheStoreLockGuard { cross_process_lock_guard, store: self.store.clone() }
97 }))
98 }
99}
100
101pub type EventCacheStoreLockState = MappedCrossProcessLockState<EventCacheStoreLockGuard>;
105
106#[derive(Clone)]
110pub struct EventCacheStoreLockGuard {
111 #[allow(unused)]
113 cross_process_lock_guard: CrossProcessLockGuard,
114
115 store: Arc<DynEventCacheStore>,
117}
118
119impl EventCacheStoreLockGuard {
120 pub fn clear_dirty(this: &Self) {
125 this.cross_process_lock_guard.clear_dirty();
126 }
127
128 pub fn is_dirty(this: &Self) -> bool {
133 this.cross_process_lock_guard.is_dirty()
134 }
135}
136
137#[cfg(not(tarpaulin_include))]
138impl fmt::Debug for EventCacheStoreLockGuard {
139 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
140 formatter.debug_struct("EventCacheStoreLockGuard").finish_non_exhaustive()
141 }
142}
143
144impl Deref for EventCacheStoreLockGuard {
145 type Target = DynEventCacheStore;
146
147 fn deref(&self) -> &Self::Target {
148 self.store.as_ref()
149 }
150}
151
152#[derive(Clone, Debug, thiserror::Error)]
154pub enum EventCacheStoreError {
155 #[error(transparent)]
157 Backend(Arc<dyn std::error::Error + Send + Sync>),
158
159 #[error("The event cache store failed to be unlocked")]
162 Locked,
163
164 #[error("The event cache store is not encrypted but tried to be opened with a passphrase")]
166 Unencrypted,
167
168 #[error("Error encrypting or decrypting data from the event cache store: {0}")]
170 Encryption(#[from] Arc<StoreEncryptionError>),
171
172 #[error("Error encoding or decoding data from the event cache store: {0}")]
174 Codec(#[from] Utf8Error),
175
176 #[error("Error serializing or deserializing data from the event cache store: {0}")]
178 Serialization(#[from] Arc<serde_json::Error>),
179
180 #[error(
182 "The database format of the event cache store changed in an incompatible way, \
183 current version: {0}, latest version: {1}"
184 )]
185 UnsupportedDatabaseVersion(usize, usize),
186
187 #[error("The store contains invalid data: {details}")]
189 InvalidData {
190 details: String,
192 },
193}
194
195impl EventCacheStoreError {
196 #[inline]
200 pub fn backend<E>(error: E) -> Self
201 where
202 E: std::error::Error + Send + Sync + 'static,
203 {
204 Self::Backend(Arc::new(error))
205 }
206}
207
208impl From<EventCacheStoreError> for CrossProcessLockError {
209 fn from(value: EventCacheStoreError) -> Self {
210 Self::TryLock(Arc::new(value))
211 }
212}
213
214pub type Result<T, E = EventCacheStoreError> = std::result::Result<T, E>;
216
217#[derive(Clone, Debug)]
220struct LockableEventCacheStore(Arc<DynEventCacheStore>);
221
222impl TryLock for LockableEventCacheStore {
223 type LockError = EventCacheStoreError;
224
225 async fn try_lock(
226 &self,
227 lease_duration_ms: u32,
228 key: &str,
229 holder: &str,
230 ) -> std::result::Result<Option<CrossProcessLockGeneration>, Self::LockError> {
231 self.0.try_take_leased_lock(lease_duration_ms, key, holder).await
232 }
233}
234
235pub fn extract_event_relation(event: &Raw<AnySyncTimelineEvent>) -> Option<(OwnedEventId, String)> {
241 #[derive(serde::Deserialize)]
242 struct RelatesTo {
243 event_id: OwnedEventId,
244 rel_type: String,
245 }
246
247 #[derive(serde::Deserialize)]
248 struct EventContent {
249 #[serde(rename = "m.relates_to")]
250 rel: Option<RelatesTo>,
251 }
252
253 match event.get_field::<EventContent>("content") {
254 Ok(event_content) => {
255 event_content.and_then(|c| c.rel).map(|rel| (rel.event_id, rel.rel_type))
256 }
257 Err(err) => {
258 trace!("when extracting relation data from an event: {err}");
259 None
260 }
261 }
262}