reddb_server/application/

entity.rs

use std::collections::HashMap;

use crate::application::ports::RuntimeEntityPort;
use crate::json::{parse_json, to_vec as json_to_vec, Map, Value as JsonValue};
use crate::storage::unified::devx::refs::{NodeRef, TableRef, VectorRef};
use crate::storage::unified::{Metadata, MetadataValue, RefTarget, SparseVector, VectorData};
use crate::storage::{EntityId, UnifiedEntity};
use crate::{RedDBError, RedDBResult};
use reddb_types::Value;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

pub(crate) struct AutocommitUpdate {
    manager: Arc<crate::storage::transaction::snapshot::SnapshotManager>,
    pub(crate) xid: u64,
    finished: AtomicBool,
}

impl AutocommitUpdate {
    pub(crate) fn new(
        manager: Arc<crate::storage::transaction::snapshot::SnapshotManager>,
    ) -> Arc<Self> {
        let xid = manager.begin_cache_coherent_update();
        Arc::new(Self {
            manager,
            xid,
            finished: AtomicBool::new(false),
        })
    }

    pub(crate) fn commit(&self) {
        if !self.finished.swap(true, Ordering::AcqRel) {
            self.manager.commit(self.xid);
        }
    }
}

impl std::fmt::Debug for AutocommitUpdate {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AutocommitUpdate")
            .field("xid", &self.xid)
            .finish()
    }
}

impl Drop for AutocommitUpdate {
    fn drop(&mut self) {
        if !self.finished.load(Ordering::Acquire) {
            self.manager.rollback(self.xid);
        }
    }
}

#[derive(Debug, Clone)]
pub struct CreateEntityOutput {
    pub id: EntityId,
    pub entity: Option<UnifiedEntity>,
}

#[derive(Debug, Clone)]
pub struct AppliedEntityMutation {
    pub(crate) autocommit_update: Option<Arc<AutocommitUpdate>>,
    pub id: EntityId,
    pub collection: String,
    pub entity: UnifiedEntity,
    pub metadata: Option<crate::storage::unified::Metadata>,
    pub modified_columns: Vec<String>,
    pub persist_metadata: bool,
    pub context_index_dirty: bool,
    /// Prior physical version retained for MVCC history. Present when an
    /// UPDATE creates a new physical entity for the same logical row.
    pub replaced_entity: Option<UnifiedEntity>,
    /// xmax carried by `replaced_entity` before this mutation stamped it.
    pub replaced_entity_previous_xmax: u64,
    /// Snapshot of the row's named fields BEFORE the mutation was
    /// applied. Carried so the post-write secondary-index hook can
    /// `delete(old) + insert(new)` for changed indexed columns.
    /// Empty when the entity isn't a row or the row carried neither a
    /// `named` map nor a `schema` Arc.
    pub pre_mutation_fields: Vec<(String, Value)>,
}

/// Damage-vector for a row update: the minimal diff between the
/// pre-mutation and post-mutation field state. Downstream consumers
/// (secondary-index maintainer, CDC emitter, eventually the pager's
/// in-place update path) work off this struct so they only touch
/// columns that actually changed.
///
/// Future work (Fase 5): the pager can decide whether `changed` fits
/// in the existing cell's slotted footprint and patch bytes in place
/// instead of rewriting the whole row.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct RowDamageVector {
    /// Columns present in both `old` and `new` whose value differs.
    /// `(column, old_value, new_value)`.
    pub changed: Vec<(String, Value, Value)>,
    /// Columns present in `new` but not in `old`.
    pub added: Vec<(String, Value)>,
    /// Columns present in `old` but not in `new`.
    pub removed: Vec<(String, Value)>,
}

impl RowDamageVector {
    /// True when no columns changed, were added, or were removed.
    /// Callers can short-circuit index/CDC/in-place work when this
    /// holds.
    pub fn is_empty(&self) -> bool {
        self.changed.is_empty() && self.added.is_empty() && self.removed.is_empty()
    }

    /// Names of every column the update touched. Equivalent to the
    /// current `modified_columns` list carried on `AppliedEntityMutation`
    /// but computed from the ground truth of old/new rather than
    /// from what the SQL executor thinks it wrote.
    pub fn touched_columns(&self) -> Vec<&str> {
        let mut out: Vec<&str> =
            Vec::with_capacity(self.changed.len() + self.added.len() + self.removed.len());
        out.extend(self.changed.iter().map(|(c, _, _)| c.as_str()));
        out.extend(self.added.iter().map(|(c, _)| c.as_str()));
        out.extend(self.removed.iter().map(|(c, _)| c.as_str()));
        out
    }
}

