OXYGENHUB
◆ v0.5.1 oxygenhub / crates / hub / src / web / flow.rs
36.0 KiB · Rust · last changed Log sections: ::group:: folds a job's output on the run page 21 days ago source blame history raw
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.
47 
pub fn why_not_run(enabled: bool, signed_in: bool, role: Role) -> Option<String> {
48 
    if !enabled {
49 
        Some("Flow is off for this repository; a site administrator can turn it on in Settings → Flow.".into())
50 
    } else if !signed_in {
51 
        Some("Sign in with write access to run the pipeline.".into())
52 
    } else if role < Role::Write {
53 
        Some("Running the pipeline needs write access to this repository.".into())
54 
    } else {
55 
        None
56 
    }
57 
}
58
59 
const PER_PAGE: i64 = 30;
60
61 
pub async fn runs(
62 
    State(app): State<Shared>,
63 
    viewer: Viewer,
64 
    Path((owner, name)): Path<(String, String)>,
65 
    Query(q): Query<RunsQuery>,
66 
) -> Result<Page<RunsPage>, Error> {
67 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
68 
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
69 
    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;
70 
    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);
71 
    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };
72 
    let mut runs = Vec::new();
73 
    for r in list.into_iter().take(PER_PAGE as usize) {
74 
        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };
75 
        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();
76 
        let duration = match r.finished_ms {
77 
            Some(f) => crate::flow::human_ms(f - r.created_ms),
78 
            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),
79 
            None => "–".into(),
80 
        };
81 
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });
82 
    }
83 
    let enabled = crate::flow::enabled(&app, &row);
84 
    let why_not = why_not_run(enabled, viewer.account().is_some(), role);
85 
    let can_run = why_not.is_none();
86 
    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };
87 
    Ok(Page(RunsPage {
88 
        shell,
89 
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
90 
        runs,
91 
        enabled,
92 
        older,
93 
        can_run,
94 
        bookmarks,
95 
        error: q.error,
96 
        why_not,
97 
    }))
98 
}
99
100 
/// The repository's bookmarks, its default first.
101 
async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {
102 
    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };
103 
    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();
104 
    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));
105 
    names
106 
}
107
108 
#[derive(Deserialize)]
109 
pub struct StartForm {
110 
    #[serde(rename = "_csrf")]
111 
    csrf_token: String,
112 
    #[serde(default)]
113 
    bookmark: String,
114 
    /// A change's full hash: run the state whose only head it is.
115 
    #[serde(default)]
116 
    change: String,
117 
}
118
119 
/// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark's
120 
/// current state, or on one committed change.
121 
pub async fn start(
122 
    State(app): State<Shared>,
123 
    viewer: Viewer,
124 
    Path((owner, name)): Path<(String, String)>,
125 
    Form(f): Form<StartForm>,
126 
) -> Result<Response, Error> {
127 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
128 
    let me = require(&viewer, &shell)?;
129 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
130 
    csrf(&viewer, &f.csrf_token, &shell)?;
131 
    let (started, what) = if f.change.trim().is_empty() {
132 
        (crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username, &[]).await, f.bookmark.trim().to_string())
133 
    } else {
134 
        (crate::flow::run_change(&app, &row, &f.change, &me.account.username, &[]).await, format!("change {}", f.change.trim()))
135 
    };
136 
    Ok(match started {
137 
        Ok(n) => {
138 
            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {what}")).await.ok();
139 
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
140 
        }
141 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
142 
    })
143 
}
144
145 
#[derive(Deserialize)]
146 
pub struct RerunForm {
147 
    #[serde(rename = "_csrf")]
148 
    csrf_token: String,
149 
}
150
151 
/// `POST /{owner}/{name}/flow/{number}/rerun`.
152 
pub async fn rerun(
153 
    State(app): State<Shared>,
154 
    viewer: Viewer,
155 
    Path((owner, name, number)): Path<(String, String, i64)>,
156 
    Form(f): Form<RerunForm>,
157 
) -> Result<Response, Error> {
158 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
159 
    let me = require(&viewer, &shell)?;
160 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
161 
    csrf(&viewer, &f.csrf_token, &shell)?;
162 
    let run = fdb::run(&app.db, row.id, number)
163 
        .await
164 
        .map_err(|e| Error::from_db(shell.clone(), e))?
165 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
166 
    Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await {
167 
        Ok(n) => {
168 
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}"))
169 
                .await
170 
                .ok();
171 
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
172 
        }
173 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
174 
    })
