OXYGENHUB
◆ v0.9.23 oxygenhub / crates / hub / src / web / flow.rs
49.0 KiB · Rust · last changed Run a pipeline's WASM jobs in the browser tab (refs andymccall/oxygenflow#80); runner builds shared with the swarm (refs andymccall/oxygenflow#84); 0.9.15 7 days ago source blame history raw
42fa9554 Andy McCall ↶ 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
1edac4b6 Andy McCall ↶ 4 
use super::settings::{csrf, require};
5 
use super::{filters, lookup, lookup_role, Error, Page, RepoShell, Shell};
6 
use crate::auth::{Access, Role, Viewer};
42fa9554 Andy McCall ↶ 7 
use crate::db::{self, flow as fdb};
8 
use crate::Shared;
9 
use askama::Template;
1edac4b6 Andy McCall ↶ 10 
use axum::extract::{Form, Path, Query, State};
11 
use axum::response::{IntoResponse, Redirect, Response};
42fa9554 Andy McCall ↶ 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)>,
bf3d0754 Andy McCall ↶ 19 
    /// "3rd in line" for a run still waiting for a runner.
20 
    pub place: Option<String>,
42fa9554 Andy McCall ↶ 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>,
1edac4b6 Andy McCall ↶ 31 
    /// Writers may start runs: the bookmarks to offer, default first.
32 
    can_run: bool,
33 
    bookmarks: Vec<String>,
34 
    error: Option<String>,
4f692533 Andy McCall ↶ 35 
    /// Why this viewer is not offered "Run pipeline", when they are not.
36 
    why_not: Option<String>,
42fa9554 Andy McCall ↶ 37 
}
38
39 
#[derive(Deserialize)]
40 
pub struct RunsQuery {
41 
    before: Option<i64>,
1edac4b6 Andy McCall ↶ 42 
    /// Why a "Run pipeline" or "Re-run" did not start.
43 
    error: Option<String>,
42fa9554 Andy McCall ↶ 44 
}
45
4f692533 Andy McCall ↶ 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 {
c7067c7b Andy McCall 49 
        Some("Flow is off for this repository; a site administrator can turn it on in Settings › Pipelines.".into())
4f692533 Andy McCall ↶ 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
42fa9554 Andy McCall ↶ 59 
const PER_PAGE: i64 = 30;
60
b644e175 Andy McCall ↶ 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).
74 
pub async fn deployments(
75 
    State(app): State<Shared>,
76 
    viewer: Viewer,
77 
    Path((owner, name)): Path<(String, String)>,
78 
) -> Result<Page<DeploymentsPage>, Error> {
79 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/deployments"), &viewer);
80 
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
81 
    let history = fdb::deployments(&app.db, row.id, 50).await.map_err(|e| Error::from_db(shell.clone(), e))?;
82 
    let mut current: Vec<fdb::Deployment> = Vec::new();
83 
    for d in &history {
84 
        if !current.iter().any(|c| c.environment == d.environment) {
85 
            current.push(d.clone());
86 
        }
87 
    }
88 
    current.sort_by(|a, b| a.environment.cmp(&b.environment));
89 
    Ok(Page(DeploymentsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), current, history }))
90 
}
91
42fa9554 Andy McCall ↶ 92 
pub async fn runs(
93 
    State(app): State<Shared>,
94 
    viewer: Viewer,
95 
    Path((owner, name)): Path<(String, String)>,
96 
    Query(q): Query<RunsQuery>,
97 
) -> Result<Page<RunsPage>, Error> {
98 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
99 
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
100 
    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;
101 
    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);
bf3d0754 Andy McCall ↶ 102 
    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };
42fa9554 Andy McCall ↶ 103 
    let mut runs = Vec::new();
104 
    for r in list.into_iter().take(PER_PAGE as usize) {
bf3d0754 Andy McCall ↶ 105 
        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };
42fa9554 Andy McCall ↶ 106 
        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();
107 
        let duration = match r.finished_ms {
108 
            Some(f) => crate::flow::human_ms(f - r.created_ms),
109 
            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),
110 
            None => "–".into(),
111 
        };
bf3d0754 Andy McCall ↶ 112 
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });
42fa9554 Andy McCall ↶ 113 
    }
9008bf6c Andy McCall ↶ 114 
    let enabled = crate::flow::enabled(&app, &row);
4f692533 Andy McCall ↶ 115 
    let why_not = why_not_run(enabled, viewer.account().is_some(), role);
116 
    let can_run = why_not.is_none();
1edac4b6 Andy McCall ↶ 117 
    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };
118 
    Ok(Page(RunsPage {
119 
        shell,
120 
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
121 
        runs,
122 
        enabled,
123 
        older,
124 
        can_run,
125 
        bookmarks,
126 
        error: q.error,
4f692533 Andy McCall ↶ 127 
        why_not,
1edac4b6 Andy McCall ↶ 128 
    }))
