lix 0.18.0

//! Browser fetch mechanics for the repository-scoped sync protocol.

use js_sys::{Array, Function, Object, Promise, Reflect, Uint8Array};
use std::cell::RefCell;
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use wasm_bindgen::{JsCast, JsValue, closure::Closure};
use wasm_bindgen_futures::JsFuture;

use super::super::http::{
    HTTP_TIMEOUT, HttpSyncTransport, RawHttpClient, RawHttpRequest, RawHttpResponse,
    SYNC_TRANSPORT_ERROR_CODE, response_too_large,
};
use crate::LixError;
use crate::authority_client::{
    ProtocolByteStream, ProtocolHttp, ProtocolHttpRequest, ProtocolHttpResponse, ProtocolHttpStream,
};
use crate::sync::{MAX_SYNC_PULL_RESPONSE_BYTES, SyncTransportFuture};
use bytes::Bytes;

#[doc(hidden)]
pub const BROWSER_TRANSPORT_CONFIG_HEADER: &str = "x-lix-internal-browser-transport";

#[derive(Clone)]
struct BrowserTransportConfig {
    header_provider: Option<Function>,
    fetch: Option<Function>,
}

thread_local! {
    static BROWSER_TRANSPORT_CONFIGS: RefCell<HashMap<String, BrowserTransportConfig>> =
        RefCell::new(HashMap::new());
}

#[doc(hidden)]
pub fn register_browser_sync_transport(
    id: String,
    header_provider: Option<Function>,
    fetch: Option<Function>,
) {
    BROWSER_TRANSPORT_CONFIGS.with(|configs| {
        configs.borrow_mut().insert(
            id,
            BrowserTransportConfig {
                header_provider,
                fetch,
            },
        );
    });
}

#[doc(hidden)]
pub fn unregister_browser_sync_transport(id: &str) {
    BROWSER_TRANSPORT_CONFIGS.with(|configs| {
        configs.borrow_mut().remove(id);
    });
}

#[derive(Clone)]
pub(crate) struct BrowserHttpClient {
    headers: Vec<(String, String)>,
    header_provider: Option<Function>,
    fetch: Option<Function>,
}

#[derive(Clone, Debug)]
pub(crate) struct AuthorityHttp(BrowserHttpClient);

pub(crate) fn authority_http(headers: &[(String, String)]) -> Result<AuthorityHttp, LixError> {
    Ok(AuthorityHttp(BrowserHttpClient::from_headers(headers)?))
}

const HOSTED_RESPONSE_LIMIT: usize = 64 * 1024;

struct UploadBodyState {
    stream: RefCell<Option<ProtocolByteStream>>,
    canceled: std::cell::Cell<bool>,
    pending: RefCell<Option<futures_util::future::AbortHandle>>,
}

impl UploadBodyState {
    fn cancel(&self) {
        self.canceled.set(true);
        self.stream.borrow_mut().take();
        if let Some(pending) = self.pending.borrow_mut().take() {
            pending.abort();
        }
    }
}

struct UploadBodyGuard(std::rc::Rc<UploadBodyState>);
impl Drop for UploadBodyGuard {
    fn drop(&mut self) {
        self.0.cancel();
    }
}

