use std::collections::{HashMap, HashSet, VecDeque};

use bytes::Bytes;
use cid::Cid as IpldCid;
use miette::{Result, WrapErr};
use smol_str::SmolStr;

use crate::car::{Blocks, CommitBlocks};
use crate::mst::{FlatEntry, MstKey, decode_node, decode_node_into, layer_for_key};

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct KeyRange {
    start: SmolStr,
    end: Option<SmolStr>,
}

impl KeyRange {
    #[inline]
    pub(crate) fn contains(&self, key: impl AsRef<[u8]>) -> bool {
        let key = key.as_ref();
        key >= self.start.as_bytes()
            && self
                .end
                .as_ref()
                .map(|end| key < end.as_bytes())
                .unwrap_or(true)
    }

    #[inline]
    fn intersects_subtree(
        &self,
        lower_exclusive: Option<&[u8]>,
        upper_exclusive: Option<&[u8]>,
    ) -> bool {
        if let Some(upper) = upper_exclusive {
            if self.start.as_bytes() >= upper {
                return false;
            }
        }

        if let (Some(end), Some(lower)) = (self.end.as_ref(), lower_exclusive) {
            if end.as_bytes() <= lower {
                return false;
            }
        }

        true
    }
}

pub(crate) fn sparse_ranges(patterns: &[SmolStr]) -> Vec<KeyRange> {
    patterns
        .iter()
        .filter_map(|pattern| {
            let prefix = pattern
                .strip_suffix(".*")
                .map(|prefix| SmolStr::new(format!("{prefix}.")))
                .unwrap_or_else(|| SmolStr::new(format!("{pattern}/")));
            Some(KeyRange {
                end: prefix_upper_bound(prefix.as_str()),
                start: prefix,
            })
        })
        .collect()
}

pub(crate) fn sparse_probe_collection(patterns: &[SmolStr]) -> Option<SmolStr> {
    patterns.iter().find_map(|pattern| {
        pattern
            .strip_suffix(".*")
            .map(|prefix| format!("{prefix}.probe"))
            .unwrap_or_else(|| pattern.to_string())
            .parse::<jacquard_common::types::string::Nsid>()
            .ok()
            .map(|_| {
                pattern
                    .strip_suffix(".*")
                    .map(|prefix| SmolStr::new(format!("{prefix}.probe")))
                    .unwrap_or_else(|| pattern.clone())
            })
    })
}

fn prefix_upper_bound(prefix: &str) -> Option<SmolStr> {
    let mut bytes = prefix.as_bytes().to_vec();
    for i in (0..bytes.len()).rev() {
        if bytes[i] < u8::MAX {
            bytes[i] += 1;
            bytes.truncate(i + 1);
            return String::from_utf8(bytes).ok().map(SmolStr::new);
        }
    }
    None
}

pub(crate) fn mst_node_layer(bytes: &[u8]) -> Result<Option<usize>> {
    let entries = decode_node(bytes).wrap_err("failed to decode mst node for layer estimate")?;
    Ok(entries.into_iter().find_map(|entry| match entry {
        FlatEntry::Leaf { key, .. } => Some(layer_for_key(key.as_str())),
        FlatEntry::Tree { .. } => None,
    }))
}

/// leaves from a completed [`SparseScanner::scan`], guaranteed sorted and deduplicated by
/// strictly increasing mst key
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct SortedLeaves(Vec<(MstKey, IpldCid)>);

impl SortedLeaves {
    fn from_raw(mut leaves: Vec<(MstKey, IpldCid)>) -> Self {
        leaves.sort_unstable_by(|a, b| a.0.cmp(&b.0));
        leaves.dedup_by(|a, b| a.0 == b.0);
        Self(leaves)
    }

    #[cfg(test)]
    pub(crate) fn from_vec(leaves: Vec<(MstKey, IpldCid)>) -> Self {
        Self::from_raw(leaves)
    }
}

impl std::ops::Deref for SortedLeaves {
    type Target = [(MstKey, IpldCid)];

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<'a> IntoIterator for &'a SortedLeaves {
    type Item = &'a (MstKey, IpldCid);
    type IntoIter = std::slice::Iter<'a, (MstKey, IpldCid)>;

    fn into_iter(self) -> Self::IntoIter {
        self.0.iter()
    }
}

impl PartialEq<Vec<(MstKey, IpldCid)>> for SortedLeaves {
    fn eq(&self, other: &Vec<(MstKey, IpldCid)>) -> bool {
        &self.0 == other
    }
}

impl PartialEq<[(MstKey, IpldCid)]> for SortedLeaves {
    fn eq(&self, other: &[(MstKey, IpldCid)]) -> bool {
        self.0.as_slice() == other
    }
}

impl PartialEq<Vec<(SmolStr, IpldCid)>> for SortedLeaves {
    fn eq(&self, other: &Vec<(SmolStr, IpldCid)>) -> bool {
        self.0.len() == other.len()
            && self
                .0
                .iter()
                .zip(other.iter())
                .all(|((k1, c1), (k2, c2))| k1 == k2 && c1 == c2)
    }
}

#[derive(Debug, Clone)]
pub(crate) struct SparseScanOutput {
    pub(crate) leaves: SortedLeaves,
    pub(crate) node_blocks_seen: usize,
    pub(crate) node_bytes_seen: usize,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct NodeBounds {
    lower: Option<MstKey>,
    upper: Option<MstKey>,
}

/// almost every mst node is reached through a single parent with a single bound; keeping the
/// first bound inline avoids a heap allocation per node in `pending_missing` and `visited`
#[derive(Debug, Clone)]
struct BoundsSet {
    first: NodeBounds,
    rest: Vec<NodeBounds>,
}

impl BoundsSet {
    fn new(first: NodeBounds) -> Self {
        Self {
            first,
            rest: Vec::new(),
        }
    }

    fn covers(&self, bounds: &NodeBounds) -> bool {
        self.first.covers(bounds) || self.rest.iter().any(|prior| prior.covers(bounds))
    }

