//! The runner API between OxygenHub and `flow runner` (docs/m0.md §5).

//! Both sides use these types, so they cannot drift.

use crate::pipeline::{Job, Sandbox};

use crate::receipt::Receipt;

use serde::{Deserialize, Serialize};

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

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct ClaimRequest {

    /// The runner's name, shown on job pages.

    pub runner: String,

    /// Tiers this runner can run (`microvm`, `syscall`).

    pub runtimes: Vec<String>,

    /// Memory the runner has free for another job, in MiB: the hub hands

    /// out only a job whose sandbox fits. Absent: no limit.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub free_memory_mib: Option<u32>,

    /// vCPUs the runner has free for another job. Absent: no limit.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub free_cpus: Option<u32>,

    /// What the runner is, for the hub's Runners page.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub info: Option<RunnerInfo>,

    /// The rootfs images the runner holds (BLAKE3 hex, as pipelines pin

    /// them): the hub offers only jobs whose image is among them

    /// (oxygenhub#73). Absent: the runner does not say, and any job may be

    /// offered.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub images: Option<Vec<String>>,

    /// The architecture of each image it holds, by hash, as the image's

    /// manifest records it (spec §11.2): the hub offers a job only to a

    /// runner whose architecture its image is for.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub image_archs: Option<std::collections::BTreeMap<String, String>>,

    /// Whether the runner fetches a job's image from the hub when it lacks

    /// it (oxygenflow#39): the hub may then offer it jobs whose image only

    /// the hub holds.

    #[serde(default, skip_serializing_if = "std::ops::Not::not")]

    pub fetches_images: bool,

    /// The jobs this runner holds now: running, or finished with a result

    /// not yet reported. The hub ends any other job it has as running on

    /// this runner, which the runner lost by restarting or never received

    /// (oxygenhub#71). Absent: the runner does not say.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub holding: Option<Vec<i64>>,

    /// What the runner measured about itself just now (spec §11.1,

    /// docs/jetstream.md §6). Absent: an older runner.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub facts: Option<Facts>,

    /// Where this runner serves image chunks to its peers (docs/jetstream.md

    /// §8.1). Absent: it is not in a swarm.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub swarm: Option<SwarmAddr>,

    /// The build caches this runner holds a disk of (oxygenflow#51), each

    /// [`cache_slot`] of a repository and job, so the hub can send a job

    /// where its cache is. Absent: an older runner.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub caches: Option<Vec<String>>,

    /// When each of [`ClaimRequest::caches`] last kept a disk, Unix ms, so

    /// the hub can send a job to the runner whose copy is freshest when

    /// several hold one (oxygenflow#54). Absent: an older runner.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub cache_times: Option<std::collections::BTreeMap<String, u64>>,

}

/// A name safe as one path segment: kept readable, with a short hash so

/// two names that clean up alike stay apart. The build cache's pool is laid

/// out by these (oxygenflow#28).

pub fn segment(s: &str) -> String {

    let clean: String = s.chars().map(|c| if c.is_ascii_alphanumeric() || "._-".contains(c) { c } else { '_' }).take(48).collect();

    format!("{clean}-{}", &blake3::hash(s.as_bytes()).to_hex()[..8])

}

/// Where a repository's job keeps its build caches, as runners report

/// them in [`ClaimRequest::caches`] (oxygenflow#51): `repo/job`, each part

/// a [`segment`].

pub fn cache_slot(repo: &str, job: &str) -> String {

    format!("{}/{}", segment(repo), segment(job))

}

/// A runner's place in the image swarm: the address peers reach it on,

/// the site it is in (peers are only ever in the same site), and the

/// BLAKE3 of its TLS certificate, which peers check it by.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct SwarmAddr {

    pub addr: String,

    #[serde(default)]

    pub site: String,

    #[serde(default)]

    pub cert: String,

}

/// `POST /flow/swarm/announce`: a runner starting, continuing or ending a

/// fetch of `image`.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Announce {

    pub image: String,

    /// `fetching`, `done` or `stopped`.

    pub state: String,

}

/// The hub's answer to an announce: the peers to ask, and the ticket that

/// admits this runner to them.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct Announced {

    pub ticket: String,

    pub peers: Vec<SwarmPeer>,

    /// Announce again after this many seconds while still fetching.

    pub interval_s: u64,

}

