hawdb-storage 0.8.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Graph record and snapshot constraints shared by storage mutation and recovery.
//!
//! These are internal validation kernels. Root storage retains transaction
//! orchestration, concrete out-of-core scans, and the WAL/publication boundary.

use crate::{cow::CowSegmentedMap, wal::WalOp, NodeId, NodeRecord, RelId, RelRecord};
use hawdb_core::{
    Catalog, ConstraintSubject, HawDBError, LabelId, PropertyType, RelTypeId, Result,
    SchemaObjectState, TableKind, Value,
};
use std::collections::{BTreeMap, BTreeSet};

/// Graph records a commit's ops can affect. Delta validation is sound because
/// pre-existing records were already valid: only touched records can carry a
/// newly introduced violation.
#[derive(Debug, Default)]
pub struct TouchedRecords {
    pub nodes: BTreeSet<NodeId>,
    pub relationships: BTreeSet<RelId>,
}

/// Collects touched records for delta validation. Returns `None` when any op
/// is not pure graph data — schema/DDL changes (new constraints, property
/// types) apply to pre-existing records and therefore require full
/// validation.
pub fn wal_ops_touched_records(ops: &[WalOp]) -> Option<TouchedRecords> {
    let mut touched = TouchedRecords::default();
    for op in ops {
        collect_touched_records(op, &mut touched)?;
    }
    Some(touched)
}

fn collect_touched_records(op: &WalOp, touched: &mut TouchedRecords) -> Option<()> {
    match op {
        WalOp::CreateNode { id, .. }
        | WalOp::SetNodeProperty { id, .. }
        | WalOp::DeleteNode { id } => {
            touched.nodes.insert(*id);
        }
        WalOp::CreateRelationship { id, .. }
        | WalOp::SetRelationshipProperty { id, .. }
        | WalOp::DeleteRelationship { id } => {
            touched.relationships.insert(*id);
        }
        WalOp::Batch(ops) => {
            for op in ops {
                collect_touched_records(op, touched)?;
            }
        }
        WalOp::ProjectGraph { .. }
        | WalOp::MarkInitialImportSource { .. }
        | WalOp::Relational { .. }
        | WalOp::RelationalSnapshot { .. }
        | WalOp::Append { .. } => {}
        _ => return None,
    }
    Some(())
}

