artifactize_core/store/
runs.rs

1use std::{
2    collections::BTreeMap,
3    path::{Path, PathBuf},
4};
5
6use rusqlite::params;
7use serde::Serialize;
8use tokio_rusqlite::Connection;
9
10use super::{
11    DATABASE,
12    receipts::{Error, check_files, schema_initialized},
13};
14use crate::workspace::{canonical_target, outside_workspace};
15
16#[derive(Debug, Clone, Serialize)]
17#[serde(rename_all = "camelCase")]
18pub struct RunSummary {
19    pub id: crate::types::RunId,
20    #[serde(serialize_with = "crate::platform::path_serde::serialize")]
21    pub repo_path: PathBuf,
22    pub created_at: crate::types::Timestamp,
23    pub completed_at: Option<crate::types::Timestamp>,
24    pub status: crate::types::RunStatus,
25    /// What the requester said it is working on (`verify --comment`).
26    #[serde(skip_serializing_if = "Option::is_none")]
27    pub comment: Option<crate::types::RunComment>,
28    pub counts: BTreeMap<crate::types::RequestStatus, u64>,
29    /// The Run's records that do not read, where the reader audited them (the monitor's
30    /// catalog); a listing reads columns only and leaves this empty.
31    #[serde(skip_serializing_if = "Vec::is_empty")]
32    pub unreadable: Vec<super::Unreadable>,
33}
34
35impl RunSummary {
36    /// Whether the Run's own record reads; a Run with only unreadable requests still opens.
37    pub fn readable(&self) -> bool {
38        !self.unreadable.iter().any(|record| record.kind == "run")
39    }
40}
41
42/// A page of Runs with their saved request-state counts, in one read-only snapshot and from
43/// columns alone: no record is decoded, so a page costs the same however large its Runs are.
44/// `status` and `run show` report the records that do not read. No repository discovery.
45pub async fn read_runs(
46    state: &Path,
47    repo: Option<&Path>,
48    limit: u32,
49    offset: u32,
50) -> Result<Vec<RunSummary>, String> {
51    let repos = repo.map(|path| vec![path.to_path_buf()]);
52    read_scoped_runs(state, repos.as_deref(), limit, offset).await
53}
54
55/// Filter workspace paths before sorting and paging; an empty scope has no Runs.
56pub async fn read_scoped_runs(
57    state: &Path,
58    repos: Option<&[PathBuf]>,
59    limit: u32,
60    offset: u32,
61) -> Result<Vec<RunSummary>, String> {
62    let state = canonical_target(state).map_err(|e| e.to_string())?;
63    let repos = repos
64        .map(|paths| {
65            paths
66                .iter()
67                .map(|path| {
68                    let path = canonical_target(path).map_err(|error| error.to_string())?;
69                    outside_workspace(&path, &state).map_err(|error| error.to_string())?;
70                    Ok(crate::platform::path_text(&path))
71                })
72                .collect::<Result<Vec<_>, String>>()
73        })
74        .transpose()?;
75    let repos = repos
76        .map(|paths| serde_json::to_string(&paths))
77        .transpose()
78        .map_err(|error| error.to_string())?;
79    check_files(&state)?;
80    if !state
81        .join(DATABASE)
82        .try_exists()
83        .map_err(|e| e.to_string())?
84    {
85        return Ok(Vec::new());
86    }
87    let connection = Connection::open_with_flags(
88        state.join(DATABASE),
89        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
90    )
91    .await
92    .map_err(|e| e.to_string())?;
93    let result = connection
94        .call(move |db| -> Result<_, Error> {
95            super::wait_when_busy(db)?;
96            let transaction = db.transaction()?;
97            if !schema_initialized(&transaction)? {
98                return Ok(Vec::new());
99            }
100            let runs = page(&transaction, &state, repos.as_deref(), limit, offset, None)?;
101            transaction.commit()?;
102            Ok(runs)
103        })
104        .await
105        .map_err(|e| e.to_string());
106    super::connection::finish(connection, result).await
107}
108
109/// A page of Runs with their counts, read from columns alone: no record of a Run is decoded.
110/// `pinned` is listed too when it is outside the page.
111pub(super) fn page(
112    db: &rusqlite::Connection,
113    state: &Path,
114    repos: Option<&str>,
115    limit: u32,
116    offset: u32,
117    pinned: Option<&crate::types::RunId>,
118) -> Result<Vec<RunSummary>, Error> {
119    let mut runs = listed(db, state, repos, limit, offset, pinned)?;
120    count(db, &mut runs)?;
121    Ok(runs)
122}
123
124fn listed(
125    db: &rusqlite::Connection,
126    state: &Path,
127    repos: Option<&str>,
128    limit: u32,
129    offset: u32,
130    pinned: Option<&crate::types::RunId>,
131) -> Result<Vec<RunSummary>, Error> {
132    let mut statement = db.prepare_cached(
133        "WITH page AS (
134         SELECT id,repo,json_extract(data,'$.createdAt') created,json_extract(data,'$.completedAt') completed,status,rowid serial,json_extract(data,'$.comment') comment
135         FROM runs WHERE (?1 IS NULL OR repo IN (SELECT value FROM json_each(?1)))
136         ORDER BY json_extract(data,'$.createdAt') DESC,rowid DESC LIMIT ?2 OFFSET ?3)
137         SELECT id,repo,created,completed,status,0,serial,comment FROM page UNION ALL
138         SELECT id,repo,json_extract(data,'$.createdAt'),json_extract(data,'$.completedAt'),status,1,rowid,json_extract(data,'$.comment') FROM runs
139         WHERE id=?4 AND id NOT IN (SELECT id FROM page)
140         AND (?1 IS NULL OR repo IN (SELECT value FROM json_each(?1))) ORDER BY 3 DESC,7 DESC",
141    )?;
142    let runs = statement
143        .query_map(params![repos, limit, offset, pinned], |row| {
144            Ok(RunSummary {
145                id: row.get(0)?,
146                repo_path: PathBuf::from(row.get::<_, String>(1)?),
147                created_at: row.get(2)?,
148                completed_at: row.get(3)?,
149                status: row.get(4)?,
150                // Only a comment that reads is shown; the Run's audit reports a record that
151                // does not.
152                comment: row
153                    .get::<_, Option<String>>(7)?
154                    .and_then(|comment| comment.parse().ok()),
155                counts: BTreeMap::new(),
156                unreadable: Vec::new(),
157            })
158        })?
159        .collect::<Result<Vec<_>, _>>()?;
160    crate::workspace::outside_workspaces(runs.iter().map(|run| run.repo_path.as_path()), state)
161        .map_err(|error| Error::Invalid(error.to_string()))?;
162    Ok(runs)
163}
164
165/// Fill in each Run's request counts by saved status.
166fn count(db: &rusqlite::Connection, runs: &mut [RunSummary]) -> Result<(), Error> {
167    let ids: Vec<_> = runs.iter().map(|run| run.id.clone()).collect();
168    let mut counts: BTreeMap<crate::types::RunId, BTreeMap<crate::types::RequestStatus, u64>> =
169        BTreeMap::new();
170    let mut statement = db.prepare_cached(
171        "SELECT run_id,status,count(*) FROM requests WHERE run_id IN (SELECT value FROM json_each(?1)) GROUP BY run_id,status",
172    )?;
173    let rows = statement.query_map([serde_json::to_string(&ids)?], |row| {
174        Ok((
175            row.get::<_, crate::types::RunId>(0)?,
176            row.get::<_, crate::types::RequestStatus>(1)?,
177            row.get::<_, i64>(2)?,
178        ))
179    })?;
180    for row in rows {
181        let (run, status, count) = row?;
182        let count = u64::try_from(count).expect("COUNT(*) is nonnegative");
183        counts.entry(run).or_default().insert(status, count);
184    }
185    for run in runs {
186        run.counts = counts.remove(&run.id).unwrap_or_default();
187    }
188    Ok(())
189}
artifactize_core/store/runs.rs

artifactize_core/store/
runs.rs
