moonlight_core/storage/
reader.rs

1use super::{scan::warn_corrupt_line, stats::StatsAccumulator};
2use crate::{run_matches_filter, ComparisonRun, ComparisonRunListItem, RunFilter, RunPage};
3use std::{
4    collections::VecDeque,
5    fs as std_fs,
6    io::{BufRead, BufReader as StdBufReader},
7    path::PathBuf,
8};
9use uuid::Uuid;
10
11#[derive(Clone)]
12pub struct JsonlStorageReader {
13    path: PathBuf,
14}
15
16impl JsonlStorageReader {
17    pub fn new(path: PathBuf) -> Self {
18        Self { path }
19    }
20
21    pub async fn stats(&self) -> anyhow::Result<crate::StatsSummary> {
22        let mut accumulator = StatsAccumulator::default();
23        self.for_each_run(|run| {
24            accumulator.record(&run);
25            true
26        })
27        .await?;
28        Ok(accumulator.finish())
29    }
30
31    pub async fn list_page(
32        &self,
33        limit: Option<usize>,
34        offset: usize,
35    ) -> anyhow::Result<Vec<ComparisonRunListItem>> {
36        Ok(self
37            .filtered_page(&RunFilter::default(), limit.unwrap_or(usize::MAX), offset)
38            .await?
39            .items)
40    }
41
42    pub async fn filtered_page(
43        &self,
44        filter: &RunFilter,
45        limit: usize,
46        offset: usize,
47    ) -> anyhow::Result<RunPage> {
48        let retained_limit = limit.saturating_add(offset);
49        let mut runs = VecDeque::new();
50        let mut total = 0;
51
52        self.for_each_run(|run| {
53            if run_matches_filter(&run, filter) {
54                total += 1;
55                if retained_limit > 0 {
56                    runs.push_back(ComparisonRunListItem::from(&run));
57                    if runs.len() > retained_limit {
58                        runs.pop_front();
59                    }
60                }
61            }
62            true
63        })
64        .await?;
65
66        let items = runs
67            .into_iter()
68            .rev()
69            .skip(offset)
70            .take(limit)
71            .collect::<Vec<_>>();
72        let next_offset = (offset + items.len() < total).then_some(offset + items.len());
73        Ok(RunPage {
74            items,
75            limit,
76            offset,
77            total,
78            next_offset,
79        })
80    }
81
82    pub async fn get(&self, id: Uuid) -> anyhow::Result<Option<ComparisonRun>> {
83        let mut found = None;
84        self.for_each_run(|run| {
85            if run.id == id {
86                found = Some(run);
87                false
88            } else {
89                true
90            }
91        })
92        .await?;
93        Ok(found)
94    }
95
96    async fn for_each_run(
97        &self,
98        mut visit: impl FnMut(ComparisonRun) -> bool,
99    ) -> anyhow::Result<()> {
100        if !self.path.try_exists()? {
101            return Ok(());
102        }
103
104        let file = std_fs::File::open(&self.path)?;
105        let lines = StdBufReader::new(file).lines();
106        for line in lines {
107            let line = line?;
108            if line.trim().is_empty() {
109                continue;
110            }
111            match serde_json::from_str::<ComparisonRun>(&line) {
112                Ok(run) => {
113                    if !visit(run) {
114                        break;
115                    }
116                }
117                Err(error) => warn_corrupt_line(&self.path, &error),
118            }
119        }
120        Ok(())
121    }
122}
moonlight_core/storage/reader.rs

moonlight_core/storage/
reader.rs
