kglite 0.16.9

//! UNIQUE-constraint declaration and enforcement on `DirGraph`.
//!
//! # The structure *is* the constraint
//!
//! `DirGraph::unique_indices` maps `(node_type, [properties]) -> tuple value ->
//! the one NodeIndex occupying it`. Unlike the other secondary index kinds,
//! which map a value to a `Vec<NodeIndex>`, the unique index holds a single
//! occupant — so "is this tuple taken by somebody else?" is one hash lookup,
//! and there is no separate constraint bookkeeping that could drift out of sync
//! with the index that answers the question.
//!
//! # Cost per write
//!
//! A graph that declares no constraint (the default, and every graph saved
//! before this feature) pays one `HashMap::is_empty` per write and allocates
//! nothing. A type that *does* carry constraints pays a scan of the
//! declaration keys, one property read per constrained property, and one hash
//! probe per constraint — never a full scan. A create allocates only the claim
//! vector; a property write additionally composes before/after maps over the
//! type's constrained properties ([`DirGraph::plan_property_write`]).
//!
//! # NULL semantics
//!
//! A node is *exempt* from a unique constraint unless **every** property in the
//! tuple reads as present and non-null — matching Neo4j, where a uniqueness
//! constraint does not apply to nodes missing the property, and a composite
//! constraint requires the whole tuple. So many nodes may share "no email"
//! while `email` is still UNIQUE. [`DirGraph::unique_claims`] encodes this by
//! producing no claim for an incomplete tuple.
//!
//! # Persistence
//!
//! `unique_indices` is `#[serde(skip)]`; the declaration list
//! `unique_constraint_keys` is persisted and replayed by
//! [`DirGraph::rebuild_unique_indices_from_keys`] on load — the same
//! persist-keys/rebuild-on-load pattern the property, range, and composite
//! indexes use. The rebuild re-verifies the constraint for free, since building
//! a single-occupant map *is* the duplicate check.

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

use petgraph::graph::NodeIndex;

use super::indexes::PropertyReader;
use super::DirGraph;
use crate::datatypes::values::Value;
use crate::graph::constraints::EntityKind;
use crate::graph::constraints::{
    normalize_properties, ConstraintKind, ConstraintResult, ConstraintViolation, NamedConstraint,
    UniqueConstraintKey,
};
use crate::graph::property_types::{self, DeclaredType};
use crate::graph::schema::{
    CompositeValue, NodeSchemaDefinition, SchemaDefinition, PROVISIONAL_KEY,
};

/// One node's occupancy of one declared unique tuple.
///
/// Produced by [`DirGraph::unique_claims`] before a write, checked by
/// [`DirGraph::check_unique_claims`], and redeemed by
/// [`DirGraph::commit_unique_claims`] once the node exists. Deriving the claim
/// once and reusing it for check + commit is what keeps the two from disagreeing
/// about *which* tuple was validated.
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct UniqueClaim {
    pub key: UniqueConstraintKey,
    pub value: CompositeValue,
}

/// The unique-index bookkeeping a property write owes once it has been applied:
/// release what the node used to occupy, claim what it now occupies. Produced by
/// [`DirGraph::plan_property_write`] *before* the write (so the write can be
/// rejected without touching storage) and redeemed by
/// [`DirGraph::apply_property_write_plan`] after.
#[derive(Debug, Clone, Default, PartialEq)]
pub(crate) struct PropertyWritePlan {
    pub release: Vec<UniqueClaim>,
    pub claim: Vec<UniqueClaim>,
}

impl DirGraph {
    // ── Declaration ──

    /// Whether *any* unique constraint is declared on this graph — the write
    /// path's fast-out.
    #[inline]
    pub(crate) fn has_unique_constraints(&self) -> bool {
        !self.unique_indices.is_empty()
    }

    /// Whether any unique constraint is declared on `node_type` — the gate the
    /// bulk fold uses to decide whether it owes occupancy bookkeeping at all.
    #[inline]
    pub(crate) fn type_has_unique_constraints(&self, node_type: &str) -> bool {
        self.has_unique_constraints() && self.unique_indices.keys().any(|(nt, _)| nt == node_type)
    }

    /// The unique tuples `node_idx` occupies **as stored**.
    ///
    /// The read half of incremental occupancy maintenance: the bulk fold takes
    /// this before a batch (what to release) and after it (what to claim), so
    /// the answer never depends on what a row *asked* for — a
    /// `conflict_handling` mode that skipped or merged the write is reflected
    /// automatically.
    ///
    /// Reads through [`DirGraph::read_indexed`] and applies the same
    /// incomplete-tuple exemption [`Self::build_unique_index`] does — here via
    /// [`Self::unique_claims`], there via `read_complete_tuple` — so a folded
    /// occupancy and a rebuilt one cannot disagree about which tuple a node holds.
    pub(crate) fn stored_unique_claims(
        &mut self,
        node_type: &str,
        node_idx: NodeIndex,
    ) -> Vec<UniqueClaim> {
        let properties: Vec<String> = {
            let mut names: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
            for (nt, tuple) in self.unique_indices.keys() {
                if nt == node_type {
                    names.extend(tuple.iter().map(String::as_str));
                }
            }
            names.into_iter().map(str::to_string).collect()
        };
        if properties.is_empty() {
            return Vec::new();
        }
        let values: HashMap<String, Option<Value>> = properties
            .into_iter()
            .map(|property| {
                let reader = self.property_reader(node_type, &property);
                let value = self.read_indexed(&reader, node_idx);
                (property, value)
            })
            .collect();
        self.unique_claims(node_type, |property| {
            values.get(property).cloned().flatten()
        })
    }

    /// Whether `(node_type, properties)` is declared UNIQUE. Property order is
    /// irrelevant — `(a, b)` and `(b, a)` are the same constraint.
    pub fn has_unique_constraint(&self, node_type: &str, properties: &[String]) -> bool {
        self.find_unique_key(node_type, properties).is_some()
    }

    /// Every declared unique constraint, as `(node_type, properties)`, sorted —
    /// `unique_indices` is a `HashMap`, so an unsorted listing would vary
    /// between runs. Backs `SHOW CONSTRAINTS`.
    pub fn list_unique_constraints(&self) -> Vec<UniqueConstraintKey> {
        let mut all: Vec<UniqueConstraintKey> = self.unique_indices.keys().cloned().collect();
        all.sort();
        all
    }

    /// The stored declaration key matching `(node_type, properties)` up to
    /// property order, if declared.
    fn find_unique_key(
        &self,
        node_type: &str,
        properties: &[String],
    ) -> Option<&UniqueConstraintKey> {
        let wanted = normalize_properties(properties);
        self.unique_indices
            .keys()
            .find(|(nt, props)| nt == node_type && normalize_properties(props) == wanted)
    }

    /// Declare a UNIQUE constraint on `(node_type, properties)` and build its
    /// index from live data. Returns the number of distinct tuples indexed.
    ///
    /// Fails with [`ConstraintViolation::preexisting`] when the existing data
    /// already contains a duplicate — declaring a constraint the data violates
    /// would otherwise install a constraint that silently lies about the rows
    /// already present. Nothing is installed on failure.
    ///
    /// Idempotent: re-declaring an existing constraint rebuilds it rather than
    /// erroring, so `CREATE CONSTRAINT ... IF NOT EXISTS` and a reload both work.
    pub(crate) fn create_unique_constraint(
        &mut self,
        node_type: &str,
        properties: &[&str],
    ) -> ConstraintResult<usize> {
        if properties.is_empty() {
            // A constraint over no properties would claim one global slot and
            // reject the type's second node. Reject the declaration instead.
            return Err(Box::new(ConstraintViolation::preexisting(
                ConstraintKind::Unique,
                node_type,
                Vec::new(),
                0,
                Vec::new(),
            )));
        }
        let owned: Vec<String> = properties.iter().map(|p| (*p).to_string()).collect();
        // Re-declaring replaces the previous spelling of the same constraint,
        // so the stored key reflects the latest declaration order.
        if let Some(existing) = self.find_unique_key(node_type, &owned).cloned() {
            self.remove_unique_declaration(&existing);
        }

        let key: UniqueConstraintKey = (node_type.to_string(), owned.clone());
        let (index, duplicates, sample) = self.build_unique_index(node_type, &owned);
        if duplicates > 0 {
            return Err(Box::new(ConstraintViolation::preexisting(
                self.unique_kind_for(node_type, &owned),
                node_type,
                owned,
                duplicates,
                sample,
            )));
        }

        let count = index.len();
        self.unique_indices.insert(key.clone(), index);
        self.unique_constraint_keys.push(key);
        Ok(count)
    }

    /// Drop a declared unique constraint. Returns whether one was removed.
    pub(crate) fn drop_unique_constraint(
        &mut self,
        node_type: &str,
        properties: &[String],
    ) -> bool {
        match self.find_unique_key(node_type, properties).cloned() {
            Some(key) => {
                self.remove_unique_declaration(&key);
                true
            }
            None => false,
        }
    }

