lora_store/memory/

impls.rs

//! `GraphStorage` / `GraphStorageMut` impls for [`InMemoryGraph`]. Both
//! delegate into the inherent helpers defined in `super`.

use std::collections::BTreeSet;

use lora_ast::Direction;

use crate::encoded::PropEdit;
use crate::{
    ConstraintDefinition, ConstraintRequest, CreateConstraintError, CreateConstraintOutcome,
    CreateIndexError, CreateIndexOutcome, DropConstraintError, DropConstraintOutcome,
    DropIndexError, DropIndexOutcome, GraphStats, GraphStorage, GraphStorageMut, IndexDefinition,
    IndexRequest, LoraVector, MutationEvent, NodeId, NodeRecord, NodeRef, Properties,
    PropertyValue, RelRef, RelationshipId, RelationshipRecord, StoredIndexEntity,
};

use super::graph::PendingAdjRemovals;
use super::property_index::PropertyIndexKey;
use super::InMemoryGraph;

impl GraphStorage for InMemoryGraph {
    // ---------- Required primitives ----------

    fn list_indexes(&self) -> Vec<IndexDefinition> {
        self.index_catalog_read().list()
    }

    fn get_index(&self, name: &str) -> Option<IndexDefinition> {
        self.index_catalog_read().get(name).cloned()
    }

    fn fulltext_search(&self, name: &str, query: &str, limit: Option<usize>) -> Vec<(u64, f64)> {
        // The index lives on exactly one entity scope; check both.
        let node_hits = self
            .fulltext_indexes_read(StoredIndexEntity::Node)
            .get(name)
            .map(|idx| idx.query(query, limit));
        if let Some(hits) = node_hits {
            if !hits.is_empty() {
                return hits;
            }
        }
        let rel_hits = self
            .fulltext_indexes_read(StoredIndexEntity::Relationship)
            .get(name)
            .map(|idx| idx.query(query, limit));
        rel_hits.unwrap_or_default()
    }

    fn vector_search(
        &self,
        name: &str,
        query: &LoraVector,
        k: usize,
        restrict_to: Option<&std::collections::BTreeSet<u64>>,
    ) -> Vec<(u64, f64)> {
        // Lazy populate: an index created with
        // `vector.populate.async: true` stays Populating until the
        // first query forces the backfill. The check is cheap (one
        // catalog read), and the populate itself takes the write
        // lock — concurrent queries pile up on the read lock and
        // proceed once the writer drops.
        self.lazy_populate_vector_index(name);

        // Like fulltext_search, an index lives on exactly one entity
        // scope. Probe both registries; the catalog has already
        // enforced kind/entity agreement so only one will hit.
        if let Some(hits) =
            self.vector_indexes_read(StoredIndexEntity::Node)
                .query(name, query, k, restrict_to)
        {
            if !hits.is_empty() {
                return hits;
            }
        }
        self.vector_indexes_read(StoredIndexEntity::Relationship)
            .query(name, query, k, restrict_to)
            .unwrap_or_default()
    }

    fn list_constraints(&self) -> Vec<ConstraintDefinition> {
        self.constraint_catalog_read().list()
    }

    fn get_constraint(&self, name: &str) -> Option<ConstraintDefinition> {
        self.constraint_catalog_read().get(name).cloned()
    }

