run-rs 0.6.33

//! The reqwest bridge, blocking and async. Only `.send()`, `.text()` and `.json()` yield futures.

use std::sync::{Arc, OnceLock};
use std::time::Duration;

use anyhow::{Result, anyhow, bail};
use base64::Engine;
use reqwest::{Client, Method};

use super::bridge::arg;
use super::bytecode::{BuiltinId, MethodName, PathId};
use super::json_bridge::{json_to_pvalue, parse_json, pvalue_to_json};
use super::native::Native;
use super::std_bridge::duration_from_value;
use super::value::{StructData, Value};
use headers::{empty_map, header_map_method, header_value_call, header_value_method};

mod headers;

/// So many one off gets reuse 1 connection pool.
fn default_client() -> Client {
    static C: OnceLock<Client> = OnceLock::new();
    C.get_or_init(Client::new).clone()
}

fn client_value(c: Client) -> Value {
    Native::HttpClient(c).wrap()
}

// the blocking client

fn build_blocking_client(
    cookie_store: bool,
    timeout: Option<Duration>,
    ua: Option<String>,
    redirect: Option<reqwest::redirect::Policy>,
) -> Result<reqwest::blocking::Client> {
    let mut b = reqwest::blocking::Client::builder().cookie_store(cookie_store);
    if let Some(d) = timeout {
        b = b.timeout(d);
    }
    if let Some(ua) = ua {
        b = b.user_agent(ua);
    }
    if let Some(policy) = redirect {
        b = b.redirect(policy);
    }
    b.build()
        .map_err(|e| anyhow!("http client build failed: {e}"))
}

/// So many one off gets don't spin up a runtime thread per call. Safe because script code always
/// runs on blocking threads.
fn default_blocking_client() -> Result<reqwest::blocking::Client> {
    static C: OnceLock<reqwest::blocking::Client> = OnceLock::new();
    if let Some(c) = C.get() {
        return Ok(c.clone());
    }
    let c = build_blocking_client(false, None, None, None)?;
    if C.set(c.clone()).is_err() {
        return C
            .get()
            .cloned()
            .ok_or_else(|| anyhow!("shared HTTP client was not initialized"));
    }
    Ok(c)
}

fn blocking_client_value(c: reqwest::blocking::Client) -> Value {
    Native::BlockingHttpClient(c).wrap()
}

// dispatch of `reqwest::..` path calls

/// Both the blocking and the async API.
pub(super) fn reqwest_call(id: PathId, args: &[Value]) -> Result<Value> {
    Ok(match id {
        // a redirect policy marker
        PathId::RedirectPolicyNone => {
            Value::struct_of("RedirectPolicy", [("kind".into(), Value::str("none"))])
        }
        PathId::RedirectPolicyLimited => Value::struct_of(
            "RedirectPolicy",
            [
                ("kind".into(), Value::str("limited")),
                ("n".into(), args.first().cloned().unwrap_or(Value::Int(10))),
            ],
        ),
        PathId::HeaderMapNew => empty_map(),
        PathId::HeaderValueFromStatic | PathId::HeaderValueFromStr => header_value_call(id, args)?,
        PathId::ReqwestBlockingClientNew => {
            blocking_client_value(build_blocking_client(false, None, None, None)?)
        }
        PathId::ReqwestClientNew => client_value(Client::new()),
        PathId::ReqwestBlockingClientBuilder => blocking_builder_value(),
        PathId::ReqwestClientBuilder => builder_value(),
        PathId::ReqwestBlockingGet => {
            let url = args.first().map(Value::display).unwrap_or_default();
            run_blocking(&request_struct("GET", &url, Value::Unit))
        }
        PathId::ReqwestGet => {
            let url = args.first().map(Value::display).unwrap_or_default();
            send_future(&request_struct("GET", &url, Value::Unit))
        }
        _ => bail!("unsupported reqwest function `{id}`, build a Client for other verbs"),
    })
}

// request and builder values

fn request_struct(method: &str, url: &str, client: Value) -> Arc<StructData> {
    let v = Value::struct_of(
        "ReqwestRequest",
        [
            ("method".into(), Value::str(method)),
            ("url".into(), Value::str(url)),
            ("headers".into(), Value::vec(vec![])),
            ("query".into(), Value::vec(vec![])),
            ("body".into(), Value::Unit),
            ("timeout".into(), Value::Unit),
            ("client".into(), client),
        ],
    );
    match v {
        Value::Struct(s) => s,
        _ => unreachable!(),
    }
}

