//! What workflows keep from run to run: for each path a workflow names
//! under `cache:`, a base, the directory its runs start from, which every
//! run sees through an overlay of its own, so runs of one workflow go side
//! by side rather than taking turns at it.
//!
//! A run's overlay has the base as its lower layer, read only, and a
//! directory of the run's own as its upper one, which takes all it writes.
//! Only a run of a push to the default branch that passes keeps what it
//! wrote: its upper layer goes into a new base, which the runs started
//! after it start from. Every other run's upper layer, a pull's among
//! them, goes when the run ends, so what a pull writes never reaches the
//! base, and a pull starts from what the default branch last left.
//!
//! A base is never changed while a run may read it: a new one is made
//! beside it, its files hard links to the old one's where the upper layer
//! left them as they were, and the old one goes once the last run reading
//! it ends. The overlays are mounted with `userxattr`, so what says a
//! directory in an upper layer hides the one below it is an extended
//! attribute the runner can read without privileges.
//!
//! A trusted pull's run that passes, which a push to the default branch
//! at the same tree may take as its own rather than run again, keeps its
//! upper layers instead, under the key of its check: the push that takes
//! the run takes its layers too, into new bases as its own run's would
//! go, but only over the base each was written over. Applied over another,
//! its whiteouts and what it left as it was would be of what that base
//! held, not what this one does, so a layer over a base that has moved on
//! is not kept, and the next run of the default branch warms it. Kept
//! layers go once taken, once their pull has moved on ([`Caches::forget`]),
//! and past [`KEPT_AGE`], [`KEPT_MAX`] of them or a budget of bytes,
//! the oldest first. Only a run the caller says to keep is kept: an
//! untrusted pull's never is, so its writes never reach a base.
//!
//! Where no push to the default branch runs, as on a runner that runs a
//! repository's pulls alone, a trusted pull's run that passes makes the
//! base instead ([`Lease::advance`]), as a push's would, so its pulls do
//! not all start cold: but only over the base each layer was written
//! over, still current, as a kept layer is taken, and only when the tip
//! of the branch the pull merged with is the one that base was made at or
//! newer, so the base never goes back to an older tip. Each base made so
//! is marked with that tip's generation, in [`MARKS`].
//!
//! The disk the caches are on is shared, with the site on the same host
//! among others, so the runner keeps a floor of it free: below it, before
//! a run starts, [`Caches::shed`] removes every kept run's layers and
//! every base no run reads, and the runs after start afresh.
//!
//! Under the root: `bases/<cache>/<generation>`, `runs/<run>/<n>/` with
//! each layer's `upper` and the `work` directory overlayfs needs beside
//! it, `kept/<run>/`, a kept run's [`KEPT`] and its `<n>/upper`s, and
//! [`MARKS`].

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeSet, HashMap, HashSet};
use std::fs::Metadata;
use std::io;
use std::os::unix::fs::{FileTypeExt, MetadataExt};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use crate::executor;
use knotless::keys;

/// The attribute overlayfs, mounted with `userxattr`, marks a directory in
/// an upper layer with when it hides the lower layer's: one deleted and
/// made again.
const OPAQUE: &str = "user.overlay.opaque";

/// What a kept run's directory says of it.
pub const KEPT: &str = "kept.json";

/// The generation of the branch's tip each cache's base a pull made was
/// made at, as `{cache: [base generation, tip generation]}`.
pub const MARKS: &str = "marks.json";

/// How long a passing pull run's layers are kept for a push to take, in
/// seconds: three days.
pub const KEPT_AGE: u64 = 3 * 24 * 60 * 60;

/// The most runs whose layers are kept at once.
pub const KEPT_MAX: usize = 16;

/// Says the bytes free on the disk holding a path.
pub type Space = Box<dyn Fn(&Path) -> io::Result<u64> + Send + Sync>;

pub struct Caches {
    root: PathBuf,
    /// The bytes free on the disk the caches are on: [`free`], but in tests.
    space: Space,
    /// This runner's start, so its runs' names are not an earlier one's.
    boot: u128,
    runs: AtomicU64,
    bases: Mutex<Bases>,
    /// Held while kept layers are made, taken or removed.
    kept: Mutex<()>,
}

/// A kept run's layers, when it was kept, and the bytes they take.
#[derive(Serialize, Deserialize, Default)]
struct Kept {
    at: u64,
    size: u64,
    layers: Vec<KeptLayer>,
}

/// A layer kept: its cache, where the run saw it, and the generation of
/// the base it was written over.
#[derive(Serialize, Deserialize)]
struct KeptLayer {
    name: String,
    path: String,
    generation: u64,
}

/// A kept run's layers as a push takes them: a lease over the bases they
/// were written over, of those still current, for [`Lease::promote`]; and
/// where the run saw those over a base that has moved on, which are not.
pub struct Taken {
    pub lease: Lease,
    pub moved: Vec<String>,
}

impl Taken {
    /// Where the run saw each layer that is taken.
    pub fn kept(&self) -> Vec<String> {
        self.lease.layers.iter().map(|l| l.path.clone()).collect()
    }
}

/// The generations of each cache's base: the one runs start from, those
/// still on disk, and how many runs read each; and of the bases a pull
/// made, the generation of the branch's tip each was made at.
#[derive(Default)]
struct Bases {
    current: HashMap<String, u64>,
    kept: HashMap<String, BTreeSet<u64>>,
    readers: HashMap<(String, u64), usize>,
    next: u64,
    marks: HashMap<String, (u64, u64)>,
}

impl Bases {
    /// Whether a layer written over `name`'s base `generation` by a pull
    /// merged with the branch's tip at generation `at` may make the next
    /// base: that base is still current, and was made at that tip or an
    /// older one, or by a push, which marks none.
    fn ahead(&self, name: &str, generation: u64, at: u64) -> bool {
        let mark = self.marks.get(name).filter(|(g, _)| *g == generation);
        self.current.get(name) == Some(&generation) && mark.is_none_or(|(_, tip)| *tip <= at)
    }

