marsdb-graph 0.9.0

//! Property indexes: `(label, property) -> node_ids` keyed by an
//! order-preserving encoding of the property's value. See
//! `marsdb_storage::tables::{INDEX_DEFS, PROPERTY_INDEX}` for the on-disk
//! layout this module builds keys for.

use std::collections::BTreeMap;

use marsdb_storage::{ReadableMultimapTable, ReadableTable, Txn};
use serde::{Deserialize, Serialize};

use crate::error::GraphError;
use crate::labels::lookup_label_id;
use crate::model::{NodeId, PropertyValue};
use crate::props::lookup_prop_id;
use crate::write_ctx::WriteCtx;

#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct IndexDef {
    pub unique: bool,
}

/// `label_id(4 bytes BE) ++ property_id(4 bytes BE)` — the key `INDEX_DEFS`
/// uses, and the fixed prefix every `PROPERTY_INDEX` entry for this
/// (label, property) pair starts with.
fn index_prefix(label_id: u32, prop_id: u32) -> [u8; 8] {
    let mut out = [0u8; 8];
    out[0..4].copy_from_slice(&label_id.to_be_bytes());
    out[4..8].copy_from_slice(&prop_id.to_be_bytes());
    out
}

/// Order-preserving byte encoding of a single `PropertyValue`, for use as
/// a `PROPERTY_INDEX` key suffix. Lexicographic byte comparison matches
/// real value ordering *within one type* (needed for a future range scan,
/// not used yet — MVP only does exact-match lookups) — a leading type tag
/// keeps different types from ever comparing as equal or interleaving.
/// `Duration` has no meaningful total order (see `PropertyValue::Duration`'s
/// own doc comment) — its encoding is only guaranteed consistent for
/// equality, not real ordering, which is fine since nothing orders by it.
pub(crate) fn encode_index_value(v: &PropertyValue) -> Vec<u8> {
    match v {
        PropertyValue::Null => vec![0x00],
        PropertyValue::Bool(b) => vec![0x01, u8::from(*b)],
        // Flip the sign bit so two's-complement ordering becomes correct
        // unsigned big-endian byte ordering (the standard trick: the most
        // negative i64 maps to all-zero bytes, the most positive to
        // all-one bytes).
        PropertyValue::Int(i) => {
            let mut out = vec![0x02];
            out.extend_from_slice(&((*i as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
            out
        }
        // Standard sortable-float transform: flip the sign bit for a
        // non-negative float (so it sorts above all negatives), flip every
        // bit for a negative float (so more-negative sorts lower).
        PropertyValue::Float(f) => {
            let bits = f.to_bits();
            let sortable = if bits & 0x8000_0000_0000_0000 != 0 {
                !bits
            } else {
                bits | 0x8000_0000_0000_0000
            };
            let mut out = vec![0x03];
            out.extend_from_slice(&sortable.to_be_bytes());
            out
        }
        // Raw UTF-8 bytes compare correctly by codepoint for ASCII and
        // "close enough" (not full Unicode collation) in general -- same
        // tradeoff most embedded databases make without pulling in ICU.
        PropertyValue::String(s) => {
            let mut out = vec![0x04];
            out.extend_from_slice(s.as_bytes());
            out
        }
        PropertyValue::Date(days) => {
            let mut out = vec![0x05];
            out.extend_from_slice(&((*days as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
            out
        }
        PropertyValue::Duration {
            months,
            days,
            seconds,
            nanos,
        } => {
            let mut out = vec![0x06];
            out.extend_from_slice(&months.to_be_bytes());
            out.extend_from_slice(&days.to_be_bytes());
            out.extend_from_slice(&seconds.to_be_bytes());
            out.extend_from_slice(&nanos.to_be_bytes());
            out
        }
        // Always non-negative by construction (`0..86_400_000_000_000`) --
        // plain BE bytes already sort correctly, no sign-flip needed.
        PropertyValue::LocalTime(nanos_of_day) => {
            let mut out = vec![0x07];
            out.extend_from_slice(&nanos_of_day.to_be_bytes());
            out
        }
        // Keyed by the UTC-equivalent instant-of-day (`nanos_of_day -
        // offset_seconds`), not the raw wall-clock fields -- matches
        // `Time`'s own equality/ordering rule (see its doc comment), so
        // two structurally-different `Time`s that represent the same
        // instant correctly collapse to the same index key.
        PropertyValue::Time {
            nanos_of_day,
            offset_seconds,
        } => {
            let instant = nanos_of_day - *offset_seconds as i64 * 1_000_000_000;
            let mut out = vec![0x08];
            out.extend_from_slice(&((instant as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
            out
        }
        PropertyValue::LocalDateTime {
            epoch_seconds,
            nanos,
        } => {
            let mut out = vec![0x09];
            out.extend_from_slice(&((*epoch_seconds as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
            out.extend_from_slice(&nanos.to_be_bytes());
            out
        }
        // `offset_seconds` deliberately excluded -- `DateTime`'s equality/
        // ordering is instant-only (see its doc comment), same reasoning
        // as `Time` above.
        PropertyValue::DateTime {
            epoch_seconds,
            nanos,
            ..
        } => {
            let mut out = vec![0x0A];
            out.extend_from_slice(&((*epoch_seconds as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
            out.extend_from_slice(&nanos.to_be_bytes());
            out
        }
        // No real ordering across two lists is defined/needed (same
        // "consistent for equality, not real ordering" carve-out
        // `Duration` above already has) -- MVP indexing only does exact-
        // match lookups. Each element's own encoding is length-prefixed
        // so two different lists can never collide onto the same byte
        // string (e.g. `["ab", "c"]` vs `["a", "bc"]`, which otherwise
        // concatenate to visually-different but genuinely ambiguous byte
        // runs once strings' own raw-UTF-8, non-length-prefixed encoding
        // is stacked back to back).
        PropertyValue::List(items) => {
            let mut out = vec![0x0B];
            for item in items {
                let encoded = encode_index_value(item);
                out.extend_from_slice(&(encoded.len() as u32).to_be_bytes());
                out.extend_from_slice(&encoded);
            }
            out
        }
        // Never reaches here: `Map` is only ever constructed on the
        // parameter-passing path (`PropertyValue`'s own doc comment), and
        // nothing ever stores -- so nothing ever indexes -- a real node/
        // edge property this way.
        PropertyValue::Map(_) => {
            unreachable!("PropertyValue::Map is never a real stored/indexed property value")
        }
    }
}

fn index_key(label_id: u32, prop_id: u32, value: &PropertyValue) -> Vec<u8> {
    let mut out = index_prefix(label_id, prop_id).to_vec();
    out.extend_from_slice(&encode_index_value(value));
    out
}

/// Declares an index on `(label, prop)` and backfills it from every
/// existing node carrying `label`. Errors (without creating the index) if
/// `unique` is requested and two existing nodes already share a value.
/// Idempotent by (label, prop) identity, not by `unique`-ness — calling
/// this again on an already-indexed pair is an error, same as most
/// databases' `CREATE INDEX` (no silent redefinition).
pub(crate) fn create_index(
    ctx: &mut WriteCtx,
    label: &str,
    prop: &str,
    unique: bool,
) -> Result<(), GraphError> {
    let label_id = crate::labels::intern_label(ctx, label)?;
    let prop_id = crate::props::intern_prop(ctx, prop)?;
    let prefix = index_prefix(label_id, prop_id);
    if ctx.index_defs()?.get(prefix.as_slice())?.is_some() {
        return Err(GraphError::CorruptData(format!(
            "index on label {label:?} property {prop:?} already exists"
        )));
    }

    // Backfill: walk every node with this label (via the existing
    // NODE_LABEL_INDEX secondary index, not a full NODES scan) and index
    // whatever value it currently has for `prop` (skipping nodes missing
    // it entirely -- a missing property never appears in the index, same
    // as `IS NULL`/absence being indistinguishable elsewhere in this
    // codebase).
    let node_ids: Vec<u64> = ctx
        .node_label_index()?
        .get(label_id)?
        .map(|entry| entry.map(|value| value.value()).map_err(GraphError::from))
        .collect::<Result<Vec<_>, GraphError>>()?;
    let mut entries: Vec<(Vec<u8>, u64)> = Vec::with_capacity(node_ids.len());
    for node_id in &node_ids {
        let Some(guard) = ctx.nodes()?.get(*node_id)? else {
            continue;
        };
        // Directory-format fast path: `prop_id` is already interned above,
        // so the backfill reads exactly the one indexed property per node —
        // no full-record decode, no name resolution at all.
        if let Some(raw) = crate::encode::node_prop_raw(guard.value(), prop_id)? {
            let value = crate::encode::decode_value(raw)?;
            entries.push((index_key(label_id, prop_id, &value), *node_id));
        }
    }
    if unique {
        let mut seen = std::collections::HashSet::with_capacity(entries.len());
        for (key, _) in &entries {
            if !seen.insert(key.clone()) {
                return Err(GraphError::UniqueConstraintViolation {
                    label: label.to_string(),
                    property: prop.to_string(),
                });
            }
        }
    }

    let encoded = postcard::to_allocvec(&IndexDef { unique })?;
    ctx.index_defs()?
        .insert(prefix.as_slice(), encoded.as_slice())?;
    for (key, node_id) in entries {
        ctx.property_index()?.insert(key.as_slice(), node_id)?;
    }
    Ok(())
}

/// `None` means no index is declared on `(label, prop)`.
pub fn lookup_index_def(txn: Txn, label: &str, prop: &str) -> Result<Option<IndexDef>, GraphError> {
    let Some(label_id) = lookup_label_id(txn, label)? else {
        return Ok(None);
    };
    let Some(prop_id) = lookup_prop_id(txn, prop)? else {
        return Ok(None);
    };
    let prefix = index_prefix(label_id, prop_id);
    let defs = txn.open_table(marsdb_storage::tables::INDEX_DEFS)?;
    let found = defs
        .get(prefix.as_slice())?
        .map(|guard| guard.value().to_vec());
    drop(defs);
    match found {
        Some(bytes) => Ok(Some(postcard::from_bytes(&bytes)?)),
        None => Ok(None),
    }
}

/// Exact-match lookup: every node currently indexed under `(label, prop) =
/// value`, up to `limit` of them if given. Caller (the planner, in a
/// later change) is responsible for checking `lookup_index_def` first —
/// this returns an empty result, not an error, if no such index exists
/// (matching a genuinely-empty index would look the same, and this
/// function has no way to tell those apart itself without the same
/// lookup its caller likely already did). `limit` bounds the underlying
/// multimap iterator itself (`.take(limit)` before collecting, not a
/// truncate after) — same real fix this same class of bug needed for
/// `NODE_LABEL_INDEX` (see `GraphStore::all_nodes_limited_in_txn`'s
/// history): truncating *after* `collect()` would still walk every
/// matching entry first, defeating the point of a `LIMIT` push-down.
/// Range lookup over one indexed `(label, prop)`: every node whose
/// stored value falls inside `[lo, hi]` (each side optional, each
/// independently inclusive/exclusive), in index order. The result is a
/// deliberate SUPERSET for numeric bounds: Cypher compares ints and
/// floats cross-type, and the index stores them in two adjacent
/// type-tagged regions, so a numeric bound scans BOTH regions with the
/// bound converted per region — widened outward where the i64<->f64
/// conversion is lossy (|v| > 2^53), never narrowed. Callers keep the
/// original predicate as a residual filter for exactness; this
/// function's job is to shrink the candidate set from "whole label" to
/// "roughly the range", not to be the final answer. Non-numeric bounds
/// scan their single type region (cross-type comparison is null in
/// Cypher, so same-type is already the complete answer; the residual
/// filter still runs).
pub fn lookup_range(
    txn: Txn,
    label: &str,
    prop: &str,
    lo: Option<(&PropertyValue, bool)>,
    hi: Option<(&PropertyValue, bool)>,
    limit: Option<usize>,
) -> Result<Vec<NodeId>, GraphError> {
    let Some(mut cursor) = IndexRangeCursor::new(txn, label, prop, lo, hi)? else {
        return Ok(Vec::new());
    };
    let mut out = Vec::new();
    loop {
        let want = match limit {
            Some(l) => {
                if out.len() >= l {
                    return Ok(out);
                }
                l - out.len()
            }
            None => usize::MAX,
        };
        let chunk = cursor.next_chunk(txn, want.min(4096))?;
        if chunk.is_empty() {
            return Ok(out);
        }
        out.extend(chunk);
    }
}

/// Resumable cursor over one indexed range — the demand-driven form of
/// `lookup_range`: each `next_chunk` re-seeks past the last `(key,
/// node)` it returned (O(log n) per refill) and pulls at most
/// `chunk_size` more ids, so a `LIMIT`ed consumer that stops early
/// never pays for the rest of the range. Region semantics (numeric
/// superset, widening) are `range_regions`'s — see `lookup_range`.
/// One scan region over full `PROPERTY_INDEX` keys — `(start, end)`.
type KeyRegion = (std::ops::Bound<Vec<u8>>, std::ops::Bound<Vec<u8>>);

pub struct IndexRangeCursor {
    regions: Vec<KeyRegion>,
    region_index: usize,
    /// Resume point within the current region: the last emitted index
    /// key and node id. The next refill scans from `Excluded`-ish this
    /// position — same key's remaining values first, then later keys.
    resume: Option<(Vec<u8>, u64)>,
}

impl IndexRangeCursor {
    /// `None` when the label/prop was never interned (nothing indexed).
    pub fn new(
        txn: Txn,
        label: &str,
        prop: &str,
        lo: Option<(&PropertyValue, bool)>,
        hi: Option<(&PropertyValue, bool)>,
    ) -> Result<Option<Self>, GraphError> {
        let Some(label_id) = lookup_label_id(txn, label)? else {
            return Ok(None);
        };
        let Some(prop_id) = lookup_prop_id(txn, prop)? else {
            return Ok(None);
        };
        let prefix = index_prefix(label_id, prop_id);
        Ok(Some(Self {
            regions: range_regions(&prefix, lo, hi),
            region_index: 0,
            resume: None,
        }))
    }

    pub fn next_chunk(&mut self, txn: Txn, chunk_size: usize) -> Result<Vec<NodeId>, GraphError> {
        use std::ops::Bound;
        if chunk_size == 0 {
            return Ok(Vec::new());
        }
        let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
        let mut out = Vec::new();
        while self.region_index < self.regions.len() {
            let (region_start, region_end) = &self.regions[self.region_index];
            // Resume from just past the last emitted key (inclusive of
            // the key itself -- its remaining values are skipped by the
            // value filter below), else the region's own start.
            let start_owned;
            let start_bound: Bound<&[u8]> = match &self.resume {
                Some((key, _)) => {
                    start_owned = key.clone();
                    Bound::Included(start_owned.as_slice())
                }
                None => match region_start {
                    Bound::Included(k) => Bound::Included(k.as_slice()),
                    Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
                    Bound::Unbounded => Bound::Unbounded,
                },
            };
            let end_bound: Bound<&[u8]> = match region_end {
                Bound::Included(k) => Bound::Included(k.as_slice()),
                Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
                Bound::Unbounded => Bound::Unbounded,
            };
            for entry in index.range::<&[u8]>((start_bound, end_bound))? {
                let (key, values) = entry?;
                let key_bytes = key.value().to_vec();
                let skip_through = match &self.resume {
                    Some((resume_key, resume_val)) if *resume_key == key_bytes => Some(*resume_val),
                    _ => None,
                };
                for value in values {
                    let node = value?.value();
                    if skip_through.is_some_and(|last| node <= last) {
                        continue;
                    }
                    out.push(NodeId(node));
                    self.resume = Some((key_bytes.clone(), node));
                    if out.len() >= chunk_size {
                        return Ok(out);
                    }
                }
            }
            // Region exhausted.
            self.region_index += 1;
            self.resume = None;
        }
        Ok(out)
    }
}

/// The byte-range regions `lookup_range` scans — one per relevant type
/// tag, each a `(start, end)` bound pair over full `PROPERTY_INDEX`
/// keys. See `lookup_range` for the superset/widening contract.
fn range_regions(
    prefix: &[u8],
    lo: Option<(&PropertyValue, bool)>,
    hi: Option<(&PropertyValue, bool)>,
) -> Vec<KeyRegion> {
    use std::ops::Bound;
    let key = |value: &PropertyValue| {
        let mut k = prefix.to_vec();
        k.extend_from_slice(&encode_index_value(value));
        k
    };
    let tag_start = |tag: u8| {
        let mut k = prefix.to_vec();
        k.push(tag);
        Bound::Included(k)
    };
    let tag_end = |tag: u8| {
        let mut k = prefix.to_vec();
        k.push(tag + 1);
        Bound::Excluded(k)
    };
    let numeric = |v: &PropertyValue| matches!(v, PropertyValue::Int(_) | PropertyValue::Float(_));

    let is_numeric = lo.map(|(v, _)| numeric(v)).unwrap_or(true)
        && hi.map(|(v, _)| numeric(v)).unwrap_or(true)
        && (lo.is_some() || hi.is_some())
        && (lo.is_some_and(|(v, _)| numeric(v)) || hi.is_some_and(|(v, _)| numeric(v)));
    if is_numeric {
        // Int region (tag 0x02): a float bound widens outward to the
        // enclosing ints. Float region (tag 0x03): an int bound converts
        // through f64, nudged one ulp outward to cover the lossy range.
        let int_bound = |side_lo: bool, bound: Option<(&PropertyValue, bool)>| match bound {
            None => {
                if side_lo {
                    tag_start(0x02)
                } else {
                    tag_end(0x02)
                }
            }
            Some((PropertyValue::Int(i), inclusive)) => {
                let k = key(&PropertyValue::Int(*i));
                if inclusive {
                    Bound::Included(k)
                } else {
                    Bound::Excluded(k)
                }
            }
            Some((PropertyValue::Float(f), _)) => {
                // Superset: floor for a lower bound, ceil for an upper,
                // both inclusive.
                let widened = if side_lo { f.floor() } else { f.ceil() };
                let clamped = widened.clamp(i64::MIN as f64, i64::MAX as f64) as i64;
                Bound::Included(key(&PropertyValue::Int(clamped)))
            }
            Some(_) => unreachable!("numeric region only built for numeric bounds"),
        };
        let float_bound = |side_lo: bool, bound: Option<(&PropertyValue, bool)>| match bound {
            None => {
                if side_lo {
                    tag_start(0x03)
                } else {
                    tag_end(0x03)
                }
            }
            Some((PropertyValue::Float(f), inclusive)) => {
                let k = key(&PropertyValue::Float(*f));
                if inclusive {
                    Bound::Included(k)
                } else {
                    Bound::Excluded(k)
                }
            }
            Some((PropertyValue::Int(i), _)) => {
                // Superset: widen outward by a couple of ulps (relative
                // epsilon) to cover |i| > 2^53 conversion lossiness --
                // overshooting is harmless, the residual filter is
                // exact. (`f64::next_down`/`next_up` say this directly
                // but are stable only since 1.86; MSRV is 1.82.)
                let f = *i as f64;
                let step = f.abs() * (2.0 * f64::EPSILON) + f64::MIN_POSITIVE;
                let widened = if side_lo { f - step } else { f + step };
                Bound::Included(key(&PropertyValue::Float(widened)))
            }
            Some(_) => unreachable!("numeric region only built for numeric bounds"),
        };
        return vec![
            (int_bound(true, lo), int_bound(false, hi)),
            (float_bound(true, lo), float_bound(false, hi)),
        ];
    }

    // Non-numeric: one region, the type tag of whichever bound exists
    // (both same-type when both exist -- a mixed-type non-numeric range
    // matches nothing in Cypher, and the residual filter enforces that;
    // scanning the lo-side region is a harmless superset).
    let tag = lo
        .or(hi)
        .map(|(v, _)| encode_index_value(v)[0])
        .unwrap_or(0x00);
    let start = match lo {
        None => tag_start(tag),
        Some((v, true)) => Bound::Included(key(v)),
        Some((v, false)) => Bound::Excluded(key(v)),
    };
    let end = match hi {
        None => tag_end(tag),
        Some((v, true)) => Bound::Included(key(v)),
        Some((v, false)) => Bound::Excluded(key(v)),
    };
    vec![(start, end)]
}

pub fn lookup_exact(
    txn: Txn,
    label: &str,
    prop: &str,
    value: &PropertyValue,
    limit: Option<usize>,
) -> Result<Vec<NodeId>, GraphError> {
    let Some(label_id) = lookup_label_id(txn, label)? else {
        return Ok(Vec::new());
    };
    let Some(prop_id) = lookup_prop_id(txn, prop)? else {
        return Ok(Vec::new());
    };
    let key = index_key(label_id, prop_id, value);
    let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
    let iter = index.get(key.as_slice())?;
    let ids: Vec<NodeId> = match limit {
        Some(limit) => iter
            .take(limit)
            .map(|entry| {
                entry
                    .map(|value| NodeId(value.value()))
                    .map_err(GraphError::from)
            })
            .collect::<Result<Vec<_>, GraphError>>()?,
        None => iter
            .map(|entry| {
                entry
                    .map(|value| NodeId(value.value()))
                    .map_err(GraphError::from)
            })
            .collect::<Result<Vec<_>, GraphError>>()?,
    };
    drop(index);
    Ok(ids)
}

/// Cheap, exact cardinality of `(label, prop) = value` under a declared
/// index — the stat the query planner uses to pick the most selective
/// candidate when several indexed equality conjuncts are available for the
/// same scan (see `marsdb_query::planner::apply_index_seeks`). O(1): redb's
/// `MultimapValue::len()` reports a count it already tracks per key, so
/// this never walks the matching entries themselves, unlike `lookup_exact`.
/// Returns 0 if no such index/value exists (same "caller already checked
/// `lookup_index_def`" contract as `lookup_exact`).
pub fn match_count(
    txn: Txn,
    label: &str,
    prop: &str,
    value: &PropertyValue,
) -> Result<u64, GraphError> {
    let Some(label_id) = lookup_label_id(txn, label)? else {
        return Ok(0);
    };
    let Some(prop_id) = lookup_prop_id(txn, prop)? else {
        return Ok(0);
    };
    let key = index_key(label_id, prop_id, value);
    let index = txn.open_multimap_table(marsdb_storage::tables::PROPERTY_INDEX)?;
    let count = index.get(key.as_slice())?.len();
    Ok(count)
}

/// Every declared index whose label is in `label_ids`, as `(label_id,
/// prop_id, prop_name, IndexDef)`. `INDEX_DEFS` is scanned in full (not a
/// prefix-range query — `TableHandle` only exposes `get`/`iter`, and the
/// number of *declared indexes* is expected to be small, unlike node
/// counts) and filtered in memory.
fn indexes_for_labels(
    ctx: &mut WriteCtx,
    label_ids: &[u32],
) -> Result<Vec<(u32, u32, String, IndexDef)>, GraphError> {
    // Collected into an owned Vec first, not resolved inline in the loop
    // below -- `ctx.index_defs()?.iter()?` holds `ctx` mutably borrowed for
    // the iterator's whole lifetime, and `resolve_prop_ctx` below needs its
    // own fresh `&mut ctx` (to lazily open `id_to_prop`), which can't
    // coexist with that borrow.
    let raw: Vec<(u32, u32, IndexDef)> = {
        let mut raw = Vec::new();
        for entry in ctx.index_defs()?.iter()? {
            let (key, value) = entry?;
            let key_bytes = key.value();
            let label_id = u32::from_be_bytes(
                key_bytes[0..4]
                    .try_into()
                    .expect("index key prefix is 8 bytes"),
            );
            if !label_ids.contains(&label_id) {
                continue;
            }
            let prop_id = u32::from_be_bytes(
                key_bytes[4..8]
                    .try_into()
                    .expect("index key prefix is 8 bytes"),
            );
            let def: IndexDef = postcard::from_bytes(value.value())?;
            raw.push((label_id, prop_id, def));
        }
        raw
    };
    raw.into_iter()
        .map(|(label_id, prop_id, def)| {
            let prop_name = resolve_prop_ctx(ctx, prop_id)?;
            Ok((label_id, prop_id, prop_name, def))
        })
        .collect()
}

/// `labels::resolve_label`/`props::resolve_prop` equivalents reading
/// directly from an already-open `WriteCtx` handle, instead of opening
/// `ID_TO_LABEL`/`ID_TO_PROP` again via `Txn` (which `WriteCtx` already
/// holds open -- a second live handle to the same table would be
/// `TableAlreadyOpen`). Small deliberate duplication, not a shared helper
/// with the `Txn`-based versions -- those stay untouched for the read
/// path (see `WriteCtx`'s own docs).
pub(crate) fn resolve_label_ctx(ctx: &mut WriteCtx, label_id: u32) -> Result<String, GraphError> {
    let value = ctx.id_to_label()?.get(label_id)?.ok_or_else(|| {
        GraphError::CorruptData(format!("label id {label_id} has no interned string"))
    })?;
    Ok(value.value().to_string())
}

pub(crate) fn resolve_prop_ctx(ctx: &mut WriteCtx, prop_id: u32) -> Result<String, GraphError> {
    let value = ctx.id_to_prop()?.get(prop_id)?.ok_or_else(|| {
        GraphError::CorruptData(format!("prop id {prop_id} has no interned string"))
    })?;
    Ok(value.value().to_string())
}

/// Identifies one declared index, both by id (for the actual key/lookup)
/// and by name (only needed for a `UniqueConstraintViolation`'s message).
/// Bundled into one struct so `insert_entry` doesn't take 8 separate
/// arguments (clippy's `too_many_arguments`, capped at 7).
struct IndexTarget<'a> {
    label_id: u32,
    prop_id: u32,
    label: &'a str,
    prop: &'a str,
}

fn insert_entry(
    ctx: &mut WriteCtx,
    target: &IndexTarget<'_>,
    value: &PropertyValue,
    node_id: u64,
    unique: bool,
) -> Result<(), GraphError> {
    let key = index_key(target.label_id, target.prop_id, value);
    if unique && ctx.property_index()?.get(key.as_slice())?.next().is_some() {
        return Err(GraphError::UniqueConstraintViolation {
            label: target.label.to_string(),
            property: target.prop.to_string(),
        });
    }
    ctx.property_index()?.insert(key.as_slice(), node_id)?;
    Ok(())
}

fn remove_entry(
    ctx: &mut WriteCtx,
    label_id: u32,
    prop_id: u32,
    value: &PropertyValue,
    node_id: u64,
) -> Result<(), GraphError> {
    let key = index_key(label_id, prop_id, value);
    ctx.property_index()?.remove(key.as_slice(), node_id)?;
    Ok(())
}

/// Inserts index entries for `node_id` into every declared index whose
/// label is in `label_ids` and whose property `props` has a value for.
/// Called on node creation (`label_ids` = every label the node was just
/// given) and on `SET n:Label` (`label_ids` = just the one newly-added
/// label — indexes on labels the node already had are untouched, since
/// nothing about their entries changed).
pub(crate) fn on_node_created(
    ctx: &mut WriteCtx,
    node_id: u64,
    label_ids: &[u32],
    props: &BTreeMap<String, PropertyValue>,
) -> Result<(), GraphError> {
    for (label_id, prop_id, prop_name, def) in indexes_for_labels(ctx, label_ids)? {
        if let Some(value) = props.get(&prop_name) {
            let label = resolve_label_ctx(ctx, label_id)?;
            let target = IndexTarget {
                label_id,
                prop_id,
                label: &label,
                prop: &prop_name,
            };
            insert_entry(ctx, &target, value, node_id, def.unique)?;
        }
    }
    Ok(())
}

/// Removes `node_id`'s index entries from every declared index whose label
/// is in `label_ids` and whose property `props` (the values *before* this
/// change) has a value for. Called on node deletion (`label_ids` = every
/// label the node had) and on `REMOVE n:Label` (`label_ids` = just the one
/// removed label).
pub(crate) fn on_node_deleted(
    ctx: &mut WriteCtx,
    node_id: u64,
    label_ids: &[u32],
    props: &BTreeMap<String, PropertyValue>,
) -> Result<(), GraphError> {
    for (label_id, prop_id, prop_name, _def) in indexes_for_labels(ctx, label_ids)? {
        if let Some(value) = props.get(&prop_name) {
            remove_entry(ctx, label_id, prop_id, value, node_id)?;
        }
    }
    Ok(())
}

/// One property's value changed on an existing node (`SET n.prop = ..`/
/// `REMOVE n.prop`) — removes the old index entry (if `old_value` is
/// `Some` and an index covers `(label, prop)` for one of `label_ids`) and
/// inserts the new one (if `new_value` is `Some`). `new_value: None`
/// means the property was removed entirely, not set to `null` — a
/// `PropertyValue::Null` value is still `Some(&PropertyValue::Null)` here
/// and gets indexed like any other value (matches `create_index`'s own
/// backfill, which only skips a property that's *absent*, not one whose
/// value is `Null`).
pub(crate) fn on_node_prop_changed(
    ctx: &mut WriteCtx,
    node_id: u64,
    label_ids: &[u32],
    prop: &str,
    old_value: Option<&PropertyValue>,
    new_value: Option<&PropertyValue>,
) -> Result<(), GraphError> {
    for (label_id, prop_id, prop_name, def) in indexes_for_labels(ctx, label_ids)? {
        if prop_name != prop {
            continue;
        }
        if let Some(old) = old_value {
            remove_entry(ctx, label_id, prop_id, old, node_id)?;
        }
        if let Some(new) = new_value {
            let label = resolve_label_ctx(ctx, label_id)?;
            let target = IndexTarget {
                label_id,
                prop_id,
                label: &label,
                prop: &prop_name,
            };
            insert_entry(ctx, &target, new, node_id, def.unique)?;
        }
    }
    Ok(())
}