//! conservative CAS garbage collection (packet §16, §20).
//!
//! every published generation is retained and participates in the CAS mark phase.
//! nginx has no reliable acknowledgement that workers using an older generation have
//! drained, so `keep_generations` cannot safely authorize deletion yet — this deliberately
//! trades unbounded generation storage for worker safety.
//!
//! manifest discovery and validation finish before the CAS sweep starts. an unknown,
//! malformed, mislabelled, or semantically corrupt generation aborts GC without
//! deleting anything.

use std::collections::BTreeSet;
use std::path::Path;
use std::sync::Arc;

use anyhow::Context as _;

use crate::config::ValidatedConfig;
use crate::manifest::GenerationManifest;

pub fn run(config: &Arc<ValidatedConfig>) -> anyhow::Result<()> {
    let (mark, retained_generations) =
        mark_retained_generations(&config.output_root.join("generations"))?;
    let swept = sweep_cas(&config.output_root.join("cas/sha256"), &mark)?;

    println!("gc: retained {retained_generations} generations, swept {swept} CAS objects");
    Ok(())
}

/// Load and validate every generation before returning any authority to sweep.
fn mark_retained_generations(generations_root: &Path) -> anyhow::Result<(BTreeSet<String>, usize)> {
    let mut mark = BTreeSet::new();
    let mut retained = 0;

    for entry in std::fs::read_dir(generations_root).context("list generations")? {
        let entry = entry.context("read generation directory entry")?;
        let dir = entry.path();
        anyhow::ensure!(
            entry
                .file_type()
                .with_context(|| format!("inspect generation entry {}", dir.display()))?
                .is_dir(),
            "unknown entry in generations directory: {}",
            dir.display()
        );

        let id = entry.file_name().into_string().map_err(|_| {
            anyhow::anyhow!("generation directory name is not UTF-8: {}", dir.display())
        })?;
        let manifest = GenerationManifest::load(&dir)
            .with_context(|| format!("load retained generation {id}"))?;
        anyhow::ensure!(
            manifest.generation_id == id,
            "retained generation directory {id} contains manifest for {}",
            manifest.generation_id
        );

        // only enforce semantic hash on current-schema manifests; older schemas hash differently.
        let manifest_is_current_schema = std::fs::read_to_string(dir.join("manifest.json"))
            .ok()
            .and_then(|bytes| serde_json::from_str::<serde_json::Value>(&bytes).ok())
            .is_some_and(|value| value.get("conf_template").is_some());
        if manifest_is_current_schema {
            let computed_id = manifest.compute_id();
            anyhow::ensure!(
                computed_id == id,
                "retained generation {id} has semantic hash {computed_id}"
            );
        }

        for hash in manifest.referenced_hashes() {
            anyhow::ensure!(
                is_sha256_hex(&hash),
                "retained generation {id} references invalid CAS hash {hash:?}"
            );
            mark.insert(hash);
        }
        retained += 1;
    }

    Ok((mark, retained))
}

fn is_sha256_hex(hash: &str) -> bool {
    hash.len() == 64
        && hash
            .bytes()
            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
}

