//! The build cache (oxygenflow#28, spec §5.4): a disk mount that names a

//! `cache` is started from a disk kept by an earlier passing job, rather

//! than empty, so a build carries its compiled dependencies forward.

//!

//! Each runner keeps a pool under `<home>/cache`, by repository, job,

//! cache name and bookmark, each disk named by its key: the BLAKE3 of the

//! image, the job, the cache name and the content of its key files. A job

//! takes, in order:

//!

//! 1. its bookmark's disk with the same key (moved: no copying);

//! 2. its bookmark's newest disk (moved; cargo rebuilds what changed);

//! 3. another bookmark's disk with the same key, then its newest (copied,

//!    so a proposal never takes main's cache away);

//! 4. else an empty disk, as without a cache.

//!

//! A passing job's disk goes back into its bookmark's place; a failing

//! job's is thrown away, so nothing it wrote reaches a later job. The

//! receipt names what the job started from. The pool is kept under its

//! size limit by dropping the disks used longest ago.

//!

//! A disk is in the [`Form`] its tier mounts (oxygenflow#70): a microVM's

//! is an ext4 image, `<key>.ext4`; the syscall tier's is the host

//! directory it binds at the mount's target, `<key>.tree`, taken and kept

//! the same way. A directory goes into the pool with nothing in it

//! writable by others or set-id, and the pool itself is the runner's

//! alone.

//!

//! Beside each disk, `<key>.sources` (`<key>.tree-sources` for a

//! directory) lists the sources it was built from (oxygenflow#52): each

//! file's content hash and the mtime it had. A job that takes the disk

//! gives every file whose content is unchanged that mtime back, so a build

//! tool that judges freshness by time (cargo) rebuilds only what changed;

//! a changed file keeps the time of its checkout, newer than anything on

//! the disk, so it is always rebuilt.

use anyhow::{Context, Result};

use flow_core::pipeline::Mount;

use std::os::unix::fs::{MetadataExt, PermissionsExt};

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

use std::time::SystemTime;

/// The pool's default size limit.

pub const DEFAULT_GB: u64 = 64;

/// The sources a cached disk was built from: path to content hash and

/// mtime, in whole seconds (what the workspace image keeps).

pub type Sources = std::collections::BTreeMap<String, (String, i64)>;

/// What a job started a cached disk from.

pub struct Taken {

    /// For the log and the receipt.

    pub how: String,

    /// The sources it was built from, when they were kept with it.

    pub sources: Option<Sources>,

}

/// How a tier holds a disk mount's files (oxygenflow#70).

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

pub enum Form {

    /// An ext4 image of the mount's size, attached to a microVM.

    Image,

    /// A host directory, bound into the syscall sandbox.

    Dir,

}

impl Form {

    fn ext(self) -> &'static str {

        match self {

            Form::Image => "ext4",

            Form::Dir => "tree",

        }

    }

}

/// Which form a kept entry of the pool is in, by its name.

fn form_of(p: &Path) -> Option<Form> {

    match p.extension()?.to_str()? {

        "ext4" => Some(Form::Image),

        "tree" => Some(Form::Dir),

        _ => None,

    }

}

fn sources_of(disk: &Path) -> PathBuf {

    match form_of(disk) {

        Some(Form::Dir) => disk.with_extension("tree-sources"),

        _ => disk.with_extension("sources"),

    }

}

fn read_sources(disk: &Path) -> Option<Sources> {

    serde_json::from_slice(&std::fs::read(sources_of(disk)).ok()?).ok()

}

/// Where a job's cached disks come from and go back to.

pub struct Pool {

    root: PathBuf,

    limit: u64,

}

/// Which job: the pool's coordinates other than the key, and the form its

/// tier keeps a disk in.

pub struct Ctx<'a> {

    pub repo: &'a str,

    pub bookmark: &'a str,

    pub job: &'a str,

    pub image: &'a str,

    pub form: Form,

}

use flow_core::api::segment;

/// Held while the pool changes: jobs run side by side, and a directory,

/// unlike an image, can be half moved or half removed while another job

/// copies it.

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

fn changing() -> std::sync::MutexGuard<'static, ()> {

    CHANGING.lock().unwrap_or_else(|p| p.into_inner())

}

impl Pool {

