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

matrix_sdk_base/event_cache/store/
memory_store.rs
