haematite 0.7.0

//! Batch mutation of a prolly tree under an explicit [`TreePolicy`].
//!
//! Every entry point takes the policy by value (CHUNKING-POLICY.md §4.1): the
//! ambient `active_detector()` thread-local the pre-v2 code used is DELETED, so
//! a mutation with no policy simply does not type-check. Production callers read
//! the policy from the stamped config; the raw public APIs accept it as an
//! argument.
//!
//! The work splits across two submodules: [`window`] locates and rebuilds the
//! affected leaf window (the both-edges sparse-window rule under v2), and
//! [`spine`] re-groups the leaf refs into the internal spine.

mod spine;
mod window;

use crate::store::NodeStore;

use super::cursor::TreeError;
use super::node::Hash;
use super::policy::TreePolicy;
use super::source_error::{ErrorCapturingStore, TreeOperationError};

/// A single leaf entry: sorted key and its inline value.
pub(super) type Entry = (Vec<u8>, Vec<u8>);

/// A single internal child: the child's separator key (its subtree's first key)
/// and the child node's content hash.
pub(super) type ChildRef = (Vec<u8>, Hash);

#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct Mutation {
    pub(super) key: Vec<u8>,
    pub(super) value: Option<Vec<u8>>,
}

/// Insert or overwrite `key` with `value` under `policy`, returning the new root.
pub fn insert<S, K, V>(
    store: &mut S,
    root_hash: Hash,
    key: K,
    value: V,
    policy: TreePolicy,
) -> Result<Hash, TreeError>
where
    S: NodeStore + ?Sized,
    K: AsRef<[u8]>,
    V: AsRef<[u8]>,
{
    let mutations = vec![(key.as_ref().to_vec(), Some(value.as_ref().to_vec()))];
    batch_mutate_owned(store, root_hash, mutations, policy)
}

/// Delete `key` under `policy`, returning the new root.
pub fn delete<S, K>(
    store: &mut S,
    root_hash: Hash,
    key: K,
    policy: TreePolicy,
) -> Result<Hash, TreeError>
where
    S: NodeStore + ?Sized,
    K: AsRef<[u8]>,
{
    let mutations = vec![(key.as_ref().to_vec(), None)];
    batch_mutate_owned(store, root_hash, mutations, policy)
}

/// Apply a batch of put/delete mutations under `policy` and return the new root.
pub fn batch_mutate<S: NodeStore + ?Sized>(
    store: &mut S,
    root_hash: Hash,
    mutations: &[(Vec<u8>, Option<Vec<u8>>)],
    policy: TreePolicy,
) -> Result<Hash, TreeError> {
    batch_mutate_owned(store, root_hash, mutations.to_vec(), policy)
}

/// Owned-input twin of [`batch_mutate`] for the durable-commit path.
pub fn batch_mutate_owned<S: NodeStore + ?Sized>(
    store: &mut S,
    root_hash: Hash,
    mutations: Vec<(Vec<u8>, Option<Vec<u8>>)>,
    policy: TreePolicy,
) -> Result<Hash, TreeError> {
    batch_mutate_owned_with_source(store, root_hash, mutations, policy)
        .map_err(TreeOperationError::into_tree_error)
}

/// Source-preserving twin used by the durable commit callers (branch/browser),
/// which need the underlying store error rather than the erased [`TreeError`].
pub(crate) fn batch_mutate_owned_with_source<S: NodeStore + ?Sized>(
    store: &mut S,
    root_hash: Hash,
    mutations: Vec<(Vec<u8>, Option<Vec<u8>>)>,
    policy: TreePolicy,
) -> Result<Hash, TreeOperationError<S::Error>> {
    let mut capturing = ErrorCapturingStore::new(store);
    let result = batch_mutate_owned_erased(&mut capturing, root_hash, mutations, policy);
    capturing.finish(result)
}

fn batch_mutate_owned_erased<S: NodeStore + ?Sized>(
    store: &mut S,
    root_hash: Hash,
    mutations: Vec<(Vec<u8>, Option<Vec<u8>>)>,
    policy: TreePolicy,
) -> Result<Hash, TreeError> {
    if mutations.is_empty() {
        return Ok(root_hash);
    }

    let normalised = normalise_mutations(mutations);
    if normalised.is_empty() {
        return Ok(root_hash);
    }

    // Collect every existing leaf in key order (cheap: only leaf *refs* are read,
    // not their contents, except the leaves the mutation actually touches).
    let leaves = spine::collect_leaf_refs(store, root_hash)?;

    let window = window::affected_window(leaves.as_slice(), normalised.as_slice())?;
    let Some(window) = window else {
        // No leaf is affected and no insert occurs (e.g. deleting absent keys).
        return Ok(root_hash);
    };

    let rebuilt = window::rebuild_window(store, &leaves, window, normalised.as_slice(), policy)?;

    spine::finish_root(store, rebuilt, policy)
}

fn normalise_mutations(mutations: Vec<(Vec<u8>, Option<Vec<u8>>)>) -> Vec<Mutation> {
    let mut normalised: Vec<Mutation> = Vec::new();

    for (key, value) in mutations {
        if let Some(last) = normalised.last_mut().filter(|last| last.key == key) {
            last.value = value;
        } else {
            normalised.push(Mutation { key, value });
        }
    }

    normalised
}