grust_core/

lib.rs

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    fmt,
};

use async_trait::async_trait;
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};

pub type Result<T> = std::result::Result<T, GrustError>;
pub type Props = BTreeMap<String, Value>;

#[derive(Debug, thiserror::Error)]
pub enum GrustError {
    #[error("backend error: {0}")]
    Backend(String),
    #[error("schema error: {0}")]
    Schema(String),
    #[error("unsupported graph feature: {0}")]
    Unsupported(String),
    #[error("Cypher syntax error: {0}")]
    CypherSyntax(String),
    #[error("Cypher unresolved identity: {0}")]
    CypherUnresolvedIdentity(String),
    #[error("Cypher unsupported cardinality: {0}")]
    CypherUnsupportedCardinality(String),
    #[error("Cypher execution error: {0}")]
    CypherExecution(String),
    #[error("serialization error: {0}")]
    Serialization(String),
}

macro_rules! string_newtype {
    ($(#[$meta:meta])* $name:ident) => {
        $(#[$meta])*
        #[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Serialize, Deserialize)]
        pub struct $name(String);

        impl $name {
            pub fn new(value: impl Into<String>) -> Self {
                Self(value.into())
            }

            pub fn as_str(&self) -> &str {
                &self.0
            }

            pub fn into_string(self) -> String {
                self.0
            }
        }

        impl fmt::Display for $name {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                f.write_str(&self.0)
            }
        }

        impl From<String> for $name {
            fn from(value: String) -> Self {
                Self::new(value)
            }
        }

        impl From<&str> for $name {
            fn from(value: &str) -> Self {
                Self::new(value)
            }
        }

        impl From<&String> for $name {
            fn from(value: &String) -> Self {
                Self::new(value.clone())
            }
        }

        impl From<&$name> for $name {
            fn from(value: &$name) -> Self {
                value.clone()
            }
        }
    };
}

string_newtype!(
    /// Stable application-level node identifier.
    NodeId
);
string_newtype!(
    /// Optional application-level edge identifier.
    EdgeId
);
string_newtype!(
    /// Node or edge label.
    Label
);

/// Validated RFC 3339 date-time string used by [`Value::DateTime`].
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct RfcDate(String);

impl RfcDate {
    /// Parses and stores an RFC 3339 date-time such as
    /// `2026-06-12T09:30:00Z` or `2026-06-12T09:30:00.123+02:00`.
    pub fn parse(value: impl Into<String>) -> Result<Self> {
        let value = value.into();
        if is_rfc3339_datetime(&value) {
            Ok(Self(value))
        } else {
            Err(GrustError::Schema(format!(
                "'{value}' is not an RFC 3339 date-time"
            )))
        }
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }

    pub fn into_string(self) -> String {
        self.0
    }
}

impl fmt::Display for RfcDate {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

impl Serialize for RfcDate {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(self.as_str())
    }
}

impl<'de> Deserialize<'de> for RfcDate {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::parse(value).map_err(de::Error::custom)
    }
}

/// Fixed-point decimal: the value is `mantissa × 10^(−scale)`. This mirrors SQL
/// `DECIMAL(38, scale)` (Postgres/Spark cap precision at 38 digits, which is what
/// an `i128` mantissa holds), and is lossless within that range — unlike `Float`.
///
/// Values are normalized on construction: trailing fractional zeros are dropped
/// (`1.50` and `1.5` are equal), and `0` always has scale `0`. Equality and
/// ordering are therefore by numeric value.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub struct Decimal {
    mantissa: i128,
    scale: u32,
}

impl Decimal {
    /// Parses a decimal numeral such as `3.14`, `-0.001`, or `42`. Rejects empty
    /// input, malformed numerals, and values exceeding 38 significant digits.
    pub fn parse(value: impl AsRef<str>) -> Result<Self> {
        let raw = value.as_ref().trim();
        let bad = || GrustError::Schema(format!("'{raw}' is not a decimal numeral"));
        if raw.is_empty() {
            return Err(bad());
        }
        let (neg, body) = match raw.strip_prefix('-') {
            Some(rest) => (true, rest),
            None => (false, raw.strip_prefix('+').unwrap_or(raw)),
        };
        let (int_part, frac_part) = match body.split_once('.') {
            Some((i, f)) => (i, f),
            None => (body, ""),
        };
        if int_part.is_empty() && frac_part.is_empty() {
            return Err(bad());
        }
        if !int_part.bytes().all(|b| b.is_ascii_digit())
            || !frac_part.bytes().all(|b| b.is_ascii_digit())
        {
            return Err(bad());
        }
        let digits: String = int_part.chars().chain(frac_part.chars()).collect();
        let mantissa: i128 = if digits.is_empty() {
            0
        } else {
            digits.parse().map_err(|_| {
                GrustError::Schema(format!("decimal '{raw}' exceeds 38 significant digits"))
            })?
        };
        let mantissa = if neg { -mantissa } else { mantissa };
        Ok(Self::normalize(mantissa, frac_part.len() as u32))
    }

    /// Builds a decimal from a raw mantissa and scale (`mantissa × 10^−scale`).
    pub fn from_parts(mantissa: i128, scale: u32) -> Self {
        Self::normalize(mantissa, scale)
    }

    fn normalize(mut mantissa: i128, mut scale: u32) -> Self {
        while scale > 0 && mantissa % 10 == 0 {
            mantissa /= 10;
            scale -= 1;
        }
        if mantissa == 0 {
            scale = 0;
        }
        Self { mantissa, scale }
    }

    pub fn mantissa(&self) -> i128 {
        self.mantissa
    }

    pub fn scale(&self) -> u32 {
        self.scale
    }

    /// Canonical string form (no superfluous trailing zeros).
    pub fn to_canonical_string(&self) -> String {
        if self.scale == 0 {
            return self.mantissa.to_string();
        }
        let neg = self.mantissa < 0;
        let digits = self.mantissa.unsigned_abs().to_string();
        let scale = self.scale as usize;
        let s = if digits.len() <= scale {
            format!("0.{:0>width$}", digits, width = scale)
        } else {
            let point = digits.len() - scale;
            format!("{}.{}", &digits[..point], &digits[point..])
        };
        if neg {
            format!("-{s}")
        } else {
            s
        }
    }

    /// Aligns two decimals to a common scale, returning their scaled mantissas.
    /// Returns `None` on `i128` overflow (operands beyond the 38-digit range).
    fn aligned(&self, other: &Self) -> Option<(i128, i128, u32)> {
        let scale = self.scale.max(other.scale);
        let lift = |m: i128, s: u32| 10i128.checked_pow(scale - s).and_then(|f| m.checked_mul(f));
        Some((lift(self.mantissa, self.scale)?, lift(other.mantissa, other.scale)?, scale))
    }

    pub fn checked_add(&self, other: &Self) -> Option<Self> {
        let (a, b, scale) = self.aligned(other)?;
        Some(Self::normalize(a.checked_add(b)?, scale))
    }

    pub fn checked_sub(&self, other: &Self) -> Option<Self> {
        let (a, b, scale) = self.aligned(other)?;
        Some(Self::normalize(a.checked_sub(b)?, scale))
    }

    pub fn checked_mul(&self, other: &Self) -> Option<Self> {
        let mantissa = self.mantissa.checked_mul(other.mantissa)?;
        Some(Self::normalize(mantissa, self.scale + other.scale))
    }

    /// Lossy conversion to `f64` (for coercion into float arithmetic).
    pub fn to_f64(&self) -> f64 {
        self.mantissa as f64 / 10f64.powi(self.scale as i32)
    }
}

impl Ord for Decimal {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        match self.aligned(other) {
            Some((a, b, _)) => a.cmp(&b),
            // Overflow on alignment: fall back to comparing the f64 magnitudes,
            // which is monotone enough to separate out-of-range operands.
            None => self
                .to_f64()
                .partial_cmp(&other.to_f64())
                .unwrap_or(std::cmp::Ordering::Equal),
        }
    }
}

impl PartialOrd for Decimal {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl fmt::Display for Decimal {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.to_canonical_string())
    }
}

impl Serialize for Decimal {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_canonical_string())
    }
}

impl<'de> Deserialize<'de> for Decimal {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::parse(value).map_err(de::Error::custom)
    }
}

/// Calendar/clock duration with month, day, second, and nanosecond components —
/// the model GQL/ISO 8601 uses (months and days are not fixed-length, so they are
/// kept distinct from seconds rather than collapsed). Construct from an ISO 8601
/// duration string such as `P1Y2M10DT2H30M` via [`Duration::parse`].
///
/// Ordering is structural (months, then days, then seconds, then nanos): a
/// deterministic total order for sorting, not a calendar-normalized comparison.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct Duration {
    pub months: i64,
    pub days: i64,
    pub seconds: i64,
    pub nanos: i32,
}

impl Duration {
    /// Parses an ISO 8601 duration: `P[nY][nM][nW][nD][T[nH][nM][nS]]`. Years map
    /// to 12 months and weeks to 7 days; the seconds field may be fractional
    /// (down to nanosecond precision). At least one component is required.
    pub fn parse(value: impl AsRef<str>) -> Result<Self> {
        let raw = value.as_ref().trim();
        let bad = || GrustError::Schema(format!("'{raw}' is not an ISO 8601 duration"));
        let body = raw.strip_prefix('P').ok_or_else(bad)?;
        let (date_part, time_part) = match body.split_once('T') {
            Some((d, t)) => (d, Some(t)),
            None => (body, None),
        };
        let mut dur = Duration::default();
        let mut any = false;
        // Date components: Y, M, W, D (integer only).
        let mut num = String::new();
        for ch in date_part.chars() {
            if ch.is_ascii_digit() || ch == '-' {
                num.push(ch);
            } else {
                let n: i64 = num.parse().map_err(|_| bad())?;
                num.clear();
                any = true;
                match ch {
                    'Y' => dur.months += n * 12,
                    'M' => dur.months += n,
                    'W' => dur.days += n * 7,
                    'D' => dur.days += n,
                    _ => return Err(bad()),
                }
            }
        }
        if !num.is_empty() {
            return Err(bad());
        }
        // Time components: H, M, S (S may be fractional).
        if let Some(time_part) = time_part {
            let mut tok = String::new();
            for ch in time_part.chars() {
                if ch.is_ascii_digit() || ch == '-' || ch == '.' {
                    tok.push(ch);
                } else {
                    any = true;
                    match ch {
                        'H' => dur.seconds += tok.parse::<i64>().map_err(|_| bad())? * 3600,
                        'M' => dur.seconds += tok.parse::<i64>().map_err(|_| bad())? * 60,
                        'S' => {
                            let (secs, nanos) = parse_fractional_seconds(&tok).ok_or_else(bad)?;
                            dur.seconds += secs;
                            dur.nanos += nanos;
                        }
                        _ => return Err(bad()),
                    }
                    tok.clear();
                }
            }
            if !tok.is_empty() {
                return Err(bad());
            }
        }
        if !any {
            return Err(bad());
        }
        Ok(dur.carry_nanos())
    }

    fn carry_nanos(mut self) -> Self {
        if self.nanos.abs() >= 1_000_000_000 {
            self.seconds += (self.nanos / 1_000_000_000) as i64;
            self.nanos %= 1_000_000_000;
        }
        self
    }

    /// Canonical ISO 8601 string (`PT0S` for the zero duration).
    pub fn to_iso_string(&self) -> String {
        let mut out = String::from("P");
        if self.months != 0 {
            out.push_str(&format!("{}M", self.months));
        }
        if self.days != 0 {
            out.push_str(&format!("{}D", self.days));
        }
        if self.seconds != 0 || self.nanos != 0 {
            out.push('T');
            if self.nanos != 0 {
                let frac = format!("{:09}", self.nanos.unsigned_abs());
                let frac = frac.trim_end_matches('0');
                out.push_str(&format!("{}.{}S", self.seconds, frac));
            } else {
                out.push_str(&format!("{}S", self.seconds));
            }
        }
        if out == "P" {
            out.push_str("T0S");
        }
        out
    }

    pub fn checked_add(&self, other: &Self) -> Option<Self> {
        Some(
            Self {
                months: self.months.checked_add(other.months)?,
                days: self.days.checked_add(other.days)?,
                seconds: self.seconds.checked_add(other.seconds)?,
                nanos: self.nanos.checked_add(other.nanos)?,
            }
            .carry_nanos(),
        )
    }

    pub fn negated(&self) -> Self {
        Self {
            months: -self.months,
            days: -self.days,
            seconds: -self.seconds,
            nanos: -self.nanos,
        }
    }
}

/// Parses a possibly-fractional seconds token into whole seconds + nanoseconds.
fn parse_fractional_seconds(tok: &str) -> Option<(i64, i32)> {
    match tok.split_once('.') {
        None => Some((tok.parse().ok()?, 0)),
        Some((whole, frac)) => {
            if frac.is_empty() || !frac.bytes().all(|b| b.is_ascii_digit()) {
                return None;
            }
            let secs: i64 = if whole.is_empty() || whole == "-" {
                0
            } else {
                whole.parse().ok()?
            };
            let frac9: String = frac.chars().chain(std::iter::repeat('0')).take(9).collect();
            let mut nanos: i32 = frac9.parse().ok()?;
            if whole.starts_with('-') {
                nanos = -nanos;
            }
            Some((secs, nanos))
        }
    }
}

impl fmt::Display for Duration {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.to_iso_string())
    }
}

impl Serialize for Duration {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_iso_string())
    }
}

impl<'de> Deserialize<'de> for Duration {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = String::deserialize(deserializer)?;
        Self::parse(value).map_err(de::Error::custom)
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", content = "value", rename_all = "snake_case")]
pub enum Value {
    Null,
    Bool(bool),
    Int(i64),
    Float(f64),
    String(String),
    /// An RFC 3339 date-time, e.g. `2026-06-12T09:30:00Z`. Construct with
    /// [`Value::datetime`] to get format validation.
    DateTime(RfcDate),
    /// A lossless fixed-point decimal (SQL DECIMAL-style). Construct with
    /// [`Value::decimal`].
    Decimal(Decimal),
    /// An ISO 8601 calendar/clock duration. Construct with [`Value::duration`].
    Duration(Duration),
    StringArray(Vec<String>),
    IntArray(Vec<i64>),
    FloatArray(Vec<f64>),
    Json(serde_json::Value),
}

impl Value {
    /// Creates a `Value::DateTime`, validating that the input is an RFC 3339
    /// date-time such as `2026-06-12T09:30:00Z` or
    /// `2026-06-12T09:30:00.123+02:00`.
    pub fn datetime(value: impl Into<String>) -> Result<Self> {
        RfcDate::parse(value).map(Self::DateTime)
    }

