//! The partial MST a commit's proof describes, and the operations over it.
//!
//! synchronous and single-threaded, since only 200 ops are allowed, and the
//! CAR (including records) can only be up to 2MB, so it should be fast enough
//! to call from an async context.
//!
//! Nodes are decoded when traversal first reaches them. A proof usually
//! contains exactly the nodes the operations touch, so in the normal case it's
//! the same thing.

use std::cmp::Ordering;
use std::mem;

use crate::blocks::Blocks;
use crate::cid::Cid;
use crate::error::{InvertError, NodeFault};
use crate::key::Key;
use crate::key::{MAX_KEY_LEN, layer_of, shared_prefix_len};
use crate::node::{EntryData, NodeData, cid_for};
use crate::op::Op;

/// Where blocks come from.
///
/// `Sync` so a [`Tree`] held across an await point keeps the future `Send`;
/// every real block source is a plain lookup table anyway.
type Source<'a> = &'a (dyn Blocks + Sync);

/// One slot in a node: either a record, or a pointer to the subtree covering
/// the keys between this slot's neighbours.
#[derive(Debug)]
enum Entry {
    Leaf {
        key: Box<[u8]>,
        value: Cid,
    },
    Child {
        cid: Option<Cid>,
        node: Option<Node>,
    },
}

impl Entry {
    fn is_child(&self) -> bool {
        matches!(self, Entry::Child { .. })
    }

    /// The subtree here, decoding its block if this is the first time anyone
    /// has looked, and checking it really does sit below `parent`.
    ///
    /// Recursion is bounded because an adversarial MST can't be arbitrarily
    /// tall, because mining higher-layer keys gets exponentially more expensive
    /// (height=80 is ~infeasible)
    fn child_mut(&mut self, blocks: Source<'_>, parent: u8) -> Result<&mut Node, InvertError> {
        let Entry::Child { cid, node } = self else {
            return Err(InvertError::Encode {
                detail: "expected a subtree pointer, found a leaf".into(),
            });
        };
        if node.is_none() {
            let expected = parent.checked_sub(1).ok_or(InvertError::MalformedNode {
                cid: *cid,
                detail: NodeFault::SubtreeAtLeafLayer,
            })?;
            *node = Some(Node::decode_at(blocks, own_cid(*cid)?, expected)?);
        }
        let child = node.as_mut().ok_or_else(|| InvertError::Encode {
            detail: "subtree vanished after decoding".into(),
        })?;
        child.check_below(parent)?;
        Ok(child)
    }

    fn into_child(self, blocks: Source<'_>, parent: u8) -> Result<Node, InvertError> {
        let child = match self {
            Entry::Child {
                node: Some(node), ..
            } => node,
            Entry::Child { cid, node: None } => {
                let expected = parent.checked_sub(1).ok_or(InvertError::MalformedNode {
                    cid,
                    detail: NodeFault::SubtreeAtLeafLayer,
                })?;
                Node::decode_at(blocks, own_cid(cid)?, expected)?
            }
            Entry::Leaf { .. } => {
                return Err(InvertError::Encode {
                    detail: "expected a subtree pointer, found a leaf".into(),
                });
            }
        };
        child.check_below(parent)?;
        Ok(child)
    }
}

/// A pointer with neither a CID nor a node is one we built ourselves and have
/// not re-encoded yet, which `root()` always does first.
fn own_cid(cid: Option<Cid>) -> Result<Cid, InvertError> {
    cid.ok_or_else(|| InvertError::Encode {
        detail: "subtree pointer with neither a CID nor a node".into(),
    })
}

#[derive(Debug, Default)]
struct Node {
    /// Slots in key order, alternating between subtrees and leaves.
    entries: Vec<Entry>,
    /// Layer this node sits at, counting up from the leaves. Meaningless while
    /// `stub` is set, since the block that would say has not been read.
    layer: u8,
    /// Entries changed since this node was decoded, so `cid` is stale.
    dirty: bool,
    cid: Option<Cid>,
    /// We know this node's CID but have not decoded its block. Fine to carry,
    /// and fine to answer `root()` with — it is only an error if an operation
    /// needs to look inside and the block turns out to be absent.
    stub: bool,
}

/// The tree a commit's proof describes, ready to have its operations inverted.
pub struct Tree<'a> {
    root: Node,
    blocks: Source<'a>,
}

impl<'a> Tree<'a> {
    /// The tree from `root`, to be decoded as operations reach into it.
    pub fn new(blocks: &'a (impl Blocks + Sync), root: Cid) -> Self {
        Self {
            root: Node {
                cid: Some(root),
                stub: true,
                ..Default::default()
            },
            blocks,
        }
    }

    /// Invert one operation.
    ///
    /// Fails with [`InvertError::OpMismatch`] if the tree does not hold what
    /// the operation says it should, which means the commit is not internally
    /// consistent.
    pub fn invert(&mut self, op: &Op) -> Result<(), InvertError> {
        let key = op.key();
        let found = match op {
            Op::Create { key, .. } => self.root.remove_root(self.blocks, key.as_bytes()),
            Op::Update { key, prev, .. } | Op::Delete { key, prev } => {
                self.root.insert_root(self.blocks, key.as_bytes(), *prev)
            }
        }
        .map_err(|err| with_key(err, key))?;

        let matched = match op {
            // the created record must have been there, holding what it claims
            Op::Create { cid, .. } => found == Some(*cid),
            // the updated record must have been there, holding the new value
            Op::Update { cid, .. } => found == Some(*cid),
            // the deleted record must not have been there
            Op::Delete { .. } => found.is_none(),
        };
        if matched {
            return Ok(());
        }

        let expected = match op {
            Op::Create { cid, .. } | Op::Update { cid, .. } => Some(*cid),
            Op::Delete { .. } => None,
        };
        Err(InvertError::OpMismatch {
            key: key.clone(),
            kind: op.kind(),
            expected,
            found,
        })
    }

