//! `flow runner` (docs/m0.md §5): claim a job from OxygenHub, fetch the

//! repository into a cache with the job's token, write the state into a

//! workspace, run it in its tier, stream the log, report the receipt.

//! One job at a time.

use crate::home::{Home, RunnerConfig, RunnerKey};

use crate::{microvm, workspace};

use anyhow::{bail, Context, Result};

use flow_core::api::{Claim, ClaimRequest, Decline, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};

use flow_core::{scope_tree_hash, Receipt};

use oxy_core::hash::ChangeHash;

use oxy_core::repo::Repo;

use std::path::PathBuf;

use std::time::{Duration, Instant};

/// A request the hub refused (4xx). On a job's endpoints, 409 means the

/// job no longer runs: it was cancelled, or given up.

#[derive(Debug)]

pub struct Refused {

    pub code: u16,

    pub text: String,

}

impl std::fmt::Display for Refused {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        write!(f, "HTTP {}: {}", self.code, self.text)

    }

}

impl std::error::Error for Refused {}

/// Whether `e` is the hub saying the job no longer runs.

fn job_gone(e: &anyhow::Error) -> bool {

    e.downcast_ref::<Refused>().is_some_and(|r| r.code == 409)

}

/// What a job ends with when its run was cancelled: nothing is reported,

/// as the hub has ended it already.

const CANCELLED: &str = "the run was cancelled";

fn cancelled() -> Finish {

    Finish { exit_code: None, duration_ms: 0, receipt: None, signed_receipt: None, error: Some(CANCELLED.into()) }

}

pub struct Hub {

    base: String,

    token: String,

    agent: ureq::Agent,

}

impl Hub {

    pub fn new(cfg: &RunnerConfig) -> Result<Hub> {

        let tls = oxy_core::protocol::tls_client_config(cfg.ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;

        let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();

        Ok(Hub { base: cfg.hub.trim_end_matches('/').to_string(), token: cfg.token.clone(), agent })

    }

    /// POST to the hub. A server error (5xx) means nothing was committed,

    /// so it is tried again after 1, 2 and 4 seconds before giving up: a

    /// busy hub costs a job a few seconds, not its log lines or its claim.

    fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {

        self.post_with(path, body, &RETRY_DELAYS)

    }

    fn post_with(&self, path: &str, body: &impl serde::Serialize, delays: &[Duration]) -> Result<Option<ureq::Response>> {

        let url = format!("{}/api/v1{path}", self.base);

        let mut delays = delays.iter();

        loop {

            match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body) {

                Ok(r) if r.status() == 204 => return Ok(None),

                Ok(r) => return Ok(Some(r)),

                Err(ureq::Error::Status(code, r)) => {

                    let text = r.into_string().unwrap_or_default();

                    match delays.next() {

                        Some(d) if code >= 500 => std::thread::sleep(*d),

                        _ if code < 500 => return Err(anyhow::Error::new(Refused { code, text }).context(url)),

                        _ => bail!("{url}: HTTP {code}: {text}"),

                    }

                }

                Err(e) => bail!("{url}: {e}"),

            }

        }

    }

    pub fn claim(&self, req: &ClaimRequest) -> Result<Option<Claim>> {

        match self.post("/flow/claim", req)? {

            Some(r) => Ok(Some(r.into_json()?)),

            None => Ok(None),

        }

    }

    /// Hand job `job` back, saying why this runner will not run it.

    pub fn decline(&self, job: i64, reason: &str) -> Result<()> {

        self.post(&format!("/flow/jobs/{job}/decline"), &Decline { reason: reason.into() })?;

        Ok(())

    }

    pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {

        self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;

        Ok(())

    }

    /// Upload artifact `name` of job `job` (spec §5.2); the hub keeps it as

    /// a blob of the job's repository and answers with its ObjectRef.

