//! Runner and job API under `/api/v1/ci` (see docs/ci.md, Runner protocol).
//!
//! Runners authenticate with `Authorization: Bearer <runner secret>`. A job's routes take
//! `Authorization: Bearer <job token>`, which `job_by_token` accepts only while the job is Running.
//! The job token also works as a read-only git credential, see `git_http::job_token_read`.
//!
//! Every response is JSON except the empty 204 of an empty claim and the body of a log append.

use forge_ci::config::JobSpec;
use forge_ci::mask::Masker;
use forge_ci::proto::{
    FinishRequest, HeartbeatResponse, JobAssignment, Outcome, RegisterRequest, RegisterResponse,
};
use forge_core::Error as CoreError;
use forge_core::ci::{CiJob, CiRun, CiRunner, JobState};
use futures_util::StreamExt;
use http::header::HeaderValue;
use serde_json::json;
use std::collections::BTreeMap;
use topcoat::context::{Cx, app_context};
use topcoat::router::request::headers;
use topcoat::router::response::Response;
use topcoat::router::{Body, HeaderMap, StatusCode, path_param, route};

use crate::api::{self, Refusal};
use crate::ci::{artifact_key, log_path};
use crate::pages::repo::https_clone_url;
use crate::self_update;
use crate::state::AppState;

path_param!(job_id);
path_param!(artifact_name);

/// Header carrying the byte offset of a log chunk.
const LOG_OFFSET_HEADER: &str = "x-forge-log-offset";
/// Header on a 409 log response carrying the size the server has.
const LOG_SIZE_HEADER: &str = "x-forge-log-size";
/// Largest log chunk, in bytes.
const MAX_LOG_CHUNK: usize = 1024 * 1024;
/// Largest JSON body on the runner routes, in bytes.
const MAX_JSON_BODY: usize = 64 * 1024;
/// How long a claim waits for work before answering 204.
#[cfg(not(test))]
const CLAIM_WAIT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(test)]
const CLAIM_WAIT: std::time::Duration = std::time::Duration::from_millis(50);
/// Longest runner name or label, in bytes.
const MAX_NAME_BYTES: usize = 64;
/// Most labels a runner may register.
const MAX_LABELS: usize = 32;
/// Text appended once when the log cap is reached.
const TRUNCATED_MARKER: &[u8] = b"\n[log truncated]\n";
const MIB: u64 = 1024 * 1024;

/// Bearer token from the `Authorization` header, or `None`.
fn bearer_token(headers: &HeaderMap) -> Option<&str> {
    let value = headers.get(http::header::AUTHORIZATION)?.to_str().ok()?;
    let (scheme, rest) = value.split_once(' ')?;
    if !scheme.eq_ignore_ascii_case("bearer") {
        return None;
    }
    let token = rest.trim();
    (!token.is_empty()).then_some(token)
}

fn unauthorized() -> Refusal {
    Box::new(api::error(
        StatusCode::UNAUTHORIZED,
        "authentication required",
    ))
}

/// The runner behind the Bearer secret. Disabled runners get 403.
async fn runner_from(state: &AppState, headers: &HeaderMap) -> Result<CiRunner, Refusal> {
    let Some(secret) = bearer_token(headers) else {
        return Err(unauthorized());
    };
    match state.store.runner_by_secret(secret).await {
        Ok(Some(runner)) if runner.disabled => Err(Box::new(api::error(
            StatusCode::FORBIDDEN,
            "the runner is disabled",
        ))),
        Ok(Some(runner)) => Ok(runner),
        Ok(None) => Err(unauthorized()),
        Err(e) => Err(api::internal(e)),
    }
}

/// The job behind the Bearer job token. Only Running jobs match, so a finished job gets 401.
async fn job_from(state: &AppState, headers: &HeaderMap, job_id: &str) -> Result<CiJob, Refusal> {
    let Some(token) = bearer_token(headers) else {
        return Err(unauthorized());
    };
    match state.store.job_by_token(job_id, token).await {
        Ok(Some(job)) => Ok(job),
        Ok(None) => Err(unauthorized()),
        Err(e) => Err(api::internal(e)),
    }
}

