OXYGENHUB

Coverage / crates/hub/src/web/flow.rs 77.9%

654 of 839 lines run, 86 of 290 functions · run #509 on main · run missed

1
//! The Flow tab (hub spec §13.2): runs of a repository, and a run's job
2
//! graph with live status and logs.
3
4
use super::settings::{csrf, require};
5
use super::{filters, lookup, lookup_role, Error, Page, RepoShell, Shell};
6
use crate::auth::{Access, Role, Viewer};
7
use crate::db::{self, flow as fdb};
8
use crate::Shared;
9
use askama::Template;
10
use axum::extract::{Form, Path, Query, State};
11
use axum::response::{IntoResponse, Redirect, Response};
12
use axum::Json;
13
use serde::Deserialize;
14
15
pub struct RunRow {
16
    pub r: fdb::Run,
17
    pub duration: String,
18
    pub jobs: Vec<(String, String)>,
19
    /// "3rd in line" for a run still waiting for a runner.
20
    pub place: Option<String>,
21
}
22
23
#[derive(Template)]
24
#[template(path = "flow_runs.html")]
25
pub struct RunsPage {
26
    shell: Shell,
27
    repo: RepoShell,
28
    runs: Vec<RunRow>,
29
    enabled: bool,
30
    older: Option<i64>,
31
    /// Writers may start runs: the bookmarks to offer, default first.
32
    can_run: bool,
33
    bookmarks: Vec<String>,
34
    error: Option<String>,
35
    /// Why this viewer is not offered "Run pipeline", when they are not.
36
    why_not: Option<String>,
37
}
38
39
#[derive(Deserialize)]
40
pub struct RunsQuery {
41
    before: Option<i64>,
42
    /// Why a "Run pipeline" or "Re-run" did not start.
43
    error: Option<String>,
44
}
45
46
/// Why `viewer` may not start runs here, or `None` when they may.
4736×
pub fn why_not_run(enabled: bool, signed_in: bool, role: Role) -> Option<String> {
4836×
    if !enabled {
4930×
        Some("Flow is off for this repository; a site administrator can turn it on in Settings › Pipelines.".into())
506×
    } else if !signed_in {
515×
        Some("Sign in with write access to run the pipeline.".into())
521×
    } else if role < Role::Write {
530×
        Some("Running the pipeline needs write access to this repository.".into())
54
    } else {
551×
        None
56
    }
5736×
}
58
59
const PER_PAGE: i64 = 30;
60
61
#[derive(Template)]
62
#[template(path = "deployments.html")]
63
pub struct DeploymentsPage {
64
    shell: Shell,
65
    repo: RepoShell,
66
    /// Each environment's newest deployment, by name.
67
    current: Vec<fdb::Deployment>,
68
    /// Every recent deployment, newest first.
69
    history: Vec<fdb::Deployment>,
70
}
71
72
/// `GET /{owner}/{name}/deployments`: what each environment runs
73
/// (oxygenflow#59).
741×
pub async fn deployments(
751×
    State(app): State<Shared>,
761×
    viewer: Viewer,
771×
    Path((owner, name)): Path<(String, String)>,
781×
) -> Result<Page<DeploymentsPage>, Error> {
791×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/deployments"), &viewer);
801×
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
811×
    let history = fdb::deployments(&app.db, row.id, 50).await.map_err(|e| Error::from_db(shell.clone(), e))?;
821×
    let mut current: Vec<fdb::Deployment> = Vec::new();
831×
    for d in &history {
841×
        if !current.iter().any(|c| c.environment == d.environment) {
851×
            current.push(d.clone());
861×
        }
87
    }
881×
    current.sort_by(|a, b| a.environment.cmp(&b.environment));
891×
    Ok(Page(DeploymentsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), current, history }))
901×
}
91
927×
pub async fn runs(
937×
    State(app): State<Shared>,
947×
    viewer: Viewer,
957×
    Path((owner, name)): Path<(String, String)>,
967×
    Query(q): Query<RunsQuery>,
977×
) -> Result<Page<RunsPage>, Error> {
987×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
997×
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
1007×
    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1017×
    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);
1027×
    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };
1037×
    let mut runs = Vec::new();
10410×
    for r in list.into_iter().take(PER_PAGE as usize) {
10510×
        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };
10610×
        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();
10710×
        let duration = match r.finished_ms {
1083×
            Some(f) => crate::flow::human_ms(f - r.created_ms),
1097×
            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),
1100×
            None => "–".into(),
111
        };
11229×
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });
113
    }
1147×
    let enabled = crate::flow::enabled(&app, &row);
1157×
    let why_not = why_not_run(enabled, viewer.account().is_some(), role);
1167×
    let can_run = why_not.is_none();
1177×
    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };
1187×
    Ok(Page(RunsPage {
1197×
        shell,
1207×
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
1217×
        runs,
1227×
        enabled,
1237×
        older,
1247×
        can_run,
1257×
        bookmarks,
1267×
        error: q.error,
1277×
        why_not,
1287×
    }))
1297×
}
130
131
/// The repository's bookmarks, its default first.
1321×
async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {
1331×
    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };
1341×
    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();
1351×
    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));
1361×
    names
1371×
}
138
139
#[derive(Deserialize)]
140
pub struct StartForm {
141
    #[serde(rename = "_csrf")]
142
    csrf_token: String,
143
    #[serde(default)]
144
    bookmark: String,
145
    /// A change's full hash: run the state whose only head it is.
146
    #[serde(default)]
147
    change: String,
148
}
149
150
/// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark's
151
/// current state, or on one committed change.
1520×
pub async fn start(
1530×
    State(app): State<Shared>,
1540×
    viewer: Viewer,
1550×
    Path((owner, name)): Path<(String, String)>,
1560×
    Form(f): Form<StartForm>,
1570×
) -> Result<Response, Error> {
1580×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
1590×
    let me = require(&viewer, &shell)?;
1600×
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
1610×
    csrf(&viewer, &f.csrf_token, &shell)?;
1620×
    let (started, what) = if f.change.trim().is_empty() {
1630×
        (crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username, &[]).await, f.bookmark.trim().to_string())
164
    } else {
1650×
        (crate::flow::run_change(&app, &row, &f.change, &me.account.username, &[]).await, format!("change {}", f.change.trim()))
166
    };
