girt 0.11.0

use std::fs::{self, File};
use std::io::{self, BufWriter, Read};
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};

use tempfile::NamedTempFile;

use super::{RetentionError, RetentionPolicy};
use crate::pack::{
    PackObject, PackWriteError, PackWriteLimits, PackWritten, Packed, ReuseError, write_reusing,
};
use crate::{DeltaOptions, ObjectId, ObjectReadError, PackCompression, Repository};

/// Bounds one additive repack.
///
/// Entries copied from existing packs stream from their files. Objects that are not packed yet
/// remain decoded in memory until the pack is written.
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct RepackLimits {
    /// Bounds object count, logical payload bytes (including copied objects), and output
    /// artifacts.
    pub write: PackWriteLimits,
    /// Selects delta search among objects not already packed (default: bounded deltas).
    pub compression: PackCompression,
}

impl RepackLimits {
    /// Sets [`Self::write`].
    #[must_use]
    pub fn with_write(mut self, write: PackWriteLimits) -> Self {
        self.write = write;
        self
    }

    /// Sets [`Self::compression`].
    #[must_use]
    pub fn with_compression(mut self, compression: PackCompression) -> Self {
        self.compression = compression;
        self
    }
}

impl Default for RepackLimits {
    fn default() -> Self {
        Self {
            write: PackWriteLimits::default(),
            compression: PackCompression::Delta(DeltaOptions::DEFAULT),
        }
    }
}

impl RepackLimits {
    /// Returns limits for a repository the caller trusts as Git trusts the user's own.
    ///
    /// Git bounds none of these values. The defaults suit untrusted input; these are large enough
    /// that only available memory bounds the work, while keeping headroom for internal arithmetic.
    pub const fn trusted() -> Self {
        Self {
            write: PackWriteLimits::trusted(),
            compression: PackCompression::Delta(DeltaOptions::DEFAULT),
        }
    }
}

/// A complete indexed pack is visible; existing storage has not been removed.
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct RepackPublished {
    /// Identity, count and artifact sizes of the newly published or reused pair.
    pub written: PackWritten,
}