/// Compute the damage-vector between an old and new row snapshot.
/// Both inputs are field lists (as carried by `CreateRowInput` and
/// `AppliedEntityMutation.pre_mutation_fields`); the order of fields
/// is not significant. Columns present in both sides with identical
/// values don't appear in any bucket.
pub fn row_damage_vector(
    old_fields: &[(String, Value)],
    new_fields: &[(String, Value)],
) -> RowDamageVector {
    // HashMap<&str, &Value> keyed by column name for O(1) membership
    // checks. Both inputs are expected to be small (tens of columns),
    // so the allocation overhead is negligible vs the O(N*M) pairwise
    // comparison we'd otherwise need.
    let old_map: HashMap<&str, &Value> = old_fields.iter().map(|(k, v)| (k.as_str(), v)).collect();
    let new_map: HashMap<&str, &Value> = new_fields.iter().map(|(k, v)| (k.as_str(), v)).collect();

    let mut changed = Vec::new();
    let mut added = Vec::new();
    let mut removed = Vec::new();

    for (name, new_value) in &new_map {
        match old_map.get(name) {
            Some(old_value) if old_value == new_value => {}
            Some(old_value) => changed.push((
                (*name).to_string(),
                (*old_value).clone(),
                (*new_value).clone(),
            )),
            None => added.push(((*name).to_string(), (*new_value).clone())),
        }
    }
    for (name, old_value) in &old_map {
        if !new_map.contains_key(name) {
            removed.push(((*name).to_string(), (*old_value).clone()));
        }
    }

    RowDamageVector {
        changed,
        added,
        removed,
    }
}

pub use reddb_engine::catalog::enforcement::{RowUpdateColumnRule, RowUpdateContractPlan};

#[derive(Debug, Clone)]
pub struct CreateRowInput {
    pub collection: String,
    pub fields: Vec<(String, Value)>,
    pub metadata: Vec<(String, MetadataValue)>,
    pub node_links: Vec<NodeRef>,
    pub vector_links: Vec<VectorRef>,
}

#[derive(Debug, Clone)]
pub struct CreateRowsBatchInput {
    pub collection: String,
    pub rows: Vec<CreateRowInput>,
    /// When true, no event subscriptions fire for this batch (SUPPRESS EVENTS).
    pub suppress_events: bool,
}

#[derive(Debug, Clone)]
pub struct CreateNodeEmbeddingInput {
    pub name: String,
    pub vector: Vec<f32>,
    pub model: Option<String>,
}

#[derive(Debug, Clone)]
pub struct CreateNodeTableLinkInput {
    pub key: String,
    pub table: TableRef,
}

#[derive(Debug, Clone)]
pub struct CreateNodeGraphLinkInput {
    pub target: EntityId,
    pub edge_label: String,
    pub weight: f32,
}

#[derive(Debug, Clone)]
pub struct CreateNodeInput {
    pub collection: String,
    pub label: String,
    pub node_type: Option<String>,
    pub properties: Vec<(String, Value)>,
    pub metadata: Vec<(String, MetadataValue)>,
    pub embeddings: Vec<CreateNodeEmbeddingInput>,
    pub table_links: Vec<CreateNodeTableLinkInput>,
    pub node_links: Vec<CreateNodeGraphLinkInput>,
}

#[derive(Debug, Clone)]
pub struct CreateEdgeInput {
    pub collection: String,
    pub label: String,
    pub from: EntityId,
    pub to: EntityId,
    pub weight: Option<f32>,
    pub properties: Vec<(String, Value)>,
    pub metadata: Vec<(String, MetadataValue)>,
}

#[derive(Debug, Clone)]
pub struct CreateVectorInput {
    pub collection: String,
    pub dense: Vec<f32>,
    pub content: Option<String>,
    pub metadata: Vec<(String, MetadataValue)>,
    pub link_row: Option<TableRef>,
    pub link_node: Option<NodeRef>,
}

#[derive(Debug, Clone)]
pub struct CreateDocumentInput {
    pub collection: String,
    pub body: JsonValue,
    pub metadata: Vec<(String, MetadataValue)>,
    pub node_links: Vec<NodeRef>,
    pub vector_links: Vec<VectorRef>,
}

#[derive(Debug, Clone)]
pub struct CreateKvInput {
    pub collection: String,
    pub key: String,
    pub value: Value,
    pub metadata: Vec<(String, MetadataValue)>,
}

#[derive(Debug, Clone)]
pub struct CreateTimeSeriesPointInput {
    pub collection: String,
    pub metric: String,
    pub value: f64,
    pub timestamp_ns: Option<u64>,
    pub tags: Vec<(String, String)>,
    pub metadata: Vec<(String, MetadataValue)>,
}