1670×
    Ok(match started {
1680×
        Ok(n) => {
1690×
            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {what}")).await.ok();
1700×
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
171
        }
1720×
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
173
    })
1740×
}
175
176
#[derive(Deserialize)]
177
pub struct RerunForm {
178
    #[serde(rename = "_csrf")]
179
    csrf_token: String,
180
}
181
182
/// `POST /{owner}/{name}/flow/{number}/rerun`.
1830×
pub async fn rerun(
1840×
    State(app): State<Shared>,
1850×
    viewer: Viewer,
1860×
    Path((owner, name, number)): Path<(String, String, i64)>,
1870×
    Form(f): Form<RerunForm>,
1880×
) -> Result<Response, Error> {
1890×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
1900×
    let me = require(&viewer, &shell)?;
1910×
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
1920×
    csrf(&viewer, &f.csrf_token, &shell)?;
1930×
    let run = fdb::run(&app.db, row.id, number)
1940×
        .await
1950×
        .map_err(|e| Error::from_db(shell.clone(), e))?
1960×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
1970×
    Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await {
1980×
        Ok(n) => {
1990×
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}"))
2000×
                .await
2010×
                .ok();
2020×
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
203
        }
2040×
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
205
    })
2060×
}
207
208
/// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31).
2090×
pub async fn rerun_job(
2100×
    State(app): State<Shared>,
2110×
    viewer: Viewer,
2120×
    Path((owner, name, number, job)): Path<(String, String, i64, String)>,
2130×
    Form(f): Form<RerunForm>,
2140×
) -> Result<Response, Error> {
2150×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
2160×
    let me = require(&viewer, &shell)?;
2170×
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
2180×
    csrf(&viewer, &f.csrf_token, &shell)?;
2190×
    let run = fdb::run(&app.db, row.id, number)
2200×
        .await
2210×
        .map_err(|e| Error::from_db(shell.clone(), e))?
2220×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
2230×
    Ok(match crate::flow::rerun_job(&app, &row, &run, &job, &me.account.username).await {
224
        Ok(_) => {
2250×
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("{job} of #{number}"))
2260×
                .await
2270×
                .ok();
2280×
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
229
        }
2300×
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
231
    })
2320×
}
233
234
/// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32).
2350×
pub async fn cancel(
2360×
    State(app): State<Shared>,
2370×
    viewer: Viewer,
2380×
    Path((owner, name, number)): Path<(String, String, i64)>,
2390×
    Form(f): Form<RerunForm>,
2400×
) -> Result<Response, Error> {
2410×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
2420×
    let me = require(&viewer, &shell)?;
2430×
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
2440×
    csrf(&viewer, &f.csrf_token, &shell)?;
2450×
    let run = fdb::run(&app.db, row.id, number)
2460×
        .await
2470×
        .map_err(|e| Error::from_db(shell.clone(), e))?
2480×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
2490×
    Ok(match crate::flow::cancel(&app, &row, &run, &me.account.username).await {
250
        Ok(()) => {
2510×
            db::accounts::audit(&app.db, Some(&me.account), "flow.cancel", &format!("{owner}/{name}"), &format!("#{number}")).await.ok();
2520×
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
253
        }
2540×
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
255
    })
2560×
}
257
258
#[derive(Deserialize)]
259
pub struct ExpediteForm {
260
    #[serde(rename = "_csrf")]
261
    csrf_token: String,
262
    #[serde(default)]
263
    reason: String,
264
}
265
266
/// `POST /{owner}/{repo}/flow/{n}/expedite` (oxygenhub#244): a maintainer
267
/// raises a running or queued run to urgent, saying why (an incident's
268
/// fix, say). Audited and shown on the run; it lapses when the run ends.
2691×
pub async fn expedite(
2701×
    State(app): State<Shared>,
2711×
    viewer: Viewer,
2721×
    Path((owner, name, number)): Path<(String, String, i64)>,
2731×
    Form(f): Form<ExpediteForm>,
2741×
) -> Result<Response, Error> {
2751×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
2761×
    let me = require(&viewer, &shell)?;
2771×
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;
2781×
    csrf(&viewer, &f.csrf_token, &shell)?;
2791×
    let run = fdb::run(&app.db, row.id, number)
2801×
        .await
2811×
        .map_err(|e| Error::from_db(shell.clone(), e))?
2821×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
2838×
    let reason: String = f.reason.chars().filter(|c| !c.is_control()).take(200).collect();
2841×
    let reason = reason.trim();
2851×
    if reason.is_empty() {
2860×
        return Ok(Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding("Say why the run is urgent."))).into_response());
2871×
    }
288
    // A cap on one owner's urgent runs at once (Admin > Runners >
289
    // Settings), so urgent can't starve everyone else.
2901×
    let cap: i64 = db::accounts::setting(&app.db, "flow_expedite_cap").await.ok().flatten().and_then(|v| v.parse().ok()).unwrap_or(2);
2911×
    let live = fdb::expedited_live(&app.db, &owner).await.map_err(|e| Error::from_db(shell.clone(), e))?;
2921×
    if live >= cap {
2930×
        let msg = format!("{owner} already has {live} expedited run(s) going, the most at once ({cap}); wait for one to end.");
2940×
        return Ok(Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response());
2951×
    }
2961×
    let note = format!("expedited by {}: {reason}", me.account.username);
