//! OxygenFlow on the hub (hub spec §13.2): runs started by pushes, the
//! runner API, job tokens and the checks that mirror jobs. Pipelines are
//! parsed by `flow-core`, the same code `flow check` runs.

use crate::api::ApiError;
use crate::db::{self, flow as fdb};
use crate::repos::RepoName;
use crate::Shared;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, MAX_RECORDS, MAX_RECORD_TEXT};
use flow_core::{Pipeline, PIPELINE_PATH};
use oxy_core::hash::ChangeHash;

/// Prefix of job tokens, so the wire protocol can tell them from personal tokens.
pub const JOB_TOKEN_PREFIX: &str = "flowjob_";
/// A job token is refused this long after its job started, whatever its state.
pub const JOB_TOKEN_MAX_AGE_MS: i64 = 2 * 3600 * 1000;

/// Whether pushes to this repository start runs: its own setting when a
/// site administrator made one, else hub.toml's `[flow] repos`.
pub fn enabled(app: &Shared, row: &db::repos::RepoRow) -> bool {
    match row.flow.as_str() {
        "on" => true,
        "off" => false,
        _ => app.config.flow.allows(&row.owner, &row.name),
    }
}

/// A bookmark moved by a push: start a run when the repository is allowed
/// and the new state has a pipeline.
pub async fn on_push(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, state: &str, heads: &[ChangeHash], actor: &str) {
    if bookmark.starts_with("tags/") || heads.is_empty() || !enabled(app, row) {
        return;
    }
    if let Err(e) = start_run(app, row, bookmark, state, heads, actor, "push").await {
        tracing::warn!("flow: starting a run of {}/{}: {e}", row.owner, row.name);
    }
}

/// "Run pipeline" (hub spec §13.2): run a bookmark's current state without
/// moving it. Returns the new run's number.
pub async fn run_bookmark(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let b = bookmark.to_string();
    let found = app
        .repos
        .read(&repo, move |r, _| {
            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
            Ok(Some((st, r.store().get_state(&st)?.heads)))
        })
        .await
        .map_err(|e| e.to_string())?;
    let Some((state, heads)) = found else { return Err(format!("no bookmark {bookmark}")) };
    if heads.is_empty() {
        return Err(format!("{bookmark} has no changes yet"));
    }
    start_run(app, row, bookmark, &state.to_hex(), &heads, actor, "manual")
        .await?
        .ok_or_else(|| format!("{bookmark} has no {PIPELINE_PATH}"))
}

/// Re-run a finished run: the same state, bookmark and pipeline, as a new
/// run whose checks replace the old ones. Returns the new run's number.
pub async fn rerun(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    if run.live() {
        return Err(format!("run #{} is still {}", run.number, run.status));
    }
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    if heads.is_empty() {
        return Err("the run names no changes".into());
    }
    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &format!("rerun:{}", run.number))
        .await?
        .ok_or_else(|| format!("that state has no {PIPELINE_PATH}"))
}