129 
}
130
c26267b3 Andy McCall ↶ 131 
/// A WASM job as the tab runner runs it (oxygenflow#80).
132 
#[derive(serde::Serialize)]
133 
pub struct TabJob {
134 
    name: String,
135 
    exec: Vec<String>,
136 
    env: std::collections::BTreeMap<String, String>,
137 
    memory_mib: u32,
138 
}
139
140 
#[derive(Template)]
141 
#[template(path = "flow_tab.html")]
142 
pub struct TabPage {
143 
    shell: Shell,
144 
    repo: RepoShell,
145 
    bookmark: String,
146 
    bookmarks: Vec<String>,
147 
    /// The pipeline's WASM jobs, as JSON for the page's script.
148 
    jobs_json: String,
149 
    jobs: usize,
150 
    /// Why there is nothing to run: no pipeline, one that does not read,
151 
    /// or no WASM job in it.
152 
    problem: Option<String>,
153 
    archive: String,
154 
}
155
156 
#[derive(Deserialize)]
157 
pub struct TabQuery {
158 
    bookmark: Option<String>,
159 
}
160
161 
/// The WASM jobs of a pipeline (`runtime "wasm"`), each with the
162 
/// environment its sandbox and itself give it.
163 
pub fn wasm_jobs(p: &flow_core::Pipeline) -> Vec<TabJob> {
164 
    p.jobs
165 
        .values()
166 
        .filter_map(|j| {
167 
            let sb = p.sandboxes.get(&j.sandbox)?;
168 
            (sb.runtime == flow_core::Runtime::Wasm && !j.exec.is_empty()).then(|| {
169 
                let mut env = sb.env.clone();
170 
                env.extend(j.env.clone());
171 
                TabJob { name: j.name.clone(), exec: j.exec.clone(), env, memory_mib: sb.memory_mib }
172 
            })
173 
        })
174 
        .collect()
175 
}
176
177 
/// `GET /{owner}/{name}/flow/tab`: run the pipeline's WASM jobs in this
178 
/// browser tab (oxygenflow#80, first part): a worker unpacks the
179 
/// bookmark's files and runs each job's modules with a WASI shim, as a
180 
/// runner's WASM tier does. Nothing is claimed or reported: it shows the
181 
/// tier works in a tab, before the tab becomes a runner.
182 
pub async fn tab(
183 
    State(app): State<Shared>,
184 
    viewer: Viewer,
185 
    Path((owner, name)): Path<(String, String)>,
186 
    Query(q): Query<TabQuery>,
187 
) -> Result<Page<TabPage>, Error> {
188 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/tab"), &viewer);
189 
    let (repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
190 
    let bookmarks = bookmark_names(&app, &row).await;
191 
    let bookmark = q.bookmark.filter(|b| bookmarks.contains(b)).unwrap_or_else(|| row.default_bookmark.clone());
192 
    let b = bookmark.clone();
193 
    let text = app
194 
        .repos
195 
        .read(&repo, move |r, cache| {
196 
            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
197 
            let heads = r.store().get_state(&st)?.heads;
198 
            let loaded = cache.load(r, &heads)?;
199 
            Ok(match loaded.snapshot.files.get(flow_core::PIPELINE_PATH).map(|f| &f.content) {
200 
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
201 
                _ => None,
202 
            })
203 
        })
204 
        .await
205 
        .unwrap_or(None);
206 
    let (jobs, problem) = match text.map(|t| flow_core::Pipeline::parse(&t)) {
207 
        None => (vec![], Some(format!("{bookmark} has no {}.", flow_core::PIPELINE_PATH))),
208 
        Some(Err(e)) => (vec![], Some(format!("The pipeline on {bookmark} does not read: {e}"))),
209 
        Some(Ok(p)) => {
210 
            let jobs = wasm_jobs(&p);
211 
            let problem =
212 
                jobs.is_empty().then(|| format!("The pipeline on {bookmark} has no WASM jobs (a sandbox with runtime \"wasm\")."));
213 
            (jobs, problem)
214 
        }
215 
    };
216 
    Ok(Page(TabPage {
217 
        shell,
218 
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
219 
        archive: format!("/{owner}/{name}/archive?ref={}&format=zip", urlencoding(&bookmark)),
220 
        bookmark,
221 
        bookmarks,
222 
        jobs: jobs.len(),
223 
        jobs_json: serde_json::to_string(&jobs).unwrap_or_else(|_| "[]".into()),
224 
        problem,
225 
    }))
226 
}
227
1edac4b6 Andy McCall ↶ 228 
/// The repository's bookmarks, its default first.
229 
async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {
230 
    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };
231 
    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();
232 
    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));
233 
    names
234 
}
235
236 
#[derive(Deserialize)]
237 
pub struct StartForm {
238 
    #[serde(rename = "_csrf")]
239 
    csrf_token: String,
4f692533 Andy McCall ↶ 240 
    #[serde(default)]
1edac4b6 Andy McCall ↶ 241 
    bookmark: String,
4f692533 Andy McCall ↶ 242 
    /// A change's full hash: run the state whose only head it is.
243 
    #[serde(default)]
244 
    change: String,
1edac4b6 Andy McCall ↶ 245 
}
246
4f692533 Andy McCall ↶ 247 
/// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark's
248 
/// current state, or on one committed change.
1edac4b6 Andy McCall ↶ 249 
pub async fn start(
250 
    State(app): State<Shared>,
251 
    viewer: Viewer,
252 
    Path((owner, name)): Path<(String, String)>,
253 
    Form(f): Form<StartForm>,
254 
) -> Result<Response, Error> {
255 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
256 
    let me = require(&viewer, &shell)?;
257 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
258 
    csrf(&viewer, &f.csrf_token, &shell)?;
4f692533 Andy McCall ↶ 259 
    let (started, what) = if f.change.trim().is_empty() {
0650d94e Andy McCall ↶ 260 
        (crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username, &[]).await, f.bookmark.trim().to_string())
4f692533 Andy McCall ↶ 261 
    } else {
0650d94e Andy McCall ↶ 262 
        (crate::flow::run_change(&app, &row, &f.change, &me.account.username, &[]).await, format!("change {}", f.change.trim()))
4f692533 Andy McCall ↶ 263 
    };
264 
    Ok(match started {
1edac4b6 Andy McCall ↶ 265 
        Ok(n) => {
4f692533 Andy McCall ↶ 266 
            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {what}")).await.ok();
1edac4b6 Andy McCall ↶ 267 
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
268 
        }
269 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
270 
    })
271 
}
272
273 
#[derive(Deserialize)]
274 
pub struct RerunForm {
275 
    #[serde(rename = "_csrf")]
276 
    csrf_token: String,
277 
}
278
279 
/// `POST /{owner}/{name}/flow/{number}/rerun`.
280 
pub async fn rerun(
281 
    State(app): State<Shared>,
282 
    viewer: Viewer,
283 
    Path((owner, name, number)): Path<(String, String, i64)>,
284 
    Form(f): Form<RerunForm>,
285 
) -> Result<Response, Error> {
286 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
287 
    let me = require(&viewer, &shell)?;
288 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
289 
    csrf(&viewer, &f.csrf_token, &shell)?;
290 
    let run = fdb::run(&app.db, row.id, number)
291 
        .await
292 
        .map_err(|e| Error::from_db(shell.clone(), e))?
293 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
294 
    Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await {
295 
        Ok(n) => {
296 
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}"))
297 
                .await
298 
                .ok();
299 
            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()
300 
        }
301 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
302 
    })
303 
}
304
e8fa5369 Andy McCall ↶ 305 
/// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31).
306 
pub async fn rerun_job(
307 
    State(app): State<Shared>,
308 
    viewer: Viewer,
309 
    Path((owner, name, number, job)): Path<(String, String, i64, String)>,
310 
    Form(f): Form<RerunForm>,
311 
) -> Result<Response, Error> {
312 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
313 
    let me = require(&viewer, &shell)?;
314 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
315 
    csrf(&viewer, &f.csrf_token, &shell)?;
316 
    let run = fdb::run(&app.db, row.id, number)
317 
        .await
318 
        .map_err(|e| Error::from_db(shell.clone(), e))?
319 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
320 
    Ok(match crate::flow::rerun_job(&app, &row, &run, &job, &me.account.username).await {
321 
        Ok(_) => {
322 
            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("{job} of #{number}"))
323 
                .await
324 
                .ok();
325 
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
326 
        }
327 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
328 
    })
