selene-db-graph 1.3.0

//! Immutable graph snapshot and read accessors.

use std::borrow::Cow;
use std::ops::RangeBounds;
use std::sync::Arc;

use imbl::HashMap;
use roaring::RoaringBitmap;
use rustc_hash::FxHashMap;
use serde::{Deserialize, Serialize};
use smallvec::SmallVec;

use selene_core::{DbString, EdgeId, GraphId, LabelSet, NodeId, PropertyMap, Value};

use crate::adjacency::AdjacencyEntry;
use crate::composite_typed_index::CompositeTypedIndex;
use crate::graph_types::GraphTypeDef;
use crate::id_map::{EngineIdMap, engine_id_map};
use crate::store::{EdgeStore, NodeStore, RowIndex};
use crate::text_index::TextIndex;
use crate::typed_index::{TypedIndex, TypedIndexKind};
use crate::vector_index::VectorIndex;

mod index_entries;

pub use index_entries::{
    CompositePropertyIndexEntry, CompositePropertyIndexEntryRow, PropertyIndexEntry,
    TextIndexEntry, TextIndexEntryRow, VectorIndexEntry, VectorIndexEntryRow,
};

/// Snapshot metadata.
#[derive(
    Clone,
    Debug,
    Deserialize,
    PartialEq,
    rkyv::Archive,
    rkyv::Deserialize,
    rkyv::Serialize,
    Serialize,
)]
pub struct GraphMeta {
    /// Graph identifier.
    pub graph_id: GraphId,
    /// Published generation counter.
    pub generation: u64,
    /// Next node ID to allocate.
    pub next_node_id: u64,
    /// Next edge ID to allocate.
    pub next_edge_id: u64,
    /// Bound closed graph type. `None` means GG01/open graph.
    pub bound_type: Option<Arc<GraphTypeDef>>,
}

/// Immutable graph snapshot.
#[derive(Clone, Debug)]
pub struct SeleneGraph {
    /// Snapshot metadata.
    pub meta: GraphMeta,
    /// Node storage.
    pub node_store: NodeStore,
    /// Edge storage.
    pub edge_store: EdgeStore,
    /// Outgoing adjacency keyed by source node.
    pub adjacency_out: EngineIdMap<NodeId, AdjacencyEntry>,
    /// Incoming adjacency keyed by target node.
    pub adjacency_in: EngineIdMap<NodeId, AdjacencyEntry>,
    /// Bitmap of node rows carrying each label.
    pub idx_label: HashMap<DbString, RoaringBitmap>,
    /// Bitmap of edge rows carrying each edge label.
    pub idx_edge_label: HashMap<DbString, RoaringBitmap>,
    /// Per-`(label, property)` node value indexes. See spec 03 section 5.2.
    pub property_index: FxHashMap<(DbString, DbString), PropertyIndexEntry>,
    /// Per-`(edge label, property)` edge value indexes.
    pub edge_property_index: FxHashMap<(DbString, DbString), PropertyIndexEntry>,
    /// Per-`(label, properties...)` node composite value indexes.
    pub composite_property_index:
        FxHashMap<(DbString, SmallVec<[DbString; 4]>), CompositePropertyIndexEntry>,
    /// Per-`(label, property)` node vector indexes.
    pub vector_index: FxHashMap<(DbString, DbString), VectorIndexEntry>,
    /// Per-`(label, property)` node BM25 text indexes.
    pub text_index: FxHashMap<(DbString, DbString), TextIndexEntry>,
    /// External `NodeId -> RowIndex` lookup (the inverse of
    /// [`NodeStore::row_to_id`]). Replaces the `id.get() - 1` arithmetic so the
    /// external id can stay stable while the row is remapped by compaction
    /// (D22 / BRIEF-Item-4a). `imbl` for cheap copy-on-write snapshot clones.
    pub node_id_to_row: EngineIdMap<NodeId, RowIndex>,
    /// External `EdgeId -> RowIndex` lookup (inverse of [`EdgeStore::row_to_id`]).
    pub edge_id_to_row: EngineIdMap<EdgeId, RowIndex>,
}

impl SeleneGraph {
    /// Construct an empty graph snapshot.
    #[must_use]
    pub fn new(graph_id: GraphId) -> Self {
        Self {
            meta: GraphMeta {
                graph_id,
                generation: 0,
                next_node_id: 1,
                next_edge_id: 1,
                bound_type: None,
            },
            node_store: NodeStore::new(),
            edge_store: EdgeStore::new(),
            adjacency_out: engine_id_map(),
            adjacency_in: engine_id_map(),
            idx_label: HashMap::new(),
            idx_edge_label: HashMap::new(),
            property_index: FxHashMap::default(),
            edge_property_index: FxHashMap::default(),
            composite_property_index: FxHashMap::default(),
            vector_index: FxHashMap::default(),
            text_index: FxHashMap::default(),
            node_id_to_row: engine_id_map(),
            edge_id_to_row: engine_id_map(),
        }
    }