#[derive(Debug, Clone)]
pub struct DeleteEntityInput {
    pub collection: String,
    pub id: EntityId,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PatchEntityOperationType {
    Set,
    Replace,
    Unset,
}

#[derive(Debug, Clone)]
pub struct PatchEntityOperation {
    pub op: PatchEntityOperationType,
    pub path: Vec<String>,
    pub value: Option<JsonValue>,
}

#[derive(Debug, Clone)]
pub struct PatchEntityInput {
    pub collection: String,
    pub id: EntityId,
    pub payload: JsonValue,
    pub operations: Vec<PatchEntityOperation>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeleteEntityOutput {
    pub deleted: bool,
    pub id: EntityId,
}

pub struct EntityUseCases<'a, P: ?Sized> {
    runtime: &'a P,
}

impl<'a, P: RuntimeEntityPort + ?Sized> EntityUseCases<'a, P> {
    pub fn new(runtime: &'a P) -> Self {
        Self { runtime }
    }

    pub fn create_row(&self, input: CreateRowInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_row(input)
    }

    pub fn create_rows_batch(
        &self,
        input: CreateRowsBatchInput,
    ) -> RedDBResult<Vec<CreateEntityOutput>> {
        self.runtime.create_rows_batch(input)
    }

    pub fn create_node(&self, input: CreateNodeInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_node(input)
    }

    pub fn create_edge(&self, input: CreateEdgeInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_edge(input)
    }

    pub fn create_vector(&self, input: CreateVectorInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_vector(input)
    }

    pub fn create_document(&self, input: CreateDocumentInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_document(input)
    }

    pub fn create_kv(&self, input: CreateKvInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_kv(input)
    }

    pub fn create_timeseries_point(
        &self,
        input: CreateTimeSeriesPointInput,
    ) -> RedDBResult<CreateEntityOutput> {
        self.runtime.create_timeseries_point(input)
    }

    pub fn get_kv(&self, collection: &str, key: &str) -> RedDBResult<Option<(Value, EntityId)>> {
        self.runtime.get_kv(collection, key)
    }

    pub fn delete_kv(&self, collection: &str, key: &str) -> RedDBResult<bool> {
        self.runtime.delete_kv(collection, key)
    }

    pub fn patch(&self, input: PatchEntityInput) -> RedDBResult<CreateEntityOutput> {
        self.runtime.patch_entity(input)
    }

