cufflink_fn/

lib.rs

//! # cufflink-fn
//!
//! Write custom Cufflink handlers in Rust that compile to WASM.
//!
//! This crate wraps the raw WASM host ABI so you write normal Rust code
//! instead of pointer manipulation. Use it with `cufflink` services running
//! in WASM mode.
//!
//! ## Quick Start
//!
//! ```rust,ignore
//! use cufflink_fn::prelude::*;
//!
//! cufflink_fn::init!();
//!
//! handler!(hello, |req: Request| {
//!     let name = req.body()["name"].as_str().unwrap_or("world");
//!     Response::json(&json!({"message": format!("Hello, {}!", name)}))
//! });
//! ```
//!
//! ## Architecture
//!
//! Organize your code in layers:
//!
//! - **Handlers** (thin) — parse request, call operation, return response
//! - **Operations** (fat) — validation, business rules, orchestration
//! - **Repos** (data) — pure SQL via [`db::query`] / [`db::execute`]

use serde_json::Value;
use std::collections::HashMap;

/// Read the host response buffer after a batch host call and parse it as a
/// JSON array. Returns `None` if the buffer was empty or non-JSON.
#[cfg(target_arch = "wasm32")]
pub(crate) fn read_batch_response() -> Option<Vec<Value>> {
    let len = unsafe { get_host_response_len() };
    if len <= 0 {
        return None;
    }
    let mut buf = vec![0u8; len as usize];
    let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
    if read <= 0 {
        return None;
    }
    buf.truncate(read as usize);
    let s = String::from_utf8(buf).ok()?;
    let v: Value = serde_json::from_str(&s).ok()?;
    v.as_array().cloned()
}

// ─── Raw FFI ─────────────────────────────────────────────────────────────────
// These are the host functions provided by the Cufflink platform WASM runtime.
// Users never call these directly — use the `db`, `nats`, and `log` modules.

#[cfg(target_arch = "wasm32")]
extern "C" {
    #[link_name = "cufflink_log"]
    fn cufflink_log(level: i32, msg_ptr: i32, msg_len: i32);
    fn db_query(sql_ptr: i32, sql_len: i32) -> i32;
    fn db_execute(sql_ptr: i32, sql_len: i32) -> i32;
    fn get_host_response_len() -> i32;
    fn get_host_response(buf_ptr: i32, buf_len: i32) -> i32;
    fn nats_publish(subj_ptr: i32, subj_len: i32, payload_ptr: i32, payload_len: i32) -> i32;
    fn nats_request(
        subj_ptr: i32,
        subj_len: i32,
        payload_ptr: i32,
        payload_len: i32,
        timeout_ms: i32,
    ) -> i32;
    fn http_fetch(
        method_ptr: i32,
        method_len: i32,
        url_ptr: i32,
        url_len: i32,
        headers_ptr: i32,
        headers_len: i32,
        body_ptr: i32,
        body_len: i32,
    ) -> i32;
    fn get_config(key_ptr: i32, key_len: i32) -> i32;
    fn s3_download(bucket_ptr: i32, bucket_len: i32, key_ptr: i32, key_len: i32) -> i32;
    fn s3_download_many(items_ptr: i32, items_len: i32) -> i32;
    fn http_fetch_many(items_ptr: i32, items_len: i32) -> i32;
    fn image_transform_jpeg(
        bucket_ptr: i32,
        bucket_len: i32,
        in_key_ptr: i32,
        in_key_len: i32,
        out_key_ptr: i32,
        out_key_len: i32,
        max_dim: i32,
        quality: i32,
    ) -> i32;
    fn s3_presign_upload(
        bucket_ptr: i32,
        bucket_len: i32,
        key_ptr: i32,
        key_len: i32,
        content_type_ptr: i32,
        content_type_len: i32,
        expires_secs: i32,
    ) -> i32;
    fn redis_get(key_ptr: i32, key_len: i32) -> i32;
    fn redis_set(key_ptr: i32, key_len: i32, val_ptr: i32, val_len: i32, ttl_secs: i32) -> i32;
    fn redis_del(key_ptr: i32, key_len: i32) -> i32;
    fn generate_uuid() -> i32;
    fn current_time() -> i32;
}

// ─── Auth ────────────────────────────────────────────────────────────────────

/// Authenticated user context, validated by the Cufflink platform.
///
/// The platform validates the JWT token (via Keycloak) and extracts claims
/// before passing them to your handler. You never need to validate tokens
/// yourself — the `auth` field is only present when the token is valid.
///
/// ```rust,ignore
/// handler!(protected, |req: Request| {
///     let auth = match req.require_auth() {
///         Ok(auth) => auth,
///         Err(resp) => return resp,
///     };
///     if !auth.has_role("admin") {
///         return Response::error("Forbidden");
///     }
///     Response::json(&json!({"user": auth.sub}))
/// });
/// ```
#[derive(Debug, Clone)]
pub struct Auth {
    /// Keycloak subject ID (unique user identifier).
    pub sub: String,
    /// Preferred username from Keycloak.
    pub preferred_username: Option<String>,
    /// Display name.
    pub name: Option<String>,
    /// Email address.
    pub email: Option<String>,
    /// Realm roles assigned to the user in Keycloak.
    pub realm_roles: Vec<String>,
    /// All other JWT claims (custom Keycloak mappers, resource_access, etc.).
    pub claims: HashMap<String, Value>,
    /// Cufflink permissions resolved from the service's tenant roles (e.g., `["staff:create", "items:*"]`).
    pub permissions: Vec<String>,
    /// Cufflink role names assigned to the user (e.g., `["admin", "manager"]`).
    pub role_names: Vec<String>,
    /// Whether this is a Keycloak service account (client credentials grant).
    /// Service accounts bypass permission checks at the platform level.
    pub is_service_account: bool,
}

impl Auth {
    /// Check if the user has a specific Keycloak realm role.
    pub fn has_role(&self, role: &str) -> bool {
        self.realm_roles.iter().any(|r| r == role)
    }

    /// Check if the user has a specific Cufflink permission.
    ///
    /// Supports wildcards: `"staff:*"` matches any operation in the "staff" area,
    /// and `"*"` matches everything.
    ///
    /// ```rust,ignore
    /// if !auth.can("staff", "create") {
    ///     return Response::error("Forbidden: missing staff:create permission");
    /// }
    /// ```
    pub fn can(&self, area: &str, operation: &str) -> bool {
        let required = format!("{}:{}", area, operation);
        let wildcard = format!("{}:*", area);
        self.permissions
            .iter()
            .any(|p| p == &required || p == &wildcard || p == "*")
    }

    /// Check if the user has a specific Cufflink role (by name).
    pub fn has_cufflink_role(&self, role: &str) -> bool {
        self.role_names.iter().any(|r| r == role)
    }

    /// Get a specific claim value by key.
    pub fn claim(&self, key: &str) -> Option<&Value> {
        self.claims.get(key)
    }
}

// ─── JobContext ──────────────────────────────────────────────────────────────

/// Context the platform attaches to a request when it was delivered via the
/// long-running jobs runtime. Absent on direct HTTP invocations.
///
/// Use this when a handler needs to distinguish a fresh invocation from a
/// worker redelivery — e.g. to skip an idempotency-sensitive side effect that
/// the prior attempt already paid for.
#[derive(Debug, Clone)]
pub struct JobContext {
    pub id: String,
    pub attempt: u32,
    pub max_attempts: u32,
}

impl JobContext {
    /// `true` once the worker has already delivered this job at least once.
    pub fn is_retry(&self) -> bool {
        self.attempt > 1
    }
}

// ─── Request ─────────────────────────────────────────────────────────────────