    /// The generations of `name` no run reads and none will start from, no
    /// longer counted as on disk.
    fn unused(&mut self, name: &str) -> Vec<u64> {
        let current = self.current.get(name).copied();
        let readers = &self.readers;
        let kept = self.kept.entry(name.to_string()).or_default();
        let unused: Vec<u64> = kept
            .iter()
            .copied()
            .filter(|g| Some(*g) != current && !readers.contains_key(&(name.to_string(), *g)))
            .collect();
        kept.retain(|g| !unused.contains(g));
        unused
    }
}

/// A cache as a run sees it: the base it starts from and the layer it
/// writes to.
#[derive(Clone, Debug)]
pub struct Layer {
    /// The cache, as its bases are named.
    pub name: String,
    /// Where the run's container sees it.
    pub path: String,
    generation: u64,
    pub lower: PathBuf,
    pub upper: PathBuf,
    pub work: PathBuf,
}

/// A run's layers over the bases it reads, which stay until it ends; its
/// own layers go then.
pub struct Lease {
    caches: Arc<Caches>,
    pub id: String,
    dir: PathBuf,
    pub layers: Vec<Layer>,
}

impl Caches {
    /// The caches under `root`, with what runs of an earlier runner left
    /// there gone, but for the layers of the runs `alive` names, whose
    /// overlays are still mounted, and the bases those may read; and the
    /// layers kept of the runs `kept` names, whose checks a push may yet
    /// take, and of no other.
    pub fn open(root: &Path, alive: &[String], kept: &[String]) -> Result<Arc<Caches>> {
        Caches::open_with(root, alive, kept, Box::new(free))
    }

    /// [`Caches::open`], told the bytes free on its disk by `space`.
    pub fn open_with(
        root: &Path,
        alive: &[String],
        kept: &[String],
        space: Space,
    ) -> Result<Arc<Caches>> {
        let (bases_dir, runs_dir) = (root.join("bases"), root.join("runs"));
        std::fs::create_dir_all(&bases_dir).context("making the caches' directory")?;
        std::fs::create_dir_all(&runs_dir)?;
        for run in std::fs::read_dir(&runs_dir)? {
            let run = run?;
            if !alive.iter().any(|a| run.file_name() == a.as_str()) {
                remove(&run.path())?;
            }
        }
        let kept_dir = root.join("kept");
        std::fs::create_dir_all(&kept_dir)?;
        let wanted: HashSet<String> = kept.iter().map(|r| keys::ref_hash(r)).collect();
        for run in std::fs::read_dir(&kept_dir)? {
            let run = run?.path();
            let named = run.file_name().map(|n| n.to_string_lossy().into_owned());
            if !named.is_some_and(|n| wanted.contains(&n)) || read_kept(&run).is_none() {
                remove(&run)?;
            }
        }
        let mut bases = Bases::default();
        for cache in std::fs::read_dir(&bases_dir)? {
            let cache = cache?;
            let name = cache.file_name().to_string_lossy().into_owned();
            // An earlier runner's, named so that its repository cannot be
            // told, and so never dropped with it: its next run warms anew.
            if name.starts_with(PREFIX) && repo_of(&name).is_none() {
                remove(&cache.path())?;
                continue;
            }
            let mut gens = BTreeSet::new();
            for base in std::fs::read_dir(cache.path())? {
                let base = base?;
                match base.file_name().to_string_lossy().parse::<u64>() {
                    Ok(generation) => gens.insert(generation),
                    // One half made when the runner stopped.
                    Err(_) => remove(&base.path()).map(|_| false)?,
                };
            }
            let Some(&newest) = gens.last() else {
                continue;
            };
            bases.next = bases.next.max(newest + 1);
            bases.current.insert(name.clone(), newest);
            bases.kept.insert(name.clone(), gens);
            if alive.is_empty() {
                for old in bases.unused(&name) {
                    remove(&cache.path().join(old.to_string()))?;
                }
            }
        }
        // Each mark of a base still current, which a pull made.
        let marks = std::fs::read(root.join(MARKS)).unwrap_or_default();
        let marks: HashMap<String, (u64, u64)> = serde_json::from_slice(&marks).unwrap_or_default();
        bases.marks = marks
            .into_iter()
            .filter(|(name, (g, _))| bases.current.get(name) == Some(g))
            .collect();
        let boot = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis();
        Ok(Arc::new(Caches {
            root: root.to_path_buf(),
            space,
            boot,
            runs: AtomicU64::new(0),
            bases: Mutex::new(bases),
            kept: Mutex::new(()),
        }))
    }

    /// Drops the caches of the repository `repo` ([`name`]), as of one
    /// deleted: each base no run reads now goes, and those that runs do go
    /// once they end, none being current any more.
    pub fn drop_repo(&self, repo: &str) -> Result<()> {
        let hash = keys::ref_hash(repo);
        let mut bases = self.bases.lock().unwrap();
        let names: Vec<String> = bases.kept.keys().cloned().collect();
        let names = names.into_iter().filter(|n| repo_of(n) == Some(&hash));
        for name in names.collect::<Vec<_>>() {
            bases.current.remove(&name);
            bases.marks.remove(&name);
            let dir = self.root.join("bases").join(&name);
            for generation in bases.unused(&name) {
                remove(&dir.join(generation.to_string()))?;
            }
            // Empty, with no run reading it.
            let _ = std::fs::remove_dir(&dir);
        }
        Ok(())
    }

    /// The bytes free on the disk the caches are on.
    pub fn free(&self) -> io::Result<u64> {
        (self.space)(&self.root)
    }

    /// Makes room on the disk, as it runs short: removes the layers kept
    /// of every run, which a push would only have saved a run's work
    /// with, and every cache's base no run reads, as [`Caches::drop_repo`]
    /// drops them, for the next run of each to start afresh. Says how
    /// many bases and kept runs went.
    pub fn shed(&self) -> Result<(usize, usize)> {
        let mut runs = 0;
        {
            let _kept = self.kept.lock().unwrap();
            for run in std::fs::read_dir(self.root.join("kept"))? {
                remove(&run?.path())?;
                runs += 1;
            }
        }
        let mut bases = self.bases.lock().unwrap();
        let mut gone = 0;
        let current: Vec<(String, u64)> = bases.current.clone().into_iter().collect();
        for (name, generation) in current {
            if bases.readers.contains_key(&(name.clone(), generation)) {
                continue;
            }
            bases.current.remove(&name);
            bases.marks.remove(&name);
            for generation in bases.unused(&name) {
                remove(&self.base(&name, generation))?;
                gone += 1;
            }
        }
        Ok((gone, runs))
    }