    pub fn new(home_root: &Path, gb: Option<u64>) -> Pool {

        Pool { root: home_root.join("cache"), limit: gb.unwrap_or(DEFAULT_GB).saturating_mul(1 << 30) }

    }

    fn cache_dir(&self, ctx: &Ctx, name: &str) -> PathBuf {

        self.root.join(segment(ctx.repo)).join(segment(ctx.job)).join(segment(name))

    }

    /// A mount's key: the image, job, cache name and its key files'

    /// content hashes (`hash_of` gives a file's, `None` when absent).

    pub fn key(ctx: &Ctx, m: &Mount, hash_of: impl Fn(&str) -> Option<String>) -> String {

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

        for part in [ctx.image, ctx.job, m.cache.as_deref().unwrap_or("")] {

            h.update(part.as_bytes());

            h.update(b"\n");

        }

        for k in &m.key {

            h.update(k.as_bytes());

            h.update(b"=");

            h.update(hash_of(k).unwrap_or_else(|| "absent".into()).as_bytes());

            h.update(b"\n");

        }

        h.finalize().to_hex()[..32].to_string()

    }

    /// Start `dest` from the pool for mount `m`: an image of `m.size_mib`,

    /// or a directory, as `ctx.form` says; `dest` must not exist yet. What

    /// it started from, or `None` (the caller makes an empty disk).

    ///

    /// A disk this job's bookmark kept is moved out of the pool, so a job

    /// of the same slot running beside this one can never take it too.

    pub fn take(&self, ctx: &Ctx, m: &Mount, key: &str, dest: &Path) -> Option<Taken> {

        let name = m.cache.as_deref()?;

        let _changing = changing();

        let dir = self.cache_dir(ctx, name);

        let size = m.size_mib as u64 * (1 << 20);

        let form = ctx.form;

        let fits = |p: &Path| match form {

            Form::Image => std::fs::metadata(p).is_ok_and(|md| md.len() == size),

            Form::Dir => std::fs::symlink_metadata(p).is_ok_and(|md| md.is_dir()),

        };

        let ext = form.ext();

        let mine = dir.join(segment(ctx.bookmark));

        let exact = mine.join(format!("{key}.{ext}"));

        // A moved disk takes its sources with it.

        if fits(&exact) {

            let sources = read_sources(&exact);

            if std::fs::rename(&exact, dest).is_ok() {

                let _ = std::fs::remove_file(sources_of(&exact));

                return Some(Taken { how: format!("cache {name}: this bookmark's, same key"), sources });

            }

        }

        if let Some(p) = newest(&mine, form, |p| fits(p)) {

            let sources = read_sources(&p);

            if std::fs::rename(&p, dest).is_ok() {

                let _ = std::fs::remove_file(sources_of(&p));

                return Some(Taken { how: format!("cache {name}: this bookmark's, an older key"), sources });

            }

        }

        let others: Vec<PathBuf> =

            std::fs::read_dir(&dir).into_iter().flatten().flatten().map(|e| e.path()).filter(|p| p.is_dir() && *p != mine).collect();

        let pick = others

            .iter()

            .map(|d| d.join(format!("{key}.{ext}")))

            .filter(|p| fits(p))

            .max_by_key(|p| modified(p))

            .map(|p| (p, "same key"))

            .or_else(|| {

                others.iter().filter_map(|d| newest(d, form, |p| fits(p))).max_by_key(|p| modified(p)).map(|p| (p, "an older key"))

            });

        if let Some((p, which)) = pick {

            let from = p.parent().and_then(|d| d.file_name()).map(|n| n.to_string_lossy().into_owned()).unwrap_or_default();

            if copy(form, &p, dest).is_ok() {

                return Some(Taken {

                    how: format!("cache {name}: copied from another bookmark's ({from}), {which}"),

                    sources: read_sources(&p),

                });

            }

        }

        None

    }

    /// Keep a passing job's disk for mount `m` under its key, then keep

    /// the pool under its limit.

