//! The Flow tab (hub spec §13.2): runs of a repository, and a run's job

//! graph with live status and logs.

use super::settings::{csrf, require};

use super::{filters, lookup, lookup_role, Error, Page, RepoShell, Shell};

use crate::auth::{Access, Role, Viewer};

use crate::db::{self, flow as fdb};

use crate::Shared;

use askama::Template;

use axum::extract::{Form, Path, Query, State};

use axum::response::{IntoResponse, Redirect, Response};

use axum::Json;

use serde::Deserialize;

pub struct RunRow {

    pub r: fdb::Run,

    pub duration: String,

    pub jobs: Vec<(String, String)>,

}

#[derive(Template)]

#[template(path = "flow_runs.html")]

pub struct RunsPage {

    shell: Shell,

    repo: RepoShell,

    runs: Vec<RunRow>,

    enabled: bool,

    older: Option<i64>,

    /// Writers may start runs: the bookmarks to offer, default first.

    can_run: bool,

    bookmarks: Vec<String>,

    error: Option<String>,

}

#[derive(Deserialize)]

pub struct RunsQuery {

    before: Option<i64>,

    /// Why a "Run pipeline" or "Re-run" did not start.

    error: Option<String>,

}

const PER_PAGE: i64 = 30;

pub async fn runs(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Query(q): Query<RunsQuery>,

) -> Result<Page<RunsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);

    let mut runs = Vec::new();

    for r in list.into_iter().take(PER_PAGE as usize) {

        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();

        let duration = match r.finished_ms {

            Some(f) => crate::flow::human_ms(f - r.created_ms),

            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),

            None => "–".into(),

        };

        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r });

    }

    let enabled = crate::flow::enabled(&app, &row);

    let can_run = enabled && viewer.account().is_some() && role >= Role::Write;

    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };

    Ok(Page(RunsPage {

        shell,

        repo: RepoShell::new(&app, &row, "flow").with_role(role),

        runs,

        enabled,

        older,

        can_run,

        bookmarks,

        error: q.error,

    }))

}

/// The repository's bookmarks, its default first.

async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {

    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };

    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();

    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));

    names

}

#[derive(Deserialize)]

pub struct StartForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    bookmark: String,

}

/// `POST /{owner}/{name}/flow/start`: "Run pipeline" on a bookmark.

pub async fn start(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Form(f): Form<StartForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    Ok(match crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username).await {

        Ok(n) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {}", f.bookmark.trim()))

                .await

                .ok();

            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()

        }

        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),

    })

}

#[derive(Deserialize)]

pub struct RerunForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

/// `POST /{owner}/{name}/flow/{number}/rerun`.

pub async fn rerun(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number)): Path<(String, String, i64)>,

    Form(f): Form<RerunForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let run = fdb::run(&app.db, row.id, number)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    Ok(match crate::flow::rerun(&app, &row, &run, &me.account.username).await {

        Ok(n) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.rerun", &format!("{owner}/{name}"), &format!("#{n} of #{number}"))

                .await

                .ok();

            Redirect::to(&format!("/{owner}/{name}/flow/{n}")).into_response()

        }

        Err(msg) => Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding(&msg))).into_response(),

    })

}

/// Percent-encode a message for a query string.

fn urlencoding(s: &str) -> String {

    s.bytes()

        .map(|b| match b {

            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),

            _ => format!("%{b:02X}"),

        })

        .collect()

}

/// A job placed in the graph.

pub struct Node {

    pub j: fdb::Job,

    pub x: i64,

    pub y: i64,

    pub duration: String,

}

pub struct Edge {

    pub d: String,

    pub status: String,

}

pub struct Graph {

    pub nodes: Vec<Node>,

    pub edges: Vec<Edge>,

    pub width: i64,

    pub height: i64,

}

pub const NODE_W: i64 = 200;

pub const NODE_H: i64 = 48;

const COL_GAP: i64 = 72;

const ROW_GAP: i64 = 18;

const PAD: i64 = 12;

/// Columns by dependency depth, jobs in each column in name order.

