| 1 |
|
| 2 |
|
| 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 |
|
| 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 |
|
| 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 |
|
| 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 |
|
| 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 |
|
| 183 |
after: Option<String>, |
| 184 |
} |
| 185 |
|
| 186 |
|
| 187 |
|
| 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 |
} |