/// Parses a JSON body of at most [`MAX_JSON_BODY`] bytes.
async fn json_body<T: serde::de::DeserializeOwned>(body: Body) -> Result<T, Refusal> {
    let bytes = api::read_body(body, MAX_JSON_BODY).await?;
    serde_json::from_slice(&bytes).map_err(|e| {
        Box::new(api::error(
            StatusCode::BAD_REQUEST,
            &format!("invalid JSON body: {e}"),
        ))
    })
}

fn valid_name(value: &str) -> bool {
    !value.trim().is_empty() && value.len() <= MAX_NAME_BYTES && !value.contains(char::is_control)
}

/// `[A-Za-z0-9._+-]{1,128}`, and not starting with a dot.
pub fn valid_artifact_name(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= 128
        && !name.starts_with('.')
        && name
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'+' | b'-'))
}

// ---------------------------------------------------------------------------------------------
// Registration and claims
// ---------------------------------------------------------------------------------------------

#[route(POST "/api/v1/ci/runners/register")]
async fn register(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let request: RegisterRequest = match json_body(body).await {
        Ok(request) => request,
        Err(res) => return Ok(*res),
    };
    if !valid_name(&request.name) || request.labels.len() > MAX_LABELS {
        return Ok(api::error(
            StatusCode::UNPROCESSABLE_ENTITY,
            "the runner name or labels are invalid",
        ));
    }
    if request.labels.iter().any(|label| !valid_name(label)) {
        return Ok(api::error(
            StatusCode::UNPROCESSABLE_ENTITY,
            "the runner name or labels are invalid",
        ));
    }
    match state
        .store
        .register_runner(&request.token, request.name.trim(), &request.labels)
        .await
    {
        Ok((runner, secret)) => {
            let body = RegisterResponse {
                runner_id: runner.id,
                runner_secret: secret,
            };
            Ok(api::json(
                StatusCode::CREATED,
                &serde_json::to_value(&body).unwrap_or_default(),
            ))
        }
        Err(CoreError::Invalid(_)) => Ok(api::error(
            StatusCode::UNAUTHORIZED,
            "the registration token is unknown, used or expired",
        )),
        Err(e) => Ok(*api::internal(e)),
    }
}

#[route(POST "/api/v1/ci/runners/claim")]
async fn claim(cx: &Cx) -> topcoat::Result<Response> {
    let mut res = claim_work(cx).await?;
    // Advertises the binary sha256 when self-update is on and the runner asked for this target.
    self_update::advertise(app_context::<AppState>(cx), headers(cx), &mut res);
    Ok(res)
}

async fn claim_work(cx: &Cx) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    // The labels in the request are ignored: the labels stored at registration are used.
    let runner = match runner_from(state, headers(cx)).await {
        Ok(runner) => runner,
        Err(res) => return Ok(*res),
    };

    // Arm the wake-up before each attempt, so a job queued during the attempt is not missed.
    // Producers use notify_waiters, which wakes every claimer, so a runner whose labels
    // don't match a job re-arms and keeps waiting for the rest of CLAIM_WAIT.
    let waker = state.ci_waker.clone();
    let deadline = tokio::time::Instant::now() + CLAIM_WAIT;
    loop {
        let mut woken = std::pin::pin!(waker.notified());
        woken.as_mut().enable();

        match try_claim(state, &runner).await {
            Ok(Some(assignment)) => return Ok(assignment_response(&assignment)),
            Ok(None) => {}
            Err(res) => return Ok(*res),
        }
        if tokio::time::timeout_at(deadline, woken.as_mut())
            .await
            .is_err()
        {
            return Ok(empty(StatusCode::NO_CONTENT));
        }
    }
}