    fn insert(&mut self, bounds: NodeBounds) {
        if !self.covers(&bounds) {
            self.rest.push(bounds);
        }
    }

    fn iter(&self) -> std::iter::Chain<std::iter::Once<&NodeBounds>, std::slice::Iter<'_, NodeBounds>> {
        std::iter::once(&self.first).chain(self.rest.iter())
    }

    fn into_iter(self) -> std::iter::Chain<std::iter::Once<NodeBounds>, std::vec::IntoIter<NodeBounds>> {
        std::iter::once(self.first).chain(self.rest)
    }
}

type CidMap<V> = HashMap<IpldCid, V, ahash::RandomState>;
type CidSet = HashSet<IpldCid, ahash::RandomState>;

impl NodeBounds {
    fn unbounded() -> Self {
        Self {
            lower: None,
            upper: None,
        }
    }

    fn covers(&self, other: &Self) -> bool {
        let lower_ok = match (&self.lower, &other.lower) {
            (None, _) => true,
            (Some(_), None) => false,
            (Some(a), Some(b)) => a <= b,
        };
        let upper_ok = match (&self.upper, &other.upper) {
            (None, _) => true,
            (Some(_), None) => false,
            (Some(a), Some(b)) => a >= b,
        };
        lower_ok && upper_ok
    }

    #[inline]
    fn contains_leaf(&self, key: impl AsRef<[u8]>) -> bool {
        let key = key.as_ref();
        self.lower
            .as_ref()
            .is_none_or(|lower| key > lower.as_bytes())
            && self
                .upper
                .as_ref()
                .is_none_or(|upper| key < upper.as_bytes())
    }

    #[inline]
    fn subtree_bounds<'a>(
        &'a self,
        lower_leaf: Option<&'a MstKey>,
        upper_leaf: Option<&'a MstKey>,
        ranges: &[KeyRange],
    ) -> Option<(Option<&'a MstKey>, Option<&'a MstKey>)> {
        let lower = match (self.lower.as_ref(), lower_leaf) {
            (Some(a), Some(b)) => Some(a.max(b)),
            (a, b) => a.or(b),
        };
        let upper = match (self.upper.as_ref(), upper_leaf) {
            (Some(a), Some(b)) => Some(a.min(b)),
            (a, b) => a.or(b),
        };
        if lower.zip(upper).is_some_and(|(l, u)| l >= u) {
            return None;
        }
        if !ranges.iter().any(|range| {
            range.intersects_subtree(lower.map(MstKey::as_bytes), upper.map(MstKey::as_bytes))
        }) {
            return None;
        }
        Some((lower, upper))
    }

    /// narrows `self` to a child subtree's `(lower_leaf, upper_leaf)` interval, returning
    /// `None` if the resulting interval is inverted/empty or cannot intersect `ranges`
    fn narrow(
        &self,
        lower_leaf: Option<&MstKey>,
        upper_leaf: Option<&MstKey>,
        ranges: &[KeyRange],
    ) -> Option<Self> {
        let (lower, upper) = self.subtree_bounds(lower_leaf, upper_leaf, ranges)?;
        Some(Self {
            lower: lower.cloned(),
            upper: upper.cloned(),
        })
    }
}

/// whether a node seen on its own, its place in the tree unknown, could hold a key in
/// `ranges`: a leaf in one, or a subtree that could reach one whatever the node's outer bounds
/// turn out to be. a scan of `ranges` takes nothing from a node this rules out, from any
/// ancestor, because real bounds only narrow the intervals checked here
pub(crate) fn node_may_reach(ranges: &[KeyRange], entries: &[FlatEntry]) -> bool {
    node_may_reach_within(&NodeBounds::unbounded(), ranges, entries)
}

fn entries_within_bounds(bounds: &NodeBounds, entries: &[FlatEntry]) -> bool {
    let first = entries.iter().find_map(|entry| match entry {
        FlatEntry::Leaf { key, .. } => Some(key),
        FlatEntry::Tree { .. } => None,
    });
    let last = entries.iter().rev().find_map(|entry| match entry {
        FlatEntry::Leaf { key, .. } => Some(key),
        FlatEntry::Tree { .. } => None,
    });
    first.is_none_or(|key| bounds.contains_leaf(key))
        && last.is_none_or(|key| bounds.contains_leaf(key))
}

fn node_may_reach_within(
    bounds: &NodeBounds,
    ranges: &[KeyRange],
    entries: &[FlatEntry],
) -> bool {
    entries.iter().enumerate().any(|(idx, entry)| match entry {
        FlatEntry::Leaf { key, .. } => {
            bounds.contains_leaf(key) && ranges.iter().any(|range| range.contains(key))
        }
        FlatEntry::Tree { .. } => bounds
            .subtree_bounds(previous_leaf(entries, idx), next_leaf(entries, idx), ranges)
            .is_some(),
    })
}

/// mst nodes [`node_may_reach`] ruled out, kept as bare sha-256 digests since a car can rule
/// out most of its nodes and a whole cid is several times bigger. a digest names the same
/// bytes whatever codec a cid puts in front of it
/// how many bytes of out-of-order mst nodes a streamed car parks while waiting for their
/// parents. a node evicted past this is fetched again if an ancestor turns out to want it
pub(crate) const SPARSE_PARKED_BUDGET: usize = 64 * 1024 * 1024;

#[derive(Debug, Default)]
pub(crate) struct RuledOut(HashSet<[u8; 32], ahash::RandomState>);

impl RuledOut {
    fn with_hasher(hasher: ahash::RandomState) -> Self {
        Self(HashSet::with_hasher(hasher))
    }

    /// remembers `cid` as ruled out. false for a cid that isn't sha-256, which can't be ruled out
    pub(crate) fn insert(&mut self, cid: &IpldCid) -> bool {
        sha256_digest(cid)
            .map(|digest| self.0.insert(digest))
            .is_some()
    }

    #[inline]
    pub(crate) fn contains(&self, cid: &IpldCid) -> bool {
        !self.0.is_empty() && sha256_digest(cid).is_some_and(|digest| self.0.contains(&digest))
    }

