moonlight_core/storage/
mod.rs

1mod reader;
2mod retention;
3mod scan;
4mod stats;
5mod writer;
6
7use crate::{run_matches_filter, ComparisonRun, ComparisonRunListItem, RunFilter, RunPage};
8use std::{
9    path::{Path, PathBuf},
10    sync::Arc,
11};
12use tokio::{
13    fs,
14    sync::{Mutex, RwLock},
15};
16use uuid::Uuid;
17
18pub use reader::JsonlStorageReader;
19pub use retention::StorageOptions;
20use retention::{atomic_write, retain_runs, serialize_runs_jsonl};
21use scan::{load_runs_from_file, load_runs_from_signature, scan_jsonl_files, JsonlFileSignature};
22use stats::StatsAccumulator;
23pub use writer::RunWriter;
24
25#[derive(Clone)]
26pub struct Storage {
27    write_path: PathBuf,
28    scan_dir: PathBuf,
29    writer: RunWriter,
30    options: StorageOptions,
31    insert_lock: Arc<Mutex<()>>,
32    runs: Arc<RwLock<Vec<ComparisonRun>>>,
33    scan_signature: Arc<Mutex<Vec<JsonlFileSignature>>>,
34}
35
36impl Storage {
37    pub async fn load(write_path: PathBuf) -> anyhow::Result<Self> {
38        Self::load_with_options(write_path, StorageOptions::default()).await
39    }
40
41    pub async fn load_with_options(
42        write_path: PathBuf,
43        options: StorageOptions,
44    ) -> anyhow::Result<Self> {
45        if let Some(parent) = write_path.parent() {
46            fs::create_dir_all(parent).await?;
47        }
48        let scan_dir = write_path
49            .parent()
50            .map(Path::to_path_buf)
51            .unwrap_or_else(|| PathBuf::from("."));
52        let scan_signature = scan_jsonl_files(&scan_dir).await?;
53        let runs = load_runs_from_signature(&scan_signature).await?;
54        let writer = RunWriter::open(write_path.clone()).await?;
55
56        Ok(Self {
57            write_path,
58            scan_dir,
59            writer,
60            options,
61            insert_lock: Arc::new(Mutex::new(())),
62            runs: Arc::new(RwLock::new(runs)),
63            scan_signature: Arc::new(Mutex::new(scan_signature)),
64        })
65    }
66
67    pub async fn insert(&self, run: ComparisonRun) -> anyhow::Result<()> {
68        let _guard = self.insert_lock.lock().await;
69        self.writer.append(&run).await?;
70        self.writer.flush().await?;
71        self.runs.write().await.push(run);
72        self.apply_retention().await?;
73        Ok(())
74    }
75
76    pub async fn refresh(&self) -> anyhow::Result<bool> {
77        let scan_signature = scan_jsonl_files(&self.scan_dir).await?;
78        {
79            let current = self.scan_signature.lock().await;
80            if *current == scan_signature {
81                return Ok(false);
82            }
83        }
84
85        let runs = load_runs_from_signature(&scan_signature).await?;
86        *self.runs.write().await = runs;
87        *self.scan_signature.lock().await = scan_signature;
88        Ok(true)
89    }
90
91    pub async fn list(&self) -> Vec<ComparisonRunListItem> {
92        self.list_page(usize::MAX, 0).await
93    }
94
95    pub async fn list_page(&self, limit: usize, offset: usize) -> Vec<ComparisonRunListItem> {
96        self.filtered_page(&RunFilter::default(), limit, offset)
97            .await
98            .items
99    }
100
101    pub async fn filtered_page(&self, filter: &RunFilter, limit: usize, offset: usize) -> RunPage {
102        let runs = self.runs.read().await;
103        let mut total = 0;
104        let mut items = Vec::new();
105
106        for run in runs
107            .iter()
108            .rev()
109            .filter(|run| run_matches_filter(run, filter))
110        {
111            if total >= offset && items.len() < limit {
112                items.push(ComparisonRunListItem::from(run));
113            }
114            total += 1;
115        }
116
117        let next_offset = (offset + items.len() < total).then_some(offset + items.len());
118        RunPage {
119            items,
120            limit,
121            offset,
122            total,
123            next_offset,
124        }
125    }
126
127    pub async fn get(&self, id: Uuid) -> Option<ComparisonRun> {
128        let runs = self.runs.read().await;
129        runs.iter().find(|run| run.id == id).cloned()
130    }
131
132    pub async fn stats(&self) -> crate::StatsSummary {
133        let runs = self.runs.read().await;
134        let mut accumulator = StatsAccumulator::default();
135
136        for run in runs.iter() {
137            accumulator.record(run);
138        }
139
140        accumulator.finish()
141    }
142
143    async fn apply_retention(&self) -> anyhow::Result<()> {
144        if !self.options.is_configured() {
145            return Ok(());
146        }
147
148        let mut active_runs = Vec::new();
149        load_runs_from_file(&self.write_path, &mut active_runs).await?;
150        active_runs.sort_by_key(|run| run.timestamp);
151
152        let retained_runs = retain_runs(active_runs.clone(), self.options)?;
153        let active_content = serialize_runs_jsonl(&active_runs)?;
154        let retained_content = serialize_runs_jsonl(&retained_runs)?;
155        if active_content == retained_content {
156            return Ok(());
157        }
158
159        atomic_write(&self.write_path, retained_content).await?;
160        self.writer.reopen(&self.write_path).await?;
161        self.force_refresh().await?;
162        Ok(())
163    }
164
165    async fn force_refresh(&self) -> anyhow::Result<()> {
166        let scan_signature = scan_jsonl_files(&self.scan_dir).await?;
167        let runs = load_runs_from_signature(&scan_signature).await?;
168        *self.runs.write().await = runs;
169        *self.scan_signature.lock().await = scan_signature;
170        Ok(())
171    }
172}
173
174#[cfg(test)]
175mod tests;
moonlight_core/storage/mod.rs

moonlight_core/storage/
mod.rs