    fn check_node_create_against_constraints(
        &self,
        labels: &[String],
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_create(&catalog, self, labels, properties)
            .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_create_against_constraints(
        &self,
        rel_type: &str,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_create(
            &catalog, self, rel_type, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_create_deferring_existence(
        &self,
        labels: &[String],
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_create_deferred(
            &catalog, self, labels, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_create_deferring_existence(
        &self,
        rel_type: &str,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_create_deferred(
            &catalog, self, rel_type, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_existence_constraints(&self, node_id: NodeId) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_existence(&catalog, self, node_id)
            .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_existence_constraints(
        &self,
        rel_id: RelationshipId,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_existence(&catalog, self, rel_id)
            .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_set_property_against_constraints(
        &self,
        node_id: NodeId,
        key: &str,
        value: &PropertyValue,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_set_property(
            &catalog, self, node_id, key, value,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_remove_property_against_constraints(
        &self,
        node_id: NodeId,
        key: &str,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_remove_property(&catalog, self, node_id, key)
            .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_replace_properties_against_constraints(
        &self,
        node_id: NodeId,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_replace_properties(
            &catalog, self, node_id, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_set_property_against_constraints(
        &self,
        rel_id: RelationshipId,
        key: &str,
        value: &PropertyValue,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_set_property(
            &catalog, self, rel_id, key, value,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_remove_property_against_constraints(
        &self,
        rel_id: RelationshipId,
        key: &str,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_remove_property(
            &catalog, self, rel_id, key,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_replace_properties_against_constraints(
        &self,
        rel_id: RelationshipId,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_replace_properties(
            &catalog, self, rel_id, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_add_label_against_constraints(
        &self,
        node_id: NodeId,
        label: &str,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_add_label(&catalog, self, node_id, label)
            .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_replace_properties_deferring_existence(
        &self,
        node_id: NodeId,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_replace_properties_deferred(
            &catalog, self, node_id, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_relationship_replace_properties_deferring_existence(
        &self,
        rel_id: RelationshipId,
        properties: &Properties,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_relationship_replace_properties_deferred(
            &catalog, self, rel_id, properties,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn check_node_add_label_deferring_existence(
        &self,
        node_id: NodeId,
        label: &str,
    ) -> Result<(), String> {
        if !self.has_active_constraints() {
            return Ok(());
        }
        let catalog = self.constraint_catalog_read();
        crate::memory::constraint_enforce::check_node_add_label_deferred(
            &catalog, self, node_id, label,
        )
        .map_err(|e| format!("[{}] {e}", e.gql_status()))
    }

    fn graph_stats(&self) -> GraphStats {
        InMemoryGraph::graph_stats(self)
    }

    fn plan_signature(&self) -> u64 {
        InMemoryGraph::plan_signature(self)
    }

    fn node_text_candidates(
        &self,
        label: &str,
        property: &str,
        query: &str,
    ) -> Option<Vec<NodeId>> {
        let registry = self.text_indexes_read(crate::StoredIndexEntity::Node);
        let candidates = registry.candidates(label, property, query)?;
        Some(candidates.into_iter().collect())
    }

    fn node_range_candidates(
        &self,
        label: &str,
        property: &str,
        lo: Option<&PropertyValue>,
        hi: Option<&PropertyValue>,
    ) -> Option<Vec<NodeId>> {
        let registry = self.sorted_indexes_read(crate::StoredIndexEntity::Node);
        // The sorted index returns a [lo, hi] inclusive candidate set.
        // The executor refilters each id against the precise predicate
        // (handling `>` vs `>=`, `<` vs `<=`).
        let candidates = registry.range_candidates(label, property, lo, hi)?;
        Some(candidates.into_iter().collect())
    }

    fn node_range_other_temporal_kind_ids(
        &self,
        label: &str,
        property: &str,
        like: &PropertyValue,
    ) -> Option<Vec<NodeId>> {
        self.sorted_indexes_read(crate::StoredIndexEntity::Node)
            .other_temporal_kind_ids(label, property, like)
    }

    fn node_range_ordered_chunk(
        &self,
        label: &str,
        property: &str,
        lo: Option<&PropertyValue>,
        hi: Option<&PropertyValue>,
        descending: bool,
        after: Option<(&PropertyValue, NodeId)>,
        max: usize,
    ) -> Option<Vec<NodeId>> {
        self.sorted_indexes_read(crate::StoredIndexEntity::Node)
            .ordered_chunk(label, property, lo, hi, descending, after, max)
    }

    fn node_point_within_bbox(
        &self,
        label: &str,
        property: &str,
        ll: (f64, f64),
        ur: (f64, f64),
    ) -> Option<Vec<NodeId>> {
        let registry = self.point_indexes_read(crate::StoredIndexEntity::Node);
        let candidates = registry.within_bbox(label, property, ll, ur)?;
        Some(candidates.into_iter().collect())
    }

    fn node_point_within_distance(
        &self,
        label: &str,
        property: &str,
        center: (f64, f64),
        max_distance: f64,
    ) -> Option<Vec<NodeId>> {
        let registry = self.point_indexes_read(crate::StoredIndexEntity::Node);
        let candidates = registry.within_distance(label, property, center, max_distance)?;
        Some(candidates.into_iter().collect())
    }

    fn relationship_text_candidates(
        &self,
        rel_type: &str,
        property: &str,
        query: &str,
    ) -> Option<Vec<RelationshipId>> {
        let registry = self.text_indexes_read(crate::StoredIndexEntity::Relationship);
        let candidates = registry.candidates(rel_type, property, query)?;
        Some(candidates.into_iter().collect())
    }

    fn relationship_range_candidates(
        &self,
        rel_type: &str,
        property: &str,
        lo: Option<&PropertyValue>,
        hi: Option<&PropertyValue>,
    ) -> Option<Vec<RelationshipId>> {
        let registry = self.sorted_indexes_read(crate::StoredIndexEntity::Relationship);
        let candidates = registry.range_candidates(rel_type, property, lo, hi)?;
        Some(candidates.into_iter().collect())
    }

    fn relationship_range_other_temporal_kind_ids(
        &self,
        rel_type: &str,
        property: &str,
        like: &PropertyValue,
    ) -> Option<Vec<RelationshipId>> {
        self.sorted_indexes_read(crate::StoredIndexEntity::Relationship)
            .other_temporal_kind_ids(rel_type, property, like)
    }

    fn relationship_point_within_bbox(
        &self,
        rel_type: &str,
        property: &str,
        ll: (f64, f64),
        ur: (f64, f64),
    ) -> Option<Vec<RelationshipId>> {
        let registry = self.point_indexes_read(crate::StoredIndexEntity::Relationship);
        let candidates = registry.within_bbox(rel_type, property, ll, ur)?;
        Some(candidates.into_iter().collect())
    }

    fn relationship_point_within_distance(
        &self,
        rel_type: &str,
        property: &str,
        center: (f64, f64),
        max_distance: f64,
    ) -> Option<Vec<RelationshipId>> {
        let registry = self.point_indexes_read(crate::StoredIndexEntity::Relationship);
        let candidates = registry.within_distance(rel_type, property, center, max_distance)?;
        Some(candidates.into_iter().collect())
    }

    fn contains_node(&self, id: NodeId) -> bool {
        self.has_node_at(id)
    }

    fn node(&self, id: NodeId) -> Option<NodeRecord> {
        self.node_at(id).map(|r| r.to_record())
    }

    fn all_node_ids(&self) -> Vec<NodeId> {
        self.iter_node_ids().collect()
    }

    fn node_ids_by_label(&self, label: &str) -> Vec<NodeId> {
        match self.nodes_by_label.get(label) {
            Some(ids) => ids.to_vec(),
            None => Vec::new(),
        }
    }

    fn scan_node_ids(
        &self,
        label: Option<&str>,
        cursor: &mut u64,
        max: usize,
        out: &mut Vec<NodeId>,
    ) -> bool {
        let max = max.max(1);
        match label {
            Some(label) => {
                let Some(ids) = self.nodes_by_label.get(label) else {
                    return false;
                };
                // The cursor is the next id the scan may return.
                fill_page(ids.iter_from(*cursor), cursor, max, out)
            }
            None => {
                // The cursor is the next slot the scan may return.
                let Ok(from) = usize::try_from(*cursor) else {
                    return false;
                };
                let slots = self.nodes.slots_from(from).map(|slot| slot as NodeId);
                fill_page(slots, cursor, max, out)
            }
        }
    }

    fn contains_relationship(&self, id: RelationshipId) -> bool {
        self.has_rel_at(id)
    }

    fn relationship(&self, id: RelationshipId) -> Option<RelationshipRecord> {
        self.rel_at(id).map(|r| r.to_record())
    }

    fn all_rel_ids(&self) -> Vec<RelationshipId> {
        self.iter_rel_ids().collect()
    }

    fn rel_ids_by_type(&self, rel_type: &str) -> Vec<RelationshipId> {
        match self.relationships_by_type.get(rel_type) {
            Some(ids) => ids.to_vec(),
            None => Vec::new(),
        }
    }

    fn relationship_endpoints(&self, id: RelationshipId) -> Option<(NodeId, NodeId)> {
        self.rel_endpoints_at(id)
    }

    fn expand_ids(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
    ) -> Vec<(RelationshipId, NodeId)> {
        if !self.has_node_at(node_id) {
            return Vec::new();
        }

        let mut out: Vec<(RelationshipId, NodeId)> = Vec::new();
        let _ = self.try_for_each_adjacent_id_unchecked(
            node_id,
            direction,
            types,
            |rel_id, other_id| {
                out.push((rel_id, other_id));
                Ok::<(), ()>(())
            },
        );
        out
    }

    fn try_for_each_expand_id<F, E>(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
        visit: F,
    ) -> Result<(), E>
    where
        F: FnMut(RelationshipId, NodeId) -> Result<(), E>,
        Self: Sized,
    {
        self.try_for_each_adjacent_id(node_id, direction, types, visit)
    }

    fn all_labels(&self) -> Vec<String> {
        self.nodes_by_label.keys().map(|k| k.to_string()).collect()
    }

    fn all_relationship_types(&self) -> Vec<String> {
        self.relationships_by_type
            .keys()
            .map(|k| k.to_string())
            .collect()
    }

    // ---------- Optimization hooks: zero-clone borrow access ----------

    fn with_node<F, R>(&self, id: NodeId, f: F) -> Option<R>
    where
        F: FnOnce(NodeRef<'_>) -> R,
        Self: Sized,
    {
        self.node_at(id).map(f)
    }

    fn with_relationship<F, R>(&self, id: RelationshipId, f: F) -> Option<R>
    where
        F: FnOnce(RelRef<'_>) -> R,
        Self: Sized,
    {
        self.rel_at(id).map(f)
    }

    // ---------- Overrides: counts + existence ----------

    fn has_node(&self, id: NodeId) -> bool {
        self.has_node_at(id)
    }

    fn has_relationship(&self, id: RelationshipId) -> bool {
        self.has_rel_at(id)
    }

    fn node_count(&self) -> usize {
        self.live_node_count
    }

    fn relationship_count(&self) -> usize {
        self.live_rel_count
    }

    fn node_count_by_label(&self, label: &str) -> usize {
        self.nodes_by_label.get(label).map_or(0, |ids| ids.len())
    }

    // ---------- Overrides: record-returning scans (direct iteration) ----------

    fn all_nodes(&self) -> Vec<NodeRecord> {
        self.iter_node_refs().map(|r| r.to_record()).collect()
    }

    fn nodes_by_label(&self, label: &str) -> Vec<NodeRecord> {
        self.nodes_by_label
            .get(label)
            .into_iter()
            .flat_map(|ids| ids.iter())
            .filter_map(|id| self.node_at(id).map(|r| r.to_record()))
            .collect()
    }

    fn all_relationships(&self) -> Vec<RelationshipRecord> {
        self.iter_rel_refs().map(|r| r.to_record()).collect()
    }

    fn relationships_by_type(&self, rel_type: &str) -> Vec<RelationshipRecord> {
        self.relationships_by_type
            .get(rel_type)
            .into_iter()
            .flat_map(|ids| ids.iter())
            .filter_map(|id| self.rel_at(id).map(|r| r.to_record()))
            .collect()
    }

    fn relationship_ids_of(&self, node_id: NodeId, direction: Direction) -> Vec<RelationshipId> {
        self.relationship_ids_for_direction(node_id, direction)
    }

    fn outgoing_relationships(&self, node_id: NodeId) -> Vec<RelationshipRecord> {
        self.outgoing_at(node_id)
            .into_iter()
            .flat_map(|adj| adj.iter())
            .filter_map(|entry| self.rel_at(entry.rel).map(|r| r.to_record()))
            .collect()
    }

    fn incoming_relationships(&self, node_id: NodeId) -> Vec<RelationshipRecord> {
        self.incoming_at(node_id)
            .into_iter()
            .flat_map(|adj| adj.iter())
            .filter_map(|entry| self.rel_at(entry.rel).map(|r| r.to_record()))
            .collect()
    }

    fn relationships_of(&self, node_id: NodeId, direction: Direction) -> Vec<RelationshipRecord> {
        let mut out = Vec::new();
        let _ = self.try_for_each_expand_id(node_id, direction, &[], |rel_id, _| {
            if let Some(rel) = self.rel_at(rel_id) {
                out.push(rel.to_record());
            }
            Ok::<(), ()>(())
        });
        out
    }

    fn degree(&self, node_id: NodeId, direction: Direction) -> usize {
        // Counting walks the list: adjacency stores no length.
        let out = || self.outgoing_at(node_id).map_or(0, |adj| adj.count());
        match direction {
            Direction::Right => out(),
            Direction::Left => self.incoming_at(node_id).map_or(0, |adj| adj.count()),
            Direction::Undirected => {
                let incoming_non_self = self
                    .incoming_at(node_id)
                    .into_iter()
                    .flat_map(|adj| adj.iter())
                    .filter(|entry| entry.neighbour != node_id)
                    .count();
                out() + incoming_non_self
            }
        }
    }

    fn expand(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
    ) -> Vec<(RelationshipRecord, NodeRecord)> {
        if !self.has_node_at(node_id) {
            return Vec::new();
        }

        let mut out = Vec::new();
        let _ = self.try_for_each_expand_id(node_id, direction, types, |rel_id, other_id| {
            if let (Some(rel), Some(other)) = (self.rel_at(rel_id), self.node_at(other_id)) {
                out.push((rel.to_record(), other.to_record()));
            }
            Ok::<(), ()>(())
        });
        out
    }

    fn neighbors(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
    ) -> Vec<NodeRecord> {
        let mut out = Vec::new();
        let _ = self.try_for_each_expand_id(node_id, direction, types, |_, other_id| {
            if let Some(node) = self.node_at(other_id) {
                out.push(node.to_record());
            }
            Ok::<(), ()>(())
        });
        out
    }

    fn all_node_property_keys(&self) -> Vec<String> {
        let mut keys = BTreeSet::new();
        for node in self.iter_node_refs() {
            for key in node.properties().keys() {
                keys.insert(key.to_string());
            }
        }
        keys.into_iter().collect()
    }

    // ---------- Overrides: traversal (direct adjacency) ----------

    fn all_relationship_property_keys(&self) -> Vec<String> {
        let mut keys = BTreeSet::new();
        for rel in self.iter_rel_refs() {
            for key in rel.properties().keys() {
                keys.insert(key.to_string());
            }
        }
        keys.into_iter().collect()
    }

    fn label_property_keys(&self, label: &str) -> Vec<String> {
        let mut keys = BTreeSet::new();

        if let Some(ids) = self.nodes_by_label.get(label) {
            for id in ids.iter() {
                if let Some(node) = self.node_at(id) {
                    for key in node.properties().keys() {
                        keys.insert(key.to_string());
                    }
                }
            }
        }

        keys.into_iter().collect()
    }

    fn rel_type_property_keys(&self, rel_type: &str) -> Vec<String> {
        let mut keys = BTreeSet::new();

        if let Some(ids) = self.relationships_by_type.get(rel_type) {
            for id in ids.iter() {
                if let Some(rel) = self.rel_at(id) {
                    for key in rel.properties().keys() {
                        keys.insert(key.to_string());
                    }
                }
            }
        }

        keys.into_iter().collect()
    }

    fn find_nodes_by_property(
        &self,
        label: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<NodeRecord>
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.scan_nodes_by_property(label, key, value);
        }

        self.ensure_node_property_index(key);
        if label.is_none() {
            self.ensure_any_scope_node_property_index(key);
        }
        let indexes = self.indexes_read();

        match label {
            Some(label) => {
                let Some(ids) = indexes.node_properties.scoped_ids_for(label, key, value) else {
                    return Vec::new();
                };
                ids.iter()
                    .filter_map(|id| self.node_at(id).map(|r| r.to_record()))
                    .collect()
            }
            None => indexes
                .node_properties
                .ids_for(key, value)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| self.node_at(id).map(|r| r.to_record()))
                .collect(),
        }
    }

    fn find_node_ids_by_property(
        &self,
        label: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<NodeId>
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.scan_node_ids_by_property(label, key, value);
        }

        self.ensure_node_property_index(key);
        if label.is_none() {
            self.ensure_any_scope_node_property_index(key);
        }
        let indexes = self.indexes_read();

        match label {
            Some(label) => indexes
                .node_properties
                .scoped_ids_for(label, key, value)
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
            None => indexes
                .node_properties
                .ids_for(key, value)
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
        }
    }
    // ---------- Overrides: schema introspection ----------

    fn find_relationships_by_property(
        &self,
        rel_type: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<RelationshipRecord>
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.scan_relationships_by_property(rel_type, key, value);
        }

        self.ensure_relationship_property_index(key);
        if rel_type.is_none() {
            self.ensure_any_scope_relationship_property_index(key);
        }
        let indexes = self.indexes_read();

        match rel_type {
            Some(rel_type) => {
                let Some(ids) = indexes
                    .relationship_properties
                    .scoped_ids_for(rel_type, key, value)
                else {
                    return Vec::new();
                };
                ids.iter()
                    .filter_map(|id| self.rel_at(id).map(|r| r.to_record()))
                    .collect()
            }
            None => indexes
                .relationship_properties
                .ids_for(key, value)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| self.rel_at(id).map(|r| r.to_record()))
                .collect(),
        }
    }

    fn find_relationship_ids_by_property(
        &self,
        rel_type: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<RelationshipId>
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.scan_relationship_ids_by_property(rel_type, key, value);
        }

        self.ensure_relationship_property_index(key);
        if rel_type.is_none() {
            self.ensure_any_scope_relationship_property_index(key);
        }
        let indexes = self.indexes_read();

        match rel_type {
            Some(rel_type) => indexes
                .relationship_properties
                .scoped_ids_for(rel_type, key, value)
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
            None => indexes
                .relationship_properties
                .ids_for(key, value)
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
        }
    }

    fn node_exists_with_label_and_property(
        &self,
        label: &str,
        key: &str,
        value: &PropertyValue,
    ) -> bool
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.any_node_by_property(label, key, value);
        }

        self.ensure_node_property_index(key);
        let indexes = self.indexes_read();
        indexes
            .node_properties
            .scoped_ids_for(label, key, value)
            .map(|ids| !ids.is_empty())
            .unwrap_or(false)
    }

    fn relationship_exists_with_type_and_property(
        &self,
        rel_type: &str,
        key: &str,
        value: &PropertyValue,
    ) -> bool
    where
        Self: Sized,
    {
        if PropertyIndexKey::from_value(value).is_none() {
            return self.any_relationship_by_property(rel_type, key, value);
        }

        self.ensure_relationship_property_index(key);
        let indexes = self.indexes_read();
        indexes
            .relationship_properties
            .scoped_ids_for(rel_type, key, value)
            .map(|ids| !ids.is_empty())
            .unwrap_or(false)
    }
}

impl InMemoryGraph {
    /// [`GraphStorageMut::delete_relationship`], leaving the edits of long
    /// adjacency lists in `pending` when it is given. The caller flushes
    /// them before anything reads adjacency again.
    fn delete_relationship_deferring(
        &mut self,
        rel_id: RelationshipId,
        pending: Option<&mut PendingAdjRemovals>,
    ) -> bool {
        let applied = match self.take_rel(rel_id) {
            Some(rel) => {
                if let Some(sink) = &self.deleted_sink {
                    sink.relationship_deleted(&rel);
                }
                self.on_relationship_deleted(&rel, pending);
                true
            }
            None => false,
        };
        if applied {
            self.emit(|| MutationEvent::DeleteRelationship { rel_id });
        }
        applied
    }

    /// [`GraphStorageMut::detach_delete_node`] with the far ends' long
    /// lists left in `pending`. The node's own lists are read and cleared
    /// first, so an entry still pending there (from an earlier node of the
    /// same batch) only names a relationship that is gone, which the
    /// delete skips, as it would have found the entry removed.
    fn detach_delete_deferring(
        &mut self,
        node_id: NodeId,
        pending: &mut PendingAdjRemovals,
    ) -> bool {
        if !self.has_node_at(node_id) {
            return false;
        }

        let rel_ids = self.incident_relationship_ids(node_id);

        // We deliberately fire per-relationship DeleteRelationship events
        // here and a DetachDeleteNode event at the end. A WAL replayer that
        // sees DetachDeleteNode can ignore the preceding DeleteRelationship
        // events, or replay them, after which the DetachDeleteNode is a
        // no-op on the remaining (now-empty) node. The
        // emit-before-delete choice costs one extra event per mutation but
        // keeps the replay contract simple: "apply every event in order".
        //
        // The node's own lists go first: each removal below then edits only
        // the other endpoint's list (a self-loop finds nothing on either side).
        self.clear_adjacency(node_id);
        for rel_id in rel_ids {
            let _ = self.delete_relationship_deferring(rel_id, Some(pending));
        }

        if self.delete_node(node_id) {
            self.emit(|| MutationEvent::DetachDeleteNode { node_id });
            true
        } else {
            false
        }
    }
}

impl GraphStorageMut for InMemoryGraph {
    fn try_create_node(
        &mut self,
        labels: Vec<String>,
        properties: Properties,
    ) -> Option<NodeRecord> {
        let (id, idx) = self.try_reserve_next_node_slot()?;
        let labels = Self::normalize_labels(labels);

        let node = NodeRecord {
            id,
            labels,
            properties,
        };

        self.put_node_at_slot(idx, &node);

        self.on_node_created(&node);

        // Reserving the slot also grew both adjacency tables to cover
        // this id.

        self.emit(|| MutationEvent::CreateNode {
            id,
            labels: node.labels.to_strings(),
            properties: node.properties.clone(),
        });

        Some(node)
    }

    fn create_relationship(
        &mut self,
        src: NodeId,
        dst: NodeId,
        rel_type: &str,
        properties: Properties,
    ) -> Option<RelationshipRecord> {
        if !self.has_node_at(src) || !self.has_node_at(dst) {
            return None;
        }

        let trimmed = rel_type.trim();
        if trimmed.is_empty() {
            return None;
        }

        let (id, idx) = self.try_reserve_next_rel_slot()?;
        let rel = RelationshipRecord {
            id,
            src,
            dst,
            rel_type: trimmed.into(),
            properties,
        };

        self.put_rel_at_slot(idx, &rel);
        self.on_relationship_created(&rel);

        self.emit(|| MutationEvent::CreateRelationship {
            id,
            src,
            dst,
            rel_type: rel.rel_type.to_string(),
            properties: rel.properties.clone(),
        });

        Some(rel)
    }

    fn set_node_property(&mut self, node_id: NodeId, key: String, value: PropertyValue) -> bool {
        let Some(old) = self.edit_node_property(node_id, &key, PropEdit::Set(&value)) else {
            return false;
        };
        self.on_node_property_set(node_id, &key, old.as_ref(), &value);

        self.emit(|| MutationEvent::SetNodeProperty {
            node_id,
            key: key.clone(),
            value: value.clone(),
        });

        true
    }

    fn remove_node_property(&mut self, node_id: NodeId, key: &str) -> bool {
        let Some(removed) = self
            .edit_node_property(node_id, key, PropEdit::Remove)
            .flatten()
        else {
            return false;
        };

        self.on_node_property_removed(node_id, key, &removed);

        self.emit(|| MutationEvent::RemoveNodeProperty {
            node_id,
            key: key.to_string(),
        });

        true
    }

    fn add_node_label(&mut self, node_id: NodeId, label: &str) -> bool {
        let label = label.trim();
        if label.is_empty() {
            return false;
        }

        let applied = match self.node_at(node_id) {
            Some(node) if node.has_label(label) => return false,
            Some(_) => self
                .update_node(node_id, |node| node.labels.push(label))
                .is_some(),
            None => false,
        };
        if applied {
            self.on_node_label_added(node_id, label);
            self.emit(|| MutationEvent::AddNodeLabel {
                node_id,
                label: label.to_string(),
            });
        }
        applied
    }

    fn remove_node_label(&mut self, node_id: NodeId, label: &str) -> bool {
        let applied = self
            .node_at(node_id)
            .is_some_and(|node| node.has_label(label))
            && self
                .update_node(node_id, |node| node.labels.retain(|l| l != label))
                .is_some();
        if applied {
            self.on_node_label_removed(node_id, label);
            self.emit(|| MutationEvent::RemoveNodeLabel {
                node_id,
                label: label.to_string(),
            });
        }
        applied
    }

    fn set_relationship_property(
        &mut self,
        rel_id: RelationshipId,
        key: String,
        value: PropertyValue,
    ) -> bool {
        let Some(old) = self.edit_rel_property(rel_id, &key, PropEdit::Set(&value)) else {
            return false;
        };
        self.on_relationship_property_set(rel_id, &key, old.as_ref(), &value);

        self.emit(|| MutationEvent::SetRelationshipProperty {
            rel_id,
            key: key.clone(),
            value: value.clone(),
        });

        true
    }

    fn remove_relationship_property(&mut self, rel_id: RelationshipId, key: &str) -> bool {
        let Some(removed) = self
            .edit_rel_property(rel_id, key, PropEdit::Remove)
            .flatten()
        else {
            return false;
        };

        self.on_relationship_property_removed(rel_id, key, &removed);

        self.emit(|| MutationEvent::RemoveRelationshipProperty {
            rel_id,
            key: key.to_string(),
        });

        true
    }

    fn delete_relationship(&mut self, rel_id: RelationshipId) -> bool {
        self.delete_relationship_deferring(rel_id, None)
    }

    fn delete_relationships(&mut self, rel_ids: &[RelationshipId]) -> usize {
        let mut pending = PendingAdjRemovals::new();
        let deleted = rel_ids
            .iter()
            .filter(|&&rel_id| self.delete_relationship_deferring(rel_id, Some(&mut pending)))
            .count();
        self.flush_adjacency_removals(&mut pending);
        deleted
    }

    fn delete_node(&mut self, node_id: NodeId) -> bool {
        if !self.has_node_at(node_id) {
            return false;
        }

        if self.has_incident_relationships(node_id) {
            return false;
        }

        let node = match self.take_node(node_id) {
            Some(node) => node,
            None => return false,
        };

        if let Some(sink) = &self.deleted_sink {
            sink.node_deleted(&node);
        }
        self.on_node_deleted(&node);

        // take_node already cleared the node's adjacency lists.

        self.emit(|| MutationEvent::DeleteNode { node_id });

        true
    }

    fn detach_delete_node(&mut self, node_id: NodeId) -> bool {
        self.detach_delete_nodes(&[node_id]) == 1
    }

    fn detach_delete_nodes(&mut self, node_ids: &[NodeId]) -> usize {
        let mut pending = PendingAdjRemovals::new();
        let deleted = node_ids
            .iter()
            .filter(|&&node_id| self.detach_delete_deferring(node_id, &mut pending))
            .count();
        self.flush_adjacency_removals(&mut pending);
        deleted
    }

    fn clear(&mut self) {
        // Keep the recorder across clear so observers can see the Clear
        // event plus whatever follows. Matches WAL semantics where the log
        // is the source of truth across a truncation.
        let recorder = self.recorder.take();
        let deleted_sink = self.deleted_sink.take();
        *self = Self::default();
        self.recorder = recorder;
        self.deleted_sink = deleted_sink;
        self.emit(|| MutationEvent::Clear);
    }

    fn create_index(
        &mut self,
        request: IndexRequest,
        if_not_exists: bool,
    ) -> Result<CreateIndexOutcome, CreateIndexError> {
        // `register_index` only needs `&self` (catalog + property-index
        // registries are behind RwLock for interior mutability), but we
        // keep the trait signature `&mut self` so downstream callers can
        // hold a uniquely-owned mutation lock. That mirrors the pattern
        // used for property writes elsewhere in this impl.
        self.register_index(request, if_not_exists)
    }

    fn drop_index(
        &mut self,
        name: &str,
        if_exists: bool,
    ) -> Result<DropIndexOutcome, DropIndexError> {
        self.drop_named_index(name, if_exists)
    }

    fn create_constraint(
        &mut self,
        request: ConstraintRequest,
        if_not_exists: bool,
    ) -> Result<CreateConstraintOutcome, CreateConstraintError> {
        self.register_constraint(request, if_not_exists)
    }

    fn drop_constraint(
        &mut self,
        name: &str,
        if_exists: bool,
    ) -> Result<DropConstraintOutcome, DropConstraintError> {
        self.drop_named_constraint(name, if_exists)
    }
}

/// Move up to `max` of the ascending `ids` into `out` and leave `cursor`
/// just past the last one moved. Returns whether any id is left.
#[inline]
fn fill_page(
    mut ids: impl Iterator<Item = NodeId>,
    cursor: &mut u64,
    max: usize,
    out: &mut Vec<NodeId>,
) -> bool {
    for id in ids.by_ref().take(max) {
        out.push(id);
        *cursor = id.saturating_add(1);
    }
    ids.next().is_some()
}