yeti_types/

queue.rs

//! Durable work queue — types shared across host components and the
//! SDK.
//!
//! **Layer:** L0 — these types live in yeti-types because the unified
//! `Context` references `QueuePayload` via the `queue_enqueue`
//! signature.

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

// ============================================================================
// QueuePayload — what lives in the yeti-queue `payload` field
// ============================================================================

/// The work a queue item represents. Persists in `yeti-queue` as msgpack
/// inside the `payload` column.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum QueuePayload {
    /// Dispatch a registered `#[queue_fn]` in the owning wasm app.
    Function {
        /// Serialized (msgpack) tuple of the user function's arguments after
        /// the leading `&Context`.
        args: Vec<u8>,
        /// `xxh3_64` of the canonicalized function signature. The worker rejects
        /// the item if the currently-registered function's `sig_hash` differs
        /// (signature changed since enqueue — fail fast, don't silently drop
        /// or mis-interpret args).
        fn_sig_hash: u64,
    },
    /// Dispatch a cross-node `fetch!()` call — routes through the existing
    /// host reqwest bridge with the same `HttpStats` + inflight cap the
    /// synchronous `fetch!` macro uses.
    Fetch {
        /// HTTP method (uppercase).
        method: String,
        /// Fully-qualified URL.
        url: String,
        /// HTTP headers, verbatim.
        headers: HashMap<String, String>,
        /// Request body bytes (may be empty).
        body: Vec<u8>,
    },
}

// ============================================================================
// QueueFn — registry of callable functions
// ============================================================================

/// Host-side registry entry for a `#[queue_fn]` discovered across loaded
/// wasm apps. The worker dispatches by looking up `(app_id, name)` in the
/// `HashMap<(AppId, String), QueueFn>` that the runtime maintains.
///
/// The trampoline is the per-function shim the attribute macro emits: it
/// deserializes the msgpack args tuple and calls the user function.
#[derive(Clone)]
pub struct QueueFn {
    /// Canonical name (matches the user function identifier).
    pub name: &'static str,
    /// Signature hash — `xxh3_64` of the canonicalized signature string.
    pub sig_hash: u64,
    /// Trampoline: takes `&Context` + serialized args, returns the user
    /// function's JSON result (or an error).
    pub trampoline:
        fn(ctx: &crate::resource::Context, args: &[u8]) -> crate::error::Result<serde_json::Value>,
}

impl std::fmt::Debug for QueueFn {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QueueFn")
            .field("name", &self.name)
            .field("sig_hash", &format_args!("{:#018x}", self.sig_hash))
            .field("trampoline", &"<fn pointer>")
            .finish()
    }
}

// Host-local registry: the `#[queue_fn]` proc-macro's `inventory::submit!`
// populates this at link time for functions compiled into the host binary
// (static plugins).
inventory::collect!(QueueFn);

/// Look up a host-local `QueueFn` by name (ignores `app_id` — static-crate
/// functions live in the same linker table as the host binary).
#[must_use]
pub fn lookup_queue_fn(name: &str) -> Option<&'static QueueFn> {
    inventory::iter::<QueueFn>().find(|f| f.name == name)
}

// ============================================================================
// App-scoped queue-fn entry — owned wrapper around `QueueFn`
// ============================================================================

/// App-scoped queue-fn entry.
///
/// `name` is owned for return-type uniformity; today every entry comes
/// out of the host's static inventory and `app_id` is always `"host"`.
#[derive(Clone)]
pub struct QueueFnEntry {
    /// Owning application ID (always `"host"` today — queue fns live in
    /// the host's static inventory; kept on the type so dispatch sites
    /// don't churn if a per-app surface returns).
    pub app_id: String,
    /// Function name (owned for return-type uniformity).
    pub name: String,
    /// `xxh3_64` of the canonicalized signature, as produced by `#[queue_fn]`.
    pub sig_hash: u64,
    /// Function pointer to the trampoline.
    pub trampoline: fn(&crate::resource::Context, &[u8]) -> crate::error::Result<serde_json::Value>,
}

impl std::fmt::Debug for QueueFnEntry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QueueFnEntry")
            .field("app_id", &self.app_id)
            .field("name", &self.name)
            .field("sig_hash", &format_args!("{:#018x}", self.sig_hash))
            .finish_non_exhaustive()
    }
}

