//! Reclaim the mission tree on the gateway.
//!
//! # Why this is filesystem-first
//!
//! `cleanup_sweeper` prunes ROWS. Deleting a row does not delete a directory,
//! and the reaper that was supposed to — `mission_runtime::teardown_container` —
//! only runs while a mission still exists to tear down. So a mission deleted by
//! any path that did not go through teardown left its directory behind forever,
//! and the gateway is the smallest disk in the fleet (150 GB, shared with
//! postgres and every checkout).
//!
//! The DB is therefore the PREDICATE here, never the enumerator: this walks the
//! filesystem and asks the database about what it finds. Enumerating from the
//! database is precisely how the orphans became invisible — a directory whose
//! row is gone is exactly the one a row-driven sweep cannot see.
//!
//! # Why deletion needs two attempts
//!
//! The server runs as uid 65532. Almost everything under a mission belongs to
//! 65532 now, but the per-mission ZeroClaw daemon still runs as root and leaves
//! ~26 of its own files (`.claude.json`, session jsonl). `remove_dir_all` then
//! fails with `PermissionDenied` and the directory survives — the
//! cleanup-that-cannot-clean-up shape, at a scale small enough to go unnoticed.
//! So a failed removal falls back to `root_copy::purge`, which deletes from
//! inside the runtime container as root.
//!
//! # What it will not touch
//!
//! Anything belonging to a mission that still has a row, and anything younger
//! than the grace window. A mission directory is created BEFORE its row is
//! committed in some paths, and reaping a directory out from under a launching
//! mission would be a far worse bug than the leak this fixes.

use std::path::Path;
use std::time::Duration;

use sqlx::PgPool;

/// How long a directory must have been untouched before it is considered
/// abandoned. Generously long: the cost of waiting is disk, and the cost of
/// being wrong is deleting a live mission's checkout.
const ORPHAN_GRACE: Duration = Duration::from_secs(2 * 60 * 60);

/// Retention for captured outputs (`_outputs`), which are artifacts a user can
/// still open. Mirrors `TOPOLOGY_RUNS_DAYS` in `cleanup_sweeper` — the run
/// history and the files it points at should not outlive each other.
const OUTPUTS_DAYS: u64 = 90;

/// Scratch trees the mission machinery makes and is supposed to remove itself:
/// `_bench`, `_gate`, `_verify`, `_merge`. Anything older than this is debris
/// from a crashed or killed run, not work in progress — every command that
/// creates one is bounded well below it.
const SCRATCH_GRACE: Duration = Duration::from_secs(6 * 60 * 60);

/// Directories under the missions root that are NOT missions.
const RESERVED: &[&str] = &["_outputs", "_home", "_cargo", "_mirrors"];

pub fn spawn(pool: PgPool, interval: Duration) {
    tokio::spawn(async move {
        // Not on the first tick. A sweep racing the server's own startup — while
        // `start_pending_phases` is still adopting in-flight missions — is the
        // one moment its "no row for this directory" predicate is least
        // trustworthy.
        tokio::time::sleep(Duration::from_secs(120)).await;
        let mut tick = tokio::time::interval(interval);
        loop {
            tick.tick().await;
            match sweep_once(&pool).await {
                Ok(r) if r.is_empty() => {}
                Ok(r) => eprintln!("mission_gc: {r}"),
                Err(e) => eprintln!("mission_gc: sweep failed: {e}"),
            }
        }
    });
}

/// What one sweep reclaimed.
#[derive(Debug, Default, PartialEq)]
pub struct Reclaimed {
    pub orphan_dirs: u64,
    pub scratch_dirs: u64,
    pub outputs: u64,
    pub bytes: u64,
    /// Directories we tried and failed to remove. Reported rather than swallowed
    /// — a GC that cannot collect is the thing being fixed.
    pub failed: u64,
    /// Rows swept from `mission_events`.
    pub events: u64,
}

impl Reclaimed {
    pub fn is_empty(&self) -> bool {
        *self == Reclaimed::default()
    }
}

impl std::fmt::Display for Reclaimed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "reclaimed {} orphan mission dir(s), {} scratch dir(s), {} output(s), \
             {} mission event(s), {:.1} MiB{}",
            self.orphan_dirs,
            self.scratch_dirs,
            self.outputs,
            self.events,
            self.bytes as f64 / (1024.0 * 1024.0),
            if self.failed > 0 {
                format!(" — {} COULD NOT BE REMOVED", self.failed)
            } else {
                String::new()
            }
        )
    }
}

async fn sweep_once(pool: &PgPool) -> Result<Reclaimed, String> {
    let root = crate::mission_workspace::missions_root();
    let mut out = Reclaimed::default();
    reap_orphan_missions(pool, &root, &mut out).await?;
    reap_scratch(&root, &mut out).await;
    reap_outputs(pool, &root, &mut out).await;
    reap_mission_events(pool, &mut out).await;
    Ok(out)
}