2971×
    if fdb::expedite(&app.db, run.id, &note).await.map_err(|e| Error::from_db(shell.clone(), e))? {
2981×
        db::accounts::audit(&app.db, Some(&me.account), "flow.expedite", &format!("{owner}/{name}"), &format!("#{number}: {reason}"))
2991×
            .await
3001×
            .ok();
301
        // Every queued job's place changes now, not at the next pass
302
        // (#291): this run's page and the pages of the runs it moved down
303
        // hear of it.
3041×
        if let Err(e) = crate::flow::explain_waits(&app).await {
3050×
            tracing::warn!("flow: placing after an expedite: {e}");
3061×
        }
3071×
        app.bus.live(&crate::events::topic::run(run.id), "jobs");
3080×
    }
3091×
    Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response())
3101×
}
311
312
/// Percent-encode a message for a query string.
3130×
fn urlencoding(s: &str) -> String {
3140×
    s.bytes()
3150×
        .map(|b| match b {
3160×
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),
3170×
            _ => format!("%{b:02X}"),
3180×
        })
3190×
        .collect()
3200×
}
321
322
/// A job placed in the graph.
323
pub struct Node {
324
    pub j: fdb::Job,
325
    pub x: i64,
326
    pub y: i64,
327
    pub duration: String,
328
}
329
330
pub struct Edge {
331
    pub d: String,
332
    pub status: String,
333
}
334
335
pub struct Graph {
336
    pub nodes: Vec<Node>,
337
    pub edges: Vec<Edge>,
338
    pub width: i64,
339
    pub height: i64,
340
}
341
342
pub const NODE_W: i64 = 200;
343
pub const NODE_H: i64 = 48;
344
const COL_GAP: i64 = 72;
345
const ROW_GAP: i64 = 18;
346
const PAD: i64 = 12;
347
348
/// Columns by dependency depth, jobs in each column in name order.
34923×
pub fn layout(jobs: Vec<fdb::Job>) -> Graph {
35023×
    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();
35154×
    for j in jobs {
35254×
        by_depth.entry(j.depth).or_default().push(j);
35354×
    }
35423×
    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;
35523×
    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;
35623×
    let mut nodes = Vec::new();
35739×
    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {
35839×
        js.sort_by(|a, b| a.name.cmp(&b.name));
35939×
        let n = js.len() as i64;
36039×
        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;
36139×
        let top = (height - column_h) / 2;
36254×
        for (i, j) in js.into_iter().enumerate() {
36354×
            let duration = crate::flow::duration(&j);
36454×
            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });
36554×
        }
366
    }
36723×
    let mut edges = Vec::new();
36854×
    for n in &nodes {
36954×
        for d in n.j.depends_on() {
37020×
            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {
37118×
                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);
37218×
                let (x2, y2) = (n.x, n.y + NODE_H / 2);
37318×
                let mid = (x1 + x2) / 2;
37418×
                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });
37518×
            }
376
        }
377
    }
37823×
    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);
37923×
    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }
38023×
}
381
382179×
pub fn glyph(status: &str) -> &'static str {
383179×
    match status {
384179×
        "passed" => "✓",
385122×
        "failed" => "✕",
386109×
        "running" => "●",
38796×
        "skipped" => "⤼",
38895×
        "cancelled" => "⊘",
38986×
        "error" => "!",
39085×
        _ => "○",
391
    }
