girt 0.4.2

use std::borrow::Cow;
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::sync::atomic::AtomicBool;

use super::{FetchError as Error, FetchLimits, check_cancelled};
use crate::{Object, ObjectId, ObjectKind, Objects};

/// Local object availability and complete-history evidence, including shallow boundaries.
///
/// [`Self::new`] verifies the reachable graph from explicit roots, retaining its payloads.
/// [`Self::from_store`] reads objects on demand and trusts only commit targets of explicit
/// destination reference roots and commits discovered from them as complete. A merely present
/// object can be an external delta base or dependency without proving that its descendants exist.
///
/// Commit discovery is breadth-first in caller root and payload edge order, capped by
/// [`FetchLimits::max_haves`]. A small cap can reduce negotiation effectiveness, never correctness.
/// No local references are inferred by these constructors. Callers must coordinate with GC until
/// installation and any subsequent reference publication. The reader's implicit empty tree is
/// not physical availability: fetch must materialize it before relying on it for publication.
#[derive(Debug, Default)]
pub struct KnownHistory {
    pub(super) objects: HashMap<ObjectId, Object>,
    /// Destination objects available as delta bases and dependencies, consulted after `objects`.
    pub(super) store: Option<Arc<Objects>>,
    /// Commit targets of explicit destination roots and commits discovered from them.
    pub(super) complete_roots: HashSet<ObjectId>,
    /// A synthetic reader value is not a physical object that can support publication.
    pub(super) missing_empty_tree: Option<ObjectId>,
    pub(super) haves: Vec<ObjectId>,
    pub(super) shallow: Vec<ObjectId>,
    pub(super) only_when_all_wants_known: bool,
}

/// Objects the destination already has, for delta bases and connectivity.
pub(crate) trait KnownObjects {
    /// Returns a known object.
    fn get(&self, id: ObjectId) -> Option<Cow<'_, Object>>;

    /// Whether a known object's history is trusted to be complete, so connectivity checks can
    /// stop at it rather than walking its history.
    fn trusted_complete(&self, _id: ObjectId) -> bool {
        false
    }
}

impl KnownObjects for HashMap<ObjectId, Object> {
    fn get(&self, id: ObjectId) -> Option<Cow<'_, Object>> {
        HashMap::get(self, &id).map(Cow::Borrowed)
    }
}

impl KnownObjects for KnownHistory {
    fn get(&self, id: ObjectId) -> Option<Cow<'_, Object>> {
        if self.missing_empty_tree == Some(id) {
            return None;
        }
        if let Some(object) = self.objects.get(&id) {
            return Some(Cow::Borrowed(object));
        }
        self.store
            .as_ref()?
            .read(id, crate::ReadLimits::trusted())
            .ok()
            .flatten()
            .map(Cow::Owned)
    }

    fn trusted_complete(&self, id: ObjectId) -> bool {
        self.complete_roots.contains(&id) && !self.shallow.contains(&id)
    }
}

impl KnownHistory {
    pub(super) fn kind(&self, id: ObjectId) -> Option<ObjectKind> {
        self.objects.get(&id).map(Object::kind).or_else(|| {
            self.store
                .as_ref()?
                .transfer_object(id, crate::ReadLimits::trusted(), &AtomicBool::new(false))
                .ok()
                .flatten()
                .map(|(kind, _, _)| kind)
        })
    }

    pub(super) fn complete_kind(&self, id: ObjectId) -> Option<ObjectKind> {
        self.objects
            .get(&id)
            .filter(|_| self.missing_empty_tree.is_none())
            .map(Object::kind)
            .or_else(|| self.trusted_complete(id).then(|| self.kind(id)).flatten())
    }

