selene-db-graph 1.2.0

//! Typed mutation funnel per spec 03 section 4.3.

mod assignment;
mod catalog;
mod composite_property_index;
mod delete;
mod delete_set;
mod factory_reset;
mod property_index;
mod remove;
mod text_index;
mod vector_index;

use std::sync::Arc;

use roaring::RoaringBitmap;
use selene_core::{
    Change, DbString, EdgeId, GraphId, LabelDiff, LabelSet, NodeId, PropertyDiff, PropertyMap,
    SchemaChange,
};

use crate::adjacency::AdjacencyEdge;
use crate::error::{GraphError, GraphResult};
use crate::graph_types::{GraphTypeDef, PropertyTypeDef};
use crate::index_provider::{IndexProvider, ProviderTag};
use crate::store::RowIndex;
use crate::type_validator::{EntityId, TypeViolation};
use crate::write_txn::WriteTxn;

/// Borrowed mutation builder for one write transaction.
pub struct Mutator<'tx, 'g> {
    txn: &'tx mut WriteTxn<'g>,
}

impl<'tx, 'g> Mutator<'tx, 'g> {
    pub(crate) fn new(txn: &'tx mut WriteTxn<'g>) -> Self {
        Self { txn }
    }

    /// Create a node, emit `Change::NodeCreated`, and return its ID.
    ///
    /// # Errors
    ///
    /// Returns [`GraphError::RowSpaceExhausted`] when the dense row store fills
    /// the v1 row-index range (max 2^32 rows).
    pub fn create_node(&mut self, labels: LabelSet, mut props: PropertyMap) -> GraphResult<NodeId> {
        fill_node_defaults(self.txn.read(), &labels, &mut props)?;
        assignment::coerce_node_properties(self.txn.read(), &labels, &mut props)?;
        let id = self.txn.allocator.allocate_node();
        {
            let graph = self.txn.guard_mut();
            // BRIEF-Item-4c: append at the dense end (row = current row count)
            // instead of `id - 1` arithmetic. After 4b compaction the monotonic
            // high-water id far exceeds the dense row count, so an arith row would
            // re-pad exactly the holes compaction reclaimed; append keeps the store
            // dense and never resurrects a reclaimed slot. The u32 row-space cap
            // therefore moves from the id value to the row count.
            let row = u32::try_from(graph.node_store.len())
                .ok()
                // u32::MAX is reserved as RowIndex::TOMBSTONE; the last real row
                // is u32::MAX - 1, so a live row never aliases the sentinel.
                .filter(|&row| row != u32::MAX)
                .ok_or(GraphError::RowSpaceExhausted {
                    kind: "node",
                    rows: graph.node_store.len() as u64,
                    max_rows: u32::MAX as u64,
                })?;
            // BRIEF-153 fix-cycle C2: run property-index admission BEFORE
            // mutating row state so a cap-exhaustion error rolls back
            // cleanly with no half-written row. Index updates only touch
            // their own maps; node_store stays untouched if any admission
            // fails. The txn boundary publishes everything atomically on
            // commit, so a partial-index publication is impossible.
            crate::property_index::apply_node_create(
                &mut graph.property_index,
                &labels,
                &props,
                row,
            )?;
            crate::composite_property_index::apply_node_create(
                &mut graph.composite_property_index,
                &labels,
                &props,
                row,
            )?;
            crate::vector_index::apply_node_create(&mut graph.vector_index, &labels, &props, row)?;
            crate::text_index::apply_node_create(&mut graph.text_index, &labels, &props, row, id);
            graph.node_store.labels.push(labels.clone());
            graph.node_store.properties.push(props.clone());
            graph.node_store.row_to_id.push(id);
            graph.node_store.alive_mut().insert(row);
            // BRIEF-Item-4a: bind the external id to its row in both directions.
            // The live commit path never re-runs `rebuild_id_maps`, so the
            // `id -> row` map must be populated here. The row is remappable once
            // 4b compaction renumbers rows under stable ids.
            graph.node_id_to_row.insert(id, RowIndex::new(row));
            insert_node_labels(&mut graph.idx_label, row, &labels);
        }
        self.txn.changes.push(Change::NodeCreated {
            id,
            labels,
            properties: props,
        });
        Ok(id)
    }

