froe 0.12.0

//! The counter index, rebuilt from Oak's own hash chain.
//!
//! This is the one rebuild whose output cannot be checked against content.
//! A property index's entries can be compared with the nodes they name; a
//! counter's `:cnt` values are a function of a `SipHash` chain, a seed and a
//! resolution, and a rebuild that got the chain subtly wrong would produce a
//! plausible map that disagreed with Oak's next cycle in a way that only
//! grows. So it is pinned to a vector generated by Oak's own editor —
//! `tests/fixtures/oak-counter-index-vectors.tsv`.
//!
//! # The rule
//!
//! For each child, one chain step from the parent's hash, folding in the
//! Java string hash code of the child name. When that step's folded 32-bit
//! hash code masked by `bit_mask` is zero, `bit_mask + 1` is added to the
//! **parent** and to every ancestor — never to the hitting node itself.
//! `bit_mask` is the highest set bit of `resolution`, doubled, less one.
//! Hidden children are never visited.
//!
//! # The seed, and a deliberate deviation
//!
//! An existing `seed` is read converting to `LONG` and then **narrowed to
//! its low 32 bits sign-extended**, exactly as the counter's editor provider
//! reads it. The interop fixture's counter carries a 64-bit seed, so a
//! builder that skipped the narrowing would place hits at different paths
//! than the store already holds.
//!
//! When a definition has no seed, one is created — and here froe
//! deliberately differs from Oak. Oak's provider draws the most significant
//! 64 bits of a random UUID and uses them **untruncated** on the run that
//! creates the seed, while narrowing them on every later run, so the first
//! cycle and every later one disagree about where hits fall. froe draws a
//! value that already fits `i32` sign-extended, so both readings agree. That
//! is strictly safer than reproducing the quirk, and it is a deviation
//! rather than an oversight.

use std::collections::{BTreeMap, HashSet};

use crate::PropertyType;
use crate::content::node::NodeState;
use crate::error::{Error, Result};
use crate::index::counter::sip_hash::SipHash;
use crate::index::definition::IndexDefinition;
use crate::segment::record::RecordIdentifier;
use crate::writer::record_writer::{
    ChildNodesToWrite, PropertyToWrite, PropertyValuesToWrite, RecordWriter, SegmentSink,
};

/// The property a counter node carries.
const COUNT_PROPERTY_NAME: &str = ":cnt";

/// `resolution`'s default, when a definition names none.
const DEFAULT_RESOLUTION: i64 = 1000;

/// What the counting walk found, for the plan.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub struct HitCount {
    /// Nodes whose hash hit the mask.
    pub hits: u64,
    /// Nodes visited.
    pub nodes_visited: u64,
    /// Nodes that will carry a `:cnt`.
    pub credited_nodes: usize,
}

/// What a build produced.
pub struct BuiltCounter {
    /// The `:index` node, or `None` when nothing hit.
    ///
    /// `None` is a real outcome, not an empty tree: Oak's editor returns
    /// before creating `:index` when no node hits, so its own reindex of a
    /// small store leaves the counter definition without one — and that is
    /// the shape the interop oracle compares against.
    pub index_record: Option<RecordIdentifier>,
    /// The seed this run created, when the definition carried none.
    pub created_seed: Option<i64>,
    /// What each node was credited, by absolute path.
    ///
    /// A node's own hit credits its *parent*, never itself, so the
    /// difference between a node's `:cnt` and the sum of its children's is
    /// exactly `(its own hitting children) × (bit_mask + 1)` — and no walk
    /// of the written index can recover that number. The map therefore
    /// outlives the build, so the tail's verification can compare against
    /// it. Bounded by hits × depth.
    pub credited_by_path: BTreeMap<String, i64>,
}

/// Builds the counter's `:index` from a state root.
pub struct CounterBuilder {
    seed: i64,
    created_seed: Option<i64>,
    bit_mask: i32,
}