    /// Re-encode every changed node bottom-up and return the resulting root.
    pub fn root(&mut self) -> Result<Cid, InvertError> {
        if self.root.stub
            && !self.root.dirty
            && let Some(cid) = self.root.cid
        {
            return Ok(cid);
        }
        self.root.rehash()
    }
}

impl std::fmt::Debug for Tree<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Tree").field("root", &self.root).finish()
    }
}

fn with_key(err: InvertError, key: &Key) -> InvertError {
    match err {
        InvertError::MissingBlock { missing } => InvertError::ProofIncomplete {
            missing,
            key: key.clone(),
        },
        other => other,
    }
}

impl Node {
    /// Decode a subtree, taking `expected` as its layer if it has no keys of
    /// its own to say. A node covering a range whose keys all sit further down
    /// is leafless and legitimate, and only its parent knows where it belongs.
    fn decode_at(blocks: Source<'_>, cid: Cid, expected: u8) -> Result<Node, InvertError> {
        let (entries, layer) = read(blocks, cid)?;
        Ok(Node {
            entries,
            layer: layer.unwrap_or(expected),
            dirty: false,
            cid: Some(cid),
            stub: false,
        })
    }

    /// Decode a root.
    ///
    /// A root has no parent to say what layer it is, and it does not need one:
    /// the root's layer *is* the highest layer any key in the tree reaches, so
    /// a non-empty tree's root always holds at least one of those keys. A
    /// leafless root would be claiming a layer with nothing at it, which is
    /// what `trim` exists to prevent.
    fn decode_root(blocks: Source<'_>, cid: Cid) -> Result<Node, InvertError> {
        let (entries, layer) = read(blocks, cid)?;
        let layer = match layer {
            Some(layer) => layer,
            None if entries.is_empty() => 0, // the empty tree, and only it
            None => {
                return Err(InvertError::MalformedNode {
                    cid: Some(cid),
                    detail: NodeFault::NoLeaves,
                });
            }
        };
        Ok(Node {
            entries,
            layer,
            dirty: false,
            cid: Some(cid),
            stub: false,
        })
    }

    /// Decode this node's own block, if it is still just a pointer.
    ///
    /// Only ever called on the root: every other node is decoded through its
    /// parent, which knows what layer to expect.
    fn wake(&mut self, blocks: Source<'_>) -> Result<(), InvertError> {
        if !self.stub {
            return Ok(());
        }
        *self = Node::decode_root(blocks, own_cid(self.cid)?)?;
        Ok(())
    }

    fn check_below(&self, parent: u8) -> Result<(), InvertError> {
        // an unread pointer has nothing to check yet; whoever wakes it will
        if self.stub || self.layer < parent {
            return Ok(());
        }
        Err(InvertError::MalformedNode {
            cid: self.cid,
            detail: NodeFault::SubtreeNotBelow {
                parent,
                child: self.layer,
            },
        })
    }

    /// The layer this node's subtrees sit at or below.
    fn below(&self) -> Result<u8, InvertError> {
        self.layer.checked_sub(1).ok_or(InvertError::MalformedNode {
            cid: self.cid,
            detail: NodeFault::SubtreeAtLeafLayer,
        })
    }

    fn rehash(&mut self) -> Result<Cid, InvertError> {
        if self.stub {
            // untouched, so its stored CID is still the answer
            return own_cid(self.cid);
        }
        for entry in &mut self.entries {
            if let Entry::Child {
                cid,
                node: Some(node),
            } = entry
                && (node.dirty || cid.is_none())
            {
                *cid = Some(node.rehash()?);
            }
        }
        let cid = cid_for(&encode(&self.entries)?);
        self.cid = Some(cid);
        self.dirty = false;
        Ok(cid)
    }

    /// Put `value` at `key` in the tree this node roots.
    fn insert_root(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        value: Cid,
    ) -> Result<Option<Cid>, InvertError> {
        self.insert(blocks, key, value, layer_of(key))
    }

    /// Put `value` at `key`, returning whatever was there before.
    fn insert(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        value: Cid,
        layer: u8,
    ) -> Result<Option<Cid>, InvertError> {
        self.wake(blocks)?;

        if layer > self.layer {
            return self.insert_above(blocks, key, value, layer);
        }
        if layer < self.layer {
            return self.insert_below(blocks, key, value, layer);
        }

        if let Some(idx) = self.find_leaf(key) {
            let Entry::Leaf { value: slot, .. } = &mut self.entries[idx] else {
                return Err(InvertError::Encode {
                    detail: "find_leaf returned a subtree pointer".into(),
                });
            };
            if *slot == value {
                return Ok(Some(value)); // no-op, leave the spine clean
            }
            let prev = mem::replace(slot, value);
            self.dirty = true;
            return Ok(Some(prev));
        }

        let (idx, splits) = self.find_slot(blocks, key)?;
        self.dirty = true;
        let leaf = Entry::Leaf {
            key: key.into(),
            value,
        };
        if !splits {
            self.entries.insert(idx, leaf);
            return Ok(None);
        }

        // the key lands inside the subtree at `idx`, which has to come apart
        let parent = self.layer;
        let child = self.entries.remove(idx).into_child(blocks, parent)?;
        let (left, right) = child.split(blocks, key)?;
        self.entries
            .splice(idx..idx, [child_of(left), leaf, child_of(right)]);
        Ok(None)
    }