/// An incoming HTTP request from the Cufflink platform.
///
/// The platform serializes the full request context (method, headers, body,
/// tenant, service name, auth) into JSON and passes it to your handler.
#[derive(Debug, Clone)]
pub struct Request {
    method: String,
    handler: String,
    headers: HashMap<String, String>,
    body: Value,
    raw_body: Vec<u8>,
    tenant: String,
    service: String,
    auth: Option<Auth>,
    job: Option<JobContext>,
}

impl Request {
    /// Parse a `Request` from the JSON the platform provides.
    pub fn from_json(json: &str) -> Option<Self> {
        let v: Value = serde_json::from_str(json).ok()?;
        let headers = v["headers"]
            .as_object()
            .map(|m| {
                m.iter()
                    .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                    .collect()
            })
            .unwrap_or_default();

        let auth = v["auth"].as_object().map(|auth_obj| {
            let a = Value::Object(auth_obj.clone());
            Auth {
                sub: a["sub"].as_str().unwrap_or("").to_string(),
                preferred_username: a["preferred_username"].as_str().map(|s| s.to_string()),
                name: a["name"].as_str().map(|s| s.to_string()),
                email: a["email"].as_str().map(|s| s.to_string()),
                realm_roles: a["realm_roles"]
                    .as_array()
                    .map(|arr| {
                        arr.iter()
                            .filter_map(|v| v.as_str().map(|s| s.to_string()))
                            .collect()
                    })
                    .unwrap_or_default(),
                claims: a["claims"]
                    .as_object()
                    .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
                    .unwrap_or_default(),
                permissions: a["permissions"]
                    .as_array()
                    .map(|arr| {
                        arr.iter()
                            .filter_map(|v| v.as_str().map(|s| s.to_string()))
                            .collect()
                    })
                    .unwrap_or_default(),
                role_names: a["role_names"]
                    .as_array()
                    .map(|arr| {
                        arr.iter()
                            .filter_map(|v| v.as_str().map(|s| s.to_string()))
                            .collect()
                    })
                    .unwrap_or_default(),
                is_service_account: a["is_service_account"].as_bool().unwrap_or(false),
            }
        });

        let raw_body = v["body_raw_b64"]
            .as_str()
            .filter(|s| !s.is_empty())
            .and_then(|s| {
                use base64::{engine::general_purpose, Engine};
                general_purpose::STANDARD.decode(s).ok()
            })
            .unwrap_or_default();

        let job = v["_job"].as_object().and_then(|j| {
            let id = j.get("id")?.as_str()?.to_string();
            let attempt = u32::try_from(j.get("attempt")?.as_u64()?).ok()?;
            let max_attempts = u32::try_from(j.get("max_attempts")?.as_u64()?).ok()?;
            Some(JobContext {
                id,
                attempt,
                max_attempts,
            })
        });

        Some(Self {
            method: v["method"].as_str().unwrap_or("GET").to_string(),
            handler: v["handler"].as_str().unwrap_or("").to_string(),
            headers,
            body: v["body"].clone(),
            raw_body,
            tenant: v["tenant"].as_str().unwrap_or("").to_string(),
            service: v["service"].as_str().unwrap_or("").to_string(),
            auth,
            job,
        })
    }

    /// The HTTP method (GET, POST, PUT, DELETE).
    pub fn method(&self) -> &str {
        &self.method
    }

    /// The handler name from the URL path.
    pub fn handler(&self) -> &str {
        &self.handler
    }

    /// All HTTP headers as a map.
    pub fn headers(&self) -> &HashMap<String, String> {
        &self.headers
    }

    /// Get a specific header value.
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(name).map(|s| s.as_str())
    }

    /// The parsed JSON body. Returns `Value::Null` if no body was sent.
    pub fn body(&self) -> &Value {
        &self.body
    }

    /// The raw, unparsed request body bytes as the caller sent them.
    ///
    /// Use this when verifying webhook signatures that HMAC over the
    /// original payload (Stripe, OpenPhone/Quo, etc.) — the parsed
    /// `body()` is canonicalized by `serde_json` and will not byte-match
    /// what the sender signed. Returns an empty slice when no body was
    /// sent.
    pub fn raw_body(&self) -> &[u8] {
        &self.raw_body
    }

    /// The tenant slug from the URL.
    pub fn tenant(&self) -> &str {
        &self.tenant
    }

    /// The service name from the URL.
    pub fn service(&self) -> &str {
        &self.service
    }

    /// Get the authenticated user context, if present.
    ///
    /// Returns `None` if no valid JWT or API key was provided with the request.
    pub fn auth(&self) -> Option<&Auth> {
        self.auth.as_ref()
    }

    /// Get the job context attached by the long-running jobs runtime.
    ///
    /// Returns `None` when the handler is invoked directly via HTTP. Present
    /// for handlers annotated with `job(...)` and dispatched by a worker.
    pub fn job(&self) -> Option<&JobContext> {
        self.job.as_ref()
    }

    /// Require authentication. Returns the auth context or an error response.
    ///
    /// ```rust,ignore
    /// handler!(protected, |req: Request| {
    ///     let auth = match req.require_auth() {
    ///         Ok(auth) => auth,
    ///         Err(resp) => return resp,
    ///     };
    ///     Response::json(&json!({"user": auth.sub}))
    /// });
    /// ```
    pub fn require_auth(&self) -> Result<&Auth, Response> {
        self.auth.as_ref().ok_or_else(|| {
            Response::json(&serde_json::json!({
                "error": "Authentication required",
                "status": 401
            }))
        })
    }
}

// ─── Response ────────────────────────────────────────────────────────────────

/// An HTTP response to return from your handler.
#[derive(Debug, Clone)]
pub struct Response {
    data: String,
    status: u16,
}

impl Response {
    /// Return a JSON response with HTTP 200.
    pub fn json(value: &Value) -> Self {
        Self {
            data: serde_json::to_string(value).unwrap_or_else(|_| "{}".to_string()),
            status: 200,
        }
    }

    /// Return a plain text response (wrapped in a JSON string).
    pub fn text(s: &str) -> Self {
        Self::json(&Value::String(s.to_string()))
    }

    /// Return an error response with HTTP 400.
    pub fn error(message: &str) -> Self {
        Self {
            data: serde_json::json!({"error": message}).to_string(),
            status: 400,
        }
    }

    /// Return a 404 Not Found error.
    pub fn not_found(message: &str) -> Self {
        Self {
            data: serde_json::json!({"error": message}).to_string(),
            status: 404,
        }
    }

    /// Return a 403 Forbidden error.
    pub fn forbidden(message: &str) -> Self {
        Self {
            data: serde_json::json!({"error": message}).to_string(),
            status: 403,
        }
    }

    /// Return an empty success response.
    pub fn empty() -> Self {
        Self::json(&serde_json::json!({"ok": true}))
    }

    /// Set a custom HTTP status code on the response.
    pub fn with_status(mut self, status: u16) -> Self {
        self.status = status;
        self
    }

    /// Get the raw response string.
    /// Encodes the status code into the response so the platform can extract it.
    pub fn into_data(self) -> String {
        if self.status == 200 {
            // No wrapping needed for 200 — backwards compatible
            self.data
        } else {
            // Wrap with __status so the platform can set the HTTP status code
            serde_json::json!({
                "__status": self.status,
                "__body": serde_json::from_str::<Value>(&self.data).unwrap_or(Value::String(self.data)),
            })
            .to_string()
        }
    }
}

// ─── db module ───────────────────────────────────────────────────────────────

/// Database access — run SQL queries against your service's tables.
///
/// All queries run in the tenant's schema automatically. You don't need
/// to qualify table names with a schema prefix.
pub mod db {
    use super::*;