impl CounterBuilder {
    /// A builder for `definition`, reading or creating its seed.
    #[must_use]
    pub fn new(definition: &IndexDefinition) -> Self {
        let resolution = definition.resolution.unwrap_or(DEFAULT_RESOLUTION);
        let bit_mask = bit_mask_for(resolution);
        // Every run after the one that created it narrows to 32 bits and
        // sign-extends, so a rebuild must too.
        if let Some(stored) = definition.seed {
            return Self {
                seed: i64::from(stored as i32),
                created_seed: None,
                bit_mask,
            };
        }
        let drawn = drawn_seed();
        Self {
            seed: drawn,
            created_seed: Some(drawn),
            bit_mask,
        }
    }

    /// The seed in use, narrowed as the editor reads it.
    #[must_use]
    pub fn seed(&self) -> i64 {
        self.seed
    }

    /// The mask a hit is tested against.
    #[must_use]
    pub fn bit_mask(&self) -> i32 {
        self.bit_mask
    }

    /// Walks without writing, for the plan.
    pub fn count_hits(&self, state_root: &NodeState<'_>) -> Result<HitCount> {
        let credited = self.accumulate(state_root)?;
        Ok(HitCount {
            hits: credited.hits,
            nodes_visited: credited.nodes_visited,
            credited_nodes: credited.by_path.len(),
        })
    }

    /// Walks, accumulates and writes the `:index` subtree.
    pub fn build<Sink: SegmentSink>(
        &self,
        state_root: &NodeState<'_>,
        writer: &mut RecordWriter<Sink>,
    ) -> Result<BuiltCounter> {
        let accumulated = self.accumulate(state_root)?;
        if accumulated.by_path.is_empty() {
            // Oak's editor returns before creating `:index`, so no hidden
            // child is written at all.
            return Ok(BuiltCounter {
                index_record: None,
                created_seed: self.created_seed,
                credited_by_path: BTreeMap::new(),
            });
        }
        let index_record = write_counter_tree(writer, &accumulated.by_path)?;
        Ok(BuiltCounter {
            index_record: Some(index_record),
            created_seed: self.created_seed,
            credited_by_path: accumulated.by_path,
        })
    }

    /// The accumulation: every ancestor of a hit gains `bit_mask + 1`.
    fn accumulate(&self, state_root: &NodeState<'_>) -> Result<Accumulated> {
        let increment = i64::from(self.bit_mask) + 1;
        let mut accumulated = Accumulated::default();

        let mut stack = vec![CounterStep::Visit {
            node: *state_root,
            path: String::new(),
            hash: SipHash::seeded(self.seed),
        }];
        let mut ancestors: HashSet<RecordIdentifier> = HashSet::new();

        while let Some(step) = stack.pop() {
            match step {
                CounterStep::Leave { record } => {
                    ancestors.remove(&record);
                }
                CounterStep::Visit { node, path, hash } => {
                    let record = node.record_identifier();
                    if !ancestors.insert(record) {
                        return Err(Error::InvalidFormat {
                            details: format!(
                                "the node at {} is its own ancestor, so the counter cannot \
                                 be rebuilt from it",
                                if path.is_empty() { "/" } else { &path }
                            ),
                        });
                    }
                    stack.push(CounterStep::Leave { record });
                    accumulated.nodes_visited += 1;

                    let mut entries = node.child_node_entries()?;
                    entries.sort_by(|left, right| left.0.as_bytes().cmp(right.0.as_bytes()));
                    for (name, child) in entries.into_iter().rev() {
                        // Hidden children are never visited.
                        if name.starts_with(':') {
                            continue;
                        }
                        let child_hash = hash.for_child(&name);
                        let child_path = format!("{path}/{name}");
                        if child_hash.hash_code() & self.bit_mask == 0 {
                            accumulated.hits += 1;
                            // The hit credits the parent and every ancestor,
                            // never the hitting node itself.
                            accumulated.credit(&path, increment);
                        }
                        stack.push(CounterStep::Visit {
                            node: child,
                            path: child_path,
                            hash: child_hash,
                        });
                    }
                }
            }
        }
        Ok(accumulated)
    }
}

/// One step of the counting walk: a node to visit with the hash chained
/// down to it, or an ancestor to release.
///
/// The hash is carried rather than re-derived, so a path's chain is computed
/// once for the whole subtree below it.
enum CounterStep<'provider> {
    Visit {
        node: NodeState<'provider>,
        path: String,
        hash: SipHash,
    },
    Leave {
        record: RecordIdentifier,
    },
}