    /// The key belongs at a higher layer than this node: grow upwards.
    fn insert_above(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        value: Cid,
        layer: u8,
    ) -> Result<Option<Cid>, InvertError> {
        if self.entries.is_empty() {
            // nothing to carry up, so this node just *is* the new layer
            *self = Node {
                layer,
                dirty: true,
                ..Default::default()
            };
        } else {
            let mut below = Node {
                layer: self.layer + 1,
                dirty: true,
                ..Default::default()
            };
            mem::swap(self, &mut below);
            self.entries.push(child_of(below));
        }
        self.insert(blocks, key, value, layer)
    }

    /// The key belongs at a lower layer: descend, making a child if needed.
    fn insert_below(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        value: Cid,
        layer: u8,
    ) -> Result<Option<Cid>, InvertError> {
        let parent = self.layer;
        if let Some(idx) = self.find_covering_child(key) {
            let child = self.entries[idx].child_mut(blocks, parent)?;
            let prev = child.insert(blocks, key, value, layer)?;
            if prev != Some(value) {
                self.dirty = true;
            }
            return Ok(prev);
        }

        let (idx, splits) = self.find_slot(blocks, key)?;
        if splits {
            return Err(InvertError::Encode {
                detail: "a key below this layer cannot split a sibling subtree".into(),
            });
        }
        self.dirty = true;
        let mut child = Node {
            layer: self.below()?,
            dirty: true,
            ..Default::default()
        };
        child.insert(blocks, key, value, layer)?;
        self.entries.insert(idx, child_of(child));
        Ok(None)
    }

    /// Take `key` out of the tree this node roots, returning what it held.
    ///
    /// Only the root may reshape the top of the tree, which is why this is
    /// separate from the recursive form.
    fn remove_root(&mut self, blocks: Source<'_>, key: &[u8]) -> Result<Option<Cid>, InvertError> {
        let removed = self.remove(blocks, key, layer_of(key))?;
        if removed.is_some() {
            self.trim(blocks)?;
        }
        Ok(removed)
    }

    fn remove(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        layer: u8,
    ) -> Result<Option<Cid>, InvertError> {
        self.wake(blocks)?;

        if layer > self.layer {
            return Ok(None); // the key would sit above this tree, so it is absent
        }
        if layer < self.layer {
            return self.remove_below(blocks, key, layer);
        }

        let Some(idx) = self.find_leaf(key) else {
            return Ok(None);
        };
        let Entry::Leaf { value: prev, .. } = self.entries[idx] else {
            return Err(InvertError::Encode {
                detail: "find_leaf returned a subtree pointer".into(),
            });
        };
        self.dirty = true;

        // a leaf sitting between two subtrees was the only thing separating
        // them, so they become one
        let parent = self.layer;
        if idx > 0
            && idx + 1 < self.entries.len()
            && self.entries[idx - 1].is_child()
            && self.entries[idx + 1].is_child()
        {
            let mut drained = self.entries.drain(idx..idx + 2);
            drained.next();
            let right = drained.next().ok_or_else(|| InvertError::Encode {
                detail: "drained fewer entries than the range covers".into(),
            })?;
            drop(drained);
            let right = right.into_child(blocks, parent)?;
            self.entries[idx - 1]
                .child_mut(blocks, parent)?
                .merge(blocks, right)?;
        } else {
            self.entries.remove(idx);
        }
        Ok(Some(prev))
    }

    fn remove_below(
        &mut self,
        blocks: Source<'_>,
        key: &[u8],
        layer: u8,
    ) -> Result<Option<Cid>, InvertError> {
        let Some(idx) = self.find_covering_child(key) else {
            return Ok(None);
        };
        let parent = self.layer;
        let child = self.entries[idx].child_mut(blocks, parent)?;
        let Some(prev) = child.remove(blocks, key, layer)? else {
            return Ok(None);
        };
        self.dirty = true;
        if child.entries.is_empty() && !child.stub {
            self.entries.remove(idx);
        }
        Ok(Some(prev))
    }

