matrix_sdk_base/event_cache/store/
memory_store.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::{
16    collections::{HashMap, HashSet},
17    sync::{Arc, RwLock as StdRwLock},
18};
19
20use async_trait::async_trait;
21use matrix_sdk_common::{
22    cross_process_lock::{
23        CrossProcessLockGeneration,
24        memory_store_helper::{Lease, try_take_leased_lock},
25    },
26    linked_chunk::{
27        ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position,
28        RawChunk, Update, relational::RelationalLinkedChunk,
29    },
30};
31use ruma::{EventId, OwnedEventId, OwnedRoomId, RoomId, events::relation::RelationType};
32use tracing::error;
33
34use super::{
35    super::{Event, Gap, thread::ThreadInfo},
36    EventCacheStore, EventCacheStoreError, Result, extract_event_relation,
37};
38
39/// In-memory, non-persistent implementation of the `EventCacheStore`.
40///
41/// Default if no other is configured at startup.
42///
43/// Note that this store is not transactional. This is particularly
44/// relevant when calling [`EventCacheStore::handle_linked_chunk_updates`],
45/// which consumes a list of [`Update`]s. When processing this list, if
46/// one of the [`Update`]s fails, the previous updates in the list
47/// will not be reversed.
48#[derive(Debug, Clone)]
49pub struct MemoryStore {
50    inner: Arc<StdRwLock<MemoryStoreInner>>,
51}
52
53#[derive(Debug)]
54struct MemoryStoreInner {
55    /// Leases for the cross-process lock.
56    leases: HashMap<String, Lease>,
57
58    /// All events organised in a `LinkedChunk`.
59    events: RelationalLinkedChunk<OwnedEventId, Event, Gap>,
60
61    /// List of all threads.
62    threads: HashMap<(OwnedRoomId, OwnedEventId), ThreadInfo>,
63}
64
65impl Default for MemoryStore {
66    fn default() -> Self {
67        Self {
68            inner: Arc::new(StdRwLock::new(MemoryStoreInner {
69                leases: Default::default(),
70                events: RelationalLinkedChunk::new(),
71                threads: HashMap::new(),
72            })),
73        }
74    }
75}
76
77impl MemoryStore {
78    /// Create a new empty MemoryStore
79    pub fn new() -> Self {
80        Self::default()
81    }
82}
83
84#[cfg_attr(target_family = "wasm", async_trait(?Send))]
85#[cfg_attr(not(target_family = "wasm"), async_trait)]
86impl EventCacheStore for MemoryStore {
87    type Error = EventCacheStoreError;
88
89    async fn close(&self) -> Result<(), Self::Error> {
90        Ok(())
91    }
92
93    async fn reopen(&self) -> Result<(), Self::Error> {
94        Ok(())
95    }
96
97    async fn try_take_leased_lock(
98        &self,
99        lease_duration_ms: u32,
100        key: &str,
101        holder: &str,
102    ) -> Result<Option<CrossProcessLockGeneration>, Self::Error> {
103        let mut inner = self.inner.write().unwrap();
104
105        Ok(try_take_leased_lock(&mut inner.leases, lease_duration_ms, key, holder))
106    }
107
108    async fn handle_linked_chunk_updates(
109        &self,
110        linked_chunk_id: LinkedChunkId<'_>,
111        updates: Vec<Update<Event, Gap>>,
112    ) -> Result<(), Self::Error> {
113        let mut inner = self.inner.write().unwrap();
114        inner
115            .events
116            .apply_updates(linked_chunk_id, updates)
117            .map_err(|e| Self::Error::Backend(Arc::new(e)))?;
118
119        Ok(())
120    }
121
122    async fn load_all_chunks(
123        &self,
124        linked_chunk_id: LinkedChunkId<'_>,
125    ) -> Result<Vec<RawChunk<Event, Gap>>, Self::Error> {
126        let inner = self.inner.read().unwrap();
127        inner
128            .events
129            .load_all_chunks(linked_chunk_id)
130            .map_err(|err| EventCacheStoreError::InvalidData { details: err })
131    }
132
133    async fn load_all_chunks_metadata(
134        &self,
135        linked_chunk_id: LinkedChunkId<'_>,
136    ) -> Result<Vec<ChunkMetadata>, Self::Error> {
137        let inner = self.inner.read().unwrap();
138        inner
139            .events
140            .load_all_chunks_metadata(linked_chunk_id)
141            .map_err(|err| EventCacheStoreError::InvalidData { details: err })
142    }
143
144    async fn load_last_chunk(
145        &self,
146        linked_chunk_id: LinkedChunkId<'_>,
147    ) -> Result<(Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator), Self::Error> {
148        let inner = self.inner.read().unwrap();
149        inner
150            .events
151            .load_last_chunk(linked_chunk_id)
152            .map_err(|err| EventCacheStoreError::InvalidData { details: err })
153    }
154
155    async fn load_previous_chunk(
156        &self,
157        linked_chunk_id: LinkedChunkId<'_>,
158        before_chunk_identifier: ChunkIdentifier,
159    ) -> Result<Option<RawChunk<Event, Gap>>, Self::Error> {
160        let inner = self.inner.read().unwrap();
161        inner
162            .events
163            .load_previous_chunk(linked_chunk_id, before_chunk_identifier)
164            .map_err(|err| EventCacheStoreError::InvalidData { details: err })
165    }
166
167    async fn load_thread_info(
168        &self,
169        room_id: &RoomId,
170        thread_id: &EventId,
171    ) -> Result<ThreadInfo, Self::Error> {
172        let mut inner = self.inner.write().unwrap();
173        let threads = &mut inner.threads;
174
175        let key = (room_id.to_owned(), thread_id.to_owned());
176
177        let thread_info = threads.entry(key).or_default();
178
179        Ok(thread_info.clone())
180    }
181
182    async fn update_thread_info(
183        &self,
184        room_id: &RoomId,
185        thread_id: &EventId,
186        thread_info: &ThreadInfo,
187    ) -> Result<(), Self::Error> {
188        let mut inner = self.inner.write().unwrap();
189        let threads = &mut inner.threads;
190
191        let key = (room_id.to_owned(), thread_id.to_owned());
192
193        *threads.get_mut(&key).expect("The thread entry must exist") = thread_info.clone();
194
195        Ok(())
196    }
197
198    async fn clear_all_events(&self, room_id: Option<&RoomId>) -> Result<(), Self::Error> {
199        match room_id {
200            Some(room_id) => {
201                self.inner.write().unwrap().events.clear_room(room_id);
202            }
203            None => {
204                self.inner.write().unwrap().events.clear();
205            }
206        }
207
208        Ok(())
209    }
210
211    async fn filter_duplicated_events(
212        &self,
213        linked_chunk_id: LinkedChunkId<'_>,
214        mut events: Vec<OwnedEventId>,
215    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error> {
216        if events.is_empty() {
217            return Ok(Vec::new());
218        }
219
220        let inner = self.inner.read().unwrap();
221
222        let mut duplicated_events = Vec::new();
223
224        for (event, position) in
225            inner.events.unordered_linked_chunk_items(&linked_chunk_id.to_owned())
226        {
227            if let Some(known_event_id) = event.event_id() {
228                // This event is a duplicate!
229                if let Some(index) =
230                    events.iter().position(|new_event_id| known_event_id == new_event_id)
231                {
232                    duplicated_events.push((events.remove(index), position));
233                }
234            }
235        }
236
237        Ok(duplicated_events)
238    }
239
240    async fn find_event(
241        &self,
242        room_id: &RoomId,
243        event_id: &EventId,
244    ) -> Result<Option<Event>, Self::Error> {
245        let inner = self.inner.read().unwrap();
246
247        let event = inner.events.items(room_id).find_map(|(_, (event, _pos))| {
248            (event.event_id()? == event_id).then_some(event.clone())
249        });
250
251        Ok(event)
252    }
253
254    async fn find_event_relations(
255        &self,
256        room_id: &RoomId,
257        event_id: &EventId,
258        filters: Option<&[RelationType]>,
259    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
260        let inner = self.inner.read().unwrap();
261
262        let related_events: Vec<_> = inner
263            .events
264            .items(room_id)
265            .filter_map(|(linked_chunk_id, (event, pos))| {
266                // Must have a relation.
267                let (related_to, rel_type) = extract_event_relation(event.raw())?;
268                let rel_type = RelationType::from(rel_type.as_str());
269
270                // Must relate to the target item.
271                if related_to != event_id {
272                    return None;
273                }
274
275                // Must not be filtered out.
276                if let Some(filters) = &filters {
277                    filters.contains(&rel_type).then_some((linked_chunk_id, (event.clone(), pos)))
278                } else {
279                    Some((linked_chunk_id, (event.clone(), pos)))
280                }
281            })
282            .collect();
283
284        // Remove any duplicate events which may exist in both a room and thread
285        // linked chunk. Additionally, remove any position information from non-room
286        // linked chunks.
287        let mut deduplicated = HashMap::new();
288        for (linked_chunk_id, (event, position)) in related_events {
289            let event_id = event
290                .event_id()
291                .ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?
292                .to_owned();
293            match linked_chunk_id.as_ref() {
294                LinkedChunkId::Room(_) => {
295                    // Prioritize events that come from a room linked chunk
296                    deduplicated.insert(event_id, (event, position));
297                }
298                _ => {
299                    // Remove position information from events that come
300                    // from any other type of linked chunk
301                    deduplicated.entry(event_id).or_insert_with(|| (event, None));
302                }
303            }
304        }
305
306        Ok(deduplicated.into_values().collect())
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        let inner = self.inner.read().unwrap();
316
317        let (_, event): (_, Vec<_>) = inner
318            .events
319            .items(room_id)
320            .map(|(_, (event, _pos))| event.clone())
321            .filter(|e| {
322                event_type
323                    .is_none_or(|event_type| Some(event_type) == e.kind.event_type().as_deref())
324            })
325            .filter(|e| session_id.is_none_or(|s| Some(s) == e.kind.session_id()))
326            .map(|e| {
327                let id = e
328                    .event_id()
329                    .ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?
330                    .to_owned();
331
332                Ok((id, e))
333            })
334            .collect::<Result<Vec<_>>>()?
335            .into_iter()
336            .fold((HashSet::new(), Vec::new()), |(mut ids, mut es), (id, e)| {
337                if !ids.contains(&id) {
338                    ids.insert(id);
339                    es.push(e);
340                }
341                (ids, es)
342            });
343        Ok(event)
344    }
345
346    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {
347        if event.event_id().is_none() {
348            error!(%room_id, "Trying to save an event with no ID");
349            return Ok(());
350        }
351        self.inner.write().unwrap().events.save_item(room_id.to_owned(), event);
352        Ok(())
353    }
354
355    async fn optimize(&self) -> Result<(), Self::Error> {
356        Ok(())
357    }
358
359    async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
360        Ok(None)
361    }
362}
363
364#[cfg(test)]
365#[allow(unused_imports)] // There seems to be a false positive when importing the test macros.
366mod tests {
367    use super::{MemoryStore, Result};
368    use crate::{event_cache_store_integration_tests, event_cache_store_integration_tests_time};
369
370    async fn get_event_cache_store() -> Result<MemoryStore> {
371        Ok(MemoryStore::new())
372    }
373
374    event_cache_store_integration_tests!();
375    #[cfg(not(target_family = "wasm"))]
376    event_cache_store_integration_tests_time!();
377}
matrix_sdk_base/event_cache/store/memory_store.rs

matrix_sdk_base/event_cache/store/
memory_store.rs