pub fn validate_property_schemas(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> Result<()> {
    for property in catalog.property_descriptors() {
        if property.state != SchemaObjectState::Public {
            continue;
        }
        let Some(table) = catalog.table_descriptor(property.table_id) else {
            continue;
        };
        if table.state != SchemaObjectState::Public {
            continue;
        }
        match table.kind {
            TableKind::Node => {
                let Some(label_id) = catalog.label_id(&table.name) else {
                    continue;
                };
                for node in nodes.values() {
                    if node.labels.contains(&label_id) {
                        validate_property_schema_value(
                            &table.name,
                            &property.name,
                            property.value_type,
                            property.nullable,
                            node.properties.get(&property.name),
                            &format!("node {}", node.id.0),
                        )?;
                    }
                }
            }
            TableKind::Relationship => {
                let Some(rel_type_id) = catalog.rel_type_id(&table.name) else {
                    continue;
                };
                for relationship in relationships.values() {
                    if relationship.rel_type == rel_type_id {
                        validate_property_schema_value(
                            &table.name,
                            &property.name,
                            property.value_type,
                            property.nullable,
                            relationship.properties.get(&property.name),
                            &format!("relationship {}", relationship.id.0),
                        )?;
                    }
                }
            }
        }
    }
    Ok(())
}

pub fn validate_node_record_constraints(catalog: &Catalog, node: &NodeRecord) -> Result<()> {
    for property in catalog.property_descriptors() {
        if property.state != SchemaObjectState::Public {
            continue;
        }
        let Some(table) = catalog.table_descriptor(property.table_id) else {
            continue;
        };
        if table.kind != TableKind::Node || table.state != SchemaObjectState::Public {
            continue;
        }
        let Some(label_id) = catalog.label_id(&table.name) else {
            continue;
        };
        if node.labels.contains(&label_id) {
            validate_property_schema_value(
                &table.name,
                &property.name,
                property.value_type,
                property.nullable,
                node.properties.get(&property.name),
                &format!("node {}", node.id.0),
            )?;
        }
    }
    for constraint in catalog.node_property_exists_constraints() {
        let ConstraintSubject::Node(label_id) = constraint.subject else {
            continue;
        };
        if node.labels.contains(&label_id)
            && !node
                .properties
                .get(&constraint.property)
                .is_some_and(|value| value != &Value::Null)
        {
            let label = catalog.label_name(label_id).unwrap_or("<unknown>");
            return Err(HawDBError::Storage(format!(
                "node property exists constraint violation on :{label}({}) for node {}",
                constraint.property, node.id.0
            )));
        }
    }
    Ok(())
}

pub fn validate_relationship_record_constraints(
    catalog: &Catalog,
    relationship: &RelRecord,
) -> Result<()> {
    for property in catalog.property_descriptors() {
        if property.state != SchemaObjectState::Public {
            continue;
        }
        let Some(table) = catalog.table_descriptor(property.table_id) else {
            continue;
        };
        if table.kind != TableKind::Relationship || table.state != SchemaObjectState::Public {
            continue;
        }
        let Some(rel_type_id) = catalog.rel_type_id(&table.name) else {
            continue;
        };
        if relationship.rel_type == rel_type_id {
            validate_property_schema_value(
                &table.name,
                &property.name,
                property.value_type,
                property.nullable,
                relationship.properties.get(&property.name),
                &format!("relationship {}", relationship.id.0),
            )?;
        }
    }
    for constraint in catalog.relationship_property_exists_constraints() {
        let ConstraintSubject::Relationship(rel_type_id) = constraint.subject else {
            continue;
        };
        if relationship.rel_type == rel_type_id
            && !relationship
                .properties
                .get(&constraint.property)
                .is_some_and(|value| value != &Value::Null)
        {
            let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
            return Err(HawDBError::Storage(format!(
                "relationship property exists constraint violation on :{rel_type}({}) for relationship {}",
                constraint.property, relationship.id.0
            )));
        }
    }
    Ok(())
}

pub fn validate_property_schema_value(
    table: &str,
    property: &str,
    value_type: PropertyType,
    nullable: bool,
    value: Option<&Value>,
    record: &str,
) -> Result<()> {
    let Some(value) = value else {
        if nullable {
            return Ok(());
        }
        return Err(property_schema_error(
            table,
            property,
            record,
            "property is not nullable",
        ));
    };
    if value == &Value::Null {
        if nullable {
            return Ok(());
        }
        return Err(property_schema_error(
            table,
            property,
            record,
            "property is not nullable",
        ));
    }
    let matches = matches!(
        (value_type, value),
        (PropertyType::Any, _)
            | (PropertyType::Bool, Value::Bool(_))
            | (PropertyType::Int, Value::Int(_))
            | (PropertyType::Float, Value::Float(_))
            | (PropertyType::String, Value::String(_))
            | (PropertyType::Text, Value::String(_))
            | (PropertyType::List, Value::List(_))
    );
    if matches {
        Ok(())
    } else {
        Err(property_schema_error(
            table,
            property,
            record,
            &format!("expected {}", encode_property_type(value_type)),
        ))
    }
}

fn property_schema_error(table: &str, property: &str, record: &str, reason: &str) -> HawDBError {
    HawDBError::Storage(format!(
        "property schema violation on {record} in {table}({property}): {reason}"
    ))
}

pub fn validate_unique_constraints(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
) -> Result<()> {
    for constraint in catalog.unique_constraints() {
        let ConstraintSubject::Node(label_id) = constraint.subject else {
            continue;
        };
        validate_unique_property(catalog, nodes, label_id, &constraint.property)?;
    }
    Ok(())
}

pub fn validate_relationship_unique_constraints(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> Result<()> {
    for constraint in catalog.relationship_unique_constraints() {
        let ConstraintSubject::Relationship(rel_type_id) = constraint.subject else {
            continue;
        };
        validate_unique_relationship_property(
            catalog,
            relationships,
            rel_type_id,
            &constraint.property,
        )?;
    }
    Ok(())
}

pub fn validate_node_property_exists_constraints(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
) -> Result<()> {
    for constraint in catalog.node_property_exists_constraints() {
        let ConstraintSubject::Node(label_id) = constraint.subject else {
            continue;
        };
        validate_node_property_exists(catalog, nodes, label_id, &constraint.property)?;
    }
    Ok(())
}

pub fn validate_relationship_property_exists_constraints(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
) -> Result<()> {
    for constraint in catalog.relationship_property_exists_constraints() {
        let ConstraintSubject::Relationship(rel_type_id) = constraint.subject else {
            continue;
        };
        validate_relationship_property_exists(
            catalog,
            relationships,
            rel_type_id,
            &constraint.property,
        )?;
    }
    Ok(())
}

pub fn validate_node_property_exists(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    label_id: LabelId,
    property: &str,
) -> Result<()> {
    for node in nodes.values() {
        if !node.labels.contains(&label_id) {
            continue;
        }
        match node.properties.get(property) {
            Some(value) if value != &Value::Null => {}
            _ => {
                let label = catalog.label_name(label_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "node property exists constraint violation on :{label}({property}) for node {}",
                    node.id.0
                )));
            }
        }
    }
    Ok(())
}