    /// Forget a declaration in both the live index and the persisted key list.
    fn remove_unique_declaration(&mut self, key: &UniqueConstraintKey) {
        self.unique_indices.remove(key);
        self.unique_constraint_keys.retain(|stored| stored != key);
    }

    /// Drop every unique constraint declared on `node_type`. Used when the type
    /// itself goes away, so a later type of the same name does not inherit a
    /// constraint whose index refers to deleted nodes.
    pub fn drop_unique_constraints_for_type(&mut self, node_type: &str) -> usize {
        let keys: Vec<UniqueConstraintKey> = self
            .unique_indices
            .keys()
            .filter(|(nt, _)| nt == node_type)
            .cloned()
            .collect();
        for key in &keys {
            self.remove_unique_declaration(key);
        }
        keys.len()
    }

    /// Report `NODE KEY` when the tuple is unique *and* every property in it is
    /// required — which is what a node key is — and `UNIQUE` otherwise. Derived
    /// rather than stored, so the declarations cannot drift apart.
    ///
    /// The type's declared primary key satisfies this by construction. The
    /// second arm is what lets `CREATE CONSTRAINT … IS NODE KEY` report itself
    /// honestly: KGLite serves that statement as uniqueness plus presence, and
    /// there is only one primary-key slot per type, so a node key declared
    /// through DDL is not the primary key and would otherwise report as a plain
    /// `UNIQUE` violation.
    pub(crate) fn unique_kind_for(&self, node_type: &str, properties: &[String]) -> ConstraintKind {
        if matches!(self.primary_key_for(node_type), Some(pk) if properties.len() == 1 && properties[0] == pk)
        {
            return ConstraintKind::NodeKey;
        }
        let required = self.required_property_names(node_type);
        if !properties.is_empty()
            && properties
                .iter()
                .all(|property| required.contains(&property.as_str()))
        {
            return ConstraintKind::NodeKey;
        }
        ConstraintKind::Unique
    }

    /// Scan `node_type` and build the single-occupant map for `properties`.
    /// Returns `(index, duplicate_tuple_count, first_duplicate_sample)`.
    ///
    /// First occupant wins a contested tuple, so the returned index is always
    /// internally consistent even when the data is not — that is what lets the
    /// load path stay non-fatal (see
    /// [`Self::rebuild_unique_indices_from_keys`]).
    pub(super) fn build_unique_index(
        &mut self,
        node_type: &str,
        properties: &[String],
    ) -> (HashMap<CompositeValue, NodeIndex>, usize, Vec<Value>) {
        let readers: Vec<PropertyReader> = properties
            .iter()
            .map(|property| self.property_reader(node_type, property))
            .collect();

        let mut index: HashMap<CompositeValue, NodeIndex> = HashMap::new();
        let mut duplicates = 0usize;
        let mut sample: Vec<Value> = Vec::new();

        if let Some(node_indices) = self.type_indices.get(node_type) {
            for idx in node_indices.iter() {
                let Some(values) = self.read_complete_tuple(&readers, idx) else {
                    // Incomplete tuple — exempt, per the NULL semantics.
                    continue;
                };
                let composite = CompositeValue(values);
                if let Some(_occupant) = index.get(&composite) {
                    duplicates += 1;
                    if sample.is_empty() {
                        sample = composite.0.clone();
                    }
                    continue;
                }
                index.insert(composite, idx);
            }
        }

        (index, duplicates, sample)
    }

    /// Read every property of a constraint tuple off one node, returning `None`
    /// as soon as one is absent or null — an incomplete tuple is exempt.
    fn read_complete_tuple(
        &self,
        readers: &[PropertyReader],
        idx: NodeIndex,
    ) -> Option<Vec<Value>> {
        let mut values = Vec::with_capacity(readers.len());
        for reader in readers {
            match self.read_indexed(reader, idx) {
                Some(Value::Null) | None => return None,
                Some(value) => values.push(value),
            }
        }
        Some(values)
    }

    // ── Write-path enforcement ──

    /// The unique tuples a node of `node_type` would occupy, given a reader for
    /// its property values. `read` is called with the **user-facing** property
    /// name, matching how the constraint was declared; the caller decides where
    /// the value comes from (a pending CREATE's property map, a node already in
    /// the graph, a bulk row).
    ///
    /// Returns an empty vector when the type declares no constraint, or when
    /// every declared tuple is incomplete on this node — so the common case
    /// allocates nothing beyond an empty `Vec`.
    pub(crate) fn unique_claims<F>(&self, node_type: &str, read: F) -> Vec<UniqueClaim>
    where
        F: Fn(&str) -> Option<Value>,
    {
        if !self.has_unique_constraints() {
            return Vec::new();
        }
        let mut claims = Vec::new();
        for key in self.unique_indices.keys() {
            if key.0 != node_type {
                continue;
            }
            let mut values = Vec::with_capacity(key.1.len());
            let mut complete = true;
            for property in &key.1 {
                match read(property) {
                    Some(Value::Null) | None => {
                        complete = false;
                        break;
                    }
                    Some(value) => values.push(value),
                }
            }
            if complete {
                claims.push(UniqueClaim {
                    key: key.clone(),
                    value: CompositeValue(values),
                });
            }
        }
        claims
    }

    /// Reject the write if any claim's tuple is already occupied by a different
    /// node. `holder` is the node being written, when it already exists — a SET
    /// that rewrites a property to its current value must not conflict with
    /// itself.
    pub(crate) fn check_unique_claims(
        &self,
        claims: &[UniqueClaim],
        holder: Option<NodeIndex>,
    ) -> ConstraintResult<()> {
        for claim in claims {
            let Some(index) = self.unique_indices.get(&claim.key) else {
                continue;
            };
            match index.get(&claim.value) {
                Some(occupant) if Some(*occupant) != holder => {
                    return Err(Box::new(ConstraintViolation::duplicate(
                        self.unique_kind_for(&claim.key.0, &claim.key.1),
                        claim.key.0.clone(),
                        claim.key.1.clone(),
                        claim.value.0.clone(),
                    )));
                }
                _ => {}
            }
        }
        Ok(())
    }

    /// Record `node_idx` as the occupant of each claimed tuple. Call **after**
    /// the node exists and [`Self::check_unique_claims`] passed.
    pub(crate) fn commit_unique_claims(&mut self, claims: &[UniqueClaim], node_idx: NodeIndex) {
        for claim in claims {
            if let Some(index) = self.unique_indices.get_mut(&claim.key) {
                index.insert(claim.value.clone(), node_idx);
            }
        }
    }

    /// Give up `node_idx`'s occupancy of each claimed tuple — the old-value half
    /// of a SET, so the vacated tuple becomes available again. Only removes an
    /// entry this node actually holds.
    pub(crate) fn release_unique_claims(&mut self, claims: &[UniqueClaim], node_idx: NodeIndex) {
        for claim in claims {
            if let Some(index) = self.unique_indices.get_mut(&claim.key) {
                if index.get(&claim.value) == Some(&node_idx) {
                    index.remove(&claim.value);
                }
            }
        }
    }

    /// Evict deleted nodes from every unique index of `node_type`. Without this
    /// a deleted node keeps its tuple reserved forever and a legitimate re-insert
    /// of the same value would be rejected.
    ///
    /// O(distinct tuples) per constraint, matching the shape the delete path
    /// already uses for the property and composite indexes.
    pub(crate) fn evict_unique_claims_for_nodes(
        &mut self,
        node_type: &str,
        deleted: &HashSet<NodeIndex>,
    ) {
        if !self.has_unique_constraints() {
            return;
        }
        for (key, index) in self.unique_indices.iter_mut() {
            if key.0 != node_type {
                continue;
            }
            index.retain(|_, occupant| !deleted.contains(occupant));
        }
    }

    /// What a property write must do to the unique indexes once it has been
    /// applied: give up the tuples the node used to occupy, take the tuples it
    /// now occupies.
    ///
    /// Empty on both sides when the type declares no unique constraint, which is
    /// the common case.
    ///
    /// A rejection is *also* parked on the graph by
    /// [`DirGraph::record_constraint_violation`], so callers whose error channel
    /// is a `String` (the Cypher `SET` / `REMOVE` tree) still surface a typed
    /// `ConstraintViolationError`. Recording here rather than at each call site
    /// keeps the violation attached to the code that produced it.
    pub(crate) fn plan_property_write(
        &mut self,
        node_type: &str,
        node_idx: NodeIndex,
        property: &str,
        new_value: Option<&Value>,
    ) -> ConstraintResult<PropertyWritePlan> {
        let planned = self.plan_property_write_uncaught(node_type, node_idx, property, new_value);
        if let Err(violation) = &planned {
            self.record_constraint_violation(violation.as_ref().clone());
        }
        planned
    }