    /// Peels reference roots to complete commits, without walking a wire have batch.
    pub(crate) fn from_store_roots(
        store: Objects,
        roots: &[ObjectId],
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        let missing_empty_tree = missing_empty_tree(&store, cancel)?;
        let mut complete_roots = HashSet::new();
        for &root in roots {
            check_cancelled(cancel)?;
            if let Ok(peeled) = store.peel(root, crate::PeelLimits::default(), cancel)
                && peeled.kind == ObjectKind::Commit
            {
                complete_roots.insert(peeled.target);
            }
        }
        check_cancelled(cancel)?;
        Ok(Self {
            shallow: store.shallow_roots().iter().collect(),
            store: Some(Arc::new(store)),
            complete_roots,
            missing_empty_tree,
            ..Self::default()
        })
    }

    /// Describes a destination store for negotiation without reading its complete history.
    ///
    /// Up to [`FetchLimits::max_haves`] commits reachable from `tips` (breadth-first, peeling
    /// tags) are offered as haves. The store's shallow roots are declared. Callers supply roots
    /// established by destination references; their peeled commit targets and commits discovered
    /// from them are trusted complete up to declared shallow boundaries. Other stored objects can
    /// supply delta bases and dependencies, but their presence alone does not stop connectivity
    /// validation. Callers must keep the roots and their histories available (e.g. exclude
    /// pruning) until the fetch publishes its references.
    ///
    /// # Errors
    ///
    /// Fails on cancellation. Unreadable tips are ignored; unreadable ancestors end their walk.
    pub fn from_store(
        store: Objects,
        tips: &[ObjectId],
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        let missing_empty_tree = missing_empty_tree(&store, cancel)?;
        let shallow: Vec<_> = store.shallow_roots().iter().collect();
        let shallow_set: HashSet<_> = shallow.iter().copied().collect();
        let mut haves = Vec::new();
        let mut complete_roots = HashSet::new();
        let mut seen = HashSet::new();
        let mut pending = VecDeque::new();
        for &tip in tips {
            check_cancelled(cancel)?;
            let Ok(peeled) = store.peel(tip, crate::PeelLimits::default(), cancel) else {
                continue;
            };
            if peeled.kind == ObjectKind::Commit {
                complete_roots.insert(peeled.target);
                if seen.insert(peeled.target) {
                    pending.push_back(peeled.target);
                }
            }
        }
        while let Some(id) = pending.pop_front() {
            if haves.len() >= limits.max_haves {
                break;
            }
            check_cancelled(cancel)?;
            let Ok(Some(object)) = store.read(id, limits.known_read) else {
                continue;
            };
            if object.kind() != ObjectKind::Commit {
                continue;
            }
            haves.push(id);
            complete_roots.insert(id);
            if shallow_set.contains(&id) {
                continue;
            }
            let Ok(commit) = crate::Commit::parse(store.object_format(), object.data()) else {
                continue;
            };
            for &parent in commit.parents() {
                if seen.insert(parent) {
                    pending.push_back(parent);
                }
            }
        }
        check_cancelled(cancel)?;
        Ok(Self {
            objects: HashMap::new(),
            store: Some(Arc::new(store)),
            complete_roots,
            missing_empty_tree,
            haves,
            shallow,
            only_when_all_wants_known: false,
        })
    }

    /// Uses this history only when it contains every selected advertised tip.
    ///
    /// A selection with any unknown tip instead requests a complete transfer. This avoids
    /// producing a pack that depends on local objects when the caller can retain only the new
    /// pack against concurrent collection. The caller must still keep verified local objects
    /// available until a known-only result publishes its references.
    /// Explicit depth requests retain the snapshot even for unknown tips: validation includes
    /// reachable local dependencies in the resulting pack before retained installation.
    pub fn only_when_all_wants_known(mut self) -> Self {
        self.only_when_all_wants_known = true;
        self
    }

    /// Lets validation read the store's synthetic empty tree, returning its ID when unstored.
    ///
    /// The caller must store the tree before publishing anything that depends on it.
    pub(crate) fn assume_empty_tree(&mut self) -> Option<ObjectId> {
        self.missing_empty_tree.take()
    }