/// The running binary, for a runner that updates itself. 404 unless self-update is enabled.
#[route(GET "/api/v1/ci/runners/binary")]
async fn runner_binary(cx: &Cx) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    if let Err(res) = runner_from(state, headers(cx)).await {
        return Ok(*res);
    }
    if !state.config.self_update.enabled {
        return Ok(api::error(StatusCode::NOT_FOUND, "self-update is disabled"));
    }
    match self_update::binary_response().await {
        Ok(res) => Ok(res),
        Err(e) => {
            tracing::error!(error = %e, "ci: serving the runner binary failed");
            Ok(api::error(
                StatusCode::INTERNAL_SERVER_ERROR,
                "the binary is unavailable",
            ))
        }
    }
}

fn assignment_response(assignment: &JobAssignment) -> Response {
    api::json(
        StatusCode::OK,
        &serde_json::to_value(assignment).unwrap_or_default(),
    )
}

fn empty(status: StatusCode) -> Response {
    let mut res = Response::new(topcoat::router::Body::empty());
    *res.status_mut() = status;
    res
}

/// One claim attempt. Builds the assignment, including the secrets the job may see.
async fn try_claim(state: &AppState, runner: &CiRunner) -> Result<Option<JobAssignment>, Refusal> {
    let Some((job, run, job_token)) = state.store.claim_job(runner).await.map_err(api::internal)?
    else {
        return Ok(None);
    };
    state.ci_live.bump_job(&job.id);
    state.ci_live.bump_run(&run.id);
    let spec: JobSpec = serde_json::from_str(&job.spec_json).map_err(api::internal)?;
    let repo = state
        .store
        .repo_by_id(&run.repo_id)
        .await
        .map_err(api::internal)?
        .ok_or_else(|| api::internal("claimed job belongs to a missing repo"))?;

    let secrets = if is_fork_run(state, &run).await? {
        BTreeMap::new()
    } else {
        state
            .store
            .ci_secret_values(&run.repo_id, &spec.secrets)
            .await
            .map_err(api::internal)?
    };

    Ok(Some(JobAssignment {
        job_id: job.id,
        job_token,
        repo: format!("{}/{}", repo.owner_name, repo.name),
        clone_url: https_clone_url(&state.config, &repo.owner_name, &repo.name),
        commit: run.commit_oid,
        git_ref: run.git_ref,
        job_name: job.name,
        image: spec.image,
        run: spec.run,
        env: spec.env,
        secrets,
        artifacts: spec.artifacts,
        timeout_minutes: spec.timeout_minutes,
        network: spec.network,
    }))
}

/// True for a pull-request run whose head is not this repo, or whose pull request is gone. Such
/// runs get no secrets. Missing data counts as a fork, so the answer errs on the safe side.
async fn is_fork_run(state: &AppState, run: &CiRun) -> Result<bool, Refusal> {
    if run.trigger != "pull_request" {
        return Ok(false);
    }
    let Some(number) = run.pull_number else {
        return Ok(true);
    };
    match state
        .store
        .pull_by_number(&run.repo_id, number)
        .await
        .map_err(api::internal)?
    {
        Some(pull) => Ok(pull.head_repo_id.as_deref() != Some(run.repo_id.as_str())),
        None => Ok(true),
    }
}

// ---------------------------------------------------------------------------------------------
// Heartbeat, logs and artifacts
// ---------------------------------------------------------------------------------------------

#[route(POST "/api/v1/ci/jobs/{job_id}/heartbeat")]
async fn heartbeat(cx: &Cx) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
        Ok(job) => job,
        Err(res) => return Ok(*res),
    };
    match state.store.heartbeat(&job.id).await {
        Ok(cancel) => Ok(api::json(
            StatusCode::OK,
            &serde_json::to_value(HeartbeatResponse { cancel }).unwrap_or_default(),
        )),
        Err(e) => Ok(*api::internal(e)),
    }
}

/// Outcome of a log append.
enum LogWrite {
    /// The chunk is stored (or was dropped by the cap); the size is the new byte count.
    Stored(u64),
    /// This exact chunk was already stored: a retry. The size is unchanged.
    Duplicate(u64),
    /// The offset does not match the stored size.
    Mismatch(u64),
}

