//! `$FLOW_HOME` (default `~/.local/share/flow`): the pinned parts of each

//! isolation tier, rootfs images by hash, scratch space and the runner's

//! configuration, all described by `flow.toml`.

//!

//! ```toml

//! [microvm]

//! firecracker = { path = "bin/firecracker", blake3 = "…" }

//! kernel = { path = "kernel/vmlinux", blake3 = "…" }

//! init = { path = "bin/flow-init", blake3 = "…" }

//!

//! [runner]                       # only for `flow runner`

//! hub = "https://hub.example:7343"

//! token = "…"                    # a runner token from the hub's [flow] runner_tokens

//! ca = "/path/to/hub-ca.pem"     # when the hub's certificate is self-signed

//! name = "earth-1"

//! ```

//!

//! A runner signs its receipts with a composite key kept beside `flow.toml`

//! (`runner.key`, `runner.identity`), made on first start.

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

use serde::Deserialize;

use std::collections::HashMap;

use std::path::{Path, PathBuf};

use std::sync::Mutex;

#[derive(Clone, Debug, Deserialize)]

pub struct Pinned {

    pub path: PathBuf,

    pub blake3: String,

}

#[derive(Clone, Debug, Deserialize)]

pub struct MicrovmConfig {

    pub firecracker: Pinned,

    pub kernel: Pinned,

    pub init: Pinned,

}

/// Where `flow runner register` keeps the runner's hub and credential.

pub const RUNNER_TOML: &str = "runner.toml";

/// The hub's CA certificate as `flow runner register --ca` keeps it (#98).

pub const CA_PEM: &str = "hub-ca.pem";

#[derive(Clone, Debug, Deserialize)]

pub struct RunnerConfig {

    pub hub: String,

    pub token: String,

    pub ca: Option<PathBuf>,

    pub name: Option<String>,

    /// Jobs run at once (default 1).

    #[serde(default)]

    pub max_jobs: Option<usize>,

    /// Labels the runner offers, e.g. `gpu`, `big-disk`.

    #[serde(default)]

    pub labels: Vec<String>,

    /// Memory (MiB) and vCPUs the jobs running at once may use together.

    #[serde(default)]

    pub memory_mib: Option<u32>,

    #[serde(default)]

    pub cpus: Option<u32>,

    /// Sharing images with other runners (docs/jetstream.md §8.1): absent,

    /// the runner neither serves nor asks peers.

    #[serde(default)]

    pub swarm: Option<SwarmConfig>,

    /// The build cache's size limit in GiB (#28; default 64).

    #[serde(default)]

    pub cache_gb: Option<u64>,

    /// The share of its disk, in percent, a kept build cache may fill

    /// (oxygenflow#72; default 60): over it, the files the job did not use

    /// are pruned, oldest first, and a cache still over it is not kept.

    #[serde(default)]

    pub cache_prune_percent: Option<u8>,

    /// Let microVM jobs reach the hosts their sandbox allows, through this

    /// runner (oxygenflow#55); off unless set. The runner then offers the

    /// `egress` label, which such jobs need.

    #[serde(default)]

    pub egress: bool,

    /// Sign each job's receipt with a fresh key the hub certifies for that

    /// job, just before signing (keyless, oxygenhub#241), rather than with

    /// the runner's own key; off unless set. When the hub cannot certify,

    /// the runner's key signs, and the job's log says why.

    #[serde(default)]

    pub keyless: bool,

    /// Serve this runner's metrics at this address, e.g. `127.0.0.1:9464`

    /// (oxygenflow#35); off unless set.

    #[serde(default)]

    pub metrics: Option<String>,

    /// A bearer token a scrape of the metrics must send.

    #[serde(default)]

    pub metrics_token: Option<String>,

    /// `[runner.update]`: whether this runner takes updates (#85).

    #[serde(default)]

    pub update: Option<UpdateConfig>,

}

/// `[runner.update]` in flow.toml (docs/jetstream.md §8.2): the runner's

/// last word on updates, whatever the hub's policy says.

#[derive(Clone, Debug, Deserialize)]

pub struct UpdateConfig {

    /// `false` refuses every update, so a site can hold a machine back.

    #[serde(default = "yes")]

    pub allow: bool,

}

fn yes() -> bool {

    true

}

impl RunnerConfig {

    /// The labels offered to the hub: the configured ones, and `egress`

    /// when egress is on.

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

        let mut l = self.labels.clone();

        if self.egress && !l.iter().any(|x| x == flow_core::EGRESS_LABEL) {

            l.push(flow_core::EGRESS_LABEL.to_string());

        }

        l

    }

}

/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.

#[derive(Clone, Debug, Deserialize)]

pub struct SwarmConfig {

    /// Where peers reach this runner, e.g. `0.0.0.0:7401`.

    pub listen: String,

    /// The address peers are given (default: `listen`).

