hawdb-storage 0.14.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.

use crate::wal::WalOp;
use hawdb_core::{
    HawDBError, PropertyType, Result, SchemaObjectState, TableKind, ValidatedRegex, Value,
};
use std::collections::BTreeMap;
use std::num::NonZeroUsize;

mod compact;
#[doc(hidden)]
pub mod evaluate;

#[doc(hidden)]
pub use compact::compact_transaction_graph_ops;

pub const DEFAULT_MAX_MUTATION_AFFECTED_ROWS: usize = 100_000;
pub const DEFAULT_MAX_MUTATION_OPERATIONS: usize = 100_000;
pub const DEFAULT_MAX_MUTATION_RESULT_ROWS: usize = 100_000;
pub const DEFAULT_MAX_MUTATION_RESULT_PAYLOAD_BYTES: usize = 64 * 1024 * 1024;

/// Hard limits applied before a mutation is appended to the WAL.
///
/// Mutation execution is atomic: exceeding a limit must reject the complete
/// mutation before durable or in-memory state changes become visible.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MutationLimits {
    pub max_affected_rows: NonZeroUsize,
    pub max_operations: NonZeroUsize,
    pub max_result_rows: NonZeroUsize,
    pub max_result_payload_bytes: NonZeroUsize,
}

/// Storage-neutral outcome of an atomically committed graph mutation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MutationSummary {
    pub rows: Vec<BTreeMap<String, Value>>,
    pub relational_mutation_outcomes: Vec<crate::relational::RelationalMutationOutcome>,
    pub append_mutation_outcomes: Vec<crate::append_table::AppendMutationOutcome>,
}

