//! Disk-pressure watch: raises a loose-ends-v2 todo when the filesystem
//! backing this agent's state gets tight, pointing at the agent's own
//! biggest directories so the todo says where the bytes actually went.
//!
//! Harness-local by design. hive-c0re cannot push a todo — the todo store
//! and its wake `Notify` live inside the container, reachable only over
//! `HIVE_AGENT_SOCKET` — and the operator explicitly ruled out core
//! wiring for this. Running in-process also means we skip the socket
//! entirely and call [`crate::todos::Todos`] directly.
//!
//! Anti-nag: the todo is keyed (`disk`/[`TODO_KEY`]), and the summary is
//! deliberately *stable* — the percentage is bucketed and no raw byte
//! counts appear in it. An unchanged summary makes `upsert` report
//! `changed == false`, so a steadily-full disk sits quiet in the
//! loose-ends list instead of waking the agent every tick. Only crossing
//! into a new bucket (or a change in which directories are big enough to
//! list) speaks up again. Dropping back under the threshold clears the
//! todo. Acknowledging the todo (`cancel_loose_end`) does not defeat this —
//! `Todos::mark_done` acks the row rather than deleting it, so the next
//! probe still has the prior summary to compare against instead of seeing
//! an empty table and re-announcing an unchanged condition as new.
//!
//! Scoped by design: the `statvfs` reads the *whole filesystem*, which on
//! a shared host volume (several agents' state dirs on the same
//! subvolume) can sit over threshold because of bytes some other agent
//! owns. The todo only fires when this agent's own tree ([`big_dirs`])
//! actually contains something big enough to name — an agent whose own
//! footprint is negligible has nothing it can safely free, so it stays
//! silent instead of nagging everyone on a crowded volume.

use std::ffi::CString;
use std::fmt::Write as _;
use std::os::unix::ffi::OsStrExt as _;
use std::os::unix::fs::MetadataExt as _;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::Notify;

use crate::todos::Todos;

/// How often the filesystem is probed. The `statvfs` is free; the
/// directory scan behind it only runs once we're already over threshold.
const CHECK_INTERVAL: Duration = Duration::from_mins(15);
/// Percent-used past which the agent gets a todo. Matches the operator's
/// ask ("more than 80%").
const WARN_PCT: u64 = 80;
/// Percentages are reported in buckets this wide. This is the anti-nag
/// knob: a disk drifting 89% → 91% keeps the same summary and stays
/// silent; a jump to 95% earns a fresh wake.
const BUCKET_PCT: u64 = 5;
/// Producing subsystem marker for the todo row.
const SUBSYSTEM: &str = "disk";
/// Fixed dedup key — there is only ever one disk todo per agent.
const TODO_KEY: &str = "usage";
/// Only directories at least this large are worth naming. Coarse on
/// purpose: the listed set shouldn't churn while a build runs.
const BIG_DIR_BYTES: u64 = 5 << 30;
/// At most this many directories are named in the todo.
const MAX_LISTED: usize = 5;
/// Directories deeper than this (relative to a scan root) are summed into
/// their parent but never listed by name — a todo pointing at
/// `…/target/debug/build/foo-1a2b/out` helps nobody.
const REPORT_DEPTH: usize = 4;
/// Hard recursion cap, independent of [`REPORT_DEPTH`]: bounds stack
/// depth on a pathological tree.
const MAX_DEPTH: usize = 24;
/// Directory entries the scan will look at before giving up. Bounds the
/// worst case on a huge tree; a truncated scan just names fewer dirs.
const SCAN_BUDGET: u64 = 400_000;