    pub(super) fn applies_to(&self, wants: &[ObjectId]) -> bool {
        !self.only_when_all_wants_known || wants.iter().all(|&id| self.complete_kind(id).is_some())
    }

    /// Keep the destination's shallow declarations without offering local objects as haves.
    /// An unknown selected tip still needs a self-contained retained pack, but the server must
    /// see existing boundaries so it can report any roots it removes during deepening.
    pub(super) fn shallow_only(&self) -> Self {
        Self {
            shallow: self.shallow.clone(),
            store: self.store.clone(),
            missing_empty_tree: self.missing_empty_tree,
            ..Self::default()
        }
    }
    /// Reads a bounded, identity-verified local graph without changing storage.
    ///
    /// Local object count, retained bytes, edges and per-read decoding use the `max_known_*` and
    /// `known_read` bounds. Root occurrences use `max_wants`. These budgets are separate from
    /// transfer/import budgets. Cancellation is checked between reads and edges; an individual
    /// filesystem read, decode, hash or parse cannot be interrupted.
    ///
    /// # Errors
    ///
    /// Fails on missing/corrupt objects, malformed payloads, mistyped edges, cancellation or
    /// bounds. Declared shallow commits stop parent traversal and are sent to the peer. To request
    /// a full transfer into a complete destination, explicitly use [`KnownHistory::default`].
    pub fn new(
        store: &Objects,
        roots: &[ObjectId],
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        Self::new_local(store, roots, limits, cancel)
    }

    /// Validates local history in either object format for native transfer decisions.
    ///
    /// Uses [`Self::new`]'s graph, resource and cancellation contract. The caller must keep
    /// verified roots available through publication.
    pub fn new_local(
        store: &Objects,
        roots: &[ObjectId],
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        let shallow: Vec<_> = store.shallow_roots().iter().collect();
        Self::new_with_boundaries(store, roots, &shallow, limits, cancel)
    }

    pub(super) fn new_with_boundaries(
        store: &Objects,
        roots: &[ObjectId],
        shallow: &[ObjectId],
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        Self::new_stopping_at(store, roots, shallow, &HashSet::new(), limits, cancel)
    }

    /// Validates history reachable from `roots`, treating `published` objects as verified.
    ///
    /// Objects named by existing references were verified when they were published, so the
    /// walk neither reads nor descends into them, like Git's connectivity check excluding
    /// everything reachable from existing references (`--not --all`). The result's objects
    /// therefore cover only history new to the repository.
    pub(super) fn new_stopping_at(
        store: &Objects,
        roots: &[ObjectId],
        shallow: &[ObjectId],
        published: &HashSet<ObjectId>,
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<Self, Error> {
        let mut result = Self {
            shallow: shallow.to_vec(),
            ..Self::default()
        };
        walk(
            WalkSource::Store(store),
            roots,
            shallow,
            published,
            limits,
            cancel,
            |id, object| {
                if object.kind() == ObjectKind::Commit && result.haves.len() < limits.max_haves {
                    result.haves.push(id);
                }
                result.objects.insert(id, object);
            },
        )?;
        let empty = store.object_format().hash_object(ObjectKind::Tree, b"");
        if result.objects.contains_key(&empty) {
            result.missing_empty_tree = missing_empty_tree(store, cancel)?;
            if result.missing_empty_tree.is_some() {
                result.haves.clear();
            }
        }
        result.shallow.retain(|id| result.objects.contains_key(id));
        Ok(result)
    }

    /// Checks that history reachable from `roots` is complete, without retaining payloads.
    ///
    /// Applies [`Self::new_stopping_at`]'s graph, resource and cancellation contract, and returns
    /// the verified object IDs. Byte-bound receive evidence can supply structure without another
    /// decode; other representations use ordinary checked reads. Memory is proportional to the
    /// number of objects, not their size.
    pub(super) fn verify_stopping_at(
        source: &super::verified::InstalledObjects<'_>,
        roots: &[ObjectId],
        shallow: &[ObjectId],
        published: &HashSet<ObjectId>,
        limits: FetchLimits,
        cancel: &AtomicBool,
    ) -> Result<HashSet<ObjectId>, Error> {
        let kinds = walk(
            WalkSource::Installed(source),
            roots,
            shallow,
            published,
            limits,
            cancel,
            |_, _| {},
        )?;
        Ok(kinds.into_keys().collect())
    }

    /// Number of distinct verified local objects retained for connectivity checks.
    pub fn object_count(&self) -> usize {
        self.objects.len()
    }

    /// Verified commits available for the single have batch, in discovery order.
    pub fn haves(&self) -> &[ObjectId] {
        &self.haves
    }
}

fn missing_empty_tree(store: &Objects, cancel: &AtomicBool) -> Result<Option<ObjectId>, Error> {
    let id = store.object_format().hash_object(ObjectKind::Tree, b"");
    check_cancelled(cancel)?;
    let present = store.has_stored_empty_tree(cancel);
    check_cancelled(cancel)?;
    let present = present.map_err(|source| Error::LocalRead { id, source })?;
    Ok((!present).then_some(id))
}

/// Installed evidence carries its own store, so the walk cannot pair it with another snapshot.
enum WalkSource<'a> {
    Store(&'a Objects),
    Installed(&'a super::verified::InstalledObjects<'a>),
}

impl WalkSource<'_> {
    fn store(&self) -> &Objects {
        match self {
            Self::Store(store) => store,
            Self::Installed(installed) => installed.store(),
        }
    }
}