fn builder_value() -> Value {
    Value::struct_of(
        "ReqwestClientBuilder",
        [
            ("cookie_store".into(), Value::Bool(false)),
            ("timeout".into(), Value::Unit),
            ("user_agent".into(), Value::Unit),
            ("redirect".into(), Value::Unit),
            ("blocking".into(), Value::Bool(false)),
        ],
    )
}

fn blocking_builder_value() -> Value {
    Value::struct_of(
        "ReqwestClientBuilder",
        [
            ("cookie_store".into(), Value::Bool(false)),
            ("timeout".into(), Value::Unit),
            ("user_agent".into(), Value::Unit),
            ("redirect".into(), Value::Unit),
            ("blocking".into(), Value::Bool(true)),
        ],
    )
}

// the redirect policy a builder stashed via `.redirect(..)`
fn redirect_policy(s: &StructData) -> Option<reqwest::redirect::Policy> {
    let Some(Value::Struct(rp)) = s.get("redirect") else {
        return None;
    };
    if &**rp.name() != "RedirectPolicy" {
        return None;
    }
    match rp.get("kind").map(|v| v.display()).as_deref() {
        Some("none") => Some(reqwest::redirect::Policy::none()),
        Some("limited") => {
            let n = match rp.get("n") {
                Some(Value::Int(n)) => usize::try_from(n).unwrap_or(10),
                _ => 10,
            };
            Some(reqwest::redirect::Policy::limited(n))
        }
        _ => None,
    }
}

// method dispatch

/// `None` when the receiver is not an http type.
pub(super) fn http_method(
    recv: &Value,
    method: &MethodName,
    args: &[Value],
) -> Option<Result<Value>> {
    match recv {
        Value::Native(n)
            if matches!(
                &*n.lock(),
                Native::HttpClient(_) | Native::BlockingHttpClient(_)
            ) =>
        {
            Some(client_method(n, method, args))
        }
        Value::Struct(s) => match &**s.name() {
            "ReqwestClientBuilder" => Some(builder_method(s, method, args)),
            "ReqwestRequest" => Some(request_method(s, method, args)),
            "ReqwestResponse" => Some(response_method(s, method)),
            "StatusCode" => Some(Ok(status_method(s, method))),
            "HeaderMap" => Some(header_map_method(s, method, args)),
            "HeaderValue" => Some(header_value_method(s, method)),
            _ => None,
        },
        _ => None,
    }
}

fn client_method(
    n: &Arc<parking_lot::Mutex<Native>>,
    method: &MethodName,
    args: &[Value],
) -> Result<Value> {
    let verb = match method.id {
        BuiltinId::Get => "GET",
        BuiltinId::Post => "POST",
        BuiltinId::Put => "PUT",
        BuiltinId::Delete => "DELETE",
        BuiltinId::Patch => "PATCH",
        BuiltinId::Head => "HEAD",
        BuiltinId::Clone => return Ok(Value::Native(n.clone())),
        _ => bail!("unknown method `{}` on a client", method.text),
    };
    let url = args.first().map(Value::display).unwrap_or_default();
    Ok(Value::Struct(request_struct(
        verb,
        &url,
        Value::Native(n.clone()),
    )))
}