329 
}
330
745d225d Andy McCall ↶ 331 
/// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32).
332 
pub async fn cancel(
333 
    State(app): State<Shared>,
334 
    viewer: Viewer,
335 
    Path((owner, name, number)): Path<(String, String, i64)>,
336 
    Form(f): Form<RerunForm>,
337 
) -> Result<Response, Error> {
338 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
339 
    let me = require(&viewer, &shell)?;
340 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
341 
    csrf(&viewer, &f.csrf_token, &shell)?;
342 
    let run = fdb::run(&app.db, row.id, number)
343 
        .await
344 
        .map_err(|e| Error::from_db(shell.clone(), e))?
345 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
346 
    Ok(match crate::flow::cancel(&app, &row, &run, &me.account.username).await {
347 
        Ok(()) => {
348 
            db::accounts::audit(&app.db, Some(&me.account), "flow.cancel", &format!("{owner}/{name}"), &format!("#{number}")).await.ok();
349 
            Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()
350 
        }
351 
        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),
352 
    })
353 
}
354
107eaba5 Andy McCall ↶ 355 
#[derive(Deserialize)]
356 
pub struct ExpediteForm {
357 
    #[serde(rename = "_csrf")]
358 
    csrf_token: String,
359 
    #[serde(default)]
360 
    reason: String,
361 
}
362
363 
/// `POST /{owner}/{repo}/flow/{n}/expedite` (oxygenhub#244): a maintainer
364 
/// raises a running or queued run to urgent, saying why (an incident's
365 
/// fix, say). Audited and shown on the run; it lapses when the run ends.
366 
pub async fn expedite(
367 
    State(app): State<Shared>,
368 
    viewer: Viewer,
369 
    Path((owner, name, number)): Path<(String, String, i64)>,
370 
    Form(f): Form<ExpediteForm>,
371 
) -> Result<Response, Error> {
372 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
373 
    let me = require(&viewer, &shell)?;
374 
    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;
375 
    csrf(&viewer, &f.csrf_token, &shell)?;
376 
    let run = fdb::run(&app.db, row.id, number)
377 
        .await
378 
        .map_err(|e| Error::from_db(shell.clone(), e))?
379 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
b1860d60 Andy McCall ↶ 380 
    if let Err(msg) = crate::flow::expedite(&app, &row, &run, &me.account, &f.reason).await {
76f93eae Andy McCall ↶ 381 
        return Ok(Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response());
107eaba5 Andy McCall ↶ 382 
    }
383 
    Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response())
384 
}
385
1edac4b6 Andy McCall ↶ 386 
/// Percent-encode a message for a query string.
387 
fn urlencoding(s: &str) -> String {
388 
    s.bytes()
389 
        .map(|b| match b {
390 
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),
391 
            _ => format!("%{b:02X}"),
392 
        })
393 
        .collect()
42fa9554 Andy McCall ↶ 394 
}
395
396 
/// A job placed in the graph.
397 
pub struct Node {
398 
    pub j: fdb::Job,
399 
    pub x: i64,
400 
    pub y: i64,
401 
    pub duration: String,
402 
}
403
404 
pub struct Edge {
405 
    pub d: String,
406 
    pub status: String,
407 
}
408
409 
pub struct Graph {
410 
    pub nodes: Vec<Node>,
411 
    pub edges: Vec<Edge>,
412 
    pub width: i64,
413 
    pub height: i64,
414 
}
415
416 
pub const NODE_W: i64 = 200;
417 
pub const NODE_H: i64 = 48;
418 
const COL_GAP: i64 = 72;
419 
const ROW_GAP: i64 = 18;
420 
const PAD: i64 = 12;
421
422 
/// Columns by dependency depth, jobs in each column in name order.
423 
pub fn layout(jobs: Vec<fdb::Job>) -> Graph {
424 
    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();
425 
    for j in jobs {
426 
        by_depth.entry(j.depth).or_default().push(j);
427 
    }
428 
    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;
429 
    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;
430 
    let mut nodes = Vec::new();
431 
    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {
432 
        js.sort_by(|a, b| a.name.cmp(&b.name));
433 
        let n = js.len() as i64;
434 
        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;
435 
        let top = (height - column_h) / 2;
436 
        for (i, j) in js.into_iter().enumerate() {
437 
            let duration = crate::flow::duration(&j);
438 
            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });
439 
        }
440 
    }
441 
    let mut edges = Vec::new();
442 
    for n in &nodes {
443 
        for d in n.j.depends_on() {
444 
            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {
445 
                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);
446 
                let (x2, y2) = (n.x, n.y + NODE_H / 2);
447 
                let mid = (x1 + x2) / 2;
448 
                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });
449 
            }
450 
        }
451 
    }
452 
    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);
453 
    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }
454 
}
455
456 
pub fn glyph(status: &str) -> &'static str {
457 
    match status {
458 
        "passed" => "✓",
459 
        "failed" => "✕",
460 
        "running" => "●",
461 
        "skipped" => "⤼",
745d225d Andy McCall ↶ 462 
        "cancelled" => "⊘",
42fa9554 Andy McCall ↶ 463 
        "error" => "!",
464 
        _ => "○",
465 
    }
466 
}
467
0b203bbf Andy McCall ↶ 468 
/// Failed tests a run page lists per job; the rest are counted.
469 
const SHOWN_FAILURES: i64 = 50;
470
d9b3b846 Andy McCall ↶ 471 
/// A stretch of a job's log as the run page shows it: a line, or a
472 
/// section the job marked with `::group::TITLE` … `::endgroup::`
473 
/// (oxygenflow#31), shown folded with how long it took.
474 
pub enum Segment {
475 
    Line(fdb::LogLine),
476 
    Group(Section),
477 
}
478
479 
pub struct Section {
480 
    pub title: String,
481 
    /// When it began, and ended (`None` while it is still going).
482 
    pub from_t: i64,
483 
    pub to_t: Option<i64>,
484 
    pub lines: Vec<fdb::LogLine>,
485 
}
486
487 
/// A log's lines in sections. A section left open is still going; a new
488 
/// `::group::` ends the one before (sections do not nest).
489 
pub fn sections(lines: &[fdb::LogLine]) -> Vec<Segment> {
490 
    let mut out: Vec<Segment> = Vec::new();
491 
    let mut open: Option<Section> = None;
492 
    for l in lines {
493 
        if let Some(title) = l.text.strip_prefix("::group::") {
494 
            if let Some(g) = open.take() {
495 
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
496 
            }
497 
            open = Some(Section { title: title.trim().to_string(), from_t: l.t, to_t: None, lines: Vec::new() });
498 
        } else if l.text.trim() == "::endgroup::" {
499 
            if let Some(g) = open.take() {
500 
                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));
501 
            }
502 
        } else {
503 
            match open.as_mut() {
504 
                Some(g) => g.lines.push(l.clone()),
505 
                None => out.push(Segment::Line(l.clone())),
506 
            }
507 
        }