    #[serde(default)]

    pub advertise: Option<String>,

    /// Peers are only ever in the same site.

    #[serde(default)]

    pub site: String,

    /// The most this runner sends to peers, MiB a second (default 50).

    #[serde(default)]

    pub upload_mib_s: Option<u32>,

}

#[derive(Clone, Debug, Default, Deserialize)]

pub struct Config {

    pub microvm: Option<MicrovmConfig>,

    pub runner: Option<RunnerConfig>,

}

pub struct Home {

    pub root: PathBuf,

    pub config: Config,

    /// Files already verified, by path, with the (len, mtime) they had then.

    verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,

}

impl Home {

    pub fn locate(explicit: Option<&Path>) -> Result<Home> {

        let root = match explicit {

            Some(p) => p.to_path_buf(),

            None => match std::env::var_os("FLOW_HOME") {

                Some(p) => PathBuf::from(p),

                None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),

            },

        };

        let file = root.join("flow.toml");

        let mut config: Config = if file.is_file() {

            let text = std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?;

            toml::from_str(&text).with_context(|| format!("parse {}", file.display()))?

        } else {

            Config::default()

        };

        // `flow runner register` writes the runner's hub and credential to

        // runner.toml, which stands in for a [runner] section flow.toml lacks.

        let registered = root.join(RUNNER_TOML);

        if config.runner.is_none() && registered.is_file() {

            let text = std::fs::read_to_string(&registered).with_context(|| format!("read {}", registered.display()))?;

            let mut runner: RunnerConfig = toml::from_str(&text).with_context(|| format!("parse {}", registered.display()))?;

            // A relative CA there is the home's (#98): register keeps its

            // copy there, and the service may start in any directory.

            runner.ca = runner.ca.map(|ca| if ca.is_absolute() { ca } else { root.join(ca) });

            config.runner = Some(runner);

        }

        for d in ["cas", "work", "repos"] {

            std::fs::create_dir_all(root.join(d)).with_context(|| format!("create {}", root.join(d).display()))?;

        }

        Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })

    }

    pub fn resolve(&self, p: &Path) -> PathBuf {

        if p.is_absolute() {

            p.to_path_buf()

        } else {

            self.root.join(p)

        }

    }

    /// Where a rootfs image with this hash lives.

    pub fn rootfs(&self, hash: &str) -> PathBuf {

        self.root.join("cas").join(format!("{hash}.ext4"))

    }

    /// An image's manifest (`cas/<hash>.json`): what `flow rootfs import`

    /// found out about it.

    pub fn image_manifest(&self, hash: &str) -> PathBuf {

        self.root.join("cas").join(format!("{hash}.json"))

    }

    /// The architecture an image is for: as its manifest records, else

    /// read from the image's own shell (spec §11.2) and recorded, so an

    /// image fetched or imported without one is read once and a runner

    /// declines what it cannot run instead of failing it. `None` when the

    /// image is not here or names no architecture.

    pub fn image_arch(&self, hash: &str) -> Option<String> {

        let path = self.image_manifest(hash);

        let mut v: serde_json::Value =

            std::fs::read_to_string(&path).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_else(|| serde_json::json!({}));

        if let Some(a) = v["arch"].as_str() {

            return Some(a.to_string());

        }

        let image = self.rootfs(hash);

        if v.get("arch_read_ms").is_some() || !image.is_file() {

            return None;

        }

        let arch = jet::rootfs::arch_of_file(&image);

        let o = v.as_object_mut()?;

        if let Some(a) = arch {

            o.insert("arch".into(), a.into());

        }

        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);

        o.insert("arch_read_ms".into(), now.into());

        let tmp = path.with_extension("json.part");

        if std::fs::write(&tmp, v.to_string()).is_ok() {

            let _ = std::fs::rename(&tmp, &path);

        }

        arch.map(String::from)

    }

    /// Every image with a known architecture (spec §11.2).

    pub fn image_archs(&self) -> std::collections::BTreeMap<String, String> {

        self.images().into_iter().filter_map(|h| self.image_arch(&h).map(|a| (h, a))).collect()

    }

    /// The rootfs images this home holds: the hashes of `cas/*.ext4`.

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

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

            return Vec::new();

        };

        let mut out: Vec<String> = dir

            .filter_map(|e| {

                let name = e.ok()?.file_name().into_string().ok()?;

                let hash = name.strip_suffix(".ext4")?;

                (hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit())).then(|| hash.to_string())

            })

            .collect();

        out.sort();

        out

    }

    /// A pinned file, checked against its BLAKE3 before use (again only

    /// when its size or modification time changed since).

    pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {

        let meta = std::fs::metadata(path).with_context(|| format!("{} is missing", path.display()))?;

        let stamp = (meta.len(), meta.modified()?);

        if self.verified.lock().unwrap().get(path) == Some(&stamp) {

            return Ok(path.to_path_buf());

        }

        let mut got = hash_file(path)?;

        if got != blake3_hex {

            // The read may have come from bad cached pages rather than the

            // disk (oxygenflow#22: a RAM fault on Earth failed every job for

            // hours while the file was intact): drop the file's pages and

            // read it again from the disk before giving up.

            drop_cached(path);

            let again = hash_file(path)?;

            if again == blake3_hex {

                eprintln!(

                    "warning: {} hashed {got} from the page cache but {again} from the disk: the cached copy was bad (check this machine's memory)",

                    path.display()

                );

            }

            got = again;

        }

        if got != blake3_hex {

            bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());

        }

        self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);

        Ok(path.to_path_buf())

    }

}

