orbit_runtime/
http_adapter.rs

1//! Built-in `orbit.http.request` task: lets a workflow call an HTTP endpoint
2//! without writing a worker. Disabled unless the server is started with
3//! `--enable-http-adapter`, because it lets workflow authors make outbound
4//! requests from the runtime host.
5
6use crate::model::{codes, TaskError};
7use serde_json::{json, Map, Value};
8use std::time::Duration;
9
10pub fn execute(input: &Value, timeout_ms: Option<u64>) -> Result<Value, TaskError> {
11    let method = input
12        .get("method")
13        .and_then(Value::as_str)
14        .unwrap_or("GET")
15        .to_uppercase();
16    let url = input
17        .get("url")
18        .and_then(Value::as_str)
19        .ok_or_else(|| TaskError::new(codes::HTTP_ERROR, "missing url", false))?;
20    if !(url.starts_with("http://") || url.starts_with("https://")) {
21        return Err(TaskError::new(
22            codes::HTTP_ERROR,
23            "only http:// and https:// URLs are allowed",
24            false,
25        ));
26    }
27    let config = ureq::Agent::config_builder()
28        .http_status_as_error(false)
29        .timeout_global(Some(Duration::from_millis(timeout_ms.unwrap_or(30_000))))
30        .build();
31    let agent: ureq::Agent = config.into();
32    let headers: Vec<(String, String)> = input
33        .get("headers")
34        .and_then(Value::as_object)
35        .map(|h| {
36            h.iter()
37                .filter_map(|(k, v)| v.as_str().map(|v| (k.clone(), v.to_string())))
38                .collect()
39        })
40        .unwrap_or_default();
41    let body = input.get("body").cloned().unwrap_or(Value::Null);
42
43    let result = match method.as_str() {
44        "GET" | "DELETE" | "HEAD" | "OPTIONS" => {
45            let mut req = match method.as_str() {
46                "GET" => agent.get(url),
47                "DELETE" => agent.delete(url),
48                "HEAD" => agent.head(url),
49                _ => agent.options(url),
50            };
51            for (k, v) in &headers {
52                req = req.header(k, v);
53            }
54            req.call()
55        }
56        "POST" | "PUT" | "PATCH" => {
57            let mut req = match method.as_str() {
58                "POST" => agent.post(url),
59                "PUT" => agent.put(url),
60                _ => agent.patch(url),
61            };
62            for (k, v) in &headers {
63                req = req.header(k, v);
64            }
65            match &body {
66                Value::Null => req.send_empty(),
67                Value::String(s) => req.send(s.as_str()),
68                other => req.send_json(other),
69            }
70        }
71        other => {
72            return Err(TaskError::new(
73                codes::HTTP_ERROR,
74                format!("unsupported method {other}"),
75                false,
76            ))
77        }
78    };
79
80    let mut resp = result
81        .map_err(|e| TaskError::new(codes::HTTP_ERROR, format!("request failed: {e}"), true))?;
82    let status = resp.status().as_u16();
83    let mut out_headers = Map::new();
84    for (k, v) in resp.headers() {
85        if let Ok(v) = v.to_str() {
86            out_headers.insert(k.as_str().to_string(), Value::String(v.to_string()));
87        }
88    }
89    let text = resp
90        .body_mut()
91        .with_config()
92        .limit(8 * 1024 * 1024)
93        .read_to_string()
94        .unwrap_or_default();
95    let body = serde_json::from_str::<Value>(&text).unwrap_or(Value::String(text));
96    if status >= 500 || status == 429 {
97        let mut e = TaskError::new(
98            codes::HTTP_ERROR,
99            format!("server responded {status}"),
100            true,
101        );
102        e.details = Some(json!({ "status": status, "body": body }));
103        return Err(e);
104    }
105    Ok(json!({ "status": status, "headers": out_headers, "body": body }))
106}
orbit_runtime/http_adapter.rs

orbit_runtime/
http_adapter.rs