    pub fn keep(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources) -> Result<()> {

        let Some(name) = m.cache.as_deref() else { return Ok(()) };

        let _changing = changing();

        // Whatever the job left, nothing in the pool is anyone else's to

        // write or runs as the runner (oxygenflow#70).

        if ctx.form == Form::Dir {

            seal(disk).with_context(|| format!("seal {}", disk.display()))?;

        }

        let mine = self.cache_dir(ctx, name).join(segment(ctx.bookmark));

        std::os::unix::fs::DirBuilderExt::mode(std::fs::DirBuilder::new().recursive(true), 0o700).create(&mine)?;

        std::fs::set_permissions(&self.root, std::fs::Permissions::from_mode(0o700))?;

        let dest = mine.join(format!("{key}.{}", ctx.form.ext()));

        if ctx.form == Form::Dir {

            // Another job of this slot kept one since: this one is newer.

            crate::workspace::remove_tree(&dest)?;

        }

        if std::fs::rename(disk, &dest).is_err() {

            copy(ctx.form, disk, &dest).with_context(|| format!("keep {}", disk.display()))?;

        }

        // Moved disks keep their old time: mark it used now.

        std::fs::File::open(&dest)?.set_modified(SystemTime::now())?;

        let sidecar = sources_of(&dest);

        let tmp = sidecar.with_extension("sources-part");

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

        {

            use std::io::Write;

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

            std::fs::OpenOptions::new().write(true).create_new(true).mode(0o600).open(&tmp)?.write_all(&serde_json::to_vec(sources)?)?;

        }

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

        // One disk per key is enough; older keys of this bookmark go once

        // the pool is over its limit.

        self.evict();

        Ok(())

    }

    /// After the job: a passing job's disk is kept; a failing one's is

    /// thrown away, so nothing it wrote reaches a later job.

    pub fn settle(&self, ctx: &Ctx, m: &Mount, key: &str, disk: &Path, sources: &Sources, passed: bool) -> Result<()> {

        if passed {

            return self.keep(ctx, m, key, disk, sources);

        }

        let gone = match ctx.form {

            Form::Dir => crate::workspace::remove_tree(disk),

            Form::Image => std::fs::remove_file(disk).or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) }),

        };

        gone.with_context(|| format!("discard {}", disk.display()))

    }

    /// The repository and job slots holding at least one disk, as

    /// [`flow_core::api::cache_slot`] names them, for the claim

    /// (oxygenflow#51), each with when its newest disk was kept (Unix ms):

    /// a disk's time is set when it is kept, so the newest is the freshest

    /// build (oxygenflow#54). Images and directories alike, so a job goes

    /// where its cache is whichever tier keeps it.

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

        let mut out = std::collections::BTreeMap::new();

        for repo in std::fs::read_dir(&self.root).into_iter().flatten().flatten() {

            for job in std::fs::read_dir(repo.path()).into_iter().flatten().flatten() {

                let mut newest: Option<SystemTime> = None;

                // `<job>/<cache>/<bookmark>/<disk>`

                kept(&job.path(), 2, &mut |p, _| {

                    let t = modified(p);

                    newest = Some(newest.map_or(t, |n| n.max(t)));

                });

                if let Some(t) = newest {

                    let ms = t.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);

                    out.insert(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()), ms);

                }

            }

        }

        out

    }

    /// Drop the disks used longest ago until the pool's blocks fit: an

    /// image's blocks, or everything in a directory. The caller holds

    /// [`changing`].

    fn evict(&self) -> u64 {

        let mut disks: Vec<(SystemTime, u64, PathBuf)> = Vec::new();

        // `<repo>/<job>/<cache>/<bookmark>/<disk>`

        kept(&self.root, 4, &mut |p, form| {

            if let Ok(md) = std::fs::symlink_metadata(p) {

                let bytes = match form {

                    Form::Image => md.blocks() * 512,

                    Form::Dir => tree_bytes(p),

                };

                disks.push((md.modified().unwrap_or(SystemTime::UNIX_EPOCH), bytes, p.to_path_buf()));

            }

        });

        let mut total: u64 = disks.iter().map(|d| d.1).sum();

        disks.sort();

        let mut freed = 0;

        for (_, bytes, p) in disks {

            if total <= self.limit {

                break;

            }

            let gone = match form_of(&p) {

                Some(Form::Dir) => crate::workspace::remove_tree(&p),

                _ => std::fs::remove_file(&p),

            };

            if gone.is_ok() {

                let _ = std::fs::remove_file(sources_of(&p));

                total -= bytes;

                freed += bytes;

            }

        }

        freed

    }

}

/// Where a job's `n`th disk mount is in its scratch directory, in the form

/// its tier mounts it: `microvm::run` attaches `mount-N.ext4`, and