    /// Create an edge between two alive nodes.
    pub fn create_edge(
        &mut self,
        label: DbString,
        source: NodeId,
        target: NodeId,
        mut props: PropertyMap,
    ) -> GraphResult<EdgeId> {
        self.require_live_node(source)?;
        self.require_live_node(target)?;
        fill_edge_defaults(self.txn.read(), label.clone(), source, target, &mut props)?;
        assignment::coerce_edge_properties(
            self.txn.read(),
            label.clone(),
            source,
            target,
            &mut props,
        )?;
        let id = self.txn.allocator.allocate_edge();
        {
            let graph = self.txn.guard_mut();
            // BRIEF-Item-4c: append at the dense end (see create_node).
            let row = u32::try_from(graph.edge_store.len())
                .ok()
                .filter(|&row| row != u32::MAX) // u32::MAX is RowIndex::TOMBSTONE
                .ok_or(GraphError::RowSpaceExhausted {
                    kind: "edge",
                    rows: graph.edge_store.len() as u64,
                    max_rows: u32::MAX as u64,
                })?;
            graph.edge_store.label.push(label.clone());
            graph.edge_store.source.push(source);
            graph.edge_store.target.push(target);
            graph.edge_store.properties.push(props.clone());
            graph.edge_store.row_to_id.push(id);
            graph.edge_store.alive_mut().insert(row);
            // BRIEF-Item-4a: bind the external edge id to its row (live path).
            graph.edge_id_to_row.insert(id, RowIndex::new(row));
            insert_index_row(&mut graph.idx_edge_label, label.clone(), row);

            graph
                .adjacency_out
                .entry(source)
                .or_default()
                .add(AdjacencyEdge {
                    label: label.clone(),
                    neighbor: target,
                    edge_id: id,
                });
            graph
                .adjacency_in
                .entry(target)
                .or_default()
                .add(AdjacencyEdge {
                    label: label.clone(),
                    neighbor: source,
                    edge_id: id,
                });
        }
        self.txn.changes.push(Change::EdgeCreated {
            id,
            label,
            source,
            target,
            properties: props,
        });
        Ok(id)
    }

    /// Update an alive node and emit `Change::NodeUpdated`.
    pub fn update_node(
        &mut self,
        id: NodeId,
        labels_diff: LabelDiff,
        mut props_diff: PropertyDiff,
    ) -> GraphResult<()> {
        let row = self.require_live_node(id)?;

        // Compute the new label set without mutating the working graph yet.
        let old_labels = self
            .txn
            .read()
            .node_store
            .labels
            .get(row)
            .cloned()
            .unwrap_or_default();
        let mut labels = old_labels.clone();
        for label in labels_diff.added.iter().cloned() {
            labels.insert(label);
        }
        for label in labels_diff.removed.iter() {
            labels.remove(label);
        }
        reject_immutable_node_update(self.txn.read(), id, &old_labels, &props_diff)?;
        reject_immutable_node_update(self.txn.read(), id, &labels, &props_diff)?;
        assignment::coerce_node_property_diff(self.txn.read(), &labels, &mut props_diff)?;

        // Apply the property diff up front; if it errors we leave the working
        // graph (including idx_label) untouched so the transaction can still
        // be safely rolled back or aborted without leaking inconsistent state.
        let old_props = self
            .txn
            .read()
            .node_store
            .properties
            .get(row)
            .cloned()
            .unwrap_or_default();
        let mut props = old_props.clone();
        apply_property_diff(&mut props, &props_diff)?;

        // BRIEF-153 fix-cycle C2: run property-index admission BEFORE
        // mutating row state so a cap-exhaustion error rolls back cleanly
        // with no half-written row. The index updates borrow `labels`/`props`;
        // both move into the row stores only after every fallible step.
        {
            let graph = self.txn.guard_mut();
            crate::property_index::apply_node_update(
                &mut graph.property_index,
                &old_labels,
                &old_props,
                &labels,
                &props,
                row as u32,
            )?;
            crate::composite_property_index::apply_node_update(
                &mut graph.composite_property_index,
                &old_labels,
                &old_props,
                &labels,
                &props,
                row as u32,
            )?;
            crate::vector_index::apply_node_update(
                &mut graph.vector_index,
                &old_labels,
                &old_props,
                &labels,
                &props,
                row as u32,
            )?;
            crate::text_index::apply_node_update(
                &mut graph.text_index,
                &old_labels,
                &old_props,
                &labels,
                &props,
                row as u32,
                id,
            );
            graph.node_store.labels.set(row, labels);
            graph.node_store.properties.set(row, props);
            for label in labels_diff.added.iter().cloned() {
                insert_index_row(&mut graph.idx_label, label, row as u32);
            }
            for label in labels_diff.removed.iter() {
                remove_index_row(&mut graph.idx_label, label, row as u32);
            }
        }

        self.txn.changes.push(Change::NodeUpdated {
            id,
            labels_diff,
            properties_diff: props_diff,
        });
        Ok(())
    }

