girt 0.1.4

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

use tempfile::NamedTempFile;

use super::RetentionPolicy;
use crate::pack::{PackObject, PackWriteError, PackWriteLimits, PackWritten, write_controlled};
use crate::{ObjectId, ObjectReadError, PackCompression, Repository};

/// Bounds one additive repack. Payloads remain in memory until the pack is written.
#[derive(Debug, Clone, Copy)]
pub struct RepackLimits {
    /// Bounds object count, decoded input bytes, and both output artifacts.
    pub write: PackWriteLimits,
    /// Selects ordinary or bounded delta compression.
    pub compression: PackCompression,
}

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

/// A complete indexed pack is visible; existing storage has not been removed.
#[derive(Debug, Clone, Copy)]
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.
#[derive(Debug, thiserror::Error)]
pub enum RepackError {
    /// A fresh retention scan did not establish complete reachability.
    #[error("retention scan incomplete: {0}")]
    Incomplete(String),
    /// 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}: {source}")]
    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`]. It reads required objects through pinned packs, 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.
    /// File contents are synchronized; directory entries are not, so power-loss durability is not
    /// promised. Publication assumes a trusted local filesystem with no in-place artifact edits.
    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> {
        check(cancel)?;
        let plan = self.plan_retention(policy, cancel);
        if let super::RetentionOutcome::Incomplete(reason) = plan.outcome {
            return Err(RepackError::Incomplete(reason));
        }
        after_plan();
        check(cancel)?;
        let objects =
            self.objects_controlled(policy.packs, crate::AlternateLimits::default(), cancel)?;
        let count = u32::try_from(plan.required.len())
            .map_err(|_| PackWriteError::Limit("object count"))?;
        if count > limits.write.max_objects {
            return Err(PackWriteError::Limit("object count").into());
        }
        let mut decoded = Vec::with_capacity(plan.required.len());
        let mut bytes = 0u64;
        for id in plan.required {
            check(cancel)?;
            let object = objects
                .read_controlled(id, policy.read, cancel)
                .map_err(|source| RepackError::Read { id, source })?
                .ok_or(RepackError::Missing(id))?;
            let len = object.data().len() as u64;
            if len > limits.write.max_object_bytes {
                return Err(PackWriteError::Limit("object bytes").into());
            }
            bytes = bytes
                .checked_add(len)
                .ok_or(PackWriteError::Limit("input bytes"))?;
            if bytes > limits.write.max_input_bytes {
                return Err(PackWriteError::Limit("input bytes").into());
            }
            decoded.push((id, object));
        }
        let input: 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_controlled(
            self.object_format(),
            &input,
            &mut pack,
            &mut index,
            limits.write,
            limits.compression,
            &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),
            result => result?,
        };
        pack.as_file().sync_all()?;
        index.as_file().sync_all()?;
        let basename = directory.join(format!("pack-{}", written.checksum));
        publish_pair(pack, index, &basename, cancel, || {})?;
        Ok(RepackPublished { written })
    }
}

fn publish_pair(
    pack: NamedTempFile,
    index: NamedTempFile,
    basename: &Path,
    cancel: &AtomicBool,
    after_pack: impl FnOnce(),
) -> Result<(), RepackError> {
    check(cancel)?;
    publish(pack, &basename.with_extension("pack"), cancel)?;
    after_pack();
    check(cancel)?;
    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 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"
        );
    }
}