matrix_sdk_base/event_cache/store/
traits.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
15use std::{fmt, sync::Arc};
16
17use async_trait::async_trait;
18use matrix_sdk_common::{
19    AsyncTraitDeps,
20    cross_process_lock::CrossProcessLockGeneration,
21    linked_chunk::{
22        ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position,
23        RawChunk, Update,
24    },
25};
26use ruma::{EventId, OwnedEventId, RoomId, events::relation::RelationType};
27
28use super::EventCacheStoreError;
29use crate::event_cache::{Event, Gap};
30
31/// A default capacity for linked chunks, when manipulating in conjunction with
32/// an `EventCacheStore` implementation.
33// TODO: move back?
34pub const DEFAULT_CHUNK_CAPACITY: usize = 128;
35
36/// An abstract trait that can be used to implement different store backends
37/// for the event cache of the SDK.
38#[cfg_attr(target_family = "wasm", async_trait(?Send))]
39#[cfg_attr(not(target_family = "wasm"), async_trait)]
40pub trait EventCacheStore: AsyncTraitDeps {
41    /// The error type used by this event cache store.
42    type Error: fmt::Debug + Into<EventCacheStoreError>;
43
44    /// Try to take a lock using the given store.
45    async fn try_take_leased_lock(
46        &self,
47        lease_duration_ms: u32,
48        key: &str,
49        holder: &str,
50    ) -> Result<Option<CrossProcessLockGeneration>, Self::Error>;
51
52    /// An [`Update`] reflects an operation that has happened inside a linked
53    /// chunk. The linked chunk is used by the event cache to store the events
54    /// in-memory. This method aims at forwarding this update inside this store.
55    async fn handle_linked_chunk_updates(
56        &self,
57        linked_chunk_id: LinkedChunkId<'_>,
58        updates: Vec<Update<Event, Gap>>,
59    ) -> Result<(), Self::Error>;
60
61    /// Return all the raw components of a linked chunk, so the caller may
62    /// reconstruct the linked chunk later.
63    #[doc(hidden)]
64    async fn load_all_chunks(
65        &self,
66        linked_chunk_id: LinkedChunkId<'_>,
67    ) -> Result<Vec<RawChunk<Event, Gap>>, Self::Error>;
68
69    /// Load all of the chunks' metadata for the given [`LinkedChunkId`].
70    ///
71    /// Chunks are unordered, and there's no guarantee that the chunks would
72    /// form a valid linked chunk after reconstruction.
73    async fn load_all_chunks_metadata(
74        &self,
75        linked_chunk_id: LinkedChunkId<'_>,
76    ) -> Result<Vec<ChunkMetadata>, Self::Error>;
77
78    /// Load the last chunk of the `LinkedChunk` holding all events of the room
79    /// identified by `room_id`.
80    ///
81    /// This is used to iteratively load events for the `EventCache`.
82    async fn load_last_chunk(
83        &self,
84        linked_chunk_id: LinkedChunkId<'_>,
85    ) -> Result<(Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator), Self::Error>;
86
87    /// Load the chunk before the chunk identified by `before_chunk_identifier`
88    /// of the `LinkedChunk` holding all events of the room identified by
89    /// `room_id`
90    ///
91    /// This is used to iteratively load events for the `EventCache`.
92    async fn load_previous_chunk(
93        &self,
94        linked_chunk_id: LinkedChunkId<'_>,
95        before_chunk_identifier: ChunkIdentifier,
96    ) -> Result<Option<RawChunk<Event, Gap>>, Self::Error>;
97
98    /// Register a new thread.
99    ///
100    /// It does nothing regarding events or linked chunks: it simply remembers
101    /// that a thread has been created. This is important if one wants to list
102    /// all threads, or remove specific events or linked chunks.
103    ///
104    /// If the thread already exists, it returns successfully.
105    async fn remember_thread(
106        &self,
107        room_id: &RoomId,
108        thread_id: &EventId,
109    ) -> Result<(), Self::Error>;
110
111    /// Clear persisted events for all the rooms if `room_id` is `None`, or a
112    /// single room otherwise.
113    ///
114    /// This will empty and remove all the linked chunks stored previously,
115    /// using the above [`Self::handle_linked_chunk_updates`] methods. It
116    /// *also* deletes all the events' content.
117    ///
118    /// ⚠ This is meant only for super specific use cases, where there shouldn't
119    /// be any live in-memory linked chunks. In general, prefer using
120    /// `EventCache::clear_all_rooms()` from the common SDK crate.
121    async fn clear_all_events(&self, room_id: Option<&RoomId>) -> Result<(), Self::Error>;
122
123    /// Given a set of event IDs, return the duplicated events along with their
124    /// position if there are any.
125    async fn filter_duplicated_events(
126        &self,
127        linked_chunk_id: LinkedChunkId<'_>,
128        events: Vec<OwnedEventId>,
129    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error>;
130
131    /// Find an event by its ID in a room.
132    ///
133    /// This method must return events saved either in any linked chunks, *or*
134    /// events saved "out-of-band" with the [`Self::save_event`] method.
135    async fn find_event(
136        &self,
137        room_id: &RoomId,
138        event_id: &EventId,
139    ) -> Result<Option<Event>, Self::Error>;
140
141    /// Find all the events (alongside their position in the room's linked
142    /// chunk, if available) that relate to a given event.
143    ///
144    /// The only events which don't have a position are those which have been
145    /// saved out-of-band using [`Self::save_event`].
146    ///
147    /// Note: it doesn't process relations recursively: for instance, if
148    /// requesting only thread events, it will NOT return the aggregated
149    /// events affecting the returned events. It is the responsibility of
150    /// the caller to do so, if needed.
151    ///
152    /// An additional filter can be provided to only retrieve related events for
153    /// a certain relationship.
154    ///
155    /// This method must return events saved either in any linked chunks, *or*
156    /// events saved "out-of-band" with the [`Self::save_event`] method.
157    async fn find_event_relations(
158        &self,
159        room_id: &RoomId,
160        event_id: &EventId,
161        filter: Option<&[RelationType]>,
162    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error>;
163
164    /// Get all events in this room.
165    ///
166    /// This method must return events saved either in any linked chunks, *or*
167    /// events saved "out-of-band" with the [`Self::save_event`] method.
168    async fn get_room_events(
169        &self,
170        room_id: &RoomId,
171        event_type: Option<&str>,
172        session_id: Option<&str>,
173    ) -> Result<Vec<Event>, Self::Error>;
174
175    /// Save an event, that might or might not be part of an existing linked
176    /// chunk.
177    ///
178    /// If the event has no event id, it will not be saved, and the function
179    /// must return an Ok result early.
180    ///
181    /// If the event was already stored with the same id, it must be replaced,
182    /// without causing an error.
183    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error>;
184
185    /// Close the store, releasing all held resources (database connections,
186    /// file descriptors, file locks).
187    ///
188    /// In-flight operations complete before this method returns. After it
189    /// returns, operations will fail until [`Self::reopen()`] is called.
190    async fn close(&self) -> Result<(), Self::Error>;
191
192    /// Reopen the store after a [`Self::close()`], re-acquiring database
193    /// connections.
194    async fn reopen(&self) -> Result<(), Self::Error>;
195
196    /// Perform database optimizations if any are available, i.e. vacuuming in
197    /// SQLite.
198    ///
199    /// **Warning:** this was added to check if SQLite fragmentation was the
200    /// source of performance issues, **DO NOT use in production**.
201    #[doc(hidden)]
202    async fn optimize(&self) -> Result<(), Self::Error>;
203
204    /// Returns the size of the store in bytes, if known.
205    async fn get_size(&self) -> Result<Option<usize>, Self::Error>;
206}
207
208#[repr(transparent)]
209struct EraseEventCacheStoreError<T>(T);
210
211#[cfg(not(tarpaulin_include))]
212impl<T: fmt::Debug> fmt::Debug for EraseEventCacheStoreError<T> {
213    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
214        self.0.fmt(f)
215    }
216}
217
218#[cfg_attr(target_family = "wasm", async_trait(?Send))]
219#[cfg_attr(not(target_family = "wasm"), async_trait)]
220impl<T: EventCacheStore> EventCacheStore for EraseEventCacheStoreError<T> {
221    type Error = EventCacheStoreError;
222
223    async fn try_take_leased_lock(
224        &self,
225        lease_duration_ms: u32,
226        key: &str,
227        holder: &str,
228    ) -> Result<Option<CrossProcessLockGeneration>, Self::Error> {
229        self.0.try_take_leased_lock(lease_duration_ms, key, holder).await.map_err(Into::into)
230    }
231
232    async fn handle_linked_chunk_updates(
233        &self,
234        linked_chunk_id: LinkedChunkId<'_>,
235        updates: Vec<Update<Event, Gap>>,
236    ) -> Result<(), Self::Error> {
237        self.0.handle_linked_chunk_updates(linked_chunk_id, updates).await.map_err(Into::into)
238    }
239
240    async fn load_all_chunks(
241        &self,
242        linked_chunk_id: LinkedChunkId<'_>,
243    ) -> Result<Vec<RawChunk<Event, Gap>>, Self::Error> {
244        self.0.load_all_chunks(linked_chunk_id).await.map_err(Into::into)
245    }
246
247    async fn load_all_chunks_metadata(
248        &self,
249        linked_chunk_id: LinkedChunkId<'_>,
250    ) -> Result<Vec<ChunkMetadata>, Self::Error> {
251        self.0.load_all_chunks_metadata(linked_chunk_id).await.map_err(Into::into)
252    }
253
254    async fn load_last_chunk(
255        &self,
256        linked_chunk_id: LinkedChunkId<'_>,
257    ) -> Result<(Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator), Self::Error> {
258        self.0.load_last_chunk(linked_chunk_id).await.map_err(Into::into)
259    }
260
261    async fn load_previous_chunk(
262        &self,
263        linked_chunk_id: LinkedChunkId<'_>,
264        before_chunk_identifier: ChunkIdentifier,
265    ) -> Result<Option<RawChunk<Event, Gap>>, Self::Error> {
266        self.0
267            .load_previous_chunk(linked_chunk_id, before_chunk_identifier)
268            .await
269            .map_err(Into::into)
270    }
271
272    async fn remember_thread(
273        &self,
274        room_id: &RoomId,
275        thread_id: &EventId,
276    ) -> Result<(), Self::Error> {
277        self.0.remember_thread(room_id, thread_id).await.map_err(Into::into)
278    }
279
280    async fn clear_all_events(&self, room_id: Option<&RoomId>) -> Result<(), Self::Error> {
281        self.0.clear_all_events(room_id).await.map_err(Into::into)
282    }
283
284    async fn filter_duplicated_events(
285        &self,
286        linked_chunk_id: LinkedChunkId<'_>,
287        events: Vec<OwnedEventId>,
288    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error> {
289        self.0.filter_duplicated_events(linked_chunk_id, events).await.map_err(Into::into)
290    }
291
292    async fn find_event(
293        &self,
294        room_id: &RoomId,
295        event_id: &EventId,
296    ) -> Result<Option<Event>, Self::Error> {
297        self.0.find_event(room_id, event_id).await.map_err(Into::into)
298    }
299
300    async fn find_event_relations(
301        &self,
302        room_id: &RoomId,
303        event_id: &EventId,
304        filter: Option<&[RelationType]>,
305    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
306        self.0.find_event_relations(room_id, event_id, filter).await.map_err(Into::into)
307    }
308
309    async fn get_room_events(
310        &self,
311        room_id: &RoomId,
312        event_type: Option<&str>,
313        session_id: Option<&str>,
314    ) -> Result<Vec<Event>, Self::Error> {
315        self.0.get_room_events(room_id, event_type, session_id).await.map_err(Into::into)
316    }
317
318    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {
319        self.0.save_event(room_id, event).await.map_err(Into::into)
320    }
321
322    async fn close(&self) -> Result<(), Self::Error> {
323        self.0.close().await.map_err(Into::into)
324    }
325
326    async fn reopen(&self) -> Result<(), Self::Error> {
327        self.0.reopen().await.map_err(Into::into)
328    }
329
330    async fn optimize(&self) -> Result<(), Self::Error> {
331        self.0.optimize().await.map_err(Into::into)?;
332        Ok(())
333    }
334
335    async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
336        Ok(self.0.get_size().await.map_err(Into::into)?)
337    }
338}
339
340/// A type-erased [`EventCacheStore`].
341pub type DynEventCacheStore = dyn EventCacheStore<Error = EventCacheStoreError>;
342
343/// A type that can be type-erased into `Arc<dyn EventCacheStore>`.
344///
345/// This trait is not meant to be implemented directly outside
346/// `matrix-sdk-base`, but it is automatically implemented for everything that
347/// implements `EventCacheStore`.
348pub trait IntoEventCacheStore {
349    #[doc(hidden)]
350    fn into_event_cache_store(self) -> Arc<DynEventCacheStore>;
351}
352
353impl IntoEventCacheStore for Arc<DynEventCacheStore> {
354    fn into_event_cache_store(self) -> Arc<DynEventCacheStore> {
355        self
356    }
357}
358
359impl<T> IntoEventCacheStore for T
360where
361    T: EventCacheStore + Sized + 'static,
362{
363    fn into_event_cache_store(self) -> Arc<DynEventCacheStore> {
364        Arc::new(EraseEventCacheStoreError(self))
365    }
366}
367
368// Turns a given `Arc<T>` into `Arc<DynEventCacheStore>` by attaching the
369// `EventCacheStore` impl vtable of `EraseEventCacheStoreError<T>`.
370impl<T> IntoEventCacheStore for Arc<T>
371where
372    T: EventCacheStore + 'static,
373{
374    fn into_event_cache_store(self) -> Arc<DynEventCacheStore> {
375        let ptr: *const T = Arc::into_raw(self);
376        let ptr_erased = ptr as *const EraseEventCacheStoreError<T>;
377        // SAFETY: EraseEventCacheStoreError is repr(transparent) so T and
378        //         EraseEventCacheStoreError<T> have the same layout and ABI
379        unsafe { Arc::from_raw(ptr_erased) }
380    }
381}
matrix_sdk_base/event_cache/store/traits.rs

matrix_sdk_base/event_cache/store/
traits.rs