    pub fn artifact(&self, job: i64, name: &str, bytes: &[u8]) -> Result<String> {

        let url = format!("{}/api/v1/flow/jobs/{job}/artifacts/{name}", self.base);

        let mut delays = RETRY_DELAYS.iter();

        loop {

            let req = self

                .agent

                .post(&url)

                .timeout(Duration::from_secs(600))

                .set("Authorization", &format!("Bearer {}", self.token))

                .set("Content-Type", "application/octet-stream");

            match req.send_bytes(bytes) {

                Ok(r) => {

                    let v: serde_json::Value = r.into_json()?;

                    return v["blob"].as_str().map(String::from).context("the hub named no blob");

                }

                Err(ureq::Error::Status(code, r)) => {

                    let text = r.into_string().unwrap_or_default();

                    match delays.next() {

                        Some(d) if code >= 500 => std::thread::sleep(*d),

                        _ => bail!("{url}: HTTP {code}: {text}"),

                    }

                }

                Err(e) => bail!("{url}: {e}"),

            }

        }

    }

    /// Report a job's end until the hub has it (oxygenhub#71): while the hub

    /// cannot be reached or fails (5xx), try again after `first_wait`,

    /// doubling to a minute, for up to `patience`. A refusal (4xx: the job was ended

    /// already, say) is final. A result lost here leaves the job running

    /// on the hub until its time limit.

    pub fn finish_until_accepted(&self, job: i64, f: &Finish, patience: Duration, first_wait: Duration) -> Result<()> {

        let url = format!("{}/api/v1/flow/jobs/{job}/finish", self.base);

        let deadline = std::time::Instant::now() + patience;

        let mut wait = first_wait;

        loop {

            let why = match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(f) {

                Ok(_) => return Ok(()),

                Err(ureq::Error::Status(code, r)) if code < 500 => {

                    bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default())

                }

                Err(ureq::Error::Status(code, _)) => format!("HTTP {code}"),

                Err(e) => e.to_string(),

            };

            if std::time::Instant::now() + wait > deadline {

                bail!("{url}: gave up after {} minutes: {why}", patience.as_secs() / 60);

            }

            eprintln!("job {job}: the hub did not take the result ({why}); trying again in {} s", wait.as_secs());

            std::thread::sleep(wait);

            wait = (wait * 2).min(Duration::from_secs(60));

        }

    }

}

/// How long a runner keeps trying to report a job's end.

const FINISH_PATIENCE: Duration = Duration::from_secs(2 * 3600);

/// Where a job's result waits until the hub has it, so a runner that

/// restarts still reports it.

fn pending_path(home: &Home, job: i64) -> PathBuf {

    home.root.join("pending").join(format!("{job}.json"))

}

/// Report `f`, keeping it on disk until the hub has it.

fn report_finish(home: &Home, hub: &Hub, job: i64, f: &Finish) {

    let path = pending_path(home, job);

    let kept = std::fs::create_dir_all(path.parent().unwrap())

        .and_then(|_| std::fs::write(&path, serde_json::to_vec(f).unwrap_or_default()))

        .is_ok();

    match hub.finish_until_accepted(job, f, FINISH_PATIENCE, Duration::from_secs(5)) {

        Ok(()) => {}

        Err(e) => eprintln!("job {job}: reporting failed: {e:#}"),

    }

    if kept {

        let _ = std::fs::remove_file(&path);

    }

}

/// Results a previous run of the runner could not report, to send again.

fn report_pending(home: &Home) -> Vec<(i64, Finish)> {

    let Ok(dir) = std::fs::read_dir(home.root.join("pending")) else {

        return Vec::new();

    };

    dir.filter_map(|e| {

        let p = e.ok()?.path();

        let job: i64 = p.file_stem()?.to_str()?.parse().ok()?;

        let f: Finish = serde_json::from_slice(&std::fs::read(&p).ok()?).ok()?;

        Some((job, f))

    })

    .collect()

}

/// `flow runner register`: exchange a one-time registration token from

/// the hub for this runner's credential, binding the runner's signing key,

/// and keep hub, credential and name in runner.toml.

pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {

    if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {

        bail!("flow.toml already has a [runner] section; remove it to register this runner instead");

    }

    let key = home.runner_key(name)?;

    let hub = hub.trim_end_matches('/').to_string();

    let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;

    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();

    let req = Register {

        token: token.trim().to_string(),

        name: name.to_string(),

        identity: hex::encode(key.identity.encode()),

        info: RunnerInfo { version: env!("OXY_VERSION_LONG").into(), ..RunnerInfo::here(labels.clone(), 1, None, None) },

    };

    let url = format!("{hub}/api/v1/flow/runners/register");

    let got: Registered = match agent.post(&url).send_json(&req) {

        Ok(r) => r.into_json()?,

        Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),

        Err(e) => bail!("{url}: {e}"),

    };

    let mut text = format!(

        "# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",

        got.credential, got.name

    );

    if let Some(ca) = &ca {

        text.push_str(&format!("ca = {:?}\n", ca.display().to_string()));

    }

    if !labels.is_empty() {

        text.push_str(&format!("labels = {labels:?}\n"));

    }

    let path = home.root.join(crate::home::RUNNER_TOML);

    {

        use std::io::Write;

        use std::os::unix::fs::OpenOptionsExt;

        let tmp = path.with_extension("toml.tmp");

        let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;

        f.write_all(text.as_bytes())?;

        f.sync_all()?;

        std::fs::rename(&tmp, &path)?;

    }

    Ok(got)

}

/// Waits before trying a hub request again after a server error.

const RETRY_DELAYS: [Duration; 3] = [Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)];

/// What the runner may run at once (flow.toml `[runner]`): at most

/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.

#[derive(Debug, Default, PartialEq, Eq)]

pub struct Budget {

    pub max_jobs: usize,

    pub memory_mib: Option<u32>,

    pub cpus: Option<u32>,

}

/// What is in use now.

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

pub struct Used {

    pub jobs: usize,

    pub memory_mib: u32,

    pub cpus: u32,

}

impl Budget {

    pub fn of(cfg: &RunnerConfig) -> Budget {

        Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }

    }

    /// The room to ask the hub for, or `None` when no job more can start.

    /// A lone job may use the whole budget even if it asks for more, so a

    /// large sandbox is never starved.

    pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {

        if used.jobs >= self.max_jobs {

            return None;

        }

        if used.jobs == 0 {

            return Some((None, None));

        }

        let mem = self.memory_mib.map(|m| m.saturating_sub(used.memory_mib));

        let cpus = self.cpus.map(|c| c.saturating_sub(used.cpus));

        if mem == Some(0) || cpus == Some(0) {

            return None;

        }

        Some((mem, cpus))

    }

}

/// The isolation tiers this machine can honour now (spec §11.1): the

/// MicroVM tier when `/dev/kvm` opens for reading and writing and

/// flow.toml pins Firecracker; the syscall tier when this binary may use

/// user namespaces, Landlock and seccomp and manage a delegated cgroup.

/// `Err` says why there are none; the reasons for missing tiers come back

/// with the ones there are.

pub fn tiers(home: &Home) -> std::result::Result<(Vec<String>, Vec<String>), String> {

    let mut have = Vec::new();

    let mut missing = Vec::new();

    let kvm = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm");

    match (kvm, home.config.microvm.is_some()) {

        (Ok(_), true) => have.push("microvm".to_string()),

        (Ok(_), false) => missing.push("microvm: flow.toml has no [microvm] section".to_string()),

        (Err(e), _) => missing.push(format!(

            "microvm: /dev/kvm cannot be opened ({e}); the tier needs hardware virtualisation and this user in the kvm group"

        )),

    }

    match crate::syscall::available() {

        Ok(()) => have.push("syscall".to_string()),

        Err(e) => missing.push(format!("syscall: {e}")),

    }

    if have.is_empty() {

        return Err(format!("no isolation tier can run here: {}", missing.join("; ")));

    }

    Ok((have, missing))

}

/// Fetch `claim`'s image from the hub when this home lacks it

/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it

/// leaves it missing, and the preflight declines the job as before.

fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) {

    let hash = &claim.sandbox.rootfs;

    if hash.is_empty() || home.rootfs(hash).is_file() || !jet::images::is_hash(hash) {

        return;

    }

    let note = |text: String| {

        eprintln!("job {}: {text}", claim.job_id);

        hub.log(claim.job_id, vec![Record { t: 0, stream: "flow".into(), text }]).ok();

    };

    // One fetch of an image at a time: two jobs wanting it wait for one.

    static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(());

    let _one = FETCHING.lock().unwrap_or_else(|p| p.into_inner());

    if home.rootfs(hash).is_file() {

        return;

    }

    let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) {

        Ok(c) => c,

        Err(e) => return note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12])),

    };

    let fetched = match swarm {

        Some(sw) => {

            note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12]));

            sw.fetch(home, hash)

        }

        None => {

            note(format!("image {} is not on this runner: fetching it from the hub", &hash[..12]));

            client.fetch(home, hash)

        }

    };

    match fetched {

        Ok(Some(m)) => note(format!(

            "image {}: fetched {} chunks, {}{} in {:.1} s",

            &hash[..12],

            m.chunks,

            crate::images::human(m.bytes),

            if swarm.is_some() { format!(" ({} from peers)", crate::images::human(m.peer_bytes)) } else { String::new() },

            m.elapsed.as_secs_f64()

        )),

        Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])),

        Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])),

    }

}

/// Why this runner cannot run `claim` now (spec §11.5), checked before

/// anything is fetched: the tier, the architecture, the image and the

/// memory. The hub chose it from what the runner last said, which may be

/// out of date.

fn preflight(home: &Home, claim: &Claim) -> Option<String> {

    let sb = &claim.sandbox;

    match sb.runtime {

        flow_core::Runtime::Microvm => {

            if let Err(e) = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm") {

                return Some(format!("the MicroVM tier needs /dev/kvm, which cannot be opened here ({e})"));

            }

            if home.config.microvm.is_none() {

                return Some("the MicroVM tier is not set up here (flow.toml has no [microvm] section)".into());

            }

        }

        flow_core::Runtime::Syscall => {

            if let Err(e) = crate::syscall::available() {

                return Some(format!("the syscall tier cannot run here: {e}"));

            }

        }

        flow_core::Runtime::Wasm => return Some("this runner has no wasm tier".into()),

    }

    sandbox_problem(home, sb, std::env::consts::ARCH, available_memory_mib())

}

/// What stops `sb` running on this home's images, on a machine of

/// architecture `here` with `free_mib` of memory free.

fn sandbox_problem(home: &Home, sb: &flow_core::Sandbox, here: &str, free_mib: Option<u64>) -> Option<String> {

    if let Some(want) = sb.arch.as_deref().filter(|a| *a != here) {

        return Some(format!("the sandbox needs {want}; this runner is {here}"));

    }

    let short = &sb.rootfs[..12.min(sb.rootfs.len())];

    let image = home.rootfs(&sb.rootfs);

    if !image.is_file() {

        return Some(format!("rootfs {short} is not on this runner"));

    }

    if let Some(arch) = home.image_arch(&sb.rootfs).filter(|a| a != here) {

        return Some(format!("rootfs {short} is for {arch}; this runner is {here}"));

    }

    if let Err(e) = home.verified(&image, &sb.rootfs) {

        return Some(format!("rootfs {short} does not verify here: {e:#}"));

    }

    if let Some(free) = free_mib.filter(|free| *free < u64::from(sb.memory_mib)) {

        return Some(format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));

    }

    None

}

/// `MemAvailable` from /proc/meminfo, in MiB.

fn available_memory_mib() -> Option<u64> {

    let info = std::fs::read_to_string("/proc/meminfo").ok()?;

    let kib: u64 = info.lines().find_map(|l| l.strip_prefix("MemAvailable:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;

    Some(kib / 1024)

}

pub fn serve(home: &Home) -> Result<()> {

    let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;

    // A runner offers only what it can run: a machine without KVM claims

    // nothing rather than taking MicroVM jobs and failing them (#36).

    let (runtimes, missing) = tiers(home).map_err(|why| anyhow::anyhow!("flow runner: {why}"))?;

    for why in &missing {

        eprintln!("flow runner: not offering {why}");

    }

    let hub = Hub::new(&cfg)?;

    let name = cfg.name.clone().unwrap_or_else(|| "flow".into());

    let budget = Budget::of(&cfg);

    let key = home.runner_key(&name)?;

    // The swarm (docs/jetstream.md §8.1): a runner that cannot serve its

    // peers still runs jobs, fetching images from the hub alone.

    let swarm = crate::swarm::Swarm::start(home, &cfg).unwrap_or_else(|e| {

        eprintln!("flow runner: not joining the swarm: {e:#}");

        None

    });