175 
}
176
177 
/// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31).
178 
pub async fn rerun_job(
179 
    State(app): State<Shared>,
180 
    viewer: Viewer,
181 
    Path((owner, name, number, job)): Path<(String, String, i64, String)>,
182 
    Form(f): Form<RerunForm>,
183 
) -> Result<Response, Error> {
184 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
185 
    let me = require(&viewer, &shell)?;
186 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
187 
    csrf(&viewer, &f.csrf_token, &shell)?;
188 
    let run = fdb::run(&app.db, row.id, number)
189 
        .await
190 
        .map_err(|e| Error::from_db(shell.clone(), e))?
191 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
192 
    Ok(match crate::flow::rerun_job(&app, &row, &run, &job, &me.account.username).await {
193 
        Ok(_) => {
194 
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("{job} of #{number}"))
195 
                .await
196 
                .ok();
197 
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
198 
        }
199 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
200 
    })
201 
}
202
203 
/// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32).
204 
pub async fn cancel(
205 
    State(app): State<Shared>,
206 
    viewer: Viewer,
207 
    Path((owner, name, number)): Path<(String, String, i64)>,
208 
    Form(f): Form<RerunForm>,
209 
) -> Result<Response, Error> {
210 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
211 
    let me = require(&viewer, &shell)?;
212 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
213 
    csrf(&viewer, &f.csrf_token, &shell)?;
214 
    let run = fdb::run(&app.db, row.id, number)
215 
        .await
216 
        .map_err(|e| Error::from_db(shell.clone(), e))?
217 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
218 
    Ok(match crate::flow::cancel(&app, &row, &run, &me.account.username).await {
219 
        Ok(()) => {
220 
            db::accounts::audit(&app.db, Some(&me.account), "flow.cancel", &format!("{owner}/{name}"), &format!("#{number}")).await.ok();
221 
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
222 
        }
223 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
224 
    })
225 
}
226
227 
/// Percent-encode a message for a query string.
228 
fn urlencoding(s: &str) -> String {
229 
    s.bytes()
230 
        .map(|b| match b {
231 
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),
232 
            _ => format!("%{b:02X}"),
233 
        })
234 
        .collect()
235 
}
236
237 
/// A job placed in the graph.
238 
pub struct Node {
239 
    pub j: fdb::Job,
240 
    pub x: i64,
241 
    pub y: i64,
242 
    pub duration: String,
243 
}
244
245 
pub struct Edge {
246 
    pub d: String,
247 
    pub status: String,
248 
}
249
250 
pub struct Graph {
251 
    pub nodes: Vec<Node>,
252 
    pub edges: Vec<Edge>,
253 
    pub width: i64,
254 
    pub height: i64,
255 
}
256
257 
pub const NODE_W: i64 = 200;
258 
pub const NODE_H: i64 = 48;
259 
const COL_GAP: i64 = 72;
260 
const ROW_GAP: i64 = 18;
261 
const PAD: i64 = 12;
262
263 
/// Columns by dependency depth, jobs in each column in name order.
264 
pub fn layout(jobs: Vec<fdb::Job>) -> Graph {
265 
    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();
266 
    for j in jobs {
267 
        by_depth.entry(j.depth).or_default().push(j);
268 
    }
269 
    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;
270 
    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;
271 
    let mut nodes = Vec::new();
272 
    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {
273 
        js.sort_by(|a, b| a.name.cmp(&b.name));
274 
        let n = js.len() as i64;
275 
        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;
276 
        let top = (height - column_h) / 2;
277 
        for (i, j) in js.into_iter().enumerate() {
278 
            let duration = crate::flow::duration(&j);
279 
            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });
280 
        }
281 
    }
282 
    let mut edges = Vec::new();
283 
    for n in &nodes {
284 
        for d in n.j.depends_on() {
285 
            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {
286 
                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);
287 
                let (x2, y2) = (n.x, n.y + NODE_H / 2);
288 
                let mid = (x1 + x2) / 2;
289 
                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });
290 
            }
291 
        }
292 
    }
293 
    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);
294 
    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }
295 
}
296
297 
pub fn glyph(status: &str) -> &'static str {
298 
    match status {
299 
        "passed" => "✓",
300 
        "failed" => "✕",
301 
        "running" => "●",
302 
        "skipped" => "⤼",
303 
        "cancelled" => "⊘",
304 
        "error" => "!",
305 
        _ => "○",
306 
    }