fn sweep_cas(cas_root: &Path, mark: &BTreeSet<String>) -> anyhow::Result<usize> {
    if !cas_root.exists() {
        return Ok(0);
    }

    let mut swept = 0;
    for shard in std::fs::read_dir(cas_root).context("list cas shards")? {
        let shard = shard.context("read cas shard entry")?;
        if !shard.file_type()?.is_dir() {
            continue;
        }
        let shard_prefix = shard.file_name();
        for object in std::fs::read_dir(shard.path())? {
            let object = object?;
            let hex = format!(
                "{}{}",
                shard_prefix.to_string_lossy(),
                object.file_name().to_string_lossy()
            );
            if mark.contains(&hex) {
                continue;
            }
            if object.file_type()?.is_dir() {
                std::fs::remove_dir_all(object.path())?;
            } else {
                std::fs::remove_file(object.path())?;
            }
            swept += 1;
        }
    }

    for shard in std::fs::read_dir(cas_root)? {
        let shard = shard?;
        if shard.file_type()?.is_dir() {
            let _ = std::fs::remove_dir(shard.path());
        }
    }

    Ok(swept)
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::path::Path;
    use std::sync::Arc;

    use crate::cas::{Cas, Digest};
    use crate::config::ValidatedConfig;
    use crate::manifest::{GENERATOR_PROFILE_VERSION, GenerationManifest, StaticResponse};

    fn config(root: &Path, keep_generations: usize) -> Arc<ValidatedConfig> {
        let config_path = root.join("config.toml");
        std::fs::write(
            &config_path,
            format!(
                r#"
[knot]
hostname = "git.example.com"
owner = "did:plc:operator"

[repo]
scan_path = "{root}/repos"
default_branch = "main"
object_format = "sha256"

[secrets]
sealed_key_file = "{root}/secrets.json"

[output]
root = "{root}"
keep_generations = {keep_generations}
"#,
                root = root.display(),
            ),
        )
        .unwrap();
        Arc::new(ValidatedConfig::load(&config_path).unwrap())
    }

    fn publish_generation(root: &Path, body_hash: &str) -> String {
        let mut routes = BTreeMap::new();
        routes.insert(
            "sh.tangled.repo.getBlob?repo=x&ref=y&path=z".to_string(),
            StaticResponse {
                body_hash: body_hash.to_string(),
                status: 200,
                content_type: "application/json".to_string(),
                headers: Vec::new(),
                cas_key: body_hash.to_string(),
            },
        );
        let mut manifest = GenerationManifest {
            generation_id: String::new(),
            generator_profile: GENERATOR_PROFILE_VERSION.to_string(),
            conf_template: crate::manifest::NGINX_TEMPLATE_VERSION.to_string(),
            knot_did: "did:web:git.example.com".to_string(),
            hostname: "git.example.com".to_string(),
            owner: "did:plc:operator".to_string(),
            static_version: "test".to_string(),
            capabilities: Vec::new(),
            profile: serde_json::json!({}),
            repos: Vec::new(),
            routes,
            sections: BTreeMap::new(),
            section_routes: BTreeMap::new(),
            section_objects: BTreeMap::new(),
            repo_aliases: BTreeMap::new(),
            ref_aliases: BTreeMap::new(),
            stats: Default::default(),
        };
        manifest.finalize_id();
        let id = manifest.generation_id.clone();
        let dir = root.join("generations").join(&id);
        std::fs::create_dir_all(&dir).unwrap();
        let cas = Cas::open(root.join("cas"));
        manifest.write(&dir, &cas).unwrap();
        id
    }

    fn put(cas: &Cas, bytes: &[u8]) -> (Digest, std::path::PathBuf) {
        let digest = cas.put(bytes).unwrap();
        let path = cas.path_for(&digest);
        (digest, path)
    }

    #[test]
    fn keep_one_still_preserves_prior_generation_and_its_cas() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path();
        std::fs::create_dir_all(root.join("generations")).unwrap();
        let cas = Cas::open(root.join("cas"));
        let (prior_hash, prior_object) = put(&cas, b"prior");
        let (current_hash, current_object) = put(&cas, b"current");
        let prior = publish_generation(root, &prior_hash.hex());
        let current = publish_generation(root, &current_hash.hex());

        super::run(&config(root, 1)).unwrap();

        assert!(root.join("generations").join(prior).is_dir());
        assert!(root.join("generations").join(current).is_dir());
        assert!(prior_object.is_file());
        assert!(current_object.is_file());
    }

    #[test]
    fn malformed_manifest_aborts_before_any_sweep() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path();
        std::fs::create_dir_all(root.join("generations/broken")).unwrap();
        std::fs::write(root.join("generations/broken/manifest.json"), b"not json").unwrap();
        let cas = Cas::open(root.join("cas"));
        let (_, orphan) = put(&cas, b"orphan");

        let error = super::run(&config(root, 1)).unwrap_err();

        assert!(format!("{error:#}").contains("load retained generation broken"));
        assert!(orphan.is_file());
    }

    #[test]
    fn orphan_sweeps_after_complete_valid_mark_phase() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path();
        std::fs::create_dir_all(root.join("generations")).unwrap();
        let cas = Cas::open(root.join("cas"));
        let (first_hash, first_object) = put(&cas, b"first retained");
        let (second_hash, second_object) = put(&cas, b"second retained");
        let (_, orphan) = put(&cas, b"orphan");
        publish_generation(root, &first_hash.hex());
        publish_generation(root, &second_hash.hex());

        super::run(&config(root, 1)).unwrap();

        assert!(first_object.is_file());
        assert!(second_object.is_file());
        assert!(!orphan.exists());
    }
}