xz_memory_engine/backends/
memory.rs

1use async_trait::async_trait;
2use std::collections::HashMap;
3use tokio::sync::Mutex;
4
5use xz_memory_core::StoreError;
6use xz_memory_core::traits::store::EntryStore;
7use xz_memory_core::types::entry::*;
8
9/// In-memory [`EntryStore`] implementation backed by a `HashMap`.
10///
11/// Partitioned entries are stored in-memory using `tokio::sync::Mutex`.
12/// Not suitable for production persistence — data is lost on drop.
13pub struct InMemoryEntryStore {
14    data: Mutex<HashMap<String, Vec<Entry>>>,
15}
16
17impl InMemoryEntryStore {
18    /// Create a new, empty `InMemoryEntryStore`.
19    pub fn new() -> Self {
20        InMemoryEntryStore { data: Mutex::new(HashMap::new()) }
21    }
22}
23
24#[async_trait]
25impl EntryStore for InMemoryEntryStore {
26    async fn append(&self, entry: Entry) -> Result<(), StoreError> {
27        let mut data = self.data.lock().await;
28        data.entry(entry.partition.clone()).or_default().push(entry);
29        Ok(())
30    }
31
32    async fn query(
33        &self,
34        partition: &str,
35        range: &TimeRange,
36        opts: &QueryOptions,
37    ) -> Result<Vec<Entry>, StoreError> {
38        let data = self.data.lock().await;
39        let mut entries: Vec<Entry> = data.get(partition).map_or(vec![], |v| {
40            v.iter()
41                .filter(|e| {
42                    let after_start = range.start.is_none_or(|s| e.recorded_at >= s);
43                    let before_end = range.end.is_none_or(|e2| e.recorded_at <= e2);
44                    after_start && before_end
45                })
46                .cloned()
47                .collect()
48        });
49        match opts.sort {
50            SortOrder::Ascending => entries.sort_by_key(|e| e.recorded_at),
51            SortOrder::Descending => entries.sort_by_key(|b| std::cmp::Reverse(b.recorded_at)),
52        }
53        entries.truncate(opts.limit);
54        Ok(entries)
55    }
56
57    async fn evict(&self, partition: &str, keep: usize) -> Result<usize, StoreError> {
58        let mut data = self.data.lock().await;
59        if let Some(entries) = data.get_mut(partition) {
60            if entries.len() <= keep {
61                return Ok(0);
62            }
63            let remove_count = entries.len() - keep;
64            entries.drain(..remove_count);
65            Ok(remove_count)
66        } else {
67            Ok(0)
68        }
69    }
70
71    async fn delete(&self, id: &str) -> Result<(), StoreError> {
72        let mut data = self.data.lock().await;
73        for entries in data.values_mut() {
74            entries.retain(|e| e.id != id);
75        }
76        Ok(())
77    }
78
79    async fn clear_partition(&self, partition: &str) -> Result<(), StoreError> {
80        let mut data = self.data.lock().await;
81        data.remove(partition);
82        Ok(())
83    }
84}
85
86impl Default for InMemoryEntryStore {
87    fn default() -> Self {
88        Self::new()
89    }
90}
xz_memory_engine/backends/memory.rs

xz_memory_engine/backends/
memory.rs
