tenant.openagents/omega

1use anyhow::anyhow;
2use futures::AsyncReadExt as _;
3use http_client::{AsyncBody, HttpClient, RedirectPolicy};
4use std::future::Future;
5use std::pin::Pin;
6use std::task::Poll;
7use wasm_bindgen::JsCast as _;
8use wasm_bindgen::prelude::*;
9
10#[wasm_bindgen]
11extern "C" {
12    #[wasm_bindgen(catch, js_name = "fetch")]
13    fn global_fetch(input: &web_sys::Request) -> Result<js_sys::Promise, JsValue>;
14}
15
16pub struct FetchHttpClient {
17    user_agent: Option<http_client::http::header::HeaderValue>,
18    credentials: FetchCredentials,
19}
20
21/// Controls whether browser Fetch requests include credentials.
22#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
23pub enum FetchCredentials {
24    /// Never send credentials in the request or include credentials in the response.
25    Omit,
26    /// Only send and include credentials for same-origin requests. This is the default.
27    #[default]
28    SameOrigin,
29    /// Always include credentials, even for cross-origin requests.
30    Include,
31}
32
33impl Default for FetchHttpClient {
34    fn default() -> Self {
35        Self {
36            user_agent: None,
37            credentials: FetchCredentials::default(),
38        }
39    }
40}
41
42impl FetchHttpClient {
43    pub fn with_credentials(mut self, credentials: FetchCredentials) -> Self {
44        self.credentials = credentials;
45        self
46    }
47}
48
49#[cfg(feature = "multithreaded")]
50impl FetchHttpClient {
51    /// # Safety
52    ///
53    /// The caller must ensure that the created `FetchHttpClient` is only used in a single thread environment.
54    pub unsafe fn new() -> Self {
55        Self::default()
56    }
57
58    /// # Safety
59    ///
60    /// The caller must ensure that the created `FetchHttpClient` is only used in a single thread environment.
61    pub unsafe fn with_user_agent(user_agent: &str) -> anyhow::Result<Self> {
62        Ok(Self {
63            user_agent: Some(http_client::http::header::HeaderValue::from_str(
64                user_agent,
65            )?),
66            credentials: FetchCredentials::default(),
67        })
68    }
69}
70
71#[cfg(not(feature = "multithreaded"))]
72impl FetchHttpClient {
73    pub fn new() -> Self {
74        Self::default()
75    }
76
77    pub fn with_user_agent(user_agent: &str) -> anyhow::Result<Self> {
78        Ok(Self {
79            user_agent: Some(http_client::http::header::HeaderValue::from_str(
80                user_agent,
81            )?),
82            credentials: FetchCredentials::default(),
83        })
84    }
85}
86
87/// Wraps a `!Send` future to satisfy the `Send` bound on `BoxFuture`.
88///
89/// Safety: only valid in WASM contexts where the `FetchHttpClient` is
90/// confined to a single thread (guaranteed by the caller via unsafe
91/// constructors when `multithreaded` is enabled, or by the absence of
92/// threads when it is not).
93struct AssertSend<F>(F);
94
95unsafe impl<F> Send for AssertSend<F> {}
96
97impl<F: Future> Future for AssertSend<F> {
98    type Output = F::Output;
99
100    fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
101        // Safety: pin projection for a single-field newtype wrapper.
102        let inner = unsafe { self.map_unchecked_mut(|this| &mut this.0) };
103        inner.poll(cx)
104    }
105}
106
107impl HttpClient for FetchHttpClient {
108    fn user_agent(&self) -> Option<&http_client::http::header::HeaderValue> {
109        self.user_agent.as_ref()
110    }
111
112    fn proxy(&self) -> Option<&http_client::Url> {
113        None
114    }
115
116    fn send(
117        &self,
118        req: http_client::http::Request<AsyncBody>,
119    ) -> futures::future::BoxFuture<'static, anyhow::Result<http_client::http::Response<AsyncBody>>>
120    {
121        let (parts, body) = req.into_parts();
122        let credentials = self.credentials;
123
124        Box::pin(AssertSend(async move {
125            let body_bytes = read_body_to_bytes(body).await?;
126
127            let init = web_sys::RequestInit::new();
128            init.set_method(parts.method.as_str());
129            init.set_credentials(match credentials {
130                FetchCredentials::Omit => web_sys::RequestCredentials::Omit,
131                FetchCredentials::SameOrigin => web_sys::RequestCredentials::SameOrigin,
132                FetchCredentials::Include => web_sys::RequestCredentials::Include,
133            });
134
135            if let Some(redirect_policy) = parts.extensions.get::<RedirectPolicy>() {
136                match redirect_policy {
137                    RedirectPolicy::NoFollow => {
138                        init.set_redirect(web_sys::RequestRedirect::Manual);
139                    }
140                    RedirectPolicy::FollowLimit(_) | RedirectPolicy::FollowAll => {
141                        init.set_redirect(web_sys::RequestRedirect::Follow);
142                    }
143                }
144            }
145
146            if let Some(ref bytes) = body_bytes {
147                let uint8array = js_sys::Uint8Array::from(bytes.as_slice());
148                init.set_body(uint8array.as_ref());
149            }
150
151            let url = parts.uri.to_string();
152            let request = web_sys::Request::new_with_str_and_init(&url, &init)
153                .map_err(|error| anyhow!("failed to create fetch Request: {error:?}"))?;
154
155            let request_headers = request.headers();
156            for (name, value) in &parts.headers {
157                let value_str = value
158                    .to_str()
159                    .map_err(|_| anyhow!("non-ASCII header value for {name}"))?;
160                request_headers
161                    .set(name.as_str(), value_str)
162                    .map_err(|error| anyhow!("failed to set header {name}: {error:?}"))?;
163            }
164
165            let promise = global_fetch(&request)
166                .map_err(|error| anyhow!("fetch threw an error: {error:?}"))?;
167            let response_value = wasm_bindgen_futures::JsFuture::from(promise)
168                .await
169                .map_err(|error| anyhow!("fetch failed: {error:?}"))?;
170
171            let web_response: web_sys::Response = response_value
172                .dyn_into()
173                .map_err(|error| anyhow!("fetch result is not a Response: {error:?}"))?;
174
175            let status = web_response.status();
176            let mut builder = http_client::http::Response::builder().status(status);
177
178            // `Headers` is a JS iterable yielding `[name, value]` pairs.
179            // `js_sys::Array::from` calls `Array.from()` which accepts any iterable.
180            let header_pairs = js_sys::Array::from(&web_response.headers());
181            for index in 0..header_pairs.length() {
182                match header_pairs.get(index).dyn_into::<js_sys::Array>() {
183                    Ok(pair) => match (pair.get(0).as_string(), pair.get(1).as_string()) {
184                        (Some(name), Some(value)) => {
185                            builder = builder.header(name, value);
186                        }
187                        (name, value) => {
188                            log::warn!(
189                                "skipping response header at index {index}: \
190                                     name={name:?}, value={value:?}"
191                            );
192                        }
193                    },
194                    Err(entry) => {
195                        log::warn!("skipping non-array header entry at index {index}: {entry:?}");
196                    }
197                }
198            }
199
200            // The entire response body is eagerly buffered into memory via
201            // `arrayBuffer()`. The Fetch API does not expose a synchronous
202            // streaming interface; streaming would require `ReadableStream`
203            // interop which is significantly more complex.
204            let body_promise = web_response
205                .array_buffer()
206                .map_err(|error| anyhow!("failed to initiate response body read: {error:?}"))?;
207            let body_value = wasm_bindgen_futures::JsFuture::from(body_promise)
208                .await
209                .map_err(|error| anyhow!("failed to read response body: {error:?}"))?;
210            let array_buffer: js_sys::ArrayBuffer = body_value
211                .dyn_into()
212                .map_err(|error| anyhow!("response body is not an ArrayBuffer: {error:?}"))?;
213            let response_bytes = js_sys::Uint8Array::new(&array_buffer).to_vec();
214
215            builder
216                .body(AsyncBody::from(response_bytes))
217                .map_err(|error| anyhow!(error))
218        }))
219    }
220}
221
222async fn read_body_to_bytes(mut body: AsyncBody) -> anyhow::Result<Option<Vec<u8>>> {
223    let mut buffer = Vec::new();
224    body.read_to_end(&mut buffer).await?;
225    if buffer.is_empty() {
226        Ok(None)
227    } else {
228        Ok(Some(buffer))
229    }
230}
231