    pub(crate) fn len(&self) -> usize {
        self.0.len()
    }
}

fn sha256_digest(cid: &IpldCid) -> Option<[u8; 32]> {
    let hash = cid.hash();
    if hash.code() != jacquard_common::types::crypto::SHA2_256 {
        return None;
    }
    hash.digest().try_into().ok()
}

#[derive(Debug)]
pub(crate) struct SparseScanner {
    ranges: Vec<KeyRange>,
    blocks: Blocks<CommitBlocks>,
    hasher: ahash::RandomState,
    parked: CidMap<Bytes>,
    unparked: CidMap<Bytes>,
    parked_order: VecDeque<IpldCid>,
    parked_bytes: usize,
    parked_budget: usize,
    evicted: CidSet,
    corrupt: CidMap<String>,
    ruled_out: RuledOut,
    root: Option<IpldCid>,
    root_bounds: Vec<NodeBounds>,
    pending_missing: CidMap<BoundsSet>,
    visited: CidMap<BoundsSet>,
    leaves: Vec<(MstKey, IpldCid)>,
    scratch: Vec<FlatEntry>,
    stack: Vec<(IpldCid, NodeBounds)>,
    stats: ScanStats,
}

impl SparseScanner {
    pub(crate) fn new(ranges: Vec<KeyRange>, blocks: Blocks<CommitBlocks>) -> Self {
        let hasher = ahash::RandomState::new();
        Self {
            ranges,
            blocks,
            parked: HashMap::with_hasher(hasher.clone()),
            unparked: HashMap::with_hasher(hasher.clone()),
            parked_order: VecDeque::new(),
            parked_bytes: 0,
            parked_budget: usize::MAX,
            evicted: HashSet::with_hasher(hasher.clone()),
            corrupt: HashMap::with_hasher(hasher.clone()),
            ruled_out: RuledOut::with_hasher(hasher.clone()),
            root: None,
            root_bounds: Vec::new(),
            pending_missing: HashMap::with_hasher(hasher.clone()),
            visited: HashMap::with_hasher(hasher.clone()),
            hasher,
            leaves: Vec::new(),
            scratch: Vec::new(),
            stack: Vec::new(),
            stats: ScanStats::default(),
        }
    }

    pub(crate) fn with_parked_budget(ranges: Vec<KeyRange>, parked_budget: usize) -> Self {
        Self {
            parked_budget,
            ..Self::new(ranges, Blocks::default())
        }
    }

    pub(crate) fn insert_blocks(&mut self, blocks: Blocks<CommitBlocks>) {
        self.blocks.extend(blocks);
    }

    /// nodes a streamed car already ruled out, which the scan then reads as empty instead of
    /// missing
    #[cfg(test)]
    pub(crate) fn rule_out(&mut self, ruled_out: RuledOut) {
        self.ruled_out = ruled_out;
    }

    pub(crate) fn is_complete(&self) -> bool {
        self.root.is_some() && self.pending_missing.is_empty()
    }

    pub(crate) fn was_evicted(&self, cid: &IpldCid) -> bool {
        !self.evicted.is_empty() && self.evicted.contains(cid)
    }

    #[cfg(test)]
    pub(crate) fn evicted_any(&self) -> bool {
        !self.evicted.is_empty()
    }

    pub(crate) fn is_pending(&self, cid: &IpldCid) -> bool {
        self.pending_missing.contains_key(cid)
    }

    pub(crate) fn ruled_out_len(&self) -> usize {
        self.ruled_out.len()
    }

    #[cfg(test)]
    pub(crate) fn parked_bytes(&self) -> usize {
        self.parked_bytes
    }

    /// resets parked state before re-streaming a whole car for nodes evicted on an earlier pass
    pub(crate) fn prepare_pass(&mut self) {
        if !self.parked.is_empty() {
            self.parked = HashMap::with_hasher(self.hasher.clone());
            self.parked_order = VecDeque::new();
            self.parked_bytes = 0;
        }
        if !self.unparked.is_empty() {
            self.unparked = HashMap::with_hasher(self.hasher.clone());
        }
        if !self.evicted.is_empty() {
            self.evicted = HashSet::with_hasher(self.hasher.clone());
        }
        if !self.corrupt.is_empty() {
            self.corrupt = HashMap::with_hasher(self.hasher.clone());
        }
    }

    /// drops parked nodes and ruled-out digests once the stream ends and no evicted node is pending
    pub(crate) fn finish_stream(&mut self) {
        if self.root.is_some() && (self.evicted.is_empty() || self.is_complete()) {
            self.clear_parked();
        }
    }

    /// sets the mst root once the car's commit block arrives and resolves any parked nodes
    /// reachable from it
    pub(crate) fn start(&mut self, root: IpldCid) -> Result<()> {
        if let Some(existing_root) = self.root {
            if existing_root != root {
                return Err(miette::miette!(
                    "sparse scanner root changed from {existing_root} to {root}"
                ));
            }
            return Ok(());
        }
        self.root = Some(root);
        self.scan_subtree(root, NodeBounds::unbounded())?;
        if self.is_complete() {
            self.clear_parked();
        }
        Ok(())
    }

    /// offers a non-record block from a streaming car. a node the walk already wants is
    /// resolved with its real bounds; one that arrives before its parent is parked if it
    /// could reach `ranges`, or remembered in `ruled_out` as a digest; once the walk is
    /// complete, everything else is dropped unparked.
    pub(crate) fn offer_block(&mut self, cid: IpldCid, bytes: &[u8]) -> Result<()> {
        if self.is_complete() {
            return Ok(());
        }
        if let Some(bounds_set) = self.pending_missing.remove(&cid) {
            self.evicted.remove(&cid);
            self.scratch.clear();
            decode_node_into(bytes, &mut self.scratch)
                .wrap_err_with(|| format!("failed to decode mst node {cid}"))?;
            let scratch = std::mem::take(&mut self.scratch);
            let mut stack = std::mem::take(&mut self.stack);
            stack.clear();
            for bounds in bounds_set.into_iter() {
                if !self.is_visited(&cid, &bounds) {
                    self.visit_entries(cid, bytes.len(), &scratch, bounds, &mut stack);
                }
            }
            self.scratch = scratch;
            let res = self.drain_stack(&mut stack);
            self.stack = stack;
            res?;
            if self.is_complete() {
                self.clear_parked();
            }
            return Ok(());
        }
        if self.ruled_out.contains(&cid)
            || self.visited.contains_key(&cid)
            || (!self.parked.is_empty() && self.parked.contains_key(&cid))
        {
            return Ok(());
        }
        self.scratch.clear();
        if let Err(err) = decode_node_into(bytes, &mut self.scratch) {
            self.corrupt.insert(cid, format!("{err:#}"));
            return Ok(());
        }
        // a node whose cid can't be remembered as ruled out has to be parked instead
        if self.scratch_may_reach() || !self.ruled_out.insert(&cid) {
            self.park(cid, Bytes::copy_from_slice(bytes));
        }
        Ok(())
    }