    pub fn delete(&self, input: DeleteEntityInput) -> RedDBResult<DeleteEntityOutput> {
        self.runtime.delete_entity(input)
    }
}

pub(crate) fn json_to_storage_value(value: &JsonValue) -> RedDBResult<Value> {
    match value {
        JsonValue::Null => Ok(Value::Null),
        JsonValue::Bool(value) => Ok(Value::Boolean(*value)),
        JsonValue::Integer(value) => Ok(Value::Integer(*value)),
        JsonValue::Number(value) => {
            if value.fract().abs() < f64::EPSILON {
                Ok(Value::Integer(*value as i64))
            } else {
                Ok(Value::Float(*value))
            }
        }
        JsonValue::Decimal(value) => Ok(Value::DecimalText(value.clone())),
        JsonValue::String(value) => Ok(Value::text(value.clone())),
        JsonValue::Array(_) | JsonValue::Object(_) => json_to_vec(value)
            .map(Value::Json)
            .map_err(|err| RedDBError::Query(format!("failed to serialize JSON value: {err}"))),
    }
}

pub(crate) fn json_to_metadata_value(value: &JsonValue) -> RedDBResult<MetadataValue> {
    match value {
        JsonValue::Null => Ok(MetadataValue::Null),
        JsonValue::Bool(value) => Ok(MetadataValue::Bool(*value)),
        JsonValue::Integer(value) => Ok(MetadataValue::Int(*value)),
        JsonValue::Number(value) => {
            if value.fract().abs() < f64::EPSILON {
                Ok(MetadataValue::Int(*value as i64))
            } else {
                Ok(MetadataValue::Float(*value))
            }
        }
        JsonValue::Decimal(value) => Ok(MetadataValue::String(value.clone())),
        JsonValue::String(value) => Ok(MetadataValue::String(value.clone())),
        JsonValue::Array(values) => {
            let mut items = Vec::with_capacity(values.len());
            for value in values {
                items.push(json_to_metadata_value(value)?);
            }
            Ok(MetadataValue::Array(items))
        }
        JsonValue::Object(map) => {
            let mut object = HashMap::with_capacity(map.len());
            for (key, value) in map {
                object.insert(key.clone(), json_to_metadata_value(value)?);
            }
            Ok(MetadataValue::Object(object))
        }
    }
}

pub(crate) fn apply_patch_operations_to_storage_map(
    fields: &mut HashMap<String, Value>,
    operations: &[PatchEntityOperation],
) -> RedDBResult<()> {
    if operations.is_empty() {
        return Ok(());
    }

    let mut patch_target = JsonValue::Object(
        fields
            .iter()
            .map(|(key, value)| (key.clone(), value.to_json()))
            .collect(),
    );
    apply_patch_operations_to_json(&mut patch_target, operations)
        .map_err(|error| RedDBError::Query(format!("patch fields failed: {error}")))?;

    let JsonValue::Object(object) = patch_target else {
        return Err(RedDBError::Query(
            "patch operations require object roots".to_string(),
        ));
    };

    let mut merged = HashMap::with_capacity(object.len());
    for (key, value) in object {
        merged.insert(key, json_to_storage_value(&value)?);
    }
    *fields = merged;
    Ok(())
}

/// `pub` re-export of [`apply_patch_operations_to_json`] for the HTTP layer.
/// The KV patch handler (issue #751) parses a JSON value, applies operations
/// against the live JSON tree, and writes back — without reaching the
/// entity-row patch core that this module otherwise owns.
pub fn apply_patch_operations_to_json_public(
    value: &mut JsonValue,
    operations: &[PatchEntityOperation],
) -> Result<(), String> {
    apply_patch_operations_to_json(value, operations)
}

pub(crate) fn apply_patch_operations_to_json(
    value: &mut JsonValue,
    operations: &[PatchEntityOperation],
) -> Result<(), String> {
    for operation in operations {
        if operation.path.is_empty() {
            return Err("patch path cannot be empty".to_string());
        }

        match operation.op {
            PatchEntityOperationType::Set | PatchEntityOperationType::Replace => {
                let Some(patch_value) = &operation.value else {
                    return Err("set/replace operations require a value".to_string());
                };
                apply_patch_json_set(value, &operation.path, patch_value.clone())?;
            }
            PatchEntityOperationType::Unset => {
                apply_patch_json_unset(value, &operation.path)?;
            }
        }
    }
    Ok(())
}

pub(crate) fn apply_patch_operations_to_vector_fields(
    vector: &mut VectorData,
    operations: &[PatchEntityOperation],
) -> RedDBResult<()> {
    if operations.is_empty() {
        return Ok(());
    }

    let mut vector_target = JsonValue::Object({
        let mut object = Map::new();
        object.insert(
            "dense".to_string(),
            JsonValue::Array(
                vector
                    .dense
                    .iter()
                    .map(|value| JsonValue::Number(*value as f64))
                    .collect(),
            ),
        );
        object.insert(
            "sparse".to_string(),
            vector.sparse.as_ref().map_or(JsonValue::Null, |sparse| {
                let mut object = Map::new();
                object.insert(
                    "indices".to_string(),
                    JsonValue::Array(
                        sparse
                            .indices
                            .iter()
                            .map(|value| JsonValue::Number(*value as f64))
                            .collect(),
                    ),
                );
                object.insert(
                    "values".to_string(),
                    JsonValue::Array(
                        sparse
                            .values
                            .iter()
                            .map(|value| JsonValue::Number(*value as f64))
                            .collect(),
                    ),
                );
                object.insert(
                    "dimension".to_string(),
                    JsonValue::Number(sparse.dimension as f64),
                );
                JsonValue::Object(object)
            }),
        );
        object.insert(
            "content".to_string(),
            match vector.content.as_ref() {
                Some(value) => JsonValue::String(value.clone()),
                None => JsonValue::Null,
            },
        );
        object
    });

    let touched_dense = operations
        .iter()
        .any(|operation| operation.path.first().is_some_and(|key| key == "dense"));
    let touched_sparse = operations
        .iter()
        .any(|operation| operation.path.first().is_some_and(|key| key == "sparse"));
    let touched_content = operations
        .iter()
        .any(|operation| operation.path.first().is_some_and(|key| key == "content"));

    apply_patch_operations_to_json(&mut vector_target, operations)
        .map_err(|error| RedDBError::Query(format!("patch fields failed: {error}")))?;

    let JsonValue::Object(object) = vector_target else {
        return Err(RedDBError::Query(
            "patch operations require object roots".to_string(),
        ));
    };

    if touched_dense {
        let Some(value) = object.get("dense") else {
            return Err(RedDBError::Query(
                "field 'dense' cannot be unset".to_string(),
            ));
        };
        vector.dense = parse_patch_f32_vector(value, "dense")?;
    }

    if touched_content {
        vector.content = match object.get("content") {
            None | Some(JsonValue::Null) => None,
            Some(value) => Some(
                value
                    .as_str()
                    .ok_or_else(|| {
                        RedDBError::Query("field 'content' must be a string".to_string())
                    })?
                    .to_string(),
            ),
        };
    }

    if touched_sparse {
        vector.sparse = match object.get("sparse") {
            Some(value) => parse_sparse_vector_value(value)?,
            None => None,
        };
    }

    Ok(())
}

pub(crate) fn metadata_to_json(metadata: &Metadata) -> JsonValue {
    JsonValue::Object(
        metadata
            .iter()
            .map(|(key, value)| (key.clone(), metadata_value_to_json(value)))
            .collect(),
    )
}

pub(crate) fn metadata_from_json(payload: &JsonValue) -> RedDBResult<Metadata> {
    let JsonValue::Object(object) = payload else {
        return Err(RedDBError::Query(
            "metadata patch requires an object".to_string(),
        ));
    };

    let mut metadata = Metadata::new();
    for (key, value) in object {
        metadata.set(key.clone(), metadata_value_from_json(value)?);
    }
    Ok(metadata)
}

fn metadata_value_to_json(value: &MetadataValue) -> JsonValue {
    match value {
        MetadataValue::Null => JsonValue::Null,
        MetadataValue::Bool(value) => JsonValue::Bool(*value),
        MetadataValue::Int(value) => JsonValue::Number(*value as f64),
        MetadataValue::Float(value) => JsonValue::Number(*value),
        MetadataValue::String(value) => JsonValue::String(value.clone()),
        MetadataValue::Bytes(value) => {
            let mut object = Map::new();
            object.insert(
                "__redb_type".to_string(),
                JsonValue::String("bytes".to_string()),
            );
            object.insert(
                "value".to_string(),
                JsonValue::Array(
                    value
                        .iter()
                        .map(|value| JsonValue::Number(*value as f64))
                        .collect(),
                ),
            );
            JsonValue::Object(object)
        }
        MetadataValue::Array(values) => {
            JsonValue::Array(values.iter().map(metadata_value_to_json).collect())
        }
        MetadataValue::Object(object) => JsonValue::Object(
            object
                .iter()
                .map(|(key, value)| (key.clone(), metadata_value_to_json(value)))
                .collect(),
        ),
        MetadataValue::Timestamp(value) => JsonValue::Number(*value as f64),
        MetadataValue::Geo { lat, lon } => {
            let mut object = Map::new();
            object.insert(
                "__redb_type".to_string(),
                JsonValue::String("geo".to_string()),
            );
            object.insert("lat".to_string(), JsonValue::Number(*lat));
            object.insert("lon".to_string(), JsonValue::Number(*lon));
            JsonValue::Object(object)
        }
        MetadataValue::Reference(value) => {
            let mut object = Map::new();
            object.insert(
                "__redb_type".to_string(),
                JsonValue::String("reference".to_string()),
            );
            let (kind, collection, id) = match value {
                RefTarget::TableRow { table, row_id } => ("table_row", table.as_str(), *row_id),
                RefTarget::Node {
                    collection,
                    node_id,
                } => ("node", collection.as_str(), node_id.raw()),
                RefTarget::Edge {
                    collection,
                    edge_id,
                } => ("edge", collection.as_str(), edge_id.raw()),
                RefTarget::Vector {
                    collection,
                    vector_id,
                } => ("vector", collection.as_str(), vector_id.raw()),
                RefTarget::Entity {
                    collection,
                    entity_id,
                } => ("entity", collection.as_str(), entity_id.raw()),
            };
            object.insert("kind".to_string(), JsonValue::String(kind.to_string()));
            object.insert(
                "collection".to_string(),
                JsonValue::String(collection.to_string()),
            );
            object.insert("id".to_string(), JsonValue::Number(id as f64));
            JsonValue::Object(object)
        }
        MetadataValue::References(values) => {
            let mut object = Map::new();
            object.insert(
                "__redb_type".to_string(),
                JsonValue::String("references".to_string()),
            );
            object.insert(
                "values".to_string(),
                JsonValue::Array(
                    values
                        .iter()
                        .map(|r| metadata_value_to_json(&MetadataValue::Reference(r.clone())))
                        .collect(),
                ),
            );
            JsonValue::Object(object)
        }
    }
}

fn metadata_value_from_json(value: &JsonValue) -> RedDBResult<MetadataValue> {
    match value {
        JsonValue::Null => Ok(MetadataValue::Null),
        JsonValue::Bool(value) => Ok(MetadataValue::Bool(*value)),
        JsonValue::Integer(value) => Ok(MetadataValue::Int(*value)),
        JsonValue::Number(value) => {
            if value.fract().abs() < f64::EPSILON {
                Ok(MetadataValue::Int(*value as i64))
            } else {
                Ok(MetadataValue::Float(*value))
            }
        }
        JsonValue::Decimal(value) => Ok(MetadataValue::String(value.clone())),
        JsonValue::String(value) => Ok(MetadataValue::String(value.clone())),
        JsonValue::Array(values) => {
            let mut out = Vec::with_capacity(values.len());
            for value in values {
                out.push(metadata_value_from_json(value)?);
            }
            Ok(MetadataValue::Array(out))
        }
        JsonValue::Object(object) => {
            if let Some(marker) = object.get("__redb_type").and_then(JsonValue::as_str) {
                match marker {
                    "bytes" => {
                        let values = object
                            .get("value")
                            .and_then(JsonValue::as_array)
                            .ok_or_else(|| {
                                RedDBError::Query(
                                    "metadata marker 'bytes' requires array value".to_string(),
                                )
                            })?;
                        let mut out = Vec::with_capacity(values.len());
                        for value in values {
                            let value = value.as_i64().ok_or_else(|| {
                                RedDBError::Query(
                                    "metadata bytes must contain integer values".to_string(),
                                )
                            })?;
                            if !(0..=255).contains(&value) {
                                return Err(RedDBError::Query(
                                    "metadata bytes must contain values between 0 and 255"
                                        .to_string(),
                                ));
                            }
                            out.push(value as u8);
                        }
                        return Ok(MetadataValue::Bytes(out));
                    }
                    "geo" => {
                        let lat =
                            object
                                .get("lat")
                                .and_then(JsonValue::as_f64)
                                .ok_or_else(|| {
                                    RedDBError::Query(
                                        "metadata marker 'geo' requires numeric 'lat'".to_string(),
                                    )
                                })?;
                        let lon =
                            object
                                .get("lon")
                                .and_then(JsonValue::as_f64)
                                .ok_or_else(|| {
                                    RedDBError::Query(
                                        "metadata marker 'geo' requires numeric 'lon'".to_string(),
                                    )
                                })?;
                        return Ok(MetadataValue::Geo { lat, lon });
                    }
                    "reference" => {
                        return parse_metadata_reference(object).map(MetadataValue::Reference)
                    }
                    "references" => {
                        let values = object
                            .get("values")
                            .and_then(JsonValue::as_array)
                            .ok_or_else(|| {
                                RedDBError::Query(
                                    "metadata marker 'references' requires array 'values'"
                                        .to_string(),
                                )
                            })?;
                        let mut references = Vec::with_capacity(values.len());
                        for value in values {
                            references.push(parse_metadata_reference_value(value)?);
                        }
                        return Ok(MetadataValue::References(references));
                    }
                    _ => {}
                }
            }

            let mut out = HashMap::with_capacity(object.len());
            for (key, value) in object {
                out.insert(key.clone(), metadata_value_from_json(value)?);
            }
            Ok(MetadataValue::Object(out))
        }
    }
}

fn parse_metadata_reference(object: &Map<String, JsonValue>) -> RedDBResult<RefTarget> {
    let kind = object
        .get("kind")
        .and_then(JsonValue::as_str)
        .ok_or_else(|| RedDBError::Query("metadata reference requires 'kind'".to_string()))?;
    let collection = object
        .get("collection")
        .and_then(JsonValue::as_str)
        .ok_or_else(|| RedDBError::Query("metadata reference requires 'collection'".to_string()))?;
    let id = object
        .get("id")
        .ok_or_else(|| RedDBError::Query("metadata reference requires 'id'".to_string()))?;
    let id = parse_patch_u64_value(id, "id")?;

    let target = match kind {
        "table_row" | "table" => RefTarget::table(collection.to_string(), id),
        "node" => RefTarget::node(collection.to_string(), EntityId::new(id)),
        "edge" => RefTarget::Edge {
            collection: collection.to_string(),
            edge_id: EntityId::new(id),
        },
        "vector" => RefTarget::vector(collection.to_string(), EntityId::new(id)),
        "entity" => RefTarget::Entity {
            collection: collection.to_string(),
            entity_id: EntityId::new(id),
        },
        _ => {
            return Err(RedDBError::Query(format!(
                "unsupported metadata reference kind '{kind}'"
            )));
        }
    };

    Ok(target)
}

fn parse_metadata_reference_value(value: &JsonValue) -> RedDBResult<RefTarget> {
    let JsonValue::Object(object) = value else {
        return Err(RedDBError::Query(
            "metadata reference entries must be objects".to_string(),
        ));
    };
    parse_metadata_reference(object)
}

fn parse_patch_u64_value(value: &JsonValue, field: &str) -> RedDBResult<u64> {
    let Some(value) = value.as_f64() else {
        return Err(RedDBError::Query(format!(
            "field '{field}' must be a number"
        )));
    };
    if value.is_sign_negative() {
        return Err(RedDBError::Query(format!(
            "field '{field}' cannot be negative"
        )));
    }
    if value.fract().abs() > f64::EPSILON {
        return Err(RedDBError::Query(format!(
            "field '{field}' must be an integer"
        )));
    }
    if value > u64::MAX as f64 {
        return Err(RedDBError::Query(format!("field '{field}' is too large")));
    }
    Ok(value as u64)
}

fn parse_patch_f32_vector(value: &JsonValue, field: &str) -> RedDBResult<Vec<f32>> {
    let values = value
        .as_array()
        .ok_or_else(|| RedDBError::Query(format!("field '{field}' must be an array")))?;
    let mut out = Vec::with_capacity(values.len());
    for value in values {
        let number = value.as_f64().ok_or_else(|| {
            RedDBError::Query(format!("field '{field}' must contain only numbers"))
        })?;
        out.push(number as f32);
    }
    if out.is_empty() {
        return Err(RedDBError::Query(format!(
            "field '{field}' cannot be empty"
        )));
    }
    Ok(out)
}

fn parse_sparse_index_array(value: &JsonValue, field: &str) -> RedDBResult<Vec<u32>> {
    let values = value
        .as_array()
        .ok_or_else(|| RedDBError::Query(format!("field '{field}' must be an array")))?;
    let mut out = Vec::with_capacity(values.len());
    for value in values {
        let value = value.as_f64().ok_or_else(|| {
            RedDBError::Query(format!("field '{field}' must contain only integers"))
        })?;
        if value.is_sign_negative() || value.fract().abs() > f64::EPSILON {
            return Err(RedDBError::Query(format!(
                "field '{field}' must contain only u32 values"
            )));
        }
        if value > u32::MAX as f64 {
            return Err(RedDBError::Query(format!(
                "field '{field}' value is too large"
            )));
        }
        out.push(value as u32);
    }
    Ok(out)
}

fn parse_sparse_value_array(value: &JsonValue, field: &str) -> RedDBResult<Vec<f32>> {
    parse_patch_f32_vector(value, field)
}

fn parse_sparse_vector_value(value: &JsonValue) -> RedDBResult<Option<SparseVector>> {
    match value {
        JsonValue::Null => Ok(None),
        JsonValue::Object(object) => {
            let indices = parse_sparse_index_array(
                object.get("indices").ok_or_else(|| {
                    RedDBError::Query("sparse metadata requires 'indices'".to_string())
                })?,
                "sparse.indices",
            )?;
            let values = parse_sparse_value_array(
                object.get("values").ok_or_else(|| {
                    RedDBError::Query("sparse metadata requires 'values'".to_string())
                })?,
                "sparse.values",
            )?;
            if indices.len() != values.len() {
                return Err(RedDBError::Query(
                    "sparse indices and values lengths must match".to_string(),
                ));
            }
            let dimension = match object.get("dimension").and_then(JsonValue::as_f64) {
                Some(value) => {
                    if value.is_sign_negative() || value.fract().abs() > f64::EPSILON {
                        return Err(RedDBError::Query(
                            "sparse dimension must be a non-negative integer".to_string(),
                        ));
                    }
                    if value > usize::MAX as f64 {
                        return Err(RedDBError::Query(
                            "sparse dimension is too large".to_string(),
                        ));
                    }
                    value as usize
                }
                None => indices
                    .iter()
                    .max()
                    .map_or(0, |index| (*index as usize) + 1),
            };
            if indices.iter().any(|index| (*index as usize) >= dimension) {
                return Err(RedDBError::Query(
                    "sparse indices must be smaller than dimension".to_string(),
                ));
            }
            Ok(Some(SparseVector::new(indices, values, dimension)))
        }
        _ => Err(RedDBError::Query(
            "field 'sparse' must be an object or null".to_string(),
        )),
    }
}

fn apply_patch_json_set(
    target: &mut JsonValue,
    path: &[String],
    value: JsonValue,
) -> Result<(), String> {
    if path.is_empty() {
        return Err("patch path cannot be empty".to_string());
    }

    let mut current = target;
    for segment in &path[..path.len() - 1] {
        let JsonValue::Object(object) = current else {
            return Err("patch path target must be an object".to_string());
        };
        let value = object
            .entry(segment.clone())
            .or_insert_with(|| JsonValue::Object(Map::new()));
        if !matches!(value, JsonValue::Object(_)) {
            *value = JsonValue::Object(Map::new());
        }
        current = value;
    }

    let JsonValue::Object(object) = current else {
        return Err("patch path target must be an object".to_string());
    };
    object.insert(path[path.len() - 1].clone(), value);
    Ok(())
}

fn apply_patch_json_unset(target: &mut JsonValue, path: &[String]) -> Result<(), String> {
    if path.is_empty() {
        return Err("patch path cannot be empty".to_string());
    }

    if path.len() == 1 {
        let JsonValue::Object(object) = target else {
            return Err("patch path target must be an object".to_string());
        };
        object.remove(&path[0]);
        return Ok(());
    }

    let mut current = target;
    for segment in &path[..path.len() - 1] {
        let Some(value) = (match current {
            JsonValue::Object(object) => object.get_mut(segment),
            _ => {
                return Err("patch path target must be an object".to_string());
            }
        }) else {
            return Ok(());
        };

        if !matches!(value, JsonValue::Object(_)) {
            return Ok(());
        }
        current = value;
    }

    let JsonValue::Object(object) = current else {
        return Err("patch path target must be an object".to_string());
    };
    object.remove(&path[path.len() - 1]);
    Ok(())
}

fn format_mac(bytes: &[u8; 6]) -> String {
    bytes
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<Vec<_>>()
        .join(":")
}

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