508 
    }
509 
    if let Some(g) = open {
510 
        out.push(Segment::Group(g));
511 
    }
512 
    out
513 
}
514
515 
/// Milliseconds as the log shows them: `12.3s`.
516 
pub fn secs(ms: impl std::borrow::Borrow<i64>) -> String {
517 
    let ms = *ms.borrow();
518 
    format!("{}.{}s", ms / 1000, ms % 1000 / 100)
519 
}
520
42fa9554 Andy McCall ↶ 521 
pub struct JobLog {
522 
    pub j: fdb::Job,
a3c8288b Andy McCall ↶ 523 
    pub artifacts: Vec<fdb::Artifact>,
42fa9554 Andy McCall ↶ 524 
    pub lines: Vec<fdb::LogLine>,
d9b3b846 Andy McCall ↶ 525 
    /// The lines in sections (oxygenflow#31).
526 
    pub segments: Vec<Segment>,
42fa9554 Andy McCall ↶ 527 
    pub duration: String,
528 
    /// The last line shown; the live view asks for what follows.
529 
    pub last_seq: i64,
554ccacf Andy McCall ↶ 530 
    /// The job's coverage report, when it uploaded one (#113): its id and percentage.
531 
    pub coverage: Option<(i64, String)>,
0b203bbf Andy McCall ↶ 532 
    /// The job's test report, when it uploaded one (#114), and its first
533 
    /// failures.
534 
    pub tests: Option<crate::junit::Report>,
535 
    pub failures: Vec<crate::junit::Case>,
42fa9554 Andy McCall ↶ 536 
}
537
538 
#[derive(Template)]
539 
#[template(path = "flow_run.html")]
540 
pub struct RunPage {
541 
    shell: Shell,
542 
    repo: RepoShell,
543 
    run: fdb::Run,
544 
    graph: Graph,
545 
    logs: Vec<JobLog>,
546 
    duration: String,
1edac4b6 Andy McCall ↶ 547 
    /// Whether the viewer may re-run it.
548 
    can_rerun: bool,
745d225d Andy McCall ↶ 549 
    /// Whether the viewer may cancel it.
550 
    can_cancel: bool,
107eaba5 Andy McCall ↶ 551 
    /// May raise the run to urgent (oxygenhub#244): maintainers, while it runs.
552 
    can_expedite: bool,
bf3d0754 Andy McCall ↶ 553 
    /// Where a live run stands among the runners' work.
554 
    queue: Option<crate::flow::QueueInfo>,
10708ab8 Andy McCall ↶ 555 
    /// Releases a writer may attach this run's artifacts to.
556 
    releases: Vec<db::releases::Release>,
417184c7 Andy McCall ↶ 557 
    /// Jobs that wait, or waited, for a person (oxygenflow#57).
558 
    approvals: Vec<ApprovalView>,
42fa9554 Andy McCall ↶ 559 
}
560
417184c7 Andy McCall ↶ 561 
/// A job's approval as the run page shows it.
562 
pub struct ApprovalView {
563 
    pub job: String,
564 
    pub by: String,
565 
    pub others_only: bool,
566 
    /// `waiting`, `approved`, `rejected` or `expired`.
567 
    pub state: &'static str,
568 
    pub who: String,
569 
    pub when_ms: i64,
570 
    /// Whether the viewer may approve or reject it now, else why not.
571 
    pub may: Result<(), String>,
47e37402 Andy McCall ↶ 572 
    /// It takes an approval signed with the approver's key (#235): the
573 
    /// command that signs it, and the approval's hash once sealed.
574 
    pub sealed: bool,
575 
    pub command: String,
576 
    pub seal: String,
417184c7 Andy McCall ↶ 577 
}
578
579 
impl ApprovalView {
580 
    pub fn may_decide(&self) -> bool {
581 
        self.may.is_ok()
582 
    }
583
584 
    pub fn why_not(&self) -> String {
585 
        self.may.clone().err().unwrap_or_default()
586 
    }
587 
}
588
42fa9554 Andy McCall ↶ 589 
/// Lines of a job's log a page shows before "the rest is loaded as it comes".
590 
const PAGE_LINES: i64 = 5000;
591
592 
pub async fn run(
593 
    State(app): State<Shared>,
594 
    viewer: Viewer,
595 
    Path((owner, name, id)): Path<(String, String, i64)>,
596 
) -> Result<Page<RunPage>, Error> {
597 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
598 
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
599 
    let run = fdb::run(&app.db, row.id, id)
600 
        .await
601 
        .map_err(|e| Error::from_db(shell.clone(), e))?
602 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
603 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
554ccacf Andy McCall ↶ 604 
    let reports = crate::coverage::for_run(&app.db, run.id).await.unwrap_or_default();
0b203bbf Andy McCall ↶ 605 
    let test_reports = crate::junit::for_run(&app.db, run.id).await.unwrap_or_default();
42fa9554 Andy McCall ↶ 606 
    let mut logs = Vec::new();
607 
    for j in &jobs {
608 
        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();
609 
        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);
a3c8288b Andy McCall ↶ 610 
        let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default();
554ccacf Andy McCall ↶ 611 
        let coverage = reports
612 
            .iter()
40dd3f86 Andy McCall ↶ 613 
            .find(|r| r.job_id == Some(j.id))
554ccacf Andy McCall ↶ 614 
            .map(|r| (r.id, r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into())));
0b203bbf Andy McCall ↶ 615 
        let tests = test_reports.iter().find(|r| r.job_id == j.id).cloned();
616 
        let failures = match &tests {
617 
            Some(r) if r.broken() > 0 => crate::junit::failures(&app.db, r.id, SHOWN_FAILURES).await.unwrap_or_default(),
618 
            _ => vec![],
619 
        };