/// Walks history reachable from `roots`, handing each verified object to `visit`.
///
/// Stops at `published` objects without reading them, and at the parents of `shallow` commits.
/// Returns the kind of every visited object.
fn walk(
    source: WalkSource<'_>,
    roots: &[ObjectId],
    shallow: &[ObjectId],
    published: &HashSet<ObjectId>,
    limits: FetchLimits,
    cancel: &AtomicBool,
    mut visit: impl FnMut(ObjectId, Object),
) -> Result<HashMap<ObjectId, ObjectKind>, Error> {
    let store = source.store();
    check_cancelled(cancel)?;
    if roots.len() > limits.max_wants {
        return Err(Error::Limit("known roots"));
    }
    if shallow.len() > limits.max_shallow_roots {
        return Err(Error::Limit("shallow roots"));
    }
    let shallow_set: HashSet<_> = shallow.iter().copied().collect();
    let mut kinds = HashMap::new();
    let mut pending = VecDeque::new();
    let mut expected = HashMap::new();
    for &id in roots.iter().filter(|id| !published.contains(id)) {
        enqueue(id, None, &mut expected, &mut pending, limits)?;
    }
    let mut bytes = limits.max_known_bytes;
    let mut edges = limits.max_known_edges;
    while let Some(id) = pending.pop_front() {
        check_cancelled(cancel)?;
        let mut read = limits.known_read;
        read.max_object_bytes = read.max_object_bytes.min(bytes);
        let proof = match &source {
            WalkSource::Store(_) => None,
            WalkSource::Installed(installed) => installed
                .get(id, read, cancel)
                .map_err(|source| Error::LocalRead { id, source })?,
        };
        let object = if proof.is_none() {
            Some(
                store
                    .read(id, read)
                    .map_err(|source| Error::LocalRead { id, source })?
                    .ok_or(Error::Missing(id))?,
            )
        } else {
            None
        };
        let (kind, length) = proof.map(|o| (o.kind, o.bytes)).unwrap_or_else(|| {
            let object = object.as_ref().unwrap();
            (object.kind(), object.data().len())
        });
        if shallow_set.contains(&id) && kind != ObjectKind::Commit {
            return Err(Error::Kind(id));
        }
        if expected[&id].is_some_and(|expected| expected != kind) {
            return Err(Error::Kind(id));
        }
        bytes = bytes
            .checked_sub(length)
            .ok_or(Error::Limit("known bytes"))?;
        let shallow_commit = kind == ObjectKind::Commit && shallow_set.contains(&id);
        let mut edge = |target, kind| {
            if (shallow_commit && kind == ObjectKind::Commit) || published.contains(&target) {
                return Ok(());
            }
            check_cancelled(cancel)?;
            edges = edges.checked_sub(1).ok_or(Error::Limit("known edges"))?;
            if kinds.get(&target).is_some_and(|&known| known != kind) {
                return Err(Error::Kind(target));
            }
            enqueue(target, Some(kind), &mut expected, &mut pending, limits)
        };
        if let Some(proof) = proof {
            let edges = proof
                .edges
                .as_ref()
                .expect("InstalledObjects::get requires parsed edges");
            for &(target, expected) in edges {
                edge(target, expected)?;
            }
        } else {
            let object = object.unwrap();
            crate::edges::visit(id, kind, object.data(), edge)?;
            visit(id, object);
        }
        kinds.insert(id, kind);
    }
    Ok(kinds)
}