    pub(crate) fn scan(&mut self, root: IpldCid) -> Result<Result<SparseScanOutput, Vec<IpldCid>>> {
        if let Some(existing_root) = self.root {
            if existing_root != root {
                return Err(miette::miette!(
                    "sparse scanner root changed from {existing_root} to {root}"
                ));
            }
        } else {
            self.root = Some(root);
            self.scan_subtree(root, NodeBounds::unbounded())?;
        }

        if !self.pending_missing.is_empty() {
            let mut ready = self
                .pending_missing
                .iter()
                .filter(|(cid, _)| {
                    self.blocks.get(cid).is_some()
                        || (!self.parked.is_empty() && self.parked.contains_key(cid))
                        || (!self.unparked.is_empty() && self.unparked.contains_key(cid))
                })
                .map(|(cid, bounds)| (*cid, bounds.clone()))
                .collect::<Vec<_>>();
            ready.sort_unstable_by_key(|(cid, _)| *cid);
            for (cid, bounds_set) in ready {
                self.pending_missing.remove(&cid);
                for bounds in bounds_set.into_iter() {
                    self.scan_subtree(cid, bounds)?;
                }
            }
        }

        if self.pending_missing.is_empty() {
            self.clear_parked();
            Ok(Ok(SparseScanOutput {
                leaves: SortedLeaves::from_raw(std::mem::take(&mut self.leaves)),
                node_blocks_seen: self.stats.node_blocks_seen,
                node_bytes_seen: self.stats.node_bytes_seen,
            }))
        } else {
            let mut missing = self.pending_missing.keys().copied().collect::<Vec<_>>();
            missing.sort_unstable();
            Ok(Err(missing))
        }
    }

    pub(crate) fn take_blocks(self) -> Blocks<CommitBlocks> {
        self.blocks
    }

    fn root_visited(&self) -> bool {
        self.root.is_some_and(|root| self.visited.contains_key(&root))
    }

    fn scratch_may_reach(&self) -> bool {
        if !node_may_reach(&self.ranges, &self.scratch) {
            return false;
        }
        if !self.root_visited() {
            return true;
        }
        if self.pending_missing.len() <= 8 {
            return self.pending_missing.values().any(|bounds_set| {
                bounds_set.iter().any(|bounds| {
                    entries_within_bounds(bounds, &self.scratch)
                        && node_may_reach_within(bounds, &self.ranges, &self.scratch)
                })
            });
        }
        self.root_bounds.iter().any(|bounds| {
            entries_within_bounds(bounds, &self.scratch)
                && node_may_reach_within(bounds, &self.ranges, &self.scratch)
        })
    }

    fn is_visited(&self, cid: &IpldCid, bounds: &NodeBounds) -> bool {
        self.visited
            .get(cid)
            .is_some_and(|seen| seen.covers(bounds))
    }

    fn record_visited(&mut self, cid: IpldCid, byte_len: usize, bounds: NodeBounds) {
        if !self.evicted.is_empty() {
            self.evicted.remove(&cid);
        }
        match self.visited.entry(cid) {
            std::collections::hash_map::Entry::Occupied(mut occupied) => {
                occupied.get_mut().insert(bounds);
            }
            std::collections::hash_map::Entry::Vacant(vacant) => {
                self.stats.node_blocks_seen += 1;
                self.stats.node_bytes_seen += byte_len;
                vacant.insert(BoundsSet::new(bounds));
            }
        }
    }

    fn record_pending(&mut self, cid: IpldCid, bounds: NodeBounds) {
        match self.pending_missing.entry(cid) {
            std::collections::hash_map::Entry::Occupied(mut occupied) => {
                occupied.get_mut().insert(bounds);
            }
            std::collections::hash_map::Entry::Vacant(vacant) => {
                vacant.insert(BoundsSet::new(bounds));
            }
        }
    }

    fn park(&mut self, cid: IpldCid, bytes: Bytes) {
        if !self.parked.is_empty() && self.parked.contains_key(&cid) {
            return;
        }
        if !self.evicted.is_empty() {
            self.evicted.remove(&cid);
        }
        self.parked_bytes += bytes.len();
        self.parked.insert(cid, bytes);
        self.parked_order.push_back(cid);
        while self.parked_bytes > self.parked_budget {
            let Some(oldest) = self.parked_order.pop_front() else {
                break;
            };
            if let Some(evicted) = self.parked.remove(&oldest) {
                self.parked_bytes -= evicted.len();
                self.evicted.insert(oldest);
            }
        }
    }

    fn unpark(&mut self, cid: &IpldCid) -> Option<(Bytes, usize)> {
        if !self.parked.is_empty()
            && let Some(bytes) = self.parked.remove(cid)
        {
            let byte_len = bytes.len();
            self.parked_bytes -= byte_len;
            if self.parked.is_empty() {
                self.parked_order.clear();
            } else {
                while self
                    .parked_order
                    .front()
                    .is_some_and(|front| !self.parked.contains_key(front))
                {
                    self.parked_order.pop_front();
                }
            }
            self.unparked.insert(*cid, bytes.clone());
            return Some((bytes, byte_len));
        }
        if !self.unparked.is_empty()
            && let Some(bytes) = self.unparked.get(cid).cloned()
        {
            return Some((bytes, 0));
        }
        None
    }