d9b3b846 Andy McCall ↶ 620 
        let segments = sections(&lines);
621 
        logs.push(JobLog {
622 
            duration: crate::flow::duration(j),
623 
            j: j.clone(),
624 
            artifacts,
625 
            lines,
626 
            segments,
627 
            last_seq,
628 
            coverage,
629 
            tests,
630 
            failures,
631 
        });
42fa9554 Andy McCall ↶ 632 
    }
633 
    let duration = match run.finished_ms {
634 
        Some(f) => crate::flow::human_ms(f - run.created_ms),
635 
        None => crate::flow::human_ms(db::now_ms() - run.created_ms),
636 
    };
417184c7 Andy McCall ↶ 637 
    let mut approvals = Vec::new();
638 
    for j in jobs.iter().filter(|j| !j.approval.is_empty()) {
639 
        let Some(a) = serde_json::from_str::<flow_core::Job>(&j.spec).ok().and_then(|s| s.approval) else { continue };
640 
        let state = match (j.approval.as_str(), j.status.as_str()) {
641 
            ("waiting", "queued") => "waiting",
642 
            ("approved", _) => "approved",
643 
            (_, "cancelled") if !j.approved_by.is_empty() => "rejected",
644 
            (_, "cancelled") if j.error.starts_with("no approval within") => "expired",
645 
            _ => continue,
646 
        };
647 
        let may = match (state, viewer.account()) {
648 
            ("waiting", Some(me)) => crate::flow::may_decide(&app, &row, &run, j, me).await.map(|_| ()),
649 
            ("waiting", None) => Err("sign in to approve".into()),
650 
            _ => Err(String::new()),
651 
        };
652 
        approvals.push(ApprovalView {
47e37402 Andy McCall ↶ 653 
            sealed: a.sealed,
654 
            command: format!("oxy seal --flow {}/{owner}/{name}/flow/{} --job {}", app.config.base_url(), run.number, j.name),
655 
            seal: j.approval_seal.clone(),
417184c7 Andy McCall ↶ 656 
            job: j.name.clone(),
657 
            by: a.by,
658 
            others_only: a.others_only,
659 
            state,
05f1a85a Andy McCall ↶ 660 
            who: if j.approved_via.is_empty() { j.approved_by.clone() } else { format!("{} via {}", j.approved_by, j.approved_via) },
417184c7 Andy McCall ↶ 661 
            when_ms: j.approved_ms.unwrap_or(0),
662 
            may,
663 
        });
664 
    }
42fa9554 Andy McCall ↶ 665 
    let graph = layout(jobs);
745d225d Andy McCall ↶ 666 
    let can_write = viewer.account().is_some() && role >= Role::Write;
667 
    let can_rerun = !run.live() && can_write && crate::flow::enabled(&app, &row);
668 
    let can_cancel = run.live() && can_write;
107eaba5 Andy McCall ↶ 669 
    let can_expedite = run.live() && run.priority.is_empty() && viewer.account().is_some() && role >= Role::Maintain;
bf3d0754 Andy McCall ↶ 670 
    let queue = if run.live() {
671 
        let q = fdb::queue(&app.db).await.unwrap_or_default();
672 
        Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some())
673 
    } else {
674 
        None
10708ab8 Andy McCall ↶ 675 
    };
676 
    let releases = if viewer.account().is_some() && role >= Role::Write && logs.iter().any(|l| !l.artifacts.is_empty()) {
677 
        db::releases::list(&app.db, row.id).await.unwrap_or_default()
678 
    } else {
679 
        vec![]
bf3d0754 Andy McCall ↶ 680 
    };
10708ab8 Andy McCall ↶ 681 
    Ok(Page(RunPage {
682 
        shell,
683 
        repo: RepoShell::new(&app, &row, "flow").with_role(role),
684 
        run,
685 
        graph,
686 
        logs,
687 
        duration,
688 
        can_rerun,
745d225d Andy McCall ↶ 689 
        can_cancel,
107eaba5 Andy McCall ↶ 690 
        can_expedite,
10708ab8 Andy McCall ↶ 691 
        queue,
692 
        releases,
417184c7 Andy McCall ↶ 693 
        approvals,
10708ab8 Andy McCall ↶ 694 
    }))
417184c7 Andy McCall ↶ 695 
}
696
697 
#[derive(Deserialize)]
698 
pub struct DecideForm {
699 
    #[serde(rename = "_csrf")]
700 
    csrf_token: String,
701 
    /// `approve` or `reject`.
702 
    action: String,
703 
}
704
705 
/// `POST /{owner}/{name}/flow/{id}/jobs/{job}/decide`: approve or reject
706 
/// a job that waits for a person (oxygenflow#57).
707 
pub async fn decide(
708 
    State(app): State<Shared>,
709 
    viewer: Viewer,
710 
    Path((owner, name, number, job)): Path<(String, String, i64, String)>,
711 
    Form(f): Form<DecideForm>,
712 
) -> Result<Response, Error> {
713 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);
714 
    let me = require(&viewer, &shell)?;
715 
    let (_, row, _) = lookup(&app, &shell, &viewer, &owner, &name).await?;
716 
    csrf(&viewer, &f.csrf_token, &shell)?;
717 
    let run = fdb::run(&app.db, row.id, number)
718 
        .await
719 
        .map_err(|e| Error::from_db(shell.clone(), e))?
720 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
721 
    let approve = match f.action.as_str() {
722 
        "approve" => true,
723 
        "reject" => false,
724 
        _ => return Err(Error::bad(shell, "approve or reject")),
725 
    };
05f1a85a Andy McCall ↶ 726 
    match crate::flow::decide(&app, &row, &run, &job, &me.account, approve, None).await {
417184c7 Andy McCall ↶ 727 
        Ok(_) => Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response()),
728 
        Err(why) => Err(Error::bad(shell, why)),
729 
    }
42fa9554 Andy McCall ↶ 730 
}
731
732 
#[derive(Deserialize)]
733 
pub struct LiveQuery {
734 
    /// `job:seq,job:seq`: the last log line each open job already shows.
735 
    after: Option<String>,
736 
}
737
738 
/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the
739 
/// run page while it is live.
740 
pub async fn live(
741 
    State(app): State<Shared>,
742 
    viewer: Viewer,
743 
    Path((owner, name, id)): Path<(String, String, i64)>,
744 
    Query(q): Query<LiveQuery>,
745 
) -> Result<Response, Error> {
746 
    let shell = Shell::new(&app, "", &viewer);
747 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
748 
    let run = fdb::run(&app.db, row.id, id)
749 
        .await
750 
        .map_err(|e| Error::from_db(shell.clone(), e))?
751 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
752 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
753 
    let after: std::collections::HashMap<i64, i64> = q
754 
        .after
755 
        .unwrap_or_default()
756 
        .split(',')
757 
        .filter_map(|p| p.split_once(':'))
758 
        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))