392179×
}
393
394
/// Failed tests a run page lists per job; the rest are counted.
395
const SHOWN_FAILURES: i64 = 50;
396
397
/// A stretch of a job's log as the run page shows it: a line, or a
398
/// section the job marked with `::group::TITLE` … `::endgroup::`
399
/// (oxygenflow#31), shown folded with how long it took.
400
pub enum Segment {
401
    Line(fdb::LogLine),
402
    Group(Section),
403
}
404
405
pub struct Section {
406
    pub title: String,
407
    /// When it began, and ended (`None` while it is still going).
408
    pub from_t: i64,
409
    pub to_t: Option<i64>,
410
    pub lines: Vec<fdb::LogLine>,
411
}
412
413
/// A log's lines in sections. A section left open is still going; a new
414
/// `::group::` ends the one before (sections do not nest).
41555×
pub fn sections(lines: &[fdb::LogLine]) -> Vec<Segment> {
41655×
    let mut out: Vec<Segment> = Vec::new();
41755×
    let mut open: Option<Section> = None;
41855×
    for l in lines {
41933×
        if let Some(title) = l.text.strip_prefix("::group::") {
4206×
            if let Some(g) = open.take() {
4211×
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
4225×
            }
4236×
            open = Some(Section { title: title.trim().to_string(), from_t: l.t, to_t: None, lines: Vec::new() });
42427×
        } else if l.text.trim() == "::endgroup::" {
4254×
            if let Some(g) = open.take() {
4264×
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
4274×
            }
428
        } else {
42923×
            match open.as_mut() {
4307×
                Some(g) => g.lines.push(l.clone()),
43116×
                None => out.push(Segment::Line(l.clone())),
432
            }
433
        }
434
    }
43555×
    if let Some(g) = open {
4361×
        out.push(Segment::Group(g));
43754×
    }
43855×
    out
43955×
}
440
441
/// Milliseconds as the log shows them: `12.3s`.
44229×
pub fn secs(ms: impl std::borrow::Borrow<i64>) -> String {
44329×
    let ms = *ms.borrow();
44429×
    format!("{}.{}s", ms / 1000, ms % 1000 / 100)
44529×
}
446
447
pub struct JobLog {
448
    pub j: fdb::Job,
449
    pub artifacts: Vec<fdb::Artifact>,
450
    pub lines: Vec<fdb::LogLine>,
451
    /// The lines in sections (oxygenflow#31).
452
    pub segments: Vec<Segment>,
453
    pub duration: String,
454
    /// The last line shown; the live view asks for what follows.
455
    pub last_seq: i64,
456
    /// The job's coverage report, when it uploaded one (#113): its id and percentage.
457
    pub coverage: Option<(i64, String)>,
458
    /// The job's test report, when it uploaded one (#114), and its first
459
    /// failures.
460
    pub tests: Option<crate::junit::Report>,
461
    pub failures: Vec<crate::junit::Case>,
462
}
463
464
#[derive(Template)]
465
#[template(path = "flow_run.html")]
466
pub struct RunPage {
467
    shell: Shell,
468
    repo: RepoShell,
469
    run: fdb::Run,
470
    graph: Graph,
471
    logs: Vec<JobLog>,
472
    duration: String,
473
    /// Whether the viewer may re-run it.
474
    can_rerun: bool,
475
    /// Whether the viewer may cancel it.
476
    can_cancel: bool,
477
    /// May raise the run to urgent (oxygenhub#244): maintainers, while it runs.
478
    can_expedite: bool,
479
    /// Where a live run stands among the runners' work.
480
    queue: Option<crate::flow::QueueInfo>,
481
    /// Releases a writer may attach this run's artifacts to.
482
    releases: Vec<db::releases::Release>,
483
    /// Jobs that wait, or waited, for a person (oxygenflow#57).
484
    approvals: Vec<ApprovalView>,
485
}
486
487
/// A job's approval as the run page shows it.
488
pub struct ApprovalView {
489
    pub job: String,
490
    pub by: String,
491
    pub others_only: bool,
492
    /// `waiting`, `approved`, `rejected` or `expired`.
493
    pub state: &'static str,
494
    pub who: String,
495
    pub when_ms: i64,
496
    /// Whether the viewer may approve or reject it now, else why not.
497
    pub may: Result<(), String>,
498
    /// It takes an approval signed with the approver's key (#235): the
499
    /// command that signs it, and the approval's hash once sealed.
500
    pub sealed: bool,
501
    pub command: String,
502
    pub seal: String,
503
}
504
505
impl ApprovalView {
5062×
    pub fn may_decide(&self) -> bool {
5072×
        self.may.is_ok()
5082×
    }
509
5102×
    pub fn why_not(&self) -> String {
5112×
        self.may.clone().err().unwrap_or_default()
5122×
    }
513
}
514
515
/// Lines of a job's log a page shows before "the rest is loaded as it comes".
516
const PAGE_LINES: i64 = 5000;
517
51824×
pub async fn run(
51924×
    State(app): State<Shared>,
52024×
    viewer: Viewer,
52124×
    Path((owner, name, id)): Path<(String, String, i64)>,
52224×
) -> Result<Page<RunPage>, Error> {
52324×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
52424×
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
52524×
    let run = fdb::run(&app.db, row.id, id)
52624×
        .await
52724×
        .map_err(|e| Error::from_db(shell.clone(), e))?
52824×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
52923×
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
53023×
    let reports = crate::coverage::for_run(&app.db, run.id).await.unwrap_or_default();
53123×
    let test_reports = crate::junit::for_run(&app.db, run.id).await.unwrap_or_default();
53223×
    let mut logs = Vec::new();
53354×
    for j in &jobs {
53454×
        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();
53554×
        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);
53654×
        let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default();
53754×
        let coverage = reports
53854×
            .iter()
53954×
            .find(|r| r.job_id == Some(j.id))
54054×
            .map(|r| (r.id, r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into())));
54154×
        let tests = test_reports.iter().find(|r| r.job_id == j.id).cloned();
54254×
        let failures = match &tests {
5431×
            Some(r) if r.broken() > 0 => crate::junit::failures(&app.db, r.id, SHOWN_FAILURES).await.unwrap_or_default(),
54453×
            _ => vec![],
545
        };
54654×
        let segments = sections(&lines);
54754×
        logs.push(JobLog {
54854×
            duration: crate::flow::duration(j),
54954×
            j: j.clone(),
55054×
            artifacts,
55154×
            lines,
55254×
            segments,
55354×
            last_seq,
55454×
            coverage,
55554×
            tests,
55654×
            failures,
55754×
        });
558
    }
55923×
    let duration = match run.finished_ms {
56011×
        Some(f) => crate::flow::human_ms(f - run.created_ms),
56112×
        None => crate::flow::human_ms(db::now_ms() - run.created_ms),
562
    };
56323×
    let mut approvals = Vec::new();
56454×
    for j in jobs.iter().filter(|j| !j.approval.is_empty()) {
5654×
        let Some(a) = serde_json::from_str::<flow_core::Job>(&j.spec).ok().and_then(|s| s.approval) else { continue };
5664×
        let state = match (j.approval.as_str(), j.status.as_str()) {
5674×
            ("waiting", "queued") => "waiting",
5682×
            ("approved", _) => "approved",
5690×
            (_, "cancelled") if !j.approved_by.is_empty() => "rejected",
5700×
            (_, "cancelled") if j.error.starts_with("no approval within") => "expired",
5710×
            _ => continue,
572
        };
5734×
        let may = match (state, viewer.account()) {
5744×
            ("waiting", Some(me)) => crate::flow::may_decide(&app, &row, &run, j, me).await.map(|_| ()),
5752×
            ("waiting", None) => Err("sign in to approve".into()),
5762×
            _ => Err(String::new()),
577
        };
5784×
        approvals.push(ApprovalView {
5794×
            sealed: a.sealed,
5804×
            command: format!("oxy seal --flow {}/{owner}/{name}/flow/{} --job {}", app.config.base_url(), run.number, j.name),
5814×
            seal: j.approval_seal.clone(),
5824×
            job: j.name.clone(),
5834×
            by: a.by,
5844×
            others_only: a.others_only,
5854×
            state,
5864×
            who: if j.approved_via.is_empty() { j.approved_by.clone() } else { format!("{} via {}", j.approved_by, j.approved_via) },
5874×
            when_ms: j.approved_ms.unwrap_or(0),
5884×
            may,
589
        });
590
    }