307 
}
308
309 
/// Failed tests a run page lists per job; the rest are counted.
310 
const SHOWN_FAILURES: i64 = 50;
311
312 
/// A stretch of a job's log as the run page shows it: a line, or a
313 
/// section the job marked with `::group::TITLE` … `::endgroup::`
314 
/// (oxygenflow#31), shown folded with how long it took.
315 
pub enum Segment {
316 
    Line(fdb::LogLine),
317 
    Group(Section),
318 
}
319
320 
pub struct Section {
321 
    pub title: String,
322 
    /// When it began, and ended (`None` while it is still going).
323 
    pub from_t: i64,
324 
    pub to_t: Option<i64>,
325 
    pub lines: Vec<fdb::LogLine>,
326 
}
327
328 
/// A log's lines in sections. A section left open is still going; a new
329 
/// `::group::` ends the one before (sections do not nest).
330 
pub fn sections(lines: &[fdb::LogLine]) -> Vec<Segment> {
331 
    let mut out: Vec<Segment> = Vec::new();
332 
    let mut open: Option<Section> = None;
333 
    for l in lines {
334 
        if let Some(title) = l.text.strip_prefix("::group::") {
335 
            if let Some(g) = open.take() {
336 
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
337 
            }
338 
            open = Some(Section { title: title.trim().to_string(), from_t: l.t, to_t: None, lines: Vec::new() });
339 
        } else if l.text.trim() == "::endgroup::" {
340 
            if let Some(g) = open.take() {
341 
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
342 
            }
343 
        } else {
344 
            match open.as_mut() {
345 
                Some(g) => g.lines.push(l.clone()),
346 
                None => out.push(Segment::Line(l.clone())),
347 
            }
348 
        }
349 
    }
350 
    if let Some(g) = open {
351 
        out.push(Segment::Group(g));
352 
    }
353 
    out
354 
}
355
356 
/// Milliseconds as the log shows them: `12.3s`.
357 
pub fn secs(ms: impl std::borrow::Borrow<i64>) -> String {
358 
    let ms = *ms.borrow();
359 
    format!("{}.{}s", ms / 1000, ms % 1000 / 100)
360 
}
361
362 
pub struct JobLog {
363 
    pub j: fdb::Job,
364 
    pub artifacts: Vec<fdb::Artifact>,
365 
    pub lines: Vec<fdb::LogLine>,
366 
    /// The lines in sections (oxygenflow#31).
367 
    pub segments: Vec<Segment>,
368 
    pub duration: String,
369 
    /// The last line shown; the live view asks for what follows.
370 
    pub last_seq: i64,
371 
    /// The job's coverage report, when it uploaded one (#113): its id and percentage.
372 
    pub coverage: Option<(i64, String)>,
373 
    /// The job's test report, when it uploaded one (#114), and its first
374 
    /// failures.
375 
    pub tests: Option<crate::junit::Report>,
376 
    pub failures: Vec<crate::junit::Case>,
377 
}
378
379 
#[derive(Template)]
380 
#[template(path = "flow_run.html")]
381 
pub struct RunPage {
382 
    shell: Shell,
383 
    repo: RepoShell,
384 
    run: fdb::Run,
385 
    graph: Graph,
386 
    logs: Vec<JobLog>,
387 
    duration: String,
388 
    /// Whether the viewer may re-run it.
389 
    can_rerun: bool,
390 
    /// Whether the viewer may cancel it.
391 
    can_cancel: bool,
392 
    /// Where a live run stands among the runners' work.
393 
    queue: Option<crate::flow::QueueInfo>,
394 
    /// Releases a writer may attach this run's artifacts to.
395 
    releases: Vec<db::releases::Release>,
396 
}
397
398 
/// Lines of a job's log a page shows before "the rest is loaded as it comes".
399 
const PAGE_LINES: i64 = 5000;
400
401 
pub async fn run(
402 
    State(app): State<Shared>,
403 
    viewer: Viewer,
404 
    Path((owner, name, id)): Path<(String, String, i64)>,
405 
) -> Result<Page<RunPage>, Error> {
406 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
407 
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
408 
    let run = fdb::run(&app.db, row.id, id)
409 
        .await
410 
        .map_err(|e| Error::from_db(shell.clone(), e))?
411 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
412 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
413 
    let reports = crate::coverage::for_run(&app.db, run.id).await.unwrap_or_default();
414 
    let test_reports = crate::junit::for_run(&app.db, run.id).await.unwrap_or_default();
415 
    let mut logs = Vec::new();
416 
    for j in &jobs {
417 
        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();
418 
        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);
419 
        let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default();
420 
        let coverage = reports
421 
            .iter()
422 
            .find(|r| r.job_id == Some(j.id))
423 
            .map(|r| (r.id, r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into())));