759 
        .collect();
760 
    let mut out = Vec::new();
761 
    for j in &jobs {
762 
        let lines = match after.get(&j.id) {
763 
            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),
764 
            None => vec![],
765 
        };
766 
        out.push(serde_json::json!({
767 
            "id": j.id,
768 
            "name": j.name,
769 
            "status": j.status,
770 
            "glyph": glyph(&j.status),
771 
            "duration": crate::flow::duration(j),
772 
            "exit_code": j.exit_code,
773 
            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),
774 
        }));
775 
    }
776 
    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())
777 
}
736f6813 Andy McCall ↶ 778
779 
// ------------------------------------------------------------ state stamps
780
781 
/// In rendered Markdown, the image path that is the viewed state's stamp.
782 
pub const STAMP_IMAGE: &str = ".flow/stamp";
783 
/// In rendered Markdown, the link path to the viewed state's run.
784 
pub const RUN_LINK: &str = ".flow/run";
785
786 
/// Resolve a ref to its state address, and whether the ref can move.
787 
async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {
788 
    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;
789 
    let spec = r.to_string();
790 
    app.repos
791 
        .read(&repo, move |repo, cache| {
792 
            let t = crate::repos::view::resolve(repo, cache, &spec)?;
793 
            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);
794 
            Ok((t.permalink().unwrap_or_default(), moves))
795 
        })
796 
        .await
797 
}
798
799 
/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).
800 
pub async fn stamp(
801 
    State(app): State<Shared>,
802 
    viewer: Viewer,
803 
    Path((owner, name, r)): Path<(String, String, String)>,
804 
) -> Result<Response, Error> {
805 
    let shell = Shell::new(&app, "", &viewer);
806 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
807 
    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
808 
    if moves && !heads.is_empty() {
809 
        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();
810 
        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));
811 
        return Ok(res);
812 
    }
813 
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
814 
    let jobs = match &run {
815 
        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,
816 
        None => vec![],
817 
    };
818 
    let svg = stamp_svg(&heads, run.as_ref(), &jobs);
819 
    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };
820 
    Ok((
821 
        [
822 
            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),
823 
            (axum::http::header::CACHE_CONTROL, cache),
824 
            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),
825 
        ],
826 
        svg,
827 
    )
828 
        .into_response())
829 
}
830
831 
/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.
832 
pub async fn at(
833 
    State(app): State<Shared>,
834 
    viewer: Viewer,
835 
    Path((owner, name, r)): Path<(String, String, String)>,
836 
) -> Result<Response, Error> {
837 
    let shell = Shell::new(&app, "", &viewer);
838 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
839 
    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
840 
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
841 
    let to = match run {
a0d4bbfb Andy McCall ↶ 842 
        Some(r) => format!("/{owner}/{name}/flow/{}", r.number),
736f6813 Andy McCall ↶ 843 
        None => format!("/{owner}/{name}/flow"),
844 
    };
845 
    Ok(axum::response::Redirect::to(&to).into_response())
846 
}
847
848 
fn xml(s: &str) -> String {
849 
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
850 
}
851
852 
/// The stamp: the mark, one glyph per job, the status, the state and the
853 
/// date; the same facts machine-readable in `<metadata>`. Fixed colours,
854 
/// because an image does not follow the page's theme.
855 
pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {
856 
    const CH: f64 = 6.6; // IBM Plex Mono at 11px
857 
    let colour = |s: &str| match s {
858 
        "passed" => "#5fe3a1",
859 
        "failed" | "error" => "#ff5e7a",
860 
        "running" => "#00e0f0",
861 
        _ => "#8592ad",
862 
    };
863 
    let short: String = heads.chars().take(12).collect();
864 
    let (status, date) = match run {
865 
        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),
866 
        None => ("no run", String::new()),
867 
    };
868 
    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();
869 
    ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name)));
870 
    let label = "OXYGENFLOW";
9991530b Andy McCall ↶ 871 
    let label_w = 18.0 + label.len() as f64 * 7.4 + 8.0; // Barlow bold 10.5px, spaced
736f6813 Andy McCall ↶ 872 
    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };
873 
    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };
874 
    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;
875 
    let w = (label_w + right_w).ceil();
876 
    let mut glyphs = String::new();
877 
    for (i, j) in ordered.iter().enumerate() {
878 
        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;
879 
        glyphs.push_str(&format!(
880 
            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,
881 
            colour(&j.status),
882 
            glyph(&j.status)
883 
        ));
884 
    }
885 
    let meta = serde_json::json!({
886 
        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),
a0d4bbfb Andy McCall ↶ 887 
        "run": run.map(|r| r.number),
736f6813 Andy McCall ↶ 888 
        "status": status,
889 
        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),
890 
    });
891 
    let title = match run {
a0d4bbfb Andy McCall ↶ 892 
        Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.number),
736f6813 Andy McCall ↶ 893 
        None => format!("OxygenFlow: no run for state {short}"),
894 
    };
895 
    format!(
896 
        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>"##,
897 
        t = xml(&title),
898 
        m = xml(&meta.to_string()),
899 
        lw = label_w,
900 
        lx = label_w - 4.0,
901 
        tx = label_w + 8.0 + glyphs_w,
902 
        sc = colour(status),
903 
        tail = xml(&tail),
904 
        wi = w - 1.0,
905 
    )