59123×
    let graph = layout(jobs);
59223×
    let can_write = viewer.account().is_some() && role >= Role::Write;
59323×
    let can_rerun = !run.live() && can_write && crate::flow::enabled(&app, &row);
59423×
    let can_cancel = run.live() && can_write;
59523×
    let can_expedite = run.live() && run.priority.is_empty() && viewer.account().is_some() && role >= Role::Maintain;
59623×
    let queue = if run.live() {
59712×
        let q = fdb::queue(&app.db).await.unwrap_or_default();
59812×
        Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some())
599
    } else {
60011×
        None
601
    };
60223×
    let releases = if viewer.account().is_some() && role >= Role::Write && logs.iter().any(|l| !l.artifacts.is_empty()) {
6031×
        db::releases::list(&app.db, row.id).await.unwrap_or_default()
604
    } else {
60522×
        vec![]
606
    };
60723×
    Ok(Page(RunPage {
60823×
        shell,
60923×
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
61023×
        run,
61123×
        graph,
61223×
        logs,
61323×
        duration,
61423×
        can_rerun,
61523×
        can_cancel,
61623×
        can_expedite,
61723×
        queue,
61823×
        releases,
61923×
        approvals,
62023×
    }))
62124×
}
622
623
#[derive(Deserialize)]
624
pub struct DecideForm {
625
    #[serde(rename = "_csrf")]
626
    csrf_token: String,
627
    /// `approve` or `reject`.
628
    action: String,
629
}
630
631
/// `POST /{owner}/{name}/flow/{id}/jobs/{job}/decide`: approve or reject
632
/// a job that waits for a person (oxygenflow#57).
6330×
pub async fn decide(
6340×
    State(app): State<Shared>,
6350×
    viewer: Viewer,
6360×
    Path((owner, name, number, job)): Path<(String, String, i64, String)>,
6370×
    Form(f): Form<DecideForm>,
6380×
) -> Result<Response, Error> {
6390×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
6400×
    let me = require(&viewer, &shell)?;
6410×
    let (_, row, _) = lookup(&app, &shell, &viewer, &owner, &name).await?;
6420×
    csrf(&viewer, &f.csrf_token, &shell)?;
6430×
    let run = fdb::run(&app.db, row.id, number)
6440×
        .await
6450×
        .map_err(|e| Error::from_db(shell.clone(), e))?
6460×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
6470×
    let approve = match f.action.as_str() {
6480×
        "approve" => true,
6490×
        "reject" => false,
6500×
        _ => return Err(Error::bad(shell, "approve or reject")),
651
    };
6520×
    match crate::flow::decide(&app, &row, &run, &job, &me.account, approve, None).await {
6530×
        Ok(_) => Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()),
6540×
        Err(why) => Err(Error::bad(shell, why)),
655
    }
6560×
}
657
658
#[derive(Deserialize)]
659
pub struct LiveQuery {
660
    /// `job:seq,job:seq`: the last log line each open job already shows.
661
    after: Option<String>,
662
}
663
664
/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the
665
/// run page while it is live.
6661×
pub async fn live(
6671×
    State(app): State<Shared>,
6681×
    viewer: Viewer,
6691×
    Path((owner, name, id)): Path<(String, String, i64)>,
6701×
    Query(q): Query<LiveQuery>,
6711×
) -> Result<Response, Error> {
6721×
    let shell = Shell::new(&app, "", &viewer);
6731×
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
6741×
    let run = fdb::run(&app.db, row.id, id)
6751×
        .await
6761×
        .map_err(|e| Error::from_db(shell.clone(), e))?
6771×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
6781×
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
6791×
    let after: std::collections::HashMap<i64, i64> = q
6801×
        .after
6811×
        .unwrap_or_default()
6821×
        .split(',')
6831×
        .filter_map(|p| p.split_once(':'))
6841×
        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))
6851×
        .collect();
6861×
    let mut out = Vec::new();
6873×
    for j in &jobs {
6883×
        let lines = match after.get(&j.id) {
6891×
            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),
6902×
            None => vec![],
691
        };
6923×
        out.push(serde_json::json!({
6933×
            "id": j.id,
6943×
            "name": j.name,
6953×
            "status": j.status,
6963×
            "glyph": glyph(&j.status),
6973×
            "duration": crate::flow::duration(j),
6983×
            "exit_code": j.exit_code,
6993×
            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),
700
        }));
701
    }
7021×
    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())
7031×
}
704
705
// ------------------------------------------------------------ state stamps
706
707
/// In rendered Markdown, the image path that is the viewed state's stamp.
708
pub const STAMP_IMAGE: &str = ".flow/stamp";
709
/// In rendered Markdown, the link path to the viewed state's run.
710
pub const RUN_LINK: &str = ".flow/run";
711
712
/// Resolve a ref to its state address, and whether the ref can move.
7138×
async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {
7148×
    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;
7158×
    let spec = r.to_string();
7168×
    app.repos
7178×
        .read(&repo, move |repo, cache| {
7188×
            let t = crate::repos::view::resolve(repo, cache, &spec)?;
7198×
            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);
7208×
            Ok((t.permalink().unwrap_or_default(), moves))
7218×
        })
7228×
        .await
7238×
}
724
725
/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).
7266×
pub async fn stamp(
7276×
    State(app): State<Shared>,
7286×
    viewer: Viewer,
7296×
    Path((owner, name, r)): Path<(String, String, String)>,
7306×
) -> Result<Response, Error> {
7316×
    let shell = Shell::new(&app, "", &viewer);
7326×
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
7336×
    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
7346×
    if moves && !heads.is_empty() {
7352×
        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();
7362×
        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));
7372×
        return Ok(res);
7384×
    }