    /// Creates a `Value::Decimal`, validating the decimal numeral (e.g. `3.14`).
    pub fn decimal(value: impl AsRef<str>) -> Result<Self> {
        Decimal::parse(value).map(Self::Decimal)
    }

    /// Creates a `Value::Duration` from an ISO 8601 duration (e.g. `P1Y2MT3H`).
    pub fn duration(value: impl AsRef<str>) -> Result<Self> {
        Duration::parse(value).map(Self::Duration)
    }

    pub fn as_decimal(&self) -> Option<&Decimal> {
        match self {
            Self::Decimal(value) => Some(value),
            _ => None,
        }
    }

    pub fn as_duration(&self) -> Option<&Duration> {
        match self {
            Self::Duration(value) => Some(value),
            _ => None,
        }
    }

    pub fn as_str(&self) -> Option<&str> {
        match self {
            Self::String(value) => Some(value),
            _ => None,
        }
    }

    pub fn as_datetime(&self) -> Option<&str> {
        match self {
            Self::DateTime(value) => Some(value.as_str()),
            _ => None,
        }
    }

    pub fn as_string_array(&self) -> Option<&[String]> {
        match self {
            Self::StringArray(values) => Some(values),
            _ => None,
        }
    }

    pub fn as_int_array(&self) -> Option<&[i64]> {
        match self {
            Self::IntArray(values) => Some(values),
            _ => None,
        }
    }

    pub fn as_float_array(&self) -> Option<&[f64]> {
        match self {
            Self::FloatArray(values) => Some(values),
            _ => None,
        }
    }

    /// Converts to a plain (untagged) JSON value. `DateTime` becomes a JSON
    /// string, so a `to_json`/`from_json` round trip yields `Value::String`.
    pub fn to_json(&self) -> serde_json::Value {
        match self {
            Self::Null => serde_json::Value::Null,
            Self::Bool(value) => serde_json::Value::Bool(*value),
            Self::Int(value) => serde_json::Value::from(*value),
            Self::Float(value) => serde_json::Value::from(*value),
            Self::String(value) => serde_json::Value::String(value.clone()),
            Self::DateTime(value) => serde_json::Value::String(value.as_str().to_string()),
            Self::Decimal(value) => serde_json::Value::String(value.to_canonical_string()),
            Self::Duration(value) => serde_json::Value::String(value.to_iso_string()),
            Self::StringArray(values) => serde_json::Value::from(values.clone()),
            Self::IntArray(values) => serde_json::Value::from(values.clone()),
            Self::FloatArray(values) => serde_json::Value::from(values.clone()),
            Self::Json(value) => value.clone(),
        }
    }

    /// Converts from JSON, accepting both plain values and the tagged
    /// `{"type": ..., "value": ...}` form that `Value`'s serde representation
    /// produces.
    pub fn from_json(value: serde_json::Value) -> Self {
        if let serde_json::Value::Object(mapping) = &value
            && mapping.contains_key("type")
            && mapping.contains_key("value")
            && let Ok(tagged) = serde_json::from_value(value.clone())
        {
            return tagged;
        }
        Self::from(value)
    }
}

/// Validates the RFC 3339 date-time shape `YYYY-MM-DDTHH:MM:SS[.frac](Z|±HH:MM)`.
fn is_rfc3339_datetime(value: &str) -> bool {
    let bytes = value.as_bytes();
    if bytes.len() < 20 {
        return false;
    }
    let digit = |i: usize| bytes[i].is_ascii_digit();
    let all_digits = |range: std::ops::Range<usize>| range.clone().all(digit);
    let pair = |i: usize| (bytes[i] - b'0') * 10 + (bytes[i + 1] - b'0');

    if !(all_digits(0..4)
        && bytes[4] == b'-'
        && all_digits(5..7)
        && bytes[7] == b'-'
        && all_digits(8..10)
        && bytes[10] == b'T'
        && all_digits(11..13)
        && bytes[13] == b':'
        && all_digits(14..16)
        && bytes[16] == b':'
        && all_digits(17..19))
    {
        return false;
    }
    let year = bytes[0..4]
        .iter()
        .fold(0u16, |acc, digit| acc * 10 + u16::from(digit - b'0'));
    let month = pair(5);
    let day = pair(8);
    let leap_year = (year % 4 == 0 && year % 100 != 0) || year % 400 == 0;
    let max_day = match month {
        1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
        4 | 6 | 9 | 11 => 30,
        2 if leap_year => 29,
        2 => 28,
        _ => return false,
    };
    if !(day >= 1 && day <= max_day && pair(11) < 24 && pair(14) < 60 && pair(17) <= 60) {
        return false;
    }

    let mut i = 19;
    if bytes[i] == b'.' {
        let start = i + 1;
        i = start;
        while i < bytes.len() && bytes[i].is_ascii_digit() {
            i += 1;
        }
        if i == start {
            return false;
        }
    }
    match bytes.get(i) {
        Some(b'Z') => i + 1 == bytes.len(),
        Some(b'+' | b'-') => {
            i + 6 == bytes.len()
                && all_digits(i + 1..i + 3)
                && bytes[i + 3] == b':'
                && all_digits(i + 4..i + 6)
                && pair(i + 1) < 24
                && pair(i + 4) < 60
        }
        _ => false,
    }
}

impl From<String> for Value {
    fn from(value: String) -> Self {
        Self::String(value)
    }
}

impl From<&str> for Value {
    fn from(value: &str) -> Self {
        Self::String(value.to_string())
    }
}

impl From<&String> for Value {
    fn from(value: &String) -> Self {
        Self::String(value.clone())
    }
}

impl From<Vec<String>> for Value {
    fn from(value: Vec<String>) -> Self {
        Self::StringArray(value)
    }
}

impl From<Vec<i64>> for Value {
    fn from(value: Vec<i64>) -> Self {
        Self::IntArray(value)
    }
}

impl From<Vec<f64>> for Value {
    fn from(value: Vec<f64>) -> Self {
        Self::FloatArray(value)
    }
}

impl From<bool> for Value {
    fn from(value: bool) -> Self {
        Self::Bool(value)
    }
}

impl From<i64> for Value {
    fn from(value: i64) -> Self {
        Self::Int(value)
    }
}

impl From<i32> for Value {
    fn from(value: i32) -> Self {
        Self::Int(i64::from(value))
    }
}

impl From<usize> for Value {
    fn from(value: usize) -> Self {
        Self::Int(value as i64)
    }
}

impl From<f64> for Value {
    fn from(value: f64) -> Self {
        Self::Float(value)
    }
}

impl From<serde_json::Value> for Value {
    fn from(value: serde_json::Value) -> Self {
        match value {
            serde_json::Value::Null => Self::Null,
            serde_json::Value::Bool(value) => Self::Bool(value),
            serde_json::Value::Number(value) => {
                if let Some(value) = value.as_i64() {
                    Self::Int(value)
                } else if let Some(value) = value.as_f64() {
                    Self::Float(value)
                } else {
                    Self::Json(serde_json::Value::Number(value))
                }
            }
            serde_json::Value::String(value) => Self::String(value),
            serde_json::Value::Array(values) => {
                let ints = values
                    .iter()
                    .filter_map(serde_json::Value::as_i64)
                    .collect::<Vec<_>>();
                if !values.is_empty() && ints.len() == values.len() {
                    return Self::IntArray(ints);
                }
                let floats = values
                    .iter()
                    .filter_map(serde_json::Value::as_f64)
                    .collect::<Vec<_>>();
                if !values.is_empty() && floats.len() == values.len() {
                    return Self::FloatArray(floats);
                }
                let strings = values
                    .iter()
                    .filter_map(|value| value.as_str().map(ToString::to_string))
                    .collect::<Vec<_>>();
                if strings.len() == values.len() {
                    Self::StringArray(strings)
                } else {
                    Self::Json(serde_json::Value::Array(values))
                }
            }
            serde_json::Value::Object(value) => Self::Json(serde_json::Value::Object(value)),
        }
    }
}

#[cfg(feature = "typed-garde")]
pub mod typed {
    use serde::{Serialize, de::DeserializeOwned};

    use crate::{
        Edge, EdgeId, Graph, GraphBuilder, GrustError, Node, NodeId, Props, PutOutcome, Result,
        Value,
    };

    pub use garde;
    #[cfg(feature = "typed-zod-rs")]
    pub use zod_rs;

    pub trait TypedNode: garde::Validate + Serialize {
        const LABEL: &'static str;

        fn node_id(&self) -> NodeId;

        fn node_props(&self) -> Result<Props> {
            props_from_serialize(self)
        }

        fn from_node(node: &Node) -> Result<Self>
        where
            Self: Sized + DeserializeOwned,
            Self::Context: Default,
        {
            let ctx = Self::Context::default();
            Self::from_node_with(node, &ctx)
        }

        fn from_node_with(node: &Node, ctx: &Self::Context) -> Result<Self>
        where
            Self: Sized + DeserializeOwned,
        {
            if node.label.as_str() != Self::LABEL {
                return Err(GrustError::Schema(format!(
                    "node '{}' has label '{}', expected '{}'",
                    node.id.as_str(),
                    node.label.as_str(),
                    Self::LABEL
                )));
            }
            let typed: Self =
                serde_json::from_value(props_to_plain_json(&node.props)).map_err(|err| {
                    GrustError::Serialization(format!("typed node decode error: {err}"))
                })?;
            typed
                .validate_with(ctx)
                .map_err(|err| validation_error(Self::LABEL, err))?;
            let typed_id = typed.node_id();
            if typed_id != node.id {
                return Err(GrustError::Schema(format!(
                    "typed node '{}' decoded id '{}', expected '{}'",
                    Self::LABEL,
                    typed_id.as_str(),
                    node.id.as_str()
                )));
            }
            Ok(typed)
        }
    }

    pub trait TypedEdge: garde::Validate + Serialize {
        const LABEL: &'static str;

        fn source_node_id(&self) -> NodeId;

        fn target_node_id(&self) -> NodeId;

        fn edge_id(&self) -> Option<EdgeId> {
            None
        }

        fn edge_props(&self) -> Result<Props> {
            props_from_serialize(self)
        }

        fn from_edge(edge: &Edge) -> Result<Self>
        where
            Self: Sized + DeserializeOwned,
            Self::Context: Default,
        {
            let ctx = Self::Context::default();
            Self::from_edge_with(edge, &ctx)
        }

        fn from_edge_with(edge: &Edge, ctx: &Self::Context) -> Result<Self>
        where
            Self: Sized + DeserializeOwned,
        {
            if edge.label.as_str() != Self::LABEL {
                return Err(GrustError::Schema(format!(
                    "edge from '{}' to '{}' has label '{}', expected '{}'",
                    edge.from.as_str(),
                    edge.to.as_str(),
                    edge.label.as_str(),
                    Self::LABEL
                )));
            }
            let typed: Self =
                serde_json::from_value(props_to_plain_json(&edge.props)).map_err(|err| {
                    GrustError::Serialization(format!("typed edge decode error: {err}"))
                })?;
            typed
                .validate_with(ctx)
                .map_err(|err| validation_error(Self::LABEL, err))?;
            if typed.source_node_id() != edge.from || typed.target_node_id() != edge.to {
                return Err(GrustError::Schema(format!(
                    "typed edge '{}' decoded endpoints '{}' -> '{}', expected '{}' -> '{}'",
                    Self::LABEL,
                    typed.source_node_id().as_str(),
                    typed.target_node_id().as_str(),
                    edge.from.as_str(),
                    edge.to.as_str()
                )));
            }
            if let Some(decoded_id) = typed.edge_id()
                && edge
                    .id
                    .as_ref()
                    .is_some_and(|edge_id| edge_id != &decoded_id)
            {
                return Err(GrustError::Schema(format!(
                    "typed edge '{}' decoded id '{}', expected '{}'",
                    Self::LABEL,
                    decoded_id.as_str(),
                    edge.id.as_ref().expect("edge id checked").as_str()
                )));
            }
            Ok(typed)
        }
    }

    #[derive(Clone, Debug, Default)]
    pub struct TypedGraphBuilder {
        builder: GraphBuilder,
    }

    impl TypedGraphBuilder {
        pub fn new() -> Self {
            Self::default()
        }

        pub fn from_builder(builder: GraphBuilder) -> Self {
            Self { builder }
        }

        pub fn from_graph(graph: Graph) -> Self {
            let mut builder = GraphBuilder::new();
            for node in graph.nodes {
                builder.add_node(node);
            }
            for edge in graph.edges {
                builder.add_edge(edge);
            }
            Self { builder }
        }

        pub fn add_raw_node(&mut self, node: Node) -> NodeId {
            self.builder.add_node(node)
        }

        pub fn add_raw_edge(&mut self, edge: Edge) -> PutOutcome {
            self.builder.add_edge(edge)
        }

        pub fn add_node<T>(&mut self, node: &T) -> Result<NodeId>
        where
            T: TypedNode,
            T::Context: Default,
        {
            node.validate()
                .map_err(|err| validation_error(T::LABEL, err))?;
            self.add_validated_node(node)
        }

        pub fn add_node_with<T>(&mut self, node: &T, ctx: &T::Context) -> Result<NodeId>
        where
            T: TypedNode,
        {
            node.validate_with(ctx)
                .map_err(|err| validation_error(T::LABEL, err))?;
            self.add_validated_node(node)
        }

        pub fn add_edge<T>(&mut self, edge: &T) -> Result<PutOutcome>
        where
            T: TypedEdge,
            T::Context: Default,
        {
            edge.validate()
                .map_err(|err| validation_error(T::LABEL, err))?;
            self.add_validated_edge(edge)
        }

        pub fn add_edge_with<T>(&mut self, edge: &T, ctx: &T::Context) -> Result<PutOutcome>
        where
            T: TypedEdge,
        {
            edge.validate_with(ctx)
                .map_err(|err| validation_error(T::LABEL, err))?;
            self.add_validated_edge(edge)
        }