impl AuthorityHttp {
    pub(crate) async fn upload(
        &self,
        request: ProtocolHttpRequest,
        body: Option<ProtocolByteStream>,
    ) -> Result<ProtocolHttpResponse, LixError> {
        let trace_headers = crate::telemetry::current_trace_context_headers();
        use futures_util::{
            StreamExt,
            future::{AbortHandle, Abortable},
        };
        let has_body = body.is_some();
        let state = std::rc::Rc::new(UploadBodyState {
            stream: RefCell::new(body),
            canceled: std::cell::Cell::new(false),
            pending: RefCell::new(None),
        });
        // Fetch may reject without canceling its request body. Cancellation must
        // also reach a Rust producer that is currently waiting for its next chunk.
        let _body_guard = UploadBodyGuard(state.clone());
        let pull_state = state.clone();
        let pull: Closure<dyn FnMut(JsValue) -> Promise> =
            Closure::wrap(Box::new(move |controller: JsValue| -> Promise {
                let state = pull_state.clone();
                wasm_bindgen_futures::future_to_promise(async move {
                    if state.canceled.get() {
                        return Ok(JsValue::UNDEFINED);
                    }
                    let mut stream = state.stream.borrow_mut().take();
                    let (abort, registration) = AbortHandle::new_pair();
                    *state.pending.borrow_mut() = Some(abort);
                    let item = Abortable::new(
                        async {
                            match stream.as_mut() {
                                Some(stream) => stream.next().await,
                                None => None,
                            }
                        },
                        registration,
                    )
                    .await;
                    state.pending.borrow_mut().take();
                    // A cancel that arrived during next() wins. Never put the source
                    // back into the shared state after cancellation.
                    if state.canceled.get() || item.is_err() {
                        return Ok(JsValue::UNDEFINED);
                    }
                    let (method, value) = match item.unwrap() {
                        Some(Ok(bytes)) => {
                            *state.stream.borrow_mut() = stream;
                            ("enqueue", Uint8Array::from(bytes.as_ref()).into())
                        }
                        Some(Err(error)) => ("error", JsValue::from_str(&error.to_string())),
                        None => ("close", JsValue::UNDEFINED),
                    };
                    let function =
                        Reflect::get(&controller, &method.into())?.dyn_into::<Function>()?;
                    function.call1(&controller, &value)?;
                    Ok(JsValue::UNDEFINED)
                })
            }));
        let cancel_state = state.clone();
        let cancel: Closure<dyn FnMut()> = Closure::wrap(Box::new(move || cancel_state.cancel()));
        let source = Object::new();
        Reflect::set(&source, &"pull".into(), &pull.into_js_value()).map_err(js_transport_error)?;
        Reflect::set(&source, &"cancel".into(), &cancel.into_js_value())
            .map_err(js_transport_error)?;
        let global = js_sys::global();
        let constructor = Reflect::get(&global, &"ReadableStream".into())
            .map_err(js_transport_error)?
            .dyn_into::<Function>()
            .map_err(js_transport_error)?;
        let args = Array::new();
        args.push(&source);
        let stream = Reflect::construct(&constructor, &args).map_err(js_transport_error)?;
        let init = Object::new();
        Reflect::set(&init, &"method".into(), &request.method.into())
            .map_err(js_transport_error)?;
        if has_body {
            Reflect::set(&init, &"body".into(), &stream).map_err(js_transport_error)?;
            Reflect::set(&init, &"duplex".into(), &"half".into()).map_err(js_transport_error)?;
        }
        Reflect::set(&init, &"credentials".into(), &"include".into())
            .map_err(js_transport_error)?;
        Reflect::set(&init, &"redirect".into(), &"error".into()).map_err(js_transport_error)?;
        Reflect::set(&init, &"cache".into(), &"no-store".into()).map_err(js_transport_error)?;
        let controller_constructor = Reflect::get(&global, &"AbortController".into())
            .map_err(js_transport_error)?
            .dyn_into::<Function>()
            .map_err(js_transport_error)?;
        let controller: Object = Reflect::construct(&controller_constructor, &Array::new())
            .map_err(js_transport_error)?
            .into();
        let signal = Reflect::get(&controller, &"signal".into()).map_err(js_transport_error)?;
        Reflect::set(&init, &"signal".into(), &signal).map_err(js_transport_error)?;
        let timeout_controller = controller.clone();
        let timeout_state = state.clone();
        let timeout_callback: Closure<dyn FnMut()> = Closure::wrap(Box::new(move || {
            timeout_state.cancel();
            abort_controller(&timeout_controller);
        }));
        let set_timeout = Reflect::get(&global, &"setTimeout".into())
            .map_err(js_transport_error)?
            .dyn_into::<Function>()
            .map_err(js_transport_error)?;
        let timeout_handle = set_timeout
            .call2(
                &global,
                timeout_callback.as_ref(),
                &JsValue::from_f64(HTTP_TIMEOUT.as_millis() as f64),
            )
            .map_err(js_transport_error)?;
        let mut abort_on_drop = AbortOnDrop {
            controller: controller.clone(),
            timeout: Some(BrowserTimeout {
                global: global.clone().into(),
                handle: timeout_handle,
                _callback: timeout_callback,
            }),
            armed: true,
        };
        let mut headers =
            resolve_request_headers(&self.0.headers, self.0.header_provider.as_ref()).await?;
        headers
            .retain(|(name, _)| !HttpSyncTransport::<BrowserHttpClient>::is_reserved_header(name));
        headers.extend(request.headers);
        replace_trace_headers(&mut headers, trace_headers);
        let pairs = Array::new();
        for (name, value) in headers {
            let pair = Array::new();
            pair.push(&name.into());
            pair.push(&value.into());
            pairs.push(&pair);
        }
        Reflect::set(&init, &"headers".into(), &pairs).map_err(js_transport_error)?;
        // Native browser Fetch must support streamed request bodies. Never turn
        // a repository into a giant ArrayBuffer as a compatibility fallback.
        if has_body && self.0.fetch.is_none() {
            let constructor = Reflect::get(&global, &"Request".into())
                .map_err(js_transport_error)?
                .dyn_into::<Function>()
                .map_err(js_transport_error)?;
            let args = Array::new();
            args.push(&request.url.clone().into());
            args.push(&init);
            let probe =
                Reflect::construct(&constructor, &args).map_err(|_| unsupported_upload())?;
            if Reflect::get(&probe, &"duplex".into())
                .ok()
                .and_then(|value| value.as_string())
                .as_deref()
                != Some("half")
            {
                return Err(unsupported_upload());
            }
        }
        let fetch = match &self.0.fetch {
            Some(fetch) => fetch.clone(),
            None => Reflect::get(&global, &"fetch".into())
                .map_err(js_transport_error)?
                .dyn_into::<Function>()
                .map_err(js_transport_error)?,
        };
        let this = if self.0.fetch.is_some() {
            JsValue::UNDEFINED
        } else {
            global.into()
        };
        let response = JsFuture::from(
            call_http_transport(
                &fetch,
                &this,
                &request.url,
                &init,
                Some(HOSTED_RESPONSE_LIMIT),
            )
            .map_err(js_transport_error)?
            .dyn_into::<Promise>()
            .map_err(js_transport_error)?,
        )
        .await
        .map_err(js_transport_error)?;
        let status = Reflect::get(&response, &"status".into())
            .map_err(js_transport_error)?
            .as_f64()
            .unwrap_or_default() as u16;
        let headers = browser_response_headers(&response)?;
        let bytes = read_response_body_limited(
            &response,
            "repository creation",
            &controller,
            HOSTED_RESPONSE_LIMIT,
        )
        .await?;
        abort_on_drop.disarm();
        Ok(ProtocolHttpResponse {
            status,
            headers,
            body: Bytes::from(bytes),
        })
    }
}

