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

artifactize_core/store/
runs.rs
