//! Garbage collection (spec §21): pack loose objects, then delete objects
//! nothing reachable refers to.
//!
//! Roots: the current view, the views of every operation newer than
//! `core.op-retention`, reference attestations, and the policy. Operation
//! objects are always kept so the log stays walkable, but an operation older
//! than the retention window no longer keeps its states alive, so undoing
//! that far back may fail once its objects are gone.

use super::*;
use crate::store::{PackReport, PruneReport};
use std::collections::HashSet;

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct GcReport {
    pub packed: PackReport,
    pub pruned: PruneReport,
    pub reachable: usize,
}

impl Repo {
    /// Pack loose objects and prune unreachable ones older than the grace
    /// period (`now` ignores the grace period).
    pub fn gc(&self, now: bool) -> Result<GcReport> {
        let packed = self.store.pack_loose()?;
        let keep = self.reachable()?;
        let grace = if now { std::time::Duration::ZERO } else { std::time::Duration::from_secs(self.config.gc_grace_days * 86_400) };
        let pruned = self.store.prune(&keep, grace)?;
        Ok(GcReport { packed, pruned, reachable: keep.len() })
    }

    /// Every object that must survive garbage collection.
    pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {
        let mut keep: HashSet<ObjectRef> = HashSet::new();
        let mut states: Vec<ObjectRef> = Vec::new();
        let mut changes: Vec<ChangeHash> = Vec::new();
        let mut attests: Vec<ObjectRef> = Vec::new();
        let mut policies: Vec<ObjectRef> = Vec::new();

        // Operations: always kept; views within retention are roots.
        let cutoff = Hlc::now().wall_ms.saturating_sub(self.config.op_retention_days * 86_400_000);
        for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {
            keep.insert(r);
            if i == 0 || op.hlc.wall_ms >= cutoff {
                states.extend(op.view.bookmarks.values().copied());
                states.extend(op.view.remotes.values().flat_map(|b| b.values().copied()));
                changes.extend(op.view.workspaces.values().copied());
                policies.extend(op.view.policy);
            }
        }
        // Reference attestations.
        for name in self.view()?.bookmarks.keys() {
            if let Some((r, _)) = self.latest_ref_attest(name)? {
                attests.push(r);
            }
        }
        for st in states {
            if keep.insert(st) {
                changes.extend(self.store.get_state(&st)?.heads);
            }
        }
        for p in policies {
            if keep.insert(p) {
                let policy = crate::policy::Policy::decode(&self.store.get_typed(&p, ObjectType::Policy)?)?;
                for id in &policy.identities {
                    keep.insert(id.hash());
                }
                for sig in self.store.signatures(&p)? {
                    keep.insert(sig.key);
                }
            }
        }
        // Changes: their closure, identities, blobs, chunks, seals.
        let mut stack = changes;
        while let Some(h) = stack.pop() {
            if !keep.insert(h) {
                continue;
            }
            let c = self.store.get_change(&h)?;
            stack.extend(c.deps.iter().copied());
            if !c.author.is_zero() {
                keep.insert(c.author);
            }
            for sig in self.store.signatures(&h)? {
                keep.insert(sig.key);
            }
            for e in &c.edits {
                if let Edit::Replace { old, new, .. } = e {
                    for b in [old, new].into_iter().flatten() {
                        if keep.insert(*b) {
                            if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {
                                keep.extend(Blob::decode(&bytes)?.chunks);
                            }
                        }
                    }
                }
            }
            attests.extend(self.store.seals_for(&h)?);
        }
        for a in attests {
            if keep.insert(a) {
                for sig in self.store.signatures(&a)? {
                    keep.insert(sig.key);
                }
            }
        }
        Ok(keep)
    }
}