OXYGENHUB
# a0d4bbfbedd1 oxygenhub / crates / hub / src / web / flow.rs
14.2 KiB · Rust · last changed Number Flow runs per repository 1 month 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::{filters, lookup, Error, Page, RepoShell, Shell};
5
use crate::auth::Viewer;
6
use crate::db::{self, flow as fdb};
7
use crate::Shared;
8
use askama::Template;
9
use axum::extract::{Path, Query, State};
10
use axum::response::{IntoResponse, Response};
11
use axum::Json;
12
use serde::Deserialize;
13
14
pub struct RunRow {
15
    pub r: fdb::Run,
16
    pub duration: String,
17
    pub jobs: Vec<(String, String)>,
18
}
19
20
#[derive(Template)]
21
#[template(path = "flow_runs.html")]
22
pub struct RunsPage {
23
    shell: Shell,
24
    repo: RepoShell,
25
    runs: Vec<RunRow>,
26
    enabled: bool,
27
    older: Option<i64>,
28
}
29
30
#[derive(Deserialize)]
31
pub struct RunsQuery {
32
    before: Option<i64>,
33
}
34
35
const PER_PAGE: i64 = 30;
36
37
pub async fn runs(
38
    State(app): State<Shared>,
39
    viewer: Viewer,
40
    Path((owner, name)): Path<(String, String)>,
41
    Query(q): Query<RunsQuery>,
42
) -> Result<Page<RunsPage>, Error> {
43
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);
44
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
45
    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;
46
    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);
47
    let mut runs = Vec::new();
48
    for r in list.into_iter().take(PER_PAGE as usize) {
49
        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();
50
        let duration = match r.finished_ms {
51
            Some(f) => crate::flow::human_ms(f - r.created_ms),
52
            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),
53
            None => "–".into(),
54
        };
55
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r });
56
    }
57
    let enabled = crate::flow::enabled(&app, &row);
58
    Ok(Page(RunsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), runs, enabled, older }))
59
}
60
61
/// A job placed in the graph.
62
pub struct Node {
63
    pub j: fdb::Job,
64
    pub x: i64,
65
    pub y: i64,
66
    pub duration: String,
67
}
68
69
pub struct Edge {
70
    pub d: String,
71
    pub status: String,
72
}
73
74
pub struct Graph {
75
    pub nodes: Vec<Node>,
76
    pub edges: Vec<Edge>,
77
    pub width: i64,
78
    pub height: i64,
79
}
80
81
pub const NODE_W: i64 = 200;
82
pub const NODE_H: i64 = 48;
83
const COL_GAP: i64 = 72;
84
const ROW_GAP: i64 = 18;
85
const PAD: i64 = 12;
86
87
/// Columns by dependency depth, jobs in each column in name order.
88
pub fn layout(jobs: Vec<fdb::Job>) -> Graph {
89
    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();
90
    for j in jobs {
91
        by_depth.entry(j.depth).or_default().push(j);
92
    }
93
    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;
94
    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;
95
    let mut nodes = Vec::new();
96
    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {
97
        js.sort_by(|a, b| a.name.cmp(&b.name));
98
        let n = js.len() as i64;
99
        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;
100
        let top = (height - column_h) / 2;
101
        for (i, j) in js.into_iter().enumerate() {
102
            let duration = crate::flow::duration(&j);
103
            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });
104
        }
105
    }
106
    let mut edges = Vec::new();
107
    for n in &nodes {
108
        for d in n.j.depends_on() {
109
            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {
110
                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);
111
                let (x2, y2) = (n.x, n.y + NODE_H / 2);
112
                let mid = (x1 + x2) / 2;
113
                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });
114
            }
115
        }
116
    }
117
    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);
118
    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }
119
}
120
121
pub fn glyph(status: &str) -> &'static str {
122
    match status {
123
        "passed" => "✓",
124
        "failed" => "✕",
125
        "running" => "●",
126
        "skipped" => "⤼",
127
        "error" => "!",
128
        _ => "○",
129
    }
130
}
131
132
pub struct JobLog {
133
    pub j: fdb::Job,
134
    pub lines: Vec<fdb::LogLine>,
135
    pub duration: String,
136
    /// The last line shown; the live view asks for what follows.
137
    pub last_seq: i64,
138
}
139
140
#[derive(Template)]
141
#[template(path = "flow_run.html")]
142
pub struct RunPage {
143
    shell: Shell,
144
    repo: RepoShell,
145
    run: fdb::Run,
146
    graph: Graph,
147
    logs: Vec<JobLog>,
148
    duration: String,
149
}
150
151
/// Lines of a job's log a page shows before "the rest is loaded as it comes".
152
const PAGE_LINES: i64 = 5000;
153
154
pub async fn run(
155
    State(app): State<Shared>,
156
    viewer: Viewer,
157
    Path((owner, name, id)): Path<(String, String, i64)>,
158
) -> Result<Page<RunPage>, Error> {
159
    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);
160
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
161
    let run = fdb::run(&app.db, row.id, id)
162
        .await
163
        .map_err(|e| Error::from_db(shell.clone(), e))?
164
        .ok_or_else(|| Error::not_found(shell.clone()))?;
165
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
166
    let mut logs = Vec::new();
167
    for j in &jobs {
168
        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();
169
        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);
170
        logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), lines, last_seq });
171
    }
172
    let duration = match run.finished_ms {
173
        Some(f) => crate::flow::human_ms(f - run.created_ms),
174
        None => crate::flow::human_ms(db::now_ms() - run.created_ms),
175
    };
176
    let graph = layout(jobs);
177
    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration }))