        #[cfg(feature = "typed-zod-rs")]
        pub fn add_node_from_json<T, S>(
            &mut self,
            schema: &S,
            value: &serde_json::Value,
        ) -> Result<NodeId>
        where
            T: TypedNode + DeserializeOwned,
            T::Context: Default,
            S: zod_rs::Schema<serde_json::Value>,
        {
            let node = parse_typed_json::<T, S>(schema, value)?;
            self.add_validated_node(&node)
        }

        #[cfg(feature = "typed-zod-rs")]
        pub fn add_node_from_json_with<T, S>(
            &mut self,
            schema: &S,
            value: &serde_json::Value,
            ctx: &T::Context,
        ) -> Result<NodeId>
        where
            T: TypedNode + DeserializeOwned,
            S: zod_rs::Schema<serde_json::Value>,
        {
            let node = parse_typed_json_with::<T, S>(schema, value, ctx)?;
            self.add_validated_node(&node)
        }

        #[cfg(feature = "typed-zod-rs")]
        pub fn add_edge_from_json<T, S>(
            &mut self,
            schema: &S,
            value: &serde_json::Value,
        ) -> Result<PutOutcome>
        where
            T: TypedEdge + DeserializeOwned,
            T::Context: Default,
            S: zod_rs::Schema<serde_json::Value>,
        {
            let edge = parse_typed_json::<T, S>(schema, value)?;
            self.add_validated_edge(&edge)
        }

        #[cfg(feature = "typed-zod-rs")]
        pub fn add_edge_from_json_with<T, S>(
            &mut self,
            schema: &S,
            value: &serde_json::Value,
            ctx: &T::Context,
        ) -> Result<PutOutcome>
        where
            T: TypedEdge + DeserializeOwned,
            S: zod_rs::Schema<serde_json::Value>,
        {
            let edge = parse_typed_json_with::<T, S>(schema, value, ctx)?;
            self.add_validated_edge(&edge)
        }

        #[must_use = "discarding this means the typed graph was not built"]
        pub fn build(self) -> Graph {
            self.builder.build()
        }

        pub fn into_builder(self) -> GraphBuilder {
            self.builder
        }

        fn add_validated_node<T>(&mut self, node: &T) -> Result<NodeId>
        where
            T: TypedNode,
        {
            let node_id = node.node_id();
            let mut props = node.node_props()?;
            props
                .entry("id".to_string())
                .or_insert_with(|| Value::from(node_id.as_str()));
            let graph_node = Node::new(T::LABEL, node_id, props);
            Ok(self.builder.add_node(graph_node))
        }

        fn add_validated_edge<T>(&mut self, edge: &T) -> Result<PutOutcome>
        where
            T: TypedEdge,
        {
            let mut graph_edge = Edge::new(
                T::LABEL,
                edge.source_node_id(),
                edge.target_node_id(),
                edge.edge_props()?,
            );
            graph_edge.id = edge.edge_id();
            Ok(self.builder.add_edge(graph_edge))
        }
    }

    pub fn props_from_serialize<T>(value: &T) -> Result<Props>
    where
        T: Serialize + ?Sized,
    {
        let serialized = serde_json::to_value(value)
            .map_err(|err| GrustError::Serialization(format!("typed props error: {err}")))?;
        let serde_json::Value::Object(fields) = serialized else {
            return Err(GrustError::Schema(
                "typed graph values must serialize as JSON objects".to_string(),
            ));
        };

        Ok(fields
            .into_iter()
            .map(|(key, value)| (key, Value::from(value)))
            .collect())
    }

    fn props_to_plain_json(props: &Props) -> serde_json::Value {
        serde_json::Value::Object(
            props
                .iter()
                .map(|(key, value)| (key.clone(), value.to_json()))
                .collect(),
        )
    }

    #[cfg(feature = "typed-zod-rs")]
    pub fn parse_typed_json<T, S>(schema: &S, value: &serde_json::Value) -> Result<T>
    where
        T: DeserializeOwned + garde::Validate,
        T::Context: Default,
        S: zod_rs::Schema<serde_json::Value>,
    {
        let ctx = T::Context::default();
        parse_typed_json_with(schema, value, &ctx)
    }

    #[cfg(feature = "typed-zod-rs")]
    pub fn parse_typed_json_with<T, S>(
        schema: &S,
        value: &serde_json::Value,
        ctx: &T::Context,
    ) -> Result<T>
    where
        T: DeserializeOwned + garde::Validate,
        S: zod_rs::Schema<serde_json::Value>,
    {
        schema
            .safe_parse(value)
            .map_err(|err| GrustError::Schema(format!("zod-rs validation failed: {err}")))?;
        let typed: T = serde_json::from_value(value.clone())
            .map_err(|err| GrustError::Serialization(format!("typed JSON decode error: {err}")))?;
        typed
            .validate_with(ctx)
            .map_err(|err| GrustError::Schema(format!("typed validation failed: {err}")))?;
        Ok(typed)
    }

    fn validation_error(label: &str, err: garde::Report) -> GrustError {
        GrustError::Schema(format!("{label} validation failed: {err}"))
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Node {
    pub id: NodeId,
    pub label: Label,
    pub props: Props,
}

impl Node {
    pub fn new(label: impl Into<Label>, id: impl Into<NodeId>, props: impl Into<Props>) -> Self {
        let id = id.into();
        let mut props = props.into();
        props
            .entry("id".to_string())
            .or_insert_with(|| Value::from(id.as_str()));
        Self {
            id,
            label: label.into(),
            props,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Edge {
    pub id: Option<EdgeId>,
    pub from: NodeId,
    pub to: NodeId,
    pub label: Label,
    pub props: Props,
}

impl Edge {
    pub fn new(
        label: impl Into<Label>,
        from: impl Into<NodeId>,
        to: impl Into<NodeId>,
        props: impl Into<Props>,
    ) -> Self {
        Self {
            id: None,
            from: from.into(),
            to: to.into(),
            label: label.into(),
            props: props.into(),
        }
    }

    pub fn with_id(mut self, id: impl Into<EdgeId>) -> Self {
        self.id = Some(id.into());
        self
    }
}

/// Normalizes an edge label into an uppercase backend relationship type.
///
/// Non-ASCII-alphanumeric characters become underscores. Empty labels fall
/// back to `RELATED_TO`.
pub fn relationship_type(value: &str) -> String {
    let relationship = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() {
                ch.to_ascii_uppercase()
            } else {
                '_'
            }
        })
        .collect::<String>();
    if relationship.is_empty() {
        "RELATED_TO".to_string()
    } else {
        relationship
    }
}

/// Normalizes arbitrary schema text into a lower_snake_case backend identifier.
///
/// This helper is for SQL-like backends that require identifiers to start with a
/// non-digit ASCII alphanumeric or underscore character after normalization.
pub fn schema_identifier(value: &str) -> Result<String> {
    let identifier = value
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() {
                ch.to_ascii_lowercase()
            } else {
                '_'
            }
        })
        .collect::<String>();
    if identifier.is_empty()
        || identifier
            .chars()
            .next()
            .is_some_and(|ch| ch.is_ascii_digit())
    {
        return Err(GrustError::Schema(format!(
            "invalid schema identifier '{value}'"
        )));
    }
    Ok(identifier)
}

/// Returns the stable key used by tabular/export backends for an edge.
///
/// Explicit edge IDs win. Otherwise the structural key joins `from`, `label`,
/// and `to` with U+001F (Unit Separator). Callers that accept arbitrary IDs or
/// labels should reject U+001F before relying on reversibility.
pub fn edge_key(edge: &Edge) -> String {
    edge.id
        .as_ref()
        .map(EdgeId::as_str)
        .map(ToString::to_string)
        .unwrap_or_else(|| {
            let from = edge.from.as_str();
            let label = edge.label.as_str();
            let to = edge.to.as_str();
            let mut key = String::with_capacity(from.len() + label.len() + to.len() + 2);
            key.push_str(from);
            key.push('\u{1f}');
            key.push_str(label);
            key.push('\u{1f}');
            key.push_str(to);
            key
        })
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct Graph {
    pub nodes: Vec<Node>,
    pub edges: Vec<Edge>,
}

impl Graph {
    pub fn new(nodes: Vec<Node>, edges: Vec<Edge>) -> Self {
        Self { nodes, edges }
    }

    pub fn from_yaml(yaml: &str) -> Result<Self> {
        yaml::graph_from_yaml(yaml)
    }

    pub fn to_yaml(&self) -> Result<String> {
        yaml::graph_to_yaml(self)
    }

    pub fn from_json(json: &str) -> Result<Self> {
        json::graph_from_json(json)
    }

    pub fn to_json(&self) -> Result<String> {
        json::graph_to_json(self)
    }

    pub fn from_xml(xml: &str) -> Result<Self> {
        xml::graph_from_xml(xml)
    }

    pub fn to_xml(&self) -> Result<String> {
        xml::graph_to_xml(self)
    }

    pub fn builder() -> GraphBuilder {
        GraphBuilder::new()
    }
}

/// Dense, reusable indexes over a [`Graph`].
///
/// This is backend-neutral: it validates edge endpoints, maps node ids to
/// stable vertex indexes, and stores edge adjacency by both node id and vertex
/// index. Higher-level crates can use it for local analytics or query planning
/// without rebuilding the same maps.
#[derive(Clone, Debug, PartialEq)]
pub struct GraphIndex {
    vertex_by_id: HashMap<NodeId, usize>,
    outgoing_by_vertex: Vec<Vec<usize>>,
    incoming_by_vertex: Vec<Vec<usize>>,
    edge_endpoints: Vec<(usize, usize)>,
}

impl GraphIndex {
    pub fn new(graph: &Graph) -> Result<Self> {
        let mut vertex_by_id = HashMap::with_capacity(graph.nodes.len());
        for (index, vertex) in graph.nodes.iter().enumerate() {
            if vertex_by_id.insert(vertex.id.clone(), index).is_some() {
                return Err(GrustError::Schema(format!(
                    "duplicate vertex id '{}'",
                    vertex.id.as_str()
                )));
            }
        }

        let mut outgoing_by_vertex = vec![Vec::<usize>::new(); graph.nodes.len()];
        let mut incoming_by_vertex = vec![Vec::<usize>::new(); graph.nodes.len()];
        let mut edge_endpoints = Vec::<(usize, usize)>::with_capacity(graph.edges.len());

        for (edge_index, edge) in graph.edges.iter().enumerate() {
            let Some(&from_index) = vertex_by_id.get(&edge.from) else {
                return Err(GrustError::Schema(format!(
                    "edge source '{}' is not present in vertices",
                    edge.from.as_str()
                )));
            };
            let Some(&to_index) = vertex_by_id.get(&edge.to) else {
                return Err(GrustError::Schema(format!(
                    "edge destination '{}' is not present in vertices",
                    edge.to.as_str()
                )));
            };

            outgoing_by_vertex[from_index].push(edge_index);
            incoming_by_vertex[to_index].push(edge_index);
            edge_endpoints.push((from_index, to_index));
        }

        Ok(Self {
            vertex_by_id,
            outgoing_by_vertex,
            incoming_by_vertex,
            edge_endpoints,
        })
    }

    pub fn vertex_index(&self, id: &NodeId) -> Option<usize> {
        self.vertex_by_id.get(id).copied()
    }

    pub fn require_vertex_index(&self, id: &NodeId) -> Result<usize> {
        self.vertex_index(id)
            .ok_or_else(|| GrustError::Schema(format!("vertex '{}' is not present", id.as_str())))
    }

    pub fn outgoing_edges(&self, id: &NodeId) -> &[usize] {
        self.vertex_index(id)
            .map(|index| self.outgoing_by_vertex(index))
            .unwrap_or(&[])
    }

    pub fn incoming_edges(&self, id: &NodeId) -> &[usize] {
        self.vertex_index(id)
            .map(|index| self.incoming_by_vertex(index))
            .unwrap_or(&[])
    }

    pub fn outgoing_by_vertex(&self, index: usize) -> &[usize] {
        self.outgoing_by_vertex
            .get(index)
            .map(Vec::as_slice)
            .unwrap_or(&[])
    }

    pub fn incoming_by_vertex(&self, index: usize) -> &[usize] {
        self.incoming_by_vertex
            .get(index)
            .map(Vec::as_slice)
            .unwrap_or(&[])
    }

    pub fn edge_endpoints(&self, edge_index: usize) -> (usize, usize) {
        self.edge_endpoints[edge_index]
    }

    pub fn edge_endpoints_slice(&self) -> &[(usize, usize)] {
        &self.edge_endpoints
    }

    pub fn out_degree(&self, index: usize) -> usize {
        self.outgoing_by_vertex[index].len()
    }

    pub fn in_degree(&self, index: usize) -> usize {
        self.incoming_by_vertex[index].len()
    }

    pub fn degree(&self, index: usize) -> usize {
        self.in_degree(index) + self.out_degree(index)
    }
}

mod graph_doc {
    use std::collections::{BTreeMap, BTreeSet};

    use serde::{Deserialize, Serialize};

    use crate::{Edge, EdgeId, Graph, GrustError, Label, Node, NodeId, Props, Value};

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct GraphDoc {
        #[serde(default)]
        pub(super) nodes: Vec<NodeDoc>,
        #[serde(default)]
        pub(super) edges: Vec<EdgeDoc>,
    }

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct NodeDoc {
        pub(super) id: NodeId,
        pub(super) label: Label,
        #[serde(default, deserialize_with = "deserialize_props")]
        pub(super) props: Props,
    }

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct NodeDocOut {
        pub(super) id: NodeId,
        pub(super) label: Label,
        #[serde(default)]
        pub(super) props: Props,
    }

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct EdgeDoc {
        #[serde(default)]
        pub(super) id: Option<EdgeId>,
        pub(super) label: Label,
        pub(super) from: NodeId,
        pub(super) to: NodeId,
        #[serde(default, deserialize_with = "deserialize_props")]
        pub(super) props: Props,
    }

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct EdgeDocOut {
        #[serde(default)]
        pub(super) id: Option<EdgeId>,
        pub(super) label: Label,
        pub(super) from: NodeId,
        pub(super) to: NodeId,
        #[serde(default)]
        pub(super) props: Props,
    }

    pub(super) fn graph_from_doc(doc: GraphDoc) -> super::Result<Graph> {
        let mut ids = BTreeSet::new();
        for node in &doc.nodes {
            if !ids.insert(node.id.clone()) {
                return Err(GrustError::Schema(format!(
                    "duplicate node id '{}'",
                    node.id
                )));
            }
        }

        let mut edges = Vec::with_capacity(doc.edges.len());
        for edge in doc.edges {
            if !ids.contains(&edge.from) {
                return Err(GrustError::Schema(format!(
                    "edge '{}' references unknown from node '{}'",
                    edge.label, edge.from
                )));
            }
            if !ids.contains(&edge.to) {
                return Err(GrustError::Schema(format!(
                    "edge '{}' references unknown to node '{}'",
                    edge.label, edge.to
                )));
            }

            let mut graph_edge = Edge::new(edge.label, edge.from, edge.to, edge.props);
            graph_edge.id = edge.id;
            edges.push(graph_edge);
        }

        let nodes = doc
            .nodes
            .into_iter()
            .map(|node| Node::new(node.label, node.id, node.props))
            .collect();

        Ok(Graph::new(nodes, edges))
    }

    pub(super) fn graph_to_doc(graph: &Graph) -> GraphDocOut {
        GraphDocOut {
            nodes: graph
                .nodes
                .iter()
                .map(|node| NodeDocOut {
                    id: node.id.clone(),
                    label: node.label.clone(),
                    props: without_generated_id(&node.props, &node.id),
                })
                .collect(),
            edges: graph
                .edges
                .iter()
                .map(|edge| EdgeDocOut {
                    id: edge.id.clone(),
                    label: edge.label.clone(),
                    from: edge.from.clone(),
                    to: edge.to.clone(),
                    props: edge.props.clone(),
                })
                .collect(),
        }
    }

    fn without_generated_id(props: &Props, id: &NodeId) -> Props {
        let mut props = props.clone();
        if props.get("id") == Some(&Value::from(id.as_str())) {
            props.remove("id");
        }
        props
    }

    fn deserialize_props<'de, D>(deserializer: D) -> std::result::Result<Props, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let raw = BTreeMap::<String, serde_json::Value>::deserialize(deserializer)?;
        raw.into_iter()
            .map(|(key, value)| {
                value_from_json(value)
                    .map(|value| (key, value))
                    .map_err(serde::de::Error::custom)
            })
            .collect()
    }

    fn value_from_json(value: serde_json::Value) -> std::result::Result<Value, String> {
        if let serde_json::Value::Object(mapping) = &value
            && mapping.contains_key("type")
            && mapping.contains_key("value")
        {
            return serde_json::from_value(value)
                .map_err(|err| format!("invalid tagged Grust value: {err}"));
        }

        Ok(Value::from_json(value))
    }

    #[derive(Debug, Serialize, Deserialize)]
    pub(super) struct GraphDocOut {
        pub(super) nodes: Vec<NodeDocOut>,
        pub(super) edges: Vec<EdgeDocOut>,
    }
}

mod yaml {
    use crate::{Graph, GrustError};

