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

    }

}

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

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

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

    };

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

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

    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)

}

/// Largest artifact the hub keeps; the runner refuses larger ones first.

pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;

/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact

/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's

/// repository. Answers `{"blob"}`.

pub async fn artifact(

    State(app): State<Shared>,

    headers: HeaderMap,

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

    body: axum::body::Bytes,

) -> Result<Response, ApiError> {

    runner_auth(&app, &headers).await?;

    let job = running_job(&app, id).await?;

    let spec: flow_core::Job =

        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;

    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));

    };