    /// Run a SELECT query and return all rows as a `Vec<Value>`.
    ///
    /// Each row is a JSON object with column names as keys.
    ///
    /// ```rust,ignore
    /// let users = db::query("SELECT id, name, email FROM users WHERE active = true");
    /// for user in &users {
    ///     log::info(&format!("User: {}", user["name"]));
    /// }
    /// ```
    pub fn query(sql: &str) -> Vec<Value> {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = sql.as_bytes();
            let result = unsafe { db_query(bytes.as_ptr() as i32, bytes.len() as i32) };
            if result < 0 {
                return vec![];
            }
            read_host_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = sql;
            vec![]
        }
    }

    /// Run a SELECT query and return the first row, or `None` if empty.
    ///
    /// ```rust,ignore
    /// if let Some(user) = db::query_one("SELECT * FROM users WHERE id = 'abc'") {
    ///     log::info(&format!("Found user: {}", user["name"]));
    /// }
    /// ```
    pub fn query_one(sql: &str) -> Option<Value> {
        query(sql).into_iter().next()
    }

    /// Run an INSERT, UPDATE, or DELETE statement.
    ///
    /// Returns the number of affected rows, or -1 on error.
    ///
    /// ```rust,ignore
    /// let affected = db::execute("UPDATE orders SET status = 'shipped' WHERE id = 'abc'");
    /// log::info(&format!("Updated {} rows", affected));
    /// ```
    pub fn execute(sql: &str) -> i32 {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = sql.as_bytes();
            unsafe { db_execute(bytes.as_ptr() as i32, bytes.len() as i32) }
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = sql;
            0
        }
    }

    /// Read the host response buffer (used internally after db_query).
    #[cfg(target_arch = "wasm32")]
    fn read_host_response() -> Vec<Value> {
        let len = unsafe { get_host_response_len() };
        if len <= 0 {
            return vec![];
        }
        let mut buf = vec![0u8; len as usize];
        let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
        if read <= 0 {
            return vec![];
        }
        buf.truncate(read as usize);
        let json_str = String::from_utf8_lossy(&buf);
        serde_json::from_str(&json_str).unwrap_or_default()
    }
}

// ─── nats module ─────────────────────────────────────────────────────────────

/// Publish messages to NATS for event-driven communication.
///
/// Use this to notify other services, trigger subscriptions, or emit
/// domain events.
pub mod nats {
    #[allow(unused_imports)]
    use super::*;

    /// Publish a message to a NATS subject.
    ///
    /// Returns `true` on success, `false` on failure.
    ///
    /// ```rust,ignore
    /// nats::publish(
    ///     "dw.acme.order-service.orders.created",
    ///     &serde_json::json!({"order_id": "abc", "total": 4500}).to_string(),
    /// );
    /// ```
    pub fn publish(subject: &str, payload: &str) -> bool {
        #[cfg(target_arch = "wasm32")]
        {
            let subj_bytes = subject.as_bytes();
            let payload_bytes = payload.as_bytes();
            let result = unsafe {
                nats_publish(
                    subj_bytes.as_ptr() as i32,
                    subj_bytes.len() as i32,
                    payload_bytes.as_ptr() as i32,
                    payload_bytes.len() as i32,
                )
            };
            result == 0
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (subject, payload);
            true
        }
    }

    /// Send a NATS request and wait for a reply (synchronous request-reply).
    ///
    /// Returns the reply payload as a string, or `None` on timeout/failure.
    ///
    /// ```rust,ignore
    /// let reply = nats::request(
    ///     "dw.acme.user-service.users.lookup",
    ///     &serde_json::json!({"customer_id": "abc"}).to_string(),
    ///     5000, // timeout in ms
    /// );
    /// ```
    pub fn request(subject: &str, payload: &str, timeout_ms: i32) -> Option<String> {
        #[cfg(target_arch = "wasm32")]
        {
            let subj_bytes = subject.as_bytes();
            let payload_bytes = payload.as_bytes();
            let result = unsafe {
                nats_request(
                    subj_bytes.as_ptr() as i32,
                    subj_bytes.len() as i32,
                    payload_bytes.as_ptr() as i32,
                    payload_bytes.len() as i32,
                    timeout_ms,
                )
            };
            if result != 0 {
                return None;
            }
            let len = unsafe { get_host_response_len() };
            if len <= 0 {
                return None;
            }
            let mut buf = vec![0u8; len as usize];
            let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
            if read <= 0 {
                return None;
            }
            String::from_utf8(buf[..read as usize].to_vec()).ok()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (subject, payload, timeout_ms);
            None
        }
    }
}

// ─── log module ──────────────────────────────────────────────────────────────

/// Structured logging from inside your WASM handler.
///
/// Messages appear in the platform's log output prefixed with `[wasm]`.
pub mod log {
    #[allow(unused_imports)]
    use super::*;

    /// Log an error message (level 0).
    pub fn error(msg: &str) {
        write(0, msg);
    }

    /// Log a warning message (level 1).
    pub fn warn(msg: &str) {
        write(1, msg);
    }

    /// Log an info message (level 2).
    pub fn info(msg: &str) {
        write(2, msg);
    }

    /// Log a debug message (level 3).
    pub fn debug(msg: &str) {
        write(3, msg);
    }

    fn write(level: i32, msg: &str) {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = msg.as_bytes();
            unsafe {
                super::cufflink_log(level, bytes.as_ptr() as i32, bytes.len() as i32);
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (level, msg);
        }
    }
}

// ─── http module ────────────────────────────────────────────────────────────

/// Make HTTP requests from inside your WASM handler.
///
/// Use this to call external APIs (Keycloak admin, third-party services, etc.)
/// from your handler code.
pub mod http {
    #[allow(unused_imports)]
    use super::*;

    /// Response from an HTTP request.
    #[derive(Debug, Clone)]
    pub struct FetchResponse {
        /// HTTP status code (e.g., 200, 404, 500).
        pub status: i32,
        /// Response body as a string (may be base64-encoded for binary content).
        pub body: String,
        /// Body encoding: "utf8" for text, "base64" for binary content.
        pub body_encoding: String,
        /// Response headers.
        pub headers: HashMap<String, String>,
    }

    impl FetchResponse {
        /// Parse the response body as JSON.
        pub fn json(&self) -> Option<Value> {
            serde_json::from_str(&self.body).ok()
        }

        /// Check if the response status indicates success (2xx).
        pub fn is_success(&self) -> bool {
            (200..300).contains(&self.status)
        }

        /// Check if the body is base64-encoded (binary content).
        pub fn is_base64(&self) -> bool {
            self.body_encoding == "base64"
        }
    }

    /// Make an HTTP request.
    ///
    /// ```rust,ignore
    /// let resp = http::fetch("GET", "https://api.example.com/data", &[], None);
    /// if let Some(resp) = resp {
    ///     if resp.is_success() {
    ///         log::info(&format!("Got: {}", resp.body));
    ///     }
    /// }
    /// ```
    pub fn fetch(
        method: &str,
        url: &str,
        headers: &[(&str, &str)],
        body: Option<&str>,
    ) -> Option<FetchResponse> {
        #[cfg(target_arch = "wasm32")]
        {
            let method_bytes = method.as_bytes();
            let url_bytes = url.as_bytes();
            let headers_map: HashMap<&str, &str> = headers.iter().copied().collect();
            let headers_json = serde_json::to_string(&headers_map).unwrap_or_default();
            let headers_bytes = headers_json.as_bytes();
            let body_bytes = body.unwrap_or("").as_bytes();
            let body_len = body.map(|b| b.len()).unwrap_or(0);

            let result = unsafe {
                http_fetch(
                    method_bytes.as_ptr() as i32,
                    method_bytes.len() as i32,
                    url_bytes.as_ptr() as i32,
                    url_bytes.len() as i32,
                    headers_bytes.as_ptr() as i32,
                    headers_bytes.len() as i32,
                    body_bytes.as_ptr() as i32,
                    body_len as i32,
                )
            };

            if result < 0 {
                return None;
            }

            read_fetch_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (method, url, headers, body);
            None
        }
    }