pub fn validate_relationship_property_exists(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
    rel_type_id: RelTypeId,
    property: &str,
) -> Result<()> {
    for relationship in relationships.values() {
        if relationship.rel_type != rel_type_id {
            continue;
        }
        match relationship.properties.get(property) {
            Some(value) if value != &Value::Null => {}
            _ => {
                let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "relationship property exists constraint violation on :{rel_type}({property}) for relationship {}",
                    relationship.id.0
                )));
            }
        }
    }
    Ok(())
}

pub fn validate_unique_property(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    label_id: LabelId,
    property: &str,
) -> Result<()> {
    let mut seen = BTreeMap::<Value, NodeId>::new();
    for node in nodes.values() {
        if !node.labels.contains(&label_id) {
            continue;
        }
        let Some(value) = node.properties.get(property) else {
            continue;
        };
        if value == &Value::Null {
            continue;
        }
        if let Some(previous) = seen.insert(value.clone(), node.id) {
            let label = catalog.label_name(label_id).unwrap_or("<unknown>");
            return Err(HawDBError::Storage(format!(
                "unique constraint violation on :{label}({property}) for nodes {} and {}",
                previous.0, node.id.0
            )));
        }
    }
    Ok(())
}

pub fn validate_unique_relationship_property(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
    rel_type_id: RelTypeId,
    property: &str,
) -> Result<()> {
    let mut seen = BTreeMap::<Value, RelId>::new();
    for relationship in relationships.values() {
        if relationship.rel_type != rel_type_id {
            continue;
        }
        let Some(value) = relationship.properties.get(property) else {
            continue;
        };
        if value == &Value::Null {
            continue;
        }
        if let Some(previous) = seen.insert(value.clone(), relationship.id) {
            let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
            return Err(HawDBError::Storage(format!(
                "relationship unique constraint violation on :{rel_type}({property}) for relationships {} and {}",
                previous.0, relationship.id.0
            )));
        }
    }
    Ok(())
}

/// Delta variant of `validate_property_schemas`: checks only the post-images
/// of touched records against their tables' public property descriptors.
pub fn validate_property_schemas_for_records(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
    touched: &TouchedRecords,
) -> Result<()> {
    for property in catalog.property_descriptors() {
        if property.state != SchemaObjectState::Public {
            continue;
        }
        let Some(table) = catalog.table_descriptor(property.table_id) else {
            continue;
        };
        if table.state != SchemaObjectState::Public {
            continue;
        }
        match table.kind {
            TableKind::Node => {
                let Some(label_id) = catalog.label_id(&table.name) else {
                    continue;
                };
                for node_id in &touched.nodes {
                    let Some(node) = nodes.get(node_id) else {
                        continue;
                    };
                    if node.labels.contains(&label_id) {
                        validate_property_schema_value(
                            &table.name,
                            &property.name,
                            property.value_type,
                            property.nullable,
                            node.properties.get(&property.name),
                            &format!("node {}", node.id.0),
                        )?;
                    }
                }
            }
            TableKind::Relationship => {
                let Some(rel_type_id) = catalog.rel_type_id(&table.name) else {
                    continue;
                };
                for rel_id in &touched.relationships {
                    let Some(relationship) = relationships.get(rel_id) else {
                        continue;
                    };
                    if relationship.rel_type == rel_type_id {
                        validate_property_schema_value(
                            &table.name,
                            &property.name,
                            property.value_type,
                            property.nullable,
                            relationship.properties.get(&property.name),
                            &format!("relationship {}", relationship.id.0),
                        )?;
                    }
                }
            }
        }
    }
    Ok(())
}

/// Delta unique validation: a violation can only involve a touched record's
/// constrained value colliding with another record carrying the same value.
///
/// Costs O(N log K) per constraint (K touched records, N records): one pass
/// collects the constrained post-image values of touched records, a second
/// single scan compares every other record's value against that map. This is
/// never worse than the full validator's O(N log N) — bulk commits where K
/// approaches N degenerate to the same bound instead of quadratic K*N scans.
pub fn validate_unique_constraints_for_records(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    touched: &TouchedRecords,
) -> Result<()> {
    for constraint in catalog.unique_constraints() {
        let ConstraintSubject::Node(label_id) = constraint.subject else {
            continue;
        };
        let mut touched_values = BTreeMap::<&Value, NodeId>::new();
        for node_id in &touched.nodes {
            let Some(node) = nodes.get(node_id) else {
                continue;
            };
            if !node.labels.contains(&label_id) {
                continue;
            }
            let Some(value) = node.properties.get(&constraint.property) else {
                continue;
            };
            if value == &Value::Null {
                continue;
            }
            if let Some(previous) = touched_values.insert(value, node.id) {
                let label = catalog.label_name(label_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "unique constraint violation on :{label}({}) for nodes {} and {}",
                    constraint.property, previous.0, node.id.0
                )));
            }
        }
        if touched_values.is_empty() {
            continue;
        }
        for node in nodes.values() {
            if touched.nodes.contains(&node.id) || !node.labels.contains(&label_id) {
                continue;
            }
            let Some(value) = node.properties.get(&constraint.property) else {
                continue;
            };
            if let Some(touched_id) = touched_values.get(value) {
                let label = catalog.label_name(label_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "unique constraint violation on :{label}({}) for nodes {} and {}",
                    constraint.property, node.id.0, touched_id.0
                )));
            }
        }
    }
    Ok(())
}