#[route(POST "/api/v1/ci/jobs/{job_id}/logs")]
async fn append_log(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let job = match job_from(state, headers, path_param::<JobId>(cx)).await {
        Ok(job) => job,
        Err(res) => return Ok(*res),
    };
    let Some(offset) = headers
        .get(LOG_OFFSET_HEADER)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok())
    else {
        return Ok(api::error(
            StatusCode::BAD_REQUEST,
            "X-Forge-Log-Offset must be a byte offset",
        ));
    };
    let chunk = match api::read_body(body, MAX_LOG_CHUNK).await {
        Ok(chunk) => chunk,
        Err(res) => return Ok(*res),
    };
    match write_log(state, &job, offset, &chunk).await {
        Ok(LogWrite::Stored(size) | LogWrite::Duplicate(size)) => {
            Ok(api::json(StatusCode::OK, &json!({ "size": size })))
        }
        Ok(LogWrite::Mismatch(size)) => {
            let mut res = api::error(StatusCode::CONFLICT, "the log offset does not match");
            res.headers_mut().insert(
                http::HeaderName::from_static(LOG_SIZE_HEADER),
                HeaderValue::from(size),
            );
            Ok(res)
        }
        Err(res) => Ok(*res),
    }
}

/// Appends one chunk of raw log bytes. Offsets count the raw bytes the runner sent, which is
/// `job.log_bytes`; the file holds the masked text, so its length may differ.
async fn write_log(
    state: &AppState,
    job: &CiJob,
    offset: u64,
    chunk: &[u8],
) -> Result<LogWrite, Refusal> {
    let size = u64::try_from(job.log_bytes).unwrap_or(0);
    let len = chunk.len() as u64;
    if offset != size {
        return Ok(if offset.checked_add(len) == Some(size) {
            LogWrite::Duplicate(size)
        } else {
            LogWrite::Mismatch(size)
        });
    }
    if chunk.is_empty() {
        return Ok(LogWrite::Stored(size));
    }

    let masked = mask_for_job(state, job, chunk).await?;
    let path = log_path(&state.config.data_dir, &job.id);
    store_log_bytes(state, &path, &masked).await?;
    state
        .store
        .add_log_bytes(&job.id, len as i64)
        .await
        .map_err(api::internal)?;
    state.ci_live.bump_job(&job.id);
    Ok(LogWrite::Stored(size + len))
}

/// Masks a chunk with the values of the job's secrets. The chunk is masked on its own, so a
/// secret split across two requests is masked by the runner, which holds back the boundary.
async fn mask_for_job(state: &AppState, job: &CiJob, chunk: &[u8]) -> Result<Vec<u8>, Refusal> {
    let spec: JobSpec = serde_json::from_str(&job.spec_json).map_err(api::internal)?;
    if spec.secrets.is_empty() {
        return Ok(chunk.to_vec());
    }
    let Some(run) = state
        .store
        .run_by_id(&job.run_id)
        .await
        .map_err(api::internal)?
    else {
        return Err(api::internal("log for a job with no run"));
    };
    let values = state
        .store
        .ci_secret_values(&run.repo_id, &spec.secrets)
        .await
        .map_err(api::internal)?;
    let masker = Masker::new(values.into_values());
    let mut stream = masker.stream();
    let mut out = stream.push(chunk);
    out.extend(stream.finish());
    Ok(out)
}

/// The bytes of `masked` that may be appended to a log of `current` bytes under a `cap`. Once the
/// log reaches the cap nothing more is kept. When the new bytes do not fit, the part that does is
/// kept and the marker follows it, so the file ends at exactly `cap` bytes.
pub fn fit_to_cap(current: u64, cap: u64, masked: &[u8]) -> Vec<u8> {
    if current >= cap {
        return Vec::new();
    }
    let data_cap = cap.saturating_sub(TRUNCATED_MARKER.len() as u64);
    let room = data_cap.saturating_sub(current);
    if masked.len() as u64 <= room {
        return masked.to_vec();
    }
    let mut out = masked[..room as usize].to_vec();
    out.extend_from_slice(TRUNCATED_MARKER);
    out
}

