//! `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.set_if("k", (), #{ n: 0 }); // insert only if ABSENT -> bool //! widgets.set_if("k", old, new); // swap only if current == old -> bool //! 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` with //! the per-call `Arc`. **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::{GroupKvService, InterceptorService, KvService, SdkCallCx, Services}; use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; use super::bridge::{ block_on, dynamic_to_json_capped, json_to_dynamic, JsonSizeError, MAX_JSON_MATERIALIZE_BYTES, }; use crate::engine::Engine; use crate::sandbox::Limits; /// Per-call handle captured by the Rhai SDK. Cheap to clone (a few Arcs plus an /// owned string). Carries the §9.4 interceptor deps so `set`/`delete` can run a /// before-op allow/deny hook (the resolver + the `invoke()` re-entry engine). #[derive(Clone)] pub struct KvHandle { collection: String, service: Arc, cx: Arc, interceptors: Arc, self_engine: Option>, limits: Limits, } /// §11.6 shared-collection handle, returned by `kv::shared_collection(name)`. A distinct /// Rhai type from `KvHandle` so the method set can diverge (e.g. shared /// collections never grow triggers) and a script's choice of private-vs-shared /// scope is explicit. Routes through the `GroupKvService`, which resolves the /// owning group from `cx.app_id` (the script never names a group). Carries the /// §9.4 interceptor deps too so a `(kv, set/delete)` guard covers shared- /// collection writes — otherwise the shared handle would be a silent bypass. #[derive(Clone)] pub struct GroupKvHandle { collection: String, service: Arc, cx: Arc, interceptors: Arc, self_engine: Option>, limits: Limits, } pub(super) fn register( engine: &mut RhaiEngine, services: &Services, cx: Arc, limits: Limits, self_engine: Option>, ) { let kv_service = services.kv.clone(); let group_kv_service = services.group_kv.clone(); let interceptors = services.interceptors.clone(); // `kv::collection(name)` / `kv::shared_collection(name)` — both constructors live in // the `kv` static module so the script-visible calls are `kv::collection` // and `kv::shared_collection` (`shared` alone is a Rhai reserved word). let mut module = Module::new(); { let kv_service = kv_service.clone(); let cx = cx.clone(); let interceptors = interceptors.clone(); let self_engine = self_engine.clone(); module.set_native_fn( "collection", move |name: &str| -> Result> { 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(), interceptors: interceptors.clone(), self_engine: self_engine.clone(), limits, }) }, ); } { let group_kv_service = group_kv_service.clone(); let cx = cx.clone(); let interceptors = interceptors.clone(); let self_engine = self_engine.clone(); module.set_native_fn( "shared_collection", move |name: &str| -> Result> { if name.is_empty() { return Err("kv::shared_collection name must not be empty".into()); } Ok(GroupKvHandle { collection: name.to_string(), service: group_kv_service.clone(), cx: cx.clone(), interceptors: interceptors.clone(), self_engine: self_engine.clone(), limits, }) }, ); } 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"); register_get(engine); register_set(engine); register_set_if(engine); register_has(engine); register_delete(engine); register_list(engine); // Same method names on GroupKvHandle — Rhai dispatches by receiver type. engine.register_type_with_name::("GroupKvHandle"); register_group_get(engine); register_group_set(engine); register_group_set_if(engine); register_group_has(engine); register_group_delete(engine); register_group_list(engine); } /// Map a Rhai `expected` argument to the CAS precondition: Rhai unit `()` means /// "expected ABSENT" (insert-if-absent); any other value is the expected current /// value. fn expected_from_dynamic(expected: &Dynamic) -> Result, JsonSizeError> { if expected.is_unit() { Ok(None) } else { Ok(Some(dynamic_to_json_capped( expected, MAX_JSON_MATERIALIZE_BYTES, )?)) } } fn register_get(engine: &mut RhaiEngine) { engine.register_fn( "get", |handle: &mut KvHandle, key: &str| -> Result> { let h = handle.clone(); block_on("kv", 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> { let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; // §9.4 before-op interceptor (allow/deny). A denial errors here and // the write below never runs. super::interceptor::run_before( &handle.interceptors, handle.self_engine.as_ref(), &handle.cx, handle.limits, "kv", "set", &handle.collection, key, Some(&json), )?; let h = handle.clone(); block_on("kv", async move { h.service.set(&h.cx, &h.collection, key, json).await }) }, ); } fn register_set_if(engine: &mut RhaiEngine) { // `handle.set_if(key, expected, new)` — CAS. `expected = ()` means "only if // absent". Returns `true` if the swap happened, `false` if the precondition // failed. engine.register_fn( "set_if", |handle: &mut KvHandle, key: &str, expected: Dynamic, new: Dynamic| -> Result> { let h = handle.clone(); let exp = expected_from_dynamic(&expected)?; let new = dynamic_to_json_capped(&new, MAX_JSON_MATERIALIZE_BYTES)?; block_on("kv", async move { h.service.set_if(&h.cx, &h.collection, key, exp, new).await }) }, ); } fn register_has(engine: &mut RhaiEngine) { engine.register_fn( "has", |handle: &mut KvHandle, key: &str| -> Result> { let h = handle.clone(); block_on("kv", 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> { super::interceptor::run_before( &handle.interceptors, handle.self_engine.as_ref(), &handle.cx, handle.limits, "kv", "delete", &handle.collection, key, None, )?; let h = handle.clone(); block_on("kv", async move { h.service.delete(&h.cx, &h.collection, key).await }) }, ); } /// §9.4: run the before-op interceptor for a shared-collection write, so a /// `(kv, set/delete)` guard covers the shared handle too (not just private KV). fn group_run_before( handle: &GroupKvHandle, op: &'static str, key: &str, value: Option<&serde_json::Value>, ) -> Result<(), Box> { super::interceptor::run_before( &handle.interceptors, handle.self_engine.as_ref(), &handle.cx, handle.limits, "kv", op, &handle.collection, key, value, ) } fn register_list(engine: &mut RhaiEngine) { // Zero-arg form — full page, no cursor. engine.register_fn( "list", |handle: &mut KvHandle| -> Result> { list_call(handle, None, 0) }, ); // One-arg form — cursor only. engine.register_fn( "list", |handle: &mut KvHandle, cursor: &str| -> Result> { 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> { 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, limit: u32, ) -> Result> { let h = handle.clone(); let page = block_on("kv", 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) } // --- GroupKvHandle methods (§11.6 shared collections) ---------------------- fn register_group_get(engine: &mut RhaiEngine) { engine.register_fn( "get", |handle: &mut GroupKvHandle, key: &str| -> Result> { let h = handle.clone(); block_on("kv", async move { h.service.get(&h.cx, &h.collection, key).await }) .map(|opt| opt.map_or(Dynamic::UNIT, json_to_dynamic)) }, ); } fn register_group_set(engine: &mut RhaiEngine) { engine.register_fn( "set", |handle: &mut GroupKvHandle, key: &str, value: Dynamic| -> Result<(), Box> { let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; // §9.4 before-op interceptor also guards shared-collection writes. group_run_before(handle, "set", key, Some(&json))?; let h = handle.clone(); block_on("kv", async move { h.service.set(&h.cx, &h.collection, key, json).await }) }, ); } fn register_group_set_if(engine: &mut RhaiEngine) { engine.register_fn( "set_if", |handle: &mut GroupKvHandle, key: &str, expected: Dynamic, new: Dynamic| -> Result> { let h = handle.clone(); let exp = expected_from_dynamic(&expected)?; let new = dynamic_to_json_capped(&new, MAX_JSON_MATERIALIZE_BYTES)?; block_on("kv", async move { h.service.set_if(&h.cx, &h.collection, key, exp, new).await }) }, ); } fn register_group_has(engine: &mut RhaiEngine) { engine.register_fn( "has", |handle: &mut GroupKvHandle, key: &str| -> Result> { let h = handle.clone(); block_on("kv", async move { h.service.has(&h.cx, &h.collection, key).await }) }, ); } fn register_group_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut GroupKvHandle, key: &str| -> Result> { group_run_before(handle, "delete", key, None)?; let h = handle.clone(); block_on("kv", async move { h.service.delete(&h.cx, &h.collection, key).await }) }, ); } fn register_group_list(engine: &mut RhaiEngine) { engine.register_fn( "list", |handle: &mut GroupKvHandle| -> Result> { group_list_call(handle, None, 0) }, ); engine.register_fn( "list", |handle: &mut GroupKvHandle, cursor: &str| -> Result> { group_list_call(handle, Some(cursor.to_string()), 0) }, ); engine.register_fn( "list", |handle: &mut GroupKvHandle, cursor: &str, limit: i64| -> Result> { let limit = u32::try_from(limit.max(0)).unwrap_or(0); group_list_call(handle, Some(cursor.to_string()), limit) }, ); } fn group_list_call( handle: &GroupKvHandle, cursor: Option, limit: u32, ) -> Result> { let h = handle.clone(); let page = block_on("kv", 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) }