zenwave 0.5.3

use core::{
    fmt,
    future::Future,
    ops::Deref,
    pin::Pin,
    task::{Context, Poll},
};

use http_kit::{
    BodyError, Endpoint, HttpError, StatusCode,
    utils::{Stream, StreamExt},
};
use std::error::Error as StdError;
use std::io;
use wasm_bindgen_futures::JsFuture;
use web_sys::wasm_bindgen::{JsCast, JsValue};

use crate::{Client, error::HttpErrorResponse};
/// HTTP client backend for browser environments using `fetch`.
pub struct WebBackend {
    scope: SingleThreaded<FetchScope>,
}

/// `globalThis` and the `fetch` it owns.
///
/// A page has a `Window`, a web worker a `WorkerGlobalScope`, and a
/// Cloudflare Worker a global whose constructor is not exposed at all, so
/// no `instanceof` test names every runtime. What every one of them has is
/// a `fetch` property on `globalThis`; the backend takes that function once
/// and calls it with `globalThis` as `this`, which is how the platform
/// itself dispatches `fetch(request)`.
struct FetchScope {
    global: JsValue,
    fetch: web_sys::js_sys::Function,
}

impl FetchScope {
    /// The scope this code is running in, or `None` when `globalThis` has
    /// no `fetch` at all (a bare JS shell, for instance).
    fn current() -> Option<Self> {
        let global: JsValue = web_sys::js_sys::global().into();
        let fetch = web_sys::js_sys::Reflect::get(&global, &JsValue::from_str("fetch")).ok()?;
        let fetch = fetch.dyn_into::<web_sys::js_sys::Function>().ok()?;
        Some(Self { global, fetch })
    }

    fn fetch_with_request(
        &self,
        request: &web_sys::Request,
    ) -> Result<web_sys::js_sys::Promise, JsValue> {
        self.fetch
            .call1(&self.global, request)
            .map(web_sys::js_sys::Promise::from)
    }
}

#[derive(Debug, thiserror::Error)]
pub enum WebError {
    #[error("{source}")]
    Transport {
        #[source]
        source: Box<dyn StdError + Send + Sync>,
        status: StatusCode,
    },
    #[error("remote error: {status}")]
    Remote {
        status: StatusCode,
        body: Option<String>,
        raw_response: http_kit::Response,
    },
}

impl WebError {
    fn new(status: StatusCode, error: impl StdError + Send + Sync + 'static) -> Self {
        Self::Transport {
            source: Box::new(error),
            status,
        }
    }

    fn remote(status: StatusCode, body: Option<String>, raw_response: http_kit::Response) -> Self {
        Self::Remote {
            status,
            body,
            raw_response,
        }
    }
}

impl HttpError for WebError {
    fn status(&self) -> StatusCode {
        match self {
            Self::Transport { status, .. } => *status,
            Self::Remote { status, .. } => *status,
        }
    }
}

// Convert WebError to unified zenwave::Error
impl From<WebError> for crate::Error {
    fn from(err: WebError) -> Self {
        match err {
            WebError::Transport { source, .. } => crate::Error::Transport(source),
            WebError::Remote {
                status,
                body,
                raw_response,
            } => crate::Error::Http {
                status,
                message: body.clone().unwrap_or_else(|| {
                    status
                        .canonical_reason()
                        .unwrap_or("Unknown error")
                        .to_string()
                }),
                response: Box::new(HttpErrorResponse {
                    response: raw_response,
                    body_text: body,
                }),
            },
        }
    }
}

impl fmt::Debug for WebBackend {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WebBackend").finish()
    }
}

// Browser is not multi-threaded, so we can safely implement `Send` and `Sync`
// since the WebBackend will only be used on the main thread
struct SingleThreaded<T>(pub T);

impl<T: Stream> Stream for SingleThreaded<T> {
    type Item = T::Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // SAFETY: SingleThreaded<T> is a newtype wrapper, and we do not move T out.
        let this = unsafe { self.get_unchecked_mut() };
        unsafe { Pin::new_unchecked(&mut this.0).poll_next(cx) }
    }
}

unsafe impl<T> Send for SingleThreaded<T> {}
unsafe impl<T> Sync for SingleThreaded<T> {}

impl<T> Deref for SingleThreaded<T> {
    type Target = T;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<T: Future> Future for SingleThreaded<T> {
    type Output = T::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: SingleThreaded<T> is a newtype wrapper, and we do not move T out.
        let this = unsafe { self.get_unchecked_mut() };
        unsafe { Pin::new_unchecked(&mut this.0).poll(cx) }
    }
}

impl WebBackend {
    /// Construct a new `WebBackend` bound to the global scope's `fetch`.
    ///
    /// # Panics
    ///
    /// Panics when `globalThis` is neither a `Window` nor a
    /// `WorkerGlobalScope`, because there is then no `fetch` to speak through.
    pub fn new() -> Self {
        let scope = FetchScope::current().expect(
            "globalThis is neither a Window nor a WorkerGlobalScope, so there is no `fetch`",
        );

        Self {
            scope: SingleThreaded(scope),
        }
    }
}

impl Default for WebBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl Endpoint for WebBackend {
    type Error = crate::Error;
    async fn respond(
        &mut self,
        request: &mut http_kit::Request,
    ) -> Result<http_kit::Response, Self::Error> {
        fetch(&self.scope, request).await.map_err(Into::into)
    }
}