/// How long a mission's structured activity is kept.
///
/// The World shows the last 24 hours of finished missions, so a week is
/// generous and still bounds a table that a single busy coding phase can add
/// hundreds of rows to. The per-phase cap bounds ONE phase; this bounds time.
const EVENT_RETENTION_DAYS: i32 = 7;

/// Sweep expired `mission_events`.
///
/// Bounded per pass rather than deleting the whole backlog in one statement: a
/// deployment that has been accumulating for months would otherwise take a long
/// lock on its first sweep after this ships. The sweep runs on a timer, so a
/// large backlog simply drains over several passes.
async fn reap_mission_events(pool: &PgPool, out: &mut Reclaimed) {
    let res = sqlx::query(
        "DELETE FROM mission_events
          WHERE id IN (
              SELECT id FROM mission_events
               WHERE created_at < now() - make_interval(days => $1)
               LIMIT 10000
          )",
    )
    .bind(EVENT_RETENTION_DAYS)
    .execute(pool)
    .await;
    match res {
        Ok(r) => out.events += r.rows_affected(),
        Err(e) => eprintln!("mission_gc: sweeping mission_events failed: {e}"),
    }
}

/// Directories under the missions root with no mission row.
async fn reap_orphan_missions(
    pool: &PgPool,
    root: &Path,
    out: &mut Reclaimed,
) -> Result<(), String> {
    let Ok(entries) = std::fs::read_dir(root) else {
        // Not an error: a deployment that has never run a mission has no tree.
        return Ok(());
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
            continue;
        };
        if RESERVED.contains(&name) || name.starts_with('_') {
            continue;
        }
        // Only well-formed mission ids. A directory this function does not
        // recognise is one it has no business deleting.
        let Ok(id) = name.parse::<uuid::Uuid>() else {
            continue;
        };
        if !older_than(&path, ORPHAN_GRACE) {
            continue;
        }
        // The DB as predicate, asked per directory.
        let exists: Option<(uuid::Uuid,)> =
            sqlx::query_as("SELECT id FROM missions WHERE id = $1")
                .bind(id)
                .fetch_optional(pool)
                .await
                .map_err(|e| format!("looking up mission {id}: {e}"))?;
        if exists.is_some() {
            continue;
        }
        let bytes = dir_size(&path);
        if remove_tree(&path).await {
            out.orphan_dirs += 1;
            out.bytes += bytes;
        } else {
            out.failed += 1;
        }
    }
    Ok(())
}

/// `_bench` / `_gate` / `_verify` / `_merge` trees older than their command
/// ceilings. These are siblings of the per-mission dirs and have leaked before.
async fn reap_scratch(root: &Path, out: &mut Reclaimed) {
    const SCRATCH: &[&str] = &["_bench", "_gate", "_verify", "_merge"];
    for name in SCRATCH {
        let path = root.join(name);
        if !path.is_dir() {
            continue;
        }
        let Ok(entries) = std::fs::read_dir(&path) else {
            continue;
        };
        for entry in entries.flatten() {
            let p = entry.path();
            if !older_than(&p, SCRATCH_GRACE) {
                continue;
            }
            let bytes = dir_size(&p);
            if remove_tree(&p).await {
                out.scratch_dirs += 1;
                out.bytes += bytes;
            } else {
                out.failed += 1;
            }
        }
    }
}

/// Captured outputs past retention, with their artifact rows marked so nothing
/// points at a file that is gone.
async fn reap_outputs(pool: &PgPool, root: &Path, out: &mut Reclaimed) {
    let outputs = root.join("_outputs");
    let Ok(entries) = std::fs::read_dir(&outputs) else {
        return;
    };
    let grace = Duration::from_secs(OUTPUTS_DAYS * 24 * 60 * 60);
    for entry in entries.flatten() {
        let p = entry.path();
        if !p.is_dir() || !older_than(&p, grace) {
            continue;
        }
        let Some(id) = p
            .file_name()
            .and_then(|n| n.to_str())
            .and_then(|n| n.parse::<uuid::Uuid>().ok())
        else {
            continue;
        };
        let bytes = dir_size(&p);
        if !remove_tree(&p).await {
            out.failed += 1;
            continue;
        }
        // The row is marked only AFTER the files are gone. The other order
        // leaves a mission whose artifacts claim to be reaped while they are
        // still on disk, which is a lie in the direction that costs disk.
        let _ = sqlx::query(
            "UPDATE mission_artifacts SET metadata = COALESCE(metadata, '{}'::jsonb)
                 || '{\"reaped\": true}'::jsonb
               WHERE mission_id = $1",
        )
        .bind(id)
        .execute(pool)
        .await;
        out.outputs += 1;
        out.bytes += bytes;
    }
}

