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, hash_map::Entry},
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        insert_default_if_missing: bool,
202    ) -> Result<Option<ThreadInfo>, Self::Error> {
203        let mut inner = self.inner.write().unwrap();
204        let threads = &mut inner.threads;
205
206        let key = (room_id.to_owned(), thread_id.to_owned());
207        let thread_info_entry = threads.entry(key);
208
209        Ok(match thread_info_entry {
210            Entry::Occupied(entry) => Some(entry.get().clone()),
211
212            Entry::Vacant(entry) if insert_default_if_missing => {
213                Some(entry.insert_entry(ThreadInfo::default()).get().clone())
214            }
215
216            Entry::Vacant(_) => None,
217        })
218    }
219
220    async fn update_thread_info(
221        &self,
222        room_id: &RoomId,
223        thread_id: &EventId,
224        thread_info: &ThreadInfo,
225    ) -> Result<(), Self::Error> {
226        let mut inner = self.inner.write().unwrap();
227        let threads = &mut inner.threads;
228
229        let key = (room_id.to_owned(), thread_id.to_owned());
230
231        *threads.get_mut(&key).expect("The thread entry must exist") = thread_info.clone();
232
233        Ok(())
234    }
235
236    async fn clear_all_events(&self, room_id: Option<&RoomId>) -> Result<(), Self::Error> {
237        match room_id {
238            Some(room_id) => {
239                self.inner.write().unwrap().events.clear_room(room_id);
240            }
241            None => {
242                self.inner.write().unwrap().events.clear();
243            }
244        }
245
246        Ok(())
247    }
248
249    async fn filter_duplicated_events(
250        &self,
251        linked_chunk_id: LinkedChunkId<'_>,
252        mut events: Vec<OwnedEventId>,
253    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error> {
254        if events.is_empty() {
255            return Ok(Vec::new());
256        }
257
258        let inner = self.inner.read().unwrap();
259
260        let mut duplicated_events = Vec::new();
261
262        for (event, position) in
263            inner.events.unordered_linked_chunk_items(&linked_chunk_id.to_owned())
264        {
265            if let Some(known_event_id) = event.event_id() {
266                // This event is a duplicate!
267                if let Some(index) =
268                    events.iter().position(|new_event_id| known_event_id == new_event_id)
269                {
270                    duplicated_events.push((events.remove(index), position));
271                }
272            }
273        }
274
275        Ok(duplicated_events)
276    }
277
278    async fn find_event(
279        &self,
280        room_id: &RoomId,
281        event_id: &EventId,
282    ) -> Result<Option<Event>, Self::Error> {
283        let inner = self.inner.read().unwrap();
284
285        let event = inner.events.items(room_id).find_map(|(_, (event, _pos))| {
286            (event.event_id()? == event_id).then_some(event.clone())
287        });
288
289        Ok(event)
290    }
291
292    async fn find_event_relations(
293        &self,
294        room_id: &RoomId,
295        event_id: &EventId,
296        filters: Option<&[RelationType]>,
297    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
298        let inner = self.inner.read().unwrap();
299
300        let related_events: Vec<_> = inner
301            .events
302            .items(room_id)
303            .filter_map(|(linked_chunk_id, (event, pos))| {
304                // Must have a relation.
305                let (related_to, rel_type) = extract_event_relation(event.raw())?;
306                let rel_type = RelationType::from(rel_type.as_str());
307
308                // Must relate to the target item.
309                if related_to != event_id {
310                    return None;
311                }
312
313                // Must not be filtered out.
314                if let Some(filters) = &filters {
315                    filters.contains(&rel_type).then_some((linked_chunk_id, (event.clone(), pos)))
316                } else {
317                    Some((linked_chunk_id, (event.clone(), pos)))
318                }
319            })
320            .collect();
321
322        // Remove any duplicate events which may exist in both a room and thread
323        // linked chunk. Additionally, remove any position information from
324        // non-room linked chunks.
325        let mut deduplicated = HashMap::new();
326        for (linked_chunk_id, (event, position)) in related_events {
327            let event_id = event
328                .event_id()
329                .ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?
330                .to_owned();
331            match linked_chunk_id.as_ref() {
332                LinkedChunkId::Room(_) => {
333                    // Prioritize events that come from a room linked chunk
334                    deduplicated.insert(event_id, (event, position));
335                }
336                _ => {
337                    // Remove position information from events that come from
338                    // any other type of linked chunk
339                    deduplicated.entry(event_id).or_insert_with(|| (event, None));
340                }
341            }
342        }
343
344        Ok(deduplicated.into_values().collect())
345    }
346
347    async fn get_room_events(
348        &self,
349        room_id: &RoomId,
350        event_type: Option<&str>,
351        session_id: Option<&str>,
352    ) -> Result<Vec<Event>, Self::Error> {
353        let inner = self.inner.read().unwrap();
354
355        let (_, event): (_, Vec<_>) = inner
356            .events
357            .items(room_id)
358            .map(|(_, (event, _pos))| event.clone())
359            .filter(|e| {
360                event_type
361                    .is_none_or(|event_type| Some(event_type) == e.kind.event_type().as_deref())
362            })
363            .filter(|e| session_id.is_none_or(|s| Some(s) == e.kind.session_id()))
364            .map(|e| {
365                let id = e
366                    .event_id()
367                    .ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?
368                    .to_owned();
369
370                Ok((id, e))
371            })
372            .collect::<Result<Vec<_>>>()?
373            .into_iter()
374            .fold((HashSet::new(), Vec::new()), |(mut ids, mut es), (id, e)| {
375                if !ids.contains(&id) {
376                    ids.insert(id);
377                    es.push(e);
378                }
379                (ids, es)
380            });
381        Ok(event)
382    }
383
384    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {
385        if event.event_id().is_none() {
386            error!(%room_id, "Trying to save an event with no ID");
387            return Ok(());
388        }
389        self.inner.write().unwrap().events.save_item(room_id.to_owned(), event);
390        Ok(())
391    }
392
393    async fn optimize(&self) -> Result<(), Self::Error> {
394        Ok(())
395    }
396
397    async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
398        Ok(None)
399    }
400}
401
402#[cfg(test)]
403#[allow(unused_imports)] // There seems to be a false positive when importing the test macros.
404mod tests {
405    use super::{MemoryStore, Result};
406    use crate::{event_cache_store_integration_tests, event_cache_store_integration_tests_time};
407
408    async fn get_event_cache_store() -> Result<MemoryStore> {
409        Ok(MemoryStore::new())
410    }
411
412    event_cache_store_integration_tests!();
413    #[cfg(not(target_family = "wasm"))]
414    event_cache_store_integration_tests_time!();
415}
matrix_sdk_base/event_cache/store/memory_store.rs

matrix_sdk_base/event_cache/store/
memory_store.rs