424 
        let tests = test_reports.iter().find(|r| r.job_id == j.id).cloned();
425 
        let failures = match &tests {
426 
            Some(r) if r.broken() > 0 => crate::junit::failures(&app.db, r.id, SHOWN_FAILURES).await.unwrap_or_default(),
427 
            _ => vec![],
428 
        };
429 
        let segments = sections(&lines);
430 
        logs.push(JobLog {
431 
            duration: crate::flow::duration(j),
432 
            j: j.clone(),
433 
            artifacts,
434 
            lines,
435 
            segments,
436 
            last_seq,
437 
            coverage,
438 
            tests,
439 
            failures,
440 
        });
441 
    }
442 
    let duration = match run.finished_ms {
443 
        Some(f) => crate::flow::human_ms(f - run.created_ms),
444 
        None => crate::flow::human_ms(db::now_ms() - run.created_ms),
445 
    };
446 
    let graph = layout(jobs);
447 
    let can_write = viewer.account().is_some() && role >= Role::Write;
448 
    let can_rerun = !run.live() && can_write && crate::flow::enabled(&app, &row);
449 
    let can_cancel = run.live() && can_write;
450 
    let queue = if run.live() {
451 
        let q = fdb::queue(&app.db).await.unwrap_or_default();
452 
        Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some())
453 
    } else {
454 
        None
455 
    };
456 
    let releases = if viewer.account().is_some() && role >= Role::Write && logs.iter().any(|l| !l.artifacts.is_empty()) {
457 
        db::releases::list(&app.db, row.id).await.unwrap_or_default()
458 
    } else {
459 
        vec![]
460 
    };
461 
    Ok(Page(RunPage {
462 
        shell,
463 
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
464 
        run,
465 
        graph,
466 
        logs,
467 
        duration,
468 
        can_rerun,
469 
        can_cancel,
470 
        queue,
471 
        releases,
472 
    }))
473 
}
474
475 
#[derive(Deserialize)]
476 
pub struct LiveQuery {
477 
    /// `job:seq,job:seq`: the last log line each open job already shows.
478 
    after: Option<String>,
479 
}
480
481 
/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the
482 
/// run page while it is live.
483 
pub async fn live(
484 
    State(app): State<Shared>,
485 
    viewer: Viewer,
486 
    Path((owner, name, id)): Path<(String, String, i64)>,
487 
    Query(q): Query<LiveQuery>,
488 
) -> Result<Response, Error> {
489 
    let shell = Shell::new(&app, "", &viewer);
490 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
491 
    let run = fdb::run(&app.db, row.id, id)
492 
        .await
493 
        .map_err(|e| Error::from_db(shell.clone(), e))?
494 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
495 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
496 
    let after: std::collections::HashMap<i64, i64> = q
497 
        .after
498 
        .unwrap_or_default()
499 
        .split(',')
500 
        .filter_map(|p| p.split_once(':'))
501 
        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))
502 
        .collect();
503 
    let mut out = Vec::new();
504 
    for j in &jobs {
505 
        let lines = match after.get(&j.id) {
506 
            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),
507 
            None => vec![],
508 
        };
509 
        out.push(serde_json::json!({
510 
            "id": j.id,
511 
            "name": j.name,
512 
            "status": j.status,
513 
            "glyph": glyph(&j.status),
514 
            "duration": crate::flow::duration(j),
515 
            "exit_code": j.exit_code,
516 
            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),
517 
        }));
518 
    }
519 
    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())
520 
}
521
522 
// ------------------------------------------------------------ state stamps
523
524 
/// In rendered Markdown, the image path that is the viewed state's stamp.
525 
pub const STAMP_IMAGE: &str = ".flow/stamp";
526 
/// In rendered Markdown, the link path to the viewed state's run.
527 
pub const RUN_LINK: &str = ".flow/run";
528
529 
/// Resolve a ref to its state address, and whether the ref can move.
530 
async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {
531 
    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;
532 
    let spec = r.to_string();
533 
    app.repos
534 
        .read(&repo, move |repo, cache| {
535 
            let t = crate::repos::view::resolve(repo, cache, &spec)?;
536 
            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);
537 
            Ok((t.permalink().unwrap_or_default(), moves))
538 
        })