906 
}
cc19697f Andy McCall ↶ 907
908 
/// A receipt kept in a repository (Oxygen spec §14.6), as pages show it.
909 
#[derive(Clone, Debug)]
910 
pub struct ReceiptRow {
911 
    pub attest: String,
912 
    pub job: String,
913 
    pub exit_code: u32,
914 
    pub duration: String,
915 
    pub runner: String,
916 
    pub run: String,
917 
    /// Why it does not verify, or that its runner's key was revoked.
918 
    pub problem: String,
919 
    pub made_ms: i64,
920 
}
921
922 
impl ReceiptRow {
923 
    pub fn passed(&self) -> bool {
924 
        self.exit_code == 0
925 
    }
926 
}
927
928 
/// The receipts kept for the state of `heads`.
929 
pub async fn receipts_of(app: &Shared, name: &crate::repos::RepoName, heads: Vec<oxy_core::hash::ChangeHash>) -> Vec<ReceiptRow> {
930 
    let base = app.config.base_url();
931 
    let found = app.repos.read(name, move |repo, _| Ok(repo.receipts(&heads)?)).await.unwrap_or_default();
932 
    found
933 
        .into_iter()
934 
        .map(|i| {
935 
            let r = &i.signed.receipt;
936 
            let run = i.signed.run().to_string();
937 
            ReceiptRow {
938 
                attest: i.attest.to_hex(),
939 
                job: r.job_name.clone(),
940 
                exit_code: r.exit_code,
941 
                duration: crate::flow::human_ms(r.duration_ms as i64),
942 
                runner: i.signed.runner.name.clone(),
943 
                // Only links to this hub are followed from the page.
944 
                run: if run.starts_with(base) { run } else { String::new() },
945 
                problem: match (i.problem, i.revoked) {
946 
                    (Some(p), _) => p,
947 
                    (None, true) => "the runner's key was revoked".into(),
948 
                    (None, false) => String::new(),
949 
                },
950 
                made_ms: i.signed.attest.hlc.wall_ms as i64,
951 
            }
952 
        })
953 
        .collect()
954 
}
955
956 
/// `GET /{owner}/{name}/receipts/{attest}`: a signed receipt, for
957 
/// `oxy receipt verify` or `oxy receipt import`.
958 
pub async fn receipt_file(
959 
    State(app): State<Shared>,
960 
    viewer: Viewer,
961 
    Path((owner, name, attest)): Path<(String, String, String)>,
962 
) -> Result<Response, Error> {
963 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/receipts/{attest}"), &viewer);
964 
    let (repo, _row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
965 
    let Some(r) = oxy_core::hash::ObjectRef::from_hex(attest.trim_end_matches(".receipt")) else {
966 
        return Err(Error::not_found(shell));
967 
    };
968 
    let (job, bytes) = app
969 
        .repos
970 
        .read(&repo, move |repo, _| {
971 
            let i = repo.receipt(&r)?;
972 
            Ok((i.signed.receipt.job_name.clone(), i.signed.encode()))
973 
        })
974 
        .await
975 
        .map_err(|_| Error::not_found(shell.clone()))?;
976 
    let file: String = job.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).collect();
977 
    Ok((
978 
        [
979 
            (axum::http::header::CONTENT_TYPE, "application/oxy-receipt".to_string()),
980 
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{file}-{}.receipt\"", r.short(12))),
a3c8288b Andy McCall ↶ 981 
        ],
982 
        bytes,
983 
    )
984 
        .into_response())
985 
}
986
987 
/// `GET /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}`: an artifact
988 
/// of a run's job, read from the repository.
989 
pub async fn artifact_file(
990 
    State(app): State<Shared>,
991 
    viewer: Viewer,
992 
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
993 
) -> Result<Response, Error> {
994 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}"), &viewer);
995 
    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
996 
    let run = fdb::run(&app.db, row.id, id)
997 
        .await
998 
        .map_err(|e| Error::from_db(shell.clone(), e))?
999 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
1000 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1001 
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
1002 
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1003 
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
1004 
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
1005 
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;
1006 
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
1007 
    Ok((
1008 
        [
1009 
            (axum::http::header::CONTENT_TYPE, "application/octet-stream".to_string()),
1010 
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}\"", file.replace('"', ""))),
cc19697f Andy McCall ↶ 1011 
        ],
1012 
        bytes,
1013 
    )
1014 
        .into_response())
10708ab8 Andy McCall ↶ 1015 
}
1016
076c7a7c Andy McCall ↶ 1017 
/// `GET …/artifacts/{job}/{artifact}/provenance` (oxygenhub#239): the
1018 
/// artifact's SLSA provenance in a DSSE envelope signed by the hub, for
1019 
/// `cosign verify-blob-attestation --key oxyhub-signing.pem --type
1020 
/// slsaprovenance1`. Only for a job whose signed receipt the hub kept.
1021 
pub async fn provenance(
1022 
    State(app): State<Shared>,
1023 
    viewer: Viewer,
1024 
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
1025 
) -> Result<Response, Error> {
1026 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/provenance"), &viewer);
1027 
    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
1028 
    let run = fdb::run(&app.db, row.id, id)
1029 
        .await
1030 
        .map_err(|e| Error::from_db(shell.clone(), e))?
1031 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
1032 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1033 
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
1034 
    if j.receipt_attest.is_empty() {
1035 
        return Err(Error::bad(shell, "This job has no signed receipt the hub kept, so the hub states no provenance for it.".to_string()));
1036 
    }
1037 
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1038 
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
1039 
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
1040 
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|_| Error::not_found(shell.clone()))?;
1041 
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
1042 
    let base = app.config.base_url();
1043 
    let subject = crate::provenance::Subject {
1044 
        name: &file,
1045 
        bytes: &bytes,
1046 
        repo: &format!("{owner}/{name}"),
1047 
        run_url: &format!("{base}/{owner}/{name}/flow/{id}"),
1048 
        bookmark: &run.bookmark,
1049 
        state: &run.state,
1050 
        heads: &run.heads,
1051 
        pipeline_hash: &run.pipeline_hash,
1052 
        job: &j.name,
1053 
        runner: &j.runner,
1054 
        receipt_signer: &j.receipt_signer,
1055 
        receipt_attest: &j.receipt_attest,
1056 
        started_ms: j.started_ms,
1057 
        finished_ms: j.finished_ms,
1058 
    };
1059 
    let statement = crate::provenance::statement(base, &subject);
1060 
    let envelope = crate::provenance::envelope(&app, &statement).map_err(|e| Error::bad(shell.clone(), e.to_string()))?;
1061 
    Ok((
1062 
        [
1063 
            (axum::http::header::CONTENT_TYPE, "application/json".to_string()),
1064 
            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{}.intoto.json\"", file.replace('"', ""))),
1065 
        ],
1066 
        serde_json::to_string(&envelope).unwrap_or_default(),
1067 
    )
1068 
        .into_response())