/// Background loop: periodically reconcile the disk todo. Detached task —
/// runs for the harness's lifetime; errors are logged, never fatal.
pub async fn run(todos: Arc<Todos>, wake: Arc<Notify>) {
    loop {
        // Sleep first: boot is the busiest the container gets, and a disk
        // that's been full for a week can wait one interval.
        tokio::time::sleep(CHECK_INTERVAL).await;
        // The scan walks the agent's tree — blocking work, off the runtime.
        let summary = match tokio::task::spawn_blocking(probe).await {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!(error = %e, "disk watch probe panicked");
                continue;
            }
        };
        reconcile(&todos, &wake, summary.as_deref());
    }
}

/// Apply one probe result to the todo store: raise/refresh while over
/// threshold, clear once back under. Split from [`run`] so the store
/// interaction is testable without a timer.
fn reconcile(todos: &Todos, wake: &Notify, summary: Option<&str>) {
    match summary {
        Some(summary) => match todos.upsert(SUBSYSTEM, Some(TODO_KEY), summary, None, false) {
            // Only a genuine change wakes the agent — an identical summary
            // means the situation is unchanged and already in its list.
            Ok((_, true)) => wake.notify_one(),
            Ok((_, false)) => {}
            Err(e) => tracing::warn!(error = ?e, "disk todo upsert failed"),
        },
        None => {
            if let Err(e) = todos.clear(SUBSYSTEM, Some(TODO_KEY)) {
                tracing::warn!(error = ?e, "disk todo clear failed");
            }
        }
    }
}

/// One probe: measure the filesystem backing the state dir and, when it's
/// over threshold, scan the agent's own tree for the big directories worth
/// naming. `None` means "nothing to report" — under threshold, or the
/// syscall failed.
fn probe() -> Option<String> {
    let state = crate::paths::state_dir();
    let pct = used_pct(&state)?;
    if pct < WARN_PCT {
        return None;
    }
    summary_for(pct, &big_dirs())
}

/// Pure rendering half of [`probe`]: the todo text for a given usage
/// percentage + set of oversized directories. Separated so the threshold
/// and — more importantly — the *stability* of the summary are unit-tested
/// without a real filesystem.
///
/// Silent unless the agent owns something worth deleting. A full disk the
/// agent did not fill is not its problem to solve: it cannot free host
/// bytes, so the todo would only ever cost a turn to conclude "not
/// actionable". The shared store filling up is the host's signal, not an
/// agent's.
fn summary_for(pct: u64, dirs: &[PathBuf]) -> Option<String> {
    if pct < WARN_PCT || dirs.is_empty() {
        return None;
    }
    let bucket = pct / BUCKET_PCT * BUCKET_PCT;
    let mut out = format!(
        "disk over {bucket}% full on the filesystem holding your state — free up space if you safely can.\n\
         Only delete things that are actually big (tens of GiB); a few MB of notes won't move the needle.\n\
         First candidates: regenerable build output in your workspace (`target/`, `node_modules/`, `dist/`).\n\
         Never delete anything still needed. If nothing is safe to drop, tell the operator you need more space \
         rather than forcing it — be cautious with deletions in general.\n\
         Biggest directories under your own tree (`du -sh` them before removing anything):"
    );
    for dir in dirs {
        // Infallible: writing into a String.
        let _ = write!(out, "\n  - {}", dir.display());
    }
    Some(out)
}

/// Percent of the filesystem containing `path` that is in use, as `df`
/// reports it (`used / (used + available)`, so root-reserved blocks count
/// as used). `None` if the `statvfs` syscall fails.
fn used_pct(path: &Path) -> Option<u64> {
    let c_path = CString::new(path.as_os_str().as_bytes()).ok()?;
    // SAFETY: `statvfs` reads only through the valid NUL-terminated
    // `c_path` pointer and writes into the zeroed `stat` we own. The
    // return code is checked before any field is read.
    let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statvfs(c_path.as_ptr(), &raw mut stat) };
    if rc != 0 {
        return None;
    }
    let used = stat.f_blocks.checked_sub(stat.f_bfree)?;
    let capacity = used.checked_add(stat.f_bavail)?;
    if capacity == 0 {
        return None;
    }
    // Integer math on block counts — no float, so no rounding to explain.
    Some(used.saturating_mul(100) / capacity)
}