    fn clear_parked(&mut self) {
        if !self.parked.is_empty() {
            self.parked = HashMap::with_hasher(self.hasher.clone());
            self.parked_order = VecDeque::new();
            self.parked_bytes = 0;
        }
        if !self.unparked.is_empty() {
            self.unparked = HashMap::with_hasher(self.hasher.clone());
        }
        if !self.corrupt.is_empty() {
            self.corrupt = HashMap::with_hasher(self.hasher.clone());
        }
        if self.ruled_out.len() > 0 {
            self.ruled_out = RuledOut::with_hasher(self.hasher.clone());
        }
        self.scratch = Vec::new();
        self.stack = Vec::new();
    }

    /// scans the subtree under `root`. proof cars come from the pds, so the scan keeps its own
    /// stack rather than recursing once per tree level.
    fn scan_subtree(&mut self, root: IpldCid, bounds: NodeBounds) -> Result<()> {
        let mut stack = std::mem::take(&mut self.stack);
        stack.clear();
        stack.push((root, bounds));
        let res = self.drain_stack(&mut stack);
        self.stack = stack;
        res
    }

    fn drain_stack(&mut self, stack: &mut Vec<(IpldCid, NodeBounds)>) -> Result<()> {
        while let Some((cid, bounds)) = stack.pop() {
            let mut bounds_set = BoundsSet::new(bounds);
            while let Some(pos) = stack.iter().rposition(|(other, _)| *other == cid) {
                let (_, other_bounds) = stack.remove(pos);
                bounds_set.insert(other_bounds);
            }
            if bounds_set.iter().all(|b| self.is_visited(&cid, b)) || self.ruled_out.contains(&cid)
            {
                continue;
            }
            let (bytes, byte_len) = if let Some((bytes, byte_len)) = self.unpark(&cid) {
                (bytes, byte_len)
            } else if let Some(bytes) = self.blocks.get(&cid).cloned() {
                let len = bytes.len();
                (bytes, len)
            } else if !self.corrupt.is_empty()
                && let Some(err) = self.corrupt.get(&cid)
            {
                return Err(
                    miette::miette!("{err}").wrap_err(format!("failed to decode mst node {cid}"))
                );
            } else {
                for b in bounds_set.into_iter() {
                    if !self.is_visited(&cid, &b) {
                        self.record_pending(cid, b);
                    }
                }
                continue;
            };

            self.scratch.clear();
            decode_node_into(&bytes, &mut self.scratch)
                .wrap_err_with(|| format!("failed to decode mst node {cid}"))?;
            let scratch = std::mem::take(&mut self.scratch);
            for b in bounds_set.into_iter() {
                if !self.is_visited(&cid, &b) {
                    self.visit_entries(cid, byte_len, &scratch, b, stack);
                }
            }
            self.scratch = scratch;
        }

        Ok(())
    }

    fn visit_entries(
        &mut self,
        cid: IpldCid,
        byte_len: usize,
        entries: &[FlatEntry],
        bounds: NodeBounds,
        stack: &mut Vec<(IpldCid, NodeBounds)>,
    ) {
        let is_root = self.root == Some(cid) && self.root_bounds.is_empty();
        self.record_visited(cid, byte_len, bounds.clone());

        let stack_start = stack.len();
        for idx in 0..entries.len() {
            match &entries[idx] {
                FlatEntry::Leaf { key, cid } => {
                    if bounds.contains_leaf(key)
                        && self.ranges.iter().any(|range| range.contains(key))
                    {
                        self.leaves.push((key.clone(), *cid));
                    }
                }
                FlatEntry::Tree { cid } => {
                    if let Some(child) = bounds.narrow(
                        previous_leaf(entries, idx),
                        next_leaf(entries, idx),
                        &self.ranges,
                    ) {
                        if is_root {
                            self.root_bounds.push(child.clone());
                        }
                        stack.push((*cid, child));
                    }
                }
            }
        }
        // reversed, so subtrees are scanned left to right, depth first
        stack[stack_start..].reverse();
    }
}

#[derive(Debug, Default)]
struct ScanStats {
    node_blocks_seen: usize,
    node_bytes_seen: usize,
}

fn previous_leaf(entries: &[FlatEntry], idx: usize) -> Option<&MstKey> {
    match idx.checked_sub(1).and_then(|prev| entries.get(prev)) {
        Some(FlatEntry::Leaf { key, .. }) => Some(key),
        _ => None,
    }
}