    fn plan_property_write_uncaught(
        &mut self,
        node_type: &str,
        node_idx: NodeIndex,
        property: &str,
        new_value: Option<&Value>,
    ) -> ConstraintResult<PropertyWritePlan> {
        // A declared property type is a pure predicate on the incoming value:
        // no read-back, no claim bookkeeping, nothing to redeem afterwards. It
        // is checked *before* the uniqueness early-out below because that
        // early-out asks only about unique/NOT NULL declarations — a type that
        // declares a property type and nothing else has no `constrained`
        // properties, and folding the check in after the early-out would skip
        // it exactly on the graphs where it is the only constraint.
        //
        // `None` is REMOVE, and a null value is a SET-to-null: both leave the
        // property absent, which satisfies every declared type (the presence
        // question belongs to NOT NULL).
        if let Some(value) = new_value {
            self.check_property_type(node_type, property, value)?;
        }

        let constrained = self.constrained_properties(node_type);
        if constrained.is_empty() {
            return Ok(PropertyWritePlan::default());
        }

        // One read pass over every property any constraint on this type cares
        // about, so the composite tuples can be rebuilt without re-reading.
        let mut before: HashMap<String, Value> = HashMap::with_capacity(constrained.len());
        for name in &constrained {
            let reader = self.property_reader(node_type, name);
            if let Some(value) = self.read_indexed(&reader, node_idx) {
                if !matches!(value, Value::Null) {
                    before.insert(name.clone(), value);
                }
            }
        }

        // The same map with this write applied — `None` models REMOVE and a SET
        // to null identically, since a constraint treats absent and null alike.
        let mut after = before.clone();
        match new_value {
            Some(Value::Null) | None => {
                after.remove(property);
            }
            Some(value) => {
                after.insert(property.to_string(), value.clone());
            }
        }

        // NOT NULL is evaluated against the post-write state, so a SET-to-null or
        // a REMOVE of a required property is caught even though the property is
        // present beforehand.
        self.check_required_fields(node_type, |name| after.get(name).cloned())?;

        let release = self.unique_claims(node_type, |name| before.get(name).cloned());
        let claim = self.unique_claims(node_type, |name| after.get(name).cloned());
        self.check_unique_claims(&claim, Some(node_idx))?;
        Ok(PropertyWritePlan { release, claim })
    }

    /// Every property name any declared constraint on `node_type` reads —
    /// unique tuples plus required fields, deduplicated.
    fn constrained_properties(&self, node_type: &str) -> Vec<String> {
        let mut names: Vec<String> = Vec::new();
        if self.has_unique_constraints() {
            for (nt, properties) in self.unique_indices.keys() {
                if nt == node_type {
                    names.extend(properties.iter().cloned());
                }
            }
        }
        names.extend(
            self.required_property_names(node_type)
                .into_iter()
                .map(str::to_string),
        );
        // The provisional marker decides whether NOT NULL applies at all, so it
        // has to be readable by the same composed map.
        if !names.is_empty() {
            names.push(PROVISIONAL_KEY.to_string());
        }
        names.sort();
        names.dedup();
        names
    }

    /// Apply a plan's index bookkeeping after the write landed.
    pub(crate) fn apply_property_write_plan(
        &mut self,
        plan: &PropertyWritePlan,
        node_idx: NodeIndex,
    ) {
        self.release_unique_claims(&plan.release, node_idx);
        self.commit_unique_claims(&plan.claim, node_idx);
    }

    /// The violation a second claim on `claim` raises, for a caller that detects
    /// the collision itself rather than through the stored index — the bulk path
    /// rejecting a repeat inside one input batch, where neither row is in the
    /// graph yet so there is no occupant to collide with.
    pub(crate) fn unique_batch_conflict(&self, claim: &UniqueClaim) -> ConstraintViolation {
        ConstraintViolation::duplicate(
            self.unique_kind_for(&claim.key.0, &claim.key.1),
            claim.key.0.clone(),
            claim.key.1.clone(),
            claim.value.0.clone(),
        )
    }

    // ── NOT NULL (required fields) ──

    /// The properties `node_type` declares as required, via
    /// `define_schema({"nodes": {"T": {"required": [...]}}})`. Empty when the
    /// type declares none.
    pub(crate) fn required_fields_for(&self, node_type: &str) -> &[String] {
        self.schema_definition
            .as_ref()
            .and_then(|schema| schema.node_schemas.get(node_type))
            .map(|node| node.required_fields.as_slice())
            .unwrap_or(&[])
    }

    /// Whether `node_type` requires any property to be present — the
    /// write-path fast-out.
    ///
    /// True for a declared `primary_key` as well as for `required_fields`, since
    /// a primary key is unique **and** present (NODE KEY). A key on `id` counts
    /// like any other: `id` is resolved by every write path but can still be
    /// nulled explicitly — see [`Self::check_required_fields`].
    #[inline]
    pub(crate) fn has_required_fields(&self, node_type: &str) -> bool {
        !self.required_fields_for(node_type).is_empty() || self.primary_key_for(node_type).is_some()
    }

    /// Every property `node_type` requires to be present: the declared
    /// `required_fields` plus the primary key. Borrow-free so the caller can
    /// hold `&self` while reading values.
    fn required_property_names(&self, node_type: &str) -> Vec<&str> {
        let mut names: Vec<&str> = self
            .required_fields_for(node_type)
            .iter()
            .map(String::as_str)
            .collect();
        // A primary key is required by definition, `id` included: a write can
        // null it explicitly, exactly as it can null `title` — see
        // [`Self::check_required_fields`].
        if let Some(pk) = self.primary_key_for(node_type) {
            if !names.contains(&pk) {
                names.push(pk);
            }
        }
        names
    }

    /// Reject a write that leaves a declared-required property absent or null.
    ///
    /// `read` is called with each required property name and returns its value
    /// *as the write will leave it* — the caller composes pending values over
    /// stored ones, so a SET that nulls a required property is caught even
    /// though the property is present beforehand.
    ///
    /// # Structural fields: `type` is exempt, `id` and `title` are not
    ///
    /// `type` is the node's label rather than a value a write supplies, so no
    /// write can leave it absent and requiring it is a genuine no-op.
    ///
    /// `id` and `title` are **not** exempt, despite also being `NodeData`
    /// fields. Each write path resolves them before this check — CREATE
    /// auto-assigns an id and synthesizes a title from `name`/`title` or the
    /// label, and the bulk path falls back to the id column — so an omitted one
    /// reads as present and the requirement is satisfied. But both accept an
    /// *explicit* null (`CREATE (:T {title: null})`, `SET t.title = null`,
    /// `REMOVE t.title`, a null title cell in a batch), and the node that
    /// results genuinely carries a null. Skipping them here made
    /// `required: ["title"]` report itself through `SHOW CONSTRAINTS` as
    /// `NODE_PROPERTY_EXISTENCE` — and, with uniqueness alongside it, as
    /// `NODE_KEY` — while admitting exactly those writes: a constraint that
    /// reported success and enforced nothing, the one outcome this module
    /// refuses everywhere else (see `reject_structural_uniqueness`).
    ///
    /// # Provisional stubs are deferred, not exempt
    ///
    /// A write that carries `_provisional = true` is auto-vivification creating
    /// a placeholder for an edge endpoint whose real row has not arrived
    /// (`mutation::maintain::vivify_stubs`). Such a stub carries only its id by
    /// construction, so enforcing NOT NULL here would make `add_connections`
    /// fail on any edge list that mentions a node before its own row loads —
    /// i.e. it would break graph building outright.
    ///
    /// The escape hatch is the **existing promotion flow**, not an exemption
    /// flag: the stub is written, and the later `add_nodes` upsert that supplies
    /// the real row clears the `_provisional` marker
    /// (`mutation::batch::flush_chunk`) — and *that* write is a normal write, so
    /// it is fully enforced. A stub that is never promoted therefore never
    /// satisfies the constraint, and stays visible as one:
    /// `validate_schema()` reports it as a missing required field, and
    /// `purge_provisional_nodes()` (which the blueprint builder runs
    /// automatically) deletes it.
    ///
    /// Writing `_provisional = true` by hand therefore deliberately opts a node
    /// out of NOT NULL until it is promoted. The blueprint builder refuses a
    /// spec that declares `_provisional` as a user property
    /// (`blueprint::build`), which is where that would most plausibly happen by
    /// accident.
    pub(crate) fn check_required_fields<F>(&self, node_type: &str, read: F) -> ConstraintResult<()>
    where
        F: Fn(&str) -> Option<Value>,
    {
        // Iterated rather than collected: this runs once per row of every bulk
        // load into a constrained type, and `required_property_names`'s `Vec`
        // would be one heap allocation per row.
        let declared = self.required_fields_for(node_type);
        let primary_key = self
            .primary_key_for(node_type)
            .filter(|pk| !declared.iter().any(|field| field == pk));
        if declared.is_empty() && primary_key.is_none() {
            return Ok(());
        }
        if matches!(read(PROVISIONAL_KEY), Some(Value::Boolean(true))) {
            return Ok(());
        }
        for property in declared.iter().map(String::as_str).chain(primary_key) {
            if property == "type" {
                continue;
            }
            match read(property) {
                Some(Value::Null) | None => {
                    return Err(Box::new(ConstraintViolation::missing(
                        self.required_kind_for(node_type, property),
                        node_type,
                        property,
                    )));
                }
                Some(_) => {}
            }
        }
        Ok(())
    }

