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