    /// Make a GET request.
    pub fn get(url: &str, headers: &[(&str, &str)]) -> Option<FetchResponse> {
        fetch("GET", url, headers, None)
    }

    /// Make a POST request with a body.
    pub fn post(url: &str, headers: &[(&str, &str)], body: &str) -> Option<FetchResponse> {
        fetch("POST", url, headers, Some(body))
    }

    /// Make a PUT request with a body.
    pub fn put(url: &str, headers: &[(&str, &str)], body: &str) -> Option<FetchResponse> {
        fetch("PUT", url, headers, Some(body))
    }

    /// Make a DELETE request.
    pub fn delete(url: &str, headers: &[(&str, &str)]) -> Option<FetchResponse> {
        fetch("DELETE", url, headers, None)
    }

    /// Make a PATCH request with a body.
    pub fn patch(url: &str, headers: &[(&str, &str)], body: &str) -> Option<FetchResponse> {
        fetch("PATCH", url, headers, Some(body))
    }

    /// A single request inside a [`fetch_many`] batch.
    #[derive(Debug, Clone)]
    pub struct FetchRequest<'a> {
        pub method: &'a str,
        pub url: &'a str,
        pub headers: &'a [(&'a str, &'a str)],
        pub body: Option<&'a str>,
    }

    /// Make N HTTP requests concurrently from the host. Per-item errors are
    /// reported in the returned slot (other slots still succeed). Use this
    /// when you'd otherwise call [`fetch`] in a loop — sequential blocking
    /// fetches inside WASM are linear, the host fan-out is parallel.
    ///
    /// The host bounds concurrency (default 32) so an unbounded list won't
    /// exhaust connections.
    ///
    /// ```rust,ignore
    /// let results = http::fetch_many(&[
    ///     http::FetchRequest { method: "GET", url: "https://a", headers: &[], body: None },
    ///     http::FetchRequest { method: "GET", url: "https://b", headers: &[], body: None },
    /// ]);
    /// ```
    pub fn fetch_many(requests: &[FetchRequest<'_>]) -> Vec<Result<FetchResponse, String>> {
        #[cfg(target_arch = "wasm32")]
        {
            let items: Vec<Value> = requests
                .iter()
                .map(|r| {
                    let h: HashMap<&str, &str> = r.headers.iter().copied().collect();
                    serde_json::json!({
                        "method": r.method,
                        "url": r.url,
                        "headers": h,
                        "body": r.body,
                    })
                })
                .collect();
            let payload = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_string());
            let bytes = payload.as_bytes();
            let rc = unsafe { super::http_fetch_many(bytes.as_ptr() as i32, bytes.len() as i32) };
            if rc < 0 {
                return requests
                    .iter()
                    .map(|_| Err("http_fetch_many host call failed".to_string()))
                    .collect();
            }
            super::read_batch_response()
                .map(|v| v.into_iter().map(parse_fetch_slot).collect())
                .unwrap_or_else(|| {
                    requests
                        .iter()
                        .map(|_| Err("malformed host response".to_string()))
                        .collect()
                })
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = requests;
            vec![]
        }
    }

    #[cfg(target_arch = "wasm32")]
    fn fetch_response_from_json(v: &Value) -> FetchResponse {
        FetchResponse {
            status: v["status"].as_i64().unwrap_or(0) as i32,
            body: v["body"].as_str().unwrap_or("").to_string(),
            body_encoding: v["body_encoding"].as_str().unwrap_or("utf8").to_string(),
            headers: v["headers"]
                .as_object()
                .map(|m| {
                    m.iter()
                        .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                        .collect()
                })
                .unwrap_or_default(),
        }
    }

    #[cfg(target_arch = "wasm32")]
    fn parse_fetch_slot(slot: Value) -> Result<FetchResponse, String> {
        if !slot["ok"].as_bool().unwrap_or(false) {
            return Err(slot["error"]
                .as_str()
                .unwrap_or("unknown error")
                .to_string());
        }
        Ok(fetch_response_from_json(&slot))
    }

    /// Read the host response buffer after http_fetch.
    #[cfg(target_arch = "wasm32")]
    fn read_fetch_response() -> Option<FetchResponse> {
        let len = unsafe { get_host_response_len() };
        if len <= 0 {
            return None;
        }
        let mut buf = vec![0u8; len as usize];
        let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
        if read <= 0 {
            return None;
        }
        buf.truncate(read as usize);
        let json_str = String::from_utf8_lossy(&buf);
        let v: Value = serde_json::from_str(&json_str).ok()?;
        Some(fetch_response_from_json(&v))
    }
}

// ─── config module ──────────────────────────────────────────────────────

/// Read service configuration values set via `cufflink config set`.
///
/// Config values are stored in the platform's `service_configs` table,
/// scoped to your service. Use `cufflink config set KEY VALUE [--secret]`
/// to set values via the CLI.
pub mod config {
    #[allow(unused_imports)]
    use super::*;

    /// Get a config value by key. Returns `None` if the key doesn't exist.
    ///
    /// ```rust,ignore
    /// let api_key = config::get("ANTHROPIC_API_KEY");
    /// if let Some(key) = api_key {
    ///     log::info(&format!("API key loaded ({} chars)", key.len()));
    /// }
    /// ```
    pub fn get(key: &str) -> Option<String> {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = key.as_bytes();
            let result = unsafe { get_config(bytes.as_ptr() as i32, bytes.len() as i32) };
            if result < 0 {
                return None;
            }
            read_config_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = key;
            None
        }
    }

    /// Read the host response buffer after get_config.
    #[cfg(target_arch = "wasm32")]
    fn read_config_response() -> Option<String> {
        let len = unsafe { get_host_response_len() };
        if len <= 0 {
            return None;
        }
        let mut buf = vec![0u8; len as usize];
        let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
        if read <= 0 {
            return None;
        }
        buf.truncate(read as usize);
        String::from_utf8(buf).ok()
    }
}

// ─── storage module ─────────────────────────────────────────────────────

/// Download files from S3-compatible object storage using the platform's credentials.
///
/// The platform uses its own S3 credentials (configured at deployment time) to
/// perform authenticated downloads. This works with AWS S3, Hetzner Object Storage,
/// MinIO, and any S3-compatible service.
pub mod storage {
    #[allow(unused_imports)]
    use super::*;

    /// Download a file from S3 and return its contents as a base64-encoded string.
    ///
    /// Returns `None` if the download fails (bucket not found, key not found,
    /// S3 not configured, etc.).
    ///
    /// ```rust,ignore
    /// if let Some(base64_data) = storage::download("my-bucket", "images/photo.jpg") {
    ///     log::info(&format!("Downloaded {} bytes of base64", base64_data.len()));
    /// }
    /// ```
    pub fn download(bucket: &str, key: &str) -> Option<String> {
        #[cfg(target_arch = "wasm32")]
        {
            let bucket_bytes = bucket.as_bytes();
            let key_bytes = key.as_bytes();
            let result = unsafe {
                s3_download(
                    bucket_bytes.as_ptr() as i32,
                    bucket_bytes.len() as i32,
                    key_bytes.as_ptr() as i32,
                    key_bytes.len() as i32,
                )
            };
            if result < 0 {
                return None;
            }
            read_storage_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (bucket, key);
            None
        }
    }

