//! Garbage collection (spec §21): delete objects

//! nothing reachable refers to.

//!

//! Roots: the current view, the views of every operation newer than

//! `core.op-retention` (their working copies only within the shorter

//! `core.snapshot-retention`), reference attestations, and the policy; receipts

//! live as long as the changes of their state. 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, RetiredReport};

use std::collections::HashSet;

#[derive(Debug, Clone, Default, PartialEq, Eq)]

pub struct GcReport {

    pub packed: PackReport,

    pub pruned: PruneReport,

    pub reachable: usize,

    /// Packs folded into the one a full repack writes.

    pub merged_packs: usize,

    /// Packs and loose objects earlier repacks retired, deleted now that

    /// [`crate::store::RETIRED_FOR`] has passed (oxygenhub#249).

    pub retired: RetiredReport,

}

impl Repo {

    /// Delete what earlier repacks retired more than

    /// [`crate::store::RETIRED_FOR`] ago, prune unreachable objects older

    /// than the grace period (`now` ignores it, but not the retired

    /// files' hour), then repack everything into one pack (spec §13.4).

    /// The packs and loose objects this repack replaces are retired in

    /// turn, for the next collection. A store that is not this disk's

    /// directory then folds its signatures and index entries (§13.7).

    pub fn gc(&self, now: bool) -> Result<GcReport> {

        self.gc_with(now, &mut |_| {})

    }

    /// [`Repo::gc`], reporting each phase and the packing's progress

    /// (oxygen#66), for a spinner or a log.

    pub fn gc_with(&self, now: bool, progress: crate::store::Progress) -> Result<GcReport> {

        self.gc_keeping(now, &[], progress)

    }

    /// [`Repo::gc_with`], keeping `extra` blobs and their chunks as well:

    /// objects a host stored for itself and names from outside the

    /// repository's history, such as a forge's job artifacts

    /// (oxygenhub#126).

    pub fn gc_keeping(

        &self,

        now: bool,

        extra: &[ObjectRef],

        progress: crate::store::Progress,

    ) -> Result<GcReport> {

        // Retired files first: the readers that used them have had their

        // hour, and what replaced them is in the packs read from here on.

        let retired = self

            .store

            .expire_retired(crate::store::RETIRED_FOR, progress)?;

        progress("finding what is reachable");

        let mut keep = self.reachable()?;

        for r in extra {

            if !self.store.has(r) || !keep.insert(*r) {

                continue;

            }

            if let Ok(body) = self.store.get_typed(r, ObjectType::Blob) {

                if let Ok(blob) = Blob::decode(&body) {

                    keep.extend(blob.chunks);

                }

            }

        }

        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_with(&keep, grace, progress)?;

        // Then everything left in one pack, compressed harder than a

        // commit's (spec §13.4), as `git gc` repacks.

        let (packed, merged_packs) = self

            .store

            .repack_all_with(crate::store::REPACK_ZSTD_LEVEL, progress)?;

        // Last, on a store that is not this disk's directory, every

        // signature and index entry into one list (spec §13.7, #116).

        self.store.fold_entries(progress)?;

        Ok(GcReport {

            packed,

            pruned,

            reachable: keep.len(),

            merged_packs,

            retired,

        })

    }

    /// 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. Past

        // working copies only within the shorter `core.snapshot-retention`

        // (oxygen#98): a file only ever snapshotted, such as a build output

        // a status took in, goes once they expire.

        let now_ms = Hlc::now().wall_ms;

        let cutoff = now_ms.saturating_sub(self.config.op_retention_days * 86_400_000);

        let wc_cutoff = now_ms

            .saturating_sub(self.config.snapshot_retention_days * 86_400_000)

            .max(cutoff);

        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()));

                policies.extend(op.view.policy);

            }

            if i == 0 || (self.config.snapshot_retention_days > 0 && op.hlc.wall_ms >= wc_cutoff) {

                changes.extend(op.view.workspaces.values().copied());

            }

        }

        // Checkpoints: the object, its payload blob, and the blob's chunks.

        for r in self.checkpoint_refs()? {

            if let Ok(bytes) = self.store.get_typed(&r, ObjectType::Checkpoint) {

                keep.insert(r);

                let cp = crate::objects::Checkpoint::decode(&bytes)?;

                keep.insert(cp.payload);

                if let Ok(b) = self.store.get_typed(&cp.payload, ObjectType::Blob) {

                    keep.extend(Blob::decode(&b)?.chunks);

                }

                // A version 2 payload's graph blobs and their chunks.

                for g in self.checkpoint_graph_blobs(&cp).unwrap_or_default() {

                    keep.insert(g);

                    if let Ok(b) = self.store.get_typed(&g, ObjectType::Blob) {

                        keep.extend(Blob::decode(&b)?.chunks);

                    }

                }

            }

        }

        // Tags: their states and attestations.

        let view_now = self.view()?;

        states.extend(view_now.tags.values().copied());

        for name in view_now.tags.keys() {

            if let Some((r, _)) = self.tag_attest(name)? {

                attests.push(r);

            }

        }

        // Reference attestations.

        for name in self.view()?.bookmarks.keys() {

            if let Some((r, _)) = self.latest_ref_attest(name)? {

                // With a host's acceptance of it (spec §14.7).

                if let Some(acc) = self.acceptance_of(&r)? {

                    attests.push(acc.attest.hash());

                }

                attests.push(r);

            }

        }

        // A historical view may name objects pruned under an earlier, shorter

        // retention; there is nothing to keep for those.

        for st in states {

            if keep.insert(st) {

                match self.store.get_state(&st) {

                    Ok(s) => changes.extend(s.heads),

                    Err(StoreError::NotFound(_)) => continue,

                    Err(e) => return Err(e.into()),

                }

            }

        }

        for p in policies {

            if keep.insert(p) {

                let bytes = match self.store.get_typed(&p, ObjectType::Policy) {

                    Ok(b) => b,

                    Err(StoreError::NotFound(_)) => continue,

                    Err(e) => return Err(e.into()),

                };

                let policy = crate::policy::Policy::decode(&bytes)?;

                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;

            }

            // Beyond the shallow boundary, or pruned under an earlier

            // retention, there is nothing to keep.

            let c = match self.store.get_change_parts(&h) {

                Ok((c, texts)) => {

                    keep.extend(texts);

                    c

                }

                Err(StoreError::NotFound(_)) => continue,

                Err(e) => return Err(e.into()),

            };

            stack.extend(c.deps.iter().copied());

            if !c.author.is_zero() {

                keep.insert(c.author);

            }

            for sig in self.store.signatures(&h)? {

                self.keep_signer(&mut keep, 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)?);

        }

        self.reachable_receipts(&mut keep)?;

        self.reachable_approvals(&mut keep)?;

        for a in attests {

            if keep.insert(a) {

                for sig in self.store.signatures(&a)? {

                    self.keep_signer(&mut keep, sig.key)?;

                }

            }

        }

        Ok(keep)

    }

    /// Keep a signer's identity and, for a short-lived key, its

    /// certificate and the certificate's issuer (spec §14.7).

    pub(crate) fn keep_signer(&self, keep: &mut HashSet<ObjectRef>, key: ObjectRef) -> Result<()> {

        keep.insert(key);

        if let Some(c) = self.store.certificate_of(&key)? {

            keep.insert(c);

            if let Ok(cert) = self.store.get_certificate(&c) {

                keep.insert(cert.cert.issuer);

            }

        }

        Ok(())

    }

}