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::sync::{MAX_SYNC_PULL_RESPONSE_BYTES, SyncTransportFuture};
use crate::authority_client::{
ProtocolByteStream, ProtocolHttp, ProtocolHttpRequest, ProtocolHttpResponse,
ProtocolHttpStream,
};
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)]
pub(crate) struct AuthorityHttp(BrowserHttpClient);
pub(crate) fn authority_http(headers: &[(String, String)]) -> Result<AuthorityHttp, LixError> {
Ok(AuthorityHttp(BrowserHttpClient::from_headers(headers)?))
}
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
}
}
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,
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 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);
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)?;
Reflect::set(&init, &"lixResponseStream".into(), &JsValue::TRUE)
.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)?;
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 = fetch
.call2(&this, &request.url.into(), &init)
.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> {
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);
fetch(
&request.url,
request.method.as_str(),
&headers,
request.body,
request.cache_immutable,
self.fetch.as_ref(),
request.operation,
)
.await
})
}
}
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,
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)?;
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)?;
}
Reflect::set(
&init,
&"lixResponseLimit".into(),
&JsValue::from_f64(MAX_SYNC_PULL_RESPONSE_BYTES as f64),
)
.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 = fetch
.call2(&this, &url.into(), &init)
.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(&response, operation, &controller).await?;
abort_on_drop.disarm();
Ok(RawHttpResponse {
status,
status_text,
body,
})
}
async fn read_response_body(
response: &JsValue,
operation: &str,
controller: &Object,
) -> 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 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 {
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) > MAX_SYNC_PULL_RESPONSE_BYTES {
abort_controller(controller);
cancel_reader(&reader);
return Err(response_too_large(operation));
}
let offset = body.len();
body.resize(offset + chunk_len, 0);
chunk.copy_to(&mut body[offset..]);
}
}
fn cancel_reader(reader: &JsValue) {
let Ok(cancel) = Reflect::get(reader, &"cancel".into()) else {
return;
};
let Ok(cancel) = cancel.dyn_into::<Function>() else {
return;
};
let Ok(result) = cancel.call0(reader) else {
return;
};
let _ = result;
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
}