    /// Generate a presigned PUT URL for uploading a file directly to S3.
    ///
    /// The returned URL is valid for `expires_secs` seconds and allows
    /// unauthenticated PUT requests. Clients can upload by sending a PUT
    /// request to the URL with the file data as the body.
    ///
    /// ```rust,ignore
    /// if let Some(url) = storage::presign_upload("my-bucket", "uploads/photo.jpg", "image/jpeg", 300) {
    ///     // Return this URL to the client for direct upload
    ///     log::info(&format!("Upload URL: {}", url));
    /// }
    /// ```
    pub fn presign_upload(
        bucket: &str,
        key: &str,
        content_type: &str,
        expires_secs: u64,
    ) -> Option<String> {
        #[cfg(target_arch = "wasm32")]
        {
            let bucket_bytes = bucket.as_bytes();
            let key_bytes = key.as_bytes();
            let ct_bytes = content_type.as_bytes();
            let result = unsafe {
                s3_presign_upload(
                    bucket_bytes.as_ptr() as i32,
                    bucket_bytes.len() as i32,
                    key_bytes.as_ptr() as i32,
                    key_bytes.len() as i32,
                    ct_bytes.as_ptr() as i32,
                    ct_bytes.len() as i32,
                    expires_secs as i32,
                )
            };
            if result < 0 {
                return None;
            }
            read_storage_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (bucket, key, content_type, expires_secs);
            None
        }
    }

    /// Read the host response buffer after s3_download or s3_presign_upload.
    #[cfg(target_arch = "wasm32")]
    fn read_storage_response() -> Option<String> {
        let len = unsafe { get_host_response_len() };
        if len <= 0 {
            return None;
        }
        let mut buf = vec![0u8; len as usize];
        let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
        if read <= 0 {
            return None;
        }
        buf.truncate(read as usize);
        String::from_utf8(buf).ok()
    }

    /// Download N files from object storage concurrently. Each slot is
    /// either the decoded bytes or a per-item error message. Use this
    /// instead of looping over [`download`] when N is large — the host
    /// fans out in parallel with a bounded semaphore (default 32).
    pub fn download_many(items: &[(&str, &str)]) -> Vec<Result<Vec<u8>, String>> {
        #[cfg(target_arch = "wasm32")]
        {
            let payload_items: Vec<Value> = items
                .iter()
                .map(|(b, k)| serde_json::json!({ "bucket": b, "key": k }))
                .collect();
            let payload = serde_json::to_string(&payload_items).unwrap_or_else(|_| "[]".into());
            let bytes = payload.as_bytes();
            let rc = unsafe { super::s3_download_many(bytes.as_ptr() as i32, bytes.len() as i32) };
            if rc < 0 {
                return items
                    .iter()
                    .map(|_| Err("s3_download_many host call failed".to_string()))
                    .collect();
            }
            super::read_batch_response()
                .map(|v| v.into_iter().map(parse_download_slot).collect())
                .unwrap_or_else(|| {
                    items
                        .iter()
                        .map(|_| Err("malformed host response".to_string()))
                        .collect()
                })
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = items;
            vec![]
        }
    }

    #[cfg(target_arch = "wasm32")]
    fn parse_download_slot(slot: Value) -> Result<Vec<u8>, String> {
        if !slot["ok"].as_bool().unwrap_or(false) {
            return Err(slot["error"]
                .as_str()
                .unwrap_or("unknown error")
                .to_string());
        }
        let b64 = slot["data_b64"]
            .as_str()
            .ok_or_else(|| "missing data_b64".to_string())?;
        use base64::{engine::general_purpose, Engine};
        general_purpose::STANDARD
            .decode(b64)
            .map_err(|e| format!("base64 decode failed: {}", e))
    }
}

// ─── image module ───────────────────────────────────────────────────────

/// Image transforms executed on the platform host.
///
/// The platform downloads the source from object storage, decodes it,
/// optionally downscales the longest edge to `max_dim` (Lanczos3),
/// JPEG-encodes at the requested quality, and writes the result back to
/// the same bucket. CPU work runs in a host-side blocking task so it
/// does not stall the async runtime.
pub mod image {
    #[allow(unused_imports)]
    use super::*;

    /// Resize an image stored at `(bucket, in_key)`, JPEG-encode it, and
    /// upload to `(bucket, out_key)`.
    ///
    /// `max_dim == 0` skips resizing. `quality` must be in `1..=100`.
    /// Returns the number of bytes written, or `None` on failure (the
    /// host logs the underlying error).
    ///
    /// ```rust,ignore
    /// let bytes = image::transform_jpeg("media", "raw/photo.jpg", "raw/photo.display.jpg", 1024, 80);
    /// ```
    pub fn transform_jpeg(
        bucket: &str,
        in_key: &str,
        out_key: &str,
        max_dim: u32,
        quality: u8,
    ) -> Option<u32> {
        #[cfg(target_arch = "wasm32")]
        {
            let bucket_bytes = bucket.as_bytes();
            let in_key_bytes = in_key.as_bytes();
            let out_key_bytes = out_key.as_bytes();
            let result = unsafe {
                super::image_transform_jpeg(
                    bucket_bytes.as_ptr() as i32,
                    bucket_bytes.len() as i32,
                    in_key_bytes.as_ptr() as i32,
                    in_key_bytes.len() as i32,
                    out_key_bytes.as_ptr() as i32,
                    out_key_bytes.len() as i32,
                    max_dim as i32,
                    quality as i32,
                )
            };
            if result < 0 {
                return None;
            }
            Some(result as u32)
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (bucket, in_key, out_key, max_dim, quality);
            None
        }
    }
}

// ─── redis module ────────────────────────────────────────────────────────

/// Read and write values in Redis (backed by the platform's Redis connection).
///
/// Use this for caching, session storage, or any key-value data that needs
/// to be shared across services or requests with low latency.
pub mod redis {
    #[allow(unused_imports)]
    use super::*;

    /// Get a value from Redis by key. Returns `None` if the key doesn't exist
    /// or Redis is not configured.
    ///
    /// ```rust,ignore
    /// if let Some(cached) = redis::get("auth:perms:user-123") {
    ///     log::info(&format!("Cache hit: {}", cached));
    /// }
    /// ```
    pub fn get(key: &str) -> Option<String> {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = key.as_bytes();
            let result = unsafe { redis_get(bytes.as_ptr() as i32, bytes.len() as i32) };
            if result < 0 {
                return None;
            }
            read_redis_response()
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = key;
            None
        }
    }

    /// Set a value in Redis. Use `ttl_secs = 0` for no expiry.
    ///
    /// Returns `true` on success, `false` on failure.
    ///
    /// ```rust,ignore
    /// redis::set("auth:perms:user-123", &perms_json, 3600); // 1 hour TTL
    /// ```
    pub fn set(key: &str, value: &str, ttl_secs: i32) -> bool {
        #[cfg(target_arch = "wasm32")]
        {
            let key_bytes = key.as_bytes();
            let val_bytes = value.as_bytes();
            let result = unsafe {
                redis_set(
                    key_bytes.as_ptr() as i32,
                    key_bytes.len() as i32,
                    val_bytes.as_ptr() as i32,
                    val_bytes.len() as i32,
                    ttl_secs,
                )
            };
            result == 0
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = (key, value, ttl_secs);
            true
        }
    }

    /// Delete a key from Redis.
    ///
    /// Returns `true` on success, `false` on failure.
    ///
    /// ```rust,ignore
    /// redis::del("auth:perms:user-123");
    /// ```
    pub fn del(key: &str) -> bool {
        #[cfg(target_arch = "wasm32")]
        {
            let bytes = key.as_bytes();
            let result = unsafe { redis_del(bytes.as_ptr() as i32, bytes.len() as i32) };
            result == 0
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            let _ = key;
            true
        }
    }

    /// Read the host response buffer after redis_get.
    #[cfg(target_arch = "wasm32")]
    fn read_redis_response() -> Option<String> {
        let len = unsafe { get_host_response_len() };
        if len <= 0 {
            return None;
        }
        let mut buf = vec![0u8; len as usize];
        let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
        if read <= 0 {
            return None;
        }
        buf.truncate(read as usize);
        String::from_utf8(buf).ok()
    }
}