    /// Restore the top of the tree to its canonical shape.
    ///
    /// A root that is nothing but a pointer is not a root: the node it points
    /// at is. And an emptied tree has no layer at all, so it must forget the
    /// one it had, or the next insert will build beneath a layer that should
    /// not exist. Indigo and rsky-relay both leave that second case alone, and
    /// both compute the wrong root for it.
    ///
    /// The promoted pointer has to be decoded to know whether it needs
    /// trimming in turn — a removal can collapse more than one layer. If its
    /// block is absent we stop there and keep the pointer: nothing above it
    /// survives to be re-encoded, so its stored CID is already the answer.
    fn trim(&mut self, blocks: Source<'_>) -> Result<(), InvertError> {
        // a node the loop is about to promote away may legitimately be
        // leafless, so track the layer separately rather than parking a
        // placeholder in the node itself
        let mut layer = Some(self.layer);
        while self.entries.len() == 1 && self.entries[0].is_child() {
            let Entry::Child { cid, node } = self.entries.remove(0) else {
                return Err(InvertError::Encode {
                    detail: "entry claimed to be a subtree pointer and was not".into(),
                });
            };
            if let Some(child) = node {
                layer = Some(child.layer);
                *self = child;
                continue;
            }
            let cid = own_cid(cid)?;
            match read(blocks, cid) {
                Ok((entries, found)) => {
                    layer = found;
                    *self = Node {
                        entries,
                        layer: found.unwrap_or_default(),
                        cid: Some(cid),
                        ..Default::default()
                    };
                }
                Err(InvertError::MissingBlock { .. }) => {
                    *self = Node {
                        cid: Some(cid),
                        stub: true,
                        ..Default::default()
                    };
                    return Ok(());
                }
                Err(e) => return Err(e),
            }
        }

        match layer {
            Some(layer) => self.layer = layer,
            // the loop only stops on a node with a leaf, an empty one, or a
            // malformed one with two subtrees side by side
            None if self.entries.is_empty() => self.layer = 0,
            None => {
                return Err(InvertError::MalformedNode {
                    cid: self.cid,
                    detail: NodeFault::NoLeaves,
                });
            }
        }
        if self.entries.is_empty() && !self.stub {
            self.layer = 0;
        }
        Ok(())
    }

    /// Absorb `other`, whose keys all sort after this node's.
    fn merge(&mut self, blocks: Source<'_>, other: Node) -> Result<(), InvertError> {
        let seam = self.entries.len();
        self.dirty = true;
        self.cid = None;
        if seam == 0 {
            self.entries = other.entries;
            return Ok(());
        }
        self.entries.extend(other.entries);
        // two subtree pointers cannot sit next to each other, so if the seam
        // put them there, they merge too
        if seam < self.entries.len()
            && self.entries[seam - 1].is_child()
            && self.entries[seam].is_child()
        {
            let parent = self.layer;
            let right = self.entries.remove(seam).into_child(blocks, parent)?;
            self.entries[seam - 1]
                .child_mut(blocks, parent)?
                .merge(blocks, right)?;
        }
        Ok(())
    }

    /// Break this node into the parts either side of `key`.
    fn split(mut self, blocks: Source<'_>, key: &[u8]) -> Result<(Node, Node), InvertError> {
        if self.entries.is_empty() {
            return Err(InvertError::MalformedNode {
                cid: self.cid,
                detail: NodeFault::Empty,
            });
        }
        let layer = self.layer;
        let (idx, splits) = self.find_slot(blocks, key)?;

        if !splits {
            if idx == 0 || idx >= self.entries.len() {
                return Err(InvertError::Encode {
                    detail: "split point falls outside the node".into(),
                });
            }
            let right = self.entries.split_off(idx);
            return Ok((node_of(self.entries, layer), node_of(right, layer)));
        }

        // the key lands inside a subtree, so that has to split too
        let mut right = self.entries.split_off(idx);
        let child = right.remove(0).into_child(blocks, layer)?;
        let (left_node, right_node) = child.split(blocks, key)?;
        self.entries.push(child_of(left_node));
        right.insert(0, child_of(right_node));
        Ok((node_of(self.entries, layer), node_of(right, layer)))
    }

    /// Index of the leaf holding exactly `key`.
    fn find_leaf(&self, key: &[u8]) -> Option<usize> {
        for (i, entry) in self.entries.iter().enumerate() {
            if let Entry::Leaf { key: leaf, .. } = entry {
                match leaf.as_ref().cmp(key) {
                    Ordering::Equal => return Some(i),
                    Ordering::Greater => return None, // entries are sorted
                    Ordering::Less => {}
                }
            }
        }
        None
    }

    /// Index of the subtree whose key range covers `key`.
    fn find_covering_child(&self, key: &[u8]) -> Option<usize> {
        let mut found = None;
        for (i, entry) in self.entries.iter().enumerate() {
            match entry {
                Entry::Leaf { key: leaf, .. } => {
                    if key <= leaf.as_ref() {
                        break;
                    }
                    found = None;
                }
                Entry::Child { .. } => found = Some(i),
            }
        }
        found
    }

    /// Where a new entry for `key` goes, and whether it lands inside the
    /// subtree already sitting there.
    fn find_slot(&mut self, blocks: Source<'_>, key: &[u8]) -> Result<(usize, bool), InvertError> {
        self.wake(blocks)?;
        let parent = self.layer;
        for idx in 0..self.entries.len() {
            match &self.entries[idx] {
                Entry::Leaf { key: leaf, .. } => {
                    if key < leaf.as_ref() {
                        return Ok((idx, false));
                    }
                }
                Entry::Child { .. } => {
                    // a following leaf below `key` rules this subtree out
                    // without having to look inside it
                    if let Some(Entry::Leaf { key: next, .. }) = self.entries.get(idx + 1)
                        && key > next.as_ref()
                    {
                        continue;
                    }
                    let order = self.entries[idx]
                        .child_mut(blocks, parent)?
                        .covers(blocks, key)?;
                    match order {
                        Ordering::Less => return Ok((idx, false)),
                        Ordering::Equal => return Ok((idx, true)),
                        Ordering::Greater => {}
                    }
                }
            }
        }
        Ok((self.entries.len(), false))
    }