fn next_leaf(entries: &[FlatEntry], idx: usize) -> Option<&MstKey> {
    match entries.get(idx + 1) {
        Some(FlatEntry::Leaf { key, .. }) => Some(key),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use super::*;
    use crate::mst::RawNodeData;
    use crate::mst::tests::{cid, left_chain, node};
    use bytes::Bytes;

    #[test]
    fn builds_exact_and_wildcard_ranges() {
        let ranges = sparse_ranges(&[SmolStr::new("sh.tangled.repo"), SmolStr::new("app.bsky.*")]);

        assert!(ranges[0].contains("sh.tangled.repo/abc"));
        assert!(!ranges[0].contains("sh.tangled.repo.comment/abc"));
        assert!(ranges[1].contains("app.bsky.feed.post/abc"));
        assert!(!ranges[1].contains("app.bskyfoo.feed.post/abc"));
        assert!(!ranges[1].contains("app.csky.feed.post/abc"));
    }

    #[test]
    fn synthesizes_probe_collection_for_wildcard() {
        assert_eq!(
            sparse_probe_collection(&[SmolStr::new("sh.tangled.*")]).as_deref(),
            Some("sh.tangled.probe")
        );
        assert_eq!(
            sparse_probe_collection(&[SmolStr::new("sh.tangled.repo")]).as_deref(),
            Some("sh.tangled.repo")
        );
    }

    #[test]
    fn reports_missing_intersecting_subtrees_only() {
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]);
        let left = cid(1);
        let middle = cid(2);
        let right = cid(3);
        let root = cid(4);
        let root_node = node(
            vec![
                ("app.bsky.feed.post/1", cid(5), Some(middle)),
                ("zz.example.record/1", cid(6), Some(right)),
            ],
            Some(left),
        );
        let root_bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap();
        let mut scanner = SparseScanner::new(
            wanted,
            Blocks::unverified(BTreeMap::from([(root, root_bytes.into())])),
        );

        let missing = scanner.scan(root).unwrap().unwrap_err();

        assert_eq!(missing, vec![middle]);
    }

    #[test]
    fn resumes_without_rescanning_known_nodes() {
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]);
        let middle = cid(1);
        let root = cid(2);
        let record = cid(3);
        let root_node = node(
            vec![
                ("app.bsky.feed.post/1", cid(4), Some(middle)),
                ("zz.example.record/1", cid(5), None),
            ],
            None,
        );
        let middle_node = node(vec![("sh.tangled.repo/1", record, None)], None);
        let mut scanner = SparseScanner::new(
            wanted,
            Blocks::unverified(BTreeMap::from([(
                root,
                serde_ipld_dagcbor::to_vec(&root_node).unwrap().into(),
            )])),
        );

        assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![middle]);

        scanner.insert_blocks(Blocks::unverified(BTreeMap::from([(
            middle,
            serde_ipld_dagcbor::to_vec(&middle_node).unwrap().into(),
        )])));
        let scan = scanner.scan(root).unwrap().unwrap();

        assert_eq!(
            scan.leaves,
            vec![(SmolStr::new("sh.tangled.repo/1"), record)]
        );
        assert_eq!(scan.node_blocks_seen, 2);
    }

    #[test]
    fn node_may_reach_checks_leaves_and_every_subtree_interval() {
        let wanted = sparse_ranges(&[SmolStr::new("mm.x")]);
        let leaf = |key: &str| FlatEntry::Leaf {
            key: MstKey::new(key),
            cid: cid(9),
        };
        let tree = FlatEntry::Tree { cid: cid(8) };

        assert!(node_may_reach(&wanted, &[leaf("mm.x/1")]));
        assert!(!node_may_reach(&wanted, &[leaf("aa.a/1"), leaf("zz.z/1")]));
        assert!(!node_may_reach(&wanted, &[]));
        // between two keys the interval is known exactly
        assert!(node_may_reach(
            &wanted,
            &[leaf("aa.a/1"), tree.clone(), leaf("zz.z/1")]
        ));
        assert!(!node_may_reach(
            &wanted,
            &[leaf("aa.a/1"), tree.clone(), leaf("bb.b/1")]
        ));
        // at either edge the node's real bound is unknown, so it could reach anything past it
        assert!(node_may_reach(&wanted, &[tree.clone(), leaf("zz.z/1")]));
        assert!(!node_may_reach(&wanted, &[tree.clone(), leaf("aa.a/1")]));
        assert!(node_may_reach(&wanted, &[leaf("aa.a/1"), tree.clone()]));
        assert!(!node_may_reach(&wanted, &[leaf("zz.z/1"), tree]));
    }

    #[test]
    fn scan_over_kept_nodes_finds_every_wanted_leaf() {
        let leaves = (0..3000usize).map(|i| {
            // mostly likes and posts with a few records of everything else, the way a real
            // account's repo looks
            let collection = match i % 300 {
                0 => "sh.tangled.repo",
                150 => "sh.tangled.repo.collaborator",
                7 | 107 | 207 => "io.atcr.hold.layer",
                n if n % 4 == 0 => "app.bsky.feed.post",
                n if n % 20 == 1 => "zz.example.record",
                _ => "app.bsky.feed.like",
            };
            (format!("{collection}/{i:013}"), cid((i % 251) as u8))
        });
        let (root, nodes) = crate::mst::build_tree(leaves).unwrap();
        let all: crate::car::CarBlocks = nodes.into_iter().collect();
        let every_leaf = crate::mst::leaves(&all, root).unwrap();

        // whether the patterns want few enough keys that the leaf-level nodes, most of a tree,
        // should all be ruled out
        for (patterns, selective) in [
            (
                vec!["sh.tangled.repo", "sh.tangled.repo.collaborator"],
                true,
            ),
            (vec!["sh.tangled.*"], true),
            (vec!["app.bsky.feed.post"], false),
            (vec!["io.atcr.*", "zz.example.record"], true),
            (vec!["nothing.here"], true),
            (vec!["app.bsky.*", "sh.tangled.repo"], false),
        ] {
            let ranges = sparse_ranges(&patterns.iter().map(SmolStr::new).collect::<Vec<_>>());
            let mut kept = BTreeMap::new();
            let mut ruled_out = RuledOut::default();
            for (cid, bytes) in &all {
                if node_may_reach(&ranges, &decode_node(bytes).unwrap()) {
                    kept.insert(*cid, bytes.clone());
                } else {
                    assert!(ruled_out.insert(cid));
                }
            }
            let mut scanner = SparseScanner::new(ranges.clone(), Blocks::unverified(kept.clone()));
            scanner.rule_out(ruled_out);

            let scan = scanner
                .scan(root)
                .unwrap()
                .unwrap_or_else(|missing| panic!("{patterns:?}: missing {missing:?}"));

            let want = every_leaf
                .iter()
                .filter(|(key, _)| ranges.iter().any(|range| range.contains(key)))
                .cloned()
                .collect::<Vec<_>>();
            assert_eq!(scan.leaves, want, "{patterns:?}");
            assert!(
                !selective || kept.len() * 3 < all.len(),
                "{patterns:?}: kept {} of {} nodes",
                kept.len(),
                all.len()
            );

            // streaming the same nodes in either order resolves the walk on the fly and drops
            // everything it parked once complete
            let ordered = all.iter().map(|(cid, bytes)| (*cid, bytes.clone())).collect::<Vec<_>>();
            for (label, start_first, pass) in [
                ("top-down", true, ordered.clone()),
                ("bottom-up", false, ordered.into_iter().rev().collect()),
            ] {
                let mut streamed = SparseScanner::with_parked_budget(ranges.clone(), usize::MAX);
                if start_first {
                    streamed.start(root).unwrap();
                }
                for (cid, bytes) in &pass {
                    streamed.offer_block(*cid, bytes).unwrap();
                }
                if !start_first {
                    streamed.start(root).unwrap();
                }
                assert!(streamed.is_complete(), "{patterns:?} ({label})");
                assert_eq!(streamed.parked_bytes(), 0, "{patterns:?} ({label})");
                assert_eq!(streamed.ruled_out_len(), 0, "{patterns:?} ({label})");
                let scan = streamed.scan(root).unwrap().unwrap();
                assert_eq!(scan.leaves, want, "{patterns:?} ({label})");
            }
        }
    }

    #[test]
    fn estimates_node_layer_from_first_leaf() {
        let root_node = node(vec![("sh.tangled.repo/1", cid(1), None)], None);
        let bytes = serde_ipld_dagcbor::to_vec(&root_node).unwrap();

        assert_eq!(
            mst_node_layer(&bytes).unwrap(),
            Some(layer_for_key("sh.tangled.repo/1"))
        );
    }

    fn car_bytes(node: &RawNodeData<'_>) -> Bytes {
        serde_ipld_dagcbor::to_vec(node).unwrap().into()
    }

    #[test]
    fn ancestor_bounds_prune_out_of_range_boundary_subtrees() {
        // two disjoint ranges; the fetched middle subtree's leftmost child sits
        // entirely between them once the ancestor lower bound is applied.
        let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]);
        let boundary_child = cid(1);
        let middle = cid(2);
        let root = cid(3);
        let root_node = node(
            vec![
                ("aa.x/1", cid(4), None),
                ("bb.q/1", cid(5), Some(middle)),
                ("zz.y/5", cid(6), None),
            ],
            None,
        );
        // boundary_child spans ("bb.q/1", "mm.a/1") given the ancestor bound,
        // which intersects neither range; without the bound its interval is
        // (-inf, "mm.a/1"), which falsely intersects the aa.x range.
        let middle_node = node(
            vec![("mm.a/1", cid(7), None), ("zz.y/1", cid(8), None)],
            Some(boundary_child),
        );
        let mut scanner = SparseScanner::new(
            wanted,
            Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])),
        );

        assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![middle]);

        scanner.insert_blocks(Blocks::unverified(BTreeMap::from([(
            middle,
            car_bytes(&middle_node),
        )])));
        let scan = scanner.scan(root).unwrap().unwrap();

        assert_eq!(
            scan.leaves,
            vec![
                (SmolStr::new("aa.x/1"), cid(4)),
                (SmolStr::new("zz.y/1"), cid(8)),
                (SmolStr::new("zz.y/5"), cid(6)),
            ]
        );
    }

    #[test]
    fn scans_a_tree_deeper_than_the_thread_stack() {
        // proof cars come from the pds, so their depth is not bounded by anything we trust
        let depth = 100_000;
        let (top, nodes) = left_chain(depth, "sh.tangled.repo/1");
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]);
        let mut scanner =
            SparseScanner::new(wanted, Blocks::unverified(nodes.into_iter().collect()));

        let scan = scanner.scan(top).unwrap().unwrap();

        assert_eq!(
            scan.leaves,
            vec![(SmolStr::new("sh.tangled.repo/1"), cid(9))]
        );
        assert_eq!(scan.node_blocks_seen, depth + 1);
    }

    #[test]
    fn stops_on_a_node_that_is_its_own_left_subtree() {
        // an unverified car can claim any cid; the narrower child bounds are already covered
        let root = cid(1);
        let root_node = node(vec![("sh.tangled.repo/1", cid(9), None)], Some(root));
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]);
        let mut scanner = SparseScanner::new(
            wanted,
            Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])),
        );

        let scan = scanner.scan(root).unwrap().unwrap();

        assert_eq!(
            scan.leaves,
            vec![(SmolStr::new("sh.tangled.repo/1"), cid(9))]
        );
        assert_eq!(scan.node_blocks_seen, 1);
    }

    #[test]
    fn duplicate_pending_cids_fetch_once_and_scan_deterministically() {
        // the same missing child CID is referenced from two positions with
        // different ancestor bounds; it must appear once in the missing list
        // and both contexts must be scanned after a single insert.
        let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]);
        let shared = cid(1);
        let root = cid(2);
        let root_node = node(
            vec![
                ("aa.w/1", cid(3), Some(shared)),
                ("bb.q/1", cid(4), None),
                ("zz.x/1", cid(5), Some(shared)),
            ],
            None,
        );
        let shared_node = node(
            vec![("aa.x/1", cid(6), None), ("zz.y/1", cid(7), None)],
            None,
        );
        let mut scanner = SparseScanner::new(
            wanted,
            Blocks::unverified(BTreeMap::from([(root, car_bytes(&root_node))])),
        );

        assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![shared]);

        scanner.insert_blocks(Blocks::unverified(BTreeMap::from([(
            shared,
            car_bytes(&shared_node),
        )])));
        let scan = scanner.scan(root).unwrap().unwrap();

        assert_eq!(
            scan.leaves,
            vec![
                (SmolStr::new("aa.x/1"), cid(6)),
                (SmolStr::new("zz.y/1"), cid(7)),
            ]
        );
        assert_eq!(scan.node_blocks_seen, 2);

        // the same shared node arriving before its parents in a streamed car (including across
        // distinct intermediate parent branches) must be visited under both ancestor bounds
        let left_branch = cid(8);
        let right_branch = cid(9);
        let left_branch_node = node(vec![("aa.w/1", cid(3), Some(shared))], None);
        let right_branch_node = node(vec![("zz.x/1", cid(5), Some(shared))], None);
        let diamond_root = node(
            vec![("bb.q/1", cid(4), Some(right_branch))],
            Some(left_branch),
        );
        let wanted = sparse_ranges(&[SmolStr::new("aa.x"), SmolStr::new("zz.y")]);
        let mut streamed = SparseScanner::with_parked_budget(wanted, usize::MAX);
        streamed
            .offer_block(shared, &car_bytes(&shared_node))
            .unwrap();
        streamed
            .offer_block(left_branch, &car_bytes(&left_branch_node))
            .unwrap();
        streamed
            .offer_block(right_branch, &car_bytes(&right_branch_node))
            .unwrap();
        streamed
            .offer_block(root, &car_bytes(&diamond_root))
            .unwrap();
        streamed.start(root).unwrap();
        let scan = streamed.scan(root).unwrap().unwrap();
        assert_eq!(
            scan.leaves,
            vec![
                (SmolStr::new("aa.x/1"), cid(6)),
                (SmolStr::new("zz.y/1"), cid(7)),
            ]
        );
        assert_eq!(scan.node_blocks_seen, 4);
        assert_eq!(streamed.parked_bytes(), 0);
    }

    #[test]
    fn enforces_intra_node_key_order_ancestor_bounds_and_pending_decode_errors() {
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.*")]);
        let root = cid(1);
        let unsorted = node(
            vec![
                ("sh.tangled.repo/b", cid(2), None),
                ("sh.tangled.repo/a", cid(3), None),
            ],
            None,
        );
        let mut scanner = SparseScanner::new(
            wanted.clone(),
            Blocks::unverified(BTreeMap::from([(root, car_bytes(&unsorted))])),
        );
        let err = scanner.scan(root).unwrap_err();
        assert!(
            format!("{err:?}").contains("does not sort after"),
            "{err:?}"
        );

        // a child cannot overwrite its parent's boundary key or smuggle out-of-bounds leaves
        let right = cid(4);
        let left = cid(5);
        let bounded_root = node(
            vec![
                ("sh.tangled.repo/b", cid(2), Some(right)),
                ("sh.tangled.repo/d", cid(3), None),
            ],
            Some(left),
        );
        let out_of_bounds_left = node(vec![("sh.tangled.repo/z", cid(8), None)], None);
        let dup_and_valid_right = node(
            vec![
                ("sh.tangled.repo/b", cid(9), None),
                ("sh.tangled.repo/c", cid(6), None),
                ("sh.tangled.repo/e", cid(7), None),
            ],
            None,
        );
        let mut scanner = SparseScanner::new(
            wanted.clone(),
            Blocks::unverified(BTreeMap::from([
                (root, car_bytes(&bounded_root)),
                (left, car_bytes(&out_of_bounds_left)),
                (right, car_bytes(&dup_and_valid_right)),
            ])),
        );
        let scan = scanner.scan(root).unwrap().unwrap();
        assert_eq!(
            scan.leaves,
            vec![
                (SmolStr::new("sh.tangled.repo/b"), cid(2)),
                (SmolStr::new("sh.tangled.repo/c"), cid(6)),
                (SmolStr::new("sh.tangled.repo/d"), cid(3)),
            ]
        );

        // wanted pending node whose block fails to decode errors immediately in offer_block,
        // whether it arrives after its parent or out of order before its parent
        let mut streamed = SparseScanner::with_parked_budget(wanted.clone(), usize::MAX);
        streamed
            .offer_block(root, &car_bytes(&bounded_root))
            .unwrap();
        streamed.start(root).unwrap();
        assert!(streamed.is_pending(&right));
        let err = streamed.offer_block(right, b"not cbor").unwrap_err();
        assert!(
            err.to_string().contains("failed to decode mst node"),
            "{err}"
        );

        let mut out_of_order = SparseScanner::with_parked_budget(wanted, usize::MAX);
        out_of_order.offer_block(right, b"not cbor").unwrap();
        out_of_order
            .offer_block(left, &car_bytes(&out_of_bounds_left))
            .unwrap();
        out_of_order
            .offer_block(root, &car_bytes(&bounded_root))
            .unwrap();
        let err = out_of_order.start(root).unwrap_err();
        assert!(
            err.to_string().contains("failed to decode mst node"),
            "{err}"
        );
    }

    #[test]
    fn evicts_oldest_parked_nodes_fifo_and_resumes_on_second_pass() {
        let wanted = sparse_ranges(&[SmolStr::new("sh.tangled.repo")]);
        let child = cid(1);
        let stale = cid(2);
        let root = cid(3);
        let record = cid(4);
        let child_bytes = car_bytes(&node(vec![("sh.tangled.repo/1", record, None)], None));
        // an unreachable historical node with an open left subtree that `node_may_reach`
        // cannot rule out without ancestor bounds
        let stale_bytes = car_bytes(&node(vec![("zz.other/1", cid(5), None)], Some(cid(6))));
        let root_bytes = car_bytes(&node(
            vec![("app.bsky.feed.post/1", cid(7), Some(child))],
            None,
        ));

        // budget fits one parked node at a time, so parking `stale` evicts `child`
        let mut scanner = SparseScanner::with_parked_budget(wanted, stale_bytes.len());
        scanner.start(root).unwrap();
        scanner.offer_block(child, &child_bytes).unwrap();
        assert_eq!(scanner.parked_bytes(), child_bytes.len());
        assert!(!scanner.evicted_any());

        scanner.offer_block(stale, &stale_bytes).unwrap();
        assert!(scanner.evicted_any());
        assert_eq!(scanner.parked_bytes(), stale_bytes.len());

        scanner.offer_block(root, &root_bytes).unwrap();
        assert!(!scanner.is_complete());
        assert_eq!(scanner.scan(root).unwrap().unwrap_err(), vec![child]);

        // a second pass resolves `child` directly because `root` already queued it in
        // `pending_missing`, without re-visiting `root`
        scanner.prepare_pass();
        assert!(!scanner.evicted_any());
        scanner.offer_block(child, &child_bytes).unwrap();
        assert!(scanner.is_complete());
        assert_eq!(scanner.parked_bytes(), 0);
        let scan = scanner.scan(root).unwrap().unwrap();
        assert_eq!(
            scan.leaves,
            vec![(SmolStr::new("sh.tangled.repo/1"), record)]
        );
        assert_eq!(scan.node_blocks_seen, 2);
    }
}