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

/// 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() || !app.config.flow.allows(&row.owner, &row.name) {

        return;

    }

    let Some(repo) = RepoName::new(&row.owner, &row.name) else { return };

    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;

    let text = match text {

        Ok(Some(t)) => t,

        Ok(None) => return,

        Err(e) => {

            tracing::warn!("flow: reading the pipeline of {repo}: {e}");

            return;

        }

    };

    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, error: None };

    let result = 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

        }

        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;

            if let Ok(id) = id {

                for j in &jobs {

                    report_check(app, row, state, id, &j.name, "queued", "waiting for a runner").await;

                }

            }

            id

        }

    };

    if let Err(e) = result {

        tracing::warn!("flow: recording a run of {repo}: {e}");

    }

}

async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_id: i64, job: &str, status: &str, summary: &str) {

    let url = format!("{}/{}/{}/flow/{run_id}", 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 Some(job) = fdb::claim(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret)).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.id, &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,

    };

    Ok(Json(claim).into_response())

}

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.id, &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)

    }

}