/// Appends masked bytes to the log file, stopping at the size cap with one marker line.
async fn store_log_bytes(
    state: &AppState,
    path: &std::path::Path,
    masked: &[u8],
) -> Result<(), Refusal> {
    use tokio::io::AsyncWriteExt;

    let cap = state.config.ci.max_log_mb.saturating_mul(MIB);
    let current = match tokio::fs::metadata(path).await {
        Ok(meta) => meta.len(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0,
        Err(e) => return Err(api::internal(e)),
    };
    let out = fit_to_cap(current, cap, masked);
    if out.is_empty() {
        return Ok(());
    }

    if let Some(parent) = path.parent() {
        tokio::fs::create_dir_all(parent)
            .await
            .map_err(api::internal)?;
    }
    let mut file = tokio::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await
        .map_err(api::internal)?;
    file.write_all(&out).await.map_err(api::internal)?;
    file.flush().await.map_err(api::internal)?;
    Ok(())
}

#[route(PUT "/api/v1/ci/jobs/{job_id}/artifacts/{artifact_name}")]
async fn upload_artifact(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
        Ok(job) => job,
        Err(res) => return Ok(*res),
    };
    let name = path_param::<ArtifactName>(cx);
    if !valid_artifact_name(name) {
        return Ok(api::error(
            StatusCode::UNPROCESSABLE_ENTITY,
            "the artifact name must be 1 to 128 of A-Z a-z 0-9 . _ + - and not start with a dot",
        ));
    }
    match upload(state, &job, name, body).await {
        Ok(artifact_id) => Ok(api::json(
            StatusCode::CREATED,
            &json!({ "id": artifact_id, "name": name }),
        )),
        Err(res) => Ok(*res),
    }
}

/// Streams the body to a temporary blob, then records the artifact and moves the blob to its
/// final key `ci/<artifact id>`. The id is only known once the row exists.
async fn upload(state: &AppState, job: &CiJob, name: &str, body: Body) -> Result<String, Refusal> {
    let max_artifacts = state.config.ci.max_artifacts_per_job;
    let existing = state
        .store
        .list_artifacts(&job.id)
        .await
        .map_err(api::internal)?;
    if existing.len() as u64 >= u64::from(max_artifacts) {
        return Err(Box::new(api::error(
            StatusCode::CONFLICT,
            "the job has too many artifacts",
        )));
    }
    if existing.iter().any(|artifact| artifact.name == name) {
        return Err(Box::new(api::error(
            StatusCode::CONFLICT,
            "an artifact with this name already exists",
        )));
    }

    let max_bytes = state.config.ci.max_artifact_mb.saturating_mul(MIB);
    let temp_key = format!("ci/tmp/{}", ulid::Ulid::generate());
    let mut writer = state
        .blobs
        .begin(&temp_key, max_bytes)
        .await
        .map_err(api::internal)?;
    let mut stream = std::pin::pin!(body.into_data_stream());
    while let Some(next) = stream.next().await {
        let outcome = match next {
            Ok(chunk) => writer.push(&chunk).await.map_err(|e| (e, false)),
            Err(_) => Err((CoreError::Invalid("upload interrupted".into()), true)),
        };
        if let Err((e, interrupted)) = outcome {
            writer.abort().await;
            return Err(upload_failure(e, interrupted));
        }
    }
    let (size, sha256) = match writer.finish().await {
        Ok(done) => done,
        Err(e) => {
            discard_temp(state, &temp_key).await;
            return Err(api::internal(e));
        }
    };

    let artifact = match state
        .store
        .add_artifact(&job.id, name, size as i64, &sha256)
        .await
    {
        Ok(artifact) => artifact,
        Err(e) => {
            discard_temp(state, &temp_key).await;
            return Err(match e {
                CoreError::Invalid(message) => Box::new(api::error(StatusCode::CONFLICT, &message)),
                other => api::internal(other),
            });
        }
    };
    let final_key = artifact_key(&artifact.id);
    if let Err(e) = state.blobs.rename(&temp_key, &final_key).await {
        discard_temp(state, &temp_key).await;
        return Err(api::internal(e));
    }
    Ok(artifact.id)
}

