nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! Mutating `KvOp` arms: put/insert/delete/increment/CAS/transfer.
//!
//! Every arm runs its write through `kv_ops::sync_capture`, which records
//! the rows it leaves for the sync push to Origin.

use nodedb_physical::physical_plan::{KvBatchIntent, KvCounterShape, ReturningSpec, UpdateValue};
use nodedb_types::{QualifiedCollection, RlsWriteCheck};

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::query::kv_ops;
use crate::query::kv_ops::sync_capture::{KvTouched, record_kv_write};
use crate::storage::engine::StorageEngine;

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

pub(super) fn put<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    value: &[u8],
    ttl_ms: u64,
    returning: &Option<ReturningSpec>,
    rls_filters: &[u8],
) -> Result<LitePhysicalFut<'a>, LiteError> {
    // Put has no rls_write_check slot: SET semantics carry no write gate.
    deny_policy(
        "KvOp::Put",
        returning.as_ref(),
        &[rls_filters],
        &RlsWriteCheck::NoPolicyApplies,
    )?;
    let col = collection.clone();
    let k = key.to_vec();
    let v = value.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_put(engine, col.as_str(), &k, &v, ttl_ms);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn insert<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    value: &[u8],
    ttl_ms: u64,
    returning: &Option<ReturningSpec>,
    rls_filters: &[u8],
) -> Result<LitePhysicalFut<'a>, LiteError> {
    // Insert has no rls_write_check slot: an unconditional first write
    // carries no write gate.
    deny_policy(
        "KvOp::Insert",
        returning.as_ref(),
        &[rls_filters],
        &RlsWriteCheck::NoPolicyApplies,
    )?;
    let col = collection.clone();
    let k = key.to_vec();
    let v = value.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_insert(engine, col.as_str(), &k, &v, ttl_ms);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn insert_if_absent<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    value: &[u8],
    ttl_ms: u64,
    returning: &Option<ReturningSpec>,
    rls_filters: &[u8],
) -> Result<LitePhysicalFut<'a>, LiteError> {
    // InsertIfAbsent has no rls_write_check slot: an unconditional
    // first write carries no write gate.
    deny_policy(
        "KvOp::InsertIfAbsent",
        returning.as_ref(),
        &[rls_filters],
        &RlsWriteCheck::NoPolicyApplies,
    )?;
    let col = collection.clone();
    let k = key.to_vec();
    let v = value.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_insert_if_absent(engine, col.as_str(), &k, &v, ttl_ms);
        record_kv_write(engine, touched, write).await
    }))
}

/// Grouped tail fields of `KvOp::InsertOnConflictUpdate`, kept out of the
/// function signature so it stays under clippy's argument-count lint.
pub(super) struct ConflictUpdateArgs<'a> {
    pub value: &'a [u8],
    pub ttl_ms: u64,
    pub updates: &'a [(String, UpdateValue)],
    pub rls_write_check: &'a RlsWriteCheck,
    pub returning: &'a Option<ReturningSpec>,
    pub rls_filters: &'a [u8],
}

pub(super) fn insert_on_conflict_update<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    args: ConflictUpdateArgs<'_>,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy(
        "KvOp::InsertOnConflictUpdate",
        args.returning.as_ref(),
        &[args.rls_filters],
        args.rls_write_check,
    )?;
    let col = collection.clone();
    let k = key.to_vec();
    let v = args.value.to_vec();
    let ttl_ms = args.ttl_ms;
    let upd = args.updates.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_insert_on_conflict_update(
            engine,
            col.as_str(),
            &k,
            &v,
            ttl_ms,
            &upd,
        );
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn delete<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    keys: &[Vec<u8>],
    rls_write_check: &RlsWriteCheck,
    returning: &Option<ReturningSpec>,
    rls_filters: &[u8],
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy(
        "KvOp::Delete",
        returning.as_ref(),
        &[rls_filters],
        rls_write_check,
    )?;
    let col = collection.clone();
    let ks = keys.to_vec();
    Ok(Box::pin(async move {
        let touched = ks
            .iter()
            .map(|k| (col.as_str().to_owned(), k.clone()))
            .collect();
        let write = kv_ops::writes::kv_delete(engine, col.as_str(), &ks);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn batch_put<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    entries: &[(Vec<u8>, Vec<u8>)],
    ttl_ms: u64,
    intent: &KvBatchIntent,
    returning: &Option<ReturningSpec>,
    rls_filters: &[u8],
) -> Result<LitePhysicalFut<'a>, LiteError> {
    // Only a merge carries a write gate. Every other intent writes the
    // entries as given.
    let no_policy = RlsWriteCheck::NoPolicyApplies;
    let write_check = match intent {
        KvBatchIntent::Merge {
            rls_write_check, ..
        } => rls_write_check,
        KvBatchIntent::Put
        | KvBatchIntent::Upsert
        | KvBatchIntent::Insert
        | KvBatchIntent::InsertIfAbsent => &no_policy,
    };
    deny_policy(
        "KvOp::BatchPut",
        returning.as_ref(),
        &[rls_filters],
        write_check,
    )?;
    let col = collection.clone();
    let ents = entries.to_vec();
    let intent = intent.clone();
    Ok(Box::pin(async move {
        let touched = ents
            .iter()
            .map(|(k, _)| (col.as_str().to_owned(), k.clone()))
            .collect();
        let write = kv_ops::writes::kv_batch_write(engine, col.as_str(), &ents, ttl_ms, &intent);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn expire<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    ttl_ms: u64,
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::Expire", None, &[], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_expire(engine, col.as_str(), &k, ttl_ms);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn persist<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::Persist", None, &[], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_persist(engine, col.as_str(), &k);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn truncate<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    restart_identity: bool,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    let col = collection.clone();
    Ok(Box::pin(async move {
        let result = kv_ops::sync_capture::truncate_recorded(engine, col.as_str()).await?;
        crate::query::truncate::restart_identity(engine, col.as_str(), restart_identity);
        Ok(result)
    }))
}

/// `KvOp::Incr`. `shape` names the row an absent key becomes, as on Origin.
pub(super) fn incr<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    delta: i64,
    ttl_ms: u64,
    shape: &KvCounterShape,
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::Incr", None, &[], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    let shape = shape.clone();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_incr(engine, col.as_str(), &k, delta, ttl_ms, &shape);
        record_kv_write(engine, touched, write).await
    }))
}

