//! The Flow tab (hub spec §13.2): runs of a repository, and a run's job //! graph with live status and logs. use super::settings::{csrf, require}; use super::{filters, lookup, lookup_role, Error, Page, RepoShell, Shell}; use crate::auth::{Access, Role, Viewer}; use crate::db::{self, flow as fdb}; use crate::Shared; use askama::Template; use axum::extract::{Form, Path, Query, State}; use axum::response::{IntoResponse, Redirect, Response}; use axum::Json; use serde::Deserialize; pub struct RunRow { pub r: fdb::Run, pub duration: String, pub jobs: Vec<(String, String)>, /// "3rd in line" for a run still waiting for a runner. pub place: Option, } #[derive(Template)] #[template(path = "flow_runs.html")] pub struct RunsPage { shell: Shell, repo: RepoShell, runs: Vec, enabled: bool, older: Option, /// Writers may start runs: the bookmarks to offer, default first. can_run: bool, bookmarks: Vec, error: Option, /// Why this viewer is not offered "Run pipeline", when they are not. why_not: Option, } #[derive(Deserialize)] pub struct RunsQuery { before: Option, /// Why a "Run pipeline" or "Re-run" did not start. error: Option, } /// Why `viewer` may not start runs here, or `None` when they may. pub fn why_not_run(enabled: bool, signed_in: bool, role: Role) -> Option { if !enabled { Some("Flow is off for this repository; a site administrator can turn it on in Settings → Flow.".into()) } else if !signed_in { Some("Sign in with write access to run the pipeline.".into()) } else if role < Role::Write { Some("Running the pipeline needs write access to this repository.".into()) } else { None } } const PER_PAGE: i64 = 30; pub async fn runs( State(app): State, viewer: Viewer, Path((owner, name)): Path<(String, String)>, Query(q): Query, ) -> Result, Error> { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer); let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?; let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id); let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] }; let mut runs = Vec::new(); for r in list.into_iter().take(PER_PAGE as usize) { let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None }; let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default(); let duration = match r.finished_ms { Some(f) => crate::flow::human_ms(f - r.created_ms), None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms), None => "–".into(), }; runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place }); } let enabled = crate::flow::enabled(&app, &row); let why_not = why_not_run(enabled, viewer.account().is_some(), role); let can_run = why_not.is_none(); let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] }; Ok(Page(RunsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), runs, enabled, older, can_run, bookmarks, error: q.error, why_not, })) } /// The repository's bookmarks, its default first. async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec { let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] }; let mut names: Vec = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default(); names.sort_by_key(|n| (n != &row.default_bookmark, n.clone())); names } #[derive(Deserialize)] pub struct StartForm { #[serde(rename = "_csrf")] csrf_token: String, #[serde(default)] bookmark: String, /// A change's full hash: run the state whose only head it is. #[serde(default)] change: String, } /// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark's /// current state, or on one committed change. pub async fn start( State(app): State, viewer: Viewer, Path((owner, name)): Path<(String, String)>, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer); let me = require(&viewer, &shell)?; let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?; csrf(&viewer, &f.csrf_token, &shell)?; let (started, what) = if f.change.trim().is_empty() { (crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username, &[]).await, f.bookmark.trim().to_string()) } else { (crate::flow::run_change(&app, &row, &f.change, &me.account.username, &[]).await, format!("change {}", f.change.trim())) }; Ok(match started { Ok(n) => { db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {what}")).await.ok(); Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response() } Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(), }) } #[derive(Deserialize)] pub struct RerunForm { #[serde(rename = "_csrf")] csrf_token: String, } /// `POST /{owner}/{name}/flow/{number}/rerun`. pub async fn rerun( State(app): State, viewer: Viewer, Path((owner, name, number)): Path<(String, String, i64)>, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer); let me = require(&viewer, &shell)?; let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?; csrf(&viewer, &f.csrf_token, &shell)?; let run = fdb::run(&app.db, row.id, number) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await { Ok(n) => { db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}")) .await .ok(); Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response() } Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(), }) } /// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31). pub async fn rerun_job( State(app): State, viewer: Viewer, Path((owner, name, number, job)): Path<(String, String, i64, String)>, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer); let me = require(&viewer, &shell)?; let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?; csrf(&viewer, &f.csrf_token, &shell)?; let run = fdb::run(&app.db, row.id, number) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; Ok(match crate::flow::rerun_job(&app, &row, &run, &job, &me.account.username).await { Ok(_) => { db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("{job} of #{number}")) .await .ok(); Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response() } Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(), }) } /// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32). pub async fn cancel( State(app): State, viewer: Viewer, Path((owner, name, number)): Path<(String, String, i64)>, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer); let me = require(&viewer, &shell)?; let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?; csrf(&viewer, &f.csrf_token, &shell)?; let run = fdb::run(&app.db, row.id, number) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; Ok(match crate::flow::cancel(&app, &row, &run, &me.account.username).await { Ok(()) => { db::accounts::audit(&app.db, Some(&me.account), "flow.cancel", &format!("{owner}/{name}"), &format!("#{number}")).await.ok(); Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response() } Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(), }) } /// Percent-encode a message for a query string. fn urlencoding(s: &str) -> String { s.bytes() .map(|b| match b { b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(), _ => format!("%{b:02X}"), }) .collect() } /// A job placed in the graph. pub struct Node { pub j: fdb::Job, pub x: i64, pub y: i64, pub duration: String, } pub struct Edge { pub d: String, pub status: String, } pub struct Graph { pub nodes: Vec, pub edges: Vec, pub width: i64, pub height: i64, } pub const NODE_W: i64 = 200; pub const NODE_H: i64 = 48; const COL_GAP: i64 = 72; const ROW_GAP: i64 = 18; const PAD: i64 = 12; /// Columns by dependency depth, jobs in each column in name order. pub fn layout(jobs: Vec) -> Graph { let mut by_depth: std::collections::BTreeMap> = Default::default(); for j in jobs { by_depth.entry(j.depth).or_default().push(j); } let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64; let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP; let mut nodes = Vec::new(); for (col, (_, mut js)) in by_depth.into_iter().enumerate() { js.sort_by(|a, b| a.name.cmp(&b.name)); let n = js.len() as i64; let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP; let top = (height - column_h) / 2; for (i, j) in js.into_iter().enumerate() { let duration = crate::flow::duration(&j); nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j }); } } let mut edges = Vec::new(); for n in &nodes { for d in n.j.depends_on() { if let Some(from) = nodes.iter().find(|m| m.j.name == d) { let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2); let (x2, y2) = (n.x, n.y + NODE_H / 2); let mid = (x1 + x2) / 2; edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() }); } } } let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2); Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) } } pub fn glyph(status: &str) -> &'static str { match status { "passed" => "✓", "failed" => "✕", "running" => "●", "skipped" => "⤼", "cancelled" => "⊘", "error" => "!", _ => "○", } } /// Failed tests a run page lists per job; the rest are counted. const SHOWN_FAILURES: i64 = 50; /// A stretch of a job's log as the run page shows it: a line, or a /// section the job marked with `::group::TITLE` … `::endgroup::` /// (oxygenflow#31), shown folded with how long it took. pub enum Segment { Line(fdb::LogLine), Group(Section), } pub struct Section { pub title: String, /// When it began, and ended (`None` while it is still going). pub from_t: i64, pub to_t: Option, pub lines: Vec, } /// A log's lines in sections. A section left open is still going; a new /// `::group::` ends the one before (sections do not nest). pub fn sections(lines: &[fdb::LogLine]) -> Vec { let mut out: Vec = Vec::new(); let mut open: Option
= None; for l in lines { if let Some(title) = l.text.strip_prefix("::group::") { if let Some(g) = open.take() { out.push(Segment::Group(Section { to_t: Some(l.t), ..g })); } open = Some(Section { title: title.trim().to_string(), from_t: l.t, to_t: None, lines: Vec::new() }); } else if l.text.trim() == "::endgroup::" { if let Some(g) = open.take() { out.push(Segment::Group(Section { to_t: Some(l.t), ..g })); } } else { match open.as_mut() { Some(g) => g.lines.push(l.clone()), None => out.push(Segment::Line(l.clone())), } } } if let Some(g) = open { out.push(Segment::Group(g)); } out } /// Milliseconds as the log shows them: `12.3s`. pub fn secs(ms: impl std::borrow::Borrow) -> String { let ms = *ms.borrow(); format!("{}.{}s", ms / 1000, ms % 1000 / 100) } pub struct JobLog { pub j: fdb::Job, pub artifacts: Vec, pub lines: Vec, /// The lines in sections (oxygenflow#31). pub segments: Vec, pub duration: String, /// The last line shown; the live view asks for what follows. pub last_seq: i64, /// The job's coverage report, when it uploaded one (#113): its id and percentage. pub coverage: Option<(i64, String)>, /// The job's test report, when it uploaded one (#114), and its first /// failures. pub tests: Option, pub failures: Vec, } #[derive(Template)] #[template(path = "flow_run.html")] pub struct RunPage { shell: Shell, repo: RepoShell, run: fdb::Run, graph: Graph, logs: Vec, duration: String, /// Whether the viewer may re-run it. can_rerun: bool, /// Whether the viewer may cancel it. can_cancel: bool, /// Where a live run stands among the runners' work. queue: Option, /// Releases a writer may attach this run's artifacts to. releases: Vec, } /// Lines of a job's log a page shows before "the rest is loaded as it comes". const PAGE_LINES: i64 = 5000; pub async fn run( State(app): State, viewer: Viewer, Path((owner, name, id)): Path<(String, String, i64)>, ) -> Result, Error> { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer); let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let run = fdb::run(&app.db, row.id, id) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let reports = crate::coverage::for_run(&app.db, run.id).await.unwrap_or_default(); let test_reports = crate::junit::for_run(&app.db, run.id).await.unwrap_or_default(); let mut logs = Vec::new(); for j in &jobs { let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default(); let last_seq = lines.last().map(|l| l.seq).unwrap_or(0); let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default(); let coverage = reports .iter() .find(|r| r.job_id == Some(j.id)) .map(|r| (r.id, r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into()))); let tests = test_reports.iter().find(|r| r.job_id == j.id).cloned(); let failures = match &tests { Some(r) if r.broken() > 0 => crate::junit::failures(&app.db, r.id, SHOWN_FAILURES).await.unwrap_or_default(), _ => vec![], }; let segments = sections(&lines); logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), artifacts, lines, segments, last_seq, coverage, tests, failures, }); } let duration = match run.finished_ms { Some(f) => crate::flow::human_ms(f - run.created_ms), None => crate::flow::human_ms(db::now_ms() - run.created_ms), }; let graph = layout(jobs); let can_write = viewer.account().is_some() && role >= Role::Write; let can_rerun = !run.live() && can_write && crate::flow::enabled(&app, &row); let can_cancel = run.live() && can_write; let queue = if run.live() { let q = fdb::queue(&app.db).await.unwrap_or_default(); Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some()) } else { None }; let releases = if viewer.account().is_some() && role >= Role::Write && logs.iter().any(|l| !l.artifacts.is_empty()) { db::releases::list(&app.db, row.id).await.unwrap_or_default() } else { vec![] }; Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun, can_cancel, queue, releases, })) } #[derive(Deserialize)] pub struct LiveQuery { /// `job:seq,job:seq`: the last log line each open job already shows. after: Option, } /// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the /// run page while it is live. pub async fn live( State(app): State, viewer: Viewer, Path((owner, name, id)): Path<(String, String, i64)>, Query(q): Query, ) -> Result { let shell = Shell::new(&app, "", &viewer); let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let run = fdb::run(&app.db, row.id, id) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let after: std::collections::HashMap = q .after .unwrap_or_default() .split(',') .filter_map(|p| p.split_once(':')) .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?))) .collect(); let mut out = Vec::new(); for j in &jobs { let lines = match after.get(&j.id) { Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(), None => vec![], }; out.push(serde_json::json!({ "id": j.id, "name": j.name, "status": j.status, "glyph": glyph(&j.status), "duration": crate::flow::duration(j), "exit_code": j.exit_code, "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::>(), })); } Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response()) } // ------------------------------------------------------------ state stamps /// In rendered Markdown, the image path that is the viewed state's stamp. pub const STAMP_IMAGE: &str = ".flow/stamp"; /// In rendered Markdown, the link path to the viewed state's run. pub const RUN_LINK: &str = ".flow/run"; /// Resolve a ref to its state address, and whether the ref can move. async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> { let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?; let spec = r.to_string(); app.repos .read(&repo, move |repo, cache| { let t = crate::repos::view::resolve(repo, cache, &spec)?; let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag); Ok((t.permalink().unwrap_or_default(), moves)) }) .await } /// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2). pub async fn stamp( State(app): State, viewer: Viewer, Path((owner, name, r)): Path<(String, String, String)>, ) -> Result { let shell = Shell::new(&app, "", &viewer); let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?; if moves && !heads.is_empty() { let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response(); res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache")); return Ok(res); } let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?; let jobs = match &run { Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?, None => vec![], }; let svg = stamp_svg(&heads, run.as_ref(), &jobs); let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" }; Ok(( [ (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"), (axum::http::header::CACHE_CONTROL, cache), (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"), ], svg, ) .into_response()) } /// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab. pub async fn at( State(app): State, viewer: Viewer, Path((owner, name, r)): Path<(String, String, String)>, ) -> Result { let shell = Shell::new(&app, "", &viewer); let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?; let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?; let to = match run { Some(r) => format!("/{owner}/{name}/flow/{}", r.number), None => format!("/{owner}/{name}/flow"), }; Ok(axum::response::Redirect::to(&to).into_response()) } fn xml(s: &str) -> String { s.replace('&', "&").replace('<', "<").replace('>', ">").replace('"', """) } /// The stamp: the mark, one glyph per job, the status, the state and the /// date; the same facts machine-readable in ``. Fixed colours, /// because an image does not follow the page's theme. pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String { const CH: f64 = 6.6; // IBM Plex Mono at 11px let colour = |s: &str| match s { "passed" => "#5fe3a1", "failed" | "error" => "#ff5e7a", "running" => "#00e0f0", _ => "#8592ad", }; let short: String = heads.chars().take(12).collect(); let (status, date) = match run { Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::()), None => ("no run", String::new()), }; let mut ordered: Vec<&fdb::Job> = jobs.iter().collect(); ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name))); let label = "OXYGENFLOW"; let label_w = 18.0 + label.len() as f64 * 7.4 + 8.0; // Barlow bold 10.5px, spaced let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 }; let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") }; let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0; let w = (label_w + right_w).ceil(); let mut glyphs = String::new(); for (i, j) in ordered.iter().enumerate() { let x = label_w + 8.0 + i as f64 * 10.0 + 4.0; glyphs.push_str(&format!( r#"{}"#, colour(&j.status), glyph(&j.status) )); } let meta = serde_json::json!({ "heads": heads.split('+').filter(|h| !h.is_empty()).collect::>(), "run": run.map(|r| r.number), "status": status, "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::>(), }); let title = match run { Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.number), None => format!("OxygenFlow: no run for state {short}"), }; format!( r##"{t}{m}{label}{glyphs}{tail}"##, t = xml(&title), m = xml(&meta.to_string()), lw = label_w, lx = label_w - 4.0, tx = label_w + 8.0 + glyphs_w, sc = colour(status), tail = xml(&tail), wi = w - 1.0, ) } /// A receipt kept in a repository (Oxygen spec §14.6), as pages show it. #[derive(Clone, Debug)] pub struct ReceiptRow { pub attest: String, pub job: String, pub exit_code: u32, pub duration: String, pub runner: String, pub run: String, /// Why it does not verify, or that its runner's key was revoked. pub problem: String, pub made_ms: i64, } impl ReceiptRow { pub fn passed(&self) -> bool { self.exit_code == 0 } } /// The receipts kept for the state of `heads`. pub async fn receipts_of(app: &Shared, name: &crate::repos::RepoName, heads: Vec) -> Vec { let base = app.config.base_url(); let found = app.repos.read(name, move |repo, _| Ok(repo.receipts(&heads)?)).await.unwrap_or_default(); found .into_iter() .map(|i| { let r = &i.signed.receipt; let run = i.signed.run().to_string(); ReceiptRow { attest: i.attest.to_hex(), job: r.job_name.clone(), exit_code: r.exit_code, duration: crate::flow::human_ms(r.duration_ms as i64), runner: i.signed.runner.name.clone(), // Only links to this hub are followed from the page. run: if run.starts_with(base) { run } else { String::new() }, problem: match (i.problem, i.revoked) { (Some(p), _) => p, (None, true) => "the runner's key was revoked".into(), (None, false) => String::new(), }, made_ms: i.signed.attest.hlc.wall_ms as i64, } }) .collect() } /// `GET /{owner}/{name}/receipts/{attest}`: a signed receipt, for /// `oxy receipt verify` or `oxy receipt import`. pub async fn receipt_file( State(app): State, viewer: Viewer, Path((owner, name, attest)): Path<(String, String, String)>, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/receipts/{attest}"), &viewer); let (repo, _row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let Some(r) = oxy_core::hash::ObjectRef::from_hex(attest.trim_end_matches(".receipt")) else { return Err(Error::not_found(shell)); }; let (job, bytes) = app .repos .read(&repo, move |repo, _| { let i = repo.receipt(&r)?; Ok((i.signed.receipt.job_name.clone(), i.signed.encode())) }) .await .map_err(|_| Error::not_found(shell.clone()))?; let file: String = job.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).collect(); Ok(( [ (axum::http::header::CONTENT_TYPE, "application/oxy-receipt".to_string()), (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{file}-{}.receipt\"", r.short(12))), ], bytes, ) .into_response()) } /// `GET /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}`: an artifact /// of a run's job, read from the repository. pub async fn artifact_file( State(app): State, viewer: Viewer, Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}"), &viewer); let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let run = fdb::run(&app.db, row.id, id) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) }; let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) }; let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) }; let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?; let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string(); Ok(( [ (axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()), (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}\"", file.replace('"', ""))), ], bytes, ) .into_response()) } #[derive(Deserialize)] pub struct AttachForm { _csrf: String, release: i64, } /// `POST /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/attach`: add /// an artifact to a release as an asset, named after its path; the release /// page then shows the run and receipt that built it. pub async fn attach( State(app): State, viewer: Viewer, Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>, Form(f): Form, ) -> Result { let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer); require(&viewer, &shell)?; csrf(&viewer, &f._csrf, &shell)?; let (repo, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?; let run = fdb::run(&app.db, row.id, id) .await .map_err(|e| Error::from_db(shell.clone(), e))? .ok_or_else(|| Error::not_found(shell.clone()))?; let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) }; // A failing job's `when "always"` artifacts are not in its receipt, so // they cannot stand for a release (oxygenflow#37). if j.status != "passed" { return Err(Error::bad(shell, "only a passing job's artifacts can be attached to a release")); } let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) }; let releases = db::releases::list(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?; let Some(rel) = releases.into_iter().find(|r| r.id == f.release) else { return Err(Error::not_found(shell)) }; let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) }; let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|e| Error::from_repo(shell.clone(), e))?; let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string(); super::releases::store_asset(&app, rel.id, &file, "application/octet-stream", &bytes) .await .map_err(|m| Error::bad(shell.clone(), m))?; Ok(Redirect::to(&format!("/{owner}/{name}/releases/tag/{}", rel.tag)).into_response()) } /// `/{owner}/{name}/flow/badge.svg`: the latest run on the default /// bookmark as a small SVG for a README, for whoever may read the /// repository. `passed` green, `failed` or `error` red, a run still going /// blue, none grey. pub async fn badge( State(app): State, viewer: Viewer, Path((owner, name)): Path<(String, String)>, ) -> Result { use axum::response::IntoResponse; let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer); let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?; let latest = fdb::latest_on(&app.db, row.id, &row.default_bookmark).await.map_err(|e| Error::from_db(shell.clone(), e))?; let (text, colour) = match latest.as_ref().map(|r| r.status.as_str()) { Some("passed") => ("passing", "#2da44e"), Some("failed") | Some("error") => ("failing", "#cf222e"), Some("cancelled") => ("cancelled", "#6e7781"), Some(_) => ("running", "#0969da"), None => ("no runs", "#6e7781"), }; let label = "OxygenFlow"; let (lw, tw) = (label.len() as u32 * 7 + 10, text.len() as u32 * 7 + 10); let svg = format!( r##"{label}: {text}{label}{text}"##, w = lw + tw, lx = lw / 2, tx = lw + tw / 2, ); Ok(( [(axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"), (axum::http::header::CACHE_CONTROL, "no-cache, max-age=60")], svg, ) .into_response()) } #[cfg(test)] mod section_tests { use super::*; fn line(seq: i64, t: i64, text: &str) -> fdb::LogLine { fdb::LogLine { seq, t, stream: "out".into(), text: text.into() } } #[test] fn marked_lines_fold_into_sections() { let lines = vec![ line(1, 0, "booting"), line(2, 100, "::group::cargo build"), line(3, 200, "Compiling a"), line(4, 900, "Compiling b"), line(5, 4_300, "::endgroup::"), line(6, 4_400, "::group::cargo test"), line(7, 4_500, "test x ... ok"), line(8, 5_000, "::group::lint"), line(9, 5_100, "clean"), ]; let segs = sections(&lines); assert_eq!(segs.len(), 4); assert!(matches!(&segs[0], Segment::Line(l) if l.text == "booting")); let Segment::Group(build) = &segs[1] else { panic!("a section") }; assert_eq!((build.title.as_str(), build.from_t, build.to_t, build.lines.len()), ("cargo build", 100, Some(4_300), 2)); let Segment::Group(test) = &segs[2] else { panic!("a section") }; assert_eq!(test.to_t, Some(5_000), "a new section ends the one before"); let Segment::Group(lint) = &segs[3] else { panic!("a section") }; assert_eq!((lint.to_t, lint.lines.len()), (None, 1), "still going"); assert_eq!(secs(4_200), "4.2s"); } }