178
}
179
180
#[derive(Deserialize)]
181
pub struct LiveQuery {
182
    /// `job:seq,job:seq`: the last log line each open job already shows.
183
    after: Option<String>,
184
}
185
186
/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the
187
/// run page while it is live.
188
pub async fn live(
189
    State(app): State<Shared>,
190
    viewer: Viewer,
191
    Path((owner, name, id)): Path<(String, String, i64)>,
192
    Query(q): Query<LiveQuery>,
193
) -> Result<Response, Error> {
194
    let shell = Shell::new(&app, "", &viewer);
195
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
196
    let run = fdb::run(&app.db, row.id, id)
197
        .await
198
        .map_err(|e| Error::from_db(shell.clone(), e))?
199
        .ok_or_else(|| Error::not_found(shell.clone()))?;
200
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
201
    let after: std::collections::HashMap<i64, i64> = q
202
        .after
203
        .unwrap_or_default()
204
        .split(',')
205
        .filter_map(|p| p.split_once(':'))
206
        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))
207
        .collect();
208
    let mut out = Vec::new();
209
    for j in &jobs {
210
        let lines = match after.get(&j.id) {
211
            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),
212
            None => vec![],
213
        };
214
        out.push(serde_json::json!({
215
            "id": j.id,
216
            "name": j.name,
217
            "status": j.status,
218
            "glyph": glyph(&j.status),
219
            "duration": crate::flow::duration(j),
220
            "exit_code": j.exit_code,
221
            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),
222
        }));
223
    }
224
    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())
225
}
226
227
// ------------------------------------------------------------ state stamps
228
229
/// In rendered Markdown, the image path that is the viewed state's stamp.
230
pub const STAMP_IMAGE: &str = ".flow/stamp";
231
/// In rendered Markdown, the link path to the viewed state's run.
232
pub const RUN_LINK: &str = ".flow/run";
233
234
/// Resolve a ref to its state address, and whether the ref can move.
235
async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {
236
    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;
237
    let spec = r.to_string();
238
    app.repos
239
        .read(&repo, move |repo, cache| {
240
            let t = crate::repos::view::resolve(repo, cache, &spec)?;
241
            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);
242
            Ok((t.permalink().unwrap_or_default(), moves))
243
        })
244
        .await
245
}
246
247
/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).
248
pub async fn stamp(
249
    State(app): State<Shared>,
250
    viewer: Viewer,
251
    Path((owner, name, r)): Path<(String, String, String)>,
252
) -> Result<Response, Error> {
253
    let shell = Shell::new(&app, "", &viewer);
254
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
255
    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
256
    if moves && !heads.is_empty() {
257
        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();
258
        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));
259
        return Ok(res);
260
    }
261
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
262
    let jobs = match &run {
263
        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,
264
        None => vec![],
265
    };
266
    let svg = stamp_svg(&heads, run.as_ref(), &jobs);
267
    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };
268
    Ok((
269
        [
270
            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),
271
            (axum::http::header::CACHE_CONTROL, cache),
272
            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),
273
        ],
274
        svg,
275
    )
276
        .into_response())
277
}
278
279
/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.
280
pub async fn at(
281
    State(app): State<Shared>,
282
    viewer: Viewer,
283
    Path((owner, name, r)): Path<(String, String, String)>,
284
) -> Result<Response, Error> {
285
    let shell = Shell::new(&app, "", &viewer);
286
    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
287
    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;
288
    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;
289
    let to = match run {
290
        Some(r) => format!("/{owner}/{name}/flow/{}", r.number),
291
        None => format!("/{owner}/{name}/flow"),
292
    };
293
    Ok(axum::response::Redirect::to(&to).into_response())
294
}
295
296
fn xml(s: &str) -> String {
297
    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
298
}
299
300
/// The stamp: the mark, one glyph per job, the status, the state and the
301
/// date; the same facts machine-readable in `<metadata>`. Fixed colours,
302
/// because an image does not follow the page's theme.
303
pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {
304
    const CH: f64 = 6.6; // IBM Plex Mono at 11px
305
    let colour = |s: &str| match s {
306
        "passed" => "#5fe3a1",
307
        "failed" | "error" => "#ff5e7a",
308
        "running" => "#00e0f0",
309
        _ => "#8592ad",
310
    };
311
    let short: String = heads.chars().take(12).collect();
312
    let (status, date) = match run {
313
        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),
314
        None => ("no run", String::new()),
315
    };
316
    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();
317
    ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name)));
318
    let label = "OXYGENFLOW";
319
    let label_w = 14.0 + label.len() as f64 * 6.2 + 8.0;
320
    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };
321
    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };
322
    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;
323
    let w = (label_w + right_w).ceil();
324
    let mut glyphs = String::new();
325
    for (i, j) in ordered.iter().enumerate() {
326
        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;
327
        glyphs.push_str(&format!(
328
            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,
329
            colour(&j.status),
330
            glyph(&j.status)
331
        ));
332
    }
333
    let meta = serde_json::json!({
334
        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),
335
        "run": run.map(|r| r.number),
336
        "status": status,
337
        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),
338
    });
339
    let title = match run {
340
        Some(r) => format!("OxygenFlow run #{} {status} on state {short}", r.number),
341
        None => format!("OxygenFlow: no run for state {short}"),
342
    };
343
    format!(
344
        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>"##,
345
        t = xml(&title),
346
        m = xml(&meta.to_string()),
347
        lw = label_w,
348
        lx = label_w - 4.0,
349
        tx = label_w + 8.0 + glyphs_w,
350
        sc = colour(status),
351
        tail = xml(&tail),
352
        wi = w - 1.0,
353
    )
354
}