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

//! `queue::` Rhai bridge — durable per-app queue producer + inspection
//! (v1.1.9).
//!
//! ```rhai
//! queue::enqueue("payments.process", #{ order_id: 123, amount: 4999 });
//! queue::enqueue("payments.process", msg, #{ delay_ms: 60000, max_attempts: 5 });
//! let total = queue::depth("payments.process");
//! let pending = queue::depth_pending("payments.process");
//! ```
//!
//! No handle pattern (queues are the grouping unit, mirroring pubsub).
//! No `peek` / `dequeue` / `purge` script-side surface — consumers are
//! `queue:receive` triggers; exposing manual dequeue would force the
//! platform to surface visibility-timeout semantics into script-land.
//!
//! `app_id` is derived from `cx.app_id` inside the service — it never
//! appears in the script-side signature.

use std::sync::Arc;

use base64::engine::general_purpose::STANDARD;
use base64::Engine as _;
use picloud_shared::{
    EnqueueOpts, GroupQueueService, QueueError, QueueService, SdkCallCx, Services,
};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use serde_json::Value as Json;
use tokio::runtime::Handle as TokioHandle;

use super::bridge::MAX_JSON_MATERIALIZE_BYTES;

pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
    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<EvalAltResult>> {
                crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
                let json = message_to_json(&message)?;
                enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default())
            },
        );
    }

    // `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<EvalAltResult>> {
                crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
                let json = message_to_json(&message)?;
                let opts = parse_opts(&opts)?;
                enqueue_blocking(&svc, &cx, name, json, opts)
            },
        );
    }

    // `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<i64, Box<EvalAltResult>> {
                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<i64, Box<EvalAltResult>> {
                let svc = svc.clone();
                let cx = cx.clone();
                let name = name.to_string();
                block_on_u64(async move { svc.depth_pending(&cx, &name).await })
            },
        );
    }

    // §11.6 D3: `queue::shared_collection("name")` → a handle over the group
    // shared queue (any subtree app enqueues into one group-owned store;
    // competing per-descendant consumers drain it). The owning group resolves
    // from `cx.app_id` inside the service — never a script arg.
    {
        let group_svc = services.group_queue.clone();
        let cx = cx.clone();
        module.set_native_fn(
            "shared_collection",
            move |name: &str| -> Result<GroupQueueHandle, Box<EvalAltResult>> {
                if name.is_empty() {
                    return Err("queue::shared_collection name must not be empty".into());
                }
                Ok(GroupQueueHandle {
                    collection: name.to_string(),
                    service: group_svc.clone(),
                    cx: cx.clone(),
                })
            },
        );
    }

    engine.register_static_module("queue", module.into());
    engine.register_type_with_name::<GroupQueueHandle>("GroupQueueHandle");
    register_shared_queue_methods(engine);
}

/// §11.6 D3 handle returned by `queue::shared_collection("name")`.
#[derive(Clone)]
pub struct GroupQueueHandle {
    collection: String,
    service: Arc<dyn GroupQueueService>,
    cx: Arc<SdkCallCx>,
}

fn group_enqueue(
    h: &mut GroupQueueHandle,
    message: Dynamic,
    opts: EnqueueOpts,
) -> Result<(), Box<EvalAltResult>> {
    crate::sdk::emit_budget::charge_emission("queue::shared_collection")?;
    let json = message_to_json(&message)?;
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(
            format!("queue: no tokio runtime available: {e}").into(),
            rhai::Position::NONE,
        )
        .into()
    })?;
    let service = h.service.clone();
    let cx = h.cx.clone();
    let collection = h.collection.clone();
    handle
        .block_on(async move { service.enqueue(&cx, &collection, json, opts).await })
        .map(|_id| ())
        .map_err(|err| -> Box<EvalAltResult> {
            EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
        })
}

fn group_depth<F, Fut>(h: &mut GroupQueueHandle, f: F) -> Result<i64, Box<EvalAltResult>>
where
    F: FnOnce(Arc<dyn GroupQueueService>, Arc<SdkCallCx>, String) -> Fut,
    Fut: std::future::Future<Output = Result<u64, picloud_shared::GroupQueueError>>,
{
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(
            format!("queue: no tokio runtime available: {e}").into(),
            rhai::Position::NONE,
        )
        .into()
    })?;
    let fut = f(h.service.clone(), h.cx.clone(), h.collection.clone());
    let n = handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
    })?;
    Ok(i64::try_from(n).unwrap_or(i64::MAX))
}