/// Start a run of the pipeline in the state `heads` make (whose state
/// hash is `state`), recorded on `bookmark`: for a push, "Run pipeline"
/// (`trigger` = `manual`) and re-runs (`rerun:N`). `actor` is the account
/// whose access the run's subrepo reads are checked against. Returns the
/// run's number, or `None` when the state has no pipeline. The caller
/// checks that Flow is enabled.
pub async fn start_run(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
) -> Result<Option<i64>, String> {
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let h = heads.to_vec();
    let text = app
        .repos
        .read(&repo, move |r, cache| {
            let loaded = cache.load(r, &h)?;
            Ok(match loaded.snapshot.files.get(PIPELINE_PATH).map(|f| &f.content) {
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
                _ => None,
            })
        })
        .await
        .map_err(|e| format!("reading the pipeline: {e}"))?;
    let Some(text) = text else { return Ok(None) };
    let mut hex: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
    hex.sort();
    let heads_s = hex.join("+");
    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, trigger, error: None };
    let id = match Pipeline::parse(&text) {
        Err(problems) => {
            let msg = problems.to_string();
            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await.map_err(|e| e.to_string())?
        }
        Ok(p) => {
            let depths = p.depths();
            let jobs: Vec<fdb::NewJob> = p
                .order()
                .into_iter()
                .map(|j| {
                    let sb = &p.sandboxes[&j.sandbox];
                    fdb::NewJob {
                        name: j.name.clone(),
                        depth: depths[j.name.as_str()] as i64,
                        depends: j.depends_on.clone(),
                        spec: serde_json::to_string(j).unwrap_or_default(),
                        sandbox: serde_json::to_string(sb).unwrap_or_default(),
                        runtime: sb.runtime.as_str().into(),
                    }
                })
                .collect();
            let hash = hex::encode(p.hash());
            let id =
                db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await.map_err(|e| e.to_string())?;
            if let Ok(Some(run)) = fdb::run_by_id(&app.db, id).await {
                for j in &jobs {
                    report_check(app, row, state, run.number, &j.name, "queued", "waiting for a runner").await;
                }
            }
            id
        }
    };
    let run = fdb::run_by_id(&app.db, id).await.map_err(|e| e.to_string())?.ok_or("the run vanished")?;
    Ok(Some(run.number))
}

/// Record a job's check; it links to the run's page by the run's number.
async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, job: &str, status: &str, summary: &str) {
    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);
    let name = format!("flow/{job}");
    let c = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state, name: &name, status, summary, url: &url, reporter: None, seal: "" },
    )
    .await;
    if let Ok(c) = c {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
}

/// The check status a job's status shows as.
fn check_status(job_status: &str) -> &'static str {
    match job_status {
        "queued" => "queued",
        "running" => "running",
        "passed" => "passed",
        "skipped" => "skipped",
        _ => "failed",
    }
}

fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {
    let Some(t) = oxy_server::bearer_token(headers) else {
        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));
    };
    if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) {
        Ok(())
    } else {
        Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into()))
    }
}

/// `POST /api/v1/flow/claim`.
pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {
    runner_auth(&app, &headers)?;
    let runner: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();
    let secret = format!("{JOB_TOKEN_PREFIX}{}", crate::auth::random_hex(32));
    let room = fdb::Room { memory_mib: req.free_memory_mib, cpus: req.free_cpus };
    let Some(job) = fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room).await? else {
        return Ok(StatusCode::NO_CONTENT.into_response());
    };
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    report_check(&app, &row, &run.state, run.number, &job.name, "running", &format!("running on {runner}")).await;
    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));
    let claim = Claim {
        job_id: job.id,
        run_id: run.id,
        repo: format!("{}/{}", row.owner, row.name),
        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),
        token: secret,
        heads: run.head_list().into_iter().map(String::from).collect(),
        pipeline_hash: run.pipeline_hash.clone(),
        job: serde_json::from_str(&job.spec).map_err(bad)?,
        sandbox: serde_json::from_str(&job.sandbox).map_err(bad)?,
        timeout_s: app.config.flow.job_timeout_minutes.max(1) * 60,
        subrepos: vec![],
        state: run.state.clone(),
        bookmark: run.bookmark.clone(),
        run_number: run.number,
    };
    let mut claim = claim;
    let (inputs, ids) = subrepo_inputs(&app, &row, &run).await;
    if !ids.is_empty() {
        fdb::set_also_read(&app.db, job.id, &ids).await?;
    }
    claim.subrepos = inputs;
    Ok(Json(claim).into_response())
}

/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
async fn pins_in(
    app: &Shared,
    name: crate::repos::RepoName,
    heads: Vec<oxy_core::hash::ChangeHash>,
) -> Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)> {
    app.repos
        .read(&name, move |repo, cache| {
            let loaded = cache.load(repo, &heads)?;
            let mut out = Vec::new();
            for (path, f) in &loaded.snapshot.files {
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
                    continue;
                }
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
                    if let Some(b) = blobs.first() {
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
                            .read_blob(b)
                            .map_err(crate::repos::RepoError::Oxy)
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
                        {
                            out.push((path.clone(), id, heads));
                        }
                    }
                }
            }
            Ok(out)
        })
        .await
        .unwrap_or_default()
}