#[derive(Default)]
struct Accumulated {
    by_path: BTreeMap<String, i64>,
    hits: u64,
    nodes_visited: u64,
}

impl Accumulated {
    /// Credits `path` and every ancestor of it, the root included.
    fn credit(&mut self, path: &str, increment: i64) {
        let mut current = path;
        loop {
            let key = if current.is_empty() { "/" } else { current };
            *self.by_path.entry(key.to_owned()).or_insert(0) += increment;
            if current.is_empty() {
                break;
            }
            current = match current.rfind('/') {
                Some(position) => &current[..position],
                None => "",
            };
        }
    }
}

/// Writes the `:index` subtree: one node per credited path, `:cnt` on each.
///
/// Bottom-up, because `write_node` needs every child's record first. The
/// paths arrive sorted, so a node's descendants are contiguous after it.
fn write_counter_tree<Sink: SegmentSink>(
    writer: &mut RecordWriter<Sink>,
    credited: &BTreeMap<String, i64>,
) -> Result<RecordIdentifier> {
    // Deepest first, so every child is written before its parent.
    //
    // The depth is the count of *non-empty* elements, not of slashes: `/`
    // and `/n0` both hold one slash, and sorting by that put the root at the
    // same level as its children — where a stable sort could write it first
    // and silently drop every child it had not collected yet.
    let mut paths: Vec<&String> = credited.keys().collect();
    paths.sort_by_key(|path| std::cmp::Reverse(path_depth(path)));

    let mut written: BTreeMap<String, RecordIdentifier> = BTreeMap::new();
    let mut children_of: BTreeMap<String, Vec<(String, RecordIdentifier)>> = BTreeMap::new();

    for path in paths {
        let count = credited.get(path).copied().unwrap_or(0);
        let value = writer.write_string(&count.to_string())?;
        let mut children = children_of.remove(path.as_str()).unwrap_or_default();
        children.sort_by(|left, right| left.0.as_bytes().cmp(right.0.as_bytes()));
        let record = writer.write_node(
            None,
            &[],
            &match children.as_slice() {
                [] => ChildNodesToWrite::Zero,
                [(name, node)] => ChildNodesToWrite::One {
                    name: name.clone(),
                    node: *node,
                },
                many => ChildNodesToWrite::Many(many.to_vec()),
            },
            &[PropertyToWrite {
                name: COUNT_PROPERTY_NAME.to_owned(),
                property_type: PropertyType::Long,
                values: PropertyValuesToWrite::Single(value),
            }],
        )?;
        written.insert(path.clone(), record);

        if path != "/" {
            let (parent, name) = split_parent(path);
            children_of.entry(parent).or_default().push((name, record));
        }
    }

    written
        .get("/")
        .copied()
        .ok_or_else(|| Error::InvalidFormat {
            details: "the counter accumulation credited no root, which cannot happen when any \
                  node hit"
                .to_owned(),
        })
}

/// How many elements a path has. `/` is zero, `/a` is one.
fn path_depth(path: &str) -> usize {
    path.split('/')
        .filter(|element| !element.is_empty())
        .count()
}

/// A path's parent and last element. `/a` yields `("/", "a")`.
fn split_parent(path: &str) -> (String, String) {
    match path.rfind('/') {
        Some(0) => ("/".to_owned(), path[1..].to_owned()),
        Some(position) => (path[..position].to_owned(), path[position + 1..].to_owned()),
        None => ("/".to_owned(), path.to_owned()),
    }
}

/// `highestOneBit(resolution) * 2 - 1`.
fn bit_mask_for(resolution: i64) -> i32 {
    let resolution = i32::try_from(resolution).unwrap_or(i32::MAX);
    if resolution <= 0 {
        return 0;
    }
    let highest = 1i32 << (31 - resolution.leading_zeros());
    highest.wrapping_mul(2).wrapping_sub(1)
}

/// A seed that already fits `i32` sign-extended, so the run that creates it
/// and every later run agree about where hits fall. See the module
/// documentation for why this differs from Oak's draw.
fn drawn_seed() -> i64 {
    i64::from(crate::writer::identifier_generator::random_u32() as i32)
}