    /// `NODE KEY` when the required property is also the type's primary key,
    /// `NOT NULL` otherwise — so one declaration does not report itself under
    /// two different names.
    fn required_kind_for(&self, node_type: &str, property: &str) -> ConstraintKind {
        match self.primary_key_for(node_type) {
            Some(pk) if pk == property => ConstraintKind::NodeKey,
            _ => ConstraintKind::NotNull,
        }
    }

    // ── NOT NULL declaration ──

    /// Declare `property` NOT NULL on `node_type` — i.e. add it to the type's
    /// `required_fields`, the list [`Self::check_required_fields`] enforces on
    /// every write path.
    ///
    /// Returns the number of nodes of the type that were checked.
    ///
    /// Fails with [`ConstraintViolation::preexisting_missing`] when existing
    /// nodes have no value for the property, and installs nothing in that case —
    /// mirroring [`Self::create_unique_constraint`], because a constraint that
    /// silently exempts the rows already present is worse than a rejected
    /// declaration. Provisional stubs are skipped, matching the write-path rule:
    /// a stub is *deferred*, not exempt, and stays reportable via
    /// `validate_schema()`.
    ///
    /// Idempotent: re-declaring an existing requirement re-verifies it and
    /// changes nothing, so `IF NOT EXISTS` and a reload both work.
    pub(crate) fn create_not_null_constraint(
        &mut self,
        node_type: &str,
        property: &str,
    ) -> ConstraintResult<usize> {
        let (checked, missing) = self.count_missing_property(node_type, property);
        if missing > 0 {
            return Err(Box::new(ConstraintViolation::preexisting_missing(
                self.required_kind_for(node_type, property),
                node_type,
                property,
                missing,
            )));
        }
        self.ddl_not_null_constraints
            .insert((node_type.to_string(), property.to_string()));
        self.require_property(node_type, property);
        Ok(checked)
    }

    /// Add `property` to `node_type`'s `required_fields`, keeping the list
    /// sorted and duplicate-free. The storage half of a presence declaration,
    /// shared by the DDL entry point and by [`Self::reapply_ddl_not_null`].
    fn require_property(&mut self, node_type: &str, property: &str) {
        let required = &mut self.node_schema_mut(node_type).required_fields;
        required.push(property.to_string());
        required.sort();
        required.dedup();
    }

    /// Re-add every DDL-declared presence constraint to the schema now installed.
    ///
    /// `required_fields` lives inside the `SchemaDefinition`, so installing a
    /// schema replaces the list a `CREATE CONSTRAINT ... IS NOT NULL` wrote into
    /// — silently un-enforcing it. The uniqueness half has no such problem: its
    /// index lives outside the schema and `set_schema` withdraws only what the
    /// *outgoing schema* declared. This restores the symmetry, so a DDL
    /// constraint is withdrawn only by `DROP CONSTRAINT`.
    pub(crate) fn reapply_ddl_not_null(&mut self) {
        if self.ddl_not_null_constraints.is_empty() {
            return;
        }
        let declared: Vec<(String, String)> =
            self.ddl_not_null_constraints.iter().cloned().collect();
        for (node_type, property) in declared {
            self.require_property(&node_type, &property);
        }
    }

    /// Withdraw a NOT NULL declaration. Reports whether one was removed.
    pub(crate) fn drop_not_null_constraint(&mut self, node_type: &str, property: &str) -> bool {
        // Forget the DDL provenance first, or the next schema install would
        // reinstate what was just dropped.
        self.ddl_not_null_constraints
            .remove(&(node_type.to_string(), property.to_string()));
        let Some(node) = self
            .schema_definition
            .as_mut()
            .and_then(|schema| schema.node_schemas.get_mut(node_type))
        else {
            return false;
        };
        let before = node.required_fields.len();
        node.required_fields.retain(|field| field != property);
        before != node.required_fields.len()
    }

    /// Whether `property` is declared NOT NULL on `node_type`. A primary key
    /// counts: it is required by definition.
    pub(crate) fn has_not_null_constraint(&self, node_type: &str, property: &str) -> bool {
        self.required_property_names(node_type).contains(&property)
    }

    /// Every declared presence constraint, as `(node_type, property)` sorted.
    /// Backs `SHOW CONSTRAINTS` together with
    /// [`Self::list_unique_constraints`].
    pub(crate) fn list_not_null_constraints(&self) -> Vec<(String, String)> {
        let Some(schema) = self.schema_definition.as_ref() else {
            return Vec::new();
        };
        let mut all: Vec<(String, String)> = schema
            .node_schemas
            .keys()
            .flat_map(|node_type| {
                self.required_property_names(node_type)
                    .into_iter()
                    .map(move |property| (node_type.clone(), property.to_string()))
            })
            .collect();
        all.sort();
        all.dedup();
        all
    }

    /// `(nodes_checked, nodes_missing_the_property)` for `node_type`.
    /// Provisional stubs are skipped — see [`Self::check_required_fields`].
    fn count_missing_property(&mut self, node_type: &str, property: &str) -> (usize, usize) {
        // `type` is the node's label rather than a supplied value, so nothing
        // can be missing. `id`/`title` are read through `read_indexed` like any
        // other property — they are resolved on every write path but can be
        // explicitly nulled, so existing nulls must block the declaration.
        if property == "type" {
            let checked = self
                .type_indices
                .get(node_type)
                .map_or(0, |nodes| nodes.iter().count());
            return (checked, 0);
        }
        let reader = self.property_reader(node_type, property);
        let provisional = self.property_reader(node_type, PROVISIONAL_KEY);
        let Some(node_indices) = self.type_indices.get(node_type) else {
            return (0, 0);
        };
        let indices: Vec<NodeIndex> = node_indices.iter().collect();
        let mut missing = 0usize;
        for idx in &indices {
            if matches!(
                self.read_indexed(&provisional, *idx),
                Some(Value::Boolean(true))
            ) {
                continue;
            }
            match self.read_indexed(&reader, *idx) {
                Some(Value::Null) | None => missing += 1,
                Some(_) => {}
            }
        }
        (indices.len(), missing)
    }

    /// The mutable `NodeSchemaDefinition` for `node_type`, creating the schema
    /// container and the per-type entry when they do not exist yet.
    ///
    /// Declaring a constraint on a graph with no `define_schema` call is
    /// legitimate — `CREATE CONSTRAINT` is exactly that — so this materializes
    /// the schema rather than refusing. It deliberately does **not** go through
    /// `set_schema`, which installs the unique constraints a *whole* schema
    /// implies; presence constraints install no index, and the caller owns the
    /// uniqueness half of a NODE KEY.
    fn node_schema_mut(&mut self, node_type: &str) -> &mut NodeSchemaDefinition {
        self.schema_definition
            .get_or_insert_with(SchemaDefinition::new)
            .node_schemas
            .entry(node_type.to_string())
            .or_default()
    }

    // ── Property type ──
    //
    // The declaration store (`ddl_property_type_constraints`) *is* the
    // constraint, the way `unique_indices` is for uniqueness: there is no
    // second copy inside the schema, so no schema install can withdraw one and
    // no `reapply_*` pass is needed to put it back.
    //
    // Cost per write has the same shape as the other kinds: one
    // `BTreeMap::is_empty`, one further map probe for a declaring type, one
    // predicate call per declared property. No allocation on any path.
    //
    // Provisional stubs need no exemption here, unlike NOT NULL. A stub carries
    // only its id, and an absent property satisfies a type constraint, so
    // enforcement never blocks auto-vivification.

    /// Whether the graph declares any property-type constraint — the
    /// write-path fast-out.
    #[inline]
    pub(crate) fn has_property_type_constraints(&self) -> bool {
        !self.ddl_property_type_constraints.is_empty()
    }

    /// Whether `node_type` declares any property type. The per-type companion
    /// to [`Self::has_property_type_constraints`], for a caller deciding once
    /// per batch whether a row gate is needed at all.
    #[inline]
    pub(crate) fn type_has_property_type_constraints(&self, node_type: &str) -> bool {
        self.ddl_property_type_constraints.contains_key(node_type)
    }

    pub(crate) fn property_type_for(
        &self,
        node_type: &str,
        property: &str,
    ) -> Option<DeclaredType> {
        self.ddl_property_type_constraints
            .get(node_type)?
            .get(property)
            .copied()
    }

    /// Reject one property write whose value has the wrong type.
    ///
    /// `value` is the value *the write will store*. Null passes — a type
    /// constraint is not an existence constraint (see
    /// [`crate::graph::property_types`]) — so a caller that nulls a property
    /// only has to consult NOT NULL.
    pub(crate) fn check_property_type(
        &self,
        node_type: &str,
        property: &str,
        value: &Value,
    ) -> ConstraintResult<()> {
        if !self.has_property_type_constraints() {
            return Ok(());
        }
        let Some(declared) = self.property_type_for(node_type, property) else {
            return Ok(());
        };
        Self::type_violation(declared, node_type, property, value)
    }