/// How deep subrepos of subrepos are followed.
const MAX_SUBREPO_DEPTH: usize = 3;

/// The Oxygen subrepos the run's state pins, and those their states pin in
/// turn (parents first, paths joined), that live on this hub and that the
/// run's pusher may read: what the job's workspace needs beside its own
/// tree. A pin the hub cannot serve is left out, with what it pins; the job
/// then fails on the missing files with an ordinary error.
async fn subrepo_inputs(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>) {
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![]) };
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
    let mut inputs = Vec::new();
    let mut ids = Vec::new();
    // (path prefix, repository, heads, depth), parents before children.
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
        if depth == MAX_SUBREPO_DEPTH {
            continue;
        }
        for (path, id, pinned) in pins_in(app, name, heads).await {
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
            for other in &rows {
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
                if !matches {
                    continue;
                }
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
                    inputs.push(flow_core::api::SubrepoInput {
                        path: path.clone(),
                        repo: format!("{}/{}", other.owner, other.name),
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
                    });
                    if !ids.contains(&other.id) {
                        ids.push(other.id);
                    }
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
                }
                break;
            }
        }
    }
    (inputs, ids)
}

async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
    match fdb::job(&app.db, id).await? {
        Some(j) if j.status == "running" => Ok(j),
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
    }
}

/// `POST /api/v1/flow/jobs/{id}/log`.
pub async fn log(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(mut batch): Json<LogBatch>,
) -> Result<StatusCode, ApiError> {
    runner_auth(&app, &headers)?;
    let job = running_job(&app, id).await?;
    for r in &mut batch.records {
        if r.text.len() > MAX_RECORD_TEXT {
            let mut cut = MAX_RECORD_TEXT;
            while !r.text.is_char_boundary(cut) {
                cut -= 1;
            }
            r.text.truncate(cut);
        }
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
            r.stream = "console".into();
        }
    }
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
        tracing::info!("flow: job {} reached the log limit", job.id);
    }
    Ok(StatusCode::NO_CONTENT)
}

/// `POST /api/v1/flow/jobs/{id}/finish`.
pub async fn finish(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(f): Json<Finish>,
) -> Result<StatusCode, ApiError> {
    runner_auth(&app, &headers)?;
    let job = running_job(&app, id).await?;
    let status = match (f.exit_code, &f.error) {
        (Some(0), None) => "passed",
        (Some(_), _) => "failed",
        (None, _) => "error",
    };
    let (receipt, receipt_hash) = match &f.receipt {
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
        }
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
        None => (String::new(), String::new()),
    };
    let error = f.error.clone().unwrap_or_default();
    let changed = fdb::finish(
        &app.db,
        job.id,
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
    )
    .await?;
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
            for j in changed {
                let summary = match j.status.as_str() {
                    "passed" => format!("passed in {}", duration(&j)),
                    "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
                    "skipped" => "skipped: a dependency did not pass".into(),
                    _ => format!("could not run: {}", j.error),
                };
                report_check(&app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
            }
        }
    }
    Ok(StatusCode::NO_CONTENT)
}

/// A job's wall time, as the pages show it.
pub fn duration(j: &fdb::Job) -> String {
    match (j.started_ms, j.finished_ms) {
        (Some(a), Some(b)) => human_ms(b - a),
        (Some(a), None) => human_ms(db::now_ms() - a),
        _ => "–".into(),
    }
}

pub fn human_ms(ms: i64) -> String {
    let ms = ms.max(0);
    if ms < 1000 {
        format!("{ms} ms")
    } else if ms < 60_000 {
        format!("{:.1} s", ms as f64 / 1000.0)
    } else {
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
    }
}