//! 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)>,

    /// "3rd in line" for a run still waiting for a runner.

    pub place: Option<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>,

    /// Why this viewer is not offered "Run pipeline", when they are not.

    why_not: Option<String>,

}

#[derive(Deserialize)]

pub struct RunsQuery {

    before: Option<i64>,

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

    error: Option<String>,

}

/// Why `viewer` may not start runs here, or `None` when they may.

pub fn why_not_run(enabled: bool, signed_in: bool, role: Role) -> Option<String> {

    if !enabled {

        Some("Flow is off for this repository; a site administrator can turn it on in Settings › Pipelines.".into())

    } else if !signed_in {

        Some("Sign in with write access to run the pipeline.".into())

    } else if role < Role::Write {

        Some("Running the pipeline needs write access to this repository.".into())

    } else {

        None

    }

}

const PER_PAGE: i64 = 30;

#[derive(Template)]

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

pub struct DeploymentsPage {

    shell: Shell,

    repo: RepoShell,

    /// Each environment's newest deployment, by name.

    current: Vec<fdb::Deployment>,

    /// Every recent deployment, newest first.

    history: Vec<fdb::Deployment>,

}

/// `GET /{owner}/{name}/deployments`: what each environment runs

/// (oxygenflow#59).

pub async fn deployments(

    State(app): State<Shared>,

    viewer: Viewer,

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

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

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

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

    let history = fdb::deployments(&app.db, row.id, 50).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let mut current: Vec<fdb::Deployment> = Vec::new();

    for d in &history {

        if !current.iter().any(|c| c.environment == d.environment) {

            current.push(d.clone());

        }

    }

    current.sort_by(|a, b| a.environment.cmp(&b.environment));

    Ok(Page(DeploymentsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), current, history }))

}

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 queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };

    let mut runs = Vec::new();

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

        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };

        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, place });

    }

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

    let why_not = why_not_run(enabled, viewer.account().is_some(), role);

    let can_run = why_not.is_none();

    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,

        why_not,

    }))

}

/// 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,

    #[serde(default)]

    bookmark: String,

    /// A change's full hash: run the state whose only head it is.

    #[serde(default)]

    change: String,

}

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

/// current state, or on one committed change.

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)?;

    let (started, what) = if f.change.trim().is_empty() {

        (crate::flow::run_bookmark(&app, &row, f.bookmark.trim(), &me.account.username, &[]).await, f.bookmark.trim().to_string())

    } else {

        (crate::flow::run_change(&app, &row, &f.change, &me.account.username, &[]).await, format!("change {}", f.change.trim()))

    };

    Ok(match started {

        Ok(n) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.run", &format!("{owner}/{name}"), &format!("#{n} {what}")).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(),

    })

}

/// `POST /{owner}/{name}/flow/{number}/jobs/{job}/rerun` (oxygenflow#31).

pub async fn rerun_job(

    State(app): State<Shared>,

    viewer: Viewer,

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

    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_job(&app, &row, &run, &job, &me.account.username).await {

        Ok(_) => {

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

                .await

                .ok();

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

        }

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

    })

}

/// `POST /{owner}/{name}/flow/{number}/cancel` (oxygenflow#32).

pub async fn cancel(

    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::cancel(&app, &row, &run, &me.account.username).await {

        Ok(()) => {

            db::accounts::audit(&app.db, Some(&me.account), "flow.cancel", &format!("{owner}/{name}"), &format!("#{number}")).await.ok();

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

        }

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

    })

}

#[derive(Deserialize)]

pub struct ExpediteForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    #[serde(default)]

    reason: String,

}

/// `POST /{owner}/{repo}/flow/{n}/expedite` (oxygenhub#244): a maintainer

/// raises a running or queued run to urgent, saying why (an incident's

/// fix, say). Audited and shown on the run; it lapses when the run ends.

