//! 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.

//!

//! A passing job's disk is pruned before it is kept (oxygenflow#72; see

//! [`flow_init::prune`]): over [`flow_init::prune::DEFAULT_PERCENT`] of

//! its size (the runner's `cache_prune_percent`), the files the job did

//! not use go, oldest first, and a cache still over it is not kept, so a

//! kept target never grows until the job's disk is full. A microVM's

//! guest prunes its image before it stops (the runner cannot look inside

//! an ext4 image without mounting it); the runner prunes a directory

//! itself, having set its files' atimes to the epoch before the job

//! ([`forget_reads`]) so that every read is seen. The job's log says what

//! was kept and pruned.

use anyhow::{Context, Result};

use flow_core::pipeline::Mount;

use flow_init::prune::Pruned;

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

}

/// How a passing job's cache disk is pruned before it is kept

/// (oxygenflow#72).

pub struct Prune<'a> {

    /// When the job started ([`flow_init::prune::start_clock`]): what it

    /// read or wrote since is what it used.

    pub started: SystemTime,

    /// The share of the disk the kept cache may fill.

    pub percent: u8,

    /// What the guest did to a microVM's image, when it said.

    pub guest: Option<&'a Pruned>,

}

/// 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));

        crate::portable::create_private_dir(&mine)?;

        crate::portable::set_mode(&self.root, 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;

            crate::portable::private(std::fs::OpenOptions::new().write(true).create_new(true))

                .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 pruned and kept, unless it

    /// is still over its cap; a failing one's is thrown away, so nothing

    /// it wrote reaches a later job. For a kept or refused cache, what to

    /// say in the job's log.

    #[allow(clippy::too_many_arguments)]

    pub fn settle(

        &self,

        ctx: &Ctx,

        m: &Mount,

        key: &str,

        disk: &Path,

        sources: &Sources,

        passed: bool,

        prune: &Prune,

    ) -> Result<Option<String>> {

        if passed {

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

            let pruned = pruned(ctx.form, m, disk, prune)?;

            let line = describe(name, &pruned);

            if pruned.fits() {

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

            } else {

                discard(ctx.form, disk)?;

            }

            return Ok(Some(line));

        }

        discard(ctx.form, disk)?;

        Ok(None)

    }

    /// 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 => crate::portable::disk_bytes(&md),

                    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

    }

}

/// Throw a disk away.

fn discard(form: Form, disk: &Path) -> Result<()> {

    let gone = match 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()))

}

/// Before a syscall job: set the atime of every file of a directory it

/// took from the pool to the epoch, so that whatever the file system's

/// atime rule, the job's first read of each is seen when the directory is

/// pruned (oxygenflow#72). A microVM's guest mounts its disk

/// `strictatime` instead.

pub fn forget_reads(form: Form, disk: &Path) -> Result<()> {

    if form == Form::Dir && disk.is_dir() {

        flow_init::prune::forget_reads(disk).with_context(|| format!("set the atimes in {}", disk.display()))?;

    }

    Ok(())

}

/// Prune a passing job's disk, or say why it was not: a directory here,

/// an image by what its guest said (else only its size is measured).

fn pruned(form: Form, m: &Mount, disk: &Path, prune: &Prune) -> Result<Pruned> {

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

    let percent = prune.percent;

    let unpruned = |before: u64, why: String| Pruned { size, percent, before, after: before, unpruned: Some(why), ..Pruned::default() };

    Ok(match form {

        Form::Dir => {

            let before = tree_bytes(disk);

            if !flow_init::prune::records_reads(disk) {

                unpruned(before, "this runner's file system records no reads (noatime), so what the job used cannot be told".into())

            } else {

                flow_init::prune::prune(disk, prune.started, &[], size, before, percent)

                    .unwrap_or_else(|e| unpruned(tree_bytes(disk), format!("pruning failed: {e}")))

            }

        }

        Form::Image => match prune.guest.filter(|g| g.size > 0) {

            Some(g) => g.clone(),

            None => {

                let (before, size) = image_usage(disk)?;

                let why = prune.guest.and_then(|g| g.unpruned.clone()).unwrap_or_else(|| "the guest did not prune it".into());

                Pruned { size, ..unpruned(before, why) }

            }

        },

    })

}

/// What an ext4 image holds and its size, as its superblock counts them.

fn image_usage(image: &Path) -> Result<(u64, u64)> {

    let out = std::process::Command::new("dumpe2fs").arg("-h").arg(image).output().context("run dumpe2fs (e2fsprogs)")?;

    if !out.status.success() {

        anyhow::bail!("dumpe2fs {}: {}", image.display(), String::from_utf8_lossy(&out.stderr).trim());

    }

    superblock_usage(&String::from_utf8_lossy(&out.stdout)).with_context(|| format!("dumpe2fs {}: no block counts", image.display()))

}

fn superblock_usage(header: &str) -> Option<(u64, u64)> {

    let field = |name: &str| -> Option<u64> {

        header.lines().find_map(|l| l.strip_prefix(name)?.strip_prefix(':')).and_then(|v| v.trim().parse().ok())

    };

    let (count, free, block) = (field("Block count")?, field("Free blocks")?, field("Block size")?);

    Some((count.saturating_sub(free) * block, count * block))

}

/// The job log's line for a passing job's cache: what was kept, or why

/// not (oxygenflow#72).

pub fn describe(name: &str, p: &Pruned) -> String {

    use crate::images::human;

    let cap = format!("{}% of its {} disk", p.percent, human(p.size));

    let pruned = format!("{} in {} file{} this run did not use", human(p.freed), p.files, if p.files == 1 { "" } else { "s" });

    if p.fits() && p.files > 0 {

        format!("cache {name}: kept {} of {} (pruned {pruned}; the cap is {cap})", human(p.after), human(p.before))

    } else if p.fits() {

        format!("cache {name}: kept {}, under its cap of {cap}", human(p.after))

    } else {

        let after = if p.files > 0 { format!("{} after pruning {pruned}", human(p.after)) } else { human(p.after) };