/// Ask the kernel to forget a file's cached pages, so the next read comes

/// from the disk. Best effort: dirty pages and other platforms stay.

fn drop_cached(path: &Path) {

    #[cfg(target_os = "linux")]

    if let Ok(f) = std::fs::File::open(path) {

        use std::os::fd::AsRawFd;

        // SAFETY: a valid descriptor; the advice only affects caching.

        unsafe {

            libc::posix_fadvise(f.as_raw_fd(), 0, 0, libc::POSIX_FADV_DONTNEED);

        }

    }

    #[cfg(not(target_os = "linux"))]

    let _ = path;

}

pub fn hash_file(path: &Path) -> Result<String> {

    let mut h = blake3::Hasher::new();

    h.update_reader(std::fs::File::open(path).with_context(|| format!("open {}", path.display()))?)?;

    Ok(h.finalize().to_hex().to_string())

}

/// The key a runner signs receipts with (Oxygen spec §14.6).

pub struct RunnerKey {

    pub identity: oxy_core::identity::Identity,

    pub secret: oxy_core::identity::SecretKey,

}

impl Home {

    /// The runner's key, made on first use as `runner.key` (owner-only)

    /// and `runner.identity`, named `name`.

    pub fn runner_key(&self, name: &str) -> Result<RunnerKey> {

        use oxy_core::identity::{Identity, SecretKey};

        let key_path = self.root.join("runner.key");

        let id_path = self.root.join("runner.identity");

        if key_path.exists() {

            let (key, secret) = SecretKey::decode(&std::fs::read(&key_path)?).context("runner.key")?;

            let identity = Identity::decode(&std::fs::read(&id_path)?).context("runner.identity")?;

            if identity.hash() != key {

                bail!("runner.identity does not belong to runner.key");

            }

            return Ok(RunnerKey { identity, secret });

        }

        let secret = SecretKey::generate();

        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);

        let identity = secret.identity(&format!("{name} (OxygenFlow runner)"), "", now);

        std::fs::write(&id_path, identity.encode())?;

        let tmp = self.root.join("runner.key.tmp");

        {

            use std::io::Write;

            let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;

            f.write_all(&secret.encode(&identity.hash()))?;

            f.sync_all()?;

        }

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

        Ok(RunnerKey { identity, secret })

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn the_runner_key_is_made_once_and_kept_private() {

        use std::os::unix::fs::PermissionsExt;

        let dir = tempfile::tempdir().unwrap();

        let home = Home::locate(Some(dir.path())).unwrap();

        let a = home.runner_key("earth-1").unwrap();

        assert_eq!(a.identity.name, "earth-1 (OxygenFlow runner)");

        let mode = std::fs::metadata(dir.path().join("runner.key")).unwrap().permissions().mode();

        assert_eq!(mode & 0o777, 0o600);

        let b = home.runner_key("renamed").unwrap();

        assert_eq!(a.identity, b.identity, "the same key after a restart");

        let sig = b.secret.sign(oxy_core::identity::Domain::Attest, &b.identity.hash(), &oxy_core::hash::ObjectRef([1; 32]));

        oxy_core::identity::verify(&a.identity, &sig, oxy_core::identity::Domain::Attest, &oxy_core::hash::ObjectRef([1; 32])).unwrap();

    }

    /// #98: runner.toml's relative CA is the home's, wherever flow starts.

    #[test]

    fn a_registered_runners_relative_ca_is_in_its_home() {

        let dir = tempfile::tempdir().unwrap();

        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"hub-ca.pem\"\n").unwrap();

        let home = Home::locate(Some(dir.path())).unwrap();

        assert_eq!(home.config.runner.unwrap().ca, Some(dir.path().join("hub-ca.pem")));

        std::fs::write(dir.path().join(RUNNER_TOML), "hub = \"https://hub\"\ntoken = \"t\"\nca = \"/etc/ssl/hub.pem\"\n").unwrap();

        let home = Home::locate(Some(dir.path())).unwrap();

        assert_eq!(home.config.runner.unwrap().ca, Some(PathBuf::from("/etc/ssl/hub.pem")));

    }

}