//! 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;

    }

    let touched = touched_since_last_run(app, row, bookmark, heads).await;

    if let Err(e) = start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).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}"))

}

/// "Run pipeline" on one committed change: the state whose only head is

/// `change`, whatever bookmarks hold it. Returns the new run's number.

pub async fn run_change(app: &Shared, row: &db::repos::RepoRow, change: &str, actor: &str) -> Result<i64, String> {

    if !enabled(app, row) {

        return Err("OxygenFlow is not enabled for this repository".into());

    }

    let h = ChangeHash::from_hex(change.trim().trim_start_matches('#')).ok_or("give the full hash of a change")?;

    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;

    let known = app.repos.read(&repo, move |r, _| Ok(r.store().get_change_meta(&h).is_ok())).await.map_err(|e| e.to_string())?;

    if !known {

        return Err(format!("no change {} in this repository", &change.trim()[..12.min(change.trim().len())]));

    }

    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(vec![h]).encode());

    start_run(app, row, "", &state.to_hex(), &[h], actor, "manual")

        .await?

        .ok_or_else(|| format!("that change's state 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}"))

}

/// The paths the changes a push brought touch: the changes in `heads`'

/// state that the bookmark's previous run did not have, and the files their

/// edits and tree operations name, as the new state places them. `None`

/// (run every job) for the bookmark's first run, or when it cannot be told.

async fn touched_since_last_run(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Option<Vec<String>> {

    let last = fdb::latest_on(&app.db, row.id, bookmark).await.ok()??;

    let old: Vec<ChangeHash> = last.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();

    if old.is_empty() {

        return None;

    }

    let repo = RepoName::new(&row.owner, &row.name)?;

    let new = heads.to_vec();

    app.repos

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

            let fresh = r.negotiate(&old, &new)?;

            let loaded = cache.load(r, &new)?;

            let by_node: std::collections::HashMap<oxy_core::ids::NodeId, &String> =

                loaded.snapshot.files.iter().map(|(p, f)| (f.node, p)).collect();

            let mut paths = std::collections::BTreeSet::new();

            for h in fresh {

                let (c, _) = r.store().get_change_lazy(&h)?;

                let nodes = c.edits.iter().map(|e| e.node()).chain(c.tree_ops.iter().map(|o| o.node()));

                for n in nodes {

                    if let Some(p) = by_node.get(&n) {

                        paths.insert((*p).clone());

                    }

                }

            }

            Ok(paths.into_iter().collect::<Vec<String>>())

        })

        .await

        .ok()

}

/// Whether a job runs for a push that touched `touched`: always without a

/// scope or without knowing what was touched, else when a path matches.

fn in_scope(scope: &[String], touched: Option<&[String]>) -> bool {

    let Some(touched) = touched else { return true };

    if scope.is_empty() {

        return true;

    }

    let mut b = globset::GlobSetBuilder::new();

    for g in scope {

        match globset::Glob::new(g) {

            Ok(g) => {

                b.add(g);

            }

            Err(_) => return true,

        }

    }

    let Ok(set) = b.build() else { return true };

    touched.iter().any(|p| set.is_match(p))

}

/// [`start_run`] for a push: jobs whose `scope` the push did not touch are

/// left out of the run, unless a job that runs depends on them.

#[allow(clippy::too_many_arguments)]

pub async fn start_run_scoped(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

    touched: Option<&[String]>,

) -> Result<Option<i64>, String> {

    start_run_with(app, row, bookmark, state, heads, actor, trigger, touched).await

}

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

    start_run_with(app, row, bookmark, state, heads, actor, trigger, None).await

}

#[allow(clippy::too_many_arguments)]

async fn start_run_with(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

    touched: Option<&[String]>,

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

            // Jobs out of the push's scope stay out, unless a kept job

            // needs them.

            let mut keep: std::collections::HashSet<String> =

                p.order().into_iter().filter(|j| in_scope(&j.scope, touched)).map(|j| j.name.clone()).collect();

            loop {

                let needed: Vec<String> = p

                    .order()

                    .into_iter()

                    .filter(|j| keep.contains(&j.name))

                    .flat_map(|j| j.depends_on.iter().cloned())

                    .filter(|d| !keep.contains(d))

                    .collect();

                if needed.is_empty() {

                    break;

                }

                keep.extend(needed);

            }

            let jobs: Vec<fdb::NewJob> = p

                .order()

                .into_iter()

                .filter(|j| keep.contains(&j.name))

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

            if jobs.is_empty() && touched.is_some() {

                // Nothing this push touched is any job's business.

                return Ok(None);

            }

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

    }

}