    /// Whether `key` falls below, inside, or above this subtree's key range.
    fn covers(&mut self, blocks: Source<'_>, key: &[u8]) -> Result<Ordering, InvertError> {
        self.wake(blocks)?;
        let parent = self.layer;
        let (Some(first), Some(last)) = (self.entries.first(), self.entries.last()) else {
            return Err(InvertError::MalformedNode {
                cid: self.cid,
                detail: NodeFault::Empty,
            });
        };
        if let Entry::Leaf { key: leaf, .. } = first
            && key < leaf.as_ref()
        {
            return Ok(Ordering::Less);
        }
        if let Entry::Leaf { key: leaf, .. } = last
            && key > leaf.as_ref()
        {
            return Ok(Ordering::Greater);
        }
        for idx in 0..self.entries.len() {
            match &self.entries[idx] {
                Entry::Leaf { key: leaf, .. } => {
                    if key < leaf.as_ref() {
                        return Ok(Ordering::Equal);
                    }
                }
                Entry::Child { .. } => {
                    if let Some(Entry::Leaf { key: next, .. }) = self.entries.get(idx + 1)
                        && key > next.as_ref()
                    {
                        continue;
                    }
                    let last_slot = idx == self.entries.len() - 1;
                    let order = self.entries[idx]
                        .child_mut(blocks, parent)?
                        .covers(blocks, key)?;
                    return Ok(match order {
                        Ordering::Less if idx == 0 => Ordering::Less,
                        Ordering::Greater if last_slot => Ordering::Greater,
                        _ => Ordering::Equal,
                    });
                }
            }
        }
        Ok(Ordering::Equal)
    }
}

fn child_of(node: Node) -> Entry {
    Entry::Child {
        cid: None,
        node: Some(node),
    }
}

fn node_of(entries: Vec<Entry>, layer: u8) -> Node {
    Node {
        entries,
        layer,
        dirty: true,
        ..Default::default()
    }
}

/// Rebuild a node's entries from its wire form, undoing prefix compression.
///
/// A node's layer is the layer of its own keys, so it is not something a
/// producer gets to assert. A node with subtrees but no keys has no layer to
/// derive and is not something a canonical tree contains.
fn read(blocks: Source<'_>, cid: Cid) -> Result<(Vec<Entry>, Option<u8>), InvertError> {
    let block = blocks
        .get(&cid)
        .ok_or(InvertError::MissingBlock { missing: cid })?;
    let data: NodeData = dasl::drisl::from_slice(block).map_err(|e| InvertError::Decode {
        cid,
        detail: e.to_string(),
    })?;
    decode(data, cid)
}

fn decode(data: NodeData, cid: Cid) -> Result<(Vec<Entry>, Option<u8>), InvertError> {
    let fault = |detail| InvertError::MalformedNode {
        cid: Some(cid),
        detail,
    };

    let mut layer = None;
    let mut entries = Vec::with_capacity(data.entries.len() * 2 + 1);
    if let Some(left) = data.left {
        entries.push(Entry::Child {
            cid: Some(left),
            node: None,
        });
    }

    // one reusable buffer for the previous key, since each entry's suffix is
    // relative to it
    let mut prev: Vec<u8> = Vec::new();
    for entry in &data.entries {
        if entry.prefix_len > prev.len() {
            return Err(fault(NodeFault::PrefixOverrun {
                claimed: entry.prefix_len,
                available: prev.len(),
            }));
        }
        // this bound is a memory defence as much as a validity check: each
        // key may extend the previous one, so without it a single crafted
        // block can expand to quadratic size on decode
        let len = entry.prefix_len + entry.key_suffix.len();
        if len > MAX_KEY_LEN {
            return Err(fault(NodeFault::KeyTooLong { len }));
        }
        let mut key = Vec::with_capacity(len);
        key.extend_from_slice(&prev[..entry.prefix_len]);
        key.extend_from_slice(&entry.key_suffix);
        layer.get_or_insert_with(|| layer_of(&key));

        prev.clear();
        prev.extend_from_slice(&key);
        entries.push(Entry::Leaf {
            key: key.into(),
            value: entry.value,
        });

        if let Some(right) = entry.right {
            entries.push(Entry::Child {
                cid: Some(right),
                node: None,
            });
        }
    }

    Ok((entries, layer))
}

/// Serialise a node back to canonical DAG-CBOR.
fn encode(entries: &[Entry]) -> Result<Vec<u8>, InvertError> {
    let mut data = NodeData {
        entries: Vec::with_capacity(entries.len()),
        left: None,
    };
    let mut prev: &[u8] = &[];
    for (idx, entry) in entries.iter().enumerate() {
        match entry {
            Entry::Leaf { key, value } => {
                let prefix_len = shared_prefix_len(key, prev);
                data.entries.push(EntryData {
                    key_suffix: key[prefix_len..].to_vec(),
                    prefix_len,
                    right: None,
                    value: *value,
                });
                prev = key;
            }
            Entry::Child { cid, .. } => {
                if idx == 0 {
                    data.left = *cid;
                    continue;
                }
                let Some(last) = data.entries.last_mut() else {
                    return Err(InvertError::Encode {
                        detail: "subtree pointer with no leaf before it".into(),
                    });
                };
                last.right = *cid;
            }
        }
    }
    dasl::drisl::to_vec(&data).map_err(|e| InvertError::Encode {
        detail: e.to_string(),
    })
}

/// A tree built in memory needs no block source; nothing in it is ever a stub.
#[cfg(test)]
struct NoBlocks;

