nodedb 0.4.0

// SPDX-License-Identifier: BUSL-1.1

//! CRDT plan builders.

use nodedb_types::protocol::TextFields;
use sonic_rs;

use crate::bridge::envelope::PhysicalPlan;
use nodedb_physical::physical_plan::CrdtOp;

use super::DispatchCtx;
use super::require_doc_id;

pub(crate) fn build_read(fields: &TextFields, collection: &str) -> crate::Result<PhysicalPlan> {
    let document_id = require_doc_id(fields)?;
    Ok(PhysicalPlan::Crdt(CrdtOp::Read {
        collection: collection.to_string(),
        document_id,
    }))
}

pub(crate) fn build_apply(
    ctx: &DispatchCtx<'_>,
    fields: &TextFields,
    collection: &str,
) -> crate::Result<PhysicalPlan> {
    let document_id = require_doc_id(fields)?;
    let delta = fields
        .delta
        .as_ref()
        .ok_or_else(|| crate::Error::BadRequest {
            detail: "missing 'delta'".to_string(),
        })?
        .clone();
    let peer_id = fields.peer_id.unwrap_or(0);

    // Use provided mutation_id, or generate deterministic one from content hash.
    let mutation_id = fields.mutation_id.unwrap_or_else(|| {
        // Deterministic dedup key from (peer_id, delta).
        let mut combined = peer_id.to_le_bytes().to_vec();
        combined.extend_from_slice(&delta);
        crate::util::fnv1a_hash(&combined)
    });

    let surrogate = ctx.state.surrogate_assigner.assign(
        ctx.database_id(),
        ctx.tenant_id(),
        collection,
        document_id.as_bytes(),
    )?;

    Ok(PhysicalPlan::Crdt(CrdtOp::Apply {
        collection: collection.to_string(),
        document_id,
        delta,
        peer_id,
        mutation_id,
        surrogate,
        provenance: None,
        // Native admission does not yet stamp a constraint fence; `0` leaves
        // the apply-time write-gate open. Catalog-based stamping here is a
        // tracked follow-up (parity with the sync admission path).
        constraint_version_required: 0,
    }))
}

pub(crate) fn build_alter_policy(
    fields: &TextFields,
    collection: &str,
) -> crate::Result<PhysicalPlan> {
    let policy = fields
        .policy
        .as_ref()
        .ok_or_else(|| crate::Error::BadRequest {
            detail: "missing 'policy'".to_string(),
        })?;
    let policy_json = sonic_rs::to_string(policy).map_err(|e| crate::Error::BadRequest {
        detail: format!("invalid policy: {e}"),
    })?;
    Ok(PhysicalPlan::Crdt(CrdtOp::SetPolicy {
        collection: collection.to_string(),
        policy_json,
    }))
}

/// Extract `list_path` from request fields.
fn require_list_path(fields: &TextFields) -> crate::Result<String> {
    fields
        .list_path
        .as_ref()
        .cloned()
        .ok_or_else(|| crate::Error::BadRequest {
            detail: "missing 'list_path'".to_string(),
        })
}

/// Extract a list index field and narrow it to `usize`.
///
/// A missing or out-of-range value is a typed decode error — never a
/// silent default to `0`, since a position-based CRDT op landing at the
/// wrong index is data corruption, not a benign fallback.
fn require_list_index(value: Option<u64>, field_name: &str) -> crate::Result<usize> {
    let raw = value.ok_or_else(|| crate::Error::BadRequest {
        detail: format!("missing '{field_name}'"),
    })?;
    usize::try_from(raw).map_err(|_| crate::Error::BadRequest {
        detail: format!("'{field_name}' value {raw} exceeds platform usize range"),
    })
}