fn unsupported_upload() -> LixError {
    LixError::new(
        "LIX_UNSUPPORTED_OPERATION",
        "creating a repository from a local Lix requires Fetch request streaming support in this browser",
    )
}

impl std::fmt::Debug for BrowserHttpClient {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("BrowserHttpClient")
            .field("headers", &self.headers)
            .field("has_header_provider", &self.header_provider.is_some())
            .field("has_fetch_override", &self.fetch.is_some())
            .finish()
    }
}

impl HttpSyncTransport<BrowserHttpClient> {
    pub(crate) async fn connect(
        repository_url: &str,
        headers: &[(String, String)],
    ) -> Result<Self, LixError> {
        let client = BrowserHttpClient::from_headers(headers)?;
        Self::connect_with(client, repository_url).await
    }

    pub(crate) async fn connect_with_progress(
        repository_url: &str,
        headers: &[(String, String)],
        progress: Option<&Arc<dyn crate::OpenProgressSink>>,
    ) -> Result<Self, LixError> {
        let client = BrowserHttpClient::from_headers(headers)?;
        Self::connect_with_progress_sink(client, repository_url, progress).await
    }
}

impl BrowserHttpClient {
    fn from_headers(headers: &[(String, String)]) -> Result<Self, LixError> {
        let config_id = headers
            .iter()
            .find(|(name, _)| name.eq_ignore_ascii_case(BROWSER_TRANSPORT_CONFIG_HEADER))
            .map(|(_, value)| value.clone());
        let config = config_id.as_deref().and_then(|id| {
            BROWSER_TRANSPORT_CONFIGS.with(|configs| configs.borrow().get(id).cloned())
        });
        if config_id.is_some() && config.is_none() {
            return Err(LixError::new(
                LixError::CODE_CLOSED,
                "browser sync transport callbacks are no longer registered",
            ));
        }
        Ok(BrowserHttpClient {
            headers: headers
                .iter()
                .filter(|(name, _)| {
                    !name.eq_ignore_ascii_case(BROWSER_TRANSPORT_CONFIG_HEADER)
                        && !HttpSyncTransport::<BrowserHttpClient>::is_reserved_header(name)
                })
                .cloned()
                .collect(),
            header_provider: config
                .as_ref()
                .and_then(|value| value.header_provider.clone()),
            fetch: config.and_then(|value| value.fetch),
        })
    }
}