/// Look up a queue function for a specific app. Wraps the static-
/// inventory lookup in a [`QueueFnEntry`] for uniform return shape.
/// `app_id` is unused today (every registered `queue_fn` lives in the
/// host's static inventory) but kept on the signature so dispatch
/// sites don't have to change if a per-app surface returns.
#[must_use]
pub fn lookup_queue_fn_for_app(_app_id: &str, name: &str) -> Option<QueueFnEntry> {
    lookup_queue_fn(name).map(|f| QueueFnEntry {
        app_id: "host".to_owned(),
        name: f.name.to_owned(),
        sig_hash: f.sig_hash,
        trampoline: f.trampoline,
    })
}

// ============================================================================
// Wasm-app queue dispatch hook
// ============================================================================

/// Process-wide hook for routing queue jobs into wasm components.
///
/// yeti-queue's dispatcher (in `crates/runtime/yeti-queue`) checks
/// the static [`inventory`]-built native registry first via
/// [`lookup_queue_fn_for_app`]. When that lookup misses AND a wasm
/// invoker is installed via [`set_wasm_queue_invoker`], the
/// dispatcher routes through here: the invoker is expected to find
/// the wasm component for `app_id`, call its `invoke-queue-fn`
/// export with `(name, payload)`, and return either the
/// msgpack-encoded result or an error message.
///
/// Layered this way (`OnceLock<Arc<dyn ...>>` in yeti-types, impl
/// installed from yeti-server at startup) to avoid pulling
/// yeti-wasm-host into yeti-queue's dep tree.
pub trait WasmQueueInvoker: Send + Sync {
    /// Invoke a `queue_fn` on a wasm component.
    ///
    /// `app_id` identifies the owning app; `name` the function;
    /// `payload` is the msgpack-encoded arg tuple. The implementor
    /// looks up the live wasm component, calls its
    /// `invoke-queue-fn` export, and returns the bytes the
    /// component handed back (typically a msgpack-encoded
    /// `serde_json::Value`).
    ///
    /// Returns `Err(msg)` if the app isn't a wasm app, the function
    /// isn't registered, or the wasm invocation traps / errors.
    fn invoke<'a>(
        &'a self,
        app_id: &'a str,
        name: &'a str,
        payload: &'a [u8],
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send + 'a>>;

    /// `true` if `app_id` is a registered wasm app this invoker
    /// knows about. Lets the dispatcher distinguish "unknown app"
    /// (let the static-inventory miss surface as
    /// "permanent failure") from "wasm app — try `invoke()`."
    fn knows_app(&self, app_id: &str) -> bool;
}

/// Process-wide wasm queue invoker — installed once at startup by
/// yeti-server.
static WASM_QUEUE_INVOKER: std::sync::OnceLock<std::sync::Arc<dyn WasmQueueInvoker>> =
    std::sync::OnceLock::new();

/// Install the process-wide wasm queue invoker. Idempotent —
/// subsequent calls are no-ops (`OnceLock`).
pub fn set_wasm_queue_invoker(invoker: std::sync::Arc<dyn WasmQueueInvoker>) {
    let _ = WASM_QUEUE_INVOKER.set(invoker);
}

/// Get the installed wasm queue invoker, if any. yeti-queue's
/// dispatcher calls this on every Function-payload job to decide
/// whether to route through wasm.
#[must_use]
pub fn wasm_queue_invoker() -> Option<&'static std::sync::Arc<dyn WasmQueueInvoker>> {
    WASM_QUEUE_INVOKER.get()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn lookup_for_app_falls_back_to_static_inventory() {
        // No `#[queue_fn]` named `never_registered_fn` exists in the
        // static inventory; the lookup misses and returns None.
        let app = format!("test-app-missing-{}", std::process::id());
        assert!(lookup_queue_fn_for_app(&app, "never_registered_fn").is_none());
    }
}

// `#[schedule(every = "...")]` types moved to the `yeti-schedule` crate.
// Queue and schedule are sibling concepts (the schedule registry enqueues
// queue jobs at fire-time, but no runtime logic is shared), so the type
// surfaces live in independent foundation crates.

// The plan originally defined a typed `QueueEvent` broadcast channel at
// 1024 capacity. The shipped worker uses the default table-pubsub
// channel (`SubscriptionMessage`, capacity 256) — every mutation of
// `yeti-queue` flows through there automatically, and the worker
// scans the row rather than decoding an event payload. The typed
// `QueueEvent` is therefore not needed: keeping it as dead code invites
// drift. Re-introduce it if a future consumer genuinely needs typed
// event payloads without reading the table.