/// Remove a tree, escalating to a root purge when our uid cannot.
///
/// The ONLY deletion path in this module. A second one is how the reap paths
/// drifted apart last time.
async fn remove_tree(path: &Path) -> bool {
    match tokio::fs::remove_dir_all(path).await {
        Ok(()) => true,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => true,
        Err(e) if e.kind() == std::io::ErrorKind::PermissionDenied => {
            let container = std::env::var("CLAWMATES_RUNTIME_CONTAINER")
                .unwrap_or_else(|_| "clawmates-runtime".to_string());
            crate::root_copy::purge(&container, path).await;
            let gone = tokio::fs::metadata(path).await.is_err();
            if !gone {
                eprintln!(
                    "mission_gc: {} survived a root purge — it will keep accumulating",
                    path.display()
                );
            }
            gone
        }
        Err(e) => {
            eprintln!("mission_gc: could not remove {}: {e}", path.display());
            false
        }
    }
}

fn older_than(path: &Path, grace: Duration) -> bool {
    let Ok(meta) = std::fs::metadata(path) else {
        return false;
    };
    // mtime, not ctime: a directory whose contents changed recently is one
    // something is still writing to.
    let Ok(modified) = meta.modified() else {
        return false;
    };
    modified
        .elapsed()
        .map(|age| age >= grace)
        .unwrap_or(false)
}

/// Apparent size, best-effort. Used only for reporting, so a read error costs a
/// wrong number in a log line rather than a wrong decision.
fn dir_size(path: &Path) -> u64 {
    let mut total = 0;
    let Ok(entries) = std::fs::read_dir(path) else {
        return 0;
    };
    for entry in entries.flatten() {
        let Ok(meta) = entry.metadata() else { continue };
        if meta.is_dir() {
            total += dir_size(&entry.path());
        } else {
            total += meta.len();
        }
    }
    total
}

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

    fn touch_dir(root: &Path, name: &str) -> std::path::PathBuf {
        let p = root.join(name);
        std::fs::create_dir_all(&p).unwrap();
        std::fs::write(p.join("f"), b"x").unwrap();
        p
    }

    /// The reserved siblings are never candidates.
    ///
    /// `_outputs`, `_home` and `_cargo` live under the same root as the mission
    /// directories. `_cargo` in particular is a SHARED cache every mission
    /// writes to, so a sweep that treated an underscore-prefixed sibling as an
    /// orphan mission would delete it out from under running work — and it would
    /// look like a slow cargo build rather than a bug.
    #[test]
    fn siblings_of_the_mission_dirs_are_not_missions() {
        for name in RESERVED {
            assert!(
                name.starts_with('_'),
                "{name} must be underscore-prefixed so the guard catches it"
            );
            assert!(
                name.parse::<uuid::Uuid>().is_err(),
                "{name} must not parse as a mission id"
            );
        }
    }

    /// Only a well-formed mission id is ever a candidate.
    ///
    /// The predicate is "no row exists", and a directory whose name is not an id
    /// can have no row BY CONSTRUCTION — so name-parsing has to gate the lookup,
    /// or every unrecognised directory looks like an orphan.
    #[test]
    fn a_directory_that_is_not_a_mission_id_is_never_a_candidate() {
        for name in ["_outputs", "_cargo", "lost+found", "notes", "019fe8", ""] {
            assert!(
                name.parse::<uuid::Uuid>().is_err(),
                "{name:?} must not parse as a mission id"
            );
        }
        assert!("019fe82e-7f0d-7481-a197-698f1d400419"
            .parse::<uuid::Uuid>()
            .is_ok());
    }

    /// The grace window is real, and measured from mtime.
    #[test]
    fn a_fresh_directory_is_never_old_enough() {
        let tmp = tempfile::tempdir().unwrap();
        let d = touch_dir(tmp.path(), "019fe82e-7f0d-7481-a197-698f1d400419");
        assert!(!older_than(&d, ORPHAN_GRACE));
        // And a zero grace makes everything eligible, which is what proves the
        // check is the window rather than an accident of the filesystem.
        assert!(older_than(&d, Duration::from_secs(0)));
    }

    /// One deletion path, and it escalates.
    ///
    /// A second removal site is how the container reap paths drifted apart and
    /// leaked for a day. The escalation is the other half: the server is uid
    /// 65532 and cannot delete what the per-mission daemon left as root.
    #[test]
    fn there_is_exactly_one_deletion_path_and_it_escalates() {
        let src = include_str!("mission_gc.rs");
        assert_eq!(
            src.matches(concat!("remove_dir", "_all(")).count(),
            1,
            "exactly one removal site"
        );
        assert!(src.contains("root_copy::purge"), "and it must escalate");
    }
}