    /// Reject a whole row's worth of property writes.
    ///
    /// `read` is called with each *constrained* property name and returns its
    /// value as the write will leave it — the same contract
    /// [`Self::check_required_fields`] uses, so a choke point that already
    /// composes pending values over stored ones can reuse the closure it has.
    /// Only declared properties are read, so an unconstrained type costs one
    /// map probe and calls `read` zero times.
    pub(crate) fn check_property_types<F>(&self, node_type: &str, read: F) -> ConstraintResult<()>
    where
        F: Fn(&str) -> Option<Value>,
    {
        if !self.has_property_type_constraints() {
            return Ok(());
        }
        let Some(declared) = self.ddl_property_type_constraints.get(node_type) else {
            return Ok(());
        };
        for (property, expected) in declared {
            // An absent property passes, exactly as a null one does.
            let Some(value) = read(property) else {
                continue;
            };
            Self::type_violation(*expected, node_type, property, &value)?;
        }
        Ok(())
    }

    /// The violation `value` raises against `declared`, or `Ok` when it
    /// satisfies it. The single place the predicate is consulted, so the
    /// write-path and row-path checks cannot disagree about what passes.
    fn type_violation(
        declared: DeclaredType,
        node_type: &str,
        property: &str,
        value: &Value,
    ) -> ConstraintResult<()> {
        if declared.accepts(value) {
            return Ok(());
        }
        Err(Box::new(ConstraintViolation::type_mismatch(
            node_type,
            property,
            declared.name(),
            property_types::value_type_name(value),
        )))
    }

    /// Declare `property` on `node_type` to hold only `declared` values.
    ///
    /// Returns the number of nodes of the type that were checked.
    ///
    /// Fails with [`ConstraintViolation::preexisting_type_mismatch`] when
    /// existing nodes hold a value of another type, and installs nothing in
    /// that case, mirroring [`Self::create_not_null_constraint`].
    ///
    /// Idempotent: re-declaring the same type re-verifies it and changes
    /// nothing. Declaring a *different* type for a property that already has
    /// one replaces it, and only when the existing data satisfies the new type
    /// — the caller owns the "a constraint already exists here" policy
    /// (`IF NOT EXISTS` and the already-declared rejection both live in the DDL
    /// executor, as they do for the other kinds).
    pub(crate) fn create_property_type_constraint(
        &mut self,
        node_type: &str,
        property: &str,
        declared: DeclaredType,
    ) -> ConstraintResult<usize> {
        let (checked, violations, sample) =
            self.count_type_violations(node_type, property, declared);
        if violations > 0 {
            return Err(Box::new(ConstraintViolation::preexisting_type_mismatch(
                node_type,
                property,
                declared.name(),
                sample.unwrap_or("a value of another type"),
                violations,
            )));
        }
        self.ddl_property_type_constraints
            .entry(node_type.to_string())
            .or_default()
            .insert(property.to_string(), declared);
        Ok(checked)
    }

    /// Withdraw a property-type declaration. Reports whether one was removed.
    ///
    /// Removes the node type's entry once its last declaration goes, so
    /// [`Self::has_property_type_constraints`] returns to `false` — and the
    /// write path to its zero-cost early-out — once every constraint is
    /// dropped.
    pub(crate) fn drop_property_type_constraint(
        &mut self,
        node_type: &str,
        property: &str,
    ) -> bool {
        let Some(declared) = self.ddl_property_type_constraints.get_mut(node_type) else {
            return false;
        };
        let removed = declared.remove(property).is_some();
        if declared.is_empty() {
            self.ddl_property_type_constraints.remove(node_type);
        }
        removed
    }

    /// Every declared property-type constraint as `(node_type, property, type)`,
    /// in deterministic order. Backs `SHOW CONSTRAINTS` alongside
    /// [`Self::list_unique_constraints`] and
    /// [`Self::list_not_null_constraints`].
    pub(crate) fn list_property_type_constraints(&self) -> Vec<(String, String, DeclaredType)> {
        self.ddl_property_type_constraints
            .iter()
            .flat_map(|(node_type, declared)| {
                declared
                    .iter()
                    .map(move |(property, kind)| (node_type.clone(), property.clone(), *kind))
            })
            .collect()
    }

    /// `(nodes_checked, nodes_violating, one_offending_type_name)` for a
    /// candidate declaration.
    ///
    /// `type` is the node's label rather than a stored value: it is a string on
    /// every node by construction, and no write can make it anything else. So a
    /// `STRING` declaration on it is satisfied for free, and any other
    /// declaration is refused here rather than installed as a constraint that
    /// would enforce nothing on the write path.
    fn count_type_violations(
        &mut self,
        node_type: &str,
        property: &str,
        declared: DeclaredType,
    ) -> (usize, usize, Option<&'static str>) {
        if property == "type" {
            let checked = self
                .type_indices
                .get(node_type)
                .map_or(0, |nodes| nodes.iter().count());
            let label_is_accepted = declared.accepts(&Value::String(node_type.to_string()));
            let violations = if label_is_accepted { 0 } else { checked };
            return (checked, violations, Some("STRING"));
        }
        let reader = self.property_reader(node_type, property);
        let Some(node_indices) = self.type_indices.get(node_type) else {
            return (0, 0, None);
        };
        let indices: Vec<NodeIndex> = node_indices.iter().collect();
        let mut violations = 0usize;
        let mut sample = None;
        for idx in &indices {
            // Absent and null both satisfy a type constraint, so neither blocks
            // the declaration — the presence question belongs to NOT NULL.
            let Some(value) = self.read_indexed(&reader, *idx) else {
                continue;
            };
            if !declared.accepts(&value) {
                violations += 1;
                sample.get_or_insert_with(|| property_types::value_type_name(&value));
            }
        }
        (indices.len(), violations, sample)
    }

    // ── Named constraints ──

    /// Record the name its author gave a constraint, so
    /// `DROP CONSTRAINT <name>` can find it. Replaces any previous registration
    /// of the same name.
    pub(crate) fn register_constraint_name(&mut self, name: &str, constraint: NamedConstraint) {
        self.constraint_names.insert(name.to_string(), constraint);
    }

    pub(crate) fn constraint_by_name(&self, name: &str) -> Option<&NamedConstraint> {
        self.constraint_names.get(name)
    }

    /// Forget a name. Called when its constraint is dropped.
    pub(crate) fn forget_constraint_name(&mut self, name: &str) {
        self.constraint_names.remove(name);
    }

    /// The name registered for `(entity, node_type, properties)`, if the
    /// constraint was declared with one. A node label and a connection type can
    /// share a name, so the entity is part of the lookup — without it a
    /// relationship constraint on `KNOWS` could answer for a node one. Property order is irrelevant, matching constraint
    /// identity. Lets `SHOW CONSTRAINTS` report the author's name.
    ///
    /// Several names can point at one tuple — `CREATE CONSTRAINT u … IS UNIQUE`
    /// and `CREATE CONSTRAINT nn … IS NOT NULL` on the same property are two
    /// declarations that `SHOW CONSTRAINTS` reports as a single `NODE_KEY` row.
    /// Picking the first match out of a `HashMap` made *which* name that row
    /// carried depend on hash order, so the same graph reported different names
    /// before and after a save/load round-trip. The lowest name in sort order
    /// wins instead: arbitrary, but stable across runs, processes and reloads,
    /// which is what a listing an operator reads during a migration needs.
    pub(crate) fn name_for_constraint(
        &self,
        entity: EntityKind,
        node_type: &str,
        properties: &[String],
    ) -> Option<&str> {
        let wanted = normalize_properties(properties);
        self.constraint_names
            .iter()
            .filter(|(_, declared)| {
                declared.entity == entity
                    && declared.node_type == node_type
                    && normalize_properties(&declared.properties) == wanted
            })
            .map(|(name, _)| name.as_str())
            .min()
    }

    /// Drop every registered name whose constraint is no longer declared.
    ///
    /// The registry is a lookup aid, not the source of truth, and several paths
    /// remove a constraint without going through `DROP CONSTRAINT` —
    /// [`Self::drop_unique_constraints_for_type`] when a type is deleted, and
    /// `set_schema` replacing a schema that declared one. Without this, those
    /// names would leak into every subsequent save and `DROP CONSTRAINT <name>`
    /// would claim to drop something that had already gone.
    pub(crate) fn prune_constraint_names(&mut self) {
        if self.constraint_names.is_empty() {
            return;
        }
        let live: Vec<String> = self
            .constraint_names
            .iter()
            .filter(|(_, declared)| self.constraint_is_declared(declared))
            .map(|(name, _)| name.clone())
            .collect();
        self.constraint_names.retain(|name, _| live.contains(name));
    }