539 
        .await
540 
}
541
542 
/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).
543 
pub async fn stamp(
544 
    State(app): State<Shared>,
545 
    viewer: Viewer,
546 
    Path((owner, name, r)): Path<(String, String, String)>,
547 
) -> Result<Response, Error> {
548 
    let shell = Shell::new(&app, "", &viewer);
549 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
550 
    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
551 
    if moves && !heads.is_empty() {
552 
        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();
553 
        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));
554 
        return Ok(res);
555 
    }
556 
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
557 
    let jobs = match &run {
558 
        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,
559 
        None => vec![],
560 
    };
561 
    let svg = stamp_svg(&heads, run.as_ref(), &jobs);
562 
    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };
563 
    Ok((
564 
        [
565 
            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),
566 
            (axum::http::header::CACHE_CONTROL, cache),
567 
            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),
568 
        ],
569 
        svg,
570 
    )
571 
        .into_response())
572 
}
573
574 
/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.
575 
pub async fn at(
576 
    State(app): State<Shared>,
577 
    viewer: Viewer,
578 
    Path((owner, name, r)): Path<(String, String, String)>,
579 
) -> Result<Response, Error> {
580 
    let shell = Shell::new(&app, "", &viewer);
581 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
582 
    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
583 
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
584 
    let to = match run {
585 
        Some(r) => format!("/{owner}/{name}/flow/{}", r.number),
586 
        None => format!("/{owner}/{name}/flow"),
587 
    };
588 
    Ok(axum::response::Redirect::to(&to).into_response())
589 
}
590
591 
fn xml(s: &str) -> String {
592 
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
593 
}
594
595 
/// The stamp: the mark, one glyph per job, the status, the state and the
596 
/// date; the same facts machine-readable in `<metadata>`. Fixed colours,
597 
/// because an image does not follow the page's theme.
598 
pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {
599 
    const CH: f64 = 6.6; // IBM Plex Mono at 11px
600 
    let colour = |s: &str| match s {
601 
        "passed" => "#5fe3a1",
602 
        "failed" | "error" => "#ff5e7a",
603 
        "running" => "#00e0f0",
604 
        _ => "#8592ad",
605 
    };
606 
    let short: String = heads.chars().take(12).collect();
607 
    let (status, date) = match run {
608 
        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),
609 
        None => ("no run", String::new()),
610 
    };
611 
    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();
612 
    ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name)));
613 
    let label = "OXYGENFLOW";
614 
    let label_w = 18.0 + label.len() as f64 * 7.4 + 8.0; // Barlow bold 10.5px, spaced
615 
    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };
616 
    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };
617 
    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;
618 
    let w = (label_w + right_w).ceil();
619 
    let mut glyphs = String::new();
620 
    for (i, j) in ordered.iter().enumerate() {
621 
        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;
622 
        glyphs.push_str(&format!(
623 
            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,
624 
            colour(&j.status),
625 
            glyph(&j.status)
626 
        ));
627 
    }
628 
    let meta = serde_json::json!({
629 
        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),
630 
        "run": run.map(|r| r.number),
631 
        "status": status,
632 
        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),
633 
    });
634 
    let title = match run {
635 
        Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.number),
636 
        None => format!("OxygenFlow: no run for state {short}"),
637 
    };
638 
    format!(
639 
        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>"##,
640 
        t = xml(&title),
641 
        m = xml(&meta.to_string()),
642 
        lw = label_w,
643 
        lx = label_w - 4.0,
644 
        tx = label_w + 8.0 + glyphs_w,
645 
        sc = colour(status),
646 
        tail = xml(&tail),
647 
        wi = w - 1.0,
648 
    )
