use super::{Branch, Discovery, Labels, Place, branches};
use crate::{
changes::Changed,
monitor::{Node, Origin, model},
store::{self, CatalogRun, RequestView, RunView, Unreadable, reader::Reader},
types::RunId,
};
use std::{path::PathBuf, sync::Arc};
use time::OffsetDateTime;
#[cfg(test)]
mod tests;
#[derive(Debug, Clone, Default)]
pub struct Input {
pub state: PathBuf,
pub launch: Option<PathBuf>,
pub open: Option<RunId>,
pub force: bool,
pub rescan: bool,
pub changed: Changed,
}
pub type Opened = (RunView, Vec<RequestView>, Arc<Vec<Node>>);
pub struct Snapshot {
pub origin: Origin,
pub branches: Vec<Branch>,
pub labels: Labels,
pub launch: Option<Place>,
pub unreadable: Vec<Unreadable>,
pub opened: Option<Opened>,
pub open: bool,
}
#[derive(Default)]
pub struct Loader {
reader: Reader,
discovery: Discovery,
facts: Option<(Arc<Vec<CatalogRun>>, Vec<Unreadable>)>,
}
impl Loader {
pub fn load(&mut self, input: Input) -> Result<Snapshot, String> {
if input.rescan {
self.discovery.rescan();
}
let database = input.state.join(store::DATABASE);
if !database.try_exists().map_err(|error| error.to_string())? {
store::check_probe_files(&input.state)?;
self.reader.reset();
self.facts = None;
return Ok(self.empty(&input, Origin::default()));
}
let facts = &mut self.facts;
let saved = self
.reader
.snapshot_optional(&input.state, |db, cache, state| {
let changed = input.changed.or_written(cache.changed());
match (facts.as_mut(), &changed) {
(Some((runs, unreadable)), Changed::Runs(ids)) if cache.has_catalog() => {
let ids: Vec<_> = ids.iter().cloned().collect();
let rows = cache.catalog_rows(db, state, &ids)?;
*runs = Arc::new(merged(runs, &ids, rows.runs));
unreadable.retain(|record| !ids.iter().any(|id| record.id == id.as_str()));
unreadable.extend(rows.unreadable);
}
(Some(_), Changed::Nothing) if cache.has_catalog() => {}
_ => {
let runs = cache.catalog(db, state)?;
*facts = Some((Arc::new(runs), cache.unreadable_runs()));
}
}
let (runs, _) = facts.as_ref().expect("read above");
let open = input
.open
.as_ref()
.filter(|id| runs.iter().any(|run| &run.id == *id));
let read = open.filter(|id| input.force || !cache.holds(id) || changed.run(id));
let opened = match read {
Some(id) => match cache.selected(db, state, id, input.force) {
Ok(opened) => Ok(opened),
Err(store::receipts::Error::Unreadable(record)) => Err(record),
Err(error) => return Err(error),
},
None => Ok(None),
};
Ok((cache.identity(db)?, open.is_some(), opened))
})?;
let Some((identity, open, opened)) = saved else {
self.facts = None;
return Ok(self.empty(&input, Origin::new(None, self.reader.database())));
};
let (opened, unopened) = match opened {
Ok(opened) => (opened, None),
Err(record) => (None, Some(record)),
};
let opened = opened.map(|(run, requests)| {
let nodes = Arc::new(model::tree(&run, &requests, OffsetDateTime::now_utc()));
(run, requests, nodes)
});
let (runs, mut unreadable) = self.facts.clone().expect("read above");
if let Some((run, _, _)) = &opened {
unreadable.extend(run.unreadable.iter().cloned());
}
unreadable.extend(unopened);
let listing = branches(&runs, input.launch.as_deref(), &mut self.discovery);
self.reader.require_current(&input.state)?;
Ok(Snapshot {
origin: Origin::new(identity, self.reader.database()),
branches: listing.branches,
labels: listing.labels,
launch: listing.launch,
unreadable,
opened,
open,
})
}
fn empty(&mut self, input: &Input, origin: Origin) -> Snapshot {
let listing = branches(&[], input.launch.as_deref(), &mut self.discovery);
Snapshot {
origin,
branches: listing.branches,
labels: listing.labels,
launch: listing.launch,
unreadable: Vec::new(),
opened: None,
open: false,
}
}
}
fn merged(runs: &[CatalogRun], ids: &[RunId], rows: Vec<CatalogRun>) -> Vec<CatalogRun> {
let mut rows: std::collections::BTreeMap<_, _> =
rows.into_iter().map(|run| (run.id.clone(), run)).collect();
let mut merged: Vec<_> = runs
.iter()
.filter_map(|run| {
if ids.contains(&run.id) {
rows.remove(&run.id)
} else {
Some(run.clone())
}
})
.collect();
merged.extend(rows.into_values());
merged
}