    /// Whether the declaration a registered name points at is still in force.
    fn constraint_is_declared(&self, declared: &NamedConstraint) -> bool {
        // Relationship declarations live in their own stores, and neither
        // uniqueness spelling can be installed on one — a name pointing at such
        // a constraint can only have come from a hand-edited file, and is
        // pruned rather than trusted.
        if declared.entity == EntityKind::Relationship {
            return match declared.kind {
                ConstraintKind::NotNull => declared.properties.iter().all(|property| {
                    self.has_rel_not_null_constraint(&declared.node_type, property)
                }),
                ConstraintKind::PropertyType => declared.properties.iter().all(|property| {
                    self.rel_property_type_for(&declared.node_type, property)
                        .is_some()
                }),
                ConstraintKind::Unique | ConstraintKind::NodeKey => false,
            };
        }
        match declared.kind {
            ConstraintKind::Unique => {
                self.has_unique_constraint(&declared.node_type, &declared.properties)
            }
            ConstraintKind::NotNull => declared
                .properties
                .iter()
                .all(|property| self.has_not_null_constraint(&declared.node_type, property)),
            // A NODE KEY is the conjunction, so it survives only while both
            // halves do — a dropped uniqueness half demotes it, and reporting it
            // as still declared would overstate what is enforced.
            ConstraintKind::NodeKey => {
                self.has_unique_constraint(&declared.node_type, &declared.properties)
                    && declared
                        .properties
                        .iter()
                        .all(|property| self.has_not_null_constraint(&declared.node_type, property))
            }
            // The declared *type* is not part of the name registration, so a
            // re-declaration that changes the type keeps the name — which is
            // right: the name points at the property, and the property is still
            // constrained.
            ConstraintKind::PropertyType => declared.properties.iter().all(|property| {
                self.property_type_for(&declared.node_type, property)
                    .is_some()
            }),
        }
    }

    // ── Load-time rebuild ──

    /// Rebuild every persisted unique constraint from live data. Returns the
    /// violations found in the loaded data, if any.
    ///
    /// **Deliberately non-fatal.** A `.kgl` file must always open: refusing to
    /// load a graph because its data violates a constraint would strand the
    /// user's data behind the very tool they need to fix it. So a contested
    /// tuple keeps its first occupant, the constraint stays declared and live
    /// for all *subsequent* writes, and the duplicates are returned for the
    /// caller to surface. A violating file can only come from a write path that
    /// predates enforcement or one that bypasses it (the RDF loaders), not from
    /// a normal write.
    pub(crate) fn rebuild_unique_indices_from_keys(&mut self) -> Vec<ConstraintViolation> {
        let keys: Vec<UniqueConstraintKey> = std::mem::take(&mut self.unique_constraint_keys);
        let mut violations = Vec::new();
        for key in &keys {
            let (index, duplicates, sample) = self.build_unique_index(&key.0, &key.1);
            if duplicates > 0 {
                violations.push(ConstraintViolation::preexisting(
                    self.unique_kind_for(&key.0, &key.1),
                    key.0.clone(),
                    key.1.clone(),
                    duplicates,
                    sample,
                ));
            }
            self.unique_indices.insert(key.clone(), index);
        }
        self.unique_constraint_keys = keys;
        violations
    }

    /// Recompute the unique-occupancy maps of every declared constraint on
    /// `node_types`, from live data.
    ///
    /// The statement-rollback counterpart of
    /// [`Self::rebuild_unique_indices_from_keys`]. `unique_indices` is parked by
    /// `rollback::swap_data_scale`, so a journal rollback leaves the *failed
    /// statement's* occupancy in place while the data underneath is restored;
    /// the claims the statement added or released have to be recomputed, or the
    /// graph keeps a phantom occupant (a permanent spurious
    /// `ConstraintViolationError` for a value nothing holds) or has silently
    /// released one (a real duplicate admitted on the next write).
    ///
    /// Scoped to the types the replay touched, so an untouched or unconstrained
    /// type costs nothing and an unconstrained graph returns immediately.
    /// Duplicates are not reported: the restored data is the pre-statement data,
    /// which the write path already accepted, so a contested tuple here could
    /// only be a pre-existing violation the load path already surfaced.
    pub(super) fn rebuild_unique_indices_for_types(&mut self, node_types: &HashSet<String>) {
        if node_types.is_empty() || self.unique_indices.is_empty() {
            return;
        }
        let keys: Vec<UniqueConstraintKey> = self
            .unique_indices
            .keys()
            .filter(|(node_type, _)| node_types.contains(node_type))
            .cloned()
            .collect();
        for key in keys {
            let (index, _duplicates, _sample) = self.build_unique_index(&key.0, &key.1);
            self.unique_indices.insert(key, index);
        }
    }

    /// Re-scan live data and report every unique-constraint violation currently
    /// present. The on-demand counterpart of the load-time rebuild, for callers
    /// that want to audit a graph filled by a path that bypasses enforcement.
    pub fn verify_unique_constraints(&mut self) -> Vec<ConstraintViolation> {
        let keys: Vec<UniqueConstraintKey> = self.unique_indices.keys().cloned().collect();
        let mut violations = Vec::new();
        for key in &keys {
            let (_, duplicates, sample) = self.build_unique_index(&key.0, &key.1);
            if duplicates > 0 {
                violations.push(ConstraintViolation::preexisting(
                    self.unique_kind_for(&key.0, &key.1),
                    key.0.clone(),
                    key.1.clone(),
                    duplicates,
                    sample,
                ));
            }
        }
        violations
    }
}

#[cfg(test)]
mod not_null_declaration_tests {
    use super::*;
    use crate::graph::schema::NodeData;
    use crate::graph::storage::GraphWrite;

    /// `Person` nodes carrying whichever properties each row supplies, so a row
    /// can deliberately omit one.
    fn person_graph(rows: &[(i64, &str, Option<&str>)]) -> DirGraph {
        let mut graph = DirGraph::new();
        for (id, name, email) in rows {
            let mut props =
                HashMap::from([("name".to_string(), Value::String((*name).to_string()))]);
            if let Some(email) = email {
                props.insert("email".to_string(), Value::String((*email).to_string()));
            }
            let node = NodeData::new(
                Value::UniqueId(*id as u32),
                Value::String((*name).to_string()),
                "Person".to_string(),
                props,
                &mut graph.interner,
            );
            let idx = graph.graph.add_node(node);
            graph
                .type_indices
                .entry_or_default("Person".to_string())
                .push(idx);
        }
        graph
    }

    #[test]
    fn declaring_not_null_on_clean_data_installs_and_enforces_it() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c")), (2, "Bob", Some("b@b.c"))]);
        assert_eq!(
            graph.create_not_null_constraint("Person", "email").unwrap(),
            2
        );
        assert!(graph.has_not_null_constraint("Person", "email"));
        assert!(graph.has_required_fields("Person"));

        let violation = graph
            .check_required_fields("Person", |_| None)
            .expect_err("a write with no email must be rejected");
        assert_eq!(violation.kind, ConstraintKind::NotNull);
        assert!(violation.to_string().contains("'email'"));