pub fn layout(jobs: Vec<fdb::Job>) -> Graph {

    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();

    for j in jobs {

        by_depth.entry(j.depth).or_default().push(j);

    }

    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;

    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;

    let mut nodes = Vec::new();

    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {

        js.sort_by(|a, b| a.name.cmp(&b.name));

        let n = js.len() as i64;

        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;

        let top = (height - column_h) / 2;

        for (i, j) in js.into_iter().enumerate() {

            let duration = crate::flow::duration(&j);

            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });

        }

    }

    let mut edges = Vec::new();

    for n in &nodes {

        for d in n.j.depends_on() {

            if let Some(from) = nodes.iter().find(|m| m.j.name == d) {

                let (x1, y1) = (from.x + NODE_W, from.y + NODE_H / 2);

                let (x2, y2) = (n.x, n.y + NODE_H / 2);

                let mid = (x1 + x2) / 2;

                edges.push(Edge { d: format!("M{x1},{y1} C{mid},{y1} {mid},{y2} {x2},{y2}"), status: from.j.status.clone() });

            }

        }

    }

    let cols = nodes.iter().map(|n| n.x).max().map(|x| x + NODE_W + PAD).unwrap_or(PAD * 2);

    Graph { nodes, edges, width: cols, height: height.max(NODE_H + PAD * 2) }

}

pub fn glyph(status: &str) -> &'static str {

    match status {

        "passed" => "✓",

        "failed" => "✕",

        "running" => "●",

        "skipped" => "⤼",

        "error" => "!",

        _ => "○",

    }

}

pub struct JobLog {

    pub j: fdb::Job,

    pub artifacts: Vec<fdb::Artifact>,

    pub lines: Vec<fdb::LogLine>,

    pub duration: String,

    /// The last line shown; the live view asks for what follows.

    pub last_seq: i64,

}

#[derive(Template)]

#[template(path = "flow_run.html")]

pub struct RunPage {

    shell: Shell,

    repo: RepoShell,

    run: fdb::Run,

    graph: Graph,

    logs: Vec<JobLog>,

    duration: String,

    /// Whether the viewer may re-run it.

    can_rerun: bool,

}

/// Lines of a job's log a page shows before "the rest is loaded as it comes".

const PAGE_LINES: i64 = 5000;

pub async fn run(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id)): Path<(String, String, i64)>,

) -> Result<Page<RunPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{id}"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let run = fdb::run(&app.db, row.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let mut logs = Vec::new();

    for j in &jobs {

        let lines = fdb::log(&app.db, j.id, 0, PAGE_LINES).await.unwrap_or_default();

        let last_seq = lines.last().map(|l| l.seq).unwrap_or(0);

        let artifacts = fdb::artifacts(&app.db, j.id).await.unwrap_or_default();

        logs.push(JobLog { duration: crate::flow::duration(j), j: j.clone(), artifacts, lines, last_seq });

    }

    let duration = match run.finished_ms {

        Some(f) => crate::flow::human_ms(f - run.created_ms),

        None => crate::flow::human_ms(db::now_ms() - run.created_ms),

    };

    let graph = layout(jobs);

    let can_rerun = !run.live() && viewer.account().is_some() && role >= Role::Write && crate::flow::enabled(&app, &row);

    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun }))

}

#[derive(Deserialize)]

pub struct LiveQuery {

    /// `job:seq,job:seq`: the last log line each open job already shows.

    after: Option<String>,

}

/// `GET /{owner}/{name}/flow/{id}/live`: statuses and new log lines, for the

/// run page while it is live.

pub async fn live(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, id)): Path<(String, String, i64)>,

    Query(q): Query<LiveQuery>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, "", &viewer);

    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let run = fdb::run(&app.db, row.id, id)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let after: std::collections::HashMap<i64, i64> = q

        .after

        .unwrap_or_default()

        .split(',')

        .filter_map(|p| p.split_once(':'))

        .filter_map(|(a, b)| Some((a.parse().ok()?, b.parse().ok()?)))

        .collect();

    let mut out = Vec::new();

    for j in &jobs {

        let lines = match after.get(&j.id) {

            Some(seq) => fdb::log(&app.db, j.id, *seq, 1000).await.unwrap_or_default(),

            None => vec![],

        };

        out.push(serde_json::json!({

            "id": j.id,

            "name": j.name,

            "status": j.status,

            "glyph": glyph(&j.status),

            "duration": crate::flow::duration(j),

            "exit_code": j.exit_code,

            "lines": lines.iter().map(|l| serde_json::json!({ "seq": l.seq, "t": l.t, "stream": l.stream, "text": l.text })).collect::<Vec<_>>(),

        }));

    }

    Ok(Json(serde_json::json!({ "status": run.status, "live": run.live(), "jobs": out })).into_response())

}

// ------------------------------------------------------------ state stamps

/// In rendered Markdown, the image path that is the viewed state's stamp.

pub const STAMP_IMAGE: &str = ".flow/stamp";

/// In rendered Markdown, the link path to the viewed state's run.

pub const RUN_LINK: &str = ".flow/run";

/// Resolve a ref to its state address, and whether the ref can move.