    fn s(n: &str) -> String {
        n.to_string()
    }

    #[test]
    fn identical_rows_produce_empty_vector() {
        let old = vec![
            (s("name"), Value::text(s("alice"))),
            (s("age"), Value::Integer(30)),
        ];
        let new = old.clone();
        let dv = row_damage_vector(&old, &new);
        assert!(dv.is_empty());
        assert!(dv.touched_columns().is_empty());
    }

    #[test]
    fn detects_changed_column_only() {
        let old = vec![
            (s("name"), Value::text(s("alice"))),
            (s("age"), Value::Integer(30)),
        ];
        let new = vec![
            (s("name"), Value::text(s("alice"))),
            (s("age"), Value::Integer(31)),
        ];
        let dv = row_damage_vector(&old, &new);
        assert_eq!(dv.changed.len(), 1);
        assert_eq!(dv.changed[0].0, "age");
        assert!(dv.added.is_empty());
        assert!(dv.removed.is_empty());
        assert_eq!(dv.touched_columns(), vec!["age"]);
    }

    #[test]
    fn detects_added_and_removed_columns() {
        let old = vec![
            (s("name"), Value::text(s("alice"))),
            (s("nickname"), Value::text(s("al"))),
        ];
        let new = vec![
            (s("name"), Value::text(s("alice"))),
            (s("email"), Value::text(s("a@x.com"))),
        ];
        let dv = row_damage_vector(&old, &new);
        assert!(dv.changed.is_empty());
        assert_eq!(dv.added.len(), 1);
        assert_eq!(dv.added[0].0, "email");
        assert_eq!(dv.removed.len(), 1);
        assert_eq!(dv.removed[0].0, "nickname");
    }

    #[test]
    fn field_order_does_not_affect_diff() {
        // `(name, age)` and `(age, name)` with identical values should
        // diff as empty — column order is a presentation concern.
        let old = vec![
            (s("name"), Value::text(s("bob"))),
            (s("age"), Value::Integer(42)),
        ];
        let new = vec![
            (s("age"), Value::Integer(42)),
            (s("name"), Value::text(s("bob"))),
        ];
        assert!(row_damage_vector(&old, &new).is_empty());
    }

    #[test]
    fn mixed_changed_added_removed() {
        let old = vec![
            (s("a"), Value::Integer(1)),
            (s("b"), Value::Integer(2)),
            (s("gone"), Value::text(s("x"))),
        ];
        let new = vec![
            (s("a"), Value::Integer(10)),     // changed
            (s("b"), Value::Integer(2)),      // unchanged
            (s("new"), Value::Boolean(true)), // added
        ];
        let dv = row_damage_vector(&old, &new);
        assert_eq!(dv.changed.len(), 1);
        assert_eq!(dv.added.len(), 1);
        assert_eq!(dv.removed.len(), 1);
        let mut touched: Vec<&str> = dv.touched_columns();
        touched.sort();
        assert_eq!(touched, vec!["a", "gone", "new"]);
    }
}