    pub(super) fn graph_from_yaml(yaml: &str) -> super::Result<Graph> {
        let doc: super::graph_doc::GraphDoc = serde_yaml::from_str(yaml)
            .map_err(|err| GrustError::Serialization(format!("YAML parse error: {err}")))?;
        super::graph_doc::graph_from_doc(doc)
    }

    pub(super) fn graph_to_yaml(graph: &Graph) -> super::Result<String> {
        serde_yaml::to_string(&super::graph_doc::graph_to_doc(graph))
            .map_err(|err| GrustError::Serialization(format!("YAML serialization error: {err}")))
    }
}

mod json {
    use crate::{Graph, GrustError};

    pub(super) fn graph_from_json(json: &str) -> super::Result<Graph> {
        let doc: super::graph_doc::GraphDoc = serde_json::from_str(json)
            .map_err(|err| GrustError::Serialization(format!("JSON parse error: {err}")))?;
        super::graph_doc::graph_from_doc(doc)
    }

    pub(super) fn graph_to_json(graph: &Graph) -> super::Result<String> {
        serde_json::to_string_pretty(&super::graph_doc::graph_to_doc(graph))
            .map_err(|err| GrustError::Serialization(format!("JSON serialization error: {err}")))
    }
}

mod xml {
    use serde::{Deserialize, Serialize};

    use crate::{Edge, EdgeId, Graph, GrustError, Label, Node, NodeId, Props, Value};

    #[derive(Debug, Serialize, Deserialize)]
    #[serde(rename = "graph")]
    struct GraphXml {
        #[serde(default)]
        nodes: NodesXml,
        #[serde(default)]
        edges: EdgesXml,
    }

    #[derive(Debug, Default, Serialize, Deserialize)]
    struct NodesXml {
        #[serde(rename = "node", default)]
        items: Vec<NodeXml>,
    }

    #[derive(Debug, Default, Serialize, Deserialize)]
    struct EdgesXml {
        #[serde(rename = "edge", default)]
        items: Vec<EdgeXml>,
    }

    #[derive(Debug, Serialize, Deserialize)]
    struct NodeXml {
        id: NodeId,
        label: Label,
        #[serde(default)]
        props: PropsXml,
    }

    #[derive(Debug, Serialize, Deserialize)]
    struct EdgeXml {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        id: Option<EdgeId>,
        label: Label,
        from: NodeId,
        to: NodeId,
        #[serde(default)]
        props: PropsXml,
    }

    #[derive(Debug, Default, Serialize, Deserialize)]
    struct PropsXml {
        #[serde(rename = "prop", default)]
        items: Vec<PropXml>,
    }

    #[derive(Debug, Serialize, Deserialize)]
    struct PropXml {
        key: String,
        value: Value,
    }

    pub(super) fn graph_from_xml(xml: &str) -> super::Result<Graph> {
        let doc: GraphXml = quick_xml::de::from_str(xml)
            .map_err(|err| GrustError::Serialization(format!("XML parse error: {err}")))?;
        super::graph_doc::graph_from_doc(doc.into())
    }

    pub(super) fn graph_to_xml(graph: &Graph) -> super::Result<String> {
        quick_xml::se::to_string(&GraphXml::from(graph))
            .map_err(|err| GrustError::Serialization(format!("XML serialization error: {err}")))
    }

    impl From<GraphXml> for super::graph_doc::GraphDoc {
        fn from(value: GraphXml) -> Self {
            Self {
                nodes: value.nodes.items.into_iter().map(Into::into).collect(),
                edges: value.edges.items.into_iter().map(Into::into).collect(),
            }
        }
    }

    impl From<NodeXml> for super::graph_doc::NodeDoc {
        fn from(value: NodeXml) -> Self {
            Self {
                id: value.id,
                label: value.label,
                props: value.props.into(),
            }
        }
    }

    impl From<EdgeXml> for super::graph_doc::EdgeDoc {
        fn from(value: EdgeXml) -> Self {
            Self {
                id: value.id,
                label: value.label,
                from: value.from,
                to: value.to,
                props: value.props.into(),
            }
        }
    }

    impl From<PropsXml> for Props {
        fn from(value: PropsXml) -> Self {
            value
                .items
                .into_iter()
                .map(|prop| (prop.key, prop.value))
                .collect()
        }
    }

    impl From<&Graph> for GraphXml {
        fn from(graph: &Graph) -> Self {
            Self {
                nodes: NodesXml {
                    items: graph.nodes.iter().map(NodeXml::from).collect(),
                },
                edges: EdgesXml {
                    items: graph.edges.iter().map(EdgeXml::from).collect(),
                },
            }
        }
    }

    impl From<&Node> for NodeXml {
        fn from(node: &Node) -> Self {
            let props = super::graph_doc::graph_to_doc(&Graph::new(vec![node.clone()], Vec::new()))
                .nodes
                .into_iter()
                .next()
                .expect("node exists")
                .props;
            Self {
                id: node.id.clone(),
                label: node.label.clone(),
                props: props.into(),
            }
        }
    }

    impl From<&Edge> for EdgeXml {
        fn from(edge: &Edge) -> Self {
            Self {
                id: edge.id.clone(),
                label: edge.label.clone(),
                from: edge.from.clone(),
                to: edge.to.clone(),
                props: edge.props.clone().into(),
            }
        }
    }

    impl From<Props> for PropsXml {
        fn from(value: Props) -> Self {
            Self {
                items: value
                    .into_iter()
                    .map(|(key, value)| PropXml { key, value })
                    .collect(),
            }
        }
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub enum EdgePolicy {
    AllowDuplicates,
    #[default]
    DedupeByFromLabelTo,
}

#[derive(Clone, Debug, Default)]
pub struct GraphBuilder {
    nodes: BTreeMap<NodeId, Node>,
    edges: Vec<Edge>,
    edge_keys: BTreeSet<(NodeId, Label, NodeId)>,
    edge_policy: EdgePolicy,
}

impl GraphBuilder {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn edge_policy(mut self, edge_policy: EdgePolicy) -> Self {
        self.edge_policy = edge_policy;
        self
    }

    pub fn node<'a>(
        &'a mut self,
        label: impl Into<Label>,
        id: impl Into<NodeId>,
    ) -> NodeBuilder<'a> {
        NodeBuilder {
            builder: self,
            label: label.into(),
            id: id.into(),
            props: Props::new(),
        }
    }

    pub fn edge<'a>(
        &'a mut self,
        label: impl Into<Label>,
        from: impl Into<NodeId>,
        to: impl Into<NodeId>,
    ) -> EdgeBuilder<'a> {
        EdgeBuilder {
            builder: self,
            id: None,
            label: label.into(),
            from: from.into(),
            to: to.into(),
            props: Props::new(),
        }
    }

    /// Adds a node, merging with any existing node that has the same id.
    ///
    /// If a node with the same id and the same label already exists, the new
    /// props are merged in (new values win). If a node with the same id but a
    /// different label exists, the new node replaces it entirely (last write
    /// wins, matching `GraphStore::put_node` overwrite semantics).
    pub fn add_node(&mut self, node: Node) -> NodeId {
        let id = node.id.clone();
        self.nodes
            .entry(id.clone())
            .and_modify(|existing| {
                if existing.label == node.label {
                    existing.props.extend(node.props.clone());
                } else {
                    *existing = node.clone();
                }
            })
            .or_insert(node);
        id
    }

    /// Adds an edge, reporting whether it was stored or dropped by the
    /// builder's [`EdgePolicy`].
    pub fn add_edge(&mut self, edge: Edge) -> PutOutcome {
        match self.edge_policy {
            EdgePolicy::AllowDuplicates => {
                self.edges.push(edge);
                PutOutcome::Inserted
            }
            EdgePolicy::DedupeByFromLabelTo => {
                let key = (edge.from.clone(), edge.label.clone(), edge.to.clone());
                if self.edge_keys.insert(key) {
                    self.edges.push(edge);
                    PutOutcome::Inserted
                } else {
                    PutOutcome::Deduped
                }
            }
        }
    }

    #[must_use = "discarding this means the graph was not built"]
    pub fn build(self) -> Graph {
        Graph {
            nodes: self.nodes.into_values().collect(),
            edges: self.edges,
        }
    }
}

pub struct NodeBuilder<'a> {
    builder: &'a mut GraphBuilder,
    label: Label,
    id: NodeId,
    props: Props,
}

impl<'a> NodeBuilder<'a> {
    pub fn prop(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
        self.props.insert(key.into(), value.into());
        self
    }

    pub fn props(mut self, props: Props) -> Self {
        self.props.extend(props);
        self
    }

    #[must_use = "discarding this means the node was not added to the builder"]
    pub fn finish(self) -> NodeId {
        let node = Node::new(self.label, self.id, self.props);
        self.builder.add_node(node)
    }
}

pub struct EdgeBuilder<'a> {
    builder: &'a mut GraphBuilder,
    id: Option<EdgeId>,
    label: Label,
    from: NodeId,
    to: NodeId,
    props: Props,
}

impl<'a> EdgeBuilder<'a> {
    pub fn id(mut self, id: impl Into<EdgeId>) -> Self {
        self.id = Some(id.into());
        self
    }

    pub fn prop(mut self, key: impl Into<String>, value: impl Into<Value>) -> Self {
        self.props.insert(key.into(), value.into());
        self
    }

    pub fn props(mut self, props: Props) -> Self {
        self.props.extend(props);
        self
    }

    #[must_use = "discarding this means the edge was not added to the builder"]
    pub fn finish(self) -> PutOutcome {
        let mut edge = Edge::new(self.label, self.from, self.to, self.props);
        edge.id = self.id;
        self.builder.add_edge(edge)
    }
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct GraphSchema {
    pub nodes: Vec<NodeType>,
    pub edges: Vec<EdgeType>,
    pub constraints: Vec<GraphConstraint>,
}

impl GraphSchema {
    pub fn builder() -> GraphSchemaBuilder {
        GraphSchemaBuilder::default()
    }

    pub fn node_type(&self, label: &Label) -> Option<&NodeType> {
        self.nodes
            .iter()
            .find(|node_type| &node_type.label == label)
    }

    pub fn edge_type(&self, label: &Label) -> Option<&EdgeType> {
        self.edges
            .iter()
            .find(|edge_type| &edge_type.label == label)
    }

    pub fn constraints_for_label(&self, label: &Label) -> Vec<&GraphConstraint> {
        self.constraints
            .iter()
            .filter(|constraint| constraint.label() == label)
            .collect()
    }

    pub fn validate_graph(&self, graph: &Graph) -> Result<()> {
        for node in &graph.nodes {
            self.validate_node(node)?;
        }
        let labels: BTreeMap<&NodeId, &Label> = graph
            .nodes
            .iter()
            .map(|node| (&node.id, &node.label))
            .collect();
        for edge in &graph.edges {
            self.validate_edge_with(edge, |id| labels.get(id).copied())?;
        }
        self.validate_edge_uniqueness(graph)?;
        self.validate_unique_property_constraints(graph)
    }

    /// Enforces each edge type's [`EdgeUniqueness`]: at most one edge of the
    /// type between a given endpoint pair (unordered for undirected types).
    fn validate_edge_uniqueness(&self, graph: &Graph) -> Result<()> {
        let mut seen = BTreeSet::new();
        for edge in &graph.edges {
            let Some(edge_type) = self.edge_type(&edge.label) else {
                continue;
            };
            if edge_type.uniqueness == EdgeUniqueness::None {
                continue;
            }
            let (a, b) = if edge_type.directed || edge.from <= edge.to {
                (&edge.from, &edge.to)
            } else {
                (&edge.to, &edge.from)
            };
            if !seen.insert((edge.label.clone(), a.clone(), b.clone())) {
                return Err(GrustError::Schema(format!(
                    "duplicate edge '{}' between '{}' and '{}' violates {:?} uniqueness",
                    edge.label.as_str(),
                    a.as_str(),
                    b.as_str(),
                    edge_type.uniqueness
                )));
            }
        }
        Ok(())
    }

