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

matrix_sdk_base/event_cache/store/
traits.rs