    /// Return this graph snapshot's stable graph ID.
    #[must_use]
    pub const fn graph_id(&self) -> GraphId {
        self.meta.graph_id
    }

    /// Number of alive nodes.
    #[must_use]
    pub fn node_count(&self) -> usize {
        self.node_store.alive.len() as usize
    }

    /// Bitmap of alive node *row indices*.
    ///
    /// Returned bitmap is row-indexed (matching `nodes_with_label`), not
    /// `NodeId`-indexed; consumers convert a row to its external `NodeId` via
    /// [`Self::node_id_for_row`] (never by `row + 1` arithmetic — the external id
    /// is stable while compaction renumbers the row). Used by `selene-algorithms`
    /// to seed the "all alive nodes" baseline of a `GraphProjection`.
    #[must_use]
    pub fn live_nodes(&self) -> &RoaringBitmap {
        // B1: alive is Arc-shared COW state; expose the bitmap, not the Arc,
        // so the crate boundary (selene-algorithms) is unchanged.
        &self.node_store.alive
    }

    /// Number of alive edges.
    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edge_store.alive.len() as usize
    }

    /// Return current row-space pressure for compaction planning.
    ///
    /// This is a cheap read over store lengths and liveness bitmaps; it does not
    /// rebuild indexes or allocate a dense graph.
    #[must_use]
    pub fn compaction_stats(&self) -> crate::compaction::CompactionStats {
        crate::compaction::CompactionStats::from_graph(self)
    }

    /// Bitmap of alive edge *row indices*.
    ///
    /// The edge-side sibling of [`Self::live_nodes`]. The returned bitmap is
    /// row-indexed (matching `edges_with_label`), not `EdgeId`-indexed; consumers
    /// convert a row to its external `EdgeId` via [`Self::edge_id_for_row`] (never
    /// by `row + 1` arithmetic). Covers every alive edge regardless of label —
    /// used by the `DROP GRAPH` factory-reset (BRIEF-152) to enumerate every live
    /// edge, including untyped/arbitrary-label ones that a per-type truncate would
    /// miss.
    #[must_use]
    pub fn live_edges(&self) -> &RoaringBitmap {
        // B1: see `live_nodes` — deref the COW Arc at the boundary.
        &self.edge_store.alive
    }

    /// Map an external [`NodeId`] to its internal [`RowIndex`].
    ///
    /// Returns `None` for a never-committed (aborted-tx hole) id. A deleted id
    /// still resolves — to its now-dead row — so liveness, not existence,
    /// distinguishes it (the row's `alive` bit is clear). This is the map-backed
    /// replacement for the old `id - 1` arithmetic; the external id stays stable
    /// while BRIEF-Item-4b compaction renumbers the row.
    #[must_use]
    pub fn row_for_node_id(&self, id: NodeId) -> Option<RowIndex> {
        self.node_id_to_row.get(&id).copied()
    }

    /// Map an external [`EdgeId`] to its internal [`RowIndex`]; see
    /// [`Self::row_for_node_id`].
    #[must_use]
    pub fn row_for_edge_id(&self, id: EdgeId) -> Option<RowIndex> {
        self.edge_id_to_row.get(&id).copied()
    }

    /// Recover the external [`NodeId`] bound to a materialized [`RowIndex`].
    ///
    /// Reads the `row_to_id` column (the persistence-stable per-row id), never
    /// synthesizing `row + 1`. Returns `None` past the column end or for a
    /// never-committed hole row (which holds [`NodeId::TOMBSTONE`]).
    #[must_use]
    pub fn node_id_for_row(&self, row: RowIndex) -> Option<NodeId> {
        self.node_store
            .row_to_id
            .get(row.get() as usize)
            .copied()
            .filter(|id| *id != NodeId::TOMBSTONE)
    }

    /// Recover the external [`EdgeId`] bound to a materialized [`RowIndex`]; see
    /// [`Self::node_id_for_row`].
    #[must_use]
    pub fn edge_id_for_row(&self, row: RowIndex) -> Option<EdgeId> {
        self.edge_store
            .row_to_id
            .get(row.get() as usize)
            .copied()
            .filter(|id| *id != EdgeId::TOMBSTONE)
    }

    /// Return true when `id` names an alive node.
    #[must_use]
    pub fn is_node_alive(&self, id: NodeId) -> bool {
        self.live_node_row(id).is_some()
    }

    /// Return true when `id` names an alive edge.
    #[must_use]
    pub fn is_edge_alive(&self, id: EdgeId) -> bool {
        self.live_edge_row(id).is_some()
    }

    /// Return node labels for an alive node.
    #[must_use]
    pub fn node_labels(&self, id: NodeId) -> Option<&LabelSet> {
        self.live_node_row(id)
            .and_then(|row| self.node_store.labels.get(row))
    }

    /// Return node properties for an alive node.
    #[must_use]
    pub fn node_properties(&self, id: NodeId) -> Option<&PropertyMap> {
        self.live_node_row(id)
            .and_then(|row| self.node_store.properties.get(row))
    }

    /// Return edge label for an alive edge.
    #[must_use]
    pub fn edge_label(&self, id: EdgeId) -> Option<&DbString> {
        self.live_edge_row(id)
            .and_then(|row| self.edge_store.label.get(row))
    }

    /// Return edge endpoints for an alive edge.
    #[must_use]
    pub fn edge_endpoints(&self, id: EdgeId) -> Option<(NodeId, NodeId)> {
        self.live_edge_row(id).and_then(|row| {
            Some((
                *self.edge_store.source.get(row)?,
                *self.edge_store.target.get(row)?,
            ))
        })
    }

    /// Return edge properties for an alive edge.
    #[must_use]
    pub fn edge_properties(&self, id: EdgeId) -> Option<&PropertyMap> {
        self.live_edge_row(id)
            .and_then(|row| self.edge_store.properties.get(row))
    }

    /// Return outgoing adjacency for `source`.
    #[must_use]
    pub fn outgoing_edges(&self, source: NodeId) -> Option<&AdjacencyEntry> {
        self.adjacency_out.get(&source)
    }

    /// Return incoming adjacency for `target`.
    #[must_use]
    pub fn incoming_edges(&self, target: NodeId) -> Option<&AdjacencyEntry> {
        self.adjacency_in.get(&target)
    }

    /// Return true when an alive node has at least one incident edge.
    #[must_use]
    pub fn node_has_incident_edges(&self, id: NodeId) -> bool {
        self.outgoing_edges(id)
            .is_some_and(|entry| !entry.is_empty())
            || self
                .incoming_edges(id)
                .is_some_and(|entry| !entry.is_empty())
    }

    /// Return the bitmap of node rows carrying `label`.
    #[must_use]
    pub fn nodes_with_label(&self, label: &DbString) -> Option<&RoaringBitmap> {
        self.idx_label.get(label)
    }

    /// Return the bitmap of edge rows carrying `label`.
    #[must_use]
    pub fn edges_with_label(&self, label: &DbString) -> Option<&RoaringBitmap> {
        self.idx_edge_label.get(label)
    }

    /// Number of distinct node labels currently indexed.
    #[must_use]
    pub fn label_count(&self) -> usize {
        self.idx_label.len()
    }

    /// Number of distinct edge labels currently indexed.
    #[must_use]
    pub fn edge_label_count(&self) -> usize {
        self.idx_edge_label.len()
    }

    /// Return a clone of the registered `(label, property)` index.
    #[must_use]
    pub fn property_index_for(
        &self,
        label: &DbString,
        property: &DbString,
    ) -> Option<Arc<TypedIndex>> {
        self.property_index
            .get(&(label.clone(), property.clone()))
            .map(|entry| Arc::clone(&entry.index))
    }

    /// Return a clone of the registered edge `(label, property)` index.
    #[must_use]
    pub fn edge_property_index_for(
        &self,
        label: &DbString,
        property: &DbString,
    ) -> Option<Arc<TypedIndex>> {
        self.edge_property_index
            .get(&(label.clone(), property.clone()))
            .map(|entry| Arc::clone(&entry.index))
    }

    /// Return a clone of the registered composite index.
    #[must_use]
    pub fn composite_property_index_for(
        &self,
        label: &DbString,
        properties: &[DbString],
    ) -> Option<Arc<CompositeTypedIndex>> {
        self.composite_property_index_entry_for(label, properties)
            .map(|entry| Arc::clone(&entry.index))
    }

    /// Return composite index metadata for a property set.
    #[must_use]
    pub fn composite_property_index_entry_for(
        &self,
        label: &DbString,
        properties: &[DbString],
    ) -> Option<&CompositePropertyIndexEntry> {
        let key = composite_property_key(properties);
        self.composite_property_index.get(&(label.clone(), key))
    }

    /// Return a clone of the registered vector index.
    #[must_use]
    pub fn vector_index_for(
        &self,
        label: &DbString,
        property: &DbString,
    ) -> Option<Arc<VectorIndex>> {
        self.vector_index
            .get(&(label.clone(), property.clone()))
            .map(|entry| Arc::clone(&entry.index))
    }

    /// Return a clone of the registered text index.
    #[must_use]
    pub fn text_index_for(&self, label: &DbString, property: &DbString) -> Option<Arc<TextIndex>> {
        self.text_index
            .get(&(label.clone(), property.clone()))
            .map(|entry| Arc::clone(&entry.index))
    }

    /// Number of distinct `(label, property)` indexes currently registered.
    #[must_use]
    pub fn property_index_count(&self) -> usize {
        self.property_index.len()
    }

    /// Number of registered edge property indexes.
    #[must_use]
    pub fn edge_property_index_count(&self) -> usize {
        self.edge_property_index.len()
    }

    /// Number of distinct `(label, properties...)` indexes currently registered.
    #[must_use]
    pub fn composite_property_index_count(&self) -> usize {
        self.composite_property_index.len()
    }

    /// Number of distinct `(label, property)` vector indexes currently registered.
    #[must_use]
    pub fn vector_index_count(&self) -> usize {
        self.vector_index.len()
    }

    /// Number of distinct `(label, property)` text indexes currently registered.
    #[must_use]
    pub fn text_index_count(&self) -> usize {
        self.text_index.len()
    }

    /// Iterate built-in property indexes as owned `(label, property, kind)` tuples.
    ///
    /// This covers only SeleneGraph's built-in property indexes.
    /// Extension-provider index state is surfaced through that provider's own
    /// procedures.
    pub fn iter_property_indexes(
        &self,
    ) -> impl Iterator<Item = (DbString, DbString, TypedIndexKind)> + '_ {
        self.property_index
            .iter()
            .map(|((label, property), entry)| (label.clone(), property.clone(), entry.kind()))
    }

    /// Iterate built-in property indexes with optional explicit catalog names.
    pub fn iter_property_index_entries(
        &self,
    ) -> impl Iterator<Item = (DbString, DbString, TypedIndexKind, Option<DbString>)> + '_ {
        self.property_index
            .iter()
            .map(|((label, property), entry)| {
                (
                    label.clone(),
                    property.clone(),
                    entry.kind(),
                    entry.name.clone(),
                )
            })
    }

    /// Iterate built-in composite property indexes with optional explicit catalog names.
    pub fn iter_composite_property_index_entries(
        &self,
    ) -> impl Iterator<Item = CompositePropertyIndexEntryRow> + '_ {
        self.composite_property_index
            .iter()
            .map(|((label, _), entry)| {
                (
                    label.clone(),
                    entry.declared_properties.clone(),
                    entry.kinds(),
                    entry.name.clone(),
                )
            })
    }

    /// Iterate built-in vector indexes with optional explicit catalog names.
    pub fn iter_vector_index_entries(&self) -> impl Iterator<Item = VectorIndexEntryRow> + '_ {
        self.vector_index.iter().map(|((label, property), entry)| {
            (
                label.clone(),
                property.clone(),
                entry.kind(),
                entry.dimension(),
                entry.hnsw_config(),
                entry.ivf_config(),
                entry.name.clone(),
            )
        })
    }

    /// Iterate built-in text indexes with optional explicit catalog names.
    pub fn iter_text_index_entries(&self) -> impl Iterator<Item = TextIndexEntryRow> + '_ {
        self.text_index.iter().map(|((label, property), entry)| {
            (
                label.clone(),
                property.clone(),
                entry.stats(),
                entry.memory_usage(),
                entry.name.clone(),
            )
        })
    }

    /// Return rows matching `value` under a registered property index.
    ///
    /// `None` means no index is registered for `(label, property)` or the
    /// supplied value cannot be used with that index kind. `Some(empty)` means
    /// the index exists but no row matches. A kind-mismatched probe returns
    /// `None` so the caller drops to a linear scan; open-graph kind drift
    /// remains discoverable via cross-variant `value_compare`.
    #[must_use]
    pub fn nodes_with_property_eq(
        &self,
        label: &DbString,
        property: &DbString,
        value: &Value,
    ) -> Option<Cow<'_, RoaringBitmap>> {
        self.property_index
            .get(&(label.clone(), property.clone()))
            .and_then(|entry| entry.index.lookup_eq(value))
    }

    /// Return the union of node rows matching any indexed scalar value.
    ///
    /// `None` means no node property index is registered for `(label, property)`
    /// or at least one supplied value cannot be used with that index kind.
    /// `Some(empty)` means the index exists but no row matches the value set.
    #[must_use]
    pub fn nodes_with_property_any(
        &self,
        label: &DbString,
        property: &DbString,
        values: &[Value],
    ) -> Option<RoaringBitmap> {
        let entry = self
            .property_index
            .get(&(label.clone(), property.clone()))?;
        let mut rows = RoaringBitmap::new();
        for value in values {
            rows |= entry.index.lookup_eq(value)?.as_ref();
        }
        Some(rows)
    }

    /// Return edge rows matching `value` under a registered edge property index.
    ///
    /// `None` means no edge index is registered for `(label, property)` or the
    /// supplied value cannot be used with that index kind. `Some(empty)` means
    /// the index exists but no edge row matches.
    #[must_use]
    pub fn edges_with_property_eq(
        &self,
        label: &DbString,
        property: &DbString,
        value: &Value,
    ) -> Option<Cow<'_, RoaringBitmap>> {
        self.edge_property_index
            .get(&(label.clone(), property.clone()))
            .and_then(|entry| entry.index.lookup_eq(value))
    }

    /// Return the union of edge rows matching any indexed scalar value.
    ///
    /// `None` means no edge property index is registered for `(label,
    /// property)` or at least one supplied value cannot be used with that index
    /// kind. `Some(empty)` means the index exists but no row matches the value
    /// set.
    #[must_use]
    pub fn edges_with_property_any(
        &self,
        label: &DbString,
        property: &DbString,
        values: &[Value],
    ) -> Option<RoaringBitmap> {
        let entry = self
            .edge_property_index
            .get(&(label.clone(), property.clone()))?;
        let mut rows = RoaringBitmap::new();
        for value in values {
            rows |= entry.index.lookup_eq(value)?.as_ref();
        }
        Some(rows)
    }

    /// Return rows matching `range` under a registered property index.
    ///
    /// `None` means no index is registered or the supplied bounds do not match
    /// the index kind. `Some(empty)` means the index exists but the range
    /// matches no rows.
    #[must_use]
    pub fn nodes_with_property_range<R>(
        &self,
        label: &DbString,
        property: &DbString,
        range: R,
    ) -> Option<RoaringBitmap>
    where
        R: RangeBounds<Value>,
    {
        self.property_index
            .get(&(label.clone(), property.clone()))
            .and_then(|entry| entry.index.lookup_range(range))
    }

    /// Return edge rows matching `range` under a registered edge property index.
    ///
    /// `None` means no edge index is registered or the supplied bounds do not
    /// match the index kind. `Some(empty)` means the index exists but no edge
    /// row matches.
    #[must_use]
    pub fn edges_with_property_range<R>(
        &self,
        label: &DbString,
        property: &DbString,
        range: R,
    ) -> Option<RoaringBitmap>
    where
        R: RangeBounds<Value>,
    {
        self.edge_property_index
            .get(&(label.clone(), property.clone()))
            .and_then(|entry| entry.index.lookup_range(range))
    }

    /// Return rows whose string property key starts with `prefix`.
    ///
    /// `None` means no index is registered or the registered index is not a
    /// string index.
    #[must_use]
    pub fn nodes_with_property_prefix(
        &self,
        label: &DbString,
        property: &DbString,
        prefix: &str,
    ) -> Option<RoaringBitmap> {
        self.property_index
            .get(&(label.clone(), property.clone()))
            .and_then(|entry| entry.index.lookup_prefix(prefix))
    }

    fn live_node_row(&self, id: NodeId) -> Option<usize> {
        let row = self.row_for_node_id(id)?.get();
        ((row as usize) < self.node_store.len() && self.node_store.is_alive(row))
            .then_some(row as usize)
    }

    fn live_edge_row(&self, id: EdgeId) -> Option<usize> {
        let row = self.row_for_edge_id(id)?.get();
        ((row as usize) < self.edge_store.len() && self.edge_store.is_alive(row))
            .then_some(row as usize)
    }
}

pub(crate) fn composite_property_key(properties: &[DbString]) -> SmallVec<[DbString; 4]> {
    let mut key: SmallVec<[DbString; 4]> = properties.iter().cloned().collect();
    key.sort();
    key
}

#[cfg(test)]
mod tests;