fn builder_method(s: &Arc<StructData>, method: &MethodName, args: &[Value]) -> Result<Value> {
    let this = || Value::Struct(s.clone());
    match method.id {
        BuiltinId::CookieStore => {
            s.set(
                "cookie_store",
                args.first().cloned().unwrap_or(Value::Bool(false)),
            );
            Ok(this())
        }
        BuiltinId::Timeout => {
            s.set("timeout", arg(args, 0)?);
            Ok(this())
        }
        BuiltinId::UserAgent => {
            s.set("user_agent", arg(args, 0)?);
            Ok(this())
        }
        BuiltinId::Redirect => {
            s.set("redirect", arg(args, 0)?);
            Ok(this())
        }
        BuiltinId::Build => {
            let cookies = matches!(s.get("cookie_store"), Some(Value::Bool(true)));
            let timeout = duration_field(s, "timeout");
            let ua = match s.get("user_agent") {
                Some(Value::Str(u)) => Some(u.to_string()),
                _ => None,
            };
            if matches!(s.get("blocking"), Some(Value::Bool(true))) {
                return Ok(
                    match build_blocking_client(cookies, timeout, ua, redirect_policy(s)) {
                        Ok(c) => Value::ok(blocking_client_value(c)),
                        Err(e) => Value::err(Value::str(e.to_string())),
                    },
                );
            }
            let mut b = Client::builder().cookie_store(cookies);
            if let Some(d) = timeout {
                b = b.timeout(d);
            }
            if let Some(ua) = ua {
                b = b.user_agent(ua);
            }
            if let Some(policy) = redirect_policy(s) {
                b = b.redirect(policy);
            }
            Ok(match b.build() {
                Ok(c) => Value::ok(client_value(c)),
                Err(e) => Value::err(Value::str(e.to_string())),
            })
        }
        _ => bail!("unknown method `{}` on a client builder", method.text),
    }
}

fn request_method(s: &Arc<StructData>, method: &MethodName, args: &[Value]) -> Result<Value> {
    let this = || Value::Struct(s.clone());
    match method.id {
        BuiltinId::Header => {
            let k = args.first().map(Value::display).unwrap_or_default();
            let v = args.get(1).map(Value::display).unwrap_or_default();
            add_header(s, &k, &v);
            Ok(this())
        }
        BuiltinId::BearerAuth => {
            let token = args.first().map(Value::display).unwrap_or_default();
            add_header(s, "Authorization", &format!("Bearer {token}"));
            Ok(this())
        }
        BuiltinId::BasicAuth => {
            let user = args.first().map(Value::display).unwrap_or_default();
            let pass = match args.get(1) {
                Some(Value::Enum { data, .. }) => {
                    data.lock().first().map(Value::display).unwrap_or_default()
                }
                Some(other) => other.display(),
                None => String::new(),
            };
            let token = base64::engine::general_purpose::STANDARD.encode(format!("{user}:{pass}"));
            add_header(s, "Authorization", &format!("Basic {token}"));
            Ok(this())
        }
        BuiltinId::Query => {
            if let Some(Value::Vec(items)) = args.first()
                && let Some(Value::Vec(q)) = s.get("query")
            {
                for item in items.lock().iter() {
                    q.lock().push(item.clone());
                }
            }
            Ok(this())
        }
        BuiltinId::Json => {
            let json = pvalue_to_json(args.first().unwrap_or(&Value::Unit))?;
            add_header(s, "Content-Type", "application/json");
            s.set("body", Value::str(serde_json::to_string(&json)?));
            Ok(this())
        }
        BuiltinId::Body => {
            s.set(
                "body",
                Value::str(args.first().map(Value::display).unwrap_or_default()),
            );
            Ok(this())
        }
        BuiltinId::Timeout => {
            s.set("timeout", arg(args, 0)?);
            Ok(this())
        }
        // a blocking client runs at once, an async one hands back a future
        BuiltinId::Send => {
            let blocking = matches!(
                s.get("client"),
                Some(Value::Native(n)) if matches!(&*n.lock(), Native::BlockingHttpClient(_))
            );
            if blocking {
                Ok(run_blocking(s))
            } else {
                Ok(send_future(s))
            }
        }
        _ => bail!("unknown method `{}` on a request", method.text),
    }
}

// blocking execution

fn run_blocking(s: &StructData) -> Value {
    match execute_blocking(s) {
        Ok(v) => Value::ok(v),
        Err(e) => Value::err(Value::str(e.to_string())),
    }
}

