artifactize_core/monitor/live/
loader.rs

1//! One consistent read of the live branches and of the Run the Main area shows. Call
2//! [`Loader::load`] on a blocking thread; it never writes. What the change hints since the last
3//! read name is read again: the whole catalog of saved Runs, or the rows of the Runs named, and
4//! the open Run when it is one of them. Git is asked again which worktrees exist on a rescan.
5use super::{Branch, Discovery, Labels, Place, branches};
6use crate::{
7    changes::Changed,
8    monitor::{Node, Origin, model},
9    store::{self, CatalogRun, RequestView, RunView, Unreadable, reader::Reader},
10    types::RunId,
11};
12use std::{path::PathBuf, sync::Arc};
13use time::OffsetDateTime;
14
15#[cfg(test)]
16mod tests;
17
18/// What one read needs.
19#[derive(Debug, Clone, Default)]
20pub struct Input {
21    pub state: PathBuf,
22    /// The folder the app was started for: its branch is listed even before its first Run.
23    pub launch: Option<PathBuf>,
24    /// The Run the Main area shows, read in full.
25    pub open: Option<RunId>,
26    /// Read the open Run again even when no hint named it.
27    pub force: bool,
28    /// Ask Git again which worktrees exist.
29    pub rescan: bool,
30    /// What the change hints since the last read named.
31    pub changed: Changed,
32}
33
34/// The open Run: its view, its requests and their projected tree.
35pub type Opened = (RunView, Vec<RequestView>, Arc<Vec<Node>>);
36
37/// One consistent read.
38pub struct Snapshot {
39    pub origin: Origin,
40    /// The live branches, the most recent Run first.
41    pub branches: Vec<Branch>,
42    /// The project and branch of every readable saved Run.
43    pub labels: Labels,
44    /// The branch of the folder the app was started for, when it lives.
45    pub launch: Option<Place>,
46    /// Saved Run records that cannot be read, and records of the open Run that cannot.
47    pub unreadable: Vec<Unreadable>,
48    /// The open Run when it was read again, `None` when its records did not change since the
49    /// last read (or no Run is open, or it is gone: see [`Snapshot::open`]).
50    pub opened: Option<Opened>,
51    /// The open Run still exists and reads.
52    pub open: bool,
53}
54
55/// A persistent read-only worker with its caches.
56#[derive(Default)]
57pub struct Loader {
58    reader: Reader,
59    discovery: Discovery,
60    /// The catalog last read, kept until a hint names a change.
61    facts: Option<(Arc<Vec<CatalogRun>>, Vec<Unreadable>)>,
62}
63
64impl Loader {
65    pub fn load(&mut self, input: Input) -> Result<Snapshot, String> {
66        if input.rescan {
67            self.discovery.rescan();
68        }
69        let database = input.state.join(store::DATABASE);
70        if !database.try_exists().map_err(|error| error.to_string())? {
71            // Empty state is a normal baseline; its file safety checks still run here.
72            store::check_probe_files(&input.state)?;
73            self.reader.reset();
74            self.facts = None;
75            return Ok(self.empty(&input, Origin::default()));
76        }
77        let facts = &mut self.facts;
78        let saved = self
79            .reader
80            .snapshot_optional(&input.state, |db, cache, state| {
81                let changed = input.changed.or_written(cache.changed());
82                match (facts.as_mut(), &changed) {
83                    (Some((runs, unreadable)), Changed::Runs(ids)) if cache.has_catalog() => {
84                        let ids: Vec<_> = ids.iter().cloned().collect();
85                        let rows = cache.catalog_rows(db, state, &ids)?;
86                        *runs = Arc::new(merged(runs, &ids, rows.runs));
87                        unreadable.retain(|record| !ids.iter().any(|id| record.id == id.as_str()));
88                        unreadable.extend(rows.unreadable);
89                    }
90                    (Some(_), Changed::Nothing) if cache.has_catalog() => {}
91                    _ => {
92                        let runs = cache.catalog(db, state)?;
93                        *facts = Some((Arc::new(runs), cache.unreadable_runs()));
94                    }
95                }
96                let (runs, _) = facts.as_ref().expect("read above");
97                let open = input
98                    .open
99                    .as_ref()
100                    .filter(|id| runs.iter().any(|run| &run.id == *id));
101                // The open Run is read when it is new here or a hint named it.
102                let read = open.filter(|id| input.force || !cache.holds(id) || changed.run(id));
103                let opened = match read {
104                    Some(id) => match cache.selected(db, state, id, input.force) {
105                        Ok(opened) => Ok(opened),
106                        // The Run's definitions do not decode: it is listed with why, not opened.
107                        Err(store::receipts::Error::Unreadable(record)) => Err(record),
108                        Err(error) => return Err(error),
109                    },
110                    None => Ok(None),
111                };
112                Ok((cache.identity(db)?, open.is_some(), opened))
113            })?;
114        let Some((identity, open, opened)) = saved else {
115            self.facts = None;
116            return Ok(self.empty(&input, Origin::new(None, self.reader.database())));
117        };
118        let (opened, unopened) = match opened {
119            Ok(opened) => (opened, None),
120            Err(record) => (None, Some(record)),
121        };
122        // No read transaction survives heavy projection or Git discovery.
123        let opened = opened.map(|(run, requests)| {
124            let nodes = Arc::new(model::tree(&run, &requests, OffsetDateTime::now_utc()));
125            (run, requests, nodes)
126        });
127        let (runs, mut unreadable) = self.facts.clone().expect("read above");
128        if let Some((run, _, _)) = &opened {
129            unreadable.extend(run.unreadable.iter().cloned());
130        }
131        unreadable.extend(unopened);
132        let listing = branches(&runs, input.launch.as_deref(), &mut self.discovery);
133        self.reader.require_current(&input.state)?;
134        Ok(Snapshot {
135            origin: Origin::new(identity, self.reader.database()),
136            branches: listing.branches,
137            labels: listing.labels,
138            launch: listing.launch,
139            unreadable,
140            opened,
141            open,
142        })
143    }
144
145    fn empty(&mut self, input: &Input, origin: Origin) -> Snapshot {
146        let listing = branches(&[], input.launch.as_deref(), &mut self.discovery);
147        Snapshot {
148            origin,
149            branches: listing.branches,
150            labels: listing.labels,
151            launch: listing.launch,
152            unreadable: Vec::new(),
153            opened: None,
154            open: false,
155        }
156    }
157}
158
159/// `runs` with the rows of the Runs `ids` replaced by `rows`: a Run without a row is gone, and
160/// a new one is the newest, listed last.
161fn merged(runs: &[CatalogRun], ids: &[RunId], rows: Vec<CatalogRun>) -> Vec<CatalogRun> {
162    let mut rows: std::collections::BTreeMap<_, _> =
163        rows.into_iter().map(|run| (run.id.clone(), run)).collect();
164    let mut merged: Vec<_> = runs
165        .iter()
166        .filter_map(|run| {
167            if ids.contains(&run.id) {
168                rows.remove(&run.id)
169            } else {
170                Some(run.clone())
171            }
172        })
173        .collect();
174    merged.extend(rows.into_values());
175    merged
176}
artifactize_core/monitor/live/loader.rs

artifactize_core/monitor/live/
loader.rs