        graph
            .check_required_fields("Person", |name| {
                (name == "email").then(|| Value::String("c@b.c".to_string()))
            })
            .expect("a write with an email must pass");
    }

    #[test]
    fn declaring_not_null_against_missing_values_is_rejected_and_changes_nothing() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c")), (2, "Bob", None)]);
        let violation = graph
            .create_not_null_constraint("Person", "email")
            .expect_err("one node has no email");

        assert!(violation.is_declaration_failure());
        let message = violation.to_string();
        assert!(message.contains("cannot declare"), "{message}");
        assert!(message.contains("1 existing node"), "{message}");

        // Nothing installed: the graph must be as permissive as before.
        assert!(!graph.has_not_null_constraint("Person", "email"));
        assert!(!graph.has_required_fields("Person"));
    }

    #[test]
    fn declaring_not_null_is_idempotent() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
        graph.create_not_null_constraint("Person", "email").unwrap();
        graph.create_not_null_constraint("Person", "email").unwrap();
        assert_eq!(
            graph.list_not_null_constraints(),
            vec![("Person".to_string(), "email".to_string())]
        );
    }

    #[test]
    fn dropping_not_null_stops_enforcement_and_reports_whether_it_existed() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
        graph.create_not_null_constraint("Person", "email").unwrap();

        assert!(graph.drop_not_null_constraint("Person", "email"));
        assert!(!graph.has_not_null_constraint("Person", "email"));
        graph
            .check_required_fields("Person", |_| None)
            .expect("no requirement remains");

        // A second drop has nothing to remove.
        assert!(!graph.drop_not_null_constraint("Person", "email"));
        assert!(!graph.drop_not_null_constraint("Person", "nonexistent"));
    }

    /// A provisional stub carries only its id by construction, so counting it as
    /// a missing value would make declaring NOT NULL impossible on any graph
    /// built from an edge list. Deferred, not exempt — see
    /// `check_required_fields`.
    #[test]
    fn provisional_stubs_do_not_block_a_declaration() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
        let stub = NodeData::new(
            Value::UniqueId(2),
            Value::String("stub".to_string()),
            "Person".to_string(),
            HashMap::from([(PROVISIONAL_KEY.to_string(), Value::Boolean(true))]),
            &mut graph.interner,
        );
        let idx = graph.graph.add_node(stub);
        graph
            .type_indices
            .entry_or_default("Person".to_string())
            .push(idx);

        graph
            .create_not_null_constraint("Person", "email")
            .expect("the stub must not block the declaration");
        assert!(graph.has_not_null_constraint("Person", "email"));
    }

    /// `type` is the node's label rather than a supplied value, so requiring it
    /// is satisfied by construction — nothing a write does can leave it absent.
    #[test]
    fn declaring_not_null_on_the_label_field_is_satisfied() {
        let mut graph = person_graph(&[(1, "Alice", None)]);
        assert_eq!(
            graph.create_not_null_constraint("Person", "type").unwrap(),
            1
        );
        graph
            .check_required_fields("Person", |_| None)
            .expect("type is the label and always present");
    }

    /// `id`/`title` are `NodeData` fields too, but a write can null them
    /// explicitly, so they are enforced rather than exempt. Every write path
    /// resolves them before the check (CREATE auto-assigns an id and synthesizes
    /// a title), which is what the reader models here: a resolved value passes,
    /// an unresolved one is the explicit-null case and must be rejected.
    /// Skipping them made `required: ["title"]` report itself through
    /// `SHOW CONSTRAINTS` while admitting `CREATE (:T {title: null})`.
    #[test]
    fn declaring_not_null_on_id_or_title_is_enforced_against_an_explicit_null() {
        for property in ["id", "title"] {
            let mut graph = person_graph(&[(1, "Alice", None)]);
            assert_eq!(
                graph
                    .create_not_null_constraint("Person", property)
                    .unwrap(),
                1
            );

            graph
                .check_required_fields("Person", |name| {
                    (name == property).then(|| Value::String("resolved".to_string()))
                })
                .expect("a write that carries the structural field passes");

            let violation = graph
                .check_required_fields("Person", |_| None)
                .expect_err("an explicitly nulled structural field must be rejected");
            assert_eq!(violation.kind, ConstraintKind::NotNull);
            assert!(violation.to_string().contains(property), "{violation}");
        }
    }

    #[test]
    fn declaring_not_null_needs_no_prior_define_schema() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
        assert!(graph.get_schema().is_none());
        graph.create_not_null_constraint("Person", "email").unwrap();
        assert!(graph.get_schema().is_some());
    }

    /// A tuple that is both unique and fully required *is* a node key, so it
    /// must report itself as one — otherwise `CREATE CONSTRAINT … IS NODE KEY`
    /// would raise `UNIQUE` violations for a constraint the user declared as a
    /// node key.
    #[test]
    fn unique_plus_required_reports_as_node_key() {
        let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        assert_eq!(
            graph.unique_kind_for("Person", &["email".to_string()]),
            ConstraintKind::Unique
        );

        graph.create_not_null_constraint("Person", "email").unwrap();
        assert_eq!(
            graph.unique_kind_for("Person", &["email".to_string()]),
            ConstraintKind::NodeKey
        );
    }

    /// A primary key is unique **and** present, and `id` is no more exempt from
    /// the presence half than a `required: ["id"]` declaration is (see
    /// `declaring_not_null_on_id_or_title_is_enforced_against_an_explicit_null`).
    /// Exempting it made `primary_key: "id"` admit `CREATE (:T {id: null})`
    /// while `primary_key: "email"` rejected the same shape.
    #[test]
    fn a_primary_key_on_id_is_required_like_any_other_primary_key() {
        use crate::graph::schema::{NodeSchemaDefinition, SchemaDefinition, SchemaInstall};

        for pk in ["id", "email"] {
            let mut graph = person_graph(&[(1, "Alice", Some("a@b.c"))]);
            let mut schema = SchemaDefinition::new();
            schema.add_node_schema(
                "Person".to_string(),
                NodeSchemaDefinition {
                    primary_key: Some(pk.to_string()),
                    ..Default::default()
                },
            );
            graph
                .set_schema(schema, SchemaInstall::Merge)
                .expect("schema install");

            assert!(graph.has_required_fields("Person"), "pk = {pk}");
            assert!(graph.has_not_null_constraint("Person", pk), "pk = {pk}");

            // The write paths resolve the key before checking, so a write that
            // carries it passes...
            graph
                .check_required_fields("Person", |name| {
                    (name == pk).then(|| Value::String("resolved".to_string()))
                })
                .unwrap_or_else(|e| panic!("pk = {pk}: {e}"));

            // ...and an explicit null (`CREATE (:Person {id: null})`) does not.
            let violation = graph
                .check_required_fields("Person", |_| None)
                .expect_err(&format!("pk = {pk}: a null primary key must be rejected"));
            assert_eq!(violation.kind, ConstraintKind::NodeKey, "pk = {pk}");
            assert!(violation.to_string().contains(pk), "{violation}");
        }
    }
}

#[cfg(test)]
mod constraint_name_registry_tests {
    use super::*;

    fn named(kind: ConstraintKind, properties: &[&str]) -> NamedConstraint {
        NamedConstraint {
            kind,
            entity: EntityKind::Node,
            node_type: "Person".to_string(),
            properties: properties.iter().map(|p| (*p).to_string()).collect(),
        }
    }

    #[test]
    fn a_registered_name_resolves_to_its_declaration() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        graph.register_constraint_name(
            "person_email_unique",
            named(ConstraintKind::Unique, &["email"]),
        );

        assert_eq!(
            graph.constraint_by_name("person_email_unique"),
            Some(&named(ConstraintKind::Unique, &["email"]))
        );
        assert_eq!(graph.constraint_by_name("nope"), None);
        assert_eq!(
            graph.name_for_constraint(EntityKind::Node, "Person", &["email".to_string()]),
            Some("person_email_unique")
        );
    }

    /// Constraint identity ignores property order, so a name lookup must too.
    #[test]
    fn a_name_resolves_regardless_of_property_order() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["first", "last"])
            .unwrap();
        graph.register_constraint_name(
            "full_name",
            named(ConstraintKind::Unique, &["first", "last"]),
        );
        assert_eq!(
            graph.name_for_constraint(
                EntityKind::Node,
                "Person",
                &["last".to_string(), "first".to_string()]
            ),
            Some("full_name")
        );
    }

    #[test]
    fn forgetting_a_name_leaves_the_constraint_in_force() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        graph.register_constraint_name("c", named(ConstraintKind::Unique, &["email"]));

        graph.forget_constraint_name("c");
        assert_eq!(graph.constraint_by_name("c"), None);
        assert!(graph.has_unique_constraint("Person", &["email".to_string()]));
    }

    /// The registry is a lookup aid, so a name whose declaration went away by
    /// another route must not survive into the next save.
    #[test]
    fn pruning_discards_names_whose_declaration_is_gone() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        graph.register_constraint_name("live", named(ConstraintKind::Unique, &["email"]));
        graph.register_constraint_name("dangling", named(ConstraintKind::Unique, &["nickname"]));

        graph.prune_constraint_names();
        assert!(graph.constraint_by_name("live").is_some());
        assert!(
            graph.constraint_by_name("dangling").is_none(),
            "a name with no declaration behind it must be pruned"
        );

        // Dropping the type's constraints strands the surviving name too.
        graph.drop_unique_constraints_for_type("Person");
        graph.prune_constraint_names();
        assert!(graph.constraint_by_name("live").is_none());
    }

    /// A NODE KEY is uniqueness *and* presence, so losing either half must
    /// demote it rather than leave the name claiming both are enforced.
    #[test]
    fn a_node_key_name_is_pruned_when_either_half_goes() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        graph.create_not_null_constraint("Person", "email").unwrap();
        graph.register_constraint_name("person_key", named(ConstraintKind::NodeKey, &["email"]));

        graph.prune_constraint_names();
        assert!(graph.constraint_by_name("person_key").is_some());

        graph.drop_not_null_constraint("Person", "email");
        graph.prune_constraint_names();
        assert!(
            graph.constraint_by_name("person_key").is_none(),
            "a NODE KEY without its presence half is no longer a NODE KEY"
        );
    }

    #[test]
    fn populate_index_keys_prunes_and_sorts_deterministically() {
        let mut graph = DirGraph::new();
        graph
            .create_unique_constraint("Person", &["email"])
            .unwrap();
        graph
            .create_unique_constraint("Person", &["nickname"])
            .unwrap();
        graph.register_constraint_name("dangling", named(ConstraintKind::Unique, &["absent"]));

        graph.populate_index_keys();
        assert!(graph.constraint_by_name("dangling").is_none());
        // Sorted, so a graph carrying constraints saves reproducible bytes.
        let mut sorted = graph.unique_constraint_keys.clone();
        sorted.sort();
        assert_eq!(graph.unique_constraint_keys, sorted);
    }
}

#[cfg(test)]
mod property_type_declaration_tests {
    use super::*;
    use crate::graph::schema::NodeData;
    use crate::graph::storage::GraphWrite;