    /// Update an alive edge and emit `Change::EdgeUpdated`.
    ///
    /// Edge labels are immutable, so property updates do not touch
    /// `idx_edge_label`.
    pub fn update_edge(&mut self, id: EdgeId, mut props_diff: PropertyDiff) -> GraphResult<()> {
        let row = self.require_live_edge(id)?;
        reject_immutable_edge_update(self.txn.read(), id, &props_diff)?;
        assignment::coerce_edge_property_diff(self.txn.read(), id, &mut props_diff)?;
        let mut props = self
            .txn
            .read()
            .edge_store
            .properties
            .get(row)
            .cloned()
            .unwrap_or_default();
        apply_property_diff(&mut props, &props_diff)?;
        self.txn.guard_mut().edge_store.properties.set(row, props);
        self.txn.changes.push(Change::EdgeUpdated {
            id,
            properties_diff: props_diff,
        });
        Ok(())
    }

    /// Append a raw [`SchemaChange`] WAL payload through the write funnel.
    ///
    /// This is a pass-through accumulator: catalog graph mutation and
    /// closed-graph validation are handled by higher-level validation layers
    /// (the typed catalog DDL methods on this `Mutator` — e.g. `create_node_type`
    /// — call those layers and then funnel here).
    ///
    /// Why: this is the single, canonical funnel entry for a `SchemaChanged`
    /// change record (hard rule 11 — every mutation routes through the one
    /// `Mutator`). It is intentionally retained as a `pub` funnel surface even
    /// though no GQL caller reaches it directly today: the catalog DDL methods
    /// are the production producers, and keeping the low-level entry public means
    /// any future schema-event producer routes through the same funnel rather
    /// than re-implementing the write path. Tests and benches drive it directly
    /// to exercise the raw funnel without the DDL validation layer on top.
    pub fn schema_change(&mut self, graph: GraphId, change: SchemaChange) {
        self.txn
            .changes
            .push(Change::SchemaChanged { graph, change });
    }

    /// Look up a registered index provider through the held write transaction.
    #[must_use]
    pub fn index_provider_by_tag(&self, tag: ProviderTag) -> Option<Arc<dyn IndexProvider>> {
        self.txn
            .providers
            .iter()
            .find(|provider| provider.provider_tag() == tag)
            .map(Arc::clone)
    }

    /// Borrow the transaction-local working graph.
    #[must_use]
    pub fn read(&self) -> &crate::SeleneGraph {
        self.txn.read()
    }

    fn require_live_node(&self, id: NodeId) -> GraphResult<usize> {
        let graph = self.txn.read();
        // Map-backed: a never-committed (aborted-tx hole) id is absent from the
        // map -> NotFound. A deleted id stays mapped to its dead row -> NotAlive.
        let row = graph
            .row_for_node_id(id)
            .ok_or(GraphError::NodeNotFound { id })?
            .get();
        if row as usize >= graph.node_store.len() {
            return Err(GraphError::NodeNotFound { id });
        }
        if !graph.node_store.is_alive(row) {
            return Err(GraphError::NodeNotAlive { id });
        }
        Ok(row as usize)
    }

    fn require_live_edge(&self, id: EdgeId) -> GraphResult<usize> {
        let graph = self.txn.read();
        let row = graph
            .row_for_edge_id(id)
            .ok_or(GraphError::EdgeNotFound { id })?
            .get();
        if row as usize >= graph.edge_store.len() {
            return Err(GraphError::EdgeNotFound { id });
        }
        if !graph.edge_store.is_alive(row) {
            return Err(GraphError::EdgeNotAlive { id });
        }
        Ok(row as usize)
    }
}

fn insert_node_labels(
    index: &mut imbl::HashMap<DbString, RoaringBitmap>,
    row: u32,
    labels: &LabelSet,
) {
    for label in labels.iter().cloned() {
        insert_index_row(index, label, row);
    }
}

fn remove_node_labels(
    index: &mut imbl::HashMap<DbString, RoaringBitmap>,
    row: u32,
    labels: &LabelSet,
) {
    for label in labels.iter() {
        remove_index_row(index, label, row);
    }
}

fn insert_index_row(index: &mut imbl::HashMap<DbString, RoaringBitmap>, label: DbString, row: u32) {
    // In-place insert via `entry().or_default()`: the rebuild path uses the same
    // idiom (see `consistency.rs` / `typed_index.rs`). `guard_mut` already gives
    // unique ownership of the bitmap (Arc::make_mut), so we never clone the whole
    // RoaringBitmap per label per node — bulk-loading one label is O(N), not O(N²).
    index.entry(label).or_default().insert(row);
}

fn remove_index_row(
    index: &mut imbl::HashMap<DbString, RoaringBitmap>,
    label: &DbString,
    row: u32,
) {
    // Mirror `insert_index_row`: mutate the bitmap behind the imbl entry
    // instead of cloning the whole RoaringBitmap for every row removed.
    let now_empty = match index.get_mut(label) {
        Some(bitmap) => {
            bitmap.remove(row);
            bitmap.is_empty()
        }
        None => false,
    };
    if now_empty {
        index.remove(label);
    }
}