impl Default for MutationLimits {
    fn default() -> Self {
        Self {
            max_affected_rows: NonZeroUsize::new(DEFAULT_MAX_MUTATION_AFFECTED_ROWS)
                .expect("default mutation affected-row limit is non-zero"),
            max_operations: NonZeroUsize::new(DEFAULT_MAX_MUTATION_OPERATIONS)
                .expect("default mutation operation limit is non-zero"),
            max_result_rows: NonZeroUsize::new(DEFAULT_MAX_MUTATION_RESULT_ROWS)
                .expect("default mutation result-row limit is non-zero"),
            max_result_payload_bytes: NonZeroUsize::new(DEFAULT_MAX_MUTATION_RESULT_PAYLOAD_BYTES)
                .expect("default mutation result payload limit is non-zero"),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConnectedNodesCreate {
    pub source_label: String,
    pub source_properties: BTreeMap<String, Value>,
    pub rel_type: String,
    pub rel_properties: BTreeMap<String, Value>,
    pub target_label: String,
    pub target_properties: BTreeMap<String, Value>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MatchedRelationshipCreate {
    pub source_label: String,
    pub source_filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub rel_properties: BTreeMap<String, Value>,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MatchedRelationshipMerge {
    pub source_label: String,
    pub source_filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub rel_match_properties: BTreeMap<String, Value>,
    pub on_create_properties: BTreeMap<String, Value>,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RelationshipOnCreatePropertyValue {
    Value(Value),
    MatchedRelationshipProperty { property: String },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MatchedRelationshipCopyMerge {
    pub source_label: String,
    pub source_filter: Option<PropertyFilter>,
    pub old_rel_type: String,
    pub old_rel_filter: BTreeMap<String, Value>,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
    pub new_rel_type: String,
    pub new_rel_match_properties: BTreeMap<String, Value>,
    pub on_create_properties: BTreeMap<String, RelationshipOnCreatePropertyValue>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MatchedRelationshipRetargetMerge {
    pub source_label: String,
    pub source_filter: Option<PropertyFilter>,
    pub old_rel_type: String,
    pub old_rel_filter: BTreeMap<String, Value>,
    pub old_target_label: String,
    pub old_target_filter: Option<PropertyFilter>,
    pub new_target_label: String,
    pub new_target_filter: Option<PropertyFilter>,
    pub new_rel_type: String,
    pub new_rel_match_properties: BTreeMap<String, Value>,
    pub on_create_properties: BTreeMap<String, Value>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MatchedRelationshipSourceRetargetMerge {
    pub old_source_label: String,
    pub old_source_filter: Option<PropertyFilter>,
    pub old_rel_type: String,
    pub old_rel_filter: BTreeMap<String, Value>,
    pub old_target_label: String,
    pub old_target_filter: Option<PropertyFilter>,
    pub new_source_label: String,
    pub new_source_filter: Option<PropertyFilter>,
    pub new_rel_type: String,
    pub new_rel_match_properties: BTreeMap<String, Value>,
    pub on_create_properties: BTreeMap<String, Value>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipPropertyUpdate {
    pub source_label: String,
    pub filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
    pub rel_filter: Option<PropertyFilter>,
    pub property: String,
    pub value: Value,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipSetAssignment {
    pub property: String,
    pub value: Value,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipPropertiesUpdate {
    pub source_label: String,
    pub filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
    pub rel_filter: Option<PropertyFilter>,
    pub assignments: Vec<RelationshipSetAssignment>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipDeleteRequest {
    pub source_label: String,
    pub filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
    pub rel_filter: Option<PropertyFilter>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RelationshipTargetNodeDelete {
    pub source_label: String,
    pub source_filter: Option<PropertyFilter>,
    pub rel_type: String,
    pub rel_filter: Option<PropertyFilter>,
    pub target_label: String,
    pub target_filter: Option<PropertyFilter>,
    pub detach: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GraphMutation {
    CreateNodeLabel {
        label: String,
    },
    CreateRelationshipType {
        rel_type: String,
    },
    CreateNodeTable {
        name: String,
    },
    CreateRelationshipTable {
        name: String,
    },
    CreateProperty {
        table_kind: TableKind,
        table: String,
        property: String,
        value_type: PropertyType,
        nullable: bool,
    },
    AlterTableState {
        table_kind: TableKind,
        table: String,
        state: SchemaObjectState,
    },
    AlterPropertyState {
        table_kind: TableKind,
        table: String,
        property: String,
        state: SchemaObjectState,
    },
    CreateIndex {
        label: String,
        property: String,
    },
    CreateCompositeIndex {
        label: String,
        properties: Vec<String>,
    },
    CreateRangeIndex {
        label: String,
        property: String,
    },
    CreateFullTextIndex {
        label: String,
        property: String,
    },
    CreateUniqueConstraint {
        label: String,
        property: String,
    },
    CreateNodePropertyExistsConstraint {
        label: String,
        property: String,
    },
    CreateRelationshipUniqueConstraint {
        rel_type: String,
        property: String,
    },
    CreateRelationshipPropertyExistsConstraint {
        rel_type: String,
        property: String,
    },
    CreateNode {
        label: String,
        properties: BTreeMap<String, Value>,
    },
    MergeNode {
        label: String,
        match_properties: BTreeMap<String, Value>,
        on_create_properties: BTreeMap<String, Value>,
        on_match_assignments: Vec<NodeSetAssignment>,
        post_merge_assignments: Vec<NodeSetAssignment>,
    },
    MergeConnectedNodes(ConnectedNodesCreate),
    SetNodeProperty {
        label: String,
        filter: Option<PropertyFilter>,
        property: String,
        value: Value,
    },
    SetNodePropertyAddInt {
        label: String,
        filter: Option<PropertyFilter>,
        property: String,
        amount: i64,
    },
    SetNodeProperties {
        label: String,
        filter: Option<PropertyFilter>,
        assignments: Vec<NodeSetAssignment>,
    },
    SetRelationshipProperty {
        source_label: String,
        filter: Option<PropertyFilter>,
        rel_type: String,
        target_label: String,
        target_filter: Option<PropertyFilter>,
        rel_filter: Option<PropertyFilter>,
        property: String,
        value: Value,
    },
    SetRelationshipProperties {
        source_label: String,
        filter: Option<PropertyFilter>,
        rel_type: String,
        target_label: String,
        target_filter: Option<PropertyFilter>,
        rel_filter: Option<PropertyFilter>,
        assignments: Vec<RelationshipSetAssignment>,
    },
    DeleteNode {
        label: String,
        filter: Option<PropertyFilter>,
        detach: bool,
    },
    DeleteRelationship {
        source_label: String,
        filter: Option<PropertyFilter>,
        rel_type: String,
        target_label: String,
        target_filter: Option<PropertyFilter>,
        rel_filter: Option<PropertyFilter>,
    },
    DeleteRelationshipTargetNodes(RelationshipTargetNodeDelete),
    CreateRelationshipsBetweenMatches(MatchedRelationshipCreate),
    MergeRelationshipsBetweenMatches(MatchedRelationshipMerge),
    MergeRelationshipsFromMatchedRelationships(MatchedRelationshipCopyMerge),
    MergeRelationshipsToMatchedTarget(MatchedRelationshipRetargetMerge),
    MergeRelationshipsFromMatchedTarget(MatchedRelationshipSourceRetargetMerge),
    CreateConnectedNodes(ConnectedNodesCreate),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NodeSetAssignment {
    pub property: String,
    pub value: NodeSetValue,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NodeSetValue {
    Value(Value),
    Coalesce { default: Value },
    AddInt { amount: i64 },
    DecrementFloorZero,
    PreserveNewerExisting { incoming: Value, preserve: bool },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PropertyFilter {
    And(Vec<PropertyFilter>),
    Or(Vec<PropertyFilter>),
    Not(Box<PropertyFilter>),
    IdEq {
        value: Value,
    },
    IdNotEq {
        value: Value,
    },
    IdRange {
        lower: Option<(Value, bool)>,
        upper: Option<(Value, bool)>,
    },
    IdIn {
        values: Vec<Value>,
    },
    Eq {
        property: String,
        value: Value,
    },
    NotEq {
        property: String,
        value: Value,
    },
    IsNull {
        property: String,
    },
    IsNotNull {
        property: String,
    },
    In {
        property: String,
        values: Vec<Value>,
    },
    ListContains {
        property: String,
        value: Value,
    },
    ListContainsLower {
        property: String,
        value: String,
    },
    Contains {
        property: String,
        value: String,
    },
    StartsWith {
        property: String,
        value: String,
    },
    EndsWith {
        property: String,
        value: String,
    },
    RegexMatch {
        property: String,
        pattern: ValidatedRegex,
    },
    DefaultIfNullOrEq {
        property: String,
        empty: Value,
        default: Value,
        value: Value,
        negated: bool,
    },
    Range {
        property: String,
        lower: Option<(Value, bool)>,
        upper: Option<(Value, bool)>,
    },
}

pub fn ensure_mutation_commit_limits(
    ops: &[WalOp],
    rows: &[BTreeMap<String, Value>],
    limits: MutationLimits,
) -> Result<()> {
    MutationCommitLimitTracker::default().check_appended(ops, rows, limits)
}

/// Tracks an append-only result prefix within one mutation preparation.
/// Previously checked rows must remain immutable until preparation ends.
#[derive(Default)]
pub(crate) struct MutationCommitLimitTracker {
    checked_rows: usize,
    payload_bytes: u64,
}

impl MutationCommitLimitTracker {
    pub(crate) fn check_appended(
        &mut self,
        ops: &[WalOp],
        rows: &[BTreeMap<String, Value>],
        limits: MutationLimits,
    ) -> Result<()> {
        ensure_additional_mutation_limits(ops.len(), rows.len(), 0, 0, limits)?;
        if rows.len() > limits.max_result_rows.get() {
            return Err(HawDBError::Execution(format!(
                "mutation would exceed max_mutation_result_rows {}",
                limits.max_result_rows
            )));
        }
        let payload_bytes =
            rows[self.checked_rows..]
                .iter()
                .fold(self.payload_bytes, |total, row| {
                    total.saturating_add(row.iter().fold(0u64, |row_total, (name, value)| {
                        row_total
                            .saturating_add(name.len() as u64)
                            .saturating_add(estimated_value_bytes(value))
                    }))
                });
        if payload_bytes > limits.max_result_payload_bytes.get() as u64 {
            return Err(HawDBError::Execution(format!(
                "mutation result payload would exceed max_mutation_result_payload_bytes {}",
                limits.max_result_payload_bytes
            )));
        }
        self.checked_rows = rows.len();
        self.payload_bytes = payload_bytes;
        Ok(())
    }
}

pub fn ensure_additional_mutation_limits(
    operation_count: usize,
    affected_row_count: usize,
    additional_operations: usize,
    additional_affected_rows: usize,
    limits: MutationLimits,
) -> Result<()> {
    let next_operations = operation_count
        .checked_add(additional_operations)
        .ok_or_else(|| HawDBError::Execution("mutation operation count overflow".to_string()))?;
    if next_operations > limits.max_operations.get() {
        return Err(HawDBError::Execution(format!(
            "mutation would exceed max_mutation_operations {}",
            limits.max_operations
        )));
    }
    let next_affected_rows = affected_row_count
        .checked_add(additional_affected_rows)
        .ok_or_else(|| HawDBError::Execution("mutation affected-row count overflow".to_string()))?;
    if next_affected_rows > limits.max_affected_rows.get() {
        return Err(HawDBError::Execution(format!(
            "mutation would exceed max_mutation_affected_rows {}",
            limits.max_affected_rows
        )));
    }
    Ok(())
}

pub fn remaining_mutation_affected_rows(current: usize, limits: MutationLimits) -> Result<usize> {
    limits
        .max_affected_rows
        .get()
        .checked_sub(current)
        .ok_or_else(|| {
            HawDBError::Execution(format!(
                "mutation would exceed max_mutation_affected_rows {}",
                limits.max_affected_rows
            ))
        })
}

pub fn remaining_mutation_operations(current: usize, limits: MutationLimits) -> Result<usize> {
    limits
        .max_operations
        .get()
        .checked_sub(current)
        .ok_or_else(|| {
            HawDBError::Execution(format!(
                "mutation would exceed max_mutation_operations {}",
                limits.max_operations
            ))
        })
}

pub fn estimated_properties_bytes(properties: &BTreeMap<String, Value>) -> u64 {
    estimated_properties_bytes_with_visit(properties, &mut || Ok::<_, std::convert::Infallible>(()))
        .unwrap_or_else(|never| match never {})
}

pub fn estimated_value_bytes(value: &Value) -> u64 {
    estimated_value_bytes_with_visit(value, &mut || Ok::<_, std::convert::Infallible>(()))
        .unwrap_or_else(|never| match never {})
}

/// Visits each borrowed property/value before its constant-time size step.
/// Ordinary estimation uses the same traversal with an infallible no-op hook.
#[doc(hidden)]
pub fn estimated_properties_bytes_with_visit<E>(
    properties: &BTreeMap<String, Value>,
    visit: &mut impl FnMut() -> std::result::Result<(), E>,
) -> std::result::Result<u64, E> {
    properties.iter().try_fold(0u64, |bytes, (key, value)| {
        visit()?;
        Ok(bytes
            .saturating_add(key.len() as u64)
            .saturating_add(estimated_value_bytes_with_visit(value, visit)?)
            .saturating_add(16))
    })
}

#[doc(hidden)]
pub fn estimated_value_bytes_with_visit<E>(
    value: &Value,
    visit: &mut impl FnMut() -> std::result::Result<(), E>,
) -> std::result::Result<u64, E> {
    visit()?;
    Ok(match value {
        Value::Null => 1,
        Value::Bool(_) => 1,
        Value::Int(_) | Value::Float(_) => 8,
        Value::String(value) => value.len() as u64,
        Value::Binary(value) => value.len() as u64,
        Value::Vector(value) => value.encoded().len() as u64,
        Value::Uuid(_) => 16,
        Value::List(values) => values.iter().try_fold(16u64, |bytes, value| {
            Ok::<_, E>(bytes.saturating_add(estimated_value_bytes_with_visit(value, visit)?))
        })?,
        Value::Map(values) => estimated_properties_bytes_with_visit(values, visit)?,
    })
}

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

    #[test]
    fn incremental_commit_limits_match_full_prefix_checks() {
        let row = BTreeMap::from([
            ("null".to_string(), Value::Null),
            ("number".to_string(), Value::Int(7)),
            (
                "nested".to_string(),
                Value::List(vec![
                    Value::String("payload".to_string()),
                    Value::Map(BTreeMap::from([(
                        "bytes".to_string(),
                        Value::Binary(vec![1, 2, 3]),
                    )])),
                ]),
            ),
        ]);
        let row_bytes = row.iter().fold(0u64, |total, (name, value)| {
            total
                .saturating_add(name.len() as u64)
                .saturating_add(estimated_value_bytes(value))
        });
        for boundary in 1..=12 {
            for kind in 0..5 {
                let mut limits = MutationLimits::default();
                let limit = NonZeroUsize::new(boundary).unwrap();
                match kind {
                    0 => limits.max_operations = limit,
                    1 => limits.max_affected_rows = limit,
                    2 => limits.max_result_rows = limit,
                    3 => {
                        limits.max_result_payload_bytes =
                            NonZeroUsize::new(boundary * row_bytes as usize).unwrap();
                    }
                    _ => {
                        limits.max_result_payload_bytes =
                            NonZeroUsize::new(boundary * row_bytes as usize - 1).unwrap();
                    }
                }
                let mut tracker = MutationCommitLimitTracker::default();
                let mut ops = Vec::new();
                let mut rows = Vec::new();
                for index in 0..16 {
                    // Some preparation branches append several rows or no op.
                    rows.extend(std::iter::repeat_n(row.clone(), index % 3));
                    if index % 2 == 0 {
                        ops.push(WalOp::CreateNodeLabel {
                            label: format!("Label{index}"),
                        });
                    }
                    let expected = ensure_mutation_commit_limits(&ops, &rows, limits)
                        .map_err(|error| error.to_string());
                    let actual = tracker
                        .check_appended(&ops, &rows, limits)
                        .map_err(|error| error.to_string());
                    assert_eq!(actual, expected);
                }
            }
        }
    }

    #[test]
    fn mutation_contract_keeps_property_filters_structured() {
        let mutation = GraphMutation::DeleteNode {
            label: "Memory".to_string(),
            filter: Some(PropertyFilter::Eq {
                property: "status".to_string(),
                value: Value::String("deleted".to_string()),
            }),
            detach: false,
        };
        assert!(matches!(
            mutation,
            GraphMutation::DeleteNode {
                filter: Some(PropertyFilter::Eq { .. }),
                ..
            }
        ));
    }
}