From cae22699320afe85195b9b519f2740155eef0eb1 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 19:33:06 +0200 Subject: [PATCH] feat(v1.1.9): queue:: Rhai SDK module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit executor-core/sdk/queue.rs registers the queue:: namespace into the per-call Rhai engine: queue::enqueue("name", message) -> () queue::enqueue("name", message, opts) -> () // opts: Map queue::depth("name") -> i64 queue::depth_pending("name") -> i64 No handle pattern (queues are the grouping unit, mirroring pubsub). Opts map accepts optional delay_ms (i64) and max_attempts (u32); unknown keys are silently ignored to match Rhai's Map semantics. Bridge details: - captures Arc + Arc per closure (cheap clone) - block_on via TokioHandle::try_current() — same pattern pubsub/kv use - message → JSON via a local message_to_json that base64-encodes Blobs at any depth (mirrors pubsub) and REJECTS FnPtr / closures with a clear error (queue messages must survive Postgres round-trip) - depth / depth_pending clamp u64 → i64::MAX (Rhai's INT is i64) register_all gains queue::register(...) positionally after pubsub. Unit tests (executor-core --lib): blob → base64, nested-map round-trip, FnPtr rejection, opts parsing (partial / max_attempts only / non-integer delay rejection). Integration tests (executor-core/tests/sdk_queue.rs, 6 binaries): enqueue map payload, enqueue with opts threads through, blob base64 encodes, depth returns service value, FnPtr in payload throws, empty queue_name throws. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/executor-core/src/sdk/mod.rs | 2 + crates/executor-core/src/sdk/queue.rs | 266 ++++++++++++++++++++++++ crates/executor-core/tests/sdk_queue.rs | 211 +++++++++++++++++++ 3 files changed, 479 insertions(+) create mode 100644 crates/executor-core/src/sdk/queue.rs create mode 100644 crates/executor-core/tests/sdk_queue.rs diff --git a/crates/executor-core/src/sdk/mod.rs b/crates/executor-core/src/sdk/mod.rs index 23dad46..a307435 100644 --- a/crates/executor-core/src/sdk/mod.rs +++ b/crates/executor-core/src/sdk/mod.rs @@ -20,6 +20,7 @@ pub mod files; pub mod http; pub mod kv; pub mod pubsub; +pub mod queue; pub mod secrets; pub mod stdlib; pub mod users; @@ -45,6 +46,7 @@ pub fn register_all(engine: &mut RhaiEngine, services: &Services, cx: Arc) { + let svc = services.queue.clone(); + let mut module = Module::new(); + + // `queue::enqueue(name, message)` — default opts. + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "enqueue", + move |name: &str, message: Dynamic| -> Result<(), Box> { + let json = message_to_json(&message)?; + enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()).map(|_| ()) + }, + ); + } + + // `queue::enqueue(name, message, opts)` — opts is a Rhai Map with + // optional `delay_ms` and `max_attempts` keys. + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "enqueue", + move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box> { + let json = message_to_json(&message)?; + let opts = parse_opts(&opts)?; + enqueue_blocking(&svc, &cx, name, json, opts).map(|_| ()) + }, + ); + } + + // `queue::depth(name)` — total rows in the queue. + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "depth", + move |name: &str| -> Result> { + let svc = svc.clone(); + let cx = cx.clone(); + let name = name.to_string(); + block_on_u64(async move { svc.depth(&cx, &name).await }) + }, + ); + } + + // `queue::depth_pending(name)` — only currently-claimable rows. + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "depth_pending", + move |name: &str| -> Result> { + let svc = svc.clone(); + let cx = cx.clone(); + let name = name.to_string(); + block_on_u64(async move { svc.depth_pending(&cx, &name).await }) + }, + ); + } + + engine.register_static_module("queue", module.into()); +} + +/// Run the async enqueue inside the synchronous Rhai context. Mirrors +/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`. +fn enqueue_blocking( + svc: &Arc, + cx: &Arc, + name: &str, + payload: Json, + opts: EnqueueOpts, +) -> Result<(), Box> { + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("queue: no tokio runtime available: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let svc = svc.clone(); + let cx = cx.clone(); + let name = name.to_string(); + handle + .block_on(async move { svc.enqueue(&cx, &name, payload, opts).await }) + .map(|_| ()) + .map_err(|err| -> Box { + EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() + }) +} + +fn block_on_u64(fut: F) -> Result> +where + F: std::future::Future> + Send + 'static, +{ + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("queue: no tokio runtime available: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let n = handle.block_on(fut).map_err(|err| -> Box { + EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() + })?; + Ok(i64::try_from(n).unwrap_or(i64::MAX)) +} + +fn parse_opts(opts: &Map) -> Result> { + let mut out = EnqueueOpts::default(); + if let Some(v) = opts.get("delay_ms") { + out.delay_ms = Some(v.as_int().map_err(|_| -> Box { + EvalAltResult::ErrorRuntime( + "queue::enqueue: opts.delay_ms must be an integer".into(), + rhai::Position::NONE, + ) + .into() + })?); + } + if let Some(v) = opts.get("max_attempts") { + let n = v.as_int().map_err(|_| -> Box { + EvalAltResult::ErrorRuntime( + "queue::enqueue: opts.max_attempts must be an integer".into(), + rhai::Position::NONE, + ) + .into() + })?; + out.max_attempts = Some(u32::try_from(n).unwrap_or(0)); + } + Ok(out) +} + +/// Convert a Rhai `Dynamic` into JSON. Mirrors `pubsub::message_to_json` +/// — blobs become base64, FnPtr / closures are rejected so a script +/// can't accidentally enqueue something that won't survive a trip +/// through Postgres + back through the bridge. +fn message_to_json(value: &Dynamic) -> Result> { + if value.is::() { + return Err(EvalAltResult::ErrorRuntime( + "queue::enqueue: messages must not contain FnPtr / closures".into(), + rhai::Position::NONE, + ) + .into()); + } + if value.is_blob() { + let blob = value.clone().into_blob().unwrap_or_default(); + return Ok(Json::String(STANDARD.encode(&blob))); + } + if value.is_unit() { + return Ok(Json::Null); + } + if let Ok(b) = value.as_bool() { + return Ok(Json::Bool(b)); + } + if let Ok(i) = value.as_int() { + return Ok(Json::Number(i.into())); + } + if let Ok(f) = value.as_float() { + return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); + } + if value.is_string() { + return Ok(Json::String(value.clone().into_string().unwrap_or_default())); + } + if let Some(arr) = value.clone().try_cast::() { + let mut out = Vec::with_capacity(arr.len()); + for v in &arr { + out.push(message_to_json(v)?); + } + return Ok(Json::Array(out)); + } + if let Some(map) = value.clone().try_cast::() { + let mut out = serde_json::Map::new(); + for (k, v) in map { + out.insert(k.to_string(), message_to_json(&v)?); + } + return Ok(Json::Object(out)); + } + Ok(Json::String(value.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + use rhai::Dynamic; + + #[test] + fn message_blob_encodes_to_base64_string() { + let blob: rhai::Blob = vec![0x01, 0x02, 0x03]; + let d = Dynamic::from(blob); + let json = message_to_json(&d).unwrap(); + // STANDARD base64 of [1,2,3] is "AQID" + assert_eq!(json, Json::String("AQID".into())); + } + + #[test] + fn message_nested_map_round_trips_through_json() { + let mut m = Map::new(); + m.insert("k".into(), Dynamic::from(42_i64)); + let d = Dynamic::from(m); + let json = message_to_json(&d).unwrap(); + let expected = serde_json::json!({ "k": 42 }); + assert_eq!(json, expected); + } + + #[test] + fn message_rejects_fnptr() { + // Build a Rhai FnPtr and confirm we reject it. + let f = rhai::FnPtr::new("anything").unwrap(); + let d = Dynamic::from(f); + let err = message_to_json(&d).unwrap_err(); + assert!(err.to_string().contains("FnPtr")); + } + + #[test] + fn parse_opts_accepts_partial_overrides() { + let mut m = Map::new(); + m.insert("delay_ms".into(), Dynamic::from(500_i64)); + let opts = parse_opts(&m).unwrap(); + assert_eq!(opts.delay_ms, Some(500)); + assert_eq!(opts.max_attempts, None); + } + + #[test] + fn parse_opts_max_attempts_only() { + let mut m = Map::new(); + m.insert("max_attempts".into(), Dynamic::from(5_i64)); + let opts = parse_opts(&m).unwrap(); + assert_eq!(opts.delay_ms, None); + assert_eq!(opts.max_attempts, Some(5)); + } + + #[test] + fn parse_opts_rejects_non_integer_delay() { + let mut m = Map::new(); + m.insert("delay_ms".into(), Dynamic::from("nope")); + assert!(parse_opts(&m).is_err()); + } +} diff --git a/crates/executor-core/tests/sdk_queue.rs b/crates/executor-core/tests/sdk_queue.rs new file mode 100644 index 0000000..b50193d --- /dev/null +++ b/crates/executor-core/tests/sdk_queue.rs @@ -0,0 +1,211 @@ +//! `queue::` SDK bridge integration tests — runs a real Rhai engine +//! against an in-memory `QueueService` that records the enqueued +//! `(queue_name, payload, opts)`. Verifies opts handling, blob +//! base64 encoding, depth/depth_pending pass-through. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits}; +use picloud_shared::{ + AppId, EnqueueOpts, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEventEmitter, + NoopFilesService, NoopHttpService, NoopKvService, NoopModuleSource, NoopPubsubService, + QueueError, QueueMessageId, QueueService, RequestId, ScriptId, ScriptSandbox, SdkCallCx, + Services, +}; +use serde_json::{json, Value}; + +#[derive(Default)] +struct RecordingQueue { + enqueues: Mutex>, + depth: Mutex, + depth_pending: Mutex, +} + +#[async_trait] +impl QueueService for RecordingQueue { + async fn enqueue( + &self, + _cx: &SdkCallCx, + queue_name: &str, + payload: Value, + opts: EnqueueOpts, + ) -> Result { + if queue_name.is_empty() { + return Err(QueueError::EmptyName); + } + self.enqueues + .lock() + .unwrap() + .push((queue_name.to_string(), payload, opts)); + Ok(QueueMessageId::new()) + } + + async fn depth(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result { + Ok(*self.depth.lock().unwrap()) + } + + async fn depth_pending( + &self, + _cx: &SdkCallCx, + _queue_name: &str, + ) -> Result { + Ok(*self.depth_pending.lock().unwrap()) + } +} + +fn make_engine(svc: Arc) -> Arc { + let services = Services::new( + Arc::new(NoopKvService), + Arc::new(NoopDocsService), + Arc::new(NoopDeadLetterService), + Arc::new(NoopEventEmitter), + Arc::new(NoopModuleSource), + Arc::new(NoopHttpService), + Arc::new(NoopFilesService), + Arc::new(NoopPubsubService), + Arc::new(picloud_shared::NoopSecretsService), + Arc::new(picloud_shared::NoopEmailService), + Arc::new(picloud_shared::NoopUsersService), + svc, + ); + Arc::new(Engine::new(Limits::default(), services)) +} + +fn baseline_request(app_id: AppId) -> ExecRequest { + let execution_id = ExecutionId::new(); + ExecRequest { + execution_id, + request_id: RequestId::new(), + script_id: ScriptId::new(), + script_name: "queue-test".into(), + invocation_type: InvocationType::Http, + path: "/queue-test".into(), + headers: BTreeMap::new(), + body: Value::Null, + params: BTreeMap::new(), + query: BTreeMap::new(), + rest: String::new(), + sandbox_overrides: ScriptSandbox::default(), + app_id, + principal: None, + trigger_depth: 0, + root_execution_id: execution_id, + is_dead_letter_handler: false, + event: None, + } +} + +async fn run(engine: Arc, src: &str, req: ExecRequest) { + let src = src.to_string(); + tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .expect("spawn_blocking should not panic") + .expect("script execution should succeed"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn enqueue_map_payload() { + let svc = Arc::new(RecordingQueue::default()); + let engine = make_engine(svc.clone()); + run( + engine, + r#"queue::enqueue("jobs", #{ id: 1, name: "x" });"#, + baseline_request(AppId::new()), + ) + .await; + let (qn, msg, opts) = svc.enqueues.lock().unwrap()[0].clone(); + assert_eq!(qn, "jobs"); + assert_eq!(msg, json!({ "id": 1, "name": "x" })); + assert!(opts.delay_ms.is_none()); + assert!(opts.max_attempts.is_none()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn enqueue_with_opts_threads_through() { + let svc = Arc::new(RecordingQueue::default()); + let engine = make_engine(svc.clone()); + run( + engine, + r#"queue::enqueue("jobs", #{ x: 1 }, #{ delay_ms: 60000, max_attempts: 5 });"#, + baseline_request(AppId::new()), + ) + .await; + let (qn, _msg, opts) = svc.enqueues.lock().unwrap()[0].clone(); + assert_eq!(qn, "jobs"); + assert_eq!(opts.delay_ms, Some(60_000)); + assert_eq!(opts.max_attempts, Some(5)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn enqueue_blob_base64_encodes() { + let svc = Arc::new(RecordingQueue::default()); + let engine = make_engine(svc.clone()); + run( + engine, + r#" + let data = base64::decode("aGVsbG8="); + queue::enqueue("blobs", #{ raw: data }); + "#, + baseline_request(AppId::new()), + ) + .await; + let (_qn, msg, _opts) = svc.enqueues.lock().unwrap()[0].clone(); + assert_eq!(msg, json!({ "raw": "aGVsbG8=" })); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn depth_returns_service_value() { + let svc = Arc::new(RecordingQueue::default()); + *svc.depth.lock().unwrap() = 99; + let engine = make_engine(svc.clone()); + // Stash result via a known-shape map; the engine returns the eval value. + let src = r#" + let n = queue::depth("any"); + #{ statusCode: 200, body: n } + "# + .to_string(); + let req = baseline_request(AppId::new()); + let resp = tokio::task::spawn_blocking({ + let engine = engine.clone(); + move || engine.execute(&src, req) + }) + .await + .expect("spawn_blocking should not panic") + .expect("script execution should succeed"); + assert_eq!(resp.body, json!(99)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn enqueue_rejects_fnptr_in_payload() { + let svc = Arc::new(RecordingQueue::default()); + let engine = make_engine(svc.clone()); + let src = r#" + let f = |x| x + 1; + queue::enqueue("jobs", #{ closure: f }); + "# + .to_string(); + let req = baseline_request(AppId::new()); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .expect("spawn_blocking should not panic"); + let err = res.unwrap_err(); + let msg = err.to_string(); + assert!( + msg.contains("FnPtr") || msg.contains("closure"), + "expected FnPtr rejection, got: {msg}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn enqueue_empty_name_throws() { + let svc = Arc::new(RecordingQueue::default()); + let engine = make_engine(svc.clone()); + let src = r#"queue::enqueue("", 1);"#.to_string(); + let req = baseline_request(AppId::new()); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .expect("spawn_blocking should not panic"); + assert!(res.is_err(), "empty queue_name should throw"); +}