    /// Where the layers kept of the run `run` are.
    fn kept_dir(&self, run: &str) -> PathBuf {
        self.root.join("kept").join(keys::ref_hash(run))
    }

    /// Keeps the layers `lease` wrote to, those with anything in them, as
    /// the run `run`'s, `now`, unless they would take more than `budget`
    /// bytes; then keeps the kept layers within their bounds.
    fn keeping(&self, lease: Lease, run: &str, budget: u64, now: u64) -> Result<()> {
        let _kept = self.kept.lock().unwrap();
        let dir = self.kept_dir(run);
        let making = dir.with_extension("new");
        remove(&dir)?;
        remove(&making)?;
        std::fs::create_dir_all(&making)?;
        let mut layers = Vec::new();
        for layer in &lease.layers {
            if std::fs::read_dir(&layer.upper)?.next().is_none() {
                continue;
            }
            let to = making.join(layers.len().to_string());
            std::fs::create_dir(&to)?;
            std::fs::rename(&layer.upper, to.join("upper"))?;
            layers.push(KeptLayer {
                name: layer.name.clone(),
                path: layer.path.clone(),
                generation: layer.generation,
            });
        }
        let size = size(&making)?;
        if layers.is_empty() || size > budget {
            return Ok(remove(&making)?);
        }
        let kept = Kept {
            at: now,
            size,
            layers,
        };
        std::fs::write(making.join(KEPT), serde_json::to_vec(&kept)?)?;
        std::fs::rename(&making, &dir)?;
        self.bound(budget, now)
    }

    /// Removes the kept layers past their bounds, `now`: older than
    /// [`KEPT_AGE`], and, the newest first, past [`KEPT_MAX`] of them or
    /// `budget` bytes in all. Called holding `kept`.
    fn bound(&self, budget: u64, now: u64) -> Result<()> {
        let mut all = Vec::new();
        for run in std::fs::read_dir(self.root.join("kept"))? {
            let run = run?.path();
            let kept = read_kept(&run).unwrap_or_default();
            all.push((kept.at, kept.size, run));
        }
        all.sort_by_key(|(at, ..)| std::cmp::Reverse(*at));
        let (mut count, mut total) = (0, 0);
        for (at, size, run) in all {
            let fits = now.saturating_sub(at) <= KEPT_AGE && count < KEPT_MAX;
            if fits && total + size <= budget {
                (count, total) = (count + 1, total + size);
            } else {
                remove(&run)?;
            }
        }
        Ok(())
    }

    /// Removes the layers kept of each of `runs`, as of runs a push can no
    /// longer take, and those past their bounds, as [`Caches::bound`].
    pub fn forget(&self, runs: &[String], budget: u64) -> Result<()> {
        self.forgetting(runs, budget, now())
    }

    /// [`Caches::forget`], `now`.
    fn forgetting(&self, runs: &[String], budget: u64, now: u64) -> Result<()> {
        let _kept = self.kept.lock().unwrap();
        for run in runs {
            remove(&self.kept_dir(run))?;
        }
        self.bound(budget, now)
    }

    /// Takes the layers kept of the run `run`, for a push that takes its
    /// check: each over a base still current, leased over it, and where
    /// the run saw each over one that has moved on since, which is not.
    /// They go once the lease does, promoted or not.
    pub fn take(self: &Arc<Self>, run: &str) -> Taken {
        let _kept = self.kept.lock().unwrap();
        let dir = self.kept_dir(run);
        let kept = read_kept(&dir).unwrap_or_default();
        let mut bases = self.bases.lock().unwrap();
        let (mut layers, mut moved) = (Vec::new(), Vec::new());
        for (i, layer) in kept.layers.into_iter().enumerate() {
            let KeptLayer {
                name,
                path,
                generation,
            } = layer;
            if bases.current.get(&name) != Some(&generation) {
                moved.push(path);
                continue;
            }
            *bases.readers.entry((name.clone(), generation)).or_default() += 1;
            let at = dir.join(i.to_string());
            layers.push(Layer {
                lower: self.base(&name, generation),
                name,
                path,
                generation,
                upper: at.join("upper"),
                work: at.join("work"),
            });
        }
        let id = keys::ref_hash(run);
        let caches = self.clone();
        let lease = Lease {
            caches,
            id,
            dir,
            layers,
        };
        Taken { lease, moved }
    }

    /// Writes down `marks`, the tips the bases pulls made were made at.
    fn mark(&self, marks: &HashMap<String, (u64, u64)>) -> Result<()> {
        let path = self.root.join(MARKS);
        let making = path.with_extension("new");
        std::fs::write(&making, serde_json::to_vec(marks)?)?;
        Ok(std::fs::rename(&making, &path)?)
    }

    fn base(&self, name: &str, generation: u64) -> PathBuf {
        self.root
            .join("bases")
            .join(name)
            .join(generation.to_string())
    }

    /// Layers for a run over the current base of each of `caches`, as
    /// `(name, path)`: an empty base for a cache that has none yet.
    pub fn lease(self: &Arc<Self>, caches: &[(String, String)]) -> Result<Lease> {
        let n = self.runs.fetch_add(1, Ordering::Relaxed);
        let id = format!("{}-{n}", self.boot);
        let dir = self.root.join("runs").join(&id);
        let mut lease = Lease {
            caches: self.clone(),
            id,
            dir,
            layers: Vec::new(),
        };
        for (i, (name, path)) in caches.iter().enumerate() {
            let generation = self.read(name)?;
            let layer = lease.dir.join(i.to_string());
            lease.layers.push(Layer {
                name: name.clone(),
                path: path.clone(),
                generation,
                lower: self.base(name, generation),
                upper: layer.join("upper"),
                work: layer.join("work"),
            });
            std::fs::create_dir_all(layer.join("upper"))?;
            std::fs::create_dir_all(layer.join("work"))?;
        }
        Ok(lease)
    }

