PiCloud/crates/executor-core/src/sdk/kv.rs

//! `kv::` Rhai bridge — collection-scoped handle pattern.
//!
//! ```rhai
//! let widgets = kv::collection("widgets");
//! widgets.set("k", #{ n: 1 });
//! let v = widgets.get("k");      // value or () if absent
//! if widgets.has("k") { ... }
//! widgets.delete("k");            // bool (was-present)
//! let page = widgets.list();      // returns #{ keys: [...], next_cursor: () }
//! ```
//!
//! The `KvHandle` custom Rhai type captures the collection name once
//! and routes each call through the injected `Arc<dyn KvService>` with
//! the per-call `Arc<SdkCallCx>`. **The service derives `app_id` from
//! `cx.app_id` — `app_id` never appears in any function signature
//! script-side, preserving cross-app isolation.**
//!
//! Sync↔async bridge: Rhai is synchronous; the underlying service is
//! async. Closures wrap each call in `Handle::current().block_on(...)`
//! — safe because `LocalExecutorClient` runs the script under
//! `spawn_blocking`, so a runtime handle is reachable and blocking on
//! it doesn't park an async worker.
//!
//! Error convention (per `docs/sdk-shape.md`):
//! - throw on failure (Rhai runtime error string)
//! - `()` for absent values (`get` on a missing key)
//! - `bool` for predicates (`has`; also `delete` returns was-present)

use std::sync::Arc;

use picloud_shared::{KvError, KvService, SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;

use super::bridge::{dynamic_to_json, json_to_dynamic};

/// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs
/// plus an owned string).
#[derive(Clone)]
pub struct KvHandle {
    collection: String,
    service: Arc<dyn KvService>,
    cx: Arc<SdkCallCx>,
}

pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
    let kv_service = services.kv.clone();

    // `kv::collection(name)` — handle constructor lives in the `kv`
    // static module so the script-visible call is `kv::collection(...)`.
    let mut module = Module::new();
    {
        let kv_service = kv_service.clone();
        let cx = cx.clone();
        module.set_native_fn(
            "collection",
            move |name: &str| -> Result<KvHandle, Box<EvalAltResult>> {
                if name.is_empty() {
                    return Err("kv::collection name must not be empty".into());
                }
                Ok(KvHandle {
                    collection: name.to_string(),
                    service: kv_service.clone(),
                    cx: cx.clone(),
                })
            },
        );
    }
    engine.register_static_module("kv", module.into());

    // Methods on KvHandle — `register_fn` with `&mut KvHandle` first
    // argument lets Rhai dispatch them as `handle.get(k)` /
    // `handle.set(k, v)` / etc. through the dot-notation.
    engine.register_type_with_name::<KvHandle>("KvHandle");

    register_get(engine);
    register_set(engine);
    register_has(engine);
    register_delete(engine);
    register_list(engine);
}

fn register_get(engine: &mut RhaiEngine) {
    engine.register_fn(
        "get",
        |handle: &mut KvHandle, key: &str| -> Result<Dynamic, Box<EvalAltResult>> {
            let h = handle.clone();
            block_on(async move { h.service.get(&h.cx, &h.collection, key).await })
                .map(|opt| opt.map_or(Dynamic::UNIT, json_to_dynamic))
        },
    );
}

fn register_set(engine: &mut RhaiEngine) {
    engine.register_fn(
        "set",
        |handle: &mut KvHandle, key: &str, value: Dynamic| -> Result<(), Box<EvalAltResult>> {
            let h = handle.clone();
            let json = dynamic_to_json(&value);
            block_on(async move { h.service.set(&h.cx, &h.collection, key, json).await })
        },
    );
}

fn register_has(engine: &mut RhaiEngine) {
    engine.register_fn(
        "has",
        |handle: &mut KvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
            let h = handle.clone();
            block_on(async move { h.service.has(&h.cx, &h.collection, key).await })
        },
    );
}

fn register_delete(engine: &mut RhaiEngine) {
    engine.register_fn(
        "delete",
        |handle: &mut KvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
            let h = handle.clone();
            block_on(async move { h.service.delete(&h.cx, &h.collection, key).await })
        },
    );
}

fn register_list(engine: &mut RhaiEngine) {
    // Zero-arg form — full page, no cursor.
    engine.register_fn(
        "list",
        |handle: &mut KvHandle| -> Result<Map, Box<EvalAltResult>> { list_call(handle, None, 0) },
    );

    // One-arg form — cursor only.
    engine.register_fn(
        "list",
        |handle: &mut KvHandle, cursor: &str| -> Result<Map, Box<EvalAltResult>> {
            list_call(handle, Some(cursor.to_string()), 0)
        },
    );

    // Two-arg form — cursor + limit.
    engine.register_fn(
        "list",
        |handle: &mut KvHandle, cursor: &str, limit: i64| -> Result<Map, Box<EvalAltResult>> {
            let limit = u32::try_from(limit.max(0)).unwrap_or(0);
            list_call(handle, Some(cursor.to_string()), limit)
        },
    );
}

fn list_call(
    handle: &KvHandle,
    cursor: Option<String>,
    limit: u32,
) -> Result<Map, Box<EvalAltResult>> {
    let h = handle.clone();
    let page = block_on(async move {
        h.service
            .list(&h.cx, &h.collection, cursor.as_deref(), limit)
            .await
    })?;
    let mut m = Map::new();
    let keys: Array = page.keys.into_iter().map(Dynamic::from).collect();
    m.insert("keys".into(), keys.into());
    m.insert(
        "next_cursor".into(),
        page.next_cursor.map_or(Dynamic::UNIT, Dynamic::from),
    );
    Ok(m)
}

/// Run an async future inside the synchronous Rhai context.
///
/// `LocalExecutorClient` wraps script execution in `spawn_blocking`, so
/// the current Tokio runtime is reachable via `Handle::current()`. We
/// block on it directly; we are NOT calling this from an async task,
/// so blocking is the correct primitive (`block_in_place` would also
/// work, but we're already on a blocking worker).
fn block_on<F, T>(fut: F) -> Result<T, Box<EvalAltResult>>
where
    F: std::future::Future<Output = Result<T, KvError>> + Send,
    T: Send,
{
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(
            format!("kv: no tokio runtime available: {e}").into(),
            rhai::Position::NONE,
        )
        .into()
    })?;
    handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(format!("kv: {err}").into(), rhai::Position::NONE).into()
    })
}