1069 
}
1070
1071 
/// `GET /.well-known/oxyhub-signing.pem`: the hub's Ed25519 public key, as
1072 
/// cosign and other verifiers take it (oxygenhub#239).
1073 
pub async fn signing_key(State(app): State<Shared>) -> Response {
1074 
    ([(axum::http::header::CONTENT_TYPE, "application/x-pem-file")], crate::provenance::public_pem(&app)).into_response()
1075 
}
1076
10708ab8 Andy McCall ↶ 1077 
#[derive(Deserialize)]
1078 
pub struct AttachForm {
1079 
    _csrf: String,
1080 
    release: i64,
1081 
}
1082
1083 
/// `POST /{owner}/{name}/flow/{id}/artifacts/{job}/{artifact}/attach`: add
1084 
/// an artifact to a release as an asset, named after its path; the release
1085 
/// page then shows the run and receipt that built it.
1086 
pub async fn attach(
1087 
    State(app): State<Shared>,
1088 
    viewer: Viewer,
1089 
    Path((owner, name, id, job, artifact)): Path<(String, String, i64, String, String)>,
1090 
    Form(f): Form<AttachForm>,
1091 
) -> Result<Response, Error> {
1092 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
1093 
    require(&viewer, &shell)?;
1094 
    csrf(&viewer, &f._csrf, &shell)?;
1095 
    let (repo, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;
1096 
    let run = fdb::run(&app.db, row.id, id)
1097 
        .await
1098 
        .map_err(|e| Error::from_db(shell.clone(), e))?
1099 
        .ok_or_else(|| Error::not_found(shell.clone()))?;
1100 
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1101 
    let Some(j) = jobs.iter().find(|j| j.name == job) else { return Err(Error::not_found(shell)) };
6f52a041 Andy McCall ↶ 1102 
    // A failing job's `when "always"` artifacts are not in its receipt, so
1103 
    // they cannot stand for a release (oxygenflow#37).
1104 
    if j.status != "passed" {
1105 
        return Err(Error::bad(shell, "only a passing job's artifacts can be attached to a release"));
1106 
    }
10708ab8 Andy McCall ↶ 1107 
    let arts = fdb::artifacts(&app.db, j.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1108 
    let Some(a) = arts.into_iter().find(|a| a.name == artifact) else { return Err(Error::not_found(shell)) };
1109 
    let releases = db::releases::list(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1110 
    let Some(rel) = releases.into_iter().find(|r| r.id == f.release) else { return Err(Error::not_found(shell)) };
1111 
    let Some(blob) = oxy_core::hash::ObjectRef::from_hex(&a.blob) else { return Err(Error::not_found(shell)) };
1112 
    let bytes = app.repos.read(&repo, move |r, _| Ok(r.read_blob(&blob)?)).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
1113 
    let file = a.path.rsplit('/').next().unwrap_or(&a.name).to_string();
1114 
    super::releases::store_asset(&app, rel.id, &file, "application/octet-stream", &bytes)
1115 
        .await
1116 
        .map_err(|m| Error::bad(shell.clone(), m))?;
1117 
    Ok(Redirect::to(&format!("/{owner}/{name}/releases/tag/{}", rel.tag)).into_response())
cc19697f Andy McCall ↶ 1118 
}
8933faff Andy McCall ↶ 1119
1120 
/// `/{owner}/{name}/flow/badge.svg`: the latest run on the default
1121 
/// bookmark as a small SVG for a README, for whoever may read the
1122 
/// repository. `passed` green, `failed` or `error` red, a run still going
1123 
/// blue, none grey.
1124 
pub async fn badge(
1125 
    State(app): State<Shared>,
1126 
    viewer: Viewer,
1127 
    Path((owner, name)): Path<(String, String)>,
1128 
) -> Result<axum::response::Response, Error> {
1129 
    use axum::response::IntoResponse;
1130 
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
1131 
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
1132 
    let latest = fdb::latest_on(&app.db, row.id, &row.default_bookmark).await.map_err(|e| Error::from_db(shell.clone(), e))?;
1133 
    let (text, colour) = match latest.as_ref().map(|r| r.status.as_str()) {
1134 
        Some("passed") => ("passing", "#2da44e"),
1135 
        Some("failed") | Some("error") => ("failing", "#cf222e"),
1136 
        Some("cancelled") => ("cancelled", "#6e7781"),
1137 
        Some(_) => ("running", "#0969da"),
1138 
        None => ("no runs", "#6e7781"),
1139 
    };
1140 
    let label = "OxygenFlow";
1141 
    let (lw, tw) = (label.len() as u32 * 7 + 10, text.len() as u32 * 7 + 10);
1142 
    let svg = format!(
1143 
        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>"##,
1144 
        w = lw + tw,
1145 
        lx = lw / 2,
1146 
        tx = lw + tw / 2,
1147 
    );
1148 
    Ok((
1149 
        [(axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"), (axum::http::header::CACHE_CONTROL, "no-cache, max-age=60")],
1150 
        svg,
1151 
    )
1152 
        .into_response())
d9b3b846 Andy McCall ↶ 1153 
}
1154
1155 
#[cfg(test)]
1156 
mod section_tests {
1157 
    use super::*;
1158
1159 
    fn line(seq: i64, t: i64, text: &str) -> fdb::LogLine {
1160 
        fdb::LogLine { seq, t, stream: "out".into(), text: text.into() }
1161 
    }
1162
1163 
    #[test]
1164 
    fn marked_lines_fold_into_sections() {
1165 
        let lines = vec![
1166 
            line(1, 0, "booting"),
1167 
            line(2, 100, "::group::cargo build"),
1168 
            line(3, 200, "Compiling a"),
1169 
            line(4, 900, "Compiling b"),
1170 
            line(5, 4_300, "::endgroup::"),
1171 
            line(6, 4_400, "::group::cargo test"),
1172 
            line(7, 4_500, "test x ... ok"),
1173 
            line(8, 5_000, "::group::lint"),
1174 
            line(9, 5_100, "clean"),
1175 
        ];
1176 
        let segs = sections(&lines);
1177 
        assert_eq!(segs.len(), 4);
1178 
        assert!(matches!(&segs[0], Segment::Line(l) if l.text == "booting"));
1179 
        let Segment::Group(build) = &segs[1] else { panic!("a section") };
1180 
        assert_eq!((build.title.as_str(), build.from_t, build.to_t, build.lines.len()), ("cargo build", 100, Some(4_300), 2));
1181 
        let Segment::Group(test) = &segs[2] else { panic!("a section") };
1182 
        assert_eq!(test.to_t, Some(5_000), "a new section ends the one before");
1183 
        let Segment::Group(lint) = &segs[3] else { panic!("a section") };
1184 
        assert_eq!((lint.to_t, lint.lines.len()), (None, 1), "still going");
1185 
        assert_eq!(secs(4_200), "4.2s");
1186 
    }
8933faff Andy McCall ↶ 1187 
}