    /// `Person` nodes carrying exactly the properties each row supplies, so a
    /// row can hold a value of the wrong type — or omit the property.
    fn person_graph(rows: &[&[(&str, Value)]]) -> DirGraph {
        let mut graph = DirGraph::new();
        for (row, properties) in rows.iter().enumerate() {
            let props: HashMap<String, Value> = properties
                .iter()
                .map(|(name, value)| ((*name).to_string(), value.clone()))
                .collect();
            let node = NodeData::new(
                Value::UniqueId(row as u32 + 1),
                Value::String(format!("person-{row}")),
                "Person".to_string(),
                props,
                &mut graph.interner,
            );
            let idx = graph.graph.add_node(node);
            graph
                .type_indices
                .entry_or_default("Person".to_string())
                .push(idx);
        }
        graph
    }

    /// A reader that answers with one property's value and nothing else, the
    /// shape a write-path choke point composes.
    fn supplying(property: &'static str, value: Value) -> impl Fn(&str) -> Option<Value> {
        move |name| (name == property).then(|| value.clone())
    }

    #[test]
    fn declaring_a_type_on_clean_data_installs_and_enforces_it() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))], &[("age", Value::Int64(9))]]);
        assert_eq!(
            graph
                .create_property_type_constraint("Person", "age", DeclaredType::Integer)
                .unwrap(),
            2
        );
        assert!(graph.has_property_type_constraints());
        assert_eq!(
            graph.property_type_for("Person", "age"),
            Some(DeclaredType::Integer)
        );
        assert_eq!(
            graph.list_property_type_constraints(),
            vec![(
                "Person".to_string(),
                "age".to_string(),
                DeclaredType::Integer
            )]
        );

        // A conforming write passes, through both the single-value and the
        // row-shaped check.
        graph
            .check_property_type("Person", "age", &Value::Int64(1))
            .expect("an integer satisfies an INTEGER declaration");
        graph
            .check_property_types("Person", supplying("age", Value::Int64(1)))
            .expect("an integer satisfies an INTEGER declaration");

        // A non-conforming one is rejected, by both, with the same facts.
        for violation in [
            graph
                .check_property_type("Person", "age", &Value::String("41".to_string()))
                .expect_err("a string must not satisfy an INTEGER declaration"),
            graph
                .check_property_types("Person", supplying("age", Value::String("41".to_string())))
                .expect_err("a string must not satisfy an INTEGER declaration"),
        ] {
            assert_eq!(violation.kind, ConstraintKind::PropertyType);
            assert!(!violation.is_declaration_failure());
            let message = violation.to_string();
            assert!(message.contains("'age'"), "{message}");
            assert!(message.contains("INTEGER"), "{message}");
            assert!(message.contains("STRING"), "{message}");
        }
    }

    /// A constraint that exempted the rows already present would report success
    /// and enforce less than it claims, so the declaration is refused instead.
    #[test]
    fn a_declaration_is_refused_when_existing_data_violates_it() {
        let mut graph = person_graph(&[
            &[("age", Value::Int64(41))],
            &[("age", Value::String("nine".to_string()))],
            &[("age", Value::String("ten".to_string()))],
        ]);
        let violation = graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .expect_err("existing strings must block an INTEGER declaration");

        assert!(violation.is_declaration_failure());
        let message = violation.to_string();
        assert!(message.contains("2 existing nodes"), "{message}");
        assert!(message.contains("not INTEGER"), "{message}");
        assert!(message.contains("STRING"), "{message}");

        // Nothing was installed, so the write path keeps its early-out.
        assert!(!graph.has_property_type_constraints());
        assert_eq!(graph.property_type_for("Person", "age"), None);
    }

    /// Neo4j semantics: a type constraint is not an existence constraint.
    #[test]
    fn null_and_absent_values_block_neither_a_declaration_nor_a_write() {
        let mut graph = person_graph(&[
            &[("age", Value::Int64(41))],
            &[("age", Value::Null)],
            &[("name", Value::String("no age at all".to_string()))],
        ]);
        assert_eq!(
            graph
                .create_property_type_constraint("Person", "age", DeclaredType::Integer)
                .unwrap(),
            3
        );
        graph
            .check_property_type("Person", "age", &Value::Null)
            .expect("null satisfies every declared type");
        graph
            .check_property_types("Person", |_| Some(Value::Null))
            .expect("null satisfies every declared type");
        graph
            .check_property_types("Person", |_| None)
            .expect("an absent property satisfies every declared type");
    }

    /// `UniqueId` is an id's compact encoding, not a second numeric type — so
    /// the single most obvious declaration anyone writes must not be refused by
    /// the graph's own ids.
    #[test]
    fn an_integer_declaration_accepts_auto_assigned_ids() {
        let mut graph = person_graph(&[&[("name", Value::String("Alice".to_string()))]]);
        assert_eq!(
            graph
                .create_property_type_constraint("Person", "id", DeclaredType::Integer)
                .unwrap(),
            1
        );
        graph
            .check_property_type("Person", "id", &Value::UniqueId(7))
            .expect("an auto-assigned id is an INTEGER");
        graph
            .check_property_type("Person", "id", &Value::Int64(7))
            .expect("an explicit numeric id is an INTEGER");
        graph
            .check_property_type("Person", "id", &Value::Float64(7.0))
            .expect_err("a float id is not an INTEGER");
    }

    /// A graph that declares nothing must not read a single value, and neither
    /// must a type that declares nothing — the reader panics if either does.
    #[test]
    fn an_unconstrained_type_reads_nothing() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))]]);
        assert!(!graph.has_property_type_constraints());
        graph
            .check_property_types("Person", |name| {
                panic!("read {name} on an unconstrained graph")
            })
            .expect("a graph with no type constraints checks nothing");

        graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .unwrap();
        graph
            .check_property_types("Company", |name| {
                panic!("read {name} on an unconstrained type")
            })
            .expect("a type with no type constraints checks nothing");
        graph
            .check_property_type("Person", "nickname", &Value::Int64(1))
            .expect("an unconstrained property is not checked");
    }

    #[test]
    fn dropping_the_last_declaration_restores_the_zero_cost_early_out() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))]]);
        graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .unwrap();
        graph
            .create_property_type_constraint("Person", "name", DeclaredType::String)
            .unwrap();

        assert!(graph.drop_property_type_constraint("Person", "age"));
        assert!(
            !graph.drop_property_type_constraint("Person", "age"),
            "dropping twice must report the second as a no-op"
        );
        assert!(!graph.drop_property_type_constraint("Company", "age"));
        assert!(
            graph.has_property_type_constraints(),
            "the surviving declaration keeps enforcement on"
        );

        assert!(graph.drop_property_type_constraint("Person", "name"));
        assert!(
            !graph.has_property_type_constraints(),
            "the empty node-type entry must go with its last declaration, or the \
             write path never returns to its early-out"
        );
        graph
            .check_property_type("Person", "age", &Value::String("x".to_string()))
            .expect("a dropped constraint enforces nothing");
    }

    /// Re-declaring is how a user corrects a type; it must still be verified
    /// against the data, and must not half-install on failure.
    #[test]
    fn redeclaring_replaces_the_type_only_when_the_data_agrees() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))]]);
        graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .unwrap();

        graph
            .create_property_type_constraint("Person", "age", DeclaredType::String)
            .expect_err("the stored integer must block a STRING re-declaration");
        assert_eq!(
            graph.property_type_for("Person", "age"),
            Some(DeclaredType::Integer),
            "a refused re-declaration must leave the old one in force"
        );

        graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .expect("re-declaring the same type is idempotent");
        assert_eq!(
            graph.property_type_for("Person", "age"),
            Some(DeclaredType::Integer)
        );
    }

    /// The label is a string on every node by construction, so a non-STRING
    /// declaration on it can never be enforced by a write and is refused at
    /// declaration time rather than installed as a no-op.
    #[test]
    fn declaring_a_type_on_the_label_field_accepts_only_string() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))]]);
        let violation = graph
            .create_property_type_constraint("Person", "type", DeclaredType::Integer)
            .expect_err("the label is never an integer");
        assert!(violation.is_declaration_failure());
        assert!(!graph.has_property_type_constraints());

        assert_eq!(
            graph
                .create_property_type_constraint("Person", "type", DeclaredType::String)
                .unwrap(),
            1
        );
    }

    /// The name registry is a lookup aid, so a name whose type declaration went
    /// away must not survive into the next save.
    #[test]
    fn a_property_type_name_is_pruned_when_its_declaration_goes() {
        let mut graph = person_graph(&[&[("age", Value::Int64(41))]]);
        graph
            .create_property_type_constraint("Person", "age", DeclaredType::Integer)
            .unwrap();
        graph.register_constraint_name(
            "person_age_typed",
            NamedConstraint {
                kind: ConstraintKind::PropertyType,
                entity: EntityKind::Node,
                node_type: "Person".to_string(),
                properties: vec!["age".to_string()],
            },
        );

        graph.prune_constraint_names();
        assert!(graph.constraint_by_name("person_age_typed").is_some());

        graph.drop_property_type_constraint("Person", "age");
        graph.prune_constraint_names();
        assert!(
            graph.constraint_by_name("person_age_typed").is_none(),
            "a name with no declaration behind it must be pruned"
        );
    }
}