//! 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)>, } #[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, } #[derive(Deserialize)] pub struct RunsQuery { before: Option, /// Why a "Run pipeline" or "Re-run" did not start. error: Option, } 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 mut runs = Vec::new(); for r in list.into_iter().take(PER_PAGE as usize) { 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 }); } let enabled = crate::flow::enabled(&app, &row); let can_run = enabled && viewer.account().is_some() && role >= Role::Write; 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, })) } /// 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, bookmark: String, } /// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark. 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)?; Ok(match crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username).await { Ok(n) => { db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {}", f.bookmark.trim())) .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(), }) } /// 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" => "⤼", "error" => "!", _ => "○", } } pub struct JobLog { pub j: fdb::Job, pub lines: Vec, pub duration: String, /// The last line shown; the live view asks for what follows. pub last_seq: i64, } #[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, } /// 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 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); logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), lines, last_seq }); } 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_rerun = !run.live() && viewer.account().is_some() && role >= Role::Write && crate::flow::enabled(&app, &row); Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun })) } #[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 = 14.0 + label.len() as f64 * 6.2 + 8.0; 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, ) }