async fn state_of(app: &Shared, owner: &str, name: &str, r: &str) -> Result<(String, bool), crate::repos::RepoError> {

    let repo = crate::repos::RepoName::new(owner, name).ok_or(crate::repos::RepoError::NotFound)?;

    let spec = r.to_string();

    app.repos

        .read(&repo, move |repo, cache| {

            let t = crate::repos::view::resolve(repo, cache, &spec)?;

            let moves = matches!(t.kind, crate::repos::view::RefKind::Bookmark | crate::repos::view::RefKind::Tag);

            Ok((t.permalink().unwrap_or_default(), moves))

        })

        .await

}

/// `GET /{owner}/{name}/flow/stamp/{ref}` (hub spec §13.2).

pub async fn stamp(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, r)): Path<(String, String, String)>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, "", &viewer);

    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let (heads, moves) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

    if moves && !heads.is_empty() {

        let mut res = axum::response::Redirect::to(&format!("/{owner}/{name}/flow/stamp/{heads}")).into_response();

        res.headers_mut().insert(axum::http::header::CACHE_CONTROL, axum::http::HeaderValue::from_static("no-cache"));

        return Ok(res);

    }

    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let jobs = match &run {

        Some(r) => fdb::jobs(&app.db, r.id).await.map_err(|e| Error::from_db(shell.clone(), e))?,

        None => vec![],

    };

    let svg = stamp_svg(&heads, run.as_ref(), &jobs);

    let cache = if run.as_ref().is_some_and(|r| !r.live()) { "public, max-age=3600" } else { "public, max-age=60" };

    Ok((

        [

            (axum::http::header::CONTENT_TYPE, "image/svg+xml; charset=utf-8"),

            (axum::http::header::CACHE_CONTROL, cache),

            (axum::http::header::CONTENT_SECURITY_POLICY, "default-src 'none'; style-src 'unsafe-inline'"),

        ],

        svg,

    )

        .into_response())

}

/// `GET /{owner}/{name}/flow/at/{ref}`: the run of the ref's state, or the Flow tab.

pub async fn at(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, r)): Path<(String, String, String)>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, "", &viewer);

    let (_repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let (heads, _) = state_of(&app, &owner, &name, &r).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

    let run = fdb::latest_for_heads(&app.db, row.id, &heads).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let to = match run {

        Some(r) => format!("/{owner}/{name}/flow/{}", r.number),

        None => format!("/{owner}/{name}/flow"),

    };

    Ok(axum::response::Redirect::to(&to).into_response())

}

fn xml(s: &str) -> String {

    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")

}

/// The stamp: the mark, one glyph per job, the status, the state and the

/// date; the same facts machine-readable in `<metadata>`. Fixed colours,

/// because an image does not follow the page's theme.

pub fn stamp_svg(heads: &str, run: Option<&fdb::Run>, jobs: &[fdb::Job]) -> String {

    const CH: f64 = 6.6; // IBM Plex Mono at 11px

    let colour = |s: &str| match s {

        "passed" => "#5fe3a1",

        "failed" | "error" => "#ff5e7a",

        "running" => "#00e0f0",

        _ => "#8592ad",

    };

    let short: String = heads.chars().take(12).collect();

    let (status, date) = match run {

        Some(r) => (r.status.as_str(), crate::render::time::absolute(r.created_ms as u64).chars().take(10).collect::<String>()),

        None => ("no run", String::new()),

    };

    let mut ordered: Vec<&fdb::Job> = jobs.iter().collect();

    ordered.sort_by(|a, b| (a.depth, &a.name).cmp(&(b.depth, &b.name)));

    let label = "OXYGENFLOW";

    let label_w = 14.0 + label.len() as f64 * 6.2 + 8.0;

    let glyphs_w = if ordered.is_empty() { 0.0 } else { ordered.len() as f64 * 10.0 + 6.0 };

    let tail = if date.is_empty() { format!("{status} · {short}") } else { format!("{status} · {short} · {date}") };

    let right_w = 8.0 + glyphs_w + tail.chars().count() as f64 * CH + 8.0;

    let w = (label_w + right_w).ceil();

    let mut glyphs = String::new();

    for (i, j) in ordered.iter().enumerate() {

        let x = label_w + 8.0 + i as f64 * 10.0 + 4.0;

        glyphs.push_str(&format!(

            r#"<text x="{x:.1}" y="15" fill="{}" text-anchor="middle">{}</text>"#,

            colour(&j.status),

            glyph(&j.status)

        ));

    }

    let meta = serde_json::json!({

        "heads": heads.split('+').filter(|h| !h.is_empty()).collect::<Vec<_>>(),

        "run": run.map(|r| r.number),

        "status": status,

        "jobs": ordered.iter().map(|j| serde_json::json!({ "name": j.name, "status": j.status, "receipt": j.receipt_hash })).collect::<Vec<_>>(),