fn register_shared_queue_methods(engine: &mut RhaiEngine) {
    engine.register_fn("enqueue", |h: &mut GroupQueueHandle, message: Dynamic| {
        group_enqueue(h, message, EnqueueOpts::default())
    });
    engine.register_fn(
        "enqueue",
        |h: &mut GroupQueueHandle, message: Dynamic, opts: Map| {
            let opts = parse_opts(&opts)?;
            group_enqueue(h, message, opts)
        },
    );
    engine.register_fn("depth", |h: &mut GroupQueueHandle| {
        group_depth(
            h,
            |svc, cx, coll| async move { svc.depth(&cx, &coll).await },
        )
    });
    engine.register_fn("depth_pending", |h: &mut GroupQueueHandle| {
        group_depth(h, |svc, cx, coll| async move {
            svc.depth_pending(&cx, &coll).await
        })
    });
}

/// 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<dyn QueueService>,
    cx: &Arc<SdkCallCx>,
    name: &str,
    payload: Json,
    opts: EnqueueOpts,
) -> Result<(), Box<EvalAltResult>> {
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        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(|_id| ())
        .map_err(|err| -> Box<EvalAltResult> {
            EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
        })
}

fn block_on_u64<F>(fut: F) -> Result<i64, Box<EvalAltResult>>
where
    F: std::future::Future<Output = Result<u64, QueueError>> + Send + 'static,
{
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        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> {
        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<EnqueueOpts, Box<EvalAltResult>> {
    let mut out = EnqueueOpts::default();
    if let Some(v) = opts.get("delay_ms") {
        out.delay_ms = Some(v.as_int().map_err(|_| -> Box<EvalAltResult> {
            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> {
            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. Bounded by
/// [`MAX_JSON_MATERIALIZE_BYTES`] so a cheaply-aliased huge message can't OOM
/// the node on materialization (the per-message payload cap is enforced later,
/// on the already-materialized value).
fn message_to_json(value: &Dynamic) -> Result<Json, Box<EvalAltResult>> {
    let mut remaining = MAX_JSON_MATERIALIZE_BYTES;
    message_to_json_bounded(value, &mut remaining)
}

fn msg_charge(remaining: &mut usize, n: usize) -> Result<(), Box<EvalAltResult>> {
    match remaining.checked_sub(n) {
        Some(left) => {
            *remaining = left;
            Ok(())
        }
        None => Err(EvalAltResult::ErrorRuntime(
            format!("queue::enqueue: message too large to serialize (exceeds the {MAX_JSON_MATERIALIZE_BYTES}-byte limit)").into(),
            rhai::Position::NONE,
        )
        .into()),
    }
}

fn message_to_json_bounded(
    value: &Dynamic,
    remaining: &mut usize,
) -> Result<Json, Box<EvalAltResult>> {
    if value.is::<rhai::FnPtr>() {
        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();
        let encoded = STANDARD.encode(&blob);
        msg_charge(remaining, encoded.len() + 2)?;
        return Ok(Json::String(encoded));
    }
    if value.is_unit() {
        msg_charge(remaining, 4)?;
        return Ok(Json::Null);
    }
    if let Ok(b) = value.as_bool() {
        msg_charge(remaining, 5)?;
        return Ok(Json::Bool(b));
    }
    if let Ok(i) = value.as_int() {
        msg_charge(remaining, 8)?;
        return Ok(Json::Number(i.into()));
    }
    if let Ok(f) = value.as_float() {
        msg_charge(remaining, 8)?;
        return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number));
    }
    if value.is_string() {
        let s = value.clone().into_string().unwrap_or_default();
        msg_charge(remaining, s.len() + 2)?;
        return Ok(Json::String(s));
    }
    if let Some(arr) = value.clone().try_cast::<Array>() {
        msg_charge(remaining, 2)?;
        let mut out = Vec::with_capacity(arr.len().min(1024));
        for v in &arr {
            msg_charge(remaining, 1)?;
            out.push(message_to_json_bounded(v, remaining)?);
        }
        return Ok(Json::Array(out));
    }
    if let Some(map) = value.clone().try_cast::<Map>() {
        msg_charge(remaining, 2)?;
        let mut out = serde_json::Map::new();
        for (k, v) in map {
            msg_charge(remaining, k.len() + 4)?;
            out.insert(k.to_string(), message_to_json_bounded(&v, remaining)?);
        }
        return Ok(Json::Object(out));
    }
    let s = value.to_string();
    msg_charge(remaining, s.len() + 2)?;
    Ok(Json::String(s))
}

#[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());
    }
}