PiCloud/crates/executor-core/tests/sdk_queue.rs

//! `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<Vec<(String, Value, EnqueueOpts)>>,
    depth: Mutex<u64>,
    depth_pending: Mutex<u64>,
}

#[async_trait]
impl QueueService for RecordingQueue {
    async fn enqueue(
        &self,
        _cx: &SdkCallCx,
        queue_name: &str,
        payload: Value,
        opts: EnqueueOpts,
    ) -> Result<QueueMessageId, QueueError> {
        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<u64, QueueError> {
        Ok(*self.depth.lock().unwrap())
    }

    async fn depth_pending(
        &self,
        _cx: &SdkCallCx,
        _queue_name: &str,
    ) -> Result<u64, QueueError> {
        Ok(*self.depth_pending.lock().unwrap())
    }
}

fn make_engine(svc: Arc<RecordingQueue>) -> Arc<Engine> {
    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(picloud_shared::NoopInvokeService),
    );
    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<Engine>, 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");
}