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

matrix_sdk_base/event_cache/store/
memory_store.rs
