//! `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};
use crate::{microvm, workspace};
use anyhow::{bail, Context, Result};
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, Record, 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};

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

    fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
        let url = format!("{}/api/v1{path}", self.base);
        match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body) {
            Ok(r) if r.status() == 204 => Ok(None),
            Ok(r) => Ok(Some(r)),
            Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
            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),
        }
    }

    pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
        self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
        Ok(())
    }

    pub fn finish(&self, job: i64, f: &Finish) -> Result<()> {
        self.post(&format!("/flow/jobs/{job}/finish"), f)?;
        Ok(())
    }
}

pub fn serve(home: &Home) -> Result<()> {
    let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
    let hub = Hub::new(&cfg)?;
    let name = cfg.name.clone().unwrap_or_else(|| "flow".into());
    let req = ClaimRequest { runner: name.clone(), runtimes: vec!["microvm".into()] };
    eprintln!("flow runner {name}: polling {}", cfg.hub);
    let mut idle = Duration::from_secs(2);
    loop {
        match hub.claim(&req) {
            Ok(Some(claim)) => {
                idle = Duration::from_secs(2);
                let id = claim.job_id;
                eprintln!("job {id}: {} {} on {}", claim.repo, claim.job.name, claim.heads.first().map(|h| &h[..12]).unwrap_or("-"));
                let finish = run_claim(home, &hub, &cfg, &claim);
                let finish =
                    finish.unwrap_or_else(|e| Finish { exit_code: None, duration_ms: 0, receipt: None, error: Some(format!("{e:#}")) });
                eprintln!(
                    "job {id}: exit {:?}{}",
                    finish.exit_code,
                    finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()
                );
                if let Err(e) = hub.finish(id, &finish) {
                    eprintln!("job {id}: reporting failed: {e:#}");
                }
            }
            Ok(None) => std::thread::sleep(idle),
            Err(e) => {
                eprintln!("claim: {e:#}");
                std::thread::sleep(idle);
                idle = (idle * 2).min(Duration::from_secs(60));
            }
        }
    }
}

/// Batches records to the hub every second (or every 200 lines).
struct LogSink<'a> {
    hub: &'a Hub,
    job: i64,
    buf: Vec<Record>,
    last: Instant,
}

impl LogSink<'_> {
    fn push(&mut self, mut r: Record) {
        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);
        }
        self.buf.push(r);
        if self.buf.len() >= 200 || self.last.elapsed() >= Duration::from_secs(1) {
            self.flush();
        }
    }

    fn flush(&mut self) {
        if !self.buf.is_empty() {
            if let Err(e) = self.hub.log(self.job, std::mem::take(&mut self.buf)) {
                eprintln!("job {}: log: {e:#}", self.job);
            }
        }
        self.last = Instant::now();
    }
}

fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, claim: &Claim) -> Result<Finish> {
    let started = Instant::now();
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now() };
    let note =
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
    note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+")));
    let repo = cached_repo(home, cfg, claim)?;
    let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
    let _ = std::fs::remove_dir_all(&scratch);
    std::fs::create_dir_all(scratch.join("tree"))?;
    let result = (|| -> Result<Finish> {
        let files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
        let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(|(p, h)| (p.as_str(), *h)));
        workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
        std::fs::remove_dir_all(scratch.join("tree"))?;
        note(
            &mut sink,
            format!(
                "booting a microVM: {} vCPU, {} MiB, rootfs {}",
                claim.sandbox.cpus,
                claim.sandbox.memory_mib,
                &claim.sandbox.rootfs[..12]
            ),
        );
        let outcome = microvm::run(
            home,
            &claim.sandbox,
            &claim.job,
            &scratch.join("workspace.ext4"),
            &scratch,
            Duration::from_secs(claim.timeout_s),
            &mut |r| sink.push(r),
        )?;
        let receipt = outcome.exit_code.map(|code| Receipt {
            job_name: claim.job.name.clone(),
            pipeline_hash: hex32(&claim.pipeline_hash),
            scope_tree_hash: scope_hash,
            inputs: vec![],
            environment: outcome.environment.hash(),
            outputs: vec![],
            exit_code: code as u32,
            duration_ms: outcome.duration_ms,
        });
        Ok(Finish {
            exit_code: outcome.exit_code,
            duration_ms: started.elapsed().as_millis() as u64,
            receipt,
            error: outcome.timed_out.then(|| "stopped at the time limit".to_string()),
        })
    })();
    let _ = std::fs::remove_dir_all(&scratch);
    sink.flush();
    result
}

fn hex32(s: &str) -> [u8; 32] {
    let mut out = [0; 32];
    if let Ok(v) = hex::decode(s) {
        if v.len() == 32 {
            out.copy_from_slice(&v);
        }
    }
    out
}

/// The runner's copy of a repository, fetched up to date with the job's token.
fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
    let dir: PathBuf = home.root.join("repos").join(claim.repo.replace('/', "__"));
    let set_remote = |repo: &mut Repo| -> Result<()> {
        repo.config.set("remote.origin.url", &claim.clone_url).map_err(|e| anyhow::anyhow!("{e}"))?;
        repo.config.set("remote.origin.token", &claim.token).map_err(|e| anyhow::anyhow!("{e}"))?;
        if let Some(ca) = &cfg.ca {
            repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(|e| anyhow::anyhow!("{e}"))?;
        }
        repo.save_config().map_err(|e| anyhow::anyhow!("{e}"))
    };
    if dir.join(".oxy").is_dir() {
        let mut repo = Repo::open(&dir).map_err(|e| anyhow::anyhow!("{e}"))?;
        set_remote(&mut repo)?;
        repo.fetch("origin").map_err(|e| anyhow::anyhow!("fetch {}: {e}", claim.clone_url))?;
        return Ok(repo);
    }
    // First job for this repository: clone with the job's token.
    std::env::set_var("OXY_TOKEN", &claim.token);
    if let Some(ca) = &cfg.ca {
        std::env::set_var("OXY_CA", ca);
    }
    let cloned = Repo::clone(&claim.clone_url, &dir);
    std::env::remove_var("OXY_TOKEN");
    std::env::remove_var("OXY_CA");
    let mut repo = cloned.map_err(|e| {
        let _ = std::fs::remove_dir_all(&dir);
        anyhow::anyhow!("clone {}: {e}", claim.clone_url)
    })?;
    set_remote(&mut repo)?;
    Ok(repo)
}