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

//! `http::` Rhai bridge — outbound HTTP from scripts (v1.1.4).
//!
//! ```rhai
//! let r = http::get("https://api.example.com/users/123");
//! let r = http::get(url, #{ headers: #{ "Authorization": "Bearer x" }, timeout_ms: 5000 });
//! let r = http::post(url, #{ text: "hello" });          // Map body → JSON
//! let r = http::post(url, "raw", #{ headers: #{ ... } }); // String body → text/plain
//! let r = http::post_form(url, #{ a: "1", b: "2" });    // form-encoded
//! let r = http::request("OPTIONS", url);
//! ```
//!
//! **Argument shape (v1.1.4 decision):** body and options are separate
//! positional arguments — `verb(url, body, opts)` — not body-inside-
//! opts. This keeps the unknown-opt-key typo guard intact and resolves
//! the brief's internal contradiction (its Slack example passed a bare
//! body map). The `opts` vocabulary is exactly
//! `{headers, timeout_ms, follow_redirects, max_redirects}`; any other
//! key throws.
//!
//! Body dispatch (positional `body`): Map/Array → JSON +
//! `application/json`; String → raw + `text/plain`; Unit `()` → no
//! body. GET/HEAD ignore any body.
//!
//! Response is a Rhai map `#{ status, headers, body, body_raw }`:
//! `body` is the parsed JSON when the response is `application/json`
//! and parses; `()` for an empty body; otherwise the raw string.
//!
//! Errors follow `docs/sdk-shape.md`: network/timeout/SSRF/size failures
//! throw (`"http: <message>"`); a non-2xx status does NOT throw — the
//! response map is returned, fetch-style.

use std::collections::BTreeMap;
use std::sync::Arc;

use picloud_shared::{HttpError, HttpRequest, HttpResponse, HttpService, SdkCallCx, Services};
use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use tokio::runtime::Handle as TokioHandle;

use super::bridge::{dynamic_to_json, json_to_dynamic};

/// Bridge-side defaults (the service clamps server-side too). The
/// `MAX_*` ceilings stay `i64` because they're compared against the
/// raw `i64` the script passed (so an over-limit value is rejected, not
/// truncated); the defaults are `u32` to match the `Opts` fields.
const DEFAULT_TIMEOUT_MS: u32 = 30_000;
const MAX_TIMEOUT_MS: i64 = 60_000;
const DEFAULT_MAX_REDIRECTS: u32 = 5;
const MAX_REDIRECTS: i64 = 10;

const ALLOWED_OPT_KEYS: [&str; 4] = ["headers", "timeout_ms", "follow_redirects", "max_redirects"];

pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
    let svc = services.http.clone();
    let mut module = Module::new();

    // Bodyless verbs: (url) / (url, opts).
    for verb in ["get", "head"] {
        register_bodyless(&mut module, verb, &svc, &cx);
    }
    // Body verbs: (url) / (url, body) / (url, body, opts).
    for verb in ["post", "put", "patch", "delete"] {
        register_body(&mut module, verb, &svc, &cx);
    }
    register_post_form(&mut module, &svc, &cx);
    register_request(&mut module, &svc, &cx);

    engine.register_static_module("http", module.into());
}