7394×
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
7404×
    let jobs = match &run {
7413×
        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,
7421×
        None => vec![],
743
    };
7444×
    let svg = stamp_svg(&heads, run.as_ref(), &jobs);
7454×
    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };
7464×
    Ok((
7474×
        [
7484×
            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),
7494×
            (axum::http::header::CACHE_CONTROL, cache),
7504×
            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),
7514×
        ],
7524×
        svg,
7534×
    )
7544×
        .into_response())
7556×
}
756
757
/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.
7582×
pub async fn at(
7592×
    State(app): State<Shared>,
7602×
    viewer: Viewer,
7612×
    Path((owner, name, r)): Path<(String, String, String)>,
7622×
) -> Result<Response, Error> {
7632×
    let shell = Shell::new(&app, "", &viewer);
7642×
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
7652×
    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
7662×
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
7672×
    let to = match run {
7681×
        Some(r) => format!("/{owner}/{name}/flow/{}", r.number),
7691×
        None => format!("/{owner}/{name}/flow"),
770
    };
7712×
    Ok(axum::response::Redirect::to(&to).into_response())
7722×
}
773
77412×
fn xml(s: &str) -> String {
77512×
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
77612×
}
777
778
/// The stamp: the mark, one glyph per job, the status, the state and the
779
/// date; the same facts machine-readable in `<metadata>`. Fixed colours,
780
/// because an image does not follow the page's theme.
7814×
pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {
782
    const CH: f64 = 6.6; // IBM Plex Mono at 11px
7837×
    let colour = |s: &str| match s {
7847×
        "passed" => "#5fe3a1",
7855×
        "failed" | "error" => "#ff5e7a",
7861×
        "running" => "#00e0f0",
7871×
        _ => "#8592ad",
7887×
    };
7894×
    let short: String = heads.chars().take(12).collect();
7904×
    let (status, date) = match run {
7913×
        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),
7921×
        None => ("no run", String::new()),
793
    };
7944×
    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();
7954×
    ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name)));
7964×
    let label = "OXYGENFLOW";
7974×
    let label_w = 18.0 + label.len() as f64 * 7.4 + 8.0; // Barlow bold 10.5px, spaced
7984×
    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };
7994×
    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };
8004×
    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;
8014×
    let w = (label_w + right_w).ceil();
8024×
    let mut glyphs = String::new();
8034×
    for (i, j) in ordered.iter().enumerate() {
8043×
        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;
8053×
        glyphs.push_str(&format!(
8063×
            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,
8073×
            colour(&j.status),
8083×
            glyph(&j.status)
8093×
        ));
8103×
    }
8114×
    let meta = serde_json::json!({
81210×
        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),
8134×
        "run": run.map(|r| r.number),
8144×
        "status": status,
8154×
        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),
816
    });
8174×
    let title = match run {
8183×
        Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.number),
8191×
        None => format!("OxygenFlow: no run for state {short}"),
820
    };
8214×
    format!(
822
        r##"<svg xmlns="http://www.w3.org/2000/svg" width="{w}" height="22" viewBox="0 0 {w} 22" role="img" aria-label="{t}"><title>{t}</title><metadata>{m}</metadata><rect width="{w}" height="22" rx="4" fill="#0e1526"/><rect width="{lw:.1}" height="22" rx="4" fill="#0b3f9e"/><rect x="{lx:.1}" width="4" height="22" fill="#0b3f9e"/><circle cx="9" cy="11" r="3.2" fill="none" stroke="#00f0ff" stroke-width="1.6"/><text x="16" y="15" fill="#ffffff" font-family="Barlow, Verdana, sans-serif" font-size="10.5" font-weight="700" letter-spacing=".6">{label}</text><g font-family="IBM Plex Mono, ui-monospace, monospace" font-size="11">{glyphs}<text x="{tx:.1}" y="15" fill="{sc}">{tail}</text></g><rect x=".5" y=".5" width="{wi}" height="21" rx="3.5" fill="none" stroke="#1f2b47"/></svg>"##,
8234×
        t = xml(&title),
8244×
        m = xml(&meta.to_string()),
825
        lw = label_w,
8264×
        lx = label_w - 4.0,
8274×
        tx = label_w + 8.0 + glyphs_w,
8284×
        sc = colour(status),
8294×
        tail = xml(&tail),
8304×
        wi = w - 1.0,
831
    )
