nodedb-lite 0.1.0

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

use nodedb_physical::physical_plan::SpatialOp;
use nodedb_types::RlsWriteCheck;

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::query::spatial_ops;
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>,
    op: &SpatialOp,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    match op {
        SpatialOp::Insert {
            collection,
            field,
            surrogate,
            geometry,
            provenance: _,
        } => {
            let col = collection.clone();
            let fld = field.clone();
            let sur = *surrogate;
            let geom = geometry.clone();
            Ok(Box::pin(async move {
                spatial_ops::writes::spatial_insert(engine, col.as_str(), &fld, sur, &geom)
            }))
        }

        SpatialOp::Delete {
            collection,
            field,
            surrogate,
            provenance: _,
        } => {
            let col = collection.clone();
            let fld = field.clone();
            let sur = *surrogate;
            Ok(Box::pin(async move {
                spatial_ops::writes::spatial_delete(engine, col.as_str(), &fld, sur)
            }))
        }

        SpatialOp::Scan {
            collection,
            field,
            predicate,
            query_geometry,
            distance_meters,
            attribute_filters,
            limit,
            projection,
            rls_filters,
            prefilter,
        } => {
            // Scan carries no rls_write_check slot: it never writes.
            deny_policy(
                "SpatialOp::Scan",
                None,
                &[rls_filters.as_slice()],
                &RlsWriteCheck::NoPolicyApplies,
            )?;
            let params = spatial_ops::reads::ScanParams {
                collection: collection.as_str().to_string(),
                field: field.clone(),
                predicate: *predicate,
                query_geometry: query_geometry.clone(),
                distance_meters: *distance_meters,
                attribute_filters: attribute_filters.clone(),
                limit: *limit,
                projection: projection.clone(),
                rls_filters: rls_filters.clone(),
                prefilter: prefilter.clone(),
            };
            Ok(Box::pin(async move {
                spatial_ops::reads::spatial_scan(engine, params)
            }))
        }
    }
}