fn fetch(
    scope: &FetchScope,
    request: &mut http_kit::Request,
) -> impl Future<Output = Result<http_kit::Response, WebError>> + Send {
    SingleThreaded(async move {
        let request_init = web_sys::RequestInit::new();
        request_init.set_method(request.method().as_str());
        let headers = web_sys::Headers::new().unwrap();
        // The request body is handed to `fetch` as bytes, never as a
        // `ReadableStream`. A streamed request body is a Chromium-only
        // feature that also needs HTTP/2 and `duplex: "half"`; Firefox and
        // Safari reject it, and a Cloudflare Worker never pulls a stream
        // driven from Rust, so the request sat pending until the runtime
        // cancelled the invocation. Buffering costs one copy of the body and
        // works everywhere `fetch` does.
        let body = std::mem::replace(request.body_mut(), http_kit::Body::empty());
        let bytes = body
            .into_bytes()
            .await
            .map_err(|e| WebError::new(StatusCode::BAD_REQUEST, e))?;
        if !bytes.is_empty() {
            let body_value: JsValue = js_sys::Uint8Array::from(bytes.as_ref()).into();
            request_init.set_body(&body_value);
        }

        for (name, value) in request.headers().iter() {
            let value = value
                .to_str()
                .map_err(|e| WebError::new(StatusCode::BAD_REQUEST, e))?;
            headers.set(name.as_str(), value).map_err(|err| {
                WebError::new(
                    StatusCode::BAD_REQUEST,
                    transport_error(format_js_value(&err)),
                )
            })?;
        }
        request_init.set_headers(headers.as_ref());

        let uri = request.uri().to_string();
        let fetch_request = web_sys::Request::new_with_str_and_init(uri.as_str(), &request_init)
            .map_err(|err| {
                WebError::new(
                    StatusCode::BAD_REQUEST,
                    transport_error(format_js_value(&err)),
                )
            })?;

        let promise = scope.fetch_with_request(&fetch_request).map_err(|err| {
            WebError::new(
                StatusCode::BAD_GATEWAY,
                transport_error(format_js_value(&err)),
            )
        })?;
        let fut = SingleThreaded(JsFuture::from(promise));
        let response = fut.await.map_err(|e| {
            WebError::new(
                StatusCode::BAD_GATEWAY,
                transport_error(format_js_value(&e)),
            )
        })?;
        let response: web_sys::Response = response.dyn_into().map_err(|_| {
            WebError::new(
                StatusCode::BAD_GATEWAY,
                transport_error("Failed to cast to Response"),
            )
        })?;

        let status = StatusCode::from_u16(response.status() as u16)
            .map_err(|e| WebError::new(StatusCode::BAD_GATEWAY, e))?;
        let mut headers = http_kit::header::HeaderMap::new();
        for pair in response.headers().entries() {
            let pair = pair.map_err(|err| {
                WebError::new(
                    StatusCode::BAD_GATEWAY,
                    transport_error(format_js_value(&err)),
                )
            })?;
            let entry: js_sys::Array = pair.dyn_into().map_err(|_| {
                WebError::new(
                    StatusCode::BAD_GATEWAY,
                    transport_error("Failed to cast header entry to Array"),
                )
            })?;
            let name = entry.get(0).as_string().ok_or_else(|| {
                WebError::new(
                    StatusCode::BAD_GATEWAY,
                    transport_error("Failed to read header name"),
                )
            })?;
            let value = entry.get(1).as_string().ok_or_else(|| {
                WebError::new(
                    StatusCode::BAD_GATEWAY,
                    transport_error("Failed to read header value"),
                )
            })?;
            headers.insert(
                http_kit::header::HeaderName::from_bytes(name.as_bytes())
                    .map_err(|e| WebError::new(StatusCode::BAD_GATEWAY, e))?,
                http_kit::header::HeaderValue::from_str(&value)
                    .map_err(|e| WebError::new(StatusCode::BAD_GATEWAY, e))?,
            );
        }

        let body = response
            .body()
            .map(|body| {
                let stream = wasm_streams::ReadableStream::from_raw(body).into_stream();

                let stream = stream.map(|result| {
                    result
                        .map(|chunk| {
                            let uint8_array = js_sys::Uint8Array::new(&chunk);
                            let mut vec = vec![0; uint8_array.length() as usize];
                            uint8_array.copy_to(&mut vec[..]);
                            let chunk: Box<[u8]> = vec.into_boxed_slice();
                            chunk
                        })
                        .map_err(|e| {
                            BodyError::Other(Box::new(io::Error::new(
                                io::ErrorKind::Other,
                                format!("Failed to read body: {e:?}"),
                            )))
                        })
                });
                http_kit::Body::from_stream(SingleThreaded(stream))
            })
            .unwrap_or_else(http_kit::Body::empty);

        let is_error = status.is_client_error() || status.is_server_error();
        let mut response: http::Response<http_kit::Body> = http::Response::new(body);

        *response.headers_mut() = headers;
        *response.status_mut() = status;

        if is_error {
            let body = response
                .body_mut()
                .as_str()
                .await
                .ok()
                .map(|text| text.to_owned());
            return Err(WebError::remote(status, body, response));
        }
        Ok(response)
    })
}

fn format_js_value(value: &JsValue) -> String {
    value.as_string().unwrap_or_else(|| format!("{value:?}"))
}

fn transport_error(message: impl Into<String>) -> io::Error {
    io::Error::new(io::ErrorKind::Other, message.into())
}

impl Client for WebBackend {}