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    #[serde(skip_serializing_if = "Vec::is_empty")]
30    pub unreadable: Vec<super::Unreadable>,
31}
32
33impl RunSummary {
34    /// Whether the Run's own record reads; a Run with only unreadable requests still opens.
35    pub fn readable(&self) -> bool {
36        !self.unreadable.iter().any(|record| record.kind == "run")
37    }
38}
39
40/// Saved request-state counts in one read-only snapshot; no repository discovery.
41pub async fn read_runs(
42    state: &Path,
43    repo: Option<&Path>,
44    limit: u32,
45    offset: u32,
46) -> Result<Vec<RunSummary>, String> {
47    let repos = repo.map(|path| vec![path.to_path_buf()]);
48    read_scoped_runs(state, repos.as_deref(), limit, offset).await
49}
50
51/// Filter workspace paths before sorting and paging; an empty scope has no Runs.
52pub async fn read_scoped_runs(
53    state: &Path,
54    repos: Option<&[PathBuf]>,
55    limit: u32,
56    offset: u32,
57) -> Result<Vec<RunSummary>, String> {
58    let state = canonical_target(state).map_err(|e| e.to_string())?;
59    let repos = repos
60        .map(|paths| {
61            paths
62                .iter()
63                .map(|path| {
64                    let path = canonical_target(path).map_err(|error| error.to_string())?;
65                    outside_workspace(&path, &state).map_err(|error| error.to_string())?;
66                    Ok(crate::platform::path_text(&path))
67                })
68                .collect::<Result<Vec<_>, String>>()
69        })
70        .transpose()?;
71    let repos = repos
72        .map(|paths| serde_json::to_string(&paths))
73        .transpose()
74        .map_err(|error| error.to_string())?;
75    check_files(&state)?;
76    if !state
77        .join(DATABASE)
78        .try_exists()
79        .map_err(|e| e.to_string())?
80    {
81        return Ok(Vec::new());
82    }
83    let connection = Connection::open_with_flags(
84        state.join(DATABASE),
85        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
86    )
87    .await
88    .map_err(|e| e.to_string())?;
89    let result = connection
90        .call(move |db| -> Result<_, Error> {
91            super::wait_when_busy(db)?;
92            let transaction = db.transaction()?;
93            if !schema_initialized(&transaction)? {
94                return Ok(Vec::new());
95            }
96            let runs = in_snapshot(&transaction, &state, repos.as_deref(), limit, offset)?;
97            transaction.commit()?;
98            Ok(runs)
99        })
100        .await
101        .map_err(|e| e.to_string());
102    super::connection::finish(connection, result).await
103}
104
105/// A page of Runs with their counts, and every record of those Runs audited: for listings
106/// that report records which do not read. A request that does not read is not counted.
107pub(super) fn in_snapshot(
108    db: &rusqlite::Connection,
109    state: &Path,
110    repos: Option<&str>,
111    limit: u32,
112    offset: u32,
113) -> Result<Vec<RunSummary>, Error> {
114    let mut runs = listed(db, state, repos, limit, offset, None)?;
115    let ids: Vec<_> = runs.iter().map(|run| run.id.clone()).collect();
116    let mut unreadable = super::unreadable::in_runs(db, &ids)?;
117    let invalid: Vec<_> = unreadable
118        .values()
119        .flatten()
120        .filter(|record| record.kind == "request")
121        .map(|record| record.id.clone())
122        .collect();
123    count(db, &mut runs, &invalid)?;
124    for run in &mut runs {
125        run.unreadable = unreadable.remove(&run.id).unwrap_or_default();
126    }
127    Ok(runs)
128}
129
130/// A page of Runs with their counts, read from columns alone: no record of a Run is decoded.
131/// `pinned` is listed too when it is outside the page.
132pub(super) fn page(
133    db: &rusqlite::Connection,
134    state: &Path,
135    repos: Option<&str>,
136    limit: u32,
137    offset: u32,
138    pinned: Option<&crate::types::RunId>,
139) -> Result<Vec<RunSummary>, Error> {
140    let mut runs = listed(db, state, repos, limit, offset, pinned)?;
141    count(db, &mut runs, &[])?;
142    Ok(runs)
143}
144
145fn listed(
146    db: &rusqlite::Connection,
147    state: &Path,
148    repos: Option<&str>,
149    limit: u32,
150    offset: u32,
151    pinned: Option<&crate::types::RunId>,
152) -> Result<Vec<RunSummary>, Error> {
153    let mut statement = db.prepare_cached(
154        "WITH page AS (
155         SELECT id,repo,json_extract(data,'$.createdAt') created,json_extract(data,'$.completedAt') completed,status,rowid serial,json_extract(data,'$.comment') comment
156         FROM runs WHERE (?1 IS NULL OR repo IN (SELECT value FROM json_each(?1)))
157         ORDER BY json_extract(data,'$.createdAt') DESC,rowid DESC LIMIT ?2 OFFSET ?3)
158         SELECT id,repo,created,completed,status,0,serial,comment FROM page UNION ALL
159         SELECT id,repo,json_extract(data,'$.createdAt'),json_extract(data,'$.completedAt'),status,1,rowid,json_extract(data,'$.comment') FROM runs
160         WHERE id=?4 AND id NOT IN (SELECT id FROM page)
161         AND (?1 IS NULL OR repo IN (SELECT value FROM json_each(?1))) ORDER BY 3 DESC,7 DESC",
162    )?;
163    let runs = statement
164        .query_map(params![repos, limit, offset, pinned], |row| {
165            Ok(RunSummary {
166                id: row.get(0)?,
167                repo_path: PathBuf::from(row.get::<_, String>(1)?),
168                created_at: row.get(2)?,
169                completed_at: row.get(3)?,
170                status: row.get(4)?,
171                // Only a comment that reads is shown; the Run's audit reports a record that
172                // does not.
173                comment: row
174                    .get::<_, Option<String>>(7)?
175                    .and_then(|comment| comment.parse().ok()),
176                counts: BTreeMap::new(),
177                unreadable: Vec::new(),
178            })
179        })?
180        .collect::<Result<Vec<_>, _>>()?;
181    crate::workspace::outside_workspaces(runs.iter().map(|run| run.repo_path.as_path()), state)
182        .map_err(|error| Error::Invalid(error.to_string()))?;
183    Ok(runs)
184}
185
186/// Fill in each Run's request counts by status, leaving out the `invalid` requests.
187fn count(
188    db: &rusqlite::Connection,
189    runs: &mut [RunSummary],
190    invalid: &[String],
191) -> Result<(), Error> {
192    let ids: Vec<_> = runs.iter().map(|run| run.id.clone()).collect();
193    let mut counts: BTreeMap<crate::types::RunId, BTreeMap<crate::types::RequestStatus, u64>> =
194        BTreeMap::new();
195    let mut statement = db.prepare_cached(
196        "SELECT run_id,status,count(*) FROM requests WHERE run_id IN (SELECT value FROM json_each(?1)) AND id NOT IN (SELECT value FROM json_each(?2)) GROUP BY run_id,status",
197    )?;
198    let rows = statement.query_map(
199        params![
200            serde_json::to_string(&ids)?,
201            serde_json::to_string(invalid)?
202        ],
203        |row| {
204            Ok((
205                row.get::<_, crate::types::RunId>(0)?,
206                row.get::<_, crate::types::RequestStatus>(1)?,
207                row.get::<_, i64>(2)?,
208            ))
209        },
210    )?;
211    for row in rows {
212        let (run, status, count) = row?;
213        let count = u64::try_from(count).expect("COUNT(*) is nonnegative");
214        counts.entry(run).or_default().insert(status, count);
215    }
216    for run in runs {
217        run.counts = counts.remove(&run.id).unwrap_or_default();
218    }
219    Ok(())
220}
artifactize_core/store/runs.rs

artifactize_core/store/
runs.rs