/// Delta variant of `validate_relationship_unique_constraints`, with the same
/// O(N log K) two-pass structure as `validate_unique_constraints_for_records`.
pub fn validate_relationship_unique_constraints_for_records(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
    touched: &TouchedRecords,
) -> Result<()> {
    for constraint in catalog.relationship_unique_constraints() {
        let ConstraintSubject::Relationship(rel_type_id) = constraint.subject else {
            continue;
        };
        let mut touched_values = BTreeMap::<&Value, RelId>::new();
        for rel_id in &touched.relationships {
            let Some(relationship) = relationships.get(rel_id) else {
                continue;
            };
            if relationship.rel_type != rel_type_id {
                continue;
            }
            let Some(value) = relationship.properties.get(&constraint.property) else {
                continue;
            };
            if value == &Value::Null {
                continue;
            }
            if let Some(previous) = touched_values.insert(value, relationship.id) {
                let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "relationship unique constraint violation on :{rel_type}({}) for relationships {} and {}",
                    constraint.property, previous.0, relationship.id.0
                )));
            }
        }
        if touched_values.is_empty() {
            continue;
        }
        for relationship in relationships.values() {
            if touched.relationships.contains(&relationship.id)
                || relationship.rel_type != rel_type_id
            {
                continue;
            }
            let Some(value) = relationship.properties.get(&constraint.property) else {
                continue;
            };
            if let Some(touched_id) = touched_values.get(value) {
                let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
                return Err(HawDBError::Storage(format!(
                    "relationship unique constraint violation on :{rel_type}({}) for relationships {} and {}",
                    constraint.property, relationship.id.0, touched_id.0
                )));
            }
        }
    }
    Ok(())
}

/// Delta variant of `validate_node_property_exists_constraints`.
pub fn validate_node_property_exists_constraints_for_records(
    catalog: &Catalog,
    nodes: &CowSegmentedMap<NodeId, NodeRecord>,
    touched: &TouchedRecords,
) -> Result<()> {
    for constraint in catalog.node_property_exists_constraints() {
        let ConstraintSubject::Node(label_id) = constraint.subject else {
            continue;
        };
        for node_id in &touched.nodes {
            let Some(node) = nodes.get(node_id) else {
                continue;
            };
            if !node.labels.contains(&label_id) {
                continue;
            }
            match node.properties.get(&constraint.property) {
                Some(value) if value != &Value::Null => {}
                _ => {
                    let label = catalog.label_name(label_id).unwrap_or("<unknown>");
                    return Err(HawDBError::Storage(format!(
                        "node property exists constraint violation on :{label}({}) for node {}",
                        constraint.property, node.id.0
                    )));
                }
            }
        }
    }
    Ok(())
}

/// Delta variant of `validate_relationship_property_exists_constraints`.
pub fn validate_relationship_property_exists_constraints_for_records(
    catalog: &Catalog,
    relationships: &CowSegmentedMap<RelId, RelRecord>,
    touched: &TouchedRecords,
) -> Result<()> {
    for constraint in catalog.relationship_property_exists_constraints() {
        let ConstraintSubject::Relationship(rel_type_id) = constraint.subject else {
            continue;
        };
        for rel_id in &touched.relationships {
            let Some(relationship) = relationships.get(rel_id) else {
                continue;
            };
            if relationship.rel_type != rel_type_id {
                continue;
            }
            match relationship.properties.get(&constraint.property) {
                Some(value) if value != &Value::Null => {}
                _ => {
                    let rel_type = catalog.rel_type_name(rel_type_id).unwrap_or("<unknown>");
                    return Err(HawDBError::Storage(format!(
                        "relationship property exists constraint violation on :{rel_type}({}) for relationship {}",
                        constraint.property, relationship.id.0
                    )));
                }
            }
        }
    }
    Ok(())
}

pub use crate::text::encode_property_type;

#[cfg(test)]
mod tests;