649 
}
650
651 
/// A receipt kept in a repository (Oxygen spec §14.6), as pages show it.
652 
#[derive(Clone, Debug)]
653 
pub struct ReceiptRow {
654 
    pub attest: String,
655 
    pub job: String,
656 
    pub exit_code: u32,
657 
    pub duration: String,
658 
    pub runner: String,
659 
    pub run: String,
660 
    /// Why it does not verify, or that its runner's key was revoked.
661 
    pub problem: String,
662 
    pub made_ms: i64,
663 
}
664
665 
impl ReceiptRow {
666 
    pub fn passed(&self) -> bool {
667 
        self.exit_code == 0
668 
    }
669 
}
670
671 
/// The receipts kept for the state of `heads`.
672 
pub async fn receipts_of(app: &Shared, name: &crate::repos::RepoName, heads: Vec<oxy_core::hash::ChangeHash>) -> Vec<ReceiptRow> {
673 
    let base = app.config.base_url();
674 
    let found = app.repos.read(name, move |repo, _| Ok(repo.receipts(&heads)?)).await.unwrap_or_default();
675 
    found
676 
        .into_iter()
677 
        .map(|i| {
678 
            let r = &i.signed.receipt;
679 
            let run = i.signed.run().to_string();
680 
            ReceiptRow {
681 
                attest: i.attest.to_hex(),
682 
                job: r.job_name.clone(),
683 
                exit_code: r.exit_code,
684 
                duration: crate::flow::human_ms(r.duration_ms as i64),
685 
                runner: i.signed.runner.name.clone(),
686 
                // Only links to this hub are followed from the page.
687 
                run: if run.starts_with(base) { run } else { String::new() },
688 
                problem: match (i.problem, i.revoked) {
689 
                    (Some(p), _) => p,
690 
                    (None, true) => "the runner's key was revoked".into(),
691 
                    (None, false) => String::new(),
692 
                },
693 
                made_ms: i.signed.attest.hlc.wall_ms as i64,
694 
            }
695 
        })
696 
        .collect()
697 
}
698
699 
/// `GET /{owner}/{name}/receipts/{attest}`: a signed receipt, for
700 
/// `oxy receipt verify` or `oxy receipt import`.
701 
pub async fn receipt_file(
702 
    State(app): State<Shared>,
703 
    viewer: Viewer,
704 
    Path((owner, name, attest)): Path<(String, String, String)>,
705 
) -> Result<Response, Error> {
706 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/receipts/{attest}"), &viewer);
707 
    let (repo, _row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
708 
    let Some(r) = oxy_core::hash::ObjectRef::from_hex(attest.trim_end_matches(".receipt")) else {
709 
        return Err(Error::not_found(shell));
710 
    };
711 
    let (job, bytes) = app
712 
        .repos
713 
        .read(&repo, move |repo, _| {
714 
            let i = repo.receipt(&r)?;
715 
            Ok((i.signed.receipt.job_name.clone(), i.signed.encode()))
716 
        })
717 
        .await
718 
        .map_err(|_| Error::not_found(shell.clone()))?;
719 
    let file: String = job.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).collect();
720 
    Ok((
721 
        [
722 
            (axum::http::header::CONTENT_TYPE, "application/oxy-receipt".to_string()),
723 
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{file}-{}.receipt\"", r.short(12))),
724 
        ],
725 
        bytes,
726 
    )
727 
        .into_response())
728 
}
729
730 
/// `GET /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}`: an artifact
731 
/// of a run's job, read from the repository.
732 
pub async fn artifact_file(
733 
    State(app): State<Shared>,
734 
    viewer: Viewer,
735 
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
736 
) -> Result<Response, Error> {
737 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}"), &viewer);
738 
    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
739 
    let run = fdb::run(&app.db, row.id, id)
740 
        .await
741 
        .map_err(|e| Error::from_db(shell.clone(), e))?
742 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
743 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
744 
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
745 
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
746 
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
747 
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
748 
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;
749 
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
750 
    Ok((
751 
        [
752 
            (axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()),
753 
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}\"", file.replace('"', ""))),
754 
        ],
755 
        bytes,
756 
    )
757 
        .into_response())