8324×
}
833
834
/// A receipt kept in a repository (Oxygen spec §14.6), as pages show it.
835
#[derive(Clone, Debug)]
836
pub struct ReceiptRow {
837
    pub attest: String,
838
    pub job: String,
839
    pub exit_code: u32,
840
    pub duration: String,
841
    pub runner: String,
842
    pub run: String,
843
    /// Why it does not verify, or that its runner's key was revoked.
844
    pub problem: String,
845
    pub made_ms: i64,
846
}
847
848
impl ReceiptRow {
8493×
    pub fn passed(&self) -> bool {
8503×
        self.exit_code == 0
8513×
    }
852
}
853
854
/// The receipts kept for the state of `heads`.
85513×
pub async fn receipts_of(app: &Shared, name: &crate::repos::RepoName, heads: Vec<oxy_core::hash::ChangeHash>) -> Vec<ReceiptRow> {
85613×
    let base = app.config.base_url();
85713×
    let found = app.repos.read(name, move |repo, _| Ok(repo.receipts(&heads)?)).await.unwrap_or_default();
85813×
    found
85913×
        .into_iter()
86013×
        .map(|i| {
8611×
            let r = &i.signed.receipt;
8621×
            let run = i.signed.run().to_string();
863
            ReceiptRow {
8641×
                attest: i.attest.to_hex(),
8651×
                job: r.job_name.clone(),
8661×
                exit_code: r.exit_code,
8671×
                duration: crate::flow::human_ms(r.duration_ms as i64),
8681×
                runner: i.signed.runner.name.clone(),
869
                // Only links to this hub are followed from the page.
8701×
                run: if run.starts_with(base) { run } else { String::new() },
8711×
                problem: match (i.problem, i.revoked) {
8720×
                    (Some(p), _) => p,
8730×
                    (None, true) => "the runner's key was revoked".into(),
8741×
                    (None, false) => String::new(),
875
                },
8761×
                made_ms: i.signed.attest.hlc.wall_ms as i64,
877
            }
8781×
        })
87913×
        .collect()
88013×
}
881
882
/// `GET /{owner}/{name}/receipts/{attest}`: a signed receipt, for
883
/// `oxy receipt verify` or `oxy receipt import`.
8842×
pub async fn receipt_file(
8852×
    State(app): State<Shared>,
8862×
    viewer: Viewer,
8872×
    Path((owner, name, attest)): Path<(String, String, String)>,
8882×
) -> Result<Response, Error> {
8892×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/receipts/{attest}"), &viewer);
8902×
    let (repo, _row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
8912×
    let Some(r) = oxy_core::hash::ObjectRef::from_hex(attest.trim_end_matches(".receipt")) else {
8920×
        return Err(Error::not_found(shell));
893
    };
8942×
    let (job, bytes) = app
8952×
        .repos
8962×
        .read(&repo, move |repo, _| {
8972×
            let i = repo.receipt(&r)?;
8981×
            Ok((i.signed.receipt.job_name.clone(), i.signed.encode()))
8992×
        })
9002×
        .await
9012×
        .map_err(|_| Error::not_found(shell.clone()))?;
9025×
    let file: String = job.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).collect();
9031×
    Ok((
9041×
        [
9051×
            (axum::http::header::CONTENT_TYPE, "application/oxy-receipt".to_string()),
9061×
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{file}-{}.receipt\"", r.short(12))),
9071×
        ],
9081×
        bytes,
9091×
    )
9101×
        .into_response())
9112×
}
912
913
/// `GET /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}`: an artifact
914
/// of a run's job, read from the repository.
9153×
pub async fn artifact_file(
9163×
    State(app): State<Shared>,
9173×
    viewer: Viewer,
9183×
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
9193×
) -> Result<Response, Error> {
9203×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}"), &viewer);
9213×
    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
9223×
    let run = fdb::run(&app.db, row.id, id)
9233×
        .await
9243×
        .map_err(|e| Error::from_db(shell.clone(), e))?
9253×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
9263×
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
9273×
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
9283×
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
9293×
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
9302×
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
9312×
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;
9322×
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
9332×
    Ok((
9342×
        [
9352×
            (axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()),
9362×
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}\"", file.replace('"', ""))),
9372×
        ],
9382×
        bytes,
9392×
    )
9402×
        .into_response())
9413×
}
942
943
/// `GET …/artifacts/{job}/{artifact}/provenance` (oxygenhub#239): the
944
/// artifact's SLSA provenance in a DSSE envelope signed by the hub, for
945
/// `cosign verify-blob-attestation --key oxyhub-signing.pem --type
946
/// slsaprovenance1`. Only for a job whose signed receipt the hub kept.
9472×
pub async fn provenance(
9482×
    State(app): State<Shared>,
9492×
    viewer: Viewer,
9502×
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
9512×
) -> Result<Response, Error> {
9522×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/provenance"), &viewer);
9532×
    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
9542×
    let run = fdb::run(&app.db, row.id, id)
9552×
        .await
9562×
        .map_err(|e| Error::from_db(shell.clone(), e))?
9572×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
9582×
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
9592×
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
9602×
    if j.receipt_attest.is_empty() {
9611×
        return Err(Error::bad(shell, "This job has no signed receipt the hub kept, so the hub states no provenance for it.".to_string()));
9621×
    }
9631×
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
9641×
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
9651×
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
9661×
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;
9671×
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
9681×
    let base = app.config.base_url();
9691×
    let subject = crate::provenance::Subject {
9701×
        name: &file,
9711×
        bytes: &bytes,
9721×
        repo: &format!("{owner}/{name}"),
9731×
        run_url: &format!("{base}/{owner}/{name}/flow/{id}"),
9741×
        bookmark: &run.bookmark,
9751×
        state: &run.state,
9761×
        heads: &run.heads,
9771×
        pipeline_hash: &run.pipeline_hash,
9781×
        job: &j.name,
9791×
        runner: &j.runner,
9801×
        receipt_signer: &j.receipt_signer,
9811×
        receipt_attest: &j.receipt_attest,
9821×
        started_ms: j.started_ms,
9831×
        finished_ms: j.finished_ms,
9841×
    };
9851×
    let statement = crate::provenance::statement(base, &subject);
9861×
    let envelope = crate::provenance::envelope(&app, &statement).map_err(|e| Error::bad(shell.clone(), e.to_string()))?;
9871×
    Ok((
9881×
        [
9891×
            (axum::http::header::CONTENT_TYPE, "application/json".to_string()),
9901×
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}.intoto.json\"", file.replace('"', ""))),
9911×
        ],
9921×
        serde_json::to_string(&envelope).unwrap_or_default(),
9931×
    )
9941×
        .into_response())