fn fill_node_defaults(
    graph: &crate::SeleneGraph,
    labels: &LabelSet,
    props: &mut PropertyMap,
) -> GraphResult<()> {
    let Some(graph_type) = graph.meta.bound_type.as_deref() else {
        return Ok(());
    };
    let Some(node_type) = graph_type.find_node_type(labels) else {
        return Ok(());
    };
    fill_property_defaults(&node_type.properties, props)
}

fn fill_edge_defaults(
    graph: &crate::SeleneGraph,
    label: DbString,
    source: NodeId,
    target: NodeId,
    props: &mut PropertyMap,
) -> GraphResult<()> {
    let Some(graph_type) = graph.meta.bound_type.as_deref() else {
        return Ok(());
    };
    let Some(source_type) = node_type_index_for_node(graph, graph_type, source) else {
        return Ok(());
    };
    let Some(target_type) = node_type_index_for_node(graph, graph_type, target) else {
        return Ok(());
    };
    let Some(edge_type) = graph_type.find_edge_type(label, source_type, target_type) else {
        return Ok(());
    };
    fill_property_defaults(&edge_type.properties, props)
}

fn fill_property_defaults(
    declarations: &[PropertyTypeDef],
    props: &mut PropertyMap,
) -> GraphResult<()> {
    for declaration in declarations {
        if props.contains_key(&declaration.name) {
            continue;
        }
        if let Some(default) = &declaration.default {
            props.set(declaration.name.clone(), default.to_value()?)?;
        }
    }
    Ok(())
}

fn reject_immutable_node_update(
    graph: &crate::SeleneGraph,
    id: NodeId,
    labels: &LabelSet,
    diff: &PropertyDiff,
) -> GraphResult<()> {
    let Some(graph_type) = graph.meta.bound_type.as_deref() else {
        return Ok(());
    };
    let Some(node_type) = graph_type.find_node_type(labels) else {
        return Ok(());
    };
    reject_immutable_property_update(
        EntityId::Node(id),
        node_type.name.clone(),
        &node_type.properties,
        diff,
    )
}

fn reject_immutable_edge_update(
    graph: &crate::SeleneGraph,
    id: EdgeId,
    diff: &PropertyDiff,
) -> GraphResult<()> {
    let Some(graph_type) = graph.meta.bound_type.as_deref() else {
        return Ok(());
    };
    let Some(label) = graph.edge_label(id).cloned() else {
        return Ok(());
    };
    let Some((source, target)) = graph.edge_endpoints(id) else {
        return Ok(());
    };
    let Some(source_type) = node_type_index_for_node(graph, graph_type, source) else {
        return Ok(());
    };
    let Some(target_type) = node_type_index_for_node(graph, graph_type, target) else {
        return Ok(());
    };
    let Some(edge_type) = graph_type.find_edge_type(label, source_type, target_type) else {
        return Ok(());
    };
    reject_immutable_property_update(
        EntityId::Edge(id),
        edge_type.name.clone(),
        &edge_type.properties,
        diff,
    )
}

fn node_type_index_for_node(
    graph: &crate::SeleneGraph,
    graph_type: &GraphTypeDef,
    node: NodeId,
) -> Option<u32> {
    let labels = graph.node_labels(node)?;
    graph_type.find_node_type_index(labels)
}

fn reject_immutable_property_update(
    entity_id: EntityId,
    declared_in: DbString,
    declarations: &[PropertyTypeDef],
    diff: &PropertyDiff,
) -> GraphResult<()> {
    for (key, _) in &diff.set {
        reject_if_immutable(entity_id, declared_in.clone(), declarations, key.clone())?;
    }
    for key in &diff.removed {
        reject_if_immutable(entity_id, declared_in.clone(), declarations, key.clone())?;
    }
    Ok(())
}

fn reject_if_immutable(
    entity_id: EntityId,
    declared_in: DbString,
    declarations: &[PropertyTypeDef],
    property: DbString,
) -> GraphResult<()> {
    if declarations
        .iter()
        .any(|declaration| declaration.name == property && declaration.immutable)
    {
        return Err(GraphError::TypeViolation(
            TypeViolation::ImmutablePropertyUpdate {
                entity_id,
                property,
                declared_in,
            },
        ));
    }
    Ok(())
}

fn apply_property_diff(map: &mut PropertyMap, diff: &PropertyDiff) -> GraphResult<()> {
    for (key, value) in diff.set.iter() {
        map.set(key.clone(), value.clone())?;
    }
    for key in diff.removed.iter() {
        map.remove(key);
    }
    Ok(())
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod id_map_tests;

#[cfg(test)]
mod row_cap_tests;

#[cfg(test)]
mod truncate_tests;

#[cfg(test)]
mod factory_reset_tests;

#[cfg(test)]
mod hub_delete_tests;

#[cfg(test)]
mod delete_set_tests;

#[cfg(test)]
mod payload_clear_tests;