/// Removes a temporary blob. A failure is logged; the blob is unreachable either way.
async fn discard_temp(state: &AppState, key: &str) {
    if let Err(e) = state.blobs.delete(key).await {
        tracing::error!(error = %e, "ci: could not remove a temporary artifact blob");
    }
}

fn upload_failure(e: CoreError, interrupted: bool) -> Refusal {
    match e {
        CoreError::Invalid(_) if !interrupted => Box::new(api::error(
            StatusCode::PAYLOAD_TOO_LARGE,
            "the artifact is larger than the upload limit",
        )),
        CoreError::Invalid(message) => Box::new(api::error(StatusCode::BAD_REQUEST, &message)),
        other => api::internal(other),
    }
}

// ---------------------------------------------------------------------------------------------
// Finish
// ---------------------------------------------------------------------------------------------

#[route(POST "/api/v1/ci/jobs/{job_id}/finish")]
async fn finish(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let job = match job_from(state, headers(cx), path_param::<JobId>(cx)).await {
        Ok(job) => job,
        Err(res) => return Ok(*res),
    };
    let request: FinishRequest = match json_body(body).await {
        Ok(request) => request,
        Err(res) => return Ok(*res),
    };
    let state_after = match request.outcome {
        Outcome::Success => JobState::Success,
        Outcome::Failure => JobState::Failure,
        Outcome::Cancelled => JobState::Cancelled,
        Outcome::Error => JobState::Error,
    };
    match finish_job(state, &job, state_after, &request).await {
        Ok(state_name) => Ok(api::json(StatusCode::OK, &json!({ "state": state_name }))),
        Err(res) => Ok(*res),
    }
}

/// Finishes the job, then syncs the statuses of the job and of every job whose state changed
/// (skipped or queued by this finish). Wakes the claimers when something was queued, and the
/// watchers of every job and run that changed.
async fn finish_job(
    state: &AppState,
    job: &CiJob,
    outcome: JobState,
    request: &FinishRequest,
) -> Result<&'static str, Refusal> {
    let before: BTreeMap<String, JobState> = state
        .store
        .jobs_for_run(&job.run_id)
        .await
        .map_err(api::internal)?
        .into_iter()
        .map(|other| (other.id, other.state))
        .collect();
    let exit_code = request.exit_code.map(i64::from);
    let run = state
        .store
        .finish_job(&job.id, outcome, exit_code, request.message.as_deref())
        .await
        .map_err(api::internal)?;

    let mut queued = false;
    let mut any_changed = false;
    for other in state
        .store
        .jobs_for_run(&job.run_id)
        .await
        .map_err(api::internal)?
    {
        let changed = before.get(&other.id) != Some(&other.state);
        if changed {
            any_changed = true;
            if other.state.is_terminal() {
                state.ci_live.close_job(&other.id);
            } else {
                state.ci_live.bump_job(&other.id);
            }
        }
        if other.id == job.id
            || (changed && matches!(other.state, JobState::Skipped | JobState::Queued))
        {
            queued |= other.state == JobState::Queued;
            crate::ci::sync_status(state, &run, &other).await;
        }
    }
    if any_changed {
        state.ci_live.bump_run(&run.id);
    }
    if queued {
        state.ci_waker.notify_waiters();
    }
    // A run that just went green may deploy the forge itself.
    self_update::maybe_deploy(state, &run).await;
    Ok(run.state.as_str())
}

#[cfg(test)]
#[path = "api_tests.rs"]
mod tests;