pub(crate) fn build_list_insert(
    ctx: &DispatchCtx<'_>,
    fields: &TextFields,
    collection: &str,
) -> crate::Result<PhysicalPlan> {
    let document_id = require_doc_id(fields)?;
    let list_path = require_list_path(fields)?;
    let index = require_list_index(fields.list_index, "list_index")?;
    let fields_json =
        fields
            .list_fields_json
            .as_ref()
            .cloned()
            .ok_or_else(|| crate::Error::BadRequest {
                detail: "missing 'list_fields_json'".to_string(),
            })?;

    let surrogate = ctx.state.surrogate_assigner.assign(
        ctx.database_id(),
        ctx.tenant_id(),
        collection,
        document_id.as_bytes(),
    )?;

    Ok(PhysicalPlan::Crdt(CrdtOp::ListInsert {
        collection: collection.to_string(),
        document_id,
        list_path,
        index,
        fields_json,
        surrogate,
    }))
}

pub(crate) fn build_list_delete(
    ctx: &DispatchCtx<'_>,
    fields: &TextFields,
    collection: &str,
) -> crate::Result<PhysicalPlan> {
    let document_id = require_doc_id(fields)?;
    let list_path = require_list_path(fields)?;
    let index = require_list_index(fields.list_index, "list_index")?;

    let surrogate = ctx.state.surrogate_assigner.assign(
        ctx.database_id(),
        ctx.tenant_id(),
        collection,
        document_id.as_bytes(),
    )?;

    Ok(PhysicalPlan::Crdt(CrdtOp::ListDelete {
        collection: collection.to_string(),
        document_id,
        list_path,
        index,
        surrogate,
    }))
}