pub async fn expedite(

    State(app): State<Shared>,

    viewer: Viewer,

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

    Form(f): Form<ExpediteForm>,

) -> 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::Maintain).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()))?;

    let reason: String = f.reason.chars().filter(|c| !c.is_control()).take(200).collect();

    let reason = reason.trim();

    if reason.is_empty() {

        return Ok(Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding("Say why the run is urgent."))).into_response());

    }

    // A cap on one owner's urgent runs at once (Admin > Runners >

    // Settings), so urgent can't starve everyone else.

    let cap: i64 = db::accounts::setting(&app.db, "flow_expedite_cap").await.ok().flatten().and_then(|v| v.parse().ok()).unwrap_or(2);

    let live = fdb::expedited_live(&app.db, &owner).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    if live >= cap {

        let msg = format!("{owner} already has {live} expedited run(s) going, the most at once ({cap}); wait for one to end.");

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

    }

    let note = format!("expedited by {}: {reason}", me.account.username);

    if fdb::expedite(&app.db, run.id, &note).await.map_err(|e| Error::from_db(shell.clone(), e))? {

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

            .await

            .ok();

        // Every queued job's place changes now, not at the next pass

        // (#291): this run's page and the pages of the runs it moved down

        // hear of it.

        if let Err(e) = crate::flow::explain_waits(&app).await {

            tracing::warn!("flow: placing after an expedite: {e}");

        }

        app.bus.live(&crate::events::topic::run(run.id), "jobs");

    }

    Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).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" => "⤼",

        "cancelled" => "⊘",

        "error" => "!",

        _ => "○",

    }

}

/// Failed tests a run page lists per job; the rest are counted.

const SHOWN_FAILURES: i64 = 50;

/// A stretch of a job's log as the run page shows it: a line, or a

/// section the job marked with `::group::TITLE` … `::endgroup::`

/// (oxygenflow#31), shown folded with how long it took.

pub enum Segment {

    Line(fdb::LogLine),

    Group(Section),

}

pub struct Section {

    pub title: String,

    /// When it began, and ended (`None` while it is still going).

    pub from_t: i64,

    pub to_t: Option<i64>,

    pub lines: Vec<fdb::LogLine>,

}

/// A log's lines in sections. A section left open is still going; a new

/// `::group::` ends the one before (sections do not nest).

pub fn sections(lines: &[fdb::LogLine]) -> Vec<Segment> {

    let mut out: Vec<Segment> = Vec::new();

    let mut open: Option<Section> = None;

    for l in lines {

        if let Some(title) = l.text.strip_prefix("::group::") {

            if let Some(g) = open.take() {

                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));

            }

            open = Some(Section { title: title.trim().to_string(), from_t: l.t, to_t: None, lines: Vec::new() });

        } else if l.text.trim() == "::endgroup::" {

            if let Some(g) = open.take() {

                out.push(Segment::Group(Section { to_t: Some(l.t), ..g }));

            }

        } else {

            match open.as_mut() {

                Some(g) => g.lines.push(l.clone()),

                None => out.push(Segment::Line(l.clone())),

            }

        }

    }

    if let Some(g) = open {

        out.push(Segment::Group(g));

    }

    out

}

/// Milliseconds as the log shows them: `12.3s`.

pub fn secs(ms: impl std::borrow::Borrow<i64>) -> String {

    let ms = *ms.borrow();

    format!("{}.{}s", ms / 1000, ms % 1000 / 100)

}

pub struct JobLog {

    pub j: fdb::Job,

    pub artifacts: Vec<fdb::Artifact>,

    pub lines: Vec<fdb::LogLine>,

    /// The lines in sections (oxygenflow#31).

    pub segments: Vec<Segment>,

    pub duration: String,

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

    pub last_seq: i64,

    /// The job's coverage report, when it uploaded one (#113): its id and percentage.

    pub coverage: Option<(i64, String)>,

    /// The job's test report, when it uploaded one (#114), and its first

    /// failures.

    pub tests: Option<crate::junit::Report>,