/// A peer in the swarm for one image.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct SwarmPeer {

    pub runner: i64,

    pub name: String,

    pub addr: String,

    /// The BLAKE3 of its TLS certificate: a peer that shows another is not

    /// this runner.

    pub cert: String,

    /// It holds the whole image (rather than fetching it now).

    pub complete: bool,

}

/// A runner's own measurements, sent with every claim (spec §11.1): cheap

/// ones taken each time, the speed from a short benchmark run at start and

/// again while idle. Whole numbers, so a claim compares exactly.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct Facts {

    /// Free disk where jobs' disks are made, in MiB.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub free_disk_mib: Option<u64>,

    /// The 1-minute load average, times 100.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub load_x100: Option<u32>,

    /// The benchmark's score: 100 on the reference machine (Earth), twice

    /// as fast scoring 200.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub speed: Option<u32>,

    /// The speed class, 1 (slow) to 5 (fast); each class twice the one

    /// below (spec §11.1).

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub speed_class: Option<u8>,

    /// The kernel release.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub kernel: Option<String>,

}

impl Facts {

    /// The class of a speed score: 3 for the reference machine's 100, each

    /// class a factor of two.

    pub fn class_of(speed: u32) -> u8 {

        match speed {

            0..=34 => 1,

            35..=69 => 2,

            70..=139 => 3,

            140..=279 => 4,

            _ => 5,

        }

    }

}

/// A runner's description: shown on the hub's Runners page, and what

/// scheduling by label and architecture will match.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct RunnerInfo {

    /// `x86_64`, `aarch64`, `riscv64`, …

    pub arch: String,

    pub os: String,

    /// The `flow` version.

    pub version: String,

    /// From flow.toml `[runner] labels`.

    #[serde(default)]

    pub labels: Vec<String>,

    /// Jobs at once, and the memory (MiB) and vCPUs they share.

    pub max_jobs: u32,

    #[serde(default)]

    pub memory_mib: Option<u32>,

    #[serde(default)]

    pub cpus: Option<u32>,

}

impl RunnerInfo {

    /// This machine and build, with the given labels and capacity.

    pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> RunnerInfo {

        RunnerInfo {

            arch: std::env::consts::ARCH.into(),

            os: std::env::consts::OS.into(),

            version: env!("CARGO_PKG_VERSION").into(),

            labels,

            max_jobs,

            memory_mib,

            cpus,

        }

    }

}

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

/// token from the hub for a runner credential, binding the runner's

/// signing identity to it.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Register {

    pub token: String,

    pub name: String,

    /// The runner's Oxygen identity, hex of its encoding: its key signs

    /// every receipt, and the hub accepts no other for this runner.

    pub identity: String,

    pub info: RunnerInfo,

}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Registered {

    pub id: i64,

    pub name: String,

    /// The runner's bearer token from now on; shown once.

    pub credential: String,

}

/// A pinned subrepo to put in the workspace: `repo`'s state `heads`, at `path`.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct SubrepoInput {

    pub path: String,

    /// `owner/name` on the hub.

    pub repo: String,

    pub clone_url: String,

    pub heads: Vec<String>,

}

/// A secret for a job, as the hub supplies it (spec §9): written by the

/// runner at `path` in the sandbox, 0400, on an in-memory mount; never in

/// the environment, the receipt or an artifact.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct SecretInput {

    pub name: String,

    pub path: String,

    pub value: String,

}