impl ProtocolHttp for AuthorityHttp {
    async fn request(
        &self,
        request: ProtocolHttpRequest,
    ) -> Result<ProtocolHttpResponse, LixError> {
        let method = request.method.parse().map_err(|error| {
            LixError::new(
                LixError::CODE_INVALID_PARAM,
                format!("invalid HTTP method: {error}"),
            )
        })?;
        let response = self
            .0
            .send(RawHttpRequest {
                method,
                url: request.url,
                headers: request.headers,
                body: request.body.map(|body| body.to_vec()),
                cache_immutable: false,
                response_limit: MAX_SYNC_PULL_RESPONSE_BYTES,
                operation: "authority request",
            })
            .await?;
        Ok(ProtocolHttpResponse {
            status: response.status,
            headers: Vec::new(),
            body: response.body.into(),
        })
    }

    async fn request_stream(
        &self,
        request: ProtocolHttpRequest,
    ) -> Result<ProtocolHttpStream, LixError> {
        authority_stream(&self.0, request).await
    }

    async fn sleep(&self, duration: Duration) {
        super::wasm::sleep(duration).await;
    }

    fn spawn(&self, future: Pin<Box<dyn Future<Output = ()>>>) {
        wasm_bindgen_futures::spawn_local(future);
    }
}