// ─── util module ────────────────────────────────────────────────────────

/// Utility functions for common operations in WASM handlers.
pub mod util {
    #[allow(unused_imports)]
    use super::*;

    /// Generate a new random UUID v4 string.
    ///
    /// ```rust,ignore
    /// let id = util::generate_uuid();
    /// log::info(&format!("New ID: {}", id));
    /// ```
    /// Get the current UTC time as an RFC3339 string.
    ///
    /// In WASM, this calls the platform host function. Outside WASM, it uses `SystemTime`.
    ///
    /// ```rust,ignore
    /// let now = util::current_time();
    /// log::info(&format!("Current time: {}", now));
    /// ```
    pub fn current_time() -> String {
        #[cfg(target_arch = "wasm32")]
        {
            let result = unsafe { super::current_time() };
            if result < 0 {
                return String::new();
            }
            let len = unsafe { get_host_response_len() };
            if len <= 0 {
                return String::new();
            }
            let mut buf = vec![0u8; len as usize];
            let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
            if read <= 0 {
                return String::new();
            }
            buf.truncate(read as usize);
            String::from_utf8(buf).unwrap_or_default()
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            let secs = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            format!("1970-01-01T00:00:00Z+{}", secs)
        }
    }

    pub fn generate_uuid() -> String {
        #[cfg(target_arch = "wasm32")]
        {
            let result = unsafe { super::generate_uuid() };
            if result < 0 {
                return String::new();
            }
            let len = unsafe { get_host_response_len() };
            if len <= 0 {
                return String::new();
            }
            let mut buf = vec![0u8; len as usize];
            let read = unsafe { get_host_response(buf.as_mut_ptr() as i32, len) };
            if read <= 0 {
                return String::new();
            }
            buf.truncate(read as usize);
            String::from_utf8(buf).unwrap_or_default()
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            format!(
                "{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_nanos() as u32)
                    .unwrap_or(0),
                std::process::id() as u16,
                0u16,
                0x8000u16,
                0u64,
            )
        }
    }
}

// ─── Handler runtime ─────────────────────────────────────────────────────────

/// Internal function used by the `handler!` macro. Do not call directly.
#[doc(hidden)]
pub fn __run_handler<F>(ptr: i32, len: i32, f: F) -> i32
where
    F: FnOnce(Request) -> Response,
{
    // Read the request JSON from guest memory
    let request_json = unsafe {
        let slice = std::slice::from_raw_parts(ptr as *const u8, len as usize);
        String::from_utf8_lossy(slice).into_owned()
    };

    // Parse the request
    let request = Request::from_json(&request_json).unwrap_or_else(|| Request {
        method: "GET".to_string(),
        handler: String::new(),
        headers: HashMap::new(),
        body: Value::Null,
        raw_body: Vec::new(),
        tenant: String::new(),
        service: String::new(),
        auth: None,
        job: None,
    });

    // Call the user's handler
    let response = f(request);
    let response_bytes = response.into_data().into_bytes();

    // Write response to guest memory: [4-byte LE length][data]
    let total = 4 + response_bytes.len();
    let layout = std::alloc::Layout::from_size_align(total, 1).expect("invalid layout");
    let out_ptr = unsafe { std::alloc::alloc(layout) };

    unsafe {
        let len_bytes = (response_bytes.len() as u32).to_le_bytes();
        std::ptr::copy_nonoverlapping(len_bytes.as_ptr(), out_ptr, 4);
        std::ptr::copy_nonoverlapping(
            response_bytes.as_ptr(),
            out_ptr.add(4),
            response_bytes.len(),
        );
    }

    out_ptr as i32
}

// ─── Macros ──────────────────────────────────────────────────────────────────

/// Initialize the cufflink-fn runtime. Call this once at the top of your `lib.rs`.
///
/// Exports the `alloc` function that the platform needs to pass data into
/// your WASM module.
///
/// ```rust,ignore
/// use cufflink_fn::prelude::*;
///
/// cufflink_fn::init!();
/// ```
#[macro_export]
macro_rules! init {
    () => {
        #[no_mangle]
        pub extern "C" fn alloc(size: i32) -> i32 {
            let layout = std::alloc::Layout::from_size_align(size as usize, 1).unwrap();
            unsafe { std::alloc::alloc(layout) as i32 }
        }
    };
}

/// Define a handler function.
///
/// This macro generates the `#[no_mangle] extern "C"` boilerplate so your
/// handler is a plain Rust closure that receives a [`Request`] and returns
/// a [`Response`].
///
/// ```rust,ignore
/// use cufflink_fn::prelude::*;
///
/// cufflink_fn::init!();
///
/// handler!(get_stats, |req: Request| {
///     let rows = db::query("SELECT COUNT(*) as total FROM orders");
///     Response::json(&json!({"total": rows[0]["total"]}))
/// });
///
/// handler!(create_order, |req: Request| {
///     let body = req.body();
///     let customer = body["customer_id"].as_str().unwrap_or("unknown");
///     db::execute(&format!(
///         "INSERT INTO orders (customer_id, status) VALUES ('{}', 'pending')",
///         customer
///     ));
///     Response::json(&json!({"status": "created"}))
/// });
/// ```
#[macro_export]
macro_rules! handler {
    ($name:ident, |$req:ident : Request| $body:expr) => {
        #[no_mangle]
        pub extern "C" fn $name(ptr: i32, len: i32) -> i32 {
            $crate::__run_handler(ptr, len, |$req: $crate::Request| $body)
        }
    };
}

// ─── Schema migration hook ───────────────────────────────────────────────────

/// Helpers for the optional `on_migrate` schema migration hook.
///
/// Declared in your service:
///
/// ```rust,ignore
/// cufflink::service! {
///     name: "logistics-service",
///     mode: wasm,
///     tables: [PickupRequest],
///     on_migrate: "handle_on_migrate",
///     // ...
/// }
/// ```
///
/// And implemented as a handler that delegates to [`migrate::run`]:
///
/// ```rust,ignore
/// use cufflink_fn::prelude::*;
///
/// handler!(handle_on_migrate, |req: Request| {
///     migrate::run(req, |diff| {
///         if diff.added_column("pickup_requests", "approximate_item_count_min")
///             && diff.dropped_column("pickup_requests", "approximate_item_count")
///         {
///             db::execute("UPDATE pickup_requests \
///                          SET approximate_item_count_min = approximate_item_count \
///                          WHERE approximate_item_count_min IS NULL");
///         }
///         Ok(())
///     })
/// });
/// ```
///
/// The closure must be **idempotent** — cufflink may invoke it on retried
/// deploys, on no-op deploys, and on first-time deploys against a fresh
/// database. Use `WHERE … IS NULL` guards on every UPDATE.
pub mod migrate {
    use super::{Request, Response};
    pub use cufflink_types::SchemaDiff;