fn enqueue(
    id: ObjectId,
    kind: Option<ObjectKind>,
    expected: &mut HashMap<ObjectId, Option<ObjectKind>>,
    pending: &mut VecDeque<ObjectId>,
    limits: FetchLimits,
) -> Result<(), Error> {
    if let Some(previous) = expected.get_mut(&id) {
        if previous.is_some() && kind.is_some() && *previous != kind {
            return Err(Error::Kind(id));
        }
        if kind.is_some() {
            *previous = kind;
        }
    } else {
        if expected.len() >= limits.max_known_objects {
            return Err(Error::Limit("known objects"));
        }
        expected.insert(id, kind);
        pending.push_back(id);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use rstest::rstest;

    use super::*;
    use crate::{
        Commit, CommitFields, EntryMode, LooseObjects, ObjectFormat, PackLimits, Signature, Tree,
        TreeEntry,
    };

    fn graph() -> (tempfile::TempDir, Objects, ObjectId, ObjectId) {
        let root = tempfile::tempdir().unwrap();
        let loose = LooseObjects::new(root.path(), ObjectFormat::Sha1);
        let blob = loose.write_blob(b"known blob").unwrap();
        let tree = loose
            .write_tree(
                &Tree::new(
                    crate::ObjectFormat::Sha1,
                    vec![
                        TreeEntry {
                            mode: EntryMode::Blob,
                            name: b"blob".to_vec(),
                            id: blob,
                        },
                        TreeEntry {
                            mode: EntryMode::Gitlink,
                            name: b"external".to_vec(),
                            id: ObjectId::for_blob(crate::ObjectFormat::Sha1, b"external"),
                        },
                    ],
                )
                .unwrap(),
            )
            .unwrap();
        let author = Signature {
            name: b"A".to_vec(),
            email: b"a@example.com".to_vec(),
            seconds: 0,
            offset_minutes: 0,
        };
        let commit = loose
            .write_commit(
                &Commit::new(CommitFields {
                    tree,
                    parents: vec![],
                    author: author.clone(),
                    committer: author,
                    extra_headers: vec![],
                    message: vec![],
                })
                .unwrap(),
            )
            .unwrap();
        let objects = Objects::open(
            crate::ObjectFormat::Sha1,
            root.path(),
            PackLimits::default(),
            crate::AlternateLimits::default(),
        )
        .unwrap();
        (root, objects, commit, blob)
    }

    #[rstest]
    #[case::objects(FetchLimits { max_known_objects: 2, ..FetchLimits::default() })]
    #[case::bytes(FetchLimits { max_known_bytes: 1, ..FetchLimits::default() })]
    #[case::edges(FetchLimits { max_known_edges: 1, ..FetchLimits::default() })]
    #[case::roots(FetchLimits { max_wants: 0, ..FetchLimits::default() })]
    fn rejects_local_work_bounds(#[case] limits: FetchLimits) {
        let (_root, objects, commit, _) = graph();
        assert!(KnownHistory::new(&objects, &[commit], limits, &AtomicBool::new(false)).is_err());
    }

    #[test]
    fn exact_count_and_edges_skip_gitlinks_and_bound_haves() {
        let (_root, objects, commit, _) = graph();
        let limits = FetchLimits {
            max_known_objects: 3,
            max_known_edges: 2,
            max_haves: 0,
            ..FetchLimits::default()
        };
        let known = KnownHistory::new(&objects, &[commit, commit], limits, &AtomicBool::new(false))
            .unwrap();
        assert_eq!(known.object_count(), 3);
        assert!(known.haves().is_empty());
    }

    fn damage(root: &std::path::Path, blob: ObjectId, corrupt: bool) {
        let id = blob.to_string();
        let path = root.join(&id[..2]).join(&id[2..]);
        if corrupt {
            std::fs::write(path, b"corrupt").unwrap();
        } else {
            std::fs::remove_file(path).unwrap();
        }
    }

    #[rstest]
    #[case::missing(false)]
    #[case::corrupt(true)]
    fn incomplete_history_cannot_be_offered(#[case] corrupt: bool) {
        let (root, objects, commit, blob) = graph();
        damage(root.path(), blob, corrupt);
        assert!(
            KnownHistory::new(
                &objects,
                &[commit],
                FetchLimits::default(),
                &AtomicBool::new(false)
            )
            .is_err()
        );
    }

    #[test]
    fn corrupt_local_dependency_reports_identity_and_storage_cause() {
        let (root, objects, commit, blob) = graph();
        damage(root.path(), blob, true);
        let error = KnownHistory::new(
            &objects,
            &[commit],
            FetchLimits::default(),
            &AtomicBool::new(false),
        )
        .unwrap_err();
        assert!(
            matches!(error, Error::LocalRead { id, source: crate::ObjectReadError::Loose(_) } if id == blob)
        );
    }

    #[test]
    fn cancelled_preparation_retains_no_partial_knowledge() {
        let (_root, objects, commit, _) = graph();
        assert!(matches!(
            KnownHistory::new(
                &objects,
                &[commit],
                FetchLimits::default(),
                &AtomicBool::new(true)
            ),
            Err(Error::Cancelled)
        ));
    }

    #[test]
    fn store_history_offers_commits_and_reads_objects_on_demand() {
        let (_root, objects, commit, blob) = graph();
        let known = KnownHistory::from_store(
            objects,
            &[commit],
            FetchLimits::default(),
            &AtomicBool::new(false),
        )
        .unwrap();
        assert_eq!(known.haves(), &[commit]);
        assert_eq!(known.object_count(), 0);
        assert!(known.trusted_complete(commit));
        assert!(!known.trusted_complete(blob));
        assert_eq!(known.get(blob).unwrap().kind(), ObjectKind::Blob);
    }

    #[test]
    fn unanchored_store_object_does_not_stop_connectivity() {
        let (root, objects, commit, blob) = graph();
        damage(root.path(), blob, false);
        let known = KnownHistory::from_store(
            objects,
            &[],
            FetchLimits::default(),
            &AtomicBool::new(false),
        )
        .unwrap();
        let result = super::super::connectivity::validate_with_boundaries(
            &HashMap::new(),
            &known,
            &[commit],
            &[],
            FetchLimits::default(),
            &AtomicBool::new(false),
        );
        assert!(matches!(result, Err(Error::Missing(missing)) if missing == blob));
    }

    #[test]
    fn connectivity_stops_at_trusted_store_objects() {
        let (_root, objects, commit, _blob) = graph();
        let known = KnownHistory::from_store(
            objects,
            &[commit],
            FetchLimits::default(),
            &AtomicBool::new(false),
        )
        .unwrap();
        let dependencies = super::super::connectivity::validate_with_boundaries(
            &HashMap::new(),
            &known,
            &[commit],
            &[],
            FetchLimits::default(),
            &AtomicBool::new(false),
        )
        .unwrap();
        assert_eq!(dependencies, [commit]);
    }
}