fn execute_blocking(s: &StructData) -> Result<Value> {
    let method = s
        .get("method")
        .map_or_else(|| "GET".into(), |v| v.display());
    let url = s.get("url").map(|v| v.display()).unwrap_or_default();
    let client = match s.get("client") {
        Some(Value::Native(h)) => match &*h.lock() {
            Native::BlockingHttpClient(c) => c.clone(),
            _ => default_blocking_client()?,
        },
        _ => default_blocking_client()?,
    };
    let m = Method::from_bytes(method.as_bytes()).unwrap_or(Method::GET);
    let mut rb = client.request(m, &url);
    let query = pairs_field(s, "query");
    if !query.is_empty() {
        rb = rb.query(&query);
    }
    for (k, v) in pairs_field(s, "headers") {
        rb = rb.header(&k, &v);
    }
    if let Some(d) = duration_field(s, "timeout") {
        rb = rb.timeout(d);
    }
    if let Some(Value::Str(body)) = s.get("body") {
        rb = rb.body(body.to_string());
    }
    let resp = rb.send()?;
    let status = resp.status().as_u16();
    let headers: Vec<(String, String)> = resp
        .headers()
        .iter()
        .map(|(k, v)| (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();
    // taken before the body is read, which consumes the response
    let length = resp.content_length();
    let text = resp.text()?;
    Ok(Value::struct_of(
        "ReqwestResponse",
        [
            ("status".into(), Value::Int(i64::from(status))),
            // a blocking body is already decoded, so `text` and `json` return directly
            ("body".into(), Value::str(text)),
            ("headers".into(), header_pairs(headers)),
            (
                "content_length".into(),
                match length {
                    Some(n) => Value::some(Value::Int(i64::try_from(n).unwrap_or(0))),
                    None => Value::none(),
                },
            ),
        ],
    ))
}

fn add_header(s: &StructData, k: &str, v: &str) {
    if let Some(Value::Vec(h)) = s.get("headers") {
        h.lock()
            .push(Value::tuple(vec![Value::str(k), Value::str(v)]));
    }
}

// execution

/// Free of any `!Send` value.
struct Plan {
    method: Method,
    url: String,
    headers: Vec<(String, String)>,
    query: Vec<(String, String)>,
    body: Option<String>,
    timeout: Option<Duration>,
    client: Client,
}

fn build_plan(s: &StructData) -> Plan {
    let method = s
        .get("method")
        .map_or_else(|| "GET".into(), |v| v.display());
    let client = match s.get("client") {
        Some(Value::Native(n)) => match &*n.lock() {
            Native::HttpClient(c) => c.clone(),
            _ => default_client(),
        },
        _ => default_client(),
    };
    Plan {
        method: Method::from_bytes(method.as_bytes()).unwrap_or(Method::GET),
        url: s.get("url").map(|v| v.display()).unwrap_or_default(),
        headers: pairs_field(s, "headers"),
        query: pairs_field(s, "query"),
        body: match s.get("body") {
            Some(Value::Str(b)) => Some(b.to_string()),
            _ => None,
        },
        timeout: duration_field(s, "timeout"),
        client,
    }
}

fn send_future(s: &StructData) -> Value {
    let plan = build_plan(s);
    Native::Future(Box::pin(async move {
        match run_plan(plan).await {
            Ok(resp) => Value::ok(resp),
            Err(e) => Value::err(Value::str(e.to_string())),
        }
    }))
    .wrap()
}

async fn run_plan(plan: Plan) -> Result<Value> {
    let mut rb = plan.client.request(plan.method, &plan.url);
    if !plan.query.is_empty() {
        rb = rb.query(&plan.query);
    }
    for (k, v) in &plan.headers {
        rb = rb.header(k, v);
    }
    if let Some(d) = plan.timeout {
        rb = rb.timeout(d);
    }
    if let Some(body) = plan.body {
        rb = rb.body(body);
    }
    let resp = rb.send().await?;
    let status = resp.status().as_u16();
    let headers: Vec<(String, String)> = resp
        .headers()
        .iter()
        .map(|(k, v)| (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()))
        .collect();
    // taken before the body is read, which consumes the response, a speed test counts these and
    // not the decoded text
    let length = resp.content_length();
    // kept in wire form, so `content_length` never pays for a UTF-8 conversion
    let raw = resp.bytes().await?.to_vec();
    Ok(Value::struct_of(
        "ReqwestResponse",
        [
            ("status".into(), Value::Int(i64::from(status))),
            ("body".into(), Native::Body(raw).wrap()),
            ("headers".into(), header_pairs(headers)),
            (
                "content_length".into(),
                match length {
                    Some(n) => Value::some(Value::Int(i64::try_from(n).unwrap_or(0))),
                    None => Value::none(),
                },
            ),
        ],
    ))
}

fn pairs_field(s: &StructData, field: &str) -> Vec<(String, String)> {
    match s.get(field) {
        Some(Value::Vec(items)) => items
            .lock()
            .iter()
            .filter_map(|item| {
                let Value::Tuple(pair) = item else {
                    return None;
                };
                let pair = pair.lock();
                Some((pair[0].display(), pair[1].display()))
            })
            .collect(),
        _ => Vec::new(),
    }
}

fn header_pairs(pairs: Vec<(String, String)>) -> Value {
    Value::vec(
        pairs
            .into_iter()
            .map(|(k, v)| Value::tuple(vec![Value::str(k), Value::str(v)]))
            .collect(),
    )
}

fn duration_field(s: &StructData, field: &str) -> Option<Duration> {
    let v = s.get(field)?;
    if let Value::Enum { data, .. } = &v {
        return data.lock().first().and_then(duration_from_value);
    }
    duration_from_value(&v)
}

// response methods

fn response_method(s: &Arc<StructData>, method: &MethodName) -> Result<Value> {
    let this = || Value::Struct(s.clone());
    let body = || body_bytes(s);
    // a blocking response holds decoded text, an async one wire bytes
    let is_blocking = matches!(s.get("body"), Some(Value::Str(_)));
    Ok(match method.id {
        BuiltinId::Status => Value::struct_of(
            "StatusCode",
            [("code".into(), s.get("status").unwrap_or(Value::Int(0)))],
        ),
        BuiltinId::Text if is_blocking => {
            Value::ok(Value::str(String::from_utf8_lossy(&body()).into_owned()))
        }
        BuiltinId::Json if is_blocking => {
            let text = String::from_utf8_lossy(&body()).into_owned();
            match parse_json(&text) {
                Ok(v) => Value::ok(v),
                Err(e) => Value::err(Value::str(e.to_string())),
            }
        }
        BuiltinId::Text => text_future(body()),
        BuiltinId::Json => json_future(body()),
        BuiltinId::ContentLength => s.get("content_length").unwrap_or_else(Value::none),
        BuiltinId::Headers => Value::struct_of(
            "HeaderMap",
            [(
                "map".into(),
                s.get("headers").unwrap_or_else(|| Value::vec(vec![])),
            )],
        ),
        BuiltinId::ErrorForStatus => {
            let code = match s.get("status") {
                Some(Value::Int(c)) => c,
                _ => 0,
            };
            if (200..400).contains(&code) {
                Value::ok(this())
            } else {
                Value::err(Value::str(format!("HTTP status {code}")))
            }
        }
        _ => bail!("unknown method `{}` on a response", method.text),
    })
}

fn body_bytes(s: &StructData) -> Vec<u8> {
    match s.get("body") {
        Some(Value::Native(n)) => match &*n.lock() {
            Native::Body(raw) => raw.clone(),
            _ => Vec::new(),
        },
        Some(other) => other.display().into_bytes(),
        None => Vec::new(),
    }
}

fn text_future(body: Vec<u8>) -> Value {
    Native::Future(Box::pin(async move {
        Value::ok(Value::str(String::from_utf8_lossy(&body).into_owned()))
    }))
    .wrap()
}

fn json_future(body: Vec<u8>) -> Value {
    Native::Future(Box::pin(async move {
        match serde_json::from_slice::<serde_json::Value>(&body) {
            Ok(v) => Value::ok(json_to_pvalue(v)),
            Err(e) => Value::err(Value::str(e.to_string())),
        }
    }))
    .wrap()
}

fn status_method(s: &StructData, method: &MethodName) -> Value {
    let code = match s.get("code") {
        Some(Value::Int(c)) => c,
        _ => 0,
    };
    match super::shared::status_core(method.id, code) {
        Some(super::shared::StatusOut::Int(i)) => Value::Int(i),
        Some(super::shared::StatusOut::Bool(b)) => Value::Bool(b),
        None => Value::Unit,
    }
}

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

    /// A json null from a response must be `Value::Null`, not Unit, otherwise a match depends on
    /// where the null came from.
    #[test]
    fn a_response_null_is_the_none_the_parsers_make() {
        assert!(json_to_pvalue(serde_json::Value::Null).is_none_value());
    }
}