    /// Parse the [`SchemaDiff`] from the request body and invoke `handler`.
    ///
    /// Returns a `200 {"ok": true}` response on success or a `400` with the
    /// error message if `handler` returns `Err`. If the request body fails
    /// to deserialise as a `SchemaDiff`, returns a `400` describing the
    /// parse error.
    pub fn run<F>(req: Request, handler: F) -> Response
    where
        F: FnOnce(SchemaDiff) -> Result<(), String>,
    {
        match serde_json::from_value::<SchemaDiff>(req.body().clone()) {
            Ok(diff) => match handler(diff) {
                Ok(()) => Response::json(&serde_json::json!({"ok": true})),
                Err(e) => Response::error(&e),
            },
            Err(e) => Response::error(&format!(
                "on_migrate: failed to parse SchemaDiff payload: {}",
                e
            )),
        }
    }
}

// ─── Prelude ─────────────────────────────────────────────────────────────────

/// Import everything you need to write handlers.
///
/// ```rust,ignore
/// use cufflink_fn::prelude::*;
/// ```
pub mod prelude {
    pub use crate::config;
    pub use crate::db;
    pub use crate::http;
    pub use crate::image;
    pub use crate::log;
    pub use crate::migrate;
    pub use crate::nats;
    pub use crate::redis;
    pub use crate::storage;
    pub use crate::util;
    pub use crate::Auth;
    pub use crate::JobContext;
    pub use crate::Request;
    pub use crate::Response;
    pub use serde_json::{json, Value};
}

