nodedb-lite 0.1.0

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

use nodedb_physical::physical_plan::MetaOp;

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

use super::LitePhysicalFut;

pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    permit: Option<&'a crate::engine::fts::coordinator::TextMutationPermit>,
    op: &MetaOp,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    match op {
        MetaOp::CreateSnapshot => Ok(Box::pin(async move {
            meta_ops::handle_create_snapshot(engine).await
        })),
        MetaOp::Compact => Ok(Box::pin(
            async move { meta_ops::handle_compact(engine).await },
        )),
        MetaOp::Checkpoint => Ok(Box::pin(async move {
            meta_ops::handle_checkpoint(engine).await
        })),
        MetaOp::UnregisterCollection {
            tenant_id,
            name,
            purge_lsn,
            // Lite is single-process/single-core: it always owns and reclaims
            // its own on-disk L1 files, so the Control-Plane fan-out flag
            // (which selects exactly one core to unlink the shared files) does
            // not apply here — this core always fully reclaims.
            reclaim_l1_files: _,
        } => {
            let tid = *tenant_id;
            let n = name.clone();
            let lsn = *purge_lsn;
            Ok(Box::pin(async move {
                meta_ops::lifecycle::handle_unregister_collection_coordinated(
                    engine, permit, tid, &n, lsn,
                )
                .await
            }))
        }
        MetaOp::UnregisterMaterializedView { tenant_id, name } => {
            let tid = *tenant_id;
            let n = name.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_unregister_materialized_view(engine, tid, &n).await
            }))
        }
        // Lite's convert probes its own per-engine stores (CRDT, strict) and
        // decodes each with that store's schema, so the plan-level source
        // mode carries nothing Lite acts on.
        MetaOp::ConvertCollection {
            collection,
            target_type,
            schema_json,
            source_storage_mode: _source_storage_mode,
        } => {
            let col = collection.clone();
            let tt = target_type.clone();
            let sj = schema_json.clone();
            Ok(Box::pin(async move {
                meta_ops::lifecycle::handle_convert_collection_coordinated(
                    engine,
                    permit,
                    col.as_str(),
                    &tt,
                    &sj,
                )
                .await
            }))
        }
        MetaOp::RegisterContinuousAggregate { def } => {
            let d = def.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_register_continuous_aggregate(engine, d).await
            }))
        }
        MetaOp::UnregisterContinuousAggregate { name } => {
            let n = name.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_unregister_continuous_aggregate(engine, &n).await
            }))
        }
        MetaOp::ListContinuousAggregates => Ok(Box::pin(async move {
            meta_ops::handle_list_continuous_aggregates(engine).await
        })),
        MetaOp::ContinuousAggregatesRegistered => Ok(Box::pin(async move {
            meta_ops::handle_continuous_aggregates_registered().await
        })),
        // Lite folds each flush into its aggregates as the flush runs and
        // never applies a refresh cut, so no cut exists to wait for.
        MetaOp::AwaitContinuousAggregateRefresh {
            collection,
            name,
            refresh_id,
        } => Err(LiteError::Unsupported {
            detail: format!(
                "MetaOp::AwaitContinuousAggregateRefresh of '{name}' on '{collection}' \
                 (refresh {refresh_id}): refresh cuts are not supported in Lite"
            ),
        }),
        MetaOp::ApplyContinuousAggRetention => Ok(Box::pin(async move {
            meta_ops::handle_apply_continuous_agg_retention(engine).await
        })),
        MetaOp::QueryAggregateWatermark { aggregate_name } => {
            let n = aggregate_name.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_query_aggregate_watermark(engine, &n).await
            }))
        }
        MetaOp::QueryLastValues { collection } => {
            let col = collection.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_query_aggregate_last_values(engine, col.as_str()).await
            }))
        }
        MetaOp::QueryLastValue {
            collection,
            series_id,
        } => {
            let col = collection.clone();
            let sid = *series_id;
            Ok(Box::pin(async move {
                meta_ops::handle_query_aggregate_last_value(engine, col.as_str(), sid).await
            }))
        }
        MetaOp::TemporalPurgeEdgeStore {
            tenant_id,
            collection,
            cutoff_system_ms,
        } => {
            let tid = *tenant_id;
            let col = collection.clone();
            let cut = *cutoff_system_ms;
            Ok(Box::pin(async move {
                meta_ops::handle_temporal_purge_edge_store(engine, tid, col.as_str(), cut).await
            }))
        }
        MetaOp::TemporalPurgeDocumentStrict {
            tenant_id,
            collection,
            cutoff_system_ms,
        } => {
            let tid = *tenant_id;
            let col = collection.clone();
            let cut = *cutoff_system_ms;
            Ok(Box::pin(async move {
                meta_ops::handle_temporal_purge_document_strict(engine, tid, col.as_str(), cut)
                    .await
            }))
        }
        MetaOp::TemporalPurgeColumnar {
            tenant_id,
            collection,
            cutoff_system_ms,
            aggregates,
        } => {
            let tid = *tenant_id;
            let col = collection.clone();
            let cut = *cutoff_system_ms;
            let aggs = aggregates.clone();
            Ok(Box::pin(async move {
                meta_ops::require_registered_aggregates(engine, "TemporalPurgeColumnar", &aggs)?;
                meta_ops::handle_temporal_purge_columnar(engine, tid, col.as_str(), cut).await
            }))
        }
        MetaOp::TemporalPurgeCrdt {
            tenant_id,
            collection,
            cutoff_system_ms,
        } => {
            let tid = *tenant_id;
            let col = collection.clone();
            let cut = *cutoff_system_ms;
            Ok(Box::pin(async move {
                meta_ops::handle_temporal_purge_crdt(engine, tid, col.as_str(), cut).await
            }))
        }
        MetaOp::TemporalPurgeArray {
            tenant_id,
            array_id,
            cutoff_system_ms,
        } => {
            let tid = *tenant_id;
            let aid = array_id.clone();
            let cut = *cutoff_system_ms;
            Ok(Box::pin(async move {
                meta_ops::handle_temporal_purge_array(engine, tid, &aid, cut).await
            }))
        }
        MetaOp::EnforceTimeseriesRetention {
            collection,
            max_age_ms,
            aggregates,
        } => {
            let col = collection.clone();
            let age = *max_age_ms;
            let aggs = aggregates.clone();
            Ok(Box::pin(async move {
                meta_ops::require_registered_aggregates(
                    engine,
                    "EnforceTimeseriesRetention",
                    &aggs,
                )?;
                meta_ops::handle_enforce_timeseries_retention(engine, col.as_str(), age).await
            }))
        }
        MetaOp::AlterArray {
            array_id,
            audit_retain_ms,
            minimum_audit_retain_ms,
        } => {
            let aid = array_id.clone();
            let arm = *audit_retain_ms;
            let marm = *minimum_audit_retain_ms;
            Ok(Box::pin(async move {
                meta_ops::handle_alter_array(engine, &aid, arm, marm).await
            }))
        }
        MetaOp::PutSynonymGroup {
            tenant_id,
            record_json,
        } => {
            let tid = *tenant_id;
            let rj = record_json.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_put_synonym_group(engine, tid, &rj).await
            }))
        }
        MetaOp::DeleteSynonymGroup { tenant_id, name } => {
            let tid = *tenant_id;
            let n = name.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_delete_synonym_group(engine, tid, &n).await
            }))
        }
        MetaOp::RebuildIndex {
            collection,
            index_name,
            concurrent,
        } => {
            let col = collection.clone();
            let idx = index_name.clone();
            let conc = *concurrent;
            Ok(Box::pin(async move {
                meta_ops::handle_rebuild_index(engine, col.as_str(), idx.as_deref(), conc).await
            }))
        }
        MetaOp::QueryCollectionSize { tenant_id, name } => {
            let tid = *tenant_id;
            let n = name.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_query_collection_size(engine, tid, &n).await
            }))
        }
        // ── Distributed ops implemented on Lite ─────────────────────────────
        MetaOp::WalAppend { payload } => {
            let bytes = payload.clone();
            let storage = engine.storage.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_wal_append(&storage, &bytes).await
            }))
        }
        MetaOp::Cancel { target_request_id } => {
            let rid = *target_request_id;
            let registry = engine.cancellation.clone();
            Ok(Box::pin(
                async move { meta_ops::handle_cancel(&registry, rid) },
            ))
        }
        MetaOp::TransactionBatch { plans, txn_id: _ } => {
            let plans = plans.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_txn_batch(engine, &plans).await
            }))
        }
        MetaOp::CalvinExecuteStatic { plans, .. } => {
            let plans = plans.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_calvin_static(engine, &plans).await
            }))
        }
        MetaOp::CalvinExecutePassive { .. } => {
            Ok(Box::pin(
                async move { meta_ops::handle_calvin_passive().await },
            ))
        }
        MetaOp::CalvinExecuteActive { plans, .. } => {
            let plans = plans.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_calvin_active(engine, &plans).await
            }))
        }
        // ── Origin-only ops that Lite's plan converter never emits ───────────
        // A Lite write has its final outcome once it commits, so a
        // consistent cut needs no wait here.
        MetaOp::CreateTenantSnapshot {
            tenant_id,
            cut_watermark: _,
            cut_capture,
            arrays: _,
        } => {
            if cut_capture.is_some() {
                return Err(LiteError::Unsupported {
                    detail: "CreateTenantSnapshot with cut_capture is a cluster backup barrier \
                             capture; unsupported on single-node Lite"
                        .into(),
                });
            }
            let tid = *tenant_id;
            let storage = engine.storage.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_create_tenant_snapshot(&*storage, tid).await
            }))
        }
        MetaOp::RestoreTenantSnapshot {
            tenant_id,
            snapshot,
            ..
        } => {
            let tid = *tenant_id;
            let snap = snapshot.clone();
            let storage = engine.storage.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_restore_tenant_snapshot(&*storage, tid, &snap).await
            }))
        }
        MetaOp::PurgeTenant { tenant_id } => {
            let tid = *tenant_id;
            let storage = engine.storage.clone();
            Ok(Box::pin(async move {
                meta_ops::handle_purge_tenant(&*storage, tid).await
            }))
        }

        // Per-core transaction staging overlay. These address the Origin Data
        // Plane's `CoreLoop`-owned `TxnOverlay`, which buffers not-yet-durable
        // writes per shard core. Lite executes transactions directly against
        // its single local store (see `nodedb/collection/transaction.rs`) and
        // has no per-core overlay to stage into, mark, or roll back, so its
        // planner never emits these.
        MetaOp::StageWrite { .. } => Err(LiteError::Unsupported {
            detail: "StageWrite targets the Data Plane per-core transaction overlay; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::DropTxnOverlay { .. } => Err(LiteError::Unsupported {
            detail: "DropTxnOverlay releases a Data Plane per-core transaction overlay; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::MarkSavepoint { .. } => Err(LiteError::Unsupported {
            detail: "MarkSavepoint marks the Data Plane per-core overlay undo journals; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::RollbackToSavepoint { .. } => Err(LiteError::Unsupported {
            detail: "RollbackToSavepoint replays Data Plane per-core overlay undo journals; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::ResolveTxn { .. } => Err(LiteError::Unsupported {
            detail: "ResolveTxn folds staged Data Plane write plans into one redo record; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::ApplyTransactionRedo { .. } => Err(LiteError::Unsupported {
            detail: "ApplyTransactionRedo installs a committed redo record from a vShard's \
                     data-group log; unsupported on the single-node Lite engine"
                .into(),
        }),

        // Calvin deterministic-scheduling ops. Calvin sequences transactions
        // across a replicated log; Lite has no sequencer, Raft group, or
        // cross-core write-version registry, so these are unreachable here.
        MetaOp::CalvinResolve { .. } => Err(LiteError::Unsupported {
            detail: "CalvinResolve is a deterministic-scheduler op; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::CalvinDrop { .. } => Err(LiteError::Unsupported {
            detail: "CalvinDrop is a deterministic-scheduler op; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),

        MetaOp::VerifyHashChain { .. } => Err(LiteError::Unsupported {
            detail: "VerifyHashChain checks a Data Plane audit hash chain; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::HomeVersions { .. } => Err(LiteError::Unsupported {
            detail: "HomeVersions reads the cluster write-version registry; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
        MetaOp::RestoreRedo(_) => Err(LiteError::Unsupported {
            detail: "RestoreRedo reinstalls a cluster transaction redo record; \
                     unsupported on the single-node Lite engine"
                .into(),
        }),
    }
}