    /// The generation of `name`'s base a run starts from, counted as read
    /// until it ends.
    fn read(&self, name: &str) -> Result<u64> {
        let mut bases = self.bases.lock().unwrap();
        let generation = match bases.current.get(name) {
            Some(g) => *g,
            None => {
                let g = bases.next;
                std::fs::create_dir_all(self.base(name, g))?;
                bases.next += 1;
                bases.current.insert(name.to_string(), g);
                bases.kept.entry(name.to_string()).or_default().insert(g);
                g
            }
        };
        *bases
            .readers
            .entry((name.to_string(), generation))
            .or_default() += 1;
        Ok(generation)
    }

    /// Removes the bases of `name` that are no longer used.
    fn sweep(&self, name: &str) {
        let unused = self.bases.lock().unwrap().unused(name);
        for generation in unused {
            let _ = remove(&self.base(name, generation));
        }
    }
}

impl Lease {
    /// The overlays to mount, one docker volume each, named for the run.
    pub fn mounts(&self) -> Vec<executor::Cache> {
        self.layers
            .iter()
            .enumerate()
            .map(|(i, l)| executor::Cache {
                volume: format!("knotless-overlay-{}-{i}", self.id),
                path: l.path.clone(),
                run: self.id.clone(),
                lower: l.lower.clone(),
                upper: l.upper.clone(),
                work: l.work.clone(),
            })
            .collect()
    }

    /// Keeps what the run wrote, once its overlays are unmounted, as the
    /// run `run`'s, for a push that takes its check to take, within the
    /// bounds of kept layers and `budget` bytes for them all.
    pub async fn keep(self, run: &str, budget: u64) -> Result<()> {
        let (caches, run) = (self.caches.clone(), run.to_string());
        let kept = move || caches.keeping(self, &run, budget, now());
        tokio::task::spawn_blocking(kept).await?
    }

    /// Makes what the run wrote, once its overlays are unmounted, the base
    /// the runs after it start from: each layer it wrote to over its base,
    /// as a new one. A base that would be over `max` bytes is dropped
    /// instead, for the next run to start afresh.
    pub async fn promote(self, max: u64) -> Result<()> {
        self.promoting(max, None).await.map(|_| ())
    }

    /// Makes what a trusted pull's run wrote, once its overlays are
    /// unmounted, the base the runs after it start from, as
    /// [`Lease::promote`] does a push's, where no push runs to: each layer
    /// only over the base it was written over, if that is still current
    /// and was made at the branch's tip of generation `at`, the one the
    /// pull merged with, or an older one. Says where the run saw the
    /// layers made the base, and those that are not.
    pub async fn advance(self, max: u64, at: u64) -> Result<(Vec<String>, Vec<String>)> {
        self.promoting(max, Some(at)).await
    }

    /// [`Lease::promote`], or, given the tip a pull merged with, `at`,
    /// [`Lease::advance`].
    async fn promoting(self, max: u64, at: Option<u64>) -> Result<(Vec<String>, Vec<String>)> {
        let (caches, layers) = (self.caches.clone(), self.layers.clone());
        let building = caches.clone();
        let made = tokio::task::spawn_blocking(move || -> Result<Vec<_>> {
            let caches = building;
            let mut made = Vec::new();
            for layer in layers {
                if std::fs::read_dir(&layer.upper)?.next().is_none() {
                    continue;
                }
                let generation = {
                    let mut bases = caches.bases.lock().unwrap();
                    bases.next += 1;
                    bases.next - 1
                };
                let base = caches.base(&layer.name, generation);
                let making = base.with_extension("new");
                clone(&layer.lower, &making)?;
                apply(&layer.upper, &making)?;
                let fits = size(&making)? <= max;
                std::fs::rename(&making, &base)?;
                made.push((layer, generation, fits));
            }
            Ok(made)
        });
        let made = made.await??;
        let (mut kept, mut moved) = (Vec::new(), Vec::new());
        for (layer, generation, fits) in made {
            let mut bases = caches.bases.lock().unwrap();
            let name = layer.name.clone();
            bases
                .kept
                .entry(name.clone())
                .or_default()
                .insert(generation);
            // Checked as it is made current, so of two pulls' runs over
            // one base, the first to end makes the next, and the other's
            // goes.
            let ahead = at.is_none_or(|at| bases.ahead(&name, layer.generation, at));
            if !ahead {
                moved.push(layer.path.clone());
            } else if fits {
                bases.current.insert(name.clone(), generation);
                kept.push(layer.path.clone());
                if let Some(at) = at {
                    bases.marks.insert(name.clone(), (generation, at));
                    caches.mark(&bases.marks)?;
                }
            } else if bases.current.get(&name) == Some(&layer.generation) {
                bases.current.remove(&name);
            }
            drop(bases);
            caches.sweep(&name);
        }
        Ok((kept, moved))
    }
}

impl Drop for Lease {
    fn drop(&mut self) {
        let _ = remove(&self.dir);
        let mut bases = self.caches.bases.lock().unwrap();
        for layer in &self.layers {
            let key = (layer.name.clone(), layer.generation);
            let n = bases.readers.entry(key.clone()).or_default();
            *n = n.saturating_sub(1);
            if *n == 0 {
                bases.readers.remove(&key);
            }
        }
        drop(bases);
        for layer in &self.layers {
            self.caches.sweep(&layer.name);
        }
    }
}

/// The bytes free to the runner on the disk holding `path`.
pub fn free(path: &Path) -> io::Result<u64> {
    let stat = rustix::fs::statvfs(path)?;
    Ok(stat.f_bavail.saturating_mul(stat.f_frsize))
}

/// The seconds since the epoch.
fn now() -> u64 {
    let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
    since.unwrap_or_default().as_secs()
}

/// What the kept run's directory `dir` says of it, if it says.
fn read_kept(dir: &Path) -> Option<Kept> {
    let text = std::fs::read(dir.join(KEPT)).ok()?;
    serde_json::from_slice(&text).ok()
}

