nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! CrdtOp dispatch for the Lite physical visitor.

use nodedb_physical::physical_plan::CrdtOp;
use nodedb_types::RlsWriteCheck;

use crate::error::LiteError;
use crate::query::crdt_ops;
use crate::query::engine::LiteQueryEngine;
use crate::storage::engine::StorageEngine;

use super::LitePhysicalFut;
use super::policy::deny_policy;

pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    permit: Option<&'a crate::engine::fts::coordinator::TextMutationPermit>,
    op: &CrdtOp,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    match op {
        CrdtOp::Read {
            collection,
            document_id,
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            Ok(Box::pin(async move {
                crdt_ops::read::handle_read(engine, col.as_str(), &doc_id).await
            }))
        }

        CrdtOp::Apply {
            collection,
            delta,
            mutation_id,
            ..
        } => {
            let col = collection.clone();
            let delta_bytes = delta.clone();
            let mid = *mutation_id;
            Ok(Box::pin(async move {
                crdt_ops::write::handle_apply_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &delta_bytes,
                    mid,
                )
                .await
            }))
        }

        CrdtOp::ImportSnapshot {
            collection, bytes, ..
        } => {
            let col = collection.clone();
            let bytes = bytes.clone();
            Ok(Box::pin(async move {
                crdt_ops::write::handle_import_snapshot_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &bytes,
                )
                .await
            }))
        }

        // Constraint install/read/drop require the `nodedb_crdt::validator`
        // Constraint-checking subsystem (UNIQUE/FK/NOT NULL validation
        // against a per-collection constraint set). Lite has no local
        // validator wiring today — it only uses `nodedb_crdt::validator`'s
        // `Violation`/`Constraint` *types* to interpret Origin-issued
        // rejections in `reject_delta_with_policy`, not to validate writes
        // itself. Building constraint enforcement locally is a real feature,
        // not a compile fix, so these are unsupported until that lands.
        // `ApplyAuthenticated` carries catalog-owned signing enforcement
        // (`delta_signature`, `signing_required`, and the authenticated
        // identity triple). Lite has no signature-verification wiring, and the
        // variant exists precisely so a transport cannot omit those fields —
        // so handling it as a plain `Apply` would silently skip the signing
        // admission it was created to enforce. Rejected until Lite verifies
        // signatures itself; that is a real feature, not a compile fix.
        CrdtOp::ApplyAuthenticated { .. } => Err(LiteError::Unsupported {
            detail: "ApplyAuthenticated carries signing enforcement that Lite cannot verify \
                     locally; applying it unverified would bypass the signing admission"
                .into(),
        }),

        CrdtOp::SetConstraints { .. } => Err(LiteError::Unsupported {
            detail: "SetConstraints requires the CRDT validator constraint subsystem, \
                     which Lite does not implement locally"
                .into(),
        }),

        CrdtOp::DropConstraints { .. } => Err(LiteError::Unsupported {
            detail: "DropConstraints requires the CRDT validator constraint subsystem, \
                     which Lite does not implement locally"
                .into(),
        }),

        CrdtOp::ReadConstraints { .. } => Err(LiteError::Unsupported {
            detail: "ReadConstraints requires the CRDT validator constraint subsystem, \
                     which Lite does not implement locally"
                .into(),
        }),

        CrdtOp::SetPolicy {
            collection,
            policy_json,
        } => {
            let col = collection.clone();
            let json = policy_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::write::handle_set_policy(engine, col.as_str(), &json).await
            }))
        }

        CrdtOp::GetPolicy { collection } => {
            let col = collection.clone();
            Ok(Box::pin(async move {
                crdt_ops::read::handle_get_policy(engine, col.as_str()).await
            }))
        }

        CrdtOp::ReadAtVersion {
            collection,
            document_id,
            version_vector_json,
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            let vv_json = version_vector_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::version::handle_read_at_version(engine, col.as_str(), &doc_id, &vv_json)
                    .await
            }))
        }

        CrdtOp::GetVersionVector { .. } => Ok(Box::pin(async move {
            crdt_ops::version::handle_get_version_vector(engine).await
        })),

        CrdtOp::ExportDelta {
            collection,
            from_version_json,
        } => {
            let col = collection.clone();
            let from_json = from_version_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::version::handle_export_delta(engine, col.as_str(), &from_json).await
            }))
        }

        CrdtOp::RestoreToVersion {
            collection,
            document_id,
            target_version_json,
            ..
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            let target_json = target_version_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::version::handle_restore_to_version_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &doc_id,
                    &target_json,
                )
                .await
            }))
        }

        CrdtOp::CompactAtVersion {
            collection,
            target_version_json,
        } => {
            let col = collection.clone();
            let target_json = target_version_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::version::handle_compact_at_version_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &target_json,
                )
                .await
            }))
        }

        CrdtOp::ListInsert {
            collection,
            document_id,
            list_path,
            index,
            fields_json,
            ..
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            let path = list_path.clone();
            let idx = *index;
            let fields = fields_json.clone();
            Ok(Box::pin(async move {
                crdt_ops::list::handle_list_insert_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &doc_id,
                    &path,
                    idx,
                    &fields,
                )
                .await
            }))
        }

        CrdtOp::ListDelete {
            collection,
            document_id,
            list_path,
            index,
            ..
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            let path = list_path.clone();
            let idx = *index;
            Ok(Box::pin(async move {
                crdt_ops::list::handle_list_delete_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &doc_id,
                    &path,
                    idx,
                )
                .await
            }))
        }

        CrdtOp::ListMove {
            collection,
            document_id,
            list_path,
            from_index,
            to_index,
            ..
        } => {
            let col = collection.clone();
            let doc_id = document_id.clone();
            let path = list_path.clone();
            let from = *from_index;
            let to = *to_index;
            Ok(Box::pin(async move {
                crdt_ops::list::handle_list_move_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &doc_id,
                    &path,
                    from,
                    to,
                )
                .await
            }))
        }

        CrdtOp::DocUpsert {
            collection,
            document_id,
            fields_json,
            partial,
            verb,
            returning,
            rls_filters,
            ..
        } => {
            // DocUpsert has no rls_write_check slot: only returning and
            // rls_filters carry a policy here.
            deny_policy(
                "CrdtOp::DocUpsert",
                returning.as_ref(),
                &[rls_filters.as_slice()],
                &RlsWriteCheck::NoPolicyApplies,
            )?;
            let col = collection.clone();
            let doc_id = document_id.clone();
            let fields = fields_json.clone();
            let partial = *partial;
            let verb = *verb;
            let returning = returning.clone();
            Ok(Box::pin(async move {
                crdt_ops::doc_row::handle_doc_upsert_coordinated(
                    engine,
                    permit,
                    crdt_ops::doc_row::DocUpsertRequest {
                        collection: col.as_str(),
                        document_id: &doc_id,
                        fields_json: &fields,
                        partial,
                        verb,
                        returning: returning.as_ref(),
                    },
                )
                .await
            }))
        }

        CrdtOp::DocDelete {
            collection,
            document_id,
            returning,
            rls_filters,
            ..
        } => {
            // DocDelete has no rls_write_check slot; see DocUpsert above.
            deny_policy(
                "CrdtOp::DocDelete",
                returning.as_ref(),
                &[rls_filters.as_slice()],
                &RlsWriteCheck::NoPolicyApplies,
            )?;
            let col = collection.clone();
            let doc_id = document_id.clone();
            let returning = returning.clone();
            Ok(Box::pin(async move {
                crdt_ops::doc_row::handle_doc_delete_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &doc_id,
                    returning.as_ref(),
                )
                .await
            }))
        }

        // Speculative preview-apply (apply a delta to a fork for preview,
        // committed or discarded later) is an Origin-plane concept: it needs
        // the fork/pending-check machinery Lite does not run locally. Applying
        // it as an ordinary delta would silently commit a write meant to stay
        // speculative, so reject it explicitly rather than misapply it.
        CrdtOp::PreviewApply { .. } => Err(LiteError::Unsupported {
            detail: "PreviewApply requires the CRDT preview/fork subsystem, \
                     which Lite does not implement locally"
                .into(),
        }),
    }
}