// ─── Tests ───────────────────────────────────────────────────────────────────

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

    #[test]
    fn test_request_parsing() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "checkout",
            "headers": {"content-type": "application/json"},
            "body": {"item": "widget", "qty": 3},
            "tenant": "acme",
            "service": "shop"
        }))
        .unwrap();

        let req = Request::from_json(&json).unwrap();
        assert_eq!(req.method(), "POST");
        assert_eq!(req.handler(), "checkout");
        assert_eq!(req.tenant(), "acme");
        assert_eq!(req.service(), "shop");
        assert_eq!(req.body()["item"], "widget");
        assert_eq!(req.body()["qty"], 3);
        assert_eq!(req.header("content-type"), Some("application/json"));
    }

    #[test]
    fn test_request_missing_fields() {
        let json = r#"{"method": "GET"}"#;
        let req = Request::from_json(json).unwrap();
        assert_eq!(req.method(), "GET");
        assert_eq!(req.handler(), "");
        assert_eq!(req.tenant(), "");
        assert_eq!(req.body(), &Value::Null);
        assert!(req.raw_body().is_empty());
    }

    #[test]
    fn test_request_raw_body_round_trip() {
        use base64::{engine::general_purpose, Engine};
        let raw = br#"{"type":"message.delivered","data":{"object":{"id":"AC1"}}}"#;
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "webhook",
            "headers": {"content-type": "application/json"},
            "body": serde_json::from_slice::<Value>(raw).unwrap(),
            "body_raw_b64": general_purpose::STANDARD.encode(raw),
            "tenant": "acme",
            "service": "shop",
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert_eq!(req.raw_body(), raw);
    }

    #[test]
    fn test_request_raw_body_invalid_base64_yields_empty() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "webhook",
            "body": Value::Null,
            "body_raw_b64": "not%base64!",
            "tenant": "acme",
            "service": "shop",
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert!(req.raw_body().is_empty());
    }

    #[test]
    fn test_request_job_context_parsed_when_present() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "handle_run_ai",
            "body": {"id": "item-1"},
            "tenant": "acme",
            "service": "asset",
            "_job": {
                "id": "11111111-1111-1111-1111-111111111111",
                "attempt": 2,
                "max_attempts": 3,
            },
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        let job = req.job().expect("job context should be parsed");
        assert_eq!(job.id, "11111111-1111-1111-1111-111111111111");
        assert_eq!(job.attempt, 2);
        assert_eq!(job.max_attempts, 3);
        assert!(job.is_retry());
    }

    #[test]
    fn test_request_job_context_absent_on_http_direct() {
        let json = serde_json::to_string(&json!({
            "method": "GET",
            "handler": "list",
            "body": Value::Null,
            "tenant": "acme",
            "service": "asset",
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert!(req.job().is_none());
    }

    #[test]
    fn test_request_job_context_first_attempt_is_not_retry() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "handle_run_ai",
            "body": Value::Null,
            "tenant": "acme",
            "service": "asset",
            "_job": {
                "id": "22222222-2222-2222-2222-222222222222",
                "attempt": 1,
                "max_attempts": 2,
            },
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        let job = req.job().expect("job context should be parsed");
        assert_eq!(job.attempt, 1);
        assert!(!job.is_retry());
    }

    #[test]
    fn test_request_job_context_rejects_malformed_envelope() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "handle_run_ai",
            "body": Value::Null,
            "tenant": "acme",
            "service": "asset",
            "_job": {
                "id": "33333333-3333-3333-3333-333333333333",
                "max_attempts": 2,
            },
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert!(
            req.job().is_none(),
            "missing attempt should yield None, not a partial JobContext"
        );
    }

    #[test]
    fn test_response_json() {
        let resp = Response::json(&json!({"status": "ok", "count": 42}));
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        assert_eq!(parsed["status"], "ok");
        assert_eq!(parsed["count"], 42);
    }

    #[test]
    fn test_response_error() {
        let resp = Response::error("something went wrong");
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        // error() returns status 400, so into_data wraps with __status/__body
        assert_eq!(parsed["__status"], 400);
        assert_eq!(parsed["__body"]["error"], "something went wrong");
    }

    #[test]
    fn test_response_not_found() {
        let resp = Response::not_found("item not found");
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        assert_eq!(parsed["__status"], 404);
        assert_eq!(parsed["__body"]["error"], "item not found");
    }

    #[test]
    fn test_response_with_status() {
        let resp = Response::json(&serde_json::json!({"ok": true})).with_status(201);
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        assert_eq!(parsed["__status"], 201);
        assert_eq!(parsed["__body"]["ok"], true);
    }

    fn migrate_request(diff: serde_json::Value) -> Request {
        let payload = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "handle_on_migrate",
            "headers": {},
            "body": diff,
            "tenant": "default",
            "service": "logistics-service",
        }))
        .unwrap();
        Request::from_json(&payload).unwrap()
    }

    #[test]
    fn test_migrate_run_success() {
        let req = migrate_request(json!({
            "added_columns": [["pickups", "min"]],
            "dropped_columns": [["pickups", "midpoint"]],
        }));
        let resp = migrate::run(req, |diff| {
            assert!(diff.added_column("pickups", "min"));
            assert!(diff.dropped_column("pickups", "midpoint"));
            Ok(())
        });
        let parsed: Value = serde_json::from_str(&resp.into_data()).unwrap();
        assert_eq!(parsed["ok"], true);
    }

    #[test]
    fn test_migrate_run_handler_error() {
        let req = migrate_request(json!({}));
        let resp = migrate::run(req, |_| Err("backfill failed".into()));
        let parsed: Value = serde_json::from_str(&resp.into_data()).unwrap();
        assert_eq!(parsed["__status"], 400);
        assert_eq!(parsed["__body"]["error"], "backfill failed");
    }

    #[test]
    fn test_migrate_run_invalid_payload() {
        // body is a string, not a SchemaDiff object
        let req = migrate_request(json!("not a diff"));
        let resp = migrate::run(req, |_| {
            panic!("closure should not be called for invalid payload")
        });
        let parsed: Value = serde_json::from_str(&resp.into_data()).unwrap();
        assert_eq!(parsed["__status"], 400);
        assert!(parsed["__body"]["error"]
            .as_str()
            .unwrap()
            .contains("on_migrate: failed to parse SchemaDiff payload"));
    }

    #[test]
    fn test_migrate_run_empty_diff() {
        let req = migrate_request(json!({}));
        let mut called = false;
        let resp = migrate::run(req, |diff| {
            assert!(diff.is_empty());
            called = true;
            Ok(())
        });
        assert!(called);
        let parsed: Value = serde_json::from_str(&resp.into_data()).unwrap();
        assert_eq!(parsed["ok"], true);
    }

    #[test]
    fn test_response_200_no_wrapper() {
        let resp = Response::json(&serde_json::json!({"data": "test"}));
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        // 200 responses should NOT be wrapped
        assert_eq!(parsed["data"], "test");
        assert!(parsed.get("__status").is_none());
    }

    #[test]
    fn test_response_empty() {
        let resp = Response::empty();
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        assert_eq!(parsed["ok"], true);
    }

    #[test]
    fn test_response_text() {
        let resp = Response::text("hello world");
        let data = resp.into_data();
        let parsed: Value = serde_json::from_str(&data).unwrap();
        assert_eq!(parsed, "hello world");
    }

    #[test]
    fn test_db_query_noop_on_native() {
        // On native (non-wasm) targets, db functions are no-ops
        let rows = db::query("SELECT 1");
        assert!(rows.is_empty());
    }

    #[test]
    fn test_db_query_one_noop_on_native() {
        let row = db::query_one("SELECT 1");
        assert!(row.is_none());
    }

    #[test]
    fn test_db_execute_noop_on_native() {
        let affected = db::execute("INSERT INTO x VALUES (1)");
        assert_eq!(affected, 0);
    }

    #[test]
    fn test_nats_publish_noop_on_native() {
        let ok = nats::publish("test.subject", "payload");
        assert!(ok);
    }

    #[test]
    fn test_request_with_auth() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "checkout",
            "headers": {},
            "body": {},
            "tenant": "acme",
            "service": "shop",
            "auth": {
                "sub": "user-123",
                "preferred_username": "john",
                "name": "John Doe",
                "email": "john@example.com",
                "realm_roles": ["admin", "manager"],
                "claims": {"department": "engineering"}
            }
        }))
        .unwrap();

        let req = Request::from_json(&json).unwrap();
        let auth = req.auth().unwrap();
        assert_eq!(auth.sub, "user-123");
        assert_eq!(auth.preferred_username.as_deref(), Some("john"));
        assert_eq!(auth.name.as_deref(), Some("John Doe"));
        assert_eq!(auth.email.as_deref(), Some("john@example.com"));
        assert!(auth.has_role("admin"));
        assert!(auth.has_role("manager"));
        assert!(!auth.has_role("viewer"));
        assert_eq!(
            auth.claim("department").and_then(|v| v.as_str()),
            Some("engineering")
        );
    }

    #[test]
    fn test_request_without_auth() {
        let json = r#"{"method": "GET"}"#;
        let req = Request::from_json(json).unwrap();
        assert!(req.auth().is_none());
    }

    #[test]
    fn test_request_null_auth() {
        let json = serde_json::to_string(&json!({
            "method": "GET",
            "auth": null
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert!(req.auth().is_none());
    }

    #[test]
    fn test_require_auth_success() {
        let json = serde_json::to_string(&json!({
            "method": "GET",
            "auth": {"sub": "user-1", "realm_roles": [], "claims": {}}
        }))
        .unwrap();
        let req = Request::from_json(&json).unwrap();
        assert!(req.require_auth().is_ok());
        assert_eq!(req.require_auth().unwrap().sub, "user-1");
    }

    #[test]
    fn test_require_auth_fails_when_unauthenticated() {
        let json = r#"{"method": "GET"}"#;
        let req = Request::from_json(json).unwrap();
        assert!(req.require_auth().is_err());
    }

    #[test]
    fn test_http_fetch_noop_on_native() {
        let resp = http::fetch("GET", "https://example.com", &[], None);
        assert!(resp.is_none());
    }

    #[test]
    fn test_http_get_noop_on_native() {
        let resp = http::get("https://example.com", &[]);
        assert!(resp.is_none());
    }

    #[test]
    fn test_http_post_noop_on_native() {
        let resp = http::post("https://example.com", &[], "{}");
        assert!(resp.is_none());
    }

    #[test]
    fn test_storage_download_noop_on_native() {
        let data = storage::download("my-bucket", "images/photo.jpg");
        assert!(data.is_none());
    }

    #[test]
    fn test_image_transform_jpeg_noop_on_native() {
        let bytes = image::transform_jpeg("my-bucket", "in.jpg", "out.jpg", 1024, 80);
        assert!(bytes.is_none());
    }

    #[test]
    fn test_auth_permissions() {
        let json = serde_json::to_string(&json!({
            "method": "POST",
            "handler": "test",
            "headers": {},
            "body": {},
            "tenant": "acme",
            "service": "shop",
            "auth": {
                "sub": "user-1",
                "realm_roles": ["admin"],
                "claims": {},
                "permissions": ["staff:create", "staff:view", "items:*"],
                "role_names": ["admin", "manager"]
            }
        }))
        .unwrap();

        let req = Request::from_json(&json).unwrap();
        let auth = req.auth().unwrap();

        // Exact permission match
        assert!(auth.can("staff", "create"));
        assert!(auth.can("staff", "view"));
        assert!(!auth.can("staff", "delete"));

        // Wildcard match
        assert!(auth.can("items", "create"));
        assert!(auth.can("items", "view"));
        assert!(auth.can("items", "delete"));

        // No match
        assert!(!auth.can("batches", "view"));

        // Cufflink roles
        assert!(auth.has_cufflink_role("admin"));
        assert!(auth.has_cufflink_role("manager"));
        assert!(!auth.has_cufflink_role("viewer"));
    }

    #[test]
    fn test_auth_super_wildcard() {
        let auth = Auth {
            sub: "user-1".to_string(),
            preferred_username: None,
            name: None,
            email: None,
            realm_roles: vec![],
            claims: HashMap::new(),
            permissions: vec!["*".to_string()],
            role_names: vec!["superadmin".to_string()],
            is_service_account: false,
        };

        assert!(auth.can("anything", "everything"));
        assert!(auth.can("staff", "create"));
    }

    #[test]
    fn test_auth_empty_permissions() {
        let auth = Auth {
            sub: "user-1".to_string(),
            preferred_username: None,
            name: None,
            email: None,
            realm_roles: vec![],
            claims: HashMap::new(),
            permissions: vec![],
            role_names: vec![],
            is_service_account: false,
        };

        assert!(!auth.can("staff", "create"));
        assert!(!auth.has_cufflink_role("admin"));
    }

    #[test]
    fn test_redis_get_noop_on_native() {
        let val = redis::get("some-key");
        assert!(val.is_none());
    }

    #[test]
    fn test_redis_set_noop_on_native() {
        let ok = redis::set("key", "value", 3600);
        assert!(ok);
    }

    #[test]
    fn test_redis_del_noop_on_native() {
        let ok = redis::del("key");
        assert!(ok);
    }

    #[test]
    fn test_http_fetch_response_helpers() {
        let resp = http::FetchResponse {
            status: 200,
            body: r#"{"key": "value"}"#.to_string(),
            body_encoding: "utf8".to_string(),
            headers: HashMap::new(),
        };
        assert!(resp.is_success());
        assert!(!resp.is_base64());
        let json = resp.json().unwrap();
        assert_eq!(json["key"], "value");

        let err_resp = http::FetchResponse {
            status: 404,
            body: "not found".to_string(),
            body_encoding: "utf8".to_string(),
            headers: HashMap::new(),
        };
        assert!(!err_resp.is_success());

        let binary_resp = http::FetchResponse {
            status: 200,
            body: "aW1hZ2VkYXRh".to_string(),
            body_encoding: "base64".to_string(),
            headers: HashMap::new(),
        };
        assert!(binary_resp.is_base64());
    }
}