/// Removes `path`, a directory and all in it or anything else, if it is
/// there.
/// What every cache's name starts with.
const PREFIX: &str = "knotless-cache-";

/// The name of a cache of the repository `repo`, `<owner did>/<key>`,
/// that `what` tells apart from its others: a hash of each, so that the
/// repository's caches can be told by name ([`Caches::drop_repo`]).
pub fn name(repo: &str, what: &str) -> String {
    format!("{PREFIX}{}-{}", keys::ref_hash(repo), keys::ref_hash(what))
}

/// The hash of the repository the cache `name` is of, as [`name`] makes
/// one; none for a name it did not make.
fn repo_of(name: &str) -> Option<&str> {
    let rest = name.strip_prefix(PREFIX)?;
    Some(rest.split_once('-')?.0)
}

fn remove(path: &Path) -> io::Result<()> {
    let removed = match std::fs::symlink_metadata(path) {
        Ok(meta) if meta.is_dir() => std::fs::remove_dir_all(path),
        Ok(_) => std::fs::remove_file(path),
        Err(err) => Err(err),
    };
    match removed {
        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
        done => done,
    }
}

/// Makes `to` as `from` is, its directories new and everything else hard
/// links to `from`'s.
fn clone(from: &Path, to: &Path) -> io::Result<()> {
    std::fs::create_dir(to)?;
    for entry in std::fs::read_dir(from)? {
        let entry = entry?;
        let dest = to.join(entry.file_name());
        if entry.file_type()?.is_dir() {
            clone(&entry.path(), &dest)?;
        } else {
            std::fs::hard_link(entry.path(), &dest)?;
        }
    }
    like(from, to)
}

/// Gives the directory `to` the permissions and owner `from` has.
fn like(from: &Path, to: &Path) -> io::Result<()> {
    let meta = std::fs::symlink_metadata(from)?;
    std::fs::set_permissions(to, meta.permissions())?;
    std::os::unix::fs::lchown(to, Some(meta.uid()), Some(meta.gid()))
}

/// Applies the upper layer `upper` to `base`, as overlayfs shows the two
/// together: each whiteout removes what it names, a directory not marked
/// opaque merges with the one below it, and anything else replaces what
/// is below it, moved rather than copied.
fn apply(upper: &Path, base: &Path) -> io::Result<()> {
    for entry in std::fs::read_dir(upper)? {
        let entry = entry?;
        let (from, to) = (entry.path(), base.join(entry.file_name()));
        let meta = entry.metadata()?;
        let below = std::fs::symlink_metadata(&to).ok();
        let merges =
            meta.is_dir() && !opaque(&from) && below.as_ref().is_some_and(Metadata::is_dir);
        if merges {
            apply(&from, &to)?;
            like(&from, &to)?;
            continue;
        }
        remove(&to)?;
        // A whiteout, once what it names is gone, has done its work.
        let moved = (!whiteout(&meta)).then(|| std::fs::rename(&from, &to));
        moved.transpose()?;
    }
    Ok(())
}

/// Whether `meta` is of a whiteout: a character device numbered 0, 0.
fn whiteout(meta: &Metadata) -> bool {
    meta.file_type().is_char_device() && meta.rdev() == 0
}

/// Whether the directory `dir` hides the one below it.
fn opaque(dir: &Path) -> bool {
    let mut value = [0u8; 1];
    let read = rustix::fs::lgetxattr(dir, OPAQUE, &mut value[..]);
    matches!(read, Ok(1)) && value[0] == b'y'
}

/// The bytes the files under `dir` take on disk, each file once however
/// many names it has, as `du` counts them: cargo links what it builds
/// under more than one.
fn size(dir: &Path) -> io::Result<u64> {
    size_of(dir, &mut HashSet::new())
}