async fn authority_stream(
    client: &BrowserHttpClient,
    request: ProtocolHttpRequest,
) -> Result<ProtocolHttpStream, LixError> {
    let trace_headers = crate::telemetry::current_trace_context_headers();
    let mut headers =
        resolve_request_headers(&client.headers, client.header_provider.as_ref()).await?;
    headers.retain(|(name, _)| !HttpSyncTransport::<BrowserHttpClient>::is_reserved_header(name));
    headers.extend(request.headers);
    replace_trace_headers(&mut headers, trace_headers);

    let init = Object::new();
    Reflect::set(&init, &"method".into(), &request.method.into()).map_err(js_transport_error)?;
    Reflect::set(&init, &"credentials".into(), &"include".into()).map_err(js_transport_error)?;
    Reflect::set(&init, &"cache".into(), &"no-store".into()).map_err(js_transport_error)?;
    let header_pairs = Array::new();
    for (name, value) in &headers {
        let pair = Array::new();
        pair.push(&name.into());
        pair.push(&value.into());
        header_pairs.push(&pair);
    }
    Reflect::set(&init, &"headers".into(), &header_pairs).map_err(js_transport_error)?;
    if let Some(body) = request.body {
        Reflect::set(&init, &"body".into(), &Uint8Array::from(body.as_ref()))
            .map_err(js_transport_error)?;
    }

    let global = js_sys::global();
    let controller_constructor = Reflect::get(&global, &"AbortController".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?;
    let controller =
        Reflect::construct(&controller_constructor, &Array::new()).map_err(js_transport_error)?;
    let signal = Reflect::get(&controller, &"signal".into()).map_err(js_transport_error)?;
    Reflect::set(&init, &"signal".into(), &signal).map_err(js_transport_error)?;
    // Dropping a `JsFuture` does not cancel its Promise. Observation membership
    // changes deliberately drop an in-flight stream-open future, so retain an
    // abort guard until the response body has been transferred to the returned
    // `ProtocolHttpStream`. Without it, every superseded open remains a live SSE
    // request and can exhaust the browser's per-origin connection budget.
    let controller: Object = controller.into();
    let mut abort_on_drop = AbortOnDrop {
        controller: controller.clone(),
        timeout: None,
        armed: true,
    };
    let cancel_controller = controller.clone();
    let cancel: Arc<dyn Fn()> = Arc::new(move || {
        if let Ok(abort) = Reflect::get(&cancel_controller, &"abort".into())
            && let Ok(abort) = abort.dyn_into::<Function>()
        {
            let _ = abort.call0(&cancel_controller);
        }
    });

    let fetch = match client.fetch.as_ref() {
        Some(fetch) => fetch.clone(),
        None => Reflect::get(&global, &"fetch".into())
            .map_err(js_transport_error)?
            .dyn_into::<Function>()
            .map_err(js_transport_error)?,
    };
    let this = if client.fetch.is_some() {
        JsValue::UNDEFINED
    } else {
        global.into()
    };
    let promise = call_http_transport(&fetch, &this, &request.url, &init, None)
        .map_err(js_transport_error)?
        .dyn_into::<Promise>()
        .map_err(js_transport_error)?;
    let response = JsFuture::from(promise).await.map_err(js_transport_error)?;
    let status = Reflect::get(&response, &"status".into())
        .map_err(js_transport_error)?
        .as_f64()
        .unwrap_or_default() as u16;
    let response_headers = browser_response_headers(&response)?;
    let body = Reflect::get(&response, &"body".into()).map_err(js_transport_error)?;
    if body.is_null() || body.is_undefined() {
        return Err(LixError::new(
            "LIX_SERVER_PROTOCOL_ERROR",
            "authority stream response has no body",
        ));
    }
    let reader = Reflect::get(&body, &"getReader".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?
        .call0(&body)
        .map_err(js_transport_error)?;
    let stream: ProtocolByteStream = Box::pin(async_stream::stream! {
        loop {
            let read = match Reflect::get(&reader, &"read".into())
                .ok()
                .and_then(|value| value.dyn_into::<Function>().ok())
            {
                Some(read) => read,
                None => break,
            };
            let result = match read.call0(&reader)
                .ok()
                .and_then(|value| value.dyn_into::<Promise>().ok())
            {
                Some(promise) => JsFuture::from(promise).await,
                None => break,
            };
            match result {
                Ok(chunk) => {
                    if Reflect::get(&chunk, &"done".into())
                        .ok()
                        .and_then(|value| value.as_bool())
                        .unwrap_or(false)
                    {
                        break;
                    }
                    let value = Reflect::get(&chunk, &"value".into()).unwrap_or(JsValue::UNDEFINED);
                    yield Ok(Bytes::from(Uint8Array::new(&value).to_vec()));
                }
                Err(error) => {
                    yield Err(js_transport_error(error));
                    break;
                }
            }
        }
    });
    abort_on_drop.disarm();
    Ok(ProtocolHttpStream {
        status,
        headers: response_headers,
        body: stream,
        cancel,
    })
}

fn browser_response_headers(response: &JsValue) -> Result<Vec<(String, String)>, LixError> {
    let headers = Reflect::get(response, &"headers".into()).map_err(js_transport_error)?;
    if headers.is_null() || headers.is_undefined() {
        return Ok(Vec::new());
    }
    let entries = Reflect::get(&headers, &"entries".into())
        .ok()
        .and_then(|value| value.dyn_into::<Function>().ok())
        .and_then(|entries| entries.call0(&headers).ok())
        .unwrap_or(headers);
    Ok(Array::from(&entries)
        .iter()
        .filter_map(|entry| {
            let pair = Array::from(&entry);
            Some((pair.get(0).as_string()?, pair.get(1).as_string()?))
        })
        .collect())
}

impl RawHttpClient for BrowserHttpClient {
    fn send(&self, request: RawHttpRequest) -> SyncTransportFuture<'_, RawHttpResponse> {
        // Capture the current Lix span while this operation is polled. The JS
        // fetch bridge runs on a different event loop and cannot read it later.
        let trace_headers = crate::telemetry::current_trace_context_headers();
        Box::pin(async move {
            let mut headers =
                resolve_request_headers(&self.headers, self.header_provider.as_ref()).await?;
            headers.retain(|(name, _)| {
                !HttpSyncTransport::<BrowserHttpClient>::is_reserved_header(name)
            });
            headers.extend(request.headers);
            replace_trace_headers(&mut headers, trace_headers);
            fetch(
                &request.url,
                request.method.as_str(),
                &headers,
                request.body,
                request.cache_immutable,
                request.response_limit,
                self.fetch.as_ref(),
                request.operation,
            )
            .await
        })
    }
}

fn replace_trace_headers(
    headers: &mut Vec<(String, String)>,
    trace_headers: Vec<(String, String)>,
) {
    if trace_headers.is_empty() {
        return;
    }
    headers.retain(|(name, _)| {
        !name.eq_ignore_ascii_case("traceparent") && !name.eq_ignore_ascii_case("tracestate")
    });
    headers.extend(trace_headers);
}

async fn resolve_request_headers(
    static_headers: &[(String, String)],
    provider: Option<&Function>,
) -> Result<Vec<(String, String)>, LixError> {
    let Some(provider) = provider else {
        return Ok(static_headers.to_vec());
    };
    let value = provider
        .call0(&JsValue::UNDEFINED)
        .map_err(js_callback_error)?;
    let value = JsFuture::from(Promise::resolve(&value))
        .await
        .map_err(js_callback_error)?;
    if !Array::is_array(&value) {
        return Err(LixError::new(
            LixError::CODE_INVALID_PARAM,
            "browser sync header provider must return [name, value] pairs",
        ));
    }
    let dynamic = Array::from(&value)
        .iter()
        .map(|pair| {
            if !Array::is_array(&pair) {
                return Err(LixError::new(
                    LixError::CODE_INVALID_PARAM,
                    "browser sync header provider must return [name, value] pairs",
                ));
            }
            let pair = Array::from(&pair);
            let name = pair.get(0).as_string();
            let value = pair.get(1).as_string();
            match (pair.length(), name, value) {
                (2, Some(name), Some(value)) => Ok((name, value)),
                _ => Err(LixError::new(
                    LixError::CODE_INVALID_PARAM,
                    "browser sync header provider must return [name, value] pairs",
                )),
            }
        })
        .collect::<Result<Vec<_>, _>>()?;
    let mut headers = static_headers.to_vec();
    headers.extend(dynamic);
    Ok(headers)
}

async fn fetch(
    url: &str,
    method: &str,
    headers: &[(String, String)],
    body: Option<Vec<u8>>,
    cache_immutable: bool,
    response_limit: usize,
    fetch_override: Option<&Function>,
    operation: &str,
) -> Result<RawHttpResponse, LixError> {
    let init = Object::new();
    let global = js_sys::global();
    let controller_constructor = Reflect::get(&global, &"AbortController".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?;
    let controller =
        Reflect::construct(&controller_constructor, &Array::new()).map_err(js_transport_error)?;
    let signal = Reflect::get(&controller, &"signal".into()).map_err(js_transport_error)?;
    Reflect::set(&init, &"signal".into(), &signal).map_err(js_transport_error)?;
    // A dropped `JsFuture` does not cancel its Promise. Keep cancellation at
    // the fetch boundary through AbortController.
    let controller: Object = controller.into();
    let timeout_controller = controller.clone();
    let timeout_callback: Closure<dyn FnMut()> = Closure::wrap(Box::new(move || {
        abort_controller(&timeout_controller);
    }));
    let set_timeout = Reflect::get(&global, &"setTimeout".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?;
    let timeout_handle = set_timeout
        .call2(
            &global,
            timeout_callback.as_ref(),
            &JsValue::from_f64(HTTP_TIMEOUT.as_millis() as f64),
        )
        .map_err(js_transport_error)?;
    let mut abort_on_drop = AbortOnDrop {
        controller: controller.clone(),
        timeout: Some(BrowserTimeout {
            global: global.clone().into(),
            handle: timeout_handle,
            _callback: timeout_callback,
        }),
        armed: true,
    };
    Reflect::set(&init, &"method".into(), &method.into()).map_err(js_transport_error)?;
    Reflect::set(&init, &"credentials".into(), &"include".into()).map_err(js_transport_error)?;
    if !cache_immutable {
        Reflect::set(&init, &"cache".into(), &"no-store".into()).map_err(js_transport_error)?;
    }
    let header_pairs = Array::new();
    for (name, value) in headers {
        let pair = Array::new();
        pair.push(&name.into());
        pair.push(&value.into());
        header_pairs.push(&pair);
    }
    Reflect::set(&init, &"headers".into(), &header_pairs).map_err(js_transport_error)?;
    if let Some(body) = body {
        let body: JsValue = Uint8Array::from(body.as_slice()).into();
        Reflect::set(&init, &"body".into(), &body).map_err(js_transport_error)?;
    }

    let fetch = match fetch_override {
        Some(fetch) => fetch.clone(),
        None => Reflect::get(&global, &"fetch".into())
            .map_err(js_transport_error)?
            .dyn_into::<Function>()
            .map_err(js_transport_error)?,
    };
    let this = if fetch_override.is_some() {
        JsValue::UNDEFINED
    } else {
        global.clone().into()
    };
    let promise = call_http_transport(&fetch, &this, url, &init, Some(response_limit))
        .map_err(js_transport_error)?
        .dyn_into::<Promise>()
        .map_err(js_transport_error)?;
    let response = JsFuture::from(promise).await.map_err(js_transport_error)?;
    let status = Reflect::get(&response, &"status".into())
        .map_err(js_transport_error)?
        .as_f64()
        .unwrap_or_default() as u16;
    let status_text = Reflect::get(&response, &"statusText".into())
        .map_err(js_transport_error)?
        .as_string()
        .unwrap_or_default();
    let body =
        read_response_body_limited(&response, operation, &controller, response_limit).await?;
    abort_on_drop.disarm();
    Ok(RawHttpResponse {
        status,
        status_text,
        body,
    })
}

async fn read_response_body_limited(
    response: &JsValue,
    operation: &str,
    controller: &Object,
    limit: usize,
) -> Result<Vec<u8>, LixError> {
    let stream = Reflect::get(response, &"body".into()).map_err(js_transport_error)?;
    if stream.is_null() || stream.is_undefined() {
        return Ok(Vec::new());
    }
    let reader = Reflect::get(&stream, &"getReader".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?
        .call0(&stream)
        .map_err(js_transport_error)?;
    let mut reader_guard = ResponseReaderGuard {
        reader: reader.clone(),
        complete: false,
    };
    let read = Reflect::get(&reader, &"read".into())
        .map_err(js_transport_error)?
        .dyn_into::<Function>()
        .map_err(js_transport_error)?;
    let mut body = Vec::new();
    loop {
        let result = read
            .call0(&reader)
            .map_err(js_transport_error)?
            .dyn_into::<Promise>()
            .map_err(js_transport_error)?;
        let result = JsFuture::from(result).await.map_err(js_transport_error)?;
        let done = Reflect::get(&result, &"done".into())
            .map_err(js_transport_error)?
            .as_bool()
            .unwrap_or(false);
        if done {
            reader_guard.complete = true;
            release_reader(&reader);
            return Ok(body);
        }
        let chunk = Reflect::get(&result, &"value".into())
            .map_err(js_transport_error)?
            .dyn_into::<Uint8Array>()
            .map_err(js_transport_error)?;
        let chunk_len = usize::try_from(chunk.length()).map_err(|_| {
            LixError::new(
                LixError::CODE_INTERNAL_ERROR,
                "browser sync response chunk length exceeds usize",
            )
        })?;
        if body.len().saturating_add(chunk_len) > limit {
            abort_controller(controller);
            return Err(if limit == MAX_SYNC_PULL_RESPONSE_BYTES {
                response_too_large(operation)
            } else {
                LixError::new(
                    "LIX_SERVER_PROTOCOL_ERROR",
                    format!("{operation} response exceeds {limit} bytes"),
                )
            });
        }
        let offset = body.len();
        body.resize(offset + chunk_len, 0);
        chunk.copy_to(&mut body[offset..]);
    }
}

struct ResponseReaderGuard {
    reader: JsValue,
    complete: bool,
}
impl Drop for ResponseReaderGuard {
    fn drop(&mut self) {
        if !self.complete {
            cancel_reader(&self.reader);
        }
    }
}

fn cancel_reader(reader: &JsValue) {
    let reader = reader.clone();
    let canceled = Reflect::get(&reader, &"cancel".into())
        .ok()
        .and_then(|value| value.dyn_into::<Function>().ok())
        .and_then(|cancel| cancel.call0(&reader).ok());
    // A pending read can keep the lock until cancellation settles. Release only
    // afterward, including when the stream's cancel hook rejects.
    wasm_bindgen_futures::spawn_local(async move {
        if let Some(result) = canceled {
            let _ = JsFuture::from(Promise::resolve(&result)).await;
        }
        release_reader(&reader);
    });
}

fn release_reader(reader: &JsValue) {
    if let Ok(release) = Reflect::get(reader, &"releaseLock".into())
        && let Ok(release) = release.dyn_into::<Function>()
    {
        let _ = release.call0(reader);
    }
}

struct AbortOnDrop {
    controller: Object,
    timeout: Option<BrowserTimeout>,
    armed: bool,
}

struct BrowserTimeout {
    global: JsValue,
    handle: JsValue,
    _callback: Closure<dyn FnMut()>,
}

impl AbortOnDrop {
    fn disarm(&mut self) {
        self.clear_timeout();
        self.armed = false;
    }

    fn clear_timeout(&mut self) {
        let Some(timeout) = self.timeout.take() else {
            return;
        };
        if let Ok(clear_timeout) = Reflect::get(&timeout.global, &"clearTimeout".into())
            && let Ok(clear_timeout) = clear_timeout.dyn_into::<Function>()
        {
            let _ = clear_timeout.call1(&timeout.global, &timeout.handle);
        }
    }
}

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.clear_timeout();
        if self.armed {
            abort_controller(&self.controller);
        }
    }
}

fn abort_controller(controller: &Object) {
    if let Ok(abort) = Reflect::get(controller, &"abort".into())
        && let Ok(abort) = abort.dyn_into::<Function>()
    {
        let _ = abort.call0(controller);
    }
}

fn js_transport_error(error: JsValue) -> LixError {
    let detail = js_error_detail(&error);
    if let Some(code) = js_error_code(&error).filter(|code| code.starts_with("LIX_")) {
        return LixError::new(code, detail);
    }
    LixError::new(
        SYNC_TRANSPORT_ERROR_CODE,
        format!("browser sync fetch failed: {detail}"),
    )
}

fn js_callback_error(error: JsValue) -> LixError {
    let code = js_error_code(&error).unwrap_or_else(|| LixError::CODE_INTERNAL_ERROR.to_owned());
    LixError::new(code, js_error_detail(&error))
}

fn js_error_code(error: &JsValue) -> Option<String> {
    Reflect::get(error, &"code".into())
        .ok()
        .and_then(|code| code.as_string())
}

fn js_error_detail(error: &JsValue) -> String {
    let detail = Reflect::get(&error, &"message".into())
        .ok()
        .and_then(|message| message.as_string())
        .unwrap_or_else(|| format!("{error:?}"));
    detail
}

/// Internal JS transports take an explicit request object. Native browser fetch
/// is adapted here, and never receives private RequestInit extensions.
fn call_http_transport(
    transport: &Function,
    this: &JsValue,
    url: &str,
    init: &Object,
    max_bytes: Option<usize>,
) -> Result<JsValue, JsValue> {
    if !this.is_undefined() {
        return transport.call2(this, &url.into(), init);
    }
    let request = Object::new();
    Reflect::set(&request, &"url".into(), &url.into())?;
    Reflect::set(&request, &"init".into(), init)?;
    let response = Object::new();
    match max_bytes {
        Some(limit) => {
            Reflect::set(&response, &"mode".into(), &"buffered".into())?;
            Reflect::set(
                &response,
                &"maxBytes".into(),
                &JsValue::from_f64(limit as f64),
            )?;
        }
        None => {
            Reflect::set(&response, &"mode".into(), &"streaming".into())?;
        }
    }
    Reflect::set(&request, &"response".into(), &response)?;
    transport.call1(&JsValue::UNDEFINED, &request)
}