/// A claimed job: everything the runner needs, nothing more.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Claim {

    pub job_id: i64,

    pub run_id: i64,

    /// `owner/name`.

    pub repo: String,

    /// Where to fetch the repository from (`oxy` wire protocol).

    pub clone_url: String,

    /// Read access to this repository only, until the job finishes.

    pub token: String,

    /// The state to build, as full change hashes.

    pub heads: Vec<String>,

    /// Hex BLAKE3 of the canonical pipeline.

    pub pipeline_hash: String,

    pub job: Job,

    pub sandbox: Sandbox,

    /// The hub's limit for this job.

    pub timeout_s: u64,

    /// Subrepos the state pins (Oxygen spec §7.8) that the job may read,

    /// fetched with the same token and written at their paths.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub subrepos: Vec<SubrepoInput>,

    /// The state's hash on the hub (what checks are keyed by).

    #[serde(default, skip_serializing_if = "String::is_empty")]

    pub state: String,

    /// The bookmark the run is for.

    #[serde(default, skip_serializing_if = "String::is_empty")]

    pub bookmark: String,

    /// The run's number within its repository.

    #[serde(default)]

    pub run_number: i64,

    /// Variables the hub holds for the repository (its site's, its

    /// organisation's and its own, inner overriding outer; oxygenhub#102).

    /// Set under `FLOW_*` and under the pipeline's own `env`, so they are

    /// part of the job's environment and of its receipt. Never secrets:

    /// those are mounted, not put in the environment (spec §9).

    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]

    pub env: std::collections::BTreeMap<String, String>,

    /// The secrets the job declares (`secret` blocks), as the hub holds

    /// them for the repository (oxygenhub#107).

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub secrets: Vec<SecretInput>,

    /// The job's inputs (oxygenflow#58), in the job's order, each with the

    /// blob the producing job's receipt names: fetched from

    /// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub inputs: Vec<ArtifactInput>,

    /// How long the job is expected to take, from its recent passing runs,

    /// when the hub has learned it (#76): a runner leaving with some notice

    /// finishes a job whose rest fits it and hands back the others.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub expected_ms: Option<u64>,

    /// The checkpoint an earlier attempt saved before its runner handed the

    /// job back (oxygenflow#77): the blob to fetch from

    /// `/api/v1/flow/jobs/{job_id}/checkpoint` and check, then restore

    /// before the job runs.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub checkpoint: Option<String>,

}

/// An artifact a job takes (oxygenflow#58), as the hub hands it over.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct ArtifactInput {

    pub artifact: String,

    /// The producing job.

    pub from: String,

    /// Where it goes, under `/src`.

    pub path: String,

    /// The Oxygen blob reference its receipt names, hex.

    pub blob: String,

}

impl Claim {

    /// What a job is told about what it builds (docs/m0.md §4):

    /// `FLOW_REPO`, `FLOW_STATE`, `FLOW_HEADS` (full hashes joined with

    /// `+`), `FLOW_BOOKMARK`, `FLOW_RUN` and `FLOW_JOB`. Unknown values are

    /// left out.

    pub fn flow_env(&self) -> Vec<(String, String)> {

        let mut v = vec![

            ("FLOW_REPO".to_string(), self.repo.clone()),

            ("FLOW_HEADS".to_string(), self.heads.join("+")),

            ("FLOW_JOB".to_string(), self.job.name.clone()),

        ];

        if !self.state.is_empty() {

            v.push(("FLOW_STATE".into(), self.state.clone()));

        }

        if !self.bookmark.is_empty() {

            v.push(("FLOW_BOOKMARK".into(), self.bookmark.clone()));

        }

        if self.run_number > 0 {

            v.push(("FLOW_RUN".into(), self.run_number.to_string()));

        }

        for (k, val) in &self.env {

            if !v.iter().any(|(n, _)| n == k) {

                v.push((k.clone(), val.clone()));

            }

        }

        v

    }

}

/// One line of output (docs/m0.md §2).

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Record {

    /// Milliseconds since the job started.

    pub t: u64,

    /// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner).

    pub stream: String,

    pub text: String,

}

/// `POST /api/v1/flow/jobs/{id}/decline` (spec §11.5): the runner will

/// not run a job it was handed, and says why. The hub queues the job

/// again and offers it to this runner no more until its profile changes.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct Decline {

    pub reason: String,

}

/// `POST /api/v1/flow/jobs/{id}/log`.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct LogBatch {

    pub records: Vec<Record>,

}

/// `POST /api/v1/flow/jobs/{id}/finish`.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Finish {

    /// The job's exit code; absent when the runner could not run it.

    pub exit_code: Option<i32>,

    pub duration_ms: u64,

    pub receipt: Option<Receipt>,

    /// The receipt bound to the job's state and signed by the runner

    /// (Oxygen spec §14.6), hex of the signed receipt's bytes.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub signed_receipt: Option<String>,

    /// Why the runner could not run the job (rootfs missing, VM failed).

    pub error: Option<String>,

    /// The most disk the job's disks and workspace took, in MiB, and the

    /// most memory its sandbox held (oxygenflow#41): the hub learns from

    /// them what the job's next run needs.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub peak_disk_mib: Option<u64>,

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub peak_memory_mib: Option<u64>,

}

/// Longest text of one record the hub keeps; longer lines are cut.

pub const MAX_RECORD_TEXT: usize = 16 * 1024;

/// Most records the hub keeps per job; later ones are dropped with a note.

pub const MAX_RECORDS: usize = 50_000;