9952×
}
996
997
/// `GET /.well-known/oxyhub-signing.pem`: the hub's Ed25519 public key, as
998
/// cosign and other verifiers take it (oxygenhub#239).
9991×
pub async fn signing_key(State(app): State<Shared>) -> Response {
10001×
    ([(axum::http::header::CONTENT_TYPE, "application/x-pem-file")], crate::provenance::public_pem(&app)).into_response()
10011×
}
1002
1003
#[derive(Deserialize)]
1004
pub struct AttachForm {
1005
    _csrf: String,
1006
    release: i64,
1007
}
1008
1009
/// `POST /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/attach`: add
1010
/// an artifact to a release as an asset, named after its path; the release
1011
/// page then shows the run and receipt that built it.
10121×
pub async fn attach(
10131×
    State(app): State<Shared>,
10141×
    viewer: Viewer,
10151×
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
10161×
    Form(f): Form<AttachForm>,
10171×
) -> Result<Response, Error> {
10181×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
10191×
    require(&viewer, &shell)?;
10201×
    csrf(&viewer, &f._csrf, &shell)?;
10211×
    let (repo, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
10221×
    let run = fdb::run(&app.db, row.id, id)
10231×
        .await
10241×
        .map_err(|e| Error::from_db(shell.clone(), e))?
10251×
        .ok_or_else(|| Error::not_found(shell.clone()))?;
10261×
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
10271×
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
1028
    // A failing job's `when "always"` artifacts are not in its receipt, so
1029
    // they cannot stand for a release (oxygenflow#37).
10301×
    if j.status != "passed" {
10310×
        return Err(Error::bad(shell, "only a passing job's artifacts can be attached to a release"));
10321×
    }
10331×
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
10341×
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
10351×
    let releases = db::releases::list(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
10361×
    let Some(rel) = releases.into_iter().find(|r| r.id == f.release) else { return Err(Error::not_found(shell)) };
10371×
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
10381×
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
10391×
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
10401×
    super::releases::store_asset(&app, rel.id, &file, "application/octet-stream", &bytes)
10411×
        .await
10421×
        .map_err(|m| Error::bad(shell.clone(), m))?;
10431×
    Ok(Redirect::to(&format!("/{owner}/{name}/releases/tag/{}", rel.tag)).into_response())
10441×
}
1045
1046
/// `/{owner}/{name}/flow/badge.svg`: the latest run on the default
1047
/// bookmark as a small SVG for a README, for whoever may read the
1048
/// repository. `passed` green, `failed` or `error` red, a run still going
1049
/// blue, none grey.
10501×
pub async fn badge(
10511×
    State(app): State<Shared>,
10521×
    viewer: Viewer,
10531×
    Path((owner, name)): Path<(String, String)>,
10541×
) -> Result<axum::response::Response, Error> {
1055
    use axum::response::IntoResponse;
10561×
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
10571×
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
10581×
    let latest = fdb::latest_on(&app.db, row.id, &row.default_bookmark).await.map_err(|e| Error::from_db(shell.clone(), e))?;
10591×
    let (text, colour) = match latest.as_ref().map(|r| r.status.as_str()) {
10600×
        Some("passed") => ("passing", "#2da44e"),
10610×
        Some("failed") | Some("error") => ("failing", "#cf222e"),
10620×
        Some("cancelled") => ("cancelled", "#6e7781"),
10630×
        Some(_) => ("running", "#0969da"),
10641×
        None => ("no runs", "#6e7781"),
1065
    };
10661×
    let label = "OxygenFlow";
10671×
    let (lw, tw) = (label.len() as u32 * 7 + 10, text.len() as u32 * 7 + 10);
10681×
    let svg = format!(
1069
        r##"<svg xmlns="http://www.w3.org/2000/svg" width="{w}" height="20" role="img" aria-label="{label}: {text}"><title>{label}: {text}</title><rect width="{lw}" height="20" rx="3" fill="#555"/><rect x="{lw}" width="{tw}" height="20" rx="3" fill="{colour}"/><rect x="{lw}" width="4" height="20" fill="{colour}"/><g fill="#fff" font-family="Verdana,DejaVu Sans,sans-serif" font-size="11" text-anchor="middle"><text x="{lx}" y="14">{label}</text><text x="{tx}" y="14">{text}</text></g></svg>"##,
10701×
        w = lw + tw,
10711×
        lx = lw / 2,
10721×
        tx = lw + tw / 2,
1073
    );
10741×
    Ok((
10751×
        [(axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"), (axum::http::header::CACHE_CONTROL, "no-cache, max-age=60")],
10761×
        svg,
10771×
    )
10781×
        .into_response())
10791×
}
1080
1081
#[cfg(test)]
1082
mod section_tests {
1083
    use super::*;
1084
10859×
    fn line(seq: i64, t: i64, text: &str) -> fdb::LogLine {
10869×
        fdb::LogLine { seq, t, stream: "out".into(), text: text.into() }
10879×
    }
1088
1089
    #[test]
10901×
    fn marked_lines_fold_into_sections() {
10911×
        let lines = vec![
10921×
            line(1, 0, "booting"),
10931×
            line(2, 100, "::group::cargo build"),
10941×
            line(3, 200, "Compiling a"),
10951×
            line(4, 900, "Compiling b"),
10961×
            line(5, 4_300, "::endgroup::"),
10971×
            line(6, 4_400, "::group::cargo test"),
10981×
            line(7, 4_500, "test x ... ok"),
10991×
            line(8, 5_000, "::group::lint"),
11001×
            line(9, 5_100, "clean"),
1101
        ];
11021×
        let segs = sections(&lines);
11031×
        assert_eq!(segs.len(), 4);
11041×
        assert!(matches!(&segs[0], Segment::Line(l) if l.text == "booting"));
11051×
        let Segment::Group(build) = &segs[1] else { panic!("a section") };
11061×
        assert_eq!((build.title.as_str(), build.from_t, build.to_t, build.lines.len()), ("cargo build", 100, Some(4_300), 2));
11071×
        let Segment::Group(test) = &segs[2] else { panic!("a section") };
11081×
        assert_eq!(test.to_t, Some(5_000), "a new section ends the one before");
11091×
        let Segment::Group(lint) = &segs[3] else { panic!("a section") };
11101×
        assert_eq!((lint.to_t, lint.lines.len()), (None, 1), "still going");
11111×
        assert_eq!(secs(4_200), "4.2s");
11121×
    }
1113
}