758 
}
759
760 
#[derive(Deserialize)]
761 
pub struct AttachForm {
762 
    _csrf: String,
763 
    release: i64,
764 
}
765
766 
/// `POST /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/attach`: add
767 
/// an artifact to a release as an asset, named after its path; the release
768 
/// page then shows the run and receipt that built it.
769 
pub async fn attach(
770 
    State(app): State<Shared>,
771 
    viewer: Viewer,
772 
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
773 
    Form(f): Form<AttachForm>,
774 
) -> Result<Response, Error> {
775 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
776 
    require(&viewer, &shell)?;
777 
    csrf(&viewer, &f._csrf, &shell)?;
778 
    let (repo, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
779 
    let run = fdb::run(&app.db, row.id, id)
780 
        .await
781 
        .map_err(|e| Error::from_db(shell.clone(), e))?
782 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
783 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
784 
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
785 
    // A failing job's `when "always"` artifacts are not in its receipt, so
786 
    // they cannot stand for a release (oxygenflow#37).
787 
    if j.status != "passed" {
788 
        return Err(Error::bad(shell, "only a passing job's artifacts can be attached to a release"));
789 
    }
790 
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
791 
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
792 
    let releases = db::releases::list(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
793 
    let Some(rel) = releases.into_iter().find(|r| r.id == f.release) else { return Err(Error::not_found(shell)) };
794 
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
795 
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
796 
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
797 
    super::releases::store_asset(&app, rel.id, &file, "application/octet-stream", &bytes)
798 
        .await
799 
        .map_err(|m| Error::bad(shell.clone(), m))?;
800 
    Ok(Redirect::to(&format!("/{owner}/{name}/releases/tag/{}", rel.tag)).into_response())
801 
}
802
803 
/// `/{owner}/{name}/flow/badge.svg`: the latest run on the default
804 
/// bookmark as a small SVG for a README, for whoever may read the
805 
/// repository. `passed` green, `failed` or `error` red, a run still going
806 
/// blue, none grey.
807 
pub async fn badge(
808 
    State(app): State<Shared>,
809 
    viewer: Viewer,
810 
    Path((owner, name)): Path<(String, String)>,
811 
) -> Result<axum::response::Response, Error> {
812 
    use axum::response::IntoResponse;
813 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
814 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
815 
    let latest = fdb::latest_on(&app.db, row.id, &row.default_bookmark).await.map_err(|e| Error::from_db(shell.clone(), e))?;
816 
    let (text, colour) = match latest.as_ref().map(|r| r.status.as_str()) {
817 
        Some("passed") => ("passing", "#2da44e"),
818 
        Some("failed") | Some("error") => ("failing", "#cf222e"),
819 
        Some("cancelled") => ("cancelled", "#6e7781"),
820 
        Some(_) => ("running", "#0969da"),
821 
        None => ("no runs", "#6e7781"),
822 
    };
823 
    let label = "OxygenFlow";
824 
    let (lw, tw) = (label.len() as u32 * 7 + 10, text.len() as u32 * 7 + 10);
825 
    let svg = format!(
826 
        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>"##,
827 
        w = lw + tw,
828 
        lx = lw / 2,
829 
        tx = lw + tw / 2,
830 
    );
831 
    Ok((
832 
        [(axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"), (axum::http::header::CACHE_CONTROL, "no-cache, max-age=60")],
833 
        svg,
834 
    )
835 
        .into_response())
836 
}
837
838 
#[cfg(test)]
839 
mod section_tests {
840 
    use super::*;
841
842 
    fn line(seq: i64, t: i64, text: &str) -> fdb::LogLine {
843 
        fdb::LogLine { seq, t, stream: "out".into(), text: text.into() }
844 
    }
845
846 
    #[test]
847 
    fn marked_lines_fold_into_sections() {
848 
        let lines = vec![
849 
            line(1, 0, "booting"),
850 
            line(2, 100, "::group::cargo build"),
851 
            line(3, 200, "Compiling a"),
852 
            line(4, 900, "Compiling b"),
853 
            line(5, 4_300, "::endgroup::"),
854 
            line(6, 4_400, "::group::cargo test"),
855 
            line(7, 4_500, "test x ... ok"),
856 
            line(8, 5_000, "::group::lint"),
857 
            line(9, 5_100, "clean"),
858 
        ];
859 
        let segs = sections(&lines);
860 
        assert_eq!(segs.len(), 4);
861 
        assert!(matches!(&segs[0], Segment::Line(l) if l.text == "booting"));
862 
        let Segment::Group(build) = &segs[1] else { panic!("a section") };
863 
        assert_eq!((build.title.as_str(), build.from_t, build.to_t, build.lines.len()), ("cargo build", 100, Some(4_300), 2));
864 
        let Segment::Group(test) = &segs[2] else { panic!("a section") };
865 
        assert_eq!(test.to_t, Some(5_000), "a new section ends the one before");
866 
        let Segment::Group(lint) = &segs[3] else { panic!("a section") };
867 
        assert_eq!((lint.to_t, lint.lines.len()), (None, 1), "still going");
868 
        assert_eq!(secs(4_200), "4.2s");
869 
    }
870 
}