/// `KvOp::IncrFloat`. `delta` is the client's decimal text, added exactly to
/// a raw body, as on Origin.
pub(super) fn incr_float<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    delta: &str,
    shape: &KvCounterShape,
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::IncrFloat", None, &[], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    let delta = delta.to_owned();
    let shape = shape.clone();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_incr_float(engine, col.as_str(), &k, &delta, &shape);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn cas<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    expected: &[u8],
    new_value: &[u8],
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::Cas", None, &[], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    let exp = expected.to_vec();
    let nv = new_value.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_cas(engine, col.as_str(), &k, &exp, &nv);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn get_set<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    key: &[u8],
    new_value: &[u8],
    rls_filters: &[u8],
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::GetSet", None, &[rls_filters], rls_write_check)?;
    let col = collection.clone();
    let k = key.to_vec();
    let nv = new_value.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_get_set(engine, col.as_str(), &k, &nv);
        record_kv_write(engine, touched, write).await
    }))
}

/// One `KvOp::FieldSet`: the fields to write and the policy slots the plan
/// carries for them.
pub(super) struct FieldSetArgs<'a> {
    pub collection: &'a QualifiedCollection,
    pub key: &'a [u8],
    pub updates: &'a [(String, Vec<u8>)],
    /// SQL `UPDATE` semantics: an absent key writes nothing and counts 0.
    pub if_present: bool,
    pub rls_write_check: &'a RlsWriteCheck,
    pub returning: &'a Option<ReturningSpec>,
    pub rls_filters: &'a [u8],
}

pub(super) fn field_set<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    args: FieldSetArgs<'_>,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    let FieldSetArgs {
        collection,
        key,
        updates,
        if_present,
        rls_write_check,
        returning,
        rls_filters,
    } = args;
    deny_policy(
        "KvOp::FieldSet",
        returning.as_ref(),
        &[rls_filters],
        rls_write_check,
    )?;
    let col = collection.clone();
    let k = key.to_vec();
    let upd = updates.to_vec();
    Ok(Box::pin(async move {
        let touched = one(&col, &k);
        let write = kv_ops::writes::kv_field_set(engine, col.as_str(), &k, &upd, if_present);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn transfer<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    collection: &QualifiedCollection,
    source_key: &[u8],
    dest_key: &[u8],
    field: &str,
    amount: f64,
    rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    deny_policy("KvOp::Transfer", None, &[], rls_write_check)?;
    let col = collection.clone();
    let src = source_key.to_vec();
    let dst = dest_key.to_vec();
    let fld = field.to_owned();
    Ok(Box::pin(async move {
        let touched = vec![
            (col.as_str().to_owned(), src.clone()),
            (col.as_str().to_owned(), dst.clone()),
        ];
        let write = kv_ops::writes::kv_transfer(engine, col.as_str(), &src, &dst, &fld, amount);
        record_kv_write(engine, touched, write).await
    }))
}

pub(super) fn transfer_item<'a, S: StorageEngine + 'a>(
    engine: &'a LiteQueryEngine<S>,
    source_collection: &QualifiedCollection,
    dest_collection: &QualifiedCollection,
    item_key: &[u8],
    dest_key: &[u8],
    source_rls_write_check: &RlsWriteCheck,
    dest_rls_write_check: &RlsWriteCheck,
) -> Result<LitePhysicalFut<'a>, LiteError> {
    // Source and destination carry independent write gates: an
    // identity may give a row up but not receive it, so both check.
    deny_policy("KvOp::TransferItem", None, &[], source_rls_write_check)?;
    deny_policy("KvOp::TransferItem", None, &[], dest_rls_write_check)?;
    let src_col = source_collection.clone();
    let dst_col = dest_collection.clone();
    let ik = item_key.to_vec();
    let dk = dest_key.to_vec();
    Ok(Box::pin(async move {
        let touched = vec![
            (src_col.as_str().to_owned(), ik.clone()),
            (dst_col.as_str().to_owned(), dk.clone()),
        ];
        let write =
            kv_ops::writes::kv_transfer_item(engine, src_col.as_str(), dst_col.as_str(), &ik, &dk);
        record_kv_write(engine, touched, write).await
    }))
}

/// The single key a point write touches.
fn one(collection: &QualifiedCollection, key: &[u8]) -> Vec<KvTouched> {
    vec![(collection.as_str().to_owned(), key.to_vec())]
}