    fn validate_unique_property_constraints(&self, graph: &Graph) -> Result<()> {
        for constraint in &self.constraints {
            match constraint {
                GraphConstraint::NodePropertyUnique { label, key } => {
                    let mut seen: Vec<(&NodeId, &Value)> = Vec::new();
                    for node in graph.nodes.iter().filter(|node| &node.label == label) {
                        let Some(value) = node.props.get(key) else {
                            continue;
                        };
                        if let Some((existing_id, _)) =
                            seen.iter().find(|(_, existing)| *existing == value)
                        {
                            return Err(GrustError::Schema(format!(
                                "node '{}' with label '{}' duplicates unique constrained property '{}' from node '{}'",
                                node.id.as_str(),
                                label.as_str(),
                                key,
                                existing_id.as_str()
                            )));
                        }
                        seen.push((&node.id, value));
                    }
                }
                GraphConstraint::EdgePropertyUnique { label, key } => {
                    let mut seen: Vec<(String, &Value)> = Vec::new();
                    for edge in graph.edges.iter().filter(|edge| &edge.label == label) {
                        let Some(value) = edge.props.get(key) else {
                            continue;
                        };
                        if let Some((existing_key, _)) =
                            seen.iter().find(|(_, existing)| *existing == value)
                        {
                            return Err(GrustError::Schema(format!(
                                "edge '{}' duplicates unique constrained property '{}' from edge '{}'",
                                edge_key(edge),
                                key,
                                existing_key
                            )));
                        }
                        seen.push((edge_key(edge), value));
                    }
                }
                GraphConstraint::NodePropertyRequired { .. }
                | GraphConstraint::EdgePropertyRequired { .. } => {}
            }
        }
        Ok(())
    }

    pub fn validate_node(&self, node: &Node) -> Result<()> {
        let node_type = self.node_type(&node.label).ok_or_else(|| {
            GrustError::Schema(format!("schema has no node type '{}'", node.label.as_str()))
        })?;
        validate_props(
            &node.props,
            &node_type.fields,
            &format!("node '{}'", node.id.as_str()),
        )?;
        for constraint in &self.constraints {
            if let GraphConstraint::NodePropertyRequired { label, key } = constraint
                && label == &node.label
                && !node.props.contains_key(key)
            {
                return Err(GrustError::Schema(format!(
                    "node '{}' with label '{}' is missing required constrained property '{}'",
                    node.id.as_str(),
                    node.label.as_str(),
                    key
                )));
            }
        }
        Ok(())
    }

    pub fn validate_edge(&self, edge: &Edge, graph: &Graph) -> Result<()> {
        self.validate_edge_with(edge, |id| {
            graph
                .nodes
                .iter()
                .find(|node| &node.id == id)
                .map(|node| &node.label)
        })
    }

    /// Validates an edge using a label lookup instead of a full `Graph`, so
    /// stores can validate against their own node index without cloning.
    pub fn validate_edge_with<'a>(
        &self,
        edge: &Edge,
        lookup: impl Fn(&NodeId) -> Option<&'a Label>,
    ) -> Result<()> {
        let edge_type = self.edge_type(&edge.label).ok_or_else(|| {
            GrustError::Schema(format!("schema has no edge type '{}'", edge.label.as_str()))
        })?;

        let from_label = lookup(&edge.from).ok_or_else(|| {
            GrustError::Schema(format!(
                "edge '{}' references unknown from node '{}'",
                edge.label.as_str(),
                edge.from.as_str()
            ))
        })?;
        let to_label = lookup(&edge.to).ok_or_else(|| {
            GrustError::Schema(format!(
                "edge '{}' references unknown to node '{}'",
                edge.label.as_str(),
                edge.to.as_str()
            ))
        })?;

        let from_matches =
            |label: &Label| edge_type.from.is_empty() || edge_type.from.contains(label);
        let to_matches = |label: &Label| edge_type.to.is_empty() || edge_type.to.contains(label);
        // Undirected edge types accept their endpoint labels in either
        // orientation.
        let endpoints_ok = (from_matches(from_label) && to_matches(to_label))
            || (!edge_type.directed && from_matches(to_label) && to_matches(from_label));
        if !endpoints_ok {
            if !from_matches(from_label) {
                return Err(GrustError::Schema(format!(
                    "edge '{}' cannot start from node label '{}'",
                    edge.label.as_str(),
                    from_label.as_str()
                )));
            }
            return Err(GrustError::Schema(format!(
                "edge '{}' cannot end at node label '{}'",
                edge.label.as_str(),
                to_label.as_str()
            )));
        }

        validate_props(
            &edge.props,
            &edge_type.fields,
            &format!("edge '{}'", edge.label.as_str()),
        )?;
        self.validate_edge_required_constraints(edge)
    }

    /// Validates an edge's label and props against the schema without
    /// checking endpoint nodes, for stores that persist a single edge and
    /// cannot cheaply resolve its endpoints.
    pub fn validate_edge_props(&self, edge: &Edge) -> Result<()> {
        let edge_type = self.edge_type(&edge.label).ok_or_else(|| {
            GrustError::Schema(format!("schema has no edge type '{}'", edge.label.as_str()))
        })?;
        validate_props(
            &edge.props,
            &edge_type.fields,
            &format!("edge '{}'", edge.label.as_str()),
        )?;
        self.validate_edge_required_constraints(edge)
    }