fn register_bodyless(
    module: &mut Module,
    verb: &'static str,
    svc: &Arc<dyn HttpService>,
    cx: &Arc<SdkCallCx>,
) {
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(verb, move |url: &str| {
            invoke(&svc, &cx, verb, url, None, None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(verb, move |url: &str, opts: Map| {
            invoke(&svc, &cx, verb, url, None, Some(&opts))
        });
    }
}

fn register_body(
    module: &mut Module,
    verb: &'static str,
    svc: &Arc<dyn HttpService>,
    cx: &Arc<SdkCallCx>,
) {
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(verb, move |url: &str| {
            invoke(&svc, &cx, verb, url, None, None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(verb, move |url: &str, body: Dynamic| {
            invoke(&svc, &cx, verb, url, Some(body), None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(verb, move |url: &str, body: Dynamic, opts: Map| {
            invoke(&svc, &cx, verb, url, Some(body), Some(&opts))
        });
    }
}

fn register_post_form(module: &mut Module, svc: &Arc<dyn HttpService>, cx: &Arc<SdkCallCx>) {
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn("post_form", move |url: &str, form: Map| {
            invoke_form(&svc, &cx, url, &form, None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn("post_form", move |url: &str, form: Map, opts: Map| {
            invoke_form(&svc, &cx, url, &form, Some(&opts))
        });
    }
}

fn register_request(module: &mut Module, svc: &Arc<dyn HttpService>, cx: &Arc<SdkCallCx>) {
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn("request", move |method: &str, url: &str| {
            invoke(&svc, &cx, method, url, None, None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn("request", move |method: &str, url: &str, body: Dynamic| {
            invoke(&svc, &cx, method, url, Some(body), None)
        });
    }
    {
        let (svc, cx) = (svc.clone(), cx.clone());
        module.set_native_fn(
            "request",
            move |method: &str, url: &str, body: Dynamic, opts: Map| {
                invoke(&svc, &cx, method, url, Some(body), Some(&opts))
            },
        );
    }
}

/// Parsed `opts` map.
struct Opts {
    headers: BTreeMap<String, String>,
    timeout_ms: u32,
    follow_redirects: bool,
    max_redirects: u32,
}

impl Default for Opts {
    fn default() -> Self {
        Self {
            headers: BTreeMap::new(),
            timeout_ms: DEFAULT_TIMEOUT_MS,
            follow_redirects: true,
            max_redirects: DEFAULT_MAX_REDIRECTS,
        }
    }
}

fn parse_opts(opts: Option<&Map>) -> Result<Opts, Box<EvalAltResult>> {
    let mut out = Opts::default();
    let Some(map) = opts else {
        return Ok(out);
    };
    for key in map.keys() {
        if !ALLOWED_OPT_KEYS.contains(&key.as_str()) {
            return Err(err(format!("unknown option key: {key}")));
        }
    }
    if let Some(h) = map.get("headers") {
        let hm = h
            .clone()
            .try_cast::<Map>()
            .ok_or_else(|| err("headers must be a map".to_string()))?;
        for (k, v) in hm {
            out.headers.insert(k.to_string(), dyn_to_string(&v));
        }
    }
    if let Some(t) = map.get("timeout_ms") {
        let ms = t
            .as_int()
            .map_err(|_| err("timeout_ms must be an integer".to_string()))?;
        if ms > MAX_TIMEOUT_MS {
            return Err(err(format!(
                "timeout_ms {ms} exceeds the {MAX_TIMEOUT_MS}ms maximum"
            )));
        }
        if ms > 0 {
            out.timeout_ms = u32::try_from(ms).unwrap_or(u32::MAX);
        }
    }
    if let Some(f) = map.get("follow_redirects") {
        out.follow_redirects = f
            .as_bool()
            .map_err(|_| err("follow_redirects must be a bool".to_string()))?;
    }
    if let Some(m) = map.get("max_redirects") {
        let n = m
            .as_int()
            .map_err(|_| err("max_redirects must be an integer".to_string()))?;
        if n > MAX_REDIRECTS {
            return Err(err(format!(
                "max_redirects {n} exceeds the {MAX_REDIRECTS} maximum"
            )));
        }
        out.max_redirects = u32::try_from(n.max(0)).unwrap_or(0);
    }
    Ok(out)
}

/// Encoded request body + the content-type chosen for it.
type EncodedBody = (Option<Vec<u8>>, Option<String>);

/// Dispatch a positional body by Rhai type. Returns the encoded bytes +
/// the chosen content-type. GET/HEAD callers pass `body = None`, so
/// this is never reached for them.
fn dispatch_body(body: Dynamic) -> Result<EncodedBody, Box<EvalAltResult>> {
    if body.is_unit() {
        return Ok((None, None));
    }
    if body.is_string() {
        let s = body.into_string().unwrap_or_default();
        return Ok((Some(s.into_bytes()), Some("text/plain".to_string())));
    }
    if body.is_map() || body.is_array() {
        let json = dynamic_to_json(&body);
        let bytes = serde_json::to_vec(&json)
            .map_err(|e| err(format!("could not encode JSON body: {e}")))?;
        return Ok((Some(bytes), Some("application/json".to_string())));
    }
    // Scalars (int/float/bool) → JSON-encode for consistency.
    let json = dynamic_to_json(&body);
    let bytes =
        serde_json::to_vec(&json).map_err(|e| err(format!("could not encode body: {e}")))?;
    Ok((Some(bytes), Some("application/json".to_string())))
}

#[allow(clippy::needless_pass_by_value)]
fn invoke(
    svc: &Arc<dyn HttpService>,
    cx: &Arc<SdkCallCx>,
    method: &str,
    url: &str,
    body: Option<Dynamic>,
    opts: Option<&Map>,
) -> Result<Dynamic, Box<EvalAltResult>> {
    let opts = parse_opts(opts)?;
    let method_uc = method.to_ascii_uppercase();
    let bodyless = matches!(method_uc.as_str(), "GET" | "HEAD");
    let (encoded, content_type) = if bodyless {
        (None, None)
    } else if let Some(b) = body {
        dispatch_body(b)?
    } else {
        (None, None)
    };

    let req = HttpRequest {
        method: method_uc,
        url: url.to_string(),
        headers: opts.headers,
        body: encoded,
        content_type,
        timeout_ms: opts.timeout_ms,
        follow_redirects: opts.follow_redirects,
        max_redirects: opts.max_redirects,
        script_id: Some(cx.script_id.to_string()),
    };
    let resp = block_on(svc, cx, req)?;
    Ok(response_to_dynamic(&resp))
}

#[allow(clippy::needless_pass_by_value)]
fn invoke_form(
    svc: &Arc<dyn HttpService>,
    cx: &Arc<SdkCallCx>,
    url: &str,
    form: &Map,
    opts: Option<&Map>,
) -> Result<Dynamic, Box<EvalAltResult>> {
    let opts = parse_opts(opts)?;
    let mut serializer = url::form_urlencoded::Serializer::new(String::new());
    for (k, v) in form {
        serializer.append_pair(k.as_str(), &dyn_to_string(v));
    }
    let encoded = serializer.finish();

    let req = HttpRequest {
        method: "POST".to_string(),
        url: url.to_string(),
        headers: opts.headers,
        body: Some(encoded.into_bytes()),
        content_type: Some("application/x-www-form-urlencoded".to_string()),
        timeout_ms: opts.timeout_ms,
        follow_redirects: opts.follow_redirects,
        max_redirects: opts.max_redirects,
        script_id: Some(cx.script_id.to_string()),
    };
    let resp = block_on(svc, cx, req)?;
    Ok(response_to_dynamic(&resp))
}

fn response_to_dynamic(resp: &HttpResponse) -> Dynamic {
    let mut m = Map::new();
    m.insert("status".into(), i64::from(resp.status).into());

    let mut headers = Map::new();
    let mut content_type = String::new();
    for (k, v) in &resp.headers {
        if k == "content-type" {
            content_type.clone_from(v);
        }
        headers.insert(k.clone().into(), v.clone().into());
    }
    m.insert("headers".into(), headers.into());

    // `body`: parsed JSON when the response is JSON and parses; () when
    // empty; otherwise the raw string.
    let body = if resp.body_raw.is_empty() {
        Dynamic::UNIT
    } else if content_type
        .to_ascii_lowercase()
        .starts_with("application/json")
    {
        match serde_json::from_str::<serde_json::Value>(&resp.body_raw) {
            Ok(json) => json_to_dynamic(json),
            Err(_) => resp.body_raw.clone().into(),
        }
    } else {
        resp.body_raw.clone().into()
    };
    m.insert("body".into(), body);
    m.insert("body_raw".into(), resp.body_raw.clone().into());
    m.into()
}

fn dyn_to_string(v: &Dynamic) -> String {
    if v.is_string() {
        v.clone().into_string().unwrap_or_default()
    } else {
        v.to_string()
    }
}

// Rhai's native-fn error channel is `Box<EvalAltResult>`, so these
// helpers return the boxed form the call sites need.
#[allow(clippy::unnecessary_box_returns)]
fn err(msg: String) -> Box<EvalAltResult> {
    EvalAltResult::ErrorRuntime(format!("http: {msg}").into(), rhai::Position::NONE).into()
}

/// Run the async service call from the synchronous Rhai context. Same
/// pattern as `kv`/`docs`: the script runs under `spawn_blocking`, so a
/// runtime handle is reachable and blocking on it is correct.
fn block_on(
    svc: &Arc<dyn HttpService>,
    cx: &Arc<SdkCallCx>,
    req: HttpRequest,
) -> Result<HttpResponse, Box<EvalAltResult>> {
    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
        EvalAltResult::ErrorRuntime(
            format!("http: no tokio runtime available: {e}").into(),
            rhai::Position::NONE,
        )
        .into()
    })?;
    let svc = svc.clone();
    let cx = cx.clone();
    handle
        .block_on(async move { svc.request(&cx, req).await })
        .map_err(map_http_err)
}

#[allow(clippy::unnecessary_box_returns)]
fn map_http_err(e: HttpError) -> Box<EvalAltResult> {
    EvalAltResult::ErrorRuntime(format!("http: {e}").into(), rhai::Position::NONE).into()
}