pub(crate) fn build_list_move(
    ctx: &DispatchCtx<'_>,
    fields: &TextFields,
    collection: &str,
) -> crate::Result<PhysicalPlan> {
    let document_id = require_doc_id(fields)?;
    let list_path = require_list_path(fields)?;
    let from_index = require_list_index(fields.list_from_index, "list_from_index")?;
    let to_index = require_list_index(fields.list_to_index, "list_to_index")?;

    let surrogate = ctx.state.surrogate_assigner.assign(
        ctx.database_id(),
        ctx.tenant_id(),
        collection,
        document_id.as_bytes(),
    )?;

    Ok(PhysicalPlan::Crdt(CrdtOp::ListMove {
        collection: collection.to_string(),
        document_id,
        list_path,
        from_index,
        to_index,
        surrogate,
    }))
}

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

    fn list_insert_fields() -> TextFields {
        TextFields {
            document_id: Some("doc-1".to_string()),
            list_path: Some("items".to_string()),
            list_index: Some(3),
            list_fields_json: Some(r#"{"title":"hello"}"#.to_string()),
            ..Default::default()
        }
    }

    fn list_move_fields() -> TextFields {
        TextFields {
            document_id: Some("doc-1".to_string()),
            list_path: Some("items".to_string()),
            list_from_index: Some(1),
            list_to_index: Some(5),
            ..Default::default()
        }
    }

    #[test]
    fn list_insert_field_parsing_carries_every_field() {
        let fields = list_insert_fields();
        let document_id = require_doc_id(&fields).unwrap();
        let list_path = require_list_path(&fields).unwrap();
        let index = require_list_index(fields.list_index, "list_index").unwrap();
        let fields_json = fields.list_fields_json.clone().unwrap();

        let surrogate = Surrogate::new(42);
        let plan = PhysicalPlan::Crdt(CrdtOp::ListInsert {
            collection: "docs".to_string(),
            document_id: document_id.clone(),
            list_path: list_path.clone(),
            index,
            fields_json: fields_json.clone(),
            surrogate,
        });

        let PhysicalPlan::Crdt(CrdtOp::ListInsert {
            collection,
            document_id: got_document_id,
            list_path: got_list_path,
            index: got_index,
            fields_json: got_fields_json,
            surrogate: got_surrogate,
        }) = plan
        else {
            panic!("expected CrdtOp::ListInsert");
        };
        assert_eq!(collection, "docs");
        assert_eq!(got_document_id, document_id);
        assert_eq!(got_list_path, list_path);
        assert_eq!(got_index, index);
        assert_eq!(got_fields_json, fields_json);
        assert_eq!(got_surrogate, surrogate);
    }

    #[test]
    fn list_insert_missing_index_is_typed_error_not_zero_default() {
        let mut fields = list_insert_fields();
        fields.list_index = None;
        let err = require_list_index(fields.list_index, "list_index").unwrap_err();
        assert!(matches!(err, crate::Error::BadRequest { .. }));
    }

    #[test]
    fn list_insert_out_of_range_index_is_typed_error() {
        // usize on this platform is 64-bit, so exercise the narrowing
        // path with a value that legitimately fits u64 but is used to
        // prove the conversion is checked rather than assumed infallible.
        let index = require_list_index(Some(u64::MAX), "list_index");
        // On 64-bit platforms usize == u64 so u64::MAX narrows cleanly;
        // the important property is that the function is fallible and
        // returns a typed error rather than silently truncating when it
        // does not fit, not that MAX specifically fails on every arch.
        if usize::try_from(u64::MAX).is_err() {
            assert!(matches!(index, Err(crate::Error::BadRequest { .. })));
        } else {
            assert_eq!(index.unwrap(), u64::MAX as usize);
        }
    }

    #[test]
    fn list_insert_missing_list_path_is_typed_error() {
        let mut fields = list_insert_fields();
        fields.list_path = None;
        assert!(matches!(
            require_list_path(&fields),
            Err(crate::Error::BadRequest { .. })
        ));
    }

    #[test]
    fn list_delete_produces_exact_plan_without_fields_json() {
        let document_id = "doc-2".to_string();
        let list_path = "notes".to_string();
        let index = 7usize;
        let surrogate = Surrogate::new(9);

        let plan = PhysicalPlan::Crdt(CrdtOp::ListDelete {
            collection: "docs".to_string(),
            document_id: document_id.clone(),
            list_path: list_path.clone(),
            index,
            surrogate,
        });

        let PhysicalPlan::Crdt(CrdtOp::ListDelete {
            collection,
            document_id: got_document_id,
            list_path: got_list_path,
            index: got_index,
            surrogate: got_surrogate,
        }) = plan
        else {
            panic!("expected CrdtOp::ListDelete");
        };
        assert_eq!(collection, "docs");
        assert_eq!(got_document_id, document_id);
        assert_eq!(got_list_path, list_path);
        assert_eq!(got_index, index);
        assert_eq!(got_surrogate, surrogate);
    }

    #[test]
    fn list_move_keeps_from_and_to_index_distinct_and_unswapped() {
        let fields = list_move_fields();
        let document_id = require_doc_id(&fields).unwrap();
        let list_path = require_list_path(&fields).unwrap();
        let from_index = require_list_index(fields.list_from_index, "list_from_index").unwrap();
        let to_index = require_list_index(fields.list_to_index, "list_to_index").unwrap();
        assert_ne!(from_index, to_index);

        let surrogate = Surrogate::new(11);
        let plan = PhysicalPlan::Crdt(CrdtOp::ListMove {
            collection: "docs".to_string(),
            document_id,
            list_path,
            from_index,
            to_index,
            surrogate,
        });

        let PhysicalPlan::Crdt(CrdtOp::ListMove {
            from_index: got_from,
            to_index: got_to,
            ..
        }) = plan
        else {
            panic!("expected CrdtOp::ListMove");
        };
        assert_eq!(got_from, 1);
        assert_eq!(got_to, 5);
        assert_ne!(got_from, got_to);
    }

    #[test]
    fn list_move_missing_from_index_is_typed_error() {
        let mut fields = list_move_fields();
        fields.list_from_index = None;
        assert!(matches!(
            require_list_index(fields.list_from_index, "list_from_index"),
            Err(crate::Error::BadRequest { .. })
        ));
        // to_index is independently present and must still parse fine.
        assert!(require_list_index(fields.list_to_index, "list_to_index").is_ok());
    }

    #[test]
    fn list_move_missing_to_index_is_typed_error() {
        let mut fields = list_move_fields();
        fields.list_to_index = None;
        assert!(matches!(
            require_list_index(fields.list_to_index, "list_to_index"),
            Err(crate::Error::BadRequest { .. })
        ));
    }
}