/// The runner a bearer token belongs to: a registered runner, or one

/// configured with a hub.toml token (tracked from its first request). A

/// disabled runner is refused.

async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {

    let Some(t) = oxy_server::bearer_token(headers) else {

        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));

    };

    let hash = blake3::hash(t.as_bytes()).to_hex().to_string();

    let runner = match db::runners::by_token(&app.db, &hash).await? {

        Some(r) => r,

        None if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) => {

            db::runners::configured(&app.db, &hash, "runner").await?

        }

        None => return Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into())),

    };

    if runner.state == "disabled" {

        return Err(ApiError(StatusCode::FORBIDDEN, format!("runner {} is disabled on this hub", runner.name)));

    }

    Ok(runner)

}

/// `POST /api/v1/flow/runners/register`: a one-time registration token

/// from Admin → Runners buys a runner credential, binding the runner's

/// signing identity (hub spec §13.2).

pub async fn register(

    State(app): State<Shared>,

    Json(req): Json<flow_core::api::Register>,

) -> Result<Json<flow_core::api::Registered>, ApiError> {

    let bytes = hex::decode(req.identity.trim()).map_err(|_| ApiError(StatusCode::BAD_REQUEST, "identity: not hex".into()))?;

    let identity = oxy_core::identity::Identity::decode(&bytes).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;

    let name: String = req.name.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).take(64).collect();

    if name.is_empty() {

        return Err(ApiError(StatusCode::BAD_REQUEST, "give the runner a name of letters, digits, - _ or .".into()));

    }

    let credential = format!("flowrunner_{}", crate::auth::random_hex(32));

    let hash = blake3::hash(credential.as_bytes()).to_hex().to_string();

    let reg = blake3::hash(req.token.trim().as_bytes()).to_hex().to_string();

    let runner =

        db::runners::register(&app.db, &reg, &name, &hash, &identity.hash().to_hex(), &identity.name, "").await?.ok_or_else(|| {

            ApiError(StatusCode::FORBIDDEN, "that registration token is unknown, used or expired; create one on Admin → Runners".into())

        })?;

    db::runners::seen(&app.db, runner.id, None, Some(&req.info)).await?;

    tracing::info!("flow: runner {} registered (key {})", runner.name, &runner.key[..12]);

    Ok(Json(flow_core::api::Registered { id: runner.id, name: runner.name, credential }))

}

/// `POST /api/v1/flow/claim`.

pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {

    let me = runner_auth(&app, &headers).await?;

    let claimed: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();

    db::runners::seen(&app.db, me.id, Some(&claimed), req.info.as_ref()).await?;

    if let Some(holding) = &req.holding {

        end_jobs_not_held(&app, &me, holding).await?;

    }

    // A draining runner finishes what it has and takes nothing new.

    if me.state == "draining" {

        return Ok(StatusCode::NO_CONTENT.into_response());

    }

    let runner = if me.source == "registered" { me.name.clone() } else { claimed };

    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,

        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),

    };

    // The labels it reports now, or those it registered with, and those

    // an administrator gave it (#98).

    let mut labels: Vec<String> = match &req.info {

        Some(i) => i.labels.clone(),

        None => me.reported_label_list().into_iter().map(String::from).collect(),

    };

    for l in me.admin_label_list() {

        if !labels.iter().any(|x| x == l) {

            labels.push(l.to_string());

        }

    }

    let Some(job) =

        fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels).await?

    else {

        return Ok(StatusCode::NO_CONTENT.into_response());

    };

    db::runners::set_job_runner(&app.db, job.id, me.id).await?;

    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}"));

    // The job's limit: its own timeout, else its sandbox's, else the site

    // default, never past the site maximum; recorded for the reaper.

    let limits = limits(&app).await;

    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;

    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;

    let mut timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);

    // ...and never past the runner's own limit, when it has one (#98).

    if let Some(m) = me.timeout_minutes.filter(|m| *m > 0) {

        timeout_s = timeout_s.min(m as u64 * 60).max(60);

    }

    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;

    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: spec,

        sandbox,

        timeout_s,

        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,