/// Scan roots: everything this agent owns and could plausibly free. The
/// `/agents/<label>` parent covers `state/` (workspaces live there) and
/// `harness/`; `$HOME` covers caches like `~/.cargo`. Roots nested inside
/// another root are dropped so nothing is counted twice.
fn roots() -> Vec<PathBuf> {
    let state = crate::paths::state_dir();
    let mut found = vec![
        state
            .parent()
            .map_or_else(|| state.clone(), Path::to_path_buf),
    ];
    if let Some(home) = std::env::var_os("HOME") {
        found.push(PathBuf::from(home));
    }
    found.retain(|p| p.is_dir());
    found.sort();
    found.dedup();
    let mut kept: Vec<PathBuf> = Vec::new();
    for root in found {
        // Sorted, so a parent is always seen before anything nested in it.
        if !kept.iter().any(|k| root.starts_with(k)) {
            kept.push(root);
        }
    }
    kept
}

/// Walk the agent's own tree and return the directories worth naming:
/// at least [`BIG_DIR_BYTES`], deepest-first (a `target/` dir, not the
/// `/agents/<label>` that contains it), capped at [`MAX_LISTED`].
fn big_dirs() -> Vec<PathBuf> {
    let mut budget = SCAN_BUDGET;
    let mut big = Vec::new();
    for root in roots() {
        // Stay on the filesystem we measured: bind mounts (`/nix`,
        // `/shared`, `/knowledge`) aren't this agent's to clean up.
        let Ok(md) = std::fs::metadata(&root) else {
            continue;
        };
        dir_bytes(&root, md.dev(), 0, &mut budget, &mut big);
    }
    let mut out = leafmost(big);
    out.truncate(MAX_LISTED);
    out
}

/// Recursive size accumulation. Pushes every directory that is both
/// shallow enough to name and at least [`BIG_DIR_BYTES`] into `big`;
/// returns the total bytes under `path`.
fn dir_bytes(path: &Path, dev: u64, depth: usize, budget: &mut u64, big: &mut Vec<PathBuf>) -> u64 {
    if depth >= MAX_DEPTH {
        return 0;
    }
    let Ok(entries) = std::fs::read_dir(path) else {
        return 0;
    };
    let mut total: u64 = 0;
    for entry in entries.flatten() {
        if *budget == 0 {
            break;
        }
        *budget -= 1;
        // `DirEntry::metadata` does not traverse symlinks, so a link into
        // the nix store is counted as the link itself, not its target.
        let Ok(md) = entry.metadata() else {
            continue;
        };
        if md.is_symlink() || md.dev() != dev {
            continue;
        }
        if md.is_dir() {
            let child = entry.path();
            let bytes = dir_bytes(&child, dev, depth + 1, budget, big);
            if depth < REPORT_DEPTH && bytes >= BIG_DIR_BYTES {
                big.push(child);
            }
            total = total.saturating_add(bytes);
        } else {
            total = total.saturating_add(md.len());
        }
    }
    total
}