    fn validate_edge_required_constraints(&self, edge: &Edge) -> Result<()> {
        for constraint in &self.constraints {
            if let GraphConstraint::EdgePropertyRequired { label, key } = constraint
                && label == &edge.label
                && !edge.props.contains_key(key)
            {
                return Err(GrustError::Schema(format!(
                    "edge '{}' from '{}' to '{}' is missing required constrained property '{}'",
                    edge.label.as_str(),
                    edge.from.as_str(),
                    edge.to.as_str(),
                    key
                )));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct NodeType {
    pub label: Label,
    pub fields: Vec<Field>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct EdgeType {
    pub label: Label,
    pub from: Vec<Label>,
    pub to: Vec<Label>,
    pub fields: Vec<Field>,
    pub directed: bool,
    pub uniqueness: EdgeUniqueness,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphConstraint {
    NodePropertyUnique { label: Label, key: String },
    NodePropertyRequired { label: Label, key: String },
    EdgePropertyUnique { label: Label, key: String },
    EdgePropertyRequired { label: Label, key: String },
}

impl GraphConstraint {
    pub fn label(&self) -> &Label {
        match self {
            Self::NodePropertyUnique { label, .. }
            | Self::NodePropertyRequired { label, .. }
            | Self::EdgePropertyUnique { label, .. }
            | Self::EdgePropertyRequired { label, .. } => label,
        }
    }

    pub fn key(&self) -> &str {
        match self {
            Self::NodePropertyUnique { key, .. }
            | Self::NodePropertyRequired { key, .. }
            | Self::EdgePropertyUnique { key, .. }
            | Self::EdgePropertyRequired { key, .. } => key,
        }
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphConstraintCapability {
    #[default]
    MetadataOnly,
    ValidateBeforeWrite,
    EnforcedByBackend,
}

/// Backend-native DDL support for a portable graph constraint.
///
/// This is intentionally separate from [`GraphConstraintCapability`].
/// A backend may validate a constraint before writes without having native DDL,
/// or it may create a query index that helps lookups but does not enforce the
/// constraint. Callers that need database-enforced guarantees should require
/// [`GraphNativeConstraintCapability::NativeConstraint`] and use
/// [`GraphStore::apply_native_constraint`] explicitly.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphNativeConstraintCapability {
    /// The backend has no native DDL mapping for this constraint.
    #[default]
    Unsupported,
    /// The backend can create a native index that may improve related lookups
    /// but does not enforce the constraint.
    NativeIndex,
    /// The backend can create a native constraint or equivalent database
    /// object that enforces the portable constraint.
    NativeConstraint,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct GraphNativeConstraintRequest {
    pub constraint: GraphConstraint,
    pub if_not_exists: bool,
}

#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct GraphNativeConstraintReport {
    pub applied: usize,
    pub skipped: usize,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Field {
    pub name: String,
    pub ty: FieldType,
    pub required: bool,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum FieldType {
    String,
    Int,
    Float,
    Bool,
    DateTime,
    StringArray,
    IntArray,
    FloatArray,
    Json,
}

/// How many edges of one type may exist between a pair of nodes.
///
/// `validate_graph` enforces `FromTo` and `FromLabelTo` identically — at most
/// one edge of the type between a given endpoint pair (unordered when the
/// type is undirected). The distinction is a storage-key hint for backends
/// that keep all edge labels in one table: `FromLabelTo` keys include the
/// label, `FromTo` keys do not.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum EdgeUniqueness {
    None,
    FromTo,
    FromLabelTo,
}

#[derive(Clone, Debug, Default)]
pub struct GraphSchemaBuilder {
    nodes: Vec<NodeType>,
    edges: Vec<EdgeType>,
    constraints: Vec<GraphConstraint>,
}

impl GraphSchemaBuilder {
    pub fn node(mut self, label: impl Into<Label>, fields: impl Into<Vec<Field>>) -> Self {
        self.nodes.push(NodeType {
            label: label.into(),
            fields: fields.into(),
        });
        self
    }

    pub fn edge(
        mut self,
        label: impl Into<Label>,
        from: impl Into<Vec<Label>>,
        to: impl Into<Vec<Label>>,
        fields: impl Into<Vec<Field>>,
    ) -> Self {
        self.edges.push(EdgeType {
            label: label.into(),
            from: from.into(),
            to: to.into(),
            fields: fields.into(),
            directed: true,
            uniqueness: EdgeUniqueness::FromLabelTo,
        });
        self
    }

    pub fn edge_type(mut self, edge_type: EdgeType) -> Self {
        self.edges.push(edge_type);
        self
    }

    pub fn constraint(mut self, constraint: GraphConstraint) -> Self {
        self.constraints.push(constraint);
        self
    }

    pub fn unique_node_property(self, label: impl Into<Label>, key: impl Into<String>) -> Self {
        self.constraint(GraphConstraint::NodePropertyUnique {
            label: label.into(),
            key: key.into(),
        })
    }

    pub fn required_node_property(self, label: impl Into<Label>, key: impl Into<String>) -> Self {
        self.constraint(GraphConstraint::NodePropertyRequired {
            label: label.into(),
            key: key.into(),
        })
    }

    pub fn unique_edge_property(self, label: impl Into<Label>, key: impl Into<String>) -> Self {
        self.constraint(GraphConstraint::EdgePropertyUnique {
            label: label.into(),
            key: key.into(),
        })
    }

    pub fn required_edge_property(self, label: impl Into<Label>, key: impl Into<String>) -> Self {
        self.constraint(GraphConstraint::EdgePropertyRequired {
            label: label.into(),
            key: key.into(),
        })
    }

    pub fn build(self) -> GraphSchema {
        GraphSchema {
            nodes: self.nodes,
            edges: self.edges,
            constraints: self.constraints,
        }
    }
}

impl Field {
    pub fn required(name: impl Into<String>, ty: FieldType) -> Self {
        Self {
            name: name.into(),
            ty,
            required: true,
        }
    }

    pub fn optional(name: impl Into<String>, ty: FieldType) -> Self {
        Self {
            name: name.into(),
            ty,
            required: false,
        }
    }
}

fn validate_props(props: &Props, fields: &[Field], context: &str) -> Result<()> {
    for field in fields {
        match props.get(&field.name) {
            Some(value) => validate_field_value(value, &field.ty, context, &field.name)?,
            None if field.required => {
                return Err(GrustError::Schema(format!(
                    "{context} missing required field '{}'",
                    field.name
                )));
            }
            None => {}
        }
    }
    Ok(())
}

fn validate_field_value(
    value: &Value,
    ty: &FieldType,
    context: &str,
    field_name: &str,
) -> Result<()> {
    let matches = match (value, ty) {
        (Value::String(_), FieldType::String)
        | (Value::Int(_), FieldType::Int)
        | (Value::Float(_), FieldType::Float)
        | (Value::Bool(_), FieldType::Bool)
        | (Value::DateTime(_), FieldType::DateTime)
        | (Value::StringArray(_), FieldType::StringArray)
        | (Value::IntArray(_), FieldType::IntArray)
        | (Value::FloatArray(_), FieldType::FloatArray)
        | (_, FieldType::Json) => true,
        // Plain strings remain valid date-times for backward compatibility,
        // but must still parse as RFC 3339.
        (Value::String(value), FieldType::DateTime) => is_rfc3339_datetime(value),
        _ => false,
    };
    if matches {
        Ok(())
    } else {
        Err(GrustError::Schema(format!(
            "{context} field '{field_name}' expected {ty:?}, got {value:?}"
        )))
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Traversal {
    pub start: Start,
    pub steps: Vec<Step>,
    pub limit: Option<u32>,
}

impl Traversal {
    pub fn from_node(id: impl Into<NodeId>) -> Self {
        Self {
            start: Start::Node(id.into()),
            steps: Vec::new(),
            limit: None,
        }
    }

    pub fn out(mut self, edge: impl Into<Label>) -> Self {
        self.steps.push(Step {
            direction: Direction::Out,
            edge: Some(edge.into()),
            node: None,
        });
        self
    }

    pub fn in_(mut self, edge: impl Into<Label>) -> Self {
        self.steps.push(Step {
            direction: Direction::In,
            edge: Some(edge.into()),
            node: None,
        });
        self
    }

    pub fn both(mut self, edge: impl Into<Label>) -> Self {
        self.steps.push(Step {
            direction: Direction::Both,
            edge: Some(edge.into()),
            node: None,
        });
        self
    }

    /// Constrains the target node label of the most recent step.
    ///
    /// # Panics
    ///
    /// Panics if called before any step has been added with `out`, `in_`, or
    /// `both`, since there is no step for the label to apply to.
    pub fn to(mut self, node: impl Into<Label>) -> Self {
        let step = self
            .steps
            .last_mut()
            .expect("Traversal::to() must follow out(), in_(), or both()");
        step.node = Some(node.into());
        self
    }

    pub fn limit(mut self, limit: u32) -> Self {
        self.limit = Some(limit);
        self
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum Start {
    Node(NodeId),
    NodesByLabel(Label),
    NodesByProperty {
        label: Label,
        key: String,
        value: Value,
    },
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Step {
    pub direction: Direction,
    pub edge: Option<Label>,
    pub node: Option<Label>,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum Direction {
    Out,
    In,
    Both,
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct EdgeQuery {
    pub from: Option<NodeId>,
    pub to: Option<NodeId>,
    pub label: Option<Label>,
}

/// Outcome of writing a single node or edge.
///
/// `Inserted` and `Updated` are precise only for stores that can cheaply
/// distinguish create from replace. Remote upsert-oriented backends commonly
/// return `Upserted` because doing otherwise would require an extra read or a
/// backend-specific write primitive. Portable callers should treat all three
/// written outcomes as success and should not rely on `Inserted` or `Updated`
/// unless they are intentionally targeting a backend that documents those
/// precise outcomes.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum PutOutcome {
    /// The element did not exist before and was created.
    Inserted,
    /// An element with the same identity existed and was overwritten or
    /// merged.
    Updated,
    /// The element was written by an upsert and the backend cannot tell
    /// whether it was an insert or an update.
    Upserted,
    /// The element was dropped by a dedupe policy and nothing was written.
    Deduped,
}

impl PutOutcome {
    /// True unless the element was deduped away.
    pub fn written(self) -> bool {
        !matches!(self, Self::Deduped)
    }
}

/// Counts of nodes and edges written by a bulk load. Each count reflects an
/// upsert applied to the backend, not distinct newly created elements.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct LoadReport {
    pub nodes: usize,
    pub edges: usize,
}

#[async_trait]
pub trait GraphStore: Send + Sync {
    /// Applies backend schema metadata.
    ///
    /// `GraphSchema` is a portable declaration of expected node labels, edge
    /// labels, fields, endpoint labels, direction, and uniqueness. The default
    /// implementation is a no-op for schemaless stores. Backend implementations
    /// may use the schema for validation, typed native tables, views, indexes,
    /// generated query shapes, or database-native schema definitions.
    ///
    /// Applying a schema does not imply the same enforcement guarantee on every
    /// backend. Callers that need portable preflight validation should call
    /// [`GraphSchema::validate_graph`] before writing, or use
    /// [`GraphStore::put_typed_graph`], which does that validation before
    /// applying the backend schema and writing the graph. Individual backends
    /// document whether they also validate each subsequent write at runtime.
    async fn apply_schema(&self, _schema: &GraphSchema) -> Result<()> {
        Ok(())
    }

    /// Reports how this backend treats a portable graph constraint.
    ///
    /// The default is metadata-only: the backend may remember or lower the
    /// constraint as schema metadata, but callers should not assume runtime
    /// enforcement. Backends that validate through [`GraphSchema`] before each
    /// write can report [`GraphConstraintCapability::ValidateBeforeWrite`].
    /// Backends with database-native guarantees can report
    /// [`GraphConstraintCapability::EnforcedByBackend`].
    fn constraint_capability(&self, _constraint: &GraphConstraint) -> GraphConstraintCapability {
        GraphConstraintCapability::MetadataOnly
    }

    /// Reports whether this backend can turn a portable graph constraint into
    /// backend-native DDL.
    ///
    /// This is an explicit opt-in surface. [`GraphStore::apply_schema`] remains
    /// the portable schema/validation hook and must not be assumed to create
    /// backend-native constraints. The default implementation reports no native
    /// support.
    fn native_constraint_capability(
        &self,
        _constraint: &GraphConstraint,
    ) -> GraphNativeConstraintCapability {
        GraphNativeConstraintCapability::Unsupported
    }

    /// Applies one backend-native constraint or index request.
    ///
    /// Backends should implement this only when they can describe the native
    /// object they create and its enforcement behavior through
    /// [`GraphNativeConstraintCapability`]. The default returns an explicit
    /// unsupported error so callers cannot mistake metadata-only schema
    /// application for native DDL.
    async fn apply_native_constraint(
        &self,
        request: GraphNativeConstraintRequest,
    ) -> Result<GraphNativeConstraintReport> {
        match self.native_constraint_capability(&request.constraint) {
            GraphNativeConstraintCapability::Unsupported => Err(GrustError::Unsupported(format!(
                "backend-native DDL is not supported for graph constraint {:?}",
                request.constraint
            ))),
            GraphNativeConstraintCapability::NativeIndex
            | GraphNativeConstraintCapability::NativeConstraint => Err(GrustError::Unsupported(
                "backend advertises native graph constraint support but does not implement apply_native_constraint"
                    .to_string(),
            )),
        }
    }

    /// Writes one node.
    ///
    /// The returned [`PutOutcome`] reports the most precise result the backend
    /// can provide. Remote upsert backends generally return
    /// [`PutOutcome::Upserted`] for both inserts and updates.
    async fn put_node(&self, node: &Node) -> Result<PutOutcome>;

    /// Writes one edge.
    ///
    /// The returned [`PutOutcome`] reports the most precise result the backend
    /// can provide. Remote upsert backends generally return
    /// [`PutOutcome::Upserted`] for both inserts and updates.
    async fn put_edge(&self, edge: &Edge) -> Result<PutOutcome>;

    async fn put_graph(&self, graph: &Graph) -> Result<LoadReport> {
        let mut report = LoadReport::default();
        for node in &graph.nodes {
            if self.put_node(node).await?.written() {
                report.nodes += 1;
            }
        }
        for edge in &graph.edges {
            if self.put_edge(edge).await?.written() {
                report.edges += 1;
            }
        }
        Ok(report)
    }

    async fn put_typed_graph(&self, schema: &GraphSchema, graph: &Graph) -> Result<LoadReport> {
        schema.validate_graph(graph)?;
        self.apply_schema(schema).await?;
        self.put_graph(graph).await
    }

    async fn get_node(&self, id: &NodeId) -> Result<Option<Node>>;

    /// Reads multiple nodes by ID.
    ///
    /// The default implementation preserves the input order and calls
    /// [`GraphStore::get_node`] once per ID. Backends with a native batch-read
    /// path should override this to avoid per-node round trips during traversal
    /// and other fan-out reads.
    async fn get_nodes(&self, ids: &[NodeId]) -> Result<Vec<Node>> {
        let mut nodes = Vec::new();
        for id in ids {
            if let Some(node) = self.get_node(id).await? {
                nodes.push(node);
            }
        }
        Ok(nodes)
    }

    async fn get_edges(&self, query: EdgeQuery) -> Result<Vec<Edge>>;
    async fn traverse(&self, traversal: Traversal) -> Result<Vec<Node>>;
}

#[async_trait]
pub trait GraphAdminStore: GraphStore {
    async fn bootstrap(&self) -> Result<()> {
        Ok(())
    }

    async fn clear(&self) -> Result<()>;
}

/// A single incremental change to a graph, for delta-oriented pipelines.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum GraphMutation {
    UpsertNode(Node),
    PatchNode {
        id: NodeId,
        props: Props,
    },
    PatchMatchingNodes {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        patch: Props,
    },
    UpdateMatchingNodeProperty {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        target_key: String,
        source_key: String,
        op: GraphNumericOp,
        operand: Value,
    },
    PatchEdge {
        from: NodeId,
        label: Label,
        to: NodeId,
        id: Option<EdgeId>,
        props: Props,
    },
    PatchMatchingEdges {
        relationship: GraphRelationshipMatch,
        patch: Props,
    },
    UpdateMatchingEdgeProperty {
        relationship: GraphRelationshipMatch,
        target_key: String,
        source_key: String,
        op: GraphNumericOp,
        operand: Value,
    },
    RemoveNodeProps {
        id: NodeId,
        keys: Vec<String>,
    },
    RemoveMatchingNodeProps {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        keys: Vec<String>,
    },
    RemoveEdgeProps {
        from: NodeId,
        label: Label,
        to: NodeId,
        id: Option<EdgeId>,
        keys: Vec<String>,
    },
    RemoveMatchingEdgeProps {
        relationship: GraphRelationshipMatch,
        keys: Vec<String>,
    },
    DeleteMatchingNodes {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
    },
    DeleteNode(NodeId),
    UpsertEdge(Edge),
    UpsertEdgesFromNodeMatches {
        kind: GraphMutationPlanKind,
        from: GraphNodeMatch,
        to: GraphNodeMatch,
        label: Label,
        props: Props,
        edge_id_policy: GraphRowEdgeIdPolicy,
    },
    DeleteEdge {
        from: NodeId,
        label: Label,
        to: NodeId,
    },
    DeleteMatchingEdges {
        relationship: GraphRelationshipMatch,
    },
    DeleteRelationshipRows {
        relationship: GraphRelationshipMatch,
        delete_edges: bool,
        endpoint_nodes: Vec<GraphRelationshipEndpoint>,
    },
    /// Cross-variable correlated node property update (Unit 10b/W3).
    SetMatchingNodeFromNode {
        target_label: Option<Label>,
        target_props: Props,
        target_predicates: Vec<GraphPropertyPredicate>,
        target_key: String,
        source_label: Option<Label>,
        source_props: Props,
        source_predicates: Vec<GraphPropertyPredicate>,
        source_key: String,
        op: Option<GraphNumericOp>,
        operand: Value,
        correlation: GraphWriteCorrelation,
        cardinality: GraphMutationCardinality,
    },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphMutationAtomicity {
    OrderedNonAtomic,
    Transactional,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphMutationPlanKind {
    Create,
    Merge,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphRowEdgeIdPolicy {
    ExplicitOnly,
    GenerateForCreate,
    GenerateForCreateAndMerge,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphRelationshipEndpoint {
    From,
    To,
}

impl Default for GraphRowEdgeIdPolicy {
    fn default() -> Self {
        Self::ExplicitOnly
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphMutationCardinality {
    SingleIdentity,
    BoundedMany,
    UnboundedMany,
}

pub fn generated_row_edge_id(from: &NodeId, label: &Label, to: &NodeId, props: &Props) -> EdgeId {
    const FNV_OFFSET: u64 = 0xcbf29ce484222325;
    const FNV_PRIME: u64 = 0x100000001b3;

    fn write_part(hash: &mut u64, value: &str) {
        for byte in value.as_bytes() {
            *hash ^= u64::from(*byte);
            *hash = hash.wrapping_mul(FNV_PRIME);
        }
        *hash ^= 0xff;
        *hash = hash.wrapping_mul(FNV_PRIME);
    }

    let mut hash = FNV_OFFSET;
    write_part(&mut hash, from.as_str());
    write_part(&mut hash, label.as_str());
    write_part(&mut hash, to.as_str());
    for (key, value) in props {
        write_part(&mut hash, key);
        write_part(&mut hash, &value.to_json().to_string());
    }
    EdgeId::new(format!("edge-{hash:016x}"))
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphNumericOp {
    Add,
    Subtract,
    Multiply,
    Divide,
}

pub fn evaluate_numeric_update(
    current: &Value,
    op: GraphNumericOp,
    operand: &Value,
) -> Result<Value> {
    match (current, operand) {
        (Value::Int(lhs), Value::Int(rhs)) if op != GraphNumericOp::Divide => {
            let value = match op {
                GraphNumericOp::Add => lhs.checked_add(*rhs),
                GraphNumericOp::Subtract => lhs.checked_sub(*rhs),
                GraphNumericOp::Multiply => lhs.checked_mul(*rhs),
                GraphNumericOp::Divide => unreachable!("division handled as floating point"),
            }
            .ok_or_else(|| GrustError::CypherExecution("numeric expression overflow".into()))?;
            Ok(Value::Int(value))
        }
        (Value::Int(lhs), Value::Int(rhs)) => numeric_float_result(*lhs as f64, op, *rhs as f64),
        (Value::Int(lhs), Value::Float(rhs)) => numeric_float_result(*lhs as f64, op, *rhs),
        (Value::Float(lhs), Value::Int(rhs)) => numeric_float_result(*lhs, op, *rhs as f64),
        (Value::Float(lhs), Value::Float(rhs)) => numeric_float_result(*lhs, op, *rhs),
        (Value::Null, _) | (_, Value::Null) => Err(GrustError::CypherExecution(
            "numeric expression cannot read null values".into(),
        )),
        _ => Err(GrustError::CypherExecution(
            "numeric expression requires integer or float values".into(),
        )),
    }
}

fn numeric_float_result(lhs: f64, op: GraphNumericOp, rhs: f64) -> Result<Value> {
    let value = match op {
        GraphNumericOp::Add => lhs + rhs,
        GraphNumericOp::Subtract => lhs - rhs,
        GraphNumericOp::Multiply => lhs * rhs,
        GraphNumericOp::Divide => {
            if rhs == 0.0 {
                return Err(GrustError::CypherExecution(
                    "numeric expression division by zero".into(),
                ));
            }
            lhs / rhs
        }
    };
    if value.is_finite() {
        Ok(Value::Float(value))
    } else {
        Err(GrustError::CypherExecution(
            "numeric expression produced a non-finite float".into(),
        ))
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphPredicateOp {
    Equal,
    NotEqual,
    IsNull,
    IsNotNull,
    StartsWith,
    NotStartsWith,
    StartsWithAny,
    NotStartsWithAny,
    EndsWith,
    NotEndsWith,
    EndsWithAny,
    NotEndsWithAny,
    Contains,
    NotContains,
    ContainsAny,
    NotContainsAny,
    In,
    NotIn,
    GreaterThan,
    GreaterThanOrEqual,
    LessThan,
    LessThanOrEqual,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GraphPropertyPredicate {
    pub key: String,
    pub op: GraphPredicateOp,
    pub value: Value,
}

impl GraphPropertyPredicate {
    pub fn matches(&self, actual: Option<&Value>) -> bool {
        if matches!(self.op, GraphPredicateOp::IsNull) {
            return actual.is_none_or(|value| matches!(value, Value::Null));
        }
        let Some(actual) = actual else {
            return false;
        };
        match self.op {
            GraphPredicateOp::Equal => actual == &self.value,
            GraphPredicateOp::NotEqual => actual != &self.value,
            GraphPredicateOp::IsNull => matches!(actual, Value::Null),
            GraphPredicateOp::IsNotNull => !matches!(actual, Value::Null),
            GraphPredicateOp::StartsWith => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| actual.starts_with(needle)),
            GraphPredicateOp::NotStartsWith => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| !actual.starts_with(needle)),
            GraphPredicateOp::StartsWithAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().any(|needle| actual.starts_with(needle))
                }),
            GraphPredicateOp::NotStartsWithAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().all(|needle| !actual.starts_with(needle))
                }),
            GraphPredicateOp::EndsWith => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| actual.ends_with(needle)),
            GraphPredicateOp::NotEndsWith => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| !actual.ends_with(needle)),
            GraphPredicateOp::EndsWithAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().any(|needle| actual.ends_with(needle))
                }),
            GraphPredicateOp::NotEndsWithAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().all(|needle| !actual.ends_with(needle))
                }),
            GraphPredicateOp::Contains => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| actual.contains(needle)),
            GraphPredicateOp::NotContains => string_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needle)| !actual.contains(needle)),
            GraphPredicateOp::ContainsAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().any(|needle| actual.contains(needle))
                }),
            GraphPredicateOp::NotContainsAny => string_list_predicate_values(actual, &self.value)
                .is_some_and(|(actual, needles)| {
                    needles.iter().all(|needle| !actual.contains(needle))
                }),
            GraphPredicateOp::In => list_predicate_values(&self.value)
                .is_some_and(|values| values.iter().any(|value| actual == value)),
            GraphPredicateOp::NotIn => list_predicate_values(&self.value)
                .is_some_and(|values| values.iter().all(|value| actual != value)),
            GraphPredicateOp::GreaterThan
            | GraphPredicateOp::GreaterThanOrEqual
            | GraphPredicateOp::LessThan
            | GraphPredicateOp::LessThanOrEqual => compare_ordered_values(actual, &self.value)
                .is_some_and(|ordering| match self.op {
                    GraphPredicateOp::GreaterThan => ordering.is_gt(),
                    GraphPredicateOp::GreaterThanOrEqual => ordering.is_gt() || ordering.is_eq(),
                    GraphPredicateOp::LessThan => ordering.is_lt(),
                    GraphPredicateOp::LessThanOrEqual => ordering.is_lt() || ordering.is_eq(),
                    GraphPredicateOp::Equal
                    | GraphPredicateOp::NotEqual
                    | GraphPredicateOp::IsNull
                    | GraphPredicateOp::IsNotNull
                    | GraphPredicateOp::StartsWith
                    | GraphPredicateOp::NotStartsWith
                    | GraphPredicateOp::StartsWithAny
                    | GraphPredicateOp::NotStartsWithAny
                    | GraphPredicateOp::EndsWith
                    | GraphPredicateOp::NotEndsWith
                    | GraphPredicateOp::EndsWithAny
                    | GraphPredicateOp::NotEndsWithAny
                    | GraphPredicateOp::Contains
                    | GraphPredicateOp::NotContains
                    | GraphPredicateOp::ContainsAny
                    | GraphPredicateOp::NotContainsAny
                    | GraphPredicateOp::In
                    | GraphPredicateOp::NotIn => unreachable!(),
                }),
        }
    }
}

fn string_predicate_values<'a>(actual: &'a Value, value: &'a Value) -> Option<(&'a str, &'a str)> {
    match (actual, value) {
        (Value::String(actual), Value::String(needle)) => Some((actual.as_str(), needle.as_str())),
        _ => None,
    }
}

fn string_list_predicate_values<'a>(
    actual: &'a Value,
    value: &'a Value,
) -> Option<(&'a str, Vec<&'a str>)> {
    match (actual, value) {
        (Value::String(actual), Value::StringArray(needles)) => Some((
            actual.as_str(),
            needles.iter().map(String::as_str).collect(),
        )),
        _ => None,
    }
}

fn list_predicate_values(value: &Value) -> Option<Vec<Value>> {
    match value {
        Value::StringArray(values) => Some(values.iter().map(Value::from).collect()),
        Value::IntArray(values) => Some(values.iter().copied().map(Value::Int).collect()),
        Value::FloatArray(values) => Some(values.iter().copied().map(Value::Float).collect()),
        Value::Json(serde_json::Value::Array(values)) => values
            .iter()
            .map(|value| match value {
                serde_json::Value::Bool(value) => Some(Value::Bool(*value)),
                serde_json::Value::Number(value) => value
                    .as_i64()
                    .map(Value::Int)
                    .or_else(|| value.as_f64().map(Value::Float)),
                serde_json::Value::String(value) => Some(Value::from(value)),
                serde_json::Value::Null
                | serde_json::Value::Array(_)
                | serde_json::Value::Object(_) => None,
            })
            .collect(),
        _ => None,
    }
}

fn compare_ordered_values(lhs: &Value, rhs: &Value) -> Option<std::cmp::Ordering> {
    match (lhs, rhs) {
        (Value::Int(lhs), Value::Int(rhs)) => Some(lhs.cmp(rhs)),
        (Value::Int(lhs), Value::Float(rhs)) => (*lhs as f64).partial_cmp(rhs),
        (Value::Float(lhs), Value::Int(rhs)) => lhs.partial_cmp(&(*rhs as f64)),
        (Value::Float(lhs), Value::Float(rhs)) => lhs.partial_cmp(rhs),
        (Value::String(lhs), Value::String(rhs)) => Some(lhs.cmp(rhs)),
        _ => None,
    }
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct GraphNodeMatch {
    pub label: Option<Label>,
    pub props: Props,
    pub predicates: Vec<GraphPropertyPredicate>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GraphRelationshipMatch {
    pub from: GraphNodeMatch,
    pub label: Label,
    pub to: GraphNodeMatch,
    pub id: Option<EdgeId>,
    pub props: Props,
    pub predicates: Vec<GraphPropertyPredicate>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum GraphMutationPlanOp {
    UpsertNode {
        kind: GraphMutationPlanKind,
        node: Node,
    },
    PatchNode {
        id: NodeId,
        props: Props,
    },
    PatchMatchingNodes {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        patch: Props,
        cardinality: GraphMutationCardinality,
    },
    UpdateMatchingNodeProperty {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        target_key: String,
        source_key: String,
        op: GraphNumericOp,
        operand: Value,
        cardinality: GraphMutationCardinality,
    },
    PatchEdge {
        from: NodeId,
        label: Label,
        to: NodeId,
        id: Option<EdgeId>,
        props: Props,
    },
    PatchMatchingEdges {
        relationship: GraphRelationshipMatch,
        patch: Props,
        cardinality: GraphMutationCardinality,
    },
    UpdateMatchingEdgeProperty {
        relationship: GraphRelationshipMatch,
        target_key: String,
        source_key: String,
        op: GraphNumericOp,
        operand: Value,
        cardinality: GraphMutationCardinality,
    },
    RemoveNodeProps {
        id: NodeId,
        keys: Vec<String>,
    },
    RemoveMatchingNodeProps {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        keys: Vec<String>,
        cardinality: GraphMutationCardinality,
    },
    RemoveEdgeProps {
        from: NodeId,
        label: Label,
        to: NodeId,
        id: Option<EdgeId>,
        keys: Vec<String>,
    },
    RemoveMatchingEdgeProps {
        relationship: GraphRelationshipMatch,
        keys: Vec<String>,
        cardinality: GraphMutationCardinality,
    },
    DeleteMatchingNodes {
        label: Option<Label>,
        props: Props,
        predicates: Vec<GraphPropertyPredicate>,
        cardinality: GraphMutationCardinality,
    },
    UpsertEdge {
        kind: GraphMutationPlanKind,
        edge: Edge,
    },
    UpsertEdgesFromNodeMatches {
        kind: GraphMutationPlanKind,
        from: GraphNodeMatch,
        to: GraphNodeMatch,
        label: Label,
        props: Props,
        edge_id_policy: GraphRowEdgeIdPolicy,
        cardinality: GraphMutationCardinality,
    },
    DeleteNode(NodeId),
    DeleteEdge {
        from: NodeId,
        label: Label,
        to: NodeId,
    },
    DeleteMatchingEdges {
        relationship: GraphRelationshipMatch,
        cardinality: GraphMutationCardinality,
    },
    DeleteRelationshipRows {
        relationship: GraphRelationshipMatch,
        delete_edges: bool,
        endpoint_nodes: Vec<GraphRelationshipEndpoint>,
        target_count: usize,
        cardinality: GraphMutationCardinality,
    },
    /// Cross-variable correlated node property update (Unit 10b/W3): for each
    /// matched `(target, source)` node pair, set
    /// `target[target_key] = source[source_key]` optionally combined with
    /// `op`/`operand`. `correlation` selects how the pairs are formed.
    SetMatchingNodeFromNode {
        target_label: Option<Label>,
        target_props: Props,
        target_predicates: Vec<GraphPropertyPredicate>,
        target_key: String,
        source_label: Option<Label>,
        source_props: Props,
        source_predicates: Vec<GraphPropertyPredicate>,
        source_key: String,
        op: Option<GraphNumericOp>,
        operand: Value,
        correlation: GraphWriteCorrelation,
        cardinality: GraphMutationCardinality,
    },
}

/// How a cross-variable write correlates its target and source node matches
/// (Unit 10b/W3).
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum GraphWriteCorrelation {
    /// Cartesian product of the target and source matches.
    Cartesian,
    /// Pairs linked by `(target)-[:label]->(source)`.
    OutgoingRelationship { label: Label },
    /// Pairs linked by `(target)<-[:label]-(source)`.
    IncomingRelationship { label: Label },
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct GraphMutationPlan {
    pub operations: Vec<GraphMutationPlanOp>,
}

impl GraphMutationPlan {
    pub fn new(operations: Vec<GraphMutationPlanOp>) -> Self {
        Self { operations }
    }

    pub fn push(&mut self, operation: GraphMutationPlanOp) {
        self.operations.push(operation);
    }

    pub fn report(&self) -> GraphMutationReport {
        let mut report = GraphMutationReport::default();
        for operation in &self.operations {
            report.record(operation);
        }
        report
    }

    pub fn into_mutations(self) -> Vec<GraphMutation> {
        self.operations
            .into_iter()
            .map(GraphMutation::from)
            .collect()
    }
}

/// Count-oriented mutation reporting.
///
/// Reports created from a [`GraphMutationPlan`] can know exact changed
/// node/edge counts only for single-identity operations. Matched or
/// row-producing operations increment their coarse operation counters at
/// planning time, while executors fill in `matched_rows` and granular
/// `changed_*` / `*_patches` / `*_deletes` / `*_upserts` counters after they
/// materialize the backend row set.
#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct GraphMutationReport {
    pub creates: usize,
    pub merges: usize,
    pub deletes: usize,
    pub patches: usize,
    pub property_removes: usize,
    pub matched_rows: usize,
    pub changed_nodes: usize,
    pub changed_edges: usize,
    pub node_upserts: usize,
    pub edge_upserts: usize,
    pub node_deletes: usize,
    pub edge_deletes: usize,
    pub node_patches: usize,
    pub edge_patches: usize,
    pub node_property_removes: usize,
    pub edge_property_removes: usize,
    /// Upserts the executor could classify as inserting a new node, i.e. no
    /// element with the same identity existed before the write.
    ///
    /// Only backends that can cheaply distinguish insert from update populate
    /// these (for example the in-memory store). Upsert-oriented backends that
    /// return [`PutOutcome::Upserted`] leave them at `0` and report the totals
    /// through `node_upserts` / `edge_upserts` instead, so a zero here does not
    /// mean "no inserts" — check the backend's classification ability first.
    pub node_inserts: usize,
    /// Upserts the executor could classify as updating an existing node.
    pub node_updates: usize,
    /// Upserts the executor could classify as inserting a new edge.
    pub edge_inserts: usize,
    /// Upserts the executor could classify as updating an existing edge.
    pub edge_updates: usize,
}

impl GraphMutationReport {
    pub fn record(&mut self, operation: &GraphMutationPlanOp) {
        match operation {
            GraphMutationPlanOp::UpsertNode { kind, .. }
            | GraphMutationPlanOp::UpsertEdge { kind, .. } => {
                match kind {
                    GraphMutationPlanKind::Create => self.creates += 1,
                    GraphMutationPlanKind::Merge => self.merges += 1,
                }
                match operation {
                    GraphMutationPlanOp::UpsertNode { .. } => {
                        self.node_upserts += 1;
                        self.changed_nodes += 1;
                    }
                    GraphMutationPlanOp::UpsertEdge { .. } => {
                        self.edge_upserts += 1;
                        self.changed_edges += 1;
                    }
                    _ => {}
                }
            }
            GraphMutationPlanOp::UpsertEdgesFromNodeMatches { kind, .. } => match kind {
                GraphMutationPlanKind::Create => self.creates += 1,
                GraphMutationPlanKind::Merge => self.merges += 1,
            },
            GraphMutationPlanOp::PatchNode { .. } => {
                self.patches += 1;
                self.node_patches += 1;
                self.changed_nodes += 1;
            }
            GraphMutationPlanOp::PatchMatchingNodes { .. } => {
                self.patches += 1;
            }
            GraphMutationPlanOp::UpdateMatchingNodeProperty { .. } => {
                self.patches += 1;
            }
            GraphMutationPlanOp::SetMatchingNodeFromNode { .. } => {
                self.patches += 1;
            }
            GraphMutationPlanOp::PatchEdge { .. } => {
                self.patches += 1;
                self.edge_patches += 1;
                self.changed_edges += 1;
            }
            GraphMutationPlanOp::PatchMatchingEdges { .. } => {
                self.patches += 1;
            }
            GraphMutationPlanOp::UpdateMatchingEdgeProperty { .. } => {
                self.patches += 1;
            }
            GraphMutationPlanOp::RemoveNodeProps { .. } => {
                self.property_removes += 1;
                self.node_property_removes += 1;
                self.changed_nodes += 1;
            }
            GraphMutationPlanOp::RemoveMatchingNodeProps { .. } => {
                self.property_removes += 1;
            }
            GraphMutationPlanOp::RemoveEdgeProps { .. } => {
                self.property_removes += 1;
                self.edge_property_removes += 1;
                self.changed_edges += 1;
            }
            GraphMutationPlanOp::RemoveMatchingEdgeProps { .. } => {
                self.property_removes += 1;
            }
            GraphMutationPlanOp::DeleteMatchingNodes { .. } => {
                self.deletes += 1;
            }
            GraphMutationPlanOp::DeleteNode(_) => {
                self.deletes += 1;
                self.node_deletes += 1;
                self.changed_nodes += 1;
            }
            GraphMutationPlanOp::DeleteEdge { .. } => {
                self.deletes += 1;
                self.edge_deletes += 1;
                self.changed_edges += 1;
            }
            GraphMutationPlanOp::DeleteMatchingEdges { .. } => {
                self.deletes += 1;
            }
            GraphMutationPlanOp::DeleteRelationshipRows { target_count, .. } => {
                self.deletes += *target_count;
            }
        }
    }
}

impl From<GraphMutationPlanOp> for GraphMutation {
    fn from(operation: GraphMutationPlanOp) -> Self {
        match operation {
            GraphMutationPlanOp::UpsertNode { node, .. } => Self::UpsertNode(node),
            GraphMutationPlanOp::PatchNode { id, props } => Self::PatchNode { id, props },
            GraphMutationPlanOp::PatchMatchingNodes {
                label,
                props,
                predicates,
                patch,
                ..
            } => Self::PatchMatchingNodes {
                label,
                props,
                predicates,
                patch,
            },
            GraphMutationPlanOp::UpdateMatchingNodeProperty {
                label,
                props,
                predicates,
                target_key,
                source_key,
                op,
                operand,
                ..
            } => Self::UpdateMatchingNodeProperty {
                label,
                props,
                predicates,
                target_key,
                source_key,
                op,
                operand,
            },
            GraphMutationPlanOp::PatchEdge {
                from,
                label,
                to,
                id,
                props,
            } => Self::PatchEdge {
                from,
                label,
                to,
                id,
                props,
            },
            GraphMutationPlanOp::PatchMatchingEdges {
                relationship,
                patch,
                ..
            } => Self::PatchMatchingEdges {
                relationship,
                patch,
            },
            GraphMutationPlanOp::UpdateMatchingEdgeProperty {
                relationship,
                target_key,
                source_key,
                op,
                operand,
                ..
            } => Self::UpdateMatchingEdgeProperty {
                relationship,
                target_key,
                source_key,
                op,
                operand,
            },
            GraphMutationPlanOp::RemoveNodeProps { id, keys } => Self::RemoveNodeProps { id, keys },
            GraphMutationPlanOp::RemoveEdgeProps {
                from,
                label,
                to,
                id,
                keys,
            } => Self::RemoveEdgeProps {
                from,
                label,
                to,
                id,
                keys,
            },
            GraphMutationPlanOp::RemoveMatchingEdgeProps {
                relationship, keys, ..
            } => Self::RemoveMatchingEdgeProps { relationship, keys },
            GraphMutationPlanOp::RemoveMatchingNodeProps {
                label,
                props,
                predicates,
                keys,
                ..
            } => Self::RemoveMatchingNodeProps {
                label,
                props,
                predicates,
                keys,
            },
            GraphMutationPlanOp::DeleteMatchingNodes {
                label,
                props,
                predicates,
                ..
            } => Self::DeleteMatchingNodes {
                label,
                props,
                predicates,
            },
            GraphMutationPlanOp::UpsertEdge { edge, .. } => Self::UpsertEdge(edge),
            GraphMutationPlanOp::UpsertEdgesFromNodeMatches {
                kind,
                from,
                to,
                label,
                props,
                edge_id_policy,
                ..
            } => Self::UpsertEdgesFromNodeMatches {
                kind,
                from,
                to,
                label,
                props,
                edge_id_policy,
            },
            GraphMutationPlanOp::DeleteNode(id) => Self::DeleteNode(id),
            GraphMutationPlanOp::DeleteEdge { from, label, to } => {
                Self::DeleteEdge { from, label, to }
            }
            GraphMutationPlanOp::DeleteMatchingEdges { relationship, .. } => {
                Self::DeleteMatchingEdges { relationship }
            }
            GraphMutationPlanOp::DeleteRelationshipRows {
                relationship,
                delete_edges,
                endpoint_nodes,
                ..
            } => Self::DeleteRelationshipRows {
                relationship,
                delete_edges,
                endpoint_nodes,
            },
            GraphMutationPlanOp::SetMatchingNodeFromNode {
                target_label,
                target_props,
                target_predicates,
                target_key,
                source_label,
                source_props,
                source_predicates,
                source_key,
                op,
                operand,
                correlation,
                cardinality,
            } => Self::SetMatchingNodeFromNode {
                target_label,
                target_props,
                target_predicates,
                target_key,
                source_label,
                source_props,
                source_predicates,
                source_key,
                op,
                operand,
                correlation,
                cardinality,
            },
        }
    }
}

/// Executes already-resolved Cypher/graph mutation plans.
///
/// This trait intentionally accepts [`GraphMutationPlan`] rather than Cypher
/// text. Parser ownership can stay with a backend or adapter until there are
/// enough consumers to justify a shared parser crate, while stores that
/// understand Grust mutation semantics can still share execution behavior.
#[async_trait]
pub trait CypherMutationExecutor: GraphMutationStore {
    async fn execute_cypher_mutation_plan(
        &self,
        plan: &GraphMutationPlan,
    ) -> Result<GraphMutationReport> {
        let mut report = plan.report();
        for operation in &plan.operations {
            match operation {
                GraphMutationPlanOp::DeleteMatchingNodes { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched node deletes require backend-specific query support".to_string(),
                    ));
                }
                GraphMutationPlanOp::PatchMatchingNodes { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched node patches require backend-specific query support".to_string(),
                    ));
                }
                GraphMutationPlanOp::UpdateMatchingNodeProperty { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched node expression updates require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutationPlanOp::RemoveMatchingNodeProps { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched node property removals require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutationPlanOp::PatchMatchingEdges { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched edge patches require backend-specific query support".to_string(),
                    ));
                }
                GraphMutationPlanOp::UpdateMatchingEdgeProperty { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched edge expression updates require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutationPlanOp::RemoveMatchingEdgeProps { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched edge property removals require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutationPlanOp::DeleteMatchingEdges { .. } => {
                    return Err(GrustError::CypherExecution(
                        "matched edge deletes require backend-specific query support".to_string(),
                    ));
                }
                GraphMutationPlanOp::UpsertEdgesFromNodeMatches { .. } => {
                    return Err(GrustError::CypherExecution(
                        "row-producing edge upserts require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutationPlanOp::UpsertNode { node, .. } => {
                    classify_node_upsert(self.put_node(node).await?, &mut report);
                }
                GraphMutationPlanOp::UpsertEdge { edge, .. } => {
                    classify_edge_upsert(self.put_edge(edge).await?, &mut report);
                }
                _ => {
                    let mutation = GraphMutation::from(operation.clone());
                    self.apply_mutations(std::slice::from_ref(&mutation))
                        .await?;
                }
            }
        }
        Ok(report)
    }
}

/// Records a single-node upsert outcome into a report's precise insert/update
/// counters. [`PutOutcome::Upserted`] and [`PutOutcome::Deduped`] carry no
/// insert-vs-update information and leave the counters unchanged.
pub fn classify_node_upsert(outcome: PutOutcome, report: &mut GraphMutationReport) {
    match outcome {
        PutOutcome::Inserted => report.node_inserts += 1,
        PutOutcome::Updated => report.node_updates += 1,
        PutOutcome::Upserted | PutOutcome::Deduped => {}
    }
}

/// Records a single-edge upsert outcome into a report's precise insert/update
/// counters. See [`classify_node_upsert`].
pub fn classify_edge_upsert(outcome: PutOutcome, report: &mut GraphMutationReport) {
    match outcome {
        PutOutcome::Inserted => report.edge_inserts += 1,
        PutOutcome::Updated => report.edge_updates += 1,
        PutOutcome::Upserted | PutOutcome::Deduped => {}
    }
}

/// Incremental mutation support for stores that can delete elements.
///
/// Deletes are idempotent: removing an element that does not exist is not an
/// error.
#[async_trait]
pub trait GraphMutationStore: GraphStore {
    fn mutation_atomicity(&self) -> GraphMutationAtomicity {
        GraphMutationAtomicity::OrderedNonAtomic
    }

    /// Deletes a node and all edges incident to it.
    async fn delete_node(&self, id: &NodeId) -> Result<()>;

    /// Deletes the edge(s) matching `(from, label, to)`.
    async fn delete_edge(&self, from: &NodeId, label: &Label, to: &NodeId) -> Result<()>;

    /// Applies mutations in order, stopping at the first error.
    ///
    /// The default implementation calls the single-mutation methods one at a
    /// time and is not atomic: if a later mutation fails, earlier successful
    /// mutations are not rolled back. Backends with transaction support should
    /// override this method and apply the whole slice in one transaction.
    async fn apply_mutations(&self, mutations: &[GraphMutation]) -> Result<()> {
        for mutation in mutations {
            match mutation {
                GraphMutation::UpsertNode(node) => {
                    self.put_node(node).await?;
                }
                GraphMutation::PatchNode { id, props } => {
                    if let Some(mut node) = self.get_node(id).await? {
                        for (key, value) in props {
                            node.props.insert(key.clone(), value.clone());
                        }
                        self.put_node(&node).await?;
                    }
                }
                GraphMutation::PatchMatchingNodes { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched node patches require backend-specific query support".to_string(),
                    ));
                }
                GraphMutation::UpdateMatchingNodeProperty { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched node expression updates require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::SetMatchingNodeFromNode { .. } => {
                    return Err(GrustError::Unsupported(
                        "cross-variable correlated updates require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::PatchEdge {
                    from,
                    label,
                    to,
                    id,
                    props,
                } => {
                    let mut edges = self
                        .get_edges(EdgeQuery {
                            from: Some(from.clone()),
                            to: Some(to.clone()),
                            label: Some(label.clone()),
                        })
                        .await?;
                    if let Some(id) = id {
                        edges.retain(|edge| edge.id.as_ref() == Some(id));
                    }
                    match edges.len() {
                        0 => {}
                        1 => {
                            let mut edge = edges.remove(0);
                            for (key, value) in props {
                                edge.props.insert(key.clone(), value.clone());
                            }
                            self.put_edge(&edge).await?;
                        }
                        count => {
                            return Err(GrustError::CypherUnsupportedCardinality(format!(
                                "edge patch matched {count} edges; add an explicit edge id"
                            )));
                        }
                    }
                }
                GraphMutation::PatchMatchingEdges { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched edge patches require backend-specific query support".to_string(),
                    ));
                }
                GraphMutation::UpdateMatchingEdgeProperty { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched edge expression updates require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::RemoveNodeProps { id, keys } => {
                    if let Some(mut node) = self.get_node(id).await? {
                        for key in keys {
                            node.props.remove(key);
                        }
                        self.put_node(&node).await?;
                    }
                }
                GraphMutation::RemoveMatchingNodeProps { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched node property removal requires backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::RemoveEdgeProps {
                    from,
                    label,
                    to,
                    id,
                    keys,
                } => {
                    let mut edges = self
                        .get_edges(EdgeQuery {
                            from: Some(from.clone()),
                            to: Some(to.clone()),
                            label: Some(label.clone()),
                        })
                        .await?;
                    if let Some(id) = id {
                        edges.retain(|edge| edge.id.as_ref() == Some(id));
                    }
                    match edges.len() {
                        0 => {}
                        1 => {
                            let mut edge = edges.remove(0);
                            for key in keys {
                                edge.props.remove(key);
                            }
                            self.put_edge(&edge).await?;
                        }
                        count => {
                            return Err(GrustError::CypherUnsupportedCardinality(format!(
                                "edge property removal matched {count} edges; add an explicit edge id"
                            )));
                        }
                    }
                }
                GraphMutation::RemoveMatchingEdgeProps { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched edge property removal requires backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::DeleteMatchingNodes { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched node deletes require backend-specific query support".to_string(),
                    ));
                }
                GraphMutation::DeleteNode(id) => self.delete_node(id).await?,
                GraphMutation::UpsertEdge(edge) => {
                    self.put_edge(edge).await?;
                }
                GraphMutation::UpsertEdgesFromNodeMatches { .. } => {
                    return Err(GrustError::Unsupported(
                        "row-producing edge upserts require backend-specific query support"
                            .to_string(),
                    ));
                }
                GraphMutation::DeleteEdge { from, label, to } => {
                    self.delete_edge(from, label, to).await?
                }
                GraphMutation::DeleteMatchingEdges { .. } => {
                    return Err(GrustError::Unsupported(
                        "matched edge deletes require backend-specific query support".to_string(),
                    ));
                }
                GraphMutation::DeleteRelationshipRows { .. } => {
                    return Err(GrustError::Unsupported(
                        "relationship-row deletes require backend-specific query support"
                            .to_string(),
                    ));
                }
            }
        }
        Ok(())
    }
}

pub mod prelude {
    pub use crate::{
        CypherMutationExecutor, Decimal, Direction, Duration, Edge, EdgeId, EdgePolicy, EdgeQuery, EdgeType,
        EdgeUniqueness, Field, FieldType, Graph, GraphAdminStore, GraphBuilder, GraphConstraint,
        GraphConstraintCapability, GraphIndex, GraphMutation, GraphMutationAtomicity,
        GraphMutationCardinality, GraphMutationPlan, GraphMutationPlanKind, GraphMutationPlanOp,
        GraphMutationReport, GraphMutationStore, GraphNativeConstraintCapability,
        GraphNativeConstraintReport, GraphNativeConstraintRequest, GraphNodeMatch, GraphNumericOp,
        GraphPredicateOp, GraphPropertyPredicate, GraphRelationshipEndpoint,
        GraphRelationshipMatch, GraphRowEdgeIdPolicy, GraphSchema, GraphSchemaBuilder, GraphStore,
        GraphWriteCorrelation,
        GrustError, Label, LoadReport, Node, NodeId, NodeType, Props, PutOutcome, Result, RfcDate,
        Start, Step, Traversal, Value, classify_edge_upsert, classify_node_upsert, edge_key,
        evaluate_numeric_update, generated_row_edge_id, relationship_type, schema_identifier,
    };

    #[cfg(feature = "typed-garde")]
    pub use crate::typed::{TypedEdge, TypedGraphBuilder, TypedNode, garde, props_from_serialize};

    #[cfg(feature = "typed-zod-rs")]
    pub use crate::typed::{parse_typed_json, parse_typed_json_with, zod_rs};
}

#[cfg(test)]
mod tests;