nyx_agent_core/store/
run.rs

1//! `runs` table - one row per scan execution.
2
3use serde::{Deserialize, Serialize};
4use sqlx::{Row, SqlitePool};
5
6pub use nyx_agent_types::run::RunRecord;
7
8use crate::store::StoreError;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
11pub enum RunStatus {
12    Pending,
13    Running,
14    Succeeded,
15    Failed,
16    Halted,
17}
18
19impl RunStatus {
20    pub fn as_str(self) -> &'static str {
21        match self {
22            RunStatus::Pending => "Pending",
23            RunStatus::Running => "Running",
24            RunStatus::Succeeded => "Succeeded",
25            RunStatus::Failed => "Failed",
26            RunStatus::Halted => "Halted",
27        }
28    }
29}
30
31#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
32pub enum TriggeredBy {
33    Manual,
34    Cron,
35    Webhook,
36    Pr,
37    Ui,
38}
39
40impl TriggeredBy {
41    pub fn as_str(self) -> &'static str {
42        match self {
43            TriggeredBy::Manual => "Manual",
44            TriggeredBy::Cron => "Cron",
45            TriggeredBy::Webhook => "Webhook",
46            TriggeredBy::Pr => "PR",
47            TriggeredBy::Ui => "UI",
48        }
49    }
50}
51
52pub struct RunStore<'a> {
53    pool: &'a SqlitePool,
54}
55
56impl<'a> RunStore<'a> {
57    pub fn new(pool: &'a SqlitePool) -> Self {
58        Self { pool }
59    }
60
61    pub async fn insert(&self, r: &RunRecord) -> Result<(), StoreError> {
62        sqlx::query(
63            r#"
64            INSERT INTO runs (
65                id, project_id, kind, started_at, finished_at, status, triggered_by,
66                git_ref, parent_run_id, wall_clock_ms, total_ai_spend_usd_micros
67            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
68            "#,
69        )
70        .bind(&r.id)
71        .bind(&r.project_id)
72        .bind(if r.kind.is_empty() { "Scan" } else { r.kind.as_str() })
73        .bind(r.started_at)
74        .bind(r.finished_at)
75        .bind(&r.status)
76        .bind(&r.triggered_by)
77        .bind(&r.git_ref)
78        .bind(&r.parent_run_id)
79        .bind(r.wall_clock_ms)
80        .bind(r.total_ai_spend_usd_micros)
81        .execute(self.pool)
82        .await?;
83        Ok(())
84    }
85
86    pub async fn get(&self, id: &str) -> Result<Option<RunRecord>, StoreError> {
87        let row = sqlx::query(
88            r#"
89            SELECT id, project_id, kind,
90                   started_at,
91                   finished_at, status,
92                   triggered_by,
93                   git_ref, parent_run_id, wall_clock_ms,
94                   total_ai_spend_usd_micros
95            FROM runs WHERE id = ?
96            "#,
97        )
98        .bind(id)
99        .fetch_optional(self.pool)
100        .await?;
101        row.map(row_to_run_record).transpose()
102    }
103
104    pub async fn list_by_status(&self, status: &str) -> Result<Vec<RunRecord>, StoreError> {
105        let rows = sqlx::query(
106            r#"
107            SELECT id, project_id, kind,
108                   started_at,
109                   finished_at, status,
110                   triggered_by,
111                   git_ref, parent_run_id, wall_clock_ms,
112                   total_ai_spend_usd_micros
113            FROM runs WHERE status = ? ORDER BY started_at DESC
114            "#,
115        )
116        .bind(status)
117        .fetch_all(self.pool)
118        .await?;
119        rows.into_iter().map(row_to_run_record).collect()
120    }
121
122    pub async fn list_by_status_for_project(
123        &self,
124        status: &str,
125        project_id: &str,
126    ) -> Result<Vec<RunRecord>, StoreError> {
127        let rows = sqlx::query(
128            r#"
129            SELECT id, project_id, kind,
130                   started_at,
131                   finished_at, status,
132                   triggered_by,
133                   git_ref, parent_run_id, wall_clock_ms,
134                   total_ai_spend_usd_micros
135            FROM runs
136            WHERE status = ? AND project_id = ?
137            ORDER BY started_at DESC
138            "#,
139        )
140        .bind(status)
141        .bind(project_id)
142        .fetch_all(self.pool)
143        .await?;
144        rows.into_iter().map(row_to_run_record).collect()
145    }
146
147    pub async fn finish(
148        &self,
149        id: &str,
150        finished_at: i64,
151        status: &str,
152        wall_clock_ms: i64,
153    ) -> Result<(), StoreError> {
154        sqlx::query(
155            r#"
156            UPDATE runs
157               SET finished_at   = ?,
158                   status        = ?,
159                   wall_clock_ms = ?
160             WHERE id = ?
161            "#,
162        )
163        .bind(finished_at)
164        .bind(status)
165        .bind(wall_clock_ms)
166        .bind(id)
167        .execute(self.pool)
168        .await?;
169        Ok(())
170    }
171
172    pub async fn add_spend(&self, id: &str, micros: i64) -> Result<(), StoreError> {
173        sqlx::query(
174            "UPDATE runs SET total_ai_spend_usd_micros = total_ai_spend_usd_micros + ? WHERE id = ?",
175        )
176        .bind(micros)
177        .bind(id)
178        .execute(self.pool)
179        .await?;
180        Ok(())
181    }
182
183    /// Most recent earlier run, ranked by `started_at`. Used by the
184    /// findings diff endpoint to anchor "new vs prior" badge logic.
185    pub async fn prior_run_id(
186        &self,
187        run_id: &str,
188        started_at: i64,
189    ) -> Result<Option<String>, StoreError> {
190        let row = sqlx::query(
191            r#"
192            SELECT id FROM runs
193            WHERE started_at < ? AND id != ?
194            ORDER BY started_at DESC LIMIT 1
195            "#,
196        )
197        .bind(started_at)
198        .bind(run_id)
199        .fetch_optional(self.pool)
200        .await?;
201        Ok(row.map(|r| r.get("id")))
202    }
203
204    pub async fn delete(&self, id: &str) -> Result<u64, StoreError> {
205        let res = sqlx::query("DELETE FROM runs WHERE id = ?").bind(id).execute(self.pool).await?;
206        Ok(res.rows_affected())
207    }
208}
209
210fn row_to_run_record(row: sqlx::sqlite::SqliteRow) -> Result<RunRecord, StoreError> {
211    Ok(RunRecord {
212        id: row.try_get("id")?,
213        project_id: row.try_get("project_id")?,
214        kind: row.try_get("kind")?,
215        started_at: row.try_get::<i64, _>("started_at")?,
216        finished_at: row.try_get("finished_at")?,
217        status: row.try_get("status")?,
218        triggered_by: row.try_get("triggered_by")?,
219        git_ref: row.try_get("git_ref")?,
220        parent_run_id: row.try_get("parent_run_id")?,
221        wall_clock_ms: row.try_get("wall_clock_ms")?,
222        total_ai_spend_usd_micros: row.try_get::<i64, _>("total_ai_spend_usd_micros")?,
223    })
224}
225
226#[cfg(test)]
227mod tests {
228    use crate::store::testutil::{fresh_store, sample_finding, sample_run};
229
230    #[tokio::test]
231    async fn insert_then_get_roundtrips() {
232        let (_tmp, s) = fresh_store().await;
233        let r = sample_run("run-1");
234        s.runs().insert(&r).await.expect("insert");
235        let got = s.runs().get("run-1").await.expect("get").expect("row");
236        assert_eq!(got, r);
237    }
238
239    #[tokio::test]
240    async fn list_by_status_filters() {
241        let (_tmp, s) = fresh_store().await;
242        let mut a = sample_run("a");
243        a.status = "Succeeded".to_string();
244        let mut b = sample_run("b");
245        b.status = "Running".to_string();
246        s.runs().insert(&a).await.expect("a");
247        s.runs().insert(&b).await.expect("b");
248        let running = s.runs().list_by_status("Running").await.expect("list");
249        assert_eq!(running.len(), 1);
250        assert_eq!(running[0].id, "b");
251    }
252
253    #[tokio::test]
254    async fn list_by_status_for_project_filters() {
255        let (_tmp, s) = fresh_store().await;
256        s.projects()
257            .create("project-two", "project-two", None, None, None, 1)
258            .await
259            .expect("project");
260        let mut a = sample_run("a");
261        a.project_id = Some(crate::store::DEFAULT_PROJECT_ID.to_string());
262        let mut b = sample_run("b");
263        b.project_id = Some("project-two".to_string());
264        s.runs().insert(&a).await.expect("a");
265        s.runs().insert(&b).await.expect("b");
266
267        let rows = s
268            .runs()
269            .list_by_status_for_project("Running", crate::store::DEFAULT_PROJECT_ID)
270            .await
271            .expect("list");
272
273        assert_eq!(rows.len(), 1);
274        assert_eq!(rows[0].id, "a");
275    }
276
277    #[tokio::test]
278    async fn finish_updates_fields() {
279        let (_tmp, s) = fresh_store().await;
280        s.runs().insert(&sample_run("r")).await.expect("insert");
281        s.runs().finish("r", 9_999, "Succeeded", 7_000).await.expect("finish");
282        let got = s.runs().get("r").await.expect("get").expect("row");
283        assert_eq!(got.finished_at, Some(9_999));
284        assert_eq!(got.status, "Succeeded");
285        assert_eq!(got.wall_clock_ms, Some(7_000));
286    }
287
288    #[tokio::test]
289    async fn add_spend_accumulates() {
290        let (_tmp, s) = fresh_store().await;
291        s.runs().insert(&sample_run("r")).await.expect("insert");
292        s.runs().add_spend("r", 1_000).await.expect("spend1");
293        s.runs().add_spend("r", 2_500).await.expect("spend2");
294        let got = s.runs().get("r").await.expect("get").expect("row");
295        assert_eq!(got.total_ai_spend_usd_micros, 3_500);
296    }
297
298    #[tokio::test]
299    async fn delete_cascades_to_findings() {
300        let (_tmp, s) = fresh_store().await;
301        s.repos().upsert(&crate::store::testutil::sample_repo("r")).await.expect("repo");
302        s.runs().insert(&sample_run("doomed")).await.expect("run");
303        let f = sample_finding("doomed", "r", "src/a.rs", "rule-1");
304        let fid = f.id.clone();
305        s.findings().upsert(&f).await.expect("finding");
306        s.runs().delete("doomed").await.expect("delete");
307        assert!(
308            s.findings().get(&fid).await.expect("get").is_none(),
309            "FK cascade should have removed the finding"
310        );
311    }
312
313    #[tokio::test]
314    async fn parent_run_id_set_null_on_parent_delete() {
315        let (_tmp, s) = fresh_store().await;
316        s.runs().insert(&sample_run("parent")).await.expect("p");
317        let mut child = sample_run("child");
318        child.parent_run_id = Some("parent".to_string());
319        s.runs().insert(&child).await.expect("c");
320        s.runs().delete("parent").await.expect("del");
321        let got = s.runs().get("child").await.expect("get").expect("row");
322        assert!(got.parent_run_id.is_none(), "expected SET NULL on parent delete");
323    }
324}
nyx_agent_core/store/run.rs

nyx_agent_core/store/
run.rs