/// Keep only the deepest oversized directory on each branch: if both a
/// workspace and the `target/` inside it are over threshold, name the
/// `target/`. Sorted output, so the todo summary is order-stable across
/// ticks.
fn leafmost(mut dirs: Vec<PathBuf>) -> Vec<PathBuf> {
    dirs.sort();
    let mut out: Vec<PathBuf> = Vec::new();
    for dir in dirs {
        // Lexicographic order puts a parent immediately before its
        // descendants, so the ancestors to drop are always on the tail.
        while out.last().is_some_and(|last| dir.starts_with(last)) {
            out.pop();
        }
        out.push(dir);
    }
    out
}

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

    fn p(s: &str) -> PathBuf {
        PathBuf::from(s)
    }

    #[test]
    fn under_threshold_is_silent() {
        assert!(summary_for(WARN_PCT - 1, &[p("/agents/a/state/hh/target")]).is_none());
    }

    #[test]
    fn at_threshold_speaks_up() {
        let s = summary_for(WARN_PCT, &[p("/agents/a/state/hh/target")]).expect("todo");
        assert!(s.contains("over 80% full"));
        assert!(s.contains("tens of GiB"));
        assert!(s.contains("tell the operator"));
        assert!(s.contains("/agents/a/state/hh/target"));
    }

    /// A full disk the agent did not fill is not its problem: it cannot
    /// free host bytes, so naming no directories means saying nothing.
    #[test]
    fn full_disk_with_nothing_big_of_ours_is_silent() {
        assert!(summary_for(99, &[]).is_none());
    }

    /// The anti-nag property: usage drifting inside one bucket must render
    /// byte-identically, so `Todos::upsert` reports `changed == false` and
    /// nothing wakes the agent.
    #[test]
    fn drift_inside_a_bucket_is_identical() {
        let dirs = [p("/agents/a/state/hh/target")];
        let a = summary_for(85, &dirs).expect("todo");
        let b = summary_for(89, &dirs).expect("todo");
        assert_eq!(a, b);
    }

    #[test]
    fn crossing_a_bucket_changes_the_summary() {
        let dirs = [p("/agents/a/state/hh/target")];
        let a = summary_for(89, &dirs).expect("todo");
        let b = summary_for(90, &dirs).expect("todo");
        assert_ne!(a, b);
        assert!(b.contains("over 90% full"));
    }

    #[test]
    fn big_dirs_are_named() {
        let s = summary_for(90, &[p("/agents/x/state/wt/target")]).expect("todo");
        assert!(s.contains("/agents/x/state/wt/target"));
        assert!(s.contains("du -sh"));
    }

    #[test]
    fn leafmost_drops_ancestors() {
        let got = leafmost(vec![
            p("/a/b"),
            p("/a"),
            p("/a/b/target"),
            p("/a/c"),
            p("/a/b-other"),
        ]);
        // `Path`'s `Ord` compares component-wise, so `/a/b/target` sorts
        // before `/a/b-other` (`b` < `b-other`) — which is exactly the
        // property `leafmost` leans on: a parent always lands immediately
        // before its own descendants, never separated by a sibling that
        // merely shares a name prefix.
        assert_eq!(
            got,
            vec![p("/a/b/target"), p("/a/b-other"), p("/a/c")],
            "only the deepest dir on each branch survives, and a sibling \
             sharing a name prefix is not treated as a descendant"
        );
    }

    #[test]
    fn leafmost_keeps_sibling_branches() {
        let got = leafmost(vec![p("/a"), p("/a/x/target"), p("/a/y/target")]);
        assert_eq!(got, vec![p("/a/x/target"), p("/a/y/target")]);
    }

    #[test]
    fn roots_are_disjoint() {
        for (i, a) in roots().iter().enumerate() {
            for (j, b) in roots().iter().enumerate() {
                assert!(i == j || !a.starts_with(b), "{a:?} nested in {b:?}");
            }
        }
    }

    /// The store round-trip: first raise wakes, an identical re-probe does
    /// not, and dropping back under threshold clears the row.
    #[test]
    fn reconcile_wakes_once_then_clears() {
        let dir = tempfile::tempdir().expect("tempdir");
        let todos = Todos::open(&dir.path().join("state.sqlite")).expect("open");
        let wake = Notify::new();
        let summary = summary_for(90, &[p("/agents/a/state/hh/target")]).expect("todo");

        reconcile(&todos, &wake, Some(&summary));
        assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);

        // Same summary again — still exactly one row, and (by
        // `upsert`'s contract) no wake was armed.
        reconcile(&todos, &wake, Some(&summary));
        assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);

        reconcile(&todos, &wake, None);
        assert!(todos.list(Some(SUBSYSTEM)).expect("list").is_empty());
    }
}