/// `syscall::run` binds the directory `mount-N`.

pub fn disk_path(form: Form, scratch: &Path, n: usize) -> PathBuf {

    match form {

        Form::Image => scratch.join(format!("mount-{n}.ext4")),

        Form::Dir => crate::syscall::disk_dir(scratch, n),

    }

}

fn modified(p: &Path) -> SystemTime {

    std::fs::symlink_metadata(p).and_then(|m| m.modified()).unwrap_or(SystemTime::UNIX_EPOCH)

}

fn newest(dir: &Path, form: Form, ok: impl Fn(&Path) -> bool) -> Option<PathBuf> {

    std::fs::read_dir(dir).ok()?.flatten().map(|e| e.path()).filter(|p| form_of(p) == Some(form) && ok(p)).max_by_key(|p| modified(p))

}

/// Each kept image or directory `depth` levels of directories below `dir`,

/// with its form; never inside a kept directory.

fn kept(dir: &Path, depth: usize, f: &mut dyn FnMut(&Path, Form)) {

    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {

        let p = e.path();

        let Ok(t) = e.file_type() else { continue };

        if depth == 0 {

            match form_of(&p) {

                Some(Form::Image) if t.is_file() => f(&p, Form::Image),

                Some(Form::Dir) if t.is_dir() => f(&p, Form::Dir),

                _ => {}

            }

        } else if t.is_dir() {

            kept(&p, depth - 1, f);

        }

    }

}

/// The blocks a directory's files take, each file once however many links

/// it has; links are not followed.

fn tree_bytes(dir: &Path) -> u64 {

    fn walk(p: &Path, seen: &mut std::collections::HashSet<(u64, u64)>) -> u64 {

        let Ok(md) = std::fs::symlink_metadata(p) else { return 0 };

        if md.nlink() > 1 && !md.is_dir() && !seen.insert((md.dev(), md.ino())) {

            return 0;

        }

        let mut bytes = md.blocks() * 512;

        if md.is_dir() {

            for e in std::fs::read_dir(p).into_iter().flatten().flatten() {

                bytes += walk(&e.path(), seen);

            }

        }

        bytes

    }

    walk(dir, &mut Default::default())

}

/// Make a directory a job left fit for the pool: nothing in it writable by

/// group or others, nothing set-id or sticky, and everything readable by

/// the runner (directories searchable and writable too, so it can be

/// copied and removed). Links are not followed. The job ran as the

/// runner's user, so all of it is the runner's to change; times are kept.

fn seal(p: &Path) -> std::io::Result<()> {

    let md = std::fs::symlink_metadata(p)?;

    if md.file_type().is_symlink() {

        return Ok(());

    }

    let mode = md.mode() & 0o7777;

    let want = (mode & !0o7022) | if md.is_dir() { 0o700 } else { 0o400 };

    if want != mode {

        std::fs::set_permissions(p, std::fs::Permissions::from_mode(want))?;

    }

    if md.is_dir() {

        for e in std::fs::read_dir(p)? {

            seal(&e?.path())?;

        }

    }

    Ok(())

}

fn copy(form: Form, from: &Path, to: &Path) -> Result<()> {

    match form {

        Form::Image => sparse_copy(from, to),

        Form::Dir => tree_copy(from, to),

    }

}

/// Copy a sparse disk without writing its holes.

fn sparse_copy(from: &Path, to: &Path) -> Result<()> {

    let tmp = to.with_extension("part");

    let st = std::process::Command::new("cp").arg("--sparse=always").arg(from).arg(&tmp).status().context("run cp")?;

    if !st.success() {

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

        anyhow::bail!("cp {} failed", from.display());

    }

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

    Ok(())

}

/// Copy a directory with its modes and times (a build tool judges by

/// them), sharing blocks where the filesystem can.

fn tree_copy(from: &Path, to: &Path) -> Result<()> {

    let tmp = to.with_extension("tree-part");

    crate::workspace::remove_tree(&tmp)?;

    let st = std::process::Command::new("cp")

        .args(["-a", "--reflink=auto", "--no-target-directory"])

        .arg(from)

        .arg(&tmp)

        .status()

        .context("run cp")?;

    if !st.success() {

        let _ = crate::workspace::remove_tree(&tmp);

        anyhow::bail!("cp {} failed", from.display());

    }