/// Failure before complete pair publication. An unindexed pack may remain after interruption.
///
/// This operation leaves old storage intact. An incomplete plan needs its typed retention cause
/// resolved; missing/corrupt input needs recovery. A publication failure may leave an unindexed
/// pack that readers ignore; do not manually delete checksum-named files without fresh ownership
/// and dependency checks. A successful additive repack is never deletion authorization.
///
/// ```
/// use girt::{InitKind, ObjectFormat, Repository};
/// use girt::retention::{RepackError, RepackLimits, RetentionError, RetentionPolicy};
///
/// let directory = tempfile::tempdir()?;
/// let path = directory.path().join("r");
/// let repo = Repository::init(ObjectFormat::Sha1, path, InitKind::Bare)?;
///
/// let missing = girt::ObjectId::for_blob(ObjectFormat::Sha1, b"not stored");
/// let policy = RetentionPolicy::default().with_heads(vec![missing]);
/// let error = repo.repack_retained(&policy, RepackLimits::default(),
///     &std::sync::atomic::AtomicBool::new(false)).unwrap_err();
/// assert!(matches!(
///     error,
///     RepackError::Incomplete(RetentionError::MissingObject(id)) if id == missing
/// ));
/// // No complete plan existed, so no replacement pair could be published.
/// # Ok::<(), Box<dyn std::error::Error>>(())
/// ```
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum RepackError {
    /// A fresh retention scan did not establish complete reachability.
    #[error("retention scan incomplete")]
    Incomplete(#[source] RetentionError),
    /// A required object disappeared between planning and reading.
    #[error("required object missing: {0}")]
    Missing(ObjectId),
    /// A required object could not be read.
    #[error("cannot read required object {id}")]
    Read {
        /// Object requested by the plan.
        id: ObjectId,
        /// Storage error.
        #[source]
        source: ObjectReadError,
    },
    /// Object storage changed or could not be opened.
    #[error(transparent)]
    Open(#[from] ObjectReadError),
    /// Input or output exceeded the configured bounds, or artifact writing failed.
    #[error(transparent)]
    Write(#[from] PackWriteError),
    /// The caller cancelled before complete pair publication.
    #[error("repack cancelled")]
    Cancelled,
    /// An existing checksum-named artifact differs from the generated bytes.
    #[error("conflicting existing pack artifact: {0}")]
    Existing(std::path::PathBuf),
    /// Temporary artifact creation, synchronization or publication failed.
    #[error(transparent)]
    Io(#[from] io::Error),
}

impl Repository {
    /// Publishes a pack of objects required by a fresh complete retention scan.
    ///
    /// This additive operation rescans roots and recent objects instead of trusting an older
    /// [`super::RetentionPlan`]. A required object already in a pinned pack has its entry copied
    /// without inflating it, in source pack order; a delta keeps its compressed instructions when
    /// its base entry from the same pack is also copied, and is otherwise rebuilt as a whole
    /// object. The scan decodes graph-bearing objects and locates blobs; the writer validates
    /// copied entry CRCs and bounds, while retirement separately verifies surviving identities.
    /// Stored entries come only from primary
    /// packs; alternate objects use the ordinary reader. Additive repack includes protected
    /// objects; retirement can instead verify a protected survivor and omit its replacement copy.
    /// Other required objects are read
    /// and identity-checked, then appended with deltas against each other under
    /// [`RepackLimits::compression`]. A repository whose only pack holds exactly the required
    /// objects can therefore reproduce that pack byte for byte. It writes two bounded temporary
    /// files in the primary pack directory, synchronizes each file, then publishes the pack
    /// before its index. The index is the visibility marker. An existing checksum-named file is
    /// reused only when every byte matches. Old packs, loose objects, reflogs and alternates
    /// remain untouched, including protected packs. Readers with pinned generations keep their
    /// handles; new readers discover the indexed pair and existing readers can refresh
    /// explicitly.
    ///
    /// The caller must exclude cooperating ref, reflog and object writers when it requires a
    /// stable maintenance observation. Girt has no lock that excludes arbitrary external Git
    /// writers. This operation never deletes data, so a concurrent new root cannot make it prune
    /// that root's object. An external writer that removes or mutates input files can still make
    /// the scan/read fail. Do not use this result as deletion authorization. Destructive
    /// maintenance must rescan under its own execution boundary and account for old pinned
    /// readers before removing files.
    ///
    /// # Errors
    ///
    /// Incomplete scans, missing/corrupt objects, limits, I/O and cancellation fail without
    /// replacing an existing artifact. Failure after pack publication can leave an unindexed pack;
    /// readers ignore it and a retry may reuse it. Cancellation after index publication reports
    /// success. Temporary files are cleaned up on ordinary failure but can survive process death.
    /// On Unix, the pack directory is synchronized before index publication. Additive
    /// publication does not promise durability of the index name; retirement synchronizes both
    /// names before removing old storage. Publication assumes a trusted local filesystem with no
    /// in-place artifact edits.
    ///
    /// # Add a pack without deleting loose objects
    ///
    /// ```
    /// use std::sync::atomic::AtomicBool;
    ///
    /// use girt::{InitKind, ObjectFormat, Repository};
    /// let directory = tempfile::tempdir()?;
    /// let repo = Repository::init(
    ///     ObjectFormat::Sha1,
    ///     directory.path().join("r"),
    ///     InitKind::Bare,
    /// )?;
    /// let cancel = AtomicBool::new(false);
    ///
    /// use girt::retention::{RepackLimits, RetentionPolicy};
    /// let id = repo.loose_objects().write_blob(b"keep me")?;
    /// let published = repo.repack_retained(
    ///     &RetentionPolicy::default(),
    ///     RepackLimits::default(),
    ///     &cancel,
    /// )?;
    /// assert_eq!(published.written.objects, 1);
    /// assert_eq!(repo.loose_objects().read_blob(id, 1024)?, b"keep me");
    /// # Ok::<(), Box<dyn std::error::Error>>(())
    /// ```
    pub fn repack_retained(
        &self,
        policy: &RetentionPolicy,
        limits: RepackLimits,
        cancel: &AtomicBool,
    ) -> Result<RepackPublished, RepackError> {
        self.repack_retained_after_plan(policy, limits, cancel, || {})
    }

    // The checkpoint makes the scan/write boundary deterministic in concurrency tests.
    fn repack_retained_after_plan(
        &self,
        policy: &RetentionPolicy,
        limits: RepackLimits,
        cancel: &AtomicBool,
        after_plan: impl FnOnce(),
    ) -> Result<RepackPublished, RepackError> {
        let _headers = crate::pack::HeaderScope::new();
        check(cancel)?;
        let mut plan = self.plan_retention(policy, cancel);
        if let super::RetentionOutcome::Incomplete(reason) = plan.outcome {
            return Err(RepackError::Incomplete(reason));
        }
        after_plan();
        // Additive repack publishes the complete retained set. Only retirement omits survivors.
        plan.protected_packs.clear();
        self.publish_required(&plan, policy, limits, cancel)
            .map(|(published, _)| published)
    }

    /// Publishes the objects required by a complete scan of this repository.
    ///
    /// Objects whose first primary copy sits in one of the plan's protected packs are decoded from
    /// that copy to prove it readable and left out of the new pack, as `git repack -a` leaves
    /// out objects in `.keep` packs. Their identities are returned so retirement can verify the
    /// protected copy again before unlinking anything.
    pub(super) fn publish_required(
        &self,
        plan: &super::RetentionPlan,
        policy: &RetentionPolicy,
        limits: RepackLimits,
        cancel: &AtomicBool,
    ) -> Result<(RepackPublished, std::collections::BTreeSet<ObjectId>), RepackError> {
        self.publish_required_after_write(plan, policy, limits, cancel, || {})
    }

    fn publish_required_after_write(
        &self,
        plan: &super::RetentionPlan,
        policy: &RetentionPolicy,
        limits: RepackLimits,
        cancel: &AtomicBool,
        after_write: impl FnOnce(),
    ) -> Result<(RepackPublished, std::collections::BTreeSet<ObjectId>), RepackError> {
        let _headers = crate::pack::HeaderScope::new();
        check(cancel)?;
        plan.check_packs_unchanged(cancel)?;
        let objects = plan
            .store
            .as_ref()
            .ok_or_else(|| RepackError::Incomplete(RetentionError::MissingPinnedPlanStore))?;
        let required = &plan.required;
        let protected = &plan.protected_packs;
        let protected_names: std::collections::BTreeSet<_> = protected
            .iter()
            .filter_map(|path| path.file_name().map(std::ffi::OsStr::to_owned))
            .collect();
        let pack_sources = objects.primary_packs_with_paths();
        let mut held = std::collections::BTreeSet::new();
        let count =
            u32::try_from(required.len()).map_err(|_| PackWriteError::Limit("object count"))?;
        if count > limits.write.max_objects {
            return Err(PackWriteError::Limit("object count").into());
        }
        let packs: Vec<_> = pack_sources.iter().map(|(_, pack)| *pack).collect();
        let mut packed = Vec::new();
        let mut decoded = Vec::new();
        let mut bytes = 0u64;
        'required: for &id in required {
            check(cancel)?;
            if !protected_names.is_empty() {
                for (path, source) in &pack_sources {
                    if !path
                        .file_name()
                        .is_some_and(|name| protected_names.contains(name))
                    {
                        continue;
                    }
                    let found = source
                        .find(id)
                        .map_err(|source| RepackError::Read { id, source })?;
                    if let Some(position) = found {
                        // A later readable protected copy can survive in place. A damaged entry
                        // cannot authorize omission; retain the ordinary first copy instead.
                        if source.read(position, policy.read, cancel).is_ok() {
                            held.insert(id);
                            continue 'required;
                        }
                        check(cancel)?;
                    }
                }
            }
            for (pack, source) in packs.iter().enumerate() {
                let found = source
                    .find(id)
                    .map_err(|source| RepackError::Read { id, source })?;
                if let Some(position) = found {
                    charge_input(plan.object_info[&id].size, &mut bytes, limits.write)?;
                    packed.push(Packed { id, pack, position });
                    continue 'required;
                }
            }
            let object = objects
                .read_controlled(id, policy.read, cancel)
                .map_err(|source| RepackError::Read { id, source })?
                .or_else(|| super::canonical_empty_tree(id))
                .ok_or(RepackError::Missing(id))?;
            charge_input(object.data().len() as u64, &mut bytes, limits.write)?;
            decoded.push((id, object));
        }
        let fresh: Vec<_> = decoded
            .iter()
            .map(|(id, object)| PackObject {
                id: *id,
                kind: object.kind(),
                data: object.data(),
            })
            .collect();
        let directory = self.object_dir().join("pack");
        fs::create_dir_all(&directory)?;
        let mut pack = NamedTempFile::new_in(&directory)?;
        let mut index = NamedTempFile::new_in(&directory)?;
        let write_result = write_reusing(
            self.object_format(),
            &packs,
            &packed,
            &fresh,
            (
                &mut BufWriter::new(pack.as_file_mut()),
                &mut BufWriter::new(index.as_file_mut()),
            ),
            limits.write,
            limits.compression,
            (policy.read, cancel),
            &mut || {
                if cancel.load(Ordering::Relaxed) {
                    Err(PackWriteError::Io(io::ErrorKind::Interrupted.into()))
                } else {
                    Ok(())
                }
            },
        );
        let written = match write_result {
            Err(_) if cancel.load(Ordering::Relaxed) => return Err(RepackError::Cancelled),
            Err(ReuseError::Read { id, source }) => return Err(RepackError::Read { id, source }),
            Err(ReuseError::Write(error)) => return Err(error.into()),
            Ok(written) => written,
        };
        // Fresh delta search must produce objects readable under the caller's limits.
        if written.deltas.max_depth > policy.read.max_delta_depth {
            return Err(PackWriteError::Limit("output delta depth").into());
        }
        pack.as_file().sync_all()?;
        index.as_file().sync_all()?;
        after_write();
        let basename = directory.join(format!("pack-{}", written.checksum));
        publish_pair_checked(
            pack,
            index,
            &basename,
            cancel,
            || {},
            || {
                plan.check_packs_unchanged(cancel)
                    .map_err(RepackError::from)
            },
        )?;
        Ok((RepackPublished { written }, held))
    }
}

fn charge_input(len: u64, bytes: &mut u64, limits: PackWriteLimits) -> Result<(), PackWriteError> {
    if len > limits.max_object_bytes {
        return Err(PackWriteError::Limit("object bytes"));
    }
    *bytes = bytes
        .checked_add(len)
        .ok_or(PackWriteError::Limit("input bytes"))?;
    if *bytes > limits.max_input_bytes {
        return Err(PackWriteError::Limit("input bytes"));
    }
    Ok(())
}

#[cfg(test)]
fn publish_pair(
    pack: NamedTempFile,
    index: NamedTempFile,
    basename: &Path,
    cancel: &AtomicBool,
    after_pack: impl FnOnce(),
) -> Result<(), RepackError> {
    publish_pair_checked(pack, index, basename, cancel, after_pack, || Ok(()))
}

fn publish_pair_checked(
    pack: NamedTempFile,
    index: NamedTempFile,
    basename: &Path,
    cancel: &AtomicBool,
    after_pack: impl FnOnce(),
    mut fresh: impl FnMut() -> Result<(), RepackError>,
) -> Result<(), RepackError> {
    check(cancel)?;
    fresh()?;
    publish(pack, &basename.with_extension("pack"), cancel)?;
    // Persist the pack name before publishing its index visibility marker.
    #[cfg(unix)]
    File::open(basename.parent().expect("pack basename has a directory"))?.sync_all()?;
    after_pack();
    check(cancel)?;
    fresh()?;
    publish(index, &basename.with_extension("idx"), cancel)?;
    Ok(())
}

fn check(cancel: &AtomicBool) -> Result<(), RepackError> {
    if cancel.load(Ordering::Relaxed) {
        Err(RepackError::Cancelled)
    } else {
        Ok(())
    }
}

fn publish(temporary: NamedTempFile, path: &Path, cancel: &AtomicBool) -> Result<(), RepackError> {
    match temporary.persist_noclobber(path) {
        Ok(_) => Ok(()),
        Err(error) if error.error.kind() == io::ErrorKind::AlreadyExists => {
            let mut existing = File::open(path)?;
            let mut candidate = error.file.reopen()?;
            if existing.metadata()?.len() != candidate.metadata()?.len() {
                return Err(RepackError::Existing(path.into()));
            }
            let mut a = [0; 8192];
            let mut b = [0; 8192];
            loop {
                check(cancel)?;
                let n = existing.read(&mut a)?;
                candidate.read_exact(&mut b[..n])?;
                if a[..n] != b[..n] {
                    return Err(RepackError::Existing(path.into()));
                }
                if n == 0 {
                    return Ok(());
                }
            }
        }
        Err(error) => Err(error.error.into()),
    }
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use super::*;

    #[test]
    fn output_depth_limit_refuses_publication_for_loose_delta_search() {
        let root = tempfile::tempdir().unwrap();
        let repository = Repository::init(
            crate::ObjectFormat::Sha1,
            root.path().join("repo"),
            crate::InitKind::Bare,
        )
        .unwrap();
        let data: Vec<u8> = (0..4096u32)
            .flat_map(|i| i.wrapping_mul(2_654_435_761).to_le_bytes())
            .collect();
        let a = repository.loose_objects().write_blob(&data).unwrap();
        let mut similar = data;
        similar[0] = b'b';
        let b = repository.loose_objects().write_blob(&similar).unwrap();
        let policy = RetentionPolicy {
            heads: vec![a, b],
            read: crate::ReadLimits {
                max_delta_depth: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        assert!(matches!(
            repository.repack_retained(&policy, RepackLimits::default(), &AtomicBool::new(false)),
            Err(RepackError::Write(PackWriteError::Limit(
                "output delta depth"
            )))
        ));
        assert_eq!(
            fs::read_dir(repository.object_dir().join("pack"))
                .unwrap()
                .count(),
            0
        );
        assert_eq!(
            repository
                .loose_objects()
                .read_blob(a, 16_384)
                .unwrap()
                .len(),
            16_384
        );
    }

    #[test]
    fn cancellation_between_publications_leaves_only_an_unindexed_pack() {
        let root = tempfile::tempdir().unwrap();
        let basename = root.path().join("pack-test");
        let mut pack = NamedTempFile::new_in(root.path()).unwrap();
        let mut index = NamedTempFile::new_in(root.path()).unwrap();
        pack.write_all(b"pack bytes").unwrap();
        index.write_all(b"index bytes").unwrap();
        let cancelled = AtomicBool::new(false);
        let result = publish_pair(pack, index, &basename, &cancelled, || {
            assert!(basename.with_extension("pack").exists());
            assert!(!basename.with_extension("idx").exists());
            cancelled.store(true, Ordering::Relaxed);
        });
        assert!(matches!(result, Err(RepackError::Cancelled)));
        assert_eq!(
            fs::read(basename.with_extension("pack")).unwrap(),
            b"pack bytes"
        );
        assert!(!basename.with_extension("idx").exists());
    }

    #[test]
    fn conflicting_existing_artifact_is_not_overwritten() {
        let root = tempfile::tempdir().unwrap();
        let path = root.path().join("pack-existing.pack");
        fs::write(&path, b"old bytes").unwrap();
        let mut candidate = NamedTempFile::new_in(root.path()).unwrap();
        candidate.write_all(b"new bytes").unwrap();
        let result = publish(candidate, &path, &AtomicBool::new(false));
        assert!(matches!(result, Err(RepackError::Existing(_))));
        assert_eq!(fs::read(path).unwrap(), b"old bytes");
    }

    #[test]
    fn new_loose_object_after_scan_survives_additive_publication() {
        let root = tempfile::tempdir().unwrap();
        let repository = Repository::init(
            crate::ObjectFormat::Sha1,
            root.path().join("repo"),
            crate::InitKind::Bare,
        )
        .unwrap();
        let mut new_id = None;
        repository
            .repack_retained_after_plan(
                &RetentionPolicy::default(),
                RepackLimits::default(),
                &AtomicBool::new(false),
                || {
                    new_id = Some(
                        repository
                            .loose_objects()
                            .write_blob(b"new writer")
                            .unwrap(),
                    )
                },
            )
            .unwrap();
        let id = new_id.unwrap();
        assert_eq!(
            repository.loose_objects().read_blob(id, 64).unwrap(),
            b"new writer"
        );
    }
    #[cfg(unix)]
    #[rstest::rstest]
    #[case::truncate(true)]
    #[case::overwrite(false)]
    fn warmed_pack_change_refuses_repack_publication(#[case] truncate: bool) {
        let (_root, repo, policy, pack) = super::super::walk_tests::header_fixture();
        let before = fs::read_dir(repo.object_dir().join("pack"))
            .unwrap()
            .count();
        let error = repo
            .repack_retained_after_plan(
                &policy,
                RepackLimits::trusted(),
                &AtomicBool::new(false),
                || super::super::walk_tests::change_pack(&pack, truncate),
            )
            .unwrap_err();
        assert!(
            matches!(error, RepackError::Open(ObjectReadError::PackChanged { path }) if path == pack)
        );
        assert_eq!(
            fs::read_dir(repo.object_dir().join("pack"))
                .unwrap()
                .count(),
            before
        );
    }
    #[cfg(unix)]
    #[rstest::rstest]
    #[case::truncate(true)]
    #[case::overwrite(false)]
    fn warmed_pack_change_after_write_refuses_publication(#[case] truncate: bool) {
        let (_root, repo, policy, pack) = super::super::walk_tests::header_fixture();
        let cancel = AtomicBool::new(false);
        let mut plan = repo.plan_retention(&policy, &cancel);
        assert!(plan.is_complete());
        plan.protected_packs.clear();
        let before = fs::read_dir(repo.object_dir().join("pack"))
            .unwrap()
            .count();
        let error = repo
            .publish_required_after_write(&plan, &policy, RepackLimits::trusted(), &cancel, || {
                super::super::walk_tests::change_pack(&pack, truncate)
            })
            .unwrap_err();
        assert!(
            matches!(error, RepackError::Open(ObjectReadError::PackChanged { path }) if path == pack)
        );
        assert_eq!(
            fs::read_dir(repo.object_dir().join("pack"))
                .unwrap()
                .count(),
            before
        );
    }
}