/// [`size`], not counting again the files `seen`, by inode.
fn size_of(dir: &Path, seen: &mut HashSet<u64>) -> io::Result<u64> {
    let mut total = 0;
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let meta = entry.metadata()?;
        if meta.nlink() > 1 && !meta.is_dir() && !seen.insert(meta.ino()) {
            continue;
        }
        total += meta.blocks() * 512;
        if meta.is_dir() {
            total += size_of(&entry.path(), seen)?;
        }
    }
    Ok(total)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn write(path: &Path, text: &str) {
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(path, text).unwrap();
    }

    fn read(path: &Path) -> String {
        std::fs::read_to_string(path).unwrap_or_default()
    }

    /// What `dir` holds, as `path=text` for each file, in order.
    fn listing(dir: &Path) -> Vec<String> {
        let mut all = Vec::new();
        let mut stack = vec![dir.to_path_buf()];
        while let Some(d) = stack.pop() {
            for e in std::fs::read_dir(&d).unwrap() {
                let p = e.unwrap().path();
                if p.is_dir() {
                    stack.push(p);
                } else {
                    let rel = p.strip_prefix(dir).unwrap().display().to_string();
                    all.push(format!("{rel}={}", read(&p)));
                }
            }
        }
        all.sort();
        all
    }

    /// Makes a whiteout at `path`, as overlayfs does: false where that
    /// takes privileges the tests do not have (macOS, not as root).
    fn whiteout_at(path: &Path) -> bool {
        let mut made = std::process::Command::new("mknod");
        made.arg(path).args(["c", "0", "0"]);
        made.stderr(std::process::Stdio::null())
            .status()
            .is_ok_and(|s| s.success())
    }

    fn names(caches: &[(&str, &str)]) -> Vec<(String, String)> {
        caches
            .iter()
            .map(|(n, p)| (n.to_string(), p.to_string()))
            .collect()
    }

    #[tokio::test]
    async fn a_deleted_repositorys_caches_go() {
        let root = tempfile::tempdir().unwrap();
        let bases = root.path().join("bases");
        // An earlier runner's, named for no repository, goes at the start.
        let old = format!("{PREFIX}{}", keys::ref_hash("x"));
        std::fs::create_dir_all(bases.join(&old).join("0")).unwrap();
        let caches = Caches::open(root.path(), &[], &[]).unwrap();
        assert!(!bases.join(&old).exists());
        let (gone, other) = (name("did:a/k", "t"), name("did:a/j", "t"));
        assert_eq!(repo_of(&gone), Some(keys::ref_hash("did:a/k").as_str()));
        assert_eq!(repo_of("t"), None);
        // One run reads the deleted repository's cache as it goes.
        let reading = caches.lease(&names(&[(gone.as_str(), "/a")])).unwrap();
        let other_base = caches.lease(&names(&[(other.as_str(), "/a")])).unwrap();
        let read = reading.layers[0].lower.clone();
        caches.drop_repo("did:a/k").unwrap();
        assert!(read.exists());
        drop(reading);
        assert!(!read.exists());
        // Another repository's stays.
        assert!(other_base.layers[0].lower.exists());
        // One no run reads goes at once, its directory with it.
        let unread = caches.lease(&names(&[(gone.as_str(), "/a")])).unwrap();
        let base = unread.layers[0].lower.clone();
        drop(unread);
        caches.drop_repo("did:a/k").unwrap();
        assert!(!base.exists());
        assert!(!bases.join(&gone).exists());
    }

    #[tokio::test]
    async fn runs_side_by_side_and_main_keeps() {
        let root = tempfile::tempdir().unwrap();
        let caches = Caches::open(root.path(), &[], &[]).unwrap();
        let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
        // Two runs at once, each over an empty base, with layers of its own.
        let a = caches.lease(&wanted).unwrap();
        let b = caches.lease(&wanted).unwrap();
        assert_ne!(a.id, b.id);
        assert_eq!(a.layers[0].lower, b.layers[0].lower);
        assert_ne!(a.layers[0].upper, b.layers[0].upper);
        let mounts = a.mounts();
        assert_eq!(mounts[1].volume, format!("knotless-overlay-{}-1", a.id));
        assert_eq!(mounts[1].path, "/usr/local/rustup");
        assert_eq!(mounts[1].run, a.id);
        // What a pull writes is gone with it, and its layers with it.
        write(&b.layers[0].upper.join("pull"), "x");
        let b_dir = b.dir.clone();
        drop(b);
        assert!(!b_dir.exists());
        // A main run that wrote makes a new base; one it did not write to
        // keeps its own.
        write(&a.layers[0].upper.join("debug/dep"), "1");
        write(&a.layers[0].upper.join("debug/.fingerprint/f"), "1");
        let first = a.layers[0].lower.clone();
        let rustup = a.layers[1].lower.clone();
        a.promote(1 << 30).await.unwrap();
        assert!(!first.exists(), "an old base no run reads goes");
        let c = caches.lease(&wanted).unwrap();
        assert_eq!(c.layers[1].lower, rustup);
        let base = c.layers[0].lower.clone();
        assert_eq!(listing(&base), ["debug/.fingerprint/f=1", "debug/dep=1"]);
        assert!(listing(&root.path().join("runs")).is_empty());

        // A pull reading that base keeps it while a later main run makes
        // the next: overwritten, removed, and made again.
        let pull = caches.lease(&wanted).unwrap();
        let up = &c.layers[0].upper;
        write(&up.join("debug/dep"), "2");
        write(&up.join("new/file"), "n");
        let whiteouts = whiteout_at(&up.join("debug/.fingerprint"));
        c.promote(1 << 30).await.unwrap();
        assert_eq!(
            listing(&pull.layers[0].lower),
            ["debug/.fingerprint/f=1", "debug/dep=1"]
        );
        let next = caches.lease(&wanted).unwrap();
        // Where no whiteout could be made, what it would remove stays.
        let stays = ["debug/.fingerprint/f=1"]
            .into_iter()
            .filter(|_| !whiteouts);
        let mut want: Vec<&str> = stays.collect();
        want.extend(["debug/dep=2", "new/file=n"]);
        assert_eq!(listing(&next.layers[0].lower), want);
        // Once the pull ends, the base it read goes.
        drop(pull);
        assert!(!base.exists());

        // A directory made again hides what was below it; a file replaces
        // a directory, and a directory a file.
        let up = &next.layers[0].upper;
        write(&up.join("debug/fresh"), "f");
        rustix::fs::lsetxattr(
            up.join("debug"),
            OPAQUE,
            b"y",
            rustix::fs::XattrFlags::empty(),
        )
        .unwrap();
        write(&up.join("new"), "now a file");
        write(&up.join("lone"), "l");
        next.promote(1 << 30).await.unwrap();
        let last = caches.lease(&wanted).unwrap();
        assert_eq!(
            listing(&last.layers[0].lower),
            ["debug/fresh=f", "lone=l", "new=now a file"]
        );
        let up = &last.layers[0].upper;
        write(&up.join("lone/inside"), "i");
        // Over the most a cache may hold, a base is dropped, and the next
        // run starts afresh.
        last.promote(1).await.unwrap();
        let fresh = caches.lease(&wanted).unwrap();
        assert!(listing(&fresh.layers[0].lower).is_empty());
        let one = fresh.layers[1].lower.clone();
        drop(fresh);
        let bases = root.path().join("bases");
        assert_eq!(std::fs::read_dir(bases.join("t")).unwrap().count(), 1);

        // Opened again, the newest base of each is kept, and nothing else.
        write(&bases.join("t/99.new/half"), "h");
        std::fs::create_dir_all(bases.join("t/0")).unwrap();
        std::fs::create_dir_all(bases.join("empty")).unwrap();
        write(&root.path().join("runs/old/0/upper/x"), "x");
        std::fs::create_dir_all(root.path().join("runs/alive")).unwrap();
        let again = Caches::open(root.path(), &[], &[]).unwrap();
        assert!(!bases.join("t/99.new").exists());
        assert!(!bases.join("t/0").exists());
        assert_eq!(
            std::fs::read_dir(root.path().join("runs")).unwrap().count(),
            0
        );
        let lease = again.lease(&wanted).unwrap();
        assert_eq!(lease.layers[1].lower, one);
        drop(lease);
        // But for what runs whose overlays are still mounted read.
        std::fs::create_dir_all(bases.join("t/0")).unwrap();
        std::fs::create_dir_all(root.path().join("runs/alive")).unwrap();
        Caches::open(root.path(), &["alive".into()], &[]).unwrap();
        assert!(bases.join("t/0").exists());
        assert!(root.path().join("runs/alive").exists());
        // A root it cannot make is an error.
        let file = root.path().join("file");
        write(&file, "");
        assert!(Caches::open(&file, &[], &[]).is_err());
        assert!(remove(&root.path().join("gone")).is_ok());
        // A file under two names counts once, as `du` counts it.
        let linked = root.path().join("linked");
        write(&linked.join("one"), &"x".repeat(1 << 16));
        let once = size(&linked).unwrap();
        std::fs::hard_link(linked.join("one"), linked.join("two")).unwrap();
        assert_eq!(size(&linked).unwrap(), once);
    }

    #[tokio::test]
    async fn pulls_advance_the_base_where_no_push_runs() {
        let root = tempfile::tempdir().unwrap();
        let caches = Caches::open(root.path(), &[], &[]).unwrap();
        let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
        let lower = |c: &Arc<Caches>| {
            let lease = c.lease(&wanted).unwrap();
            listing(&lease.layers[0].lower)
        };
        // Two pulls' runs over one base, merged with the branch's tip at
        // generation 5: the first to end makes the next base, marked as of
        // that tip; the other's, over a base that has moved on, goes.
        let a = caches.lease(&wanted).unwrap();
        let b = caches.lease(&wanted).unwrap();
        write(&a.layers[0].upper.join("a"), "a");
        write(&b.layers[0].upper.join("b"), "b");
        let made = a.advance(1 << 30, 5).await.unwrap();
        assert_eq!(made, (vec!["/work/target".to_string()], Vec::new()));
        let made = b.advance(1 << 30, 5).await.unwrap();
        assert_eq!(made, (Vec::new(), vec!["/work/target".to_string()]));
        assert_eq!(lower(&caches), ["a=a"]);
        assert_eq!(
            std::fs::read_dir(root.path().join("bases/t"))
                .unwrap()
                .count(),
            1
        );
        // One merged with an older tip never takes the base back to it,
        // though it ran over that base; one merged with the same or a
        // newer one goes on from it.
        let old = caches.lease(&wanted).unwrap();
        write(&old.layers[0].upper.join("old"), "o");
        assert_eq!(
            old.advance(1 << 30, 4).await.unwrap().0,
            Vec::<String>::new()
        );
        let same = caches.lease(&wanted).unwrap();
        write(&same.layers[0].upper.join("same"), "s");
        assert_eq!(
            same.advance(1 << 30, 5).await.unwrap().1,
            Vec::<String>::new()
        );
        assert_eq!(lower(&caches), ["a=a", "same=s"]);
        // The mark outlasts the runner: opened again, an older tip still
        // does not take it back, and a newer one goes on.
        let again = Caches::open(root.path(), &[], &[]).unwrap();
        let old = again.lease(&wanted).unwrap();
        write(&old.layers[0].upper.join("old"), "o");
        assert!(old.advance(1 << 30, 4).await.unwrap().0.is_empty());
        let newer = again.lease(&wanted).unwrap();
        write(&newer.layers[0].upper.join("new"), "n");
        newer.advance(1 << 30, 6).await.unwrap();
        assert_eq!(lower(&again), ["a=a", "new=n", "same=s"]);
        // A push's base is marked with no tip, so any pull goes on from
        // it; and a mark of a base no longer current is let go on opening.
        let push = again.lease(&wanted).unwrap();
        write(&push.layers[0].upper.join("push"), "p");
        push.promote(1 << 30).await.unwrap();
        let again = Caches::open(root.path(), &[], &[]).unwrap();
        assert!(again.bases.lock().unwrap().marks.is_empty());
        let after = again.lease(&wanted).unwrap();
        write(&after.layers[0].upper.join("after"), "f");
        after.advance(1 << 30, 1).await.unwrap();
        assert_eq!(
            lower(&again),
            ["a=a", "after=f", "new=n", "push=p", "same=s"]
        );
        // Over the most a cache may hold, the base is dropped, as a push's.
        let big = again.lease(&wanted).unwrap();
        write(&big.layers[0].upper.join("big"), "b");
        assert_eq!(big.advance(1, 9).await.unwrap(), (Vec::new(), Vec::new()));
        assert!(lower(&again).is_empty());
    }

    /// The runs whose layers are kept, by the names their directories have.
    fn kept_of(root: &Path) -> Vec<String> {
        let all = std::fs::read_dir(root.join("kept")).unwrap();
        let mut all: Vec<String> = all
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        all.sort();
        all
    }

    #[tokio::test]
    async fn pulls_keep_what_a_push_takes() {
        let root = tempfile::tempdir().unwrap();
        let caches = Caches::open(root.path(), &[], &[]).unwrap();
        let wanted = names(&[("t", "/work/target"), ("r", "/usr/local/rustup")]);
        let main = caches.lease(&wanted).unwrap();
        write(&main.layers[0].upper.join("dep"), "1");
        main.promote(1 << 30).await.unwrap();

        // A pull's run that passed keeps the layers it wrote to, and only
        // those, its own run's layers gone.
        let pull = caches.lease(&wanted).unwrap();
        let (over, dir) = (pull.layers[0].lower.clone(), pull.dir.clone());
        write(&pull.layers[0].upper.join("dep"), "2");
        write(&pull.layers[0].upper.join("new"), "n");
        pull.keep("pull-run", 1 << 30).await.unwrap();
        assert!(!dir.exists());
        assert_eq!(kept_of(root.path()), [keys::ref_hash("pull-run")]);
        // The push that takes its check takes them, into a new base over
        // the one it was written over, as its own run's would go.
        let taken = caches.take("pull-run");
        assert_eq!(taken.kept(), ["/work/target"]);
        assert!(taken.moved.is_empty());
        taken.lease.promote(1 << 30).await.unwrap();
        assert!(kept_of(root.path()).is_empty());
        assert!(!over.exists(), "the base it was over goes");
        let next = caches.lease(&wanted).unwrap();
        assert_eq!(listing(&next.layers[0].lower), ["dep=2", "new=n"]);
        drop(next);
        // A run nothing was kept of has none to take.
        let none = caches.take("never");
        assert!(none.kept().is_empty() && none.moved.is_empty());
        none.lease.promote(1 << 30).await.unwrap();

        // One written over a base that has moved on since is not kept: its
        // writes and whiteouts are of what that base held.
        let stale = caches.lease(&wanted).unwrap();
        write(&stale.layers[0].upper.join("stale"), "s");
        write(&stale.layers[1].upper.join("toolchain"), "t");
        let main = caches.lease(&wanted).unwrap();
        write(&main.layers[0].upper.join("main"), "m");
        stale.keep("stale-run", 1 << 30).await.unwrap();
        main.promote(1 << 30).await.unwrap();
        let taken = caches.take("stale-run");
        assert_eq!(taken.kept(), ["/usr/local/rustup"]);
        assert_eq!(taken.moved, ["/work/target"]);
        taken.lease.promote(1 << 30).await.unwrap();
        assert!(kept_of(root.path()).is_empty());
        let next = caches.lease(&wanted).unwrap();
        assert_eq!(listing(&next.layers[0].lower), ["dep=2", "main=m", "new=n"]);
        assert_eq!(listing(&next.layers[1].lower), ["toolchain=t"]);
        drop(next);

        // A run that wrote nothing keeps nothing, and nor does one over
        // the budget.
        let bare = caches.lease(&wanted).unwrap();
        bare.keep("bare", 1 << 30).await.unwrap();
        let big = caches.lease(&wanted).unwrap();
        write(&big.layers[0].upper.join("big"), "b");
        big.keep("big", 1).await.unwrap();
        assert!(kept_of(root.path()).is_empty());

        // Kept, they are bounded: in number, the oldest going first...
        let t0 = 1_000_000_000;
        let keep = |run: &str, at: u64, budget: u64| {
            let lease = caches.lease(&wanted).unwrap();
            write(&lease.layers[0].upper.join(run), run);
            caches.keeping(lease, run, budget, at).unwrap();
        };
        for n in 0..=KEPT_MAX as u64 {
            keep(&format!("k{n}"), t0 + n, 1 << 30);
        }
        let all = kept_of(root.path());
        assert_eq!(all.len(), KEPT_MAX);
        assert!(!all.contains(&keys::ref_hash("k0")));
        // ...in age...
        let k1 = keys::ref_hash("k1");
        let late = t0 + 2 + KEPT_AGE;
        caches.forgetting(&[], 1 << 30, late).unwrap();
        assert!(!kept_of(root.path()).contains(&k1));
        assert_eq!(kept_of(root.path()).len(), KEPT_MAX - 1);
        // ...and in bytes, the newest kept.
        let newest = format!("k{KEPT_MAX}");
        let one = read_kept(&caches.kept_dir(&newest)).unwrap().size;
        caches.forgetting(&[], one, late).unwrap();
        assert_eq!(kept_of(root.path()), [keys::ref_hash(&newest)]);
        // A run kept again replaces what it kept; one forgotten goes.
        keep(&newest, late, 1 << 30);
        assert_eq!(kept_of(root.path()), [keys::ref_hash(&newest)]);
        caches.forgetting(&[newest], 1 << 30, late).unwrap();
        assert!(kept_of(root.path()).is_empty());

        // Opened again, it keeps the layers of the runs it is told a push
        // may yet take, and nothing else.
        keep("wanted", late, 1 << 30);
        keep("unwanted", late, 1 << 30);
        std::fs::create_dir_all(caches.kept_dir("broken")).unwrap();
        let half = caches.kept_dir("half").with_extension("new");
        std::fs::create_dir_all(half).unwrap();
        let named = ["wanted".into(), "broken".into()];
        Caches::open(root.path(), &[], &named).unwrap();
        assert_eq!(kept_of(root.path()), [keys::ref_hash("wanted")]);
        // A kept run's directory that says nothing of it counts for none.
        write(&root.path().join("kept/junk/0/upper/x"), "x");
        caches.forgetting(&[], 1 << 30, late).unwrap();
        assert_eq!(kept_of(root.path()), [keys::ref_hash("wanted")]);
    }

    #[tokio::test]
    async fn short_of_disk_it_sheds_what_no_run_reads() {
        let root = tempfile::tempdir().unwrap();
        // The disk as statvfs tells it, of somewhere real.
        assert!(free(root.path()).unwrap() > 0);
        assert!(free(&root.path().join("gone")).is_err());
        let space: Space = Box::new(|_| Ok(7));
        let caches = Caches::open_with(root.path(), &[], &[], space).unwrap();
        assert_eq!(caches.free().unwrap(), 7);
        let wanted = names(&[("read", "/a"), ("unread", "/b")]);
        // Bases for both caches, and a pull's run kept.
        let main = caches.lease(&wanted).unwrap();
        write(&main.layers[0].upper.join("x"), "x");
        write(&main.layers[1].upper.join("y"), "y");
        main.promote(1 << 30).await.unwrap();
        let pull = caches.lease(&wanted).unwrap();
        write(&pull.layers[0].upper.join("z"), "z");
        pull.keep("pull-run", 1 << 30).await.unwrap();
        assert!(caches.kept_dir("pull-run").exists());
        // A run reads one cache's base as the disk runs short.
        let reading = caches.lease(&names(&[("read", "/a")])).unwrap();
        let (read, unread) = {
            let all = caches.lease(&wanted).unwrap();
            (all.layers[0].lower.clone(), all.layers[1].lower.clone())
        };
        assert_eq!(caches.shed().unwrap(), (1, 1));
        assert!(!caches.kept_dir("pull-run").exists());
        assert!(!unread.exists());
        assert!(read.exists());
        // The next run of the one shed starts afresh; the one read goes
        // on from its base.
        let next = caches.lease(&wanted).unwrap();
        assert_eq!(next.layers[0].lower, read);
        assert!(listing(&next.layers[1].lower).is_empty());
        drop((next, reading));
        // Nothing left to shed but the bases no run reads now.
        assert_eq!(caches.shed().unwrap(), (2, 0));
    }
}