artifactize_core/store/
catalog.rs

1//! Global monitor catalog, intentionally independent of any visible Run page.
2use super::{
3    DATABASE,
4    receipts::{Error, check_files, schema_initialized},
5};
6use crate::{
7    repository::Identity,
8    types::{ExecutionId, RequestId, RequestStatus, RunStatus, Sha256Bytes},
9    workspace::{canonical_target, outside_workspaces},
10};
11use std::path::{Path, PathBuf};
12
13#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
14pub enum Signoff {
15    Execution(ExecutionId),
16    Request(RequestId),
17}
18
19#[derive(Debug, Clone)]
20pub struct CatalogRun {
21    pub repo_path: PathBuf,
22    pub repository: Identity,
23    pub status: RunStatus,
24    pub red: u64,
25    pub error: u64,
26    pub waiting: Vec<Signoff>,
27}
28
29/// Global attention facts; unreadable Runs are skipped individually with their reason.
30pub async fn read_catalog(state: &Path) -> Result<Vec<CatalogRun>, String> {
31    let state = canonical_target(state).map_err(|error| error.to_string())?;
32    check_files(&state)?;
33    if !state
34        .join(DATABASE)
35        .try_exists()
36        .map_err(|error| error.to_string())?
37    {
38        return Ok(Vec::new());
39    }
40    let connection = tokio_rusqlite::Connection::open_with_flags(
41        state.join(DATABASE),
42        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
43    )
44    .await
45    .map_err(|error| error.to_string())?;
46    connection
47        .call(move |db| -> Result<_, Error> {
48            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
49            let transaction = db.transaction()?;
50            if !schema_initialized(&transaction)? {
51                return Ok(Vec::new());
52            }
53            let catalog = in_snapshot(&transaction, &state)?;
54            transaction.commit()?;
55            Ok(catalog)
56        })
57        .await
58        .map_err(|error| error.to_string())
59}
60
61pub(super) fn in_snapshot(
62    db: &rusqlite::Connection,
63    state: &Path,
64) -> Result<Vec<CatalogRun>, Error> {
65    in_snapshot_cached(db, state, &mut Facts::default())
66}
67
68#[derive(Default)]
69pub(super) struct Facts {
70    runs: std::collections::BTreeMap<crate::types::RunId, (Sha256Bytes, Option<CatalogRun>)>,
71}
72
73pub(super) fn in_snapshot_cached(
74    db: &rusqlite::Connection,
75    state: &Path,
76    facts: &mut Facts,
77) -> Result<Vec<CatalogRun>, Error> {
78    let mut statement = db.prepare_cached("SELECT id,data FROM runs ORDER BY rowid")?;
79    let runs = statement
80        .query_map([], |row| {
81            Ok((
82                row.get::<_, crate::types::RunId>(0)?,
83                row.get::<_, String>(1)?,
84            ))
85        })?
86        .collect::<Result<Vec<_>, _>>()?;
87    let mut catalog = std::collections::BTreeMap::new();
88    let mut order = Vec::new();
89    let mut retained = std::collections::BTreeMap::new();
90    for (id, data) in runs {
91        let digest = Sha256Bytes::of(data.as_bytes());
92        let fact = match facts.runs.remove(&id) {
93            Some((saved, fact)) if saved == digest => fact,
94            _ => super::unreadable::evidence::<super::Run>("run", &data).map(|run| {
95                let status = run.status();
96                CatalogRun {
97                    repo_path: run.repo_path,
98                    repository: run.repository,
99                    status,
100                    red: 0,
101                    error: 0,
102                    waiting: Vec::new(),
103                }
104            }),
105        };
106        retained.insert(id.clone(), (digest, fact.clone()));
107        if let Some(fact) = fact {
108            order.push(id.clone());
109            catalog.insert(id, fact);
110        }
111    }
112    outside_workspaces(catalog.values().map(|run| run.repo_path.as_path()), state)
113        .map_err(|error| Error::Invalid(error.to_string()))?;
114    facts.runs = retained;
115    let mut statement = db.prepare_cached(
116        "SELECT q.run_id,q.status,q.execution_id,coalesce(json_extract(e.data,'$.provenance.requestId'),json_extract(q.data,'$.provenance.requestId'),q.id)
117        FROM requests q LEFT JOIN executions e ON e.id=q.execution_id WHERE q.status IN ('RED','ERROR','WAITING_HUMAN') ORDER BY q.run_id,q.ordinal",
118    )?;
119    let attention = statement.query_map([], |row| {
120        Ok((
121            row.get::<_, crate::types::RunId>(0)?,
122            row.get::<_, RequestStatus>(1)?,
123            row.get::<_, Option<ExecutionId>>(2)?,
124            row.get::<_, RequestId>(3)?,
125        ))
126    })?;
127    for row in attention {
128        let (id, status, execution, request) = row?;
129        if let Some(run) = catalog.get_mut(&id) {
130            match status {
131                RequestStatus::Red => run.red += 1,
132                RequestStatus::Error => run.error += 1,
133                RequestStatus::WaitingHuman => run
134                    .waiting
135                    .push(execution.map_or(Signoff::Request(request), Signoff::Execution)),
136                _ => unreachable!("attention query filters states"),
137            }
138        }
139    }
140    Ok(order
141        .into_iter()
142        .filter_map(|id| catalog.remove(&id))
143        .collect())
144}
artifactize_core/store/catalog.rs

artifactize_core/store/
catalog.rs