#[cfg(test)]
impl Blocks for NoBlocks {
    fn get(&self, _: &Cid) -> Option<&[u8]> {
        None
    }
}

#[cfg(test)]
impl Tree<'static> {
    fn empty() -> Self {
        Self {
            root: Node {
                layer: 0,
                dirty: true,
                ..Default::default()
            },
            blocks: &NoBlocks,
        }
    }

    fn insert(&mut self, key: &str, value: Cid) -> Result<Option<Cid>, InvertError> {
        self.root.insert_root(self.blocks, key.as_bytes(), value)
    }

    fn remove(&mut self, key: &str) -> Result<Option<Cid>, InvertError> {
        self.root.remove_root(self.blocks, key.as_bytes())
    }

    /// Every node in the tree, encoded, as inversion would need to read them.
    fn blocks(&mut self) -> Result<std::collections::HashMap<Cid, Vec<u8>>, InvertError> {
        self.root()?;
        let mut blocks = std::collections::HashMap::new();
        collect(&self.root, &mut blocks)?;
        Ok(blocks)
    }
}

#[cfg(test)]
fn collect(
    node: &Node,
    into: &mut std::collections::HashMap<Cid, Vec<u8>>,
) -> Result<(), InvertError> {
    if node.stub {
        return Ok(());
    }
    let block = encode(&node.entries)?;
    into.insert(cid_for(&block), block);
    for entry in &node.entries {
        if let Entry::Child {
            node: Some(child), ..
        } = entry
        {
            collect(child, into)?;
        }
    }
    Ok(())
}

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

    use serde::Deserialize;

    use super::*;
    use crate::cid::Codec;
    use crate::{OpOrder, invert_ops};

    fn cid(seed: &str) -> Cid {
        Cid::digest_sha2(Codec::Drisl, seed.as_bytes())
    }

    fn parse(s: &str) -> Cid {
        s.parse().expect("fixture cid parses")
    }

    /// Commit-proof vectors from bluesky-social/atproto-interop-tests. Each
    /// case names a structural situation someone thought worth pinning down —
    /// "two deep leafless split", "merge and split in multi-op commit".
    #[derive(Deserialize)]
    #[serde(rename_all = "camelCase")]
    struct CommitProof {
        comment: String,
        leaf_value: String,
        keys: Vec<String>,
        adds: Vec<String>,
        dels: Vec<String>,
        root_before_commit: String,
        root_after_commit: String,
        blocks_in_proof: Vec<String>,
    }

    #[test]
    fn interop_commit_proofs() {
        let json = include_str!("../tests/fixtures/atproto-interop/commit-proof-fixtures.json");
        let cases: Vec<CommitProof> = serde_json::from_str(json).expect("fixtures parse");
        assert!(!cases.is_empty());

        for case in &cases {
            let leaf = parse(&case.leaf_value);
            let context = &case.comment;

            let mut before = Tree::empty();
            for key in &case.keys {
                before.insert(key, leaf).expect(context);
            }
            assert_eq!(
                before.root().expect(context),
                parse(&case.root_before_commit),
                "{context}: tree built from keys"
            );

            let mut after = Tree::empty();
            for key in &case.keys {
                after.insert(key, leaf).expect(context);
            }
            let mut ops = Vec::new();
            for key in &case.dels {
                after.remove(key).expect(context);
                ops.push(Op::Delete {
                    key: Key::new(key).expect(context),
                    prev: leaf,
                });
            }
            for key in &case.adds {
                after.insert(key, leaf).expect(context);
                ops.push(Op::Create {
                    key: Key::new(key).expect(context),
                    cid: leaf,
                });
            }
            let root_after = after.root().expect(context);
            assert_eq!(
                root_after,
                parse(&case.root_after_commit),
                "{context}: tree after the commit"
            );

            // inverting from the whole tree must recover the previous root
            let all = after.blocks().expect(context);
            assert_eq!(
                invert_ops(&all, root_after, &ops, OpOrder::AsGiven).expect(context),
                parse(&case.root_before_commit),
                "{context}: inverted from the full tree"
            );

            // and from nothing but the blocks the fixture says a producer
            // sends, which is the claim that actually matters
            let in_proof: HashMap<Cid, Vec<u8>> = case
                .blocks_in_proof
                .iter()
                .map(|s| parse(s))
                .filter_map(|cid| all.get(&cid).map(|block| (cid, block.clone())))
                .collect();
            assert_eq!(
                invert_ops(&in_proof, root_after, &ops, OpOrder::AsGiven).expect(context),
                parse(&case.root_before_commit),
                "{context}: inverted from blocksInProof alone"
            );
        }
    }

    /// Emptying the top layer and then inserting lower leaves a stale layer
    /// behind unless the root is trimmed. Indigo and rsky-relay both get this
    /// wrong; it needs a tree whose top layer holds exactly one key.
    #[test]
    fn emptying_the_top_layer_forgets_its_height() {
        // sha2 puts these at layers 0 and 1 respectively
        let low = "app.bsky.feed.post/3lrbpi5ot35kpjb";
        let high = "app.bsky.feed.post/3lmq4wfloltgcdw";
        assert_eq!(layer_of(low.as_bytes()), 0);
        assert_eq!(layer_of(high.as_bytes()), 1);

        let (old, new) = (cid("old"), cid("new"));

        let mut before = Tree::empty();
        before.insert(low, old).expect("insert low");
        let root_before = before.root().expect("root before");

        let mut after = Tree::empty();
        after.insert(high, new).expect("insert high");
        let root_after = after.root().expect("root after");

        let mut blocks = after.blocks().expect("after blocks");
        blocks.extend(before.blocks().expect("before blocks"));

        let ops = vec![
            Op::Create {
                key: Key::new(high).expect("key"),
                cid: new,
            },
            Op::Delete {
                key: Key::new(low).expect("key"),
                prev: old,
            },
        ];
        assert_eq!(
            invert_ops(&blocks, root_after, &ops, OpOrder::AsGiven).expect("invert"),
            root_before
        );
    }

    #[test]
    fn round_trips_an_emptied_repo() {
        let key = "app.bsky.feed.post/3lrbpi5ot35kpjb";
        let value = cid("only");

        let mut before = Tree::empty();
        before.insert(key, value).expect("insert");
        let root_before = before.root().expect("root");

        let mut after = Tree::empty();
        let root_after = after.root().expect("empty root");

        let mut blocks = after.blocks().expect("after blocks");
        blocks.extend(before.blocks().expect("before blocks"));

        let ops = vec![Op::Delete {
            key: Key::new(key).expect("key"),
            prev: value,
        }];
        assert_eq!(
            invert_ops(&blocks, root_after, &ops, OpOrder::AsGiven).expect("invert"),
            root_before
        );
    }

    /// Counts how many blocks a traversal actually asks for.
    struct Counting<'a> {
        inner: &'a HashMap<Cid, Vec<u8>>,
        reads: std::sync::atomic::AtomicUsize,
    }

    impl Blocks for Counting<'_> {
        fn get(&self, cid: &Cid) -> Option<&[u8]> {
            self.reads
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            self.inner.get(cid).map(Vec::as_slice)
        }
    }

    /// A producer may send its whole tree instead of the minimal proof, and
    /// extra blocks in a diff are legal. Decoding all of them to then walk
    /// eight is work we must not be made to do: one operation should read a
    /// path's worth of nodes, not a repo's worth.
    #[test]
    fn one_op_reads_a_paths_worth_of_nodes() {
        let leaf = cid("leaf");
        let keys: Vec<String> = (0..1000)
            .map(|i| format!("app.bsky.feed.post/3l{i:013}"))
            .collect();

        let mut before = Tree::empty();
        for key in &keys {
            before.insert(key, leaf).expect("insert");
        }
        let root_before = before.root().expect("root");

        let added = "app.bsky.feed.post/3lzzzzzzzzzzzzz";
        let mut after = Tree::empty();
        for key in &keys {
            after.insert(key, leaf).expect("insert");
        }
        after.insert(added, leaf).expect("insert");
        let root_after = after.root().expect("root");

        // the whole tree, not the covering proof
        let blocks = after.blocks().expect("blocks");
        assert!(blocks.len() > 200, "test tree too small: {}", blocks.len());

        let counting = Counting {
            inner: &blocks,
            reads: std::sync::atomic::AtomicUsize::new(0),
        };
        let ops = vec![Op::Create {
            key: Key::new(added).expect("key"),
            cid: leaf,
        }];
        assert_eq!(
            invert_ops(&counting, root_after, &ops, OpOrder::AsGiven).expect("invert"),
            root_before
        );

        let reads = counting.reads.load(std::sync::atomic::Ordering::Relaxed);
        assert!(
            reads < 32,
            "read {reads} blocks out of {} to invert one op",
            blocks.len()
        );
    }

    #[test]
    fn a_commit_that_changed_nothing_needs_no_blocks() {
        let root = cid("some root");
        let blocks: HashMap<Cid, Vec<u8>> = HashMap::new();
        assert_eq!(
            invert_ops(&blocks, root, &[], OpOrder::AsGiven).expect("invert"),
            root
        );
    }

    #[test]
    fn reports_the_block_a_proof_is_missing() {
        let key = "app.bsky.feed.post/3lrbpi5ot35kpjb";
        let value = cid("only");
        let mut tree = Tree::empty();
        tree.insert(key, value).expect("insert");
        let root = tree.root().expect("root");

        let blocks: HashMap<Cid, Vec<u8>> = HashMap::new();
        let ops = vec![Op::Create {
            key: Key::new(key).expect("key"),
            cid: value,
        }];
        match invert_ops(&blocks, root, &ops, OpOrder::AsGiven) {
            Err(InvertError::ProofIncomplete { missing, key: k }) => {
                assert_eq!(missing, root);
                assert_eq!(k.as_bytes(), key.as_bytes());
            }
            other => panic!("expected ProofIncomplete, got {other:?}"),
        }
    }

    #[test]
    fn reports_an_op_the_tree_disagrees_with() {
        let key = "app.bsky.feed.post/3lrbpi5ot35kpjb";
        let mut tree = Tree::empty();
        tree.insert(key, cid("actual")).expect("insert");
        let root = tree.root().expect("root");
        let blocks = tree.blocks().expect("blocks");

        let ops = vec![Op::Create {
            key: Key::new(key).expect("key"),
            cid: cid("claimed"),
        }];
        match invert_ops(&blocks, root, &ops, OpOrder::AsGiven) {
            Err(InvertError::OpMismatch {
                kind,
                expected,
                found,
                ..
            }) => {
                assert_eq!(kind, crate::OpKind::Create);
                assert_eq!(expected, Some(cid("claimed")));
                assert_eq!(found, Some(cid("actual")));
            }
            other => panic!("expected OpMismatch, got {other:?}"),
        }
    }

    /// Deterministic xorshift, so a failure is reproducible from its seed
    /// without pulling in a dependency the crate does not otherwise need.
    struct Rng(u64);

    impl Rng {
        fn next(&mut self) -> u64 {
            self.0 ^= self.0 << 13;
            self.0 ^= self.0 >> 7;
            self.0 ^= self.0 << 17;
            self.0
        }

        fn below(&mut self, n: usize) -> usize {
            (self.next() % n as u64) as usize
        }
    }

    /// mst-test-suite enumerates creates and deletes exhaustively but never
    /// updates, so those need covering here. Random repos, random mixed
    /// commits, invert back to where we started.
    #[test]
    fn round_trips_generated_commits() {
        for seed in 1..300u64 {
            let mut rng = Rng(seed);
            let size = [0usize, 1, 2, 3, 17, 250][seed as usize % 6];

            let mut keys: Vec<String> = (0..size)
                .map(|_| format!("app.bsky.feed.post/3l{:013}", rng.next() % 1_000_000_000))
                .collect();
            keys.sort();
            keys.dedup();

            let value = |n: u64| cid(&format!("record {n}"));
            let mut state: Vec<(String, Cid)> = keys
                .iter()
                .cloned()
                .map(|k| (k, value(rng.next())))
                .collect();

            let mut before = Tree::empty();
            for (key, cid) in &state {
                before.insert(key, *cid).expect("insert");
            }
            let root_before = before.root().expect("root");

            // a commit of 1..=6 ops, each on a distinct key
            let mut ops = Vec::new();
            let mut touched: Vec<String> = Vec::new();
            for _ in 0..1 + rng.below(6) {
                let choice = if state.is_empty() { 0 } else { rng.below(3) };
                match choice {
                    0 => {
                        let key =
                            format!("app.bsky.feed.post/3l{:013}", rng.next() % 1_000_000_000);
                        if touched.contains(&key) || state.iter().any(|(k, _)| *k == key) {
                            continue;
                        }
                        let cid = value(rng.next());
                        touched.push(key.clone());
                        state.push((key.clone(), cid));
                        ops.push(Op::Create {
                            key: Key::new(&key).expect("key"),
                            cid,
                        });
                    }
                    1 => {
                        let idx = rng.below(state.len());
                        let (key, prev) = state[idx].clone();
                        if touched.contains(&key) {
                            continue;
                        }
                        let cid = value(rng.next());
                        touched.push(key.clone());
                        state[idx].1 = cid;
                        ops.push(Op::Update {
                            key: Key::new(&key).expect("key"),
                            cid,
                            prev,
                        });
                    }
                    _ => {
                        let idx = rng.below(state.len());
                        let (key, prev) = state[idx].clone();
                        if touched.contains(&key) {
                            continue;
                        }
                        touched.push(key.clone());
                        state.remove(idx);
                        ops.push(Op::Delete {
                            key: Key::new(&key).expect("key"),
                            prev,
                        });
                    }
                }
            }
            if ops.is_empty() {
                continue;
            }

            let mut after = Tree::empty();
            for (key, cid) in &state {
                after.insert(key, *cid).expect("insert");
            }
            let root_after = after.root().expect("root");

            let mut blocks = after.blocks().expect("after blocks");
            blocks.extend(before.blocks().expect("before blocks"));

            assert_eq!(
                invert_ops(&blocks, root_after, &ops, OpOrder::AsGiven).expect("invert"),
                root_before,
                "seed {seed}\n  after keys: {:?}\n  ops: {:?}",
                state
                    .iter()
                    .map(|(k, _)| (&k[19..], layer_of(k.as_bytes())))
                    .collect::<Vec<_>>(),
                ops.iter()
                    .map(|o| (
                        o.kind(),
                        String::from_utf8_lossy(&o.key().as_bytes()[19..]).to_string(),
                        layer_of(o.key().as_bytes())
                    ))
                    .collect::<Vec<_>>()
            );
        }
    }

    #[test]
    fn op_order_does_not_change_the_recovered_root() {
        let leaf = cid("leaf");
        let keys: Vec<String> = (0..40)
            .map(|i| format!("app.bsky.feed.post/3l{i:013}"))
            .collect();

        let mut before = Tree::empty();
        for key in &keys {
            before.insert(key, leaf).expect("insert");
        }
        let root_before = before.root().expect("root");

        let mut after = Tree::empty();
        for key in &keys {
            after.insert(key, leaf).expect("insert");
        }
        let added: Vec<String> = (100..105)
            .map(|i| format!("app.bsky.feed.post/3l{i:013}"))
            .collect();
        for key in &added {
            after.insert(key, leaf).expect("insert");
        }
        let root_after = after.root().expect("root");
        let blocks = after.blocks().expect("blocks");

        let mut ops: Vec<Op> = added
            .iter()
            .map(|key| Op::Create {
                key: Key::new(key).expect("key"),
                cid: leaf,
            })
            .collect();

        for order in [OpOrder::AsGiven, OpOrder::HighestKeyFirst] {
            assert_eq!(
                invert_ops(&blocks, root_after, &ops, order).expect("invert"),
                root_before,
                "{order:?}"
            );
            ops.reverse();
            assert_eq!(
                invert_ops(&blocks, root_after, &ops, order).expect("invert"),
                root_before,
                "{order:?} reversed"
            );
        }
    }
}