#![allow(deprecated)]
pub mod application_crypto;
pub mod application_crypto_streams;
pub mod client;
pub mod connection;
pub mod coordination;
pub mod explicit_transfer_crypto;
pub(crate) mod generated;
pub mod heartbeat;
#[cfg(feature = "iroh-carrier-core")]
pub mod iroh_carrier;
#[cfg(feature = "iroh-carrier-core")]
pub mod iroh_carrier_bootstrap;
#[cfg(feature = "iroh-carrier-core")]
pub mod iroh_carrier_kind;
#[cfg(feature = "iroh-carrier-core")]
pub mod iroh_carrier_proof;
pub(crate) mod iroh_connection_policy;
pub mod key_agreement;
pub mod lifecycle_reason;
pub(crate) mod native_moq_policy;
pub mod native_protocol;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) mod native_send_policy;
pub(crate) mod native_webrtc_policy;
#[cfg(feature = "iroh-carrier-core")]
pub mod packet_carrier_transport;
pub mod presence;
pub(crate) mod presence_policy;
pub mod protocol_config;
pub mod route_policy;
pub mod runtime_policy;
pub mod session_token;
pub mod signaling;
pub mod stream_metadata;
pub(crate) mod transport_generation;
pub(crate) mod transport_label;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
pub mod local_discovery;
#[cfg(all(target_arch = "wasm32", feature = "transport-webrtc"))]
compile_error!(
"feature `transport-webrtc` is native-only and must not be enabled for wasm32 targets"
);
#[cfg(all(target_arch = "wasm32", feature = "transport-moq"))]
compile_error!("feature `transport-moq` is native-only and must not be enabled for wasm32 targets");
#[cfg(test)]
pub mod test_constants {
pub const TEST_PROJECT_ID: &str = "test-project";
pub const TEST_API_KEY: &str = "pk_test_0000000000000000000000000000000000000000";
}
pub const LIVE_PROJECT_ID: &str = "pluto-rtc-prod";
pub fn validate_v2_public_api_key(api_key: &str) -> anyhow::Result<&str> {
let trimmed = api_key.trim();
let valid_prefix = trimmed.starts_with("pk_live_") || trimmed.starts_with("pk_test_");
let suffix = trimmed.get(8..).unwrap_or_default();
if !valid_prefix || suffix.len() != 40 || !suffix.bytes().all(|byte| byte.is_ascii_hexdigit()) {
anyhow::bail!("OpenRTC 2.0 requires a public pk_live_ or pk_test_ API key");
}
Ok(trimmed)
}
pub fn app_tag_from_api_key(api_key: &str) -> String {
let trimmed = api_key.trim();
if trimmed.is_empty() {
return "app_anonymous".to_string();
}
let suffix_len = trimmed.len().min(16);
format!("app_{}", &trimmed[trimmed.len() - suffix_len..])
}
pub fn space_app_tag_from_keys(api_key: &str, space_key: &str) -> String {
let input = format!("{}:{}", api_key.trim(), space_key.trim());
let digest = <sha2::Sha256 as sha2::Digest>::digest(input.as_bytes());
format!("space::{}", hex::encode(digest))
}
#[cfg(test)]
mod v2_constructor_contract_tests {
use super::*;
#[test]
fn rust_v2_constructor_is_provider_neutral_and_side_effect_free() {
let api_key = test_constants::TEST_API_KEY;
let client = client::Client::new_v2(api_key.to_string()).expect("valid public API key");
assert_eq!(client.app_tag(), app_tag_from_api_key(api_key));
assert!(client::Client::new_v2("firebase-project-id".to_string()).is_err());
}
}
#[cfg(not(target_arch = "wasm32"))]
pub fn ensure_default_rustls_provider() {
if rustls::crypto::CryptoProvider::get_default().is_none() {
let _ = rustls::crypto::ring::default_provider().install_default();
}
}
#[cfg(not(target_arch = "wasm32"))]
pub mod adapters;
#[cfg(not(target_arch = "wasm32"))]
pub mod native_coordination_gateway;
#[cfg(not(target_arch = "wasm32"))]
pub mod native_v2;
pub mod connection_manager;
pub mod logging;
pub mod protocol_registry;
#[cfg(not(target_arch = "wasm32"))]
pub mod runtime_manager;
#[cfg(not(target_arch = "wasm32"))]
pub mod transport;
#[cfg(not(target_arch = "wasm32"))]
pub use client::EndpointHandle;
#[cfg(all(
not(target_arch = "wasm32"),
not(any(target_os = "ios", target_os = "android"))
))]
pub mod sso;
#[cfg(not(target_arch = "wasm32"))]
pub mod native_node;
#[cfg(not(target_arch = "wasm32"))]
pub mod native_device;
#[cfg(all(
test,
not(target_arch = "wasm32"),
feature = "iroh-protocols-wasm",
any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq")
))]
mod native_carrier_protocol_test;
#[cfg(all(not(target_arch = "wasm32"), feature = "iroh-transport-moq"))]
pub mod native_moq_carrier;
#[cfg(all(not(target_arch = "wasm32"), feature = "iroh-transport-webrtc"))]
pub mod native_webrtc_carrier;
#[cfg(all(not(target_arch = "wasm32"), feature = "legacy-v1"))]
pub mod native_auth;
#[cfg(all(target_arch = "wasm32", feature = "iroh-protocols-wasm"))]
mod wasm_docs_persistence;
#[cfg(all(target_arch = "wasm32", feature = "iroh-protocols-wasm"))]
mod wasm_indexeddb_blob_store;
#[cfg(all(target_arch = "wasm32", feature = "iroh-transport-moq"))]
pub mod wasm_moq_carrier;
#[cfg(target_arch = "wasm32")]
pub mod wasm_node;
#[cfg(all(target_arch = "wasm32", feature = "iroh-transport-webrtc"))]
pub mod wasm_webrtc_carrier;
#[cfg(target_arch = "wasm32")]
#[macro_export]
macro_rules! console_log {
($($t:tt)*) => (web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format_args!($($t)*).to_string())))
}
#[cfg(target_arch = "wasm32")]
pub mod wasm_api {
use crate::client::Client;
use crate::session_token::split_compound_ticket;
use crate::wasm_node::{
into_js_readable_stream, peer_uni_stream_from_send, BiStream, PeerUniStream,
};
use iroh_tickets::endpoint::EndpointTicket;
#[cfg(any(
feature = "iroh-protocols-wasm",
feature = "iroh-transport-webrtc",
feature = "iroh-transport-moq"
))]
use std::rc::Rc;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
use std::{cell::RefCell, collections::HashMap};
use wasm_bindgen::prelude::*;
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
use wasm_bindgen_futures::spawn_local;
use wasm_streams::readable::sys::ReadableStream as JsReadableStream;
#[cfg(feature = "iroh-transport-webrtc")]
#[derive(Debug, Clone)]
struct WasmWebRtcCarrierAttempt {
connection_id: String,
remote_endpoint_id: iroh::EndpointId,
bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
generation: crate::client::WasmPeerDataGeneration,
role: &'static str,
offer_started: bool,
remote_ready: bool,
completion_started: bool,
retry_count: u8,
external_fallback_allowed: bool,
inbound_authorization_expires_at_ms: Option<f64>,
}
#[cfg(feature = "iroh-transport-moq")]
#[derive(Debug, Clone)]
struct WasmMoqCarrierAttempt {
connection_id: String,
remote_endpoint_id: iroh::EndpointId,
bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
generation: crate::client::WasmPeerDataGeneration,
role: &'static str,
external_fallback_allowed: bool,
inbound_authorization_expires_at_ms: Option<f64>,
}
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
#[derive(Debug, Clone, Copy)]
struct WasmRemoteCarrierCapabilities {
webrtc: bool,
webrtc_external: bool,
moq: bool,
moq_external: bool,
}
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
fn emit_wasm_carrier_action(
handler: &Rc<RefCell<Option<js_sys::Function>>>,
action: serde_json::Value,
) {
let Some(handler) = handler.borrow().as_ref().cloned() else {
return;
};
let Ok(value) = serde::Serialize::serialize(
&action,
&serde_wasm_bindgen::Serializer::json_compatible(),
) else {
return;
};
if let Err(error) = handler.call1(&JsValue::UNDEFINED, &value) {
web_sys::console::error_2(
&JsValue::from_str("[OpenRTC][WASM carrier] action handler failed"),
&error,
);
}
}
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
fn wasm_carrier_failure_code(error: &anyhow::Error) -> &'static str {
let message = format!("{error:#}");
if message.contains("incumbent generation changed")
|| message.contains("replacement incumbent is stale")
{
"carrier-base-generation-stale"
} else if message.contains("authorization epoch changed") {
"carrier-authorization-stale"
} else if message.contains("stale before atomic commit")
|| message.contains("became stale during atomic commit")
{
"carrier-logical-generation-stale"
} else if message.contains("attempt was retired") || message.contains("retired upgrade") {
"carrier-attempt-retired"
} else {
"carrier-proof-failed"
}
}
#[wasm_bindgen]
pub struct WasmClient {
inner: Arc<Client>,
identity_credential: Arc<Mutex<Option<String>>>,
last_auth_log: Arc<Mutex<Option<(bool, usize)>>>,
#[cfg(feature = "iroh-protocols-wasm")]
persistent_protocols:
Rc<tokio::sync::Mutex<Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>>>,
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_sessions:
Rc<RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>>,
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_remote_external: Rc<RefCell<HashMap<String, bool>>>,
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
wasm_carrier_action_handler: Rc<RefCell<Option<js_sys::Function>>>,
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_sessions:
Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_remote_external: Rc<RefCell<HashMap<String, bool>>>,
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
wasm_remote_carrier_capabilities:
Rc<RefCell<HashMap<String, WasmRemoteCarrierCapabilities>>>,
}
#[cfg(feature = "iroh-transport-webrtc")]
async fn fail_wasm_webrtc_attempt(
inner: Arc<Client>,
attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
sessions: Rc<
RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>,
>,
handler: Rc<RefCell<Option<js_sys::Function>>>,
attempt: WasmWebRtcCarrierAttempt,
failure_code: &'static str,
notify_peer: bool,
) {
let kind = crate::client::IrohPathKind::IrohWebRtc;
if !inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await
{
return;
}
if attempts
.borrow()
.get(&attempt.connection_id)
.is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
{
attempts.borrow_mut().remove(&attempt.connection_id);
}
sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
if notify_peer {
if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
&attempt.bootstrap,
failure_code,
) {
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "send-control",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": failed,
}),
);
}
}
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "retire-webrtc",
"connectionId": attempt.connection_id,
"upgradeId": attempt.bootstrap.upgrade_id,
"failureCode": failure_code,
}),
);
let retry_pending = attempt.retry_count == 0
&& matches!(
failure_code,
"data-channel-failed" | "ice-failed" | "signaling-failed"
);
if !attempt.external_fallback_allowed && attempt.role == "initiator" && !retry_pending {
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "advance-carrier",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"failedRoute": "iroh-webrtc",
}),
);
}
if attempt.external_fallback_allowed
&& attempt.role == "initiator"
&& matches!(
inner
.iroh_path_kind(&attempt.remote_endpoint_id.to_string())
.await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
)
{
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "fallback-external-webrtc",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"failureCode": failure_code,
}),
);
}
}
#[cfg(feature = "iroh-transport-webrtc")]
impl WasmClient {
fn schedule_wasm_webrtc_carrier_watchdog(
&self,
bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
) {
let inner = self.inner.clone();
let attempts = self.wasm_webrtc_carrier_attempts.clone();
let sessions = self.wasm_webrtc_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
gloo_timers::future::sleep(std::time::Duration::from_secs(30)).await;
let attempt = attempts
.borrow()
.values()
.find(|attempt| attempt.bootstrap.upgrade_id == bootstrap.upgrade_id)
.cloned();
if let Some(attempt) = attempt {
if !inner
.wasm_carrier_upgrade_is_current(
&attempt.connection_id,
crate::client::IrohPathKind::IrohWebRtc,
&attempt.bootstrap.upgrade_id,
)
.await
{
return;
}
fail_wasm_webrtc_attempt(
inner,
attempts,
sessions,
handler,
attempt,
"carrier-timeout",
true,
)
.await;
}
});
}
async fn fail_wasm_webrtc_carrier_attempt(
&self,
attempt: WasmWebRtcCarrierAttempt,
failure_code: &'static str,
notify_peer: bool,
) {
fail_wasm_webrtc_attempt(
self.inner.clone(),
self.wasm_webrtc_carrier_attempts.clone(),
self.wasm_webrtc_carrier_sessions.clone(),
self.wasm_carrier_action_handler.clone(),
attempt,
failure_code,
notify_peer,
)
.await;
}
fn spawn_outbound_wasm_webrtc_carrier_completion(&self, attempt: WasmWebRtcCarrierAttempt) {
let inner = self.inner.clone();
let attempts = self.wasm_webrtc_carrier_attempts.clone();
let sessions = self.wasm_webrtc_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
let kind = crate::client::IrohPathKind::IrohWebRtc;
let result = inner
.complete_outbound_wasm_carrier_upgrade(
&attempt.connection_id,
&attempt.remote_endpoint_id.to_string(),
&attempt.bootstrap.upgrade_id,
attempt.generation,
kind,
)
.await;
if let Err(error) = result {
let failure_code = wasm_carrier_failure_code(&error);
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OpenRTC][WebRTC carrier] outbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
)));
fail_wasm_webrtc_attempt(
inner,
attempts,
sessions,
handler,
attempt,
failure_code,
true,
)
.await;
return;
}
inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await;
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "selected",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"upgradeId": attempt.bootstrap.upgrade_id,
"family": "iroh",
"carrier": "webrtc",
"implementation": "iroh-carrier",
"transportGeneration": attempt.generation.transport_generation.saturating_add(1),
"routeGeneration": 0,
}),
);
});
}
fn take_ready_outbound_wasm_webrtc_carrier_attempt(
&self,
connection_id: &str,
upgrade_id: &str,
) -> Option<WasmWebRtcCarrierAttempt> {
if !self
.wasm_webrtc_carrier_sessions
.borrow()
.contains_key(upgrade_id)
{
return None;
}
let mut attempts = self.wasm_webrtc_carrier_attempts.borrow_mut();
let attempt = attempts.get_mut(connection_id).filter(|attempt| {
attempt.bootstrap.upgrade_id == upgrade_id
&& attempt.role == "initiator"
&& attempt.remote_ready
&& !attempt.completion_started
})?;
attempt.completion_started = true;
Some(attempt.clone())
}
fn spawn_inbound_wasm_webrtc_carrier_completion(&self, attempt: WasmWebRtcCarrierAttempt) {
let inner = self.inner.clone();
let attempts = self.wasm_webrtc_carrier_attempts.clone();
let sessions = self.wasm_webrtc_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
let kind = crate::client::IrohPathKind::IrohWebRtc;
let node = inner.iroh_node.read().await.as_ref().cloned();
let result = async {
let node = node.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
let candidate = node
.wait_for_inbound_replacement_candidate(
attempt.remote_endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
std::time::Duration::from_secs(40),
)
.await?;
let proof = inner.wasm_candidate_proof_probe(
&attempt.connection_id,
&attempt.bootstrap.upgrade_id,
attempt.bootstrap.base.transport_generation,
attempt.bootstrap.base.route_generation,
kind,
crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
)?;
let (send, probe, ack) = inner
.receive_inbound_wasm_carrier_candidate_proof(
&attempt.connection_id,
&candidate,
&proof,
)
.await?;
inner
.send_inbound_wasm_carrier_candidate_ack(send, &ack)
.await?;
let (commit_send, committed) = inner
.receive_inbound_wasm_carrier_commit(
&attempt.connection_id,
&candidate,
&probe,
)
.await?;
let incumbent = inner
.commit_proven_wasm_carrier_candidate(
&attempt.connection_id,
&attempt.bootstrap.upgrade_id,
attempt.generation,
kind,
candidate,
)
.await?;
inner
.send_inbound_wasm_carrier_candidate_ack(commit_send, &committed)
.await?;
incumbent.close(0u8.into(), b"wasm-custom-transport-upgrade");
Ok::<(), anyhow::Error>(())
}
.await;
if let Some(expires_at_ms) = attempt.inbound_authorization_expires_at_ms {
if let Some(node) = inner.iroh_node.read().await.as_ref().cloned() {
node.revoke_inbound_replacement_if_current(
attempt.remote_endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
expires_at_ms,
)
.await;
}
}
if let Err(error) = result {
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OpenRTC][WebRTC carrier] inbound candidate completion failed connection_id={} upgrade_id={} error={error:#}",
attempt.connection_id, attempt.bootstrap.upgrade_id,
)));
fail_wasm_webrtc_attempt(
inner,
attempts,
sessions,
handler,
attempt,
"carrier-proof-failed",
true,
)
.await;
return;
}
inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await;
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "selected",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"upgradeId": attempt.bootstrap.upgrade_id,
"family": "iroh",
"carrier": "webrtc",
"implementation": "iroh-carrier",
"transportGeneration": attempt.generation.transport_generation.saturating_add(1),
"routeGeneration": 0,
}),
);
});
}
}
#[cfg(feature = "iroh-transport-moq")]
fn wasm_moq_carrier_namespaces(
local_endpoint_id: &str,
remote_endpoint_id: &str,
carrier_session_id: &str,
) -> (String, String, &'static str) {
let (first, second) = if local_endpoint_id <= remote_endpoint_id {
(local_endpoint_id, remote_endpoint_id)
} else {
(remote_endpoint_id, local_endpoint_id)
};
let base = format!("openrtc/iroh-carrier/moq/{first}/{second}/{carrier_session_id}");
(
format!("{base}/from/{local_endpoint_id}"),
format!("{base}/from/{remote_endpoint_id}"),
"iroh-packets",
)
}
#[cfg(feature = "iroh-transport-moq")]
async fn fail_wasm_moq_attempt(
inner: Arc<Client>,
attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
sessions: Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
handler: Rc<RefCell<Option<js_sys::Function>>>,
attempt: WasmMoqCarrierAttempt,
failure_code: &'static str,
notify_peer: bool,
) {
let kind = crate::client::IrohPathKind::IrohMoq;
if !inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await
{
return;
}
if attempts
.borrow()
.get(&attempt.connection_id)
.is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
{
attempts.borrow_mut().remove(&attempt.connection_id);
}
sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
if notify_peer {
if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
&attempt.bootstrap,
failure_code,
) {
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "send-control",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": failed,
}),
);
}
}
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "retire-moq",
"connectionId": attempt.connection_id,
"upgradeId": attempt.bootstrap.upgrade_id,
"failureCode": failure_code,
}),
);
if !attempt.external_fallback_allowed && attempt.role == "initiator" {
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "advance-carrier",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"failedRoute": "iroh-moq",
}),
);
}
if attempt.external_fallback_allowed
&& attempt.role == "initiator"
&& matches!(
inner
.iroh_path_kind(&attempt.remote_endpoint_id.to_string())
.await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
)
{
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "fallback-external-moq",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"failureCode": failure_code,
}),
);
}
}
#[cfg(feature = "iroh-transport-moq")]
impl WasmClient {
fn schedule_wasm_moq_carrier_watchdog(
&self,
bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
) {
let inner = self.inner.clone();
let attempts = self.wasm_moq_carrier_attempts.clone();
let sessions = self.wasm_moq_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
gloo_timers::future::sleep(std::time::Duration::from_secs(30)).await;
let attempt = attempts
.borrow()
.values()
.find(|attempt| attempt.bootstrap.upgrade_id == bootstrap.upgrade_id)
.cloned();
if let Some(attempt) = attempt {
if !inner
.wasm_carrier_upgrade_is_current(
&attempt.connection_id,
crate::client::IrohPathKind::IrohMoq,
&attempt.bootstrap.upgrade_id,
)
.await
{
return;
}
fail_wasm_moq_attempt(
inner,
attempts,
sessions,
handler,
attempt,
"carrier-timeout",
true,
)
.await;
}
});
}
async fn fail_wasm_moq_carrier_attempt(
&self,
attempt: WasmMoqCarrierAttempt,
failure_code: &'static str,
notify_peer: bool,
) {
fail_wasm_moq_attempt(
self.inner.clone(),
self.wasm_moq_carrier_attempts.clone(),
self.wasm_moq_carrier_sessions.clone(),
self.wasm_carrier_action_handler.clone(),
attempt,
failure_code,
notify_peer,
)
.await;
}
fn spawn_outbound_wasm_moq_carrier_completion(&self, attempt: WasmMoqCarrierAttempt) {
let inner = self.inner.clone();
let attempts = self.wasm_moq_carrier_attempts.clone();
let sessions = self.wasm_moq_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
let kind = crate::client::IrohPathKind::IrohMoq;
let result = inner
.complete_outbound_wasm_carrier_upgrade(
&attempt.connection_id,
&attempt.remote_endpoint_id.to_string(),
&attempt.bootstrap.upgrade_id,
attempt.generation,
kind,
)
.await;
if let Err(error) = result {
let failure_code = wasm_carrier_failure_code(&error);
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OpenRTC][MoQ carrier] outbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
)));
fail_wasm_moq_attempt(
inner,
attempts,
sessions,
handler,
attempt,
failure_code,
true,
)
.await;
return;
}
inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await;
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "selected",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"upgradeId": attempt.bootstrap.upgrade_id,
"family": "iroh",
"carrier": "moq",
"implementation": "iroh-carrier",
"transportGeneration": attempt.generation.transport_generation.saturating_add(1),
"routeGeneration": 0,
}),
);
});
}
fn spawn_inbound_wasm_moq_carrier_completion(&self, attempt: WasmMoqCarrierAttempt) {
let inner = self.inner.clone();
let attempts = self.wasm_moq_carrier_attempts.clone();
let sessions = self.wasm_moq_carrier_sessions.clone();
let handler = self.wasm_carrier_action_handler.clone();
spawn_local(async move {
let kind = crate::client::IrohPathKind::IrohMoq;
let node = inner.iroh_node.read().await.as_ref().cloned();
let result = async {
let node = node.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
let candidate = node
.wait_for_inbound_replacement_candidate(
attempt.remote_endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
std::time::Duration::from_secs(40),
)
.await?;
let proof = inner.wasm_candidate_proof_probe(
&attempt.connection_id,
&attempt.bootstrap.upgrade_id,
attempt.bootstrap.base.transport_generation,
attempt.bootstrap.base.route_generation,
kind,
crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
)?;
let (send, probe, ack) = inner
.receive_inbound_wasm_carrier_candidate_proof(
&attempt.connection_id,
&candidate,
&proof,
)
.await?;
inner
.send_inbound_wasm_carrier_candidate_ack(send, &ack)
.await?;
let (commit_send, committed) = inner
.receive_inbound_wasm_carrier_commit(
&attempt.connection_id,
&candidate,
&probe,
)
.await?;
let incumbent = inner
.commit_proven_wasm_carrier_candidate(
&attempt.connection_id,
&attempt.bootstrap.upgrade_id,
attempt.generation,
kind,
candidate,
)
.await?;
inner
.send_inbound_wasm_carrier_candidate_ack(commit_send, &committed)
.await?;
incumbent.close(0u8.into(), b"wasm-custom-transport-upgrade");
Ok::<(), anyhow::Error>(())
}
.await;
if let Some(expires_at_ms) = attempt.inbound_authorization_expires_at_ms {
if let Some(node) = inner.iroh_node.read().await.as_ref().cloned() {
node.revoke_inbound_replacement_if_current(
attempt.remote_endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
expires_at_ms,
)
.await;
}
}
if let Err(error) = result {
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OpenRTC][MoQ carrier] inbound candidate completion failed connection_id={} upgrade_id={} error={error:#}",
attempt.connection_id, attempt.bootstrap.upgrade_id,
)));
fail_wasm_moq_attempt(
inner,
attempts,
sessions,
handler,
attempt,
"carrier-proof-failed",
true,
)
.await;
return;
}
inner
.retire_wasm_carrier_upgrade(
&attempt.connection_id,
kind,
&attempt.bootstrap.upgrade_id,
)
.await;
emit_wasm_carrier_action(
&handler,
serde_json::json!({
"type": "selected",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"upgradeId": attempt.bootstrap.upgrade_id,
"family": "iroh",
"carrier": "moq",
"implementation": "iroh-carrier",
"transportGeneration": attempt.generation.transport_generation.saturating_add(1),
"routeGeneration": 0,
}),
);
});
}
async fn handle_wasm_moq_bootstrap(
&self,
connection_id: String,
remote_endpoint_id: String,
bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
) -> Result<bool, JsValue> {
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let kind = crate::client::IrohPathKind::IrohMoq;
match bootstrap.action {
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
let local_endpoint_id = self
.inner
.current_node_id()
.await
.ok_or_else(|| JsValue::from_str("local endpoint id is unavailable"))?;
if local_endpoint_id.as_str() >= remote_endpoint_id.as_str()
|| !self.inner.is_iroh_moq_carrier_enabled().await
{
return Ok(true);
}
if !matches!(
self.inner.iroh_path_kind(&remote_endpoint_id).await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
) {
return Ok(true);
}
self.inner
.get_connection(endpoint_id)
.await
.ok_or_else(|| JsValue::from_str("MoQ carrier base is unavailable"))?;
let generation = self
.inner
.current_wasm_peer_data_generation(&connection_id, None)
.await
.ok_or_else(|| {
JsValue::from_str("MoQ carrier generation is unavailable")
})?;
if !self
.inner
.reserve_wasm_carrier_upgrade(&connection_id, kind, &bootstrap.upgrade_id)
.await
{
return Ok(true);
}
let node = self
.inner
.iroh_node
.read()
.await
.as_ref()
.cloned()
.ok_or_else(|| JsValue::from_str("Iroh node is unavailable"))?;
let authorization_expiry = node
.authorize_pending_inbound_replacement(
endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
std::time::Duration::from_secs(45),
)
.await;
let fallback_allowed = self.inner.wasm_moq_external_fallback_allowed().await
&& self
.wasm_moq_remote_external
.borrow()
.get(&connection_id)
.copied()
.unwrap_or(false);
let previous = self.wasm_moq_carrier_attempts.borrow_mut().insert(
connection_id.clone(),
WasmMoqCarrierAttempt {
connection_id: connection_id.clone(),
remote_endpoint_id: endpoint_id,
bootstrap: bootstrap.clone(),
generation,
role: "responder",
external_fallback_allowed: fallback_allowed,
inbound_authorization_expires_at_ms: Some(authorization_expiry),
},
);
if let Some(previous) = previous {
self.wasm_moq_carrier_sessions
.borrow_mut()
.remove(&previous.bootstrap.upgrade_id);
}
let (publish_namespace, subscribe_namespace, track_name) =
wasm_moq_carrier_namespaces(
&local_endpoint_id,
&remote_endpoint_id,
&bootstrap.carrier_session_id,
);
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "prepare-moq",
"connectionId": connection_id,
"remoteEndpointId": remote_endpoint_id,
"role": "responder",
"upgradeId": bootstrap.upgrade_id,
"carrierSessionId": bootstrap.carrier_session_id,
"transportGeneration": generation.transport_generation.saturating_add(1),
"publishNamespace": publish_namespace,
"subscribeNamespace": subscribe_namespace,
"trackName": track_name,
}),
);
self.schedule_wasm_moq_carrier_watchdog(bootstrap);
}
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
let attempt = self
.wasm_moq_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
.cloned();
if let Some(attempt) = attempt {
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "activate-moq",
"connectionId": attempt.connection_id,
"upgradeId": attempt.bootstrap.upgrade_id,
}),
);
}
}
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
let attempt = self
.wasm_moq_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
.cloned();
if let Some(attempt) = attempt {
self.fail_wasm_moq_carrier_attempt(attempt, "peer-rejected-carrier", false)
.await;
}
}
}
Ok(true)
}
}
#[wasm_bindgen]
impl WasmClient {
#[cfg(feature = "legacy-v1")]
#[deprecated(
since = "2.0.0",
note = "rollback-only: use WasmClient.newV2(apiKey) through the openrtc/runtime adapter"
)]
#[wasm_bindgen(constructor)]
pub fn new(project_id: String, tag: String) -> Result<WasmClient, JsValue> {
Self::new_with_app_tag(project_id, tag)
}
#[cfg(feature = "legacy-v1")]
#[deprecated(
since = "2.0.0",
note = "rollback-only: use WasmClient.newV2(apiKey) through the openrtc/runtime adapter"
)]
#[wasm_bindgen(js_name = newWithAppTag)]
pub fn new_with_app_tag(
project_id: String,
app_tag: String,
) -> Result<WasmClient, JsValue> {
let identity_credential: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let token_state = identity_credential.clone();
let token_provider = Box::new(move || -> Option<String> {
token_state.lock().ok().and_then(|guard| guard.clone())
});
Ok(Self {
inner: Arc::new(Client::new_with_app_tag(
project_id,
app_tag,
token_provider,
)),
identity_credential,
last_auth_log: Arc::new(Mutex::new(None)),
#[cfg(feature = "iroh-protocols-wasm")]
persistent_protocols: Rc::new(tokio::sync::Mutex::new(None)),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_remote_external: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_carrier_action_handler: Rc::new(RefCell::new(None)),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_remote_external: Rc::new(RefCell::new(HashMap::new())),
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
wasm_remote_carrier_capabilities: Rc::new(RefCell::new(HashMap::new())),
})
}
#[wasm_bindgen(js_name = newV2)]
pub fn new_v2(api_key: String) -> Result<WasmClient, JsValue> {
let api_key = crate::validate_v2_public_api_key(&api_key)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let app_tag = crate::app_tag_from_api_key(api_key);
let identity_credential: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
let credential_state = identity_credential.clone();
let identity_credential_provider = Box::new(move || -> Option<String> {
credential_state.lock().ok().and_then(|guard| guard.clone())
});
Ok(Self {
inner: Arc::new(Client::new_provider_neutral(
app_tag,
identity_credential_provider,
)),
identity_credential,
last_auth_log: Arc::new(Mutex::new(None)),
#[cfg(feature = "iroh-protocols-wasm")]
persistent_protocols: Rc::new(tokio::sync::Mutex::new(None)),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-webrtc")]
wasm_webrtc_remote_external: Rc::new(RefCell::new(HashMap::new())),
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
wasm_carrier_action_handler: Rc::new(RefCell::new(None)),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
#[cfg(feature = "iroh-transport-moq")]
wasm_moq_remote_external: Rc::new(RefCell::new(HashMap::new())),
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
wasm_remote_carrier_capabilities: Rc::new(RefCell::new(HashMap::new())),
})
}
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
#[wasm_bindgen(js_name = __setIrohCarrierActionHandler)]
pub fn set_iroh_carrier_action_handler(&self, handler: Option<js_sys::Function>) {
*self.wasm_carrier_action_handler.borrow_mut() = handler;
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __configureIrohWebRtcCarrier)]
pub async fn configure_iroh_webrtc_carrier(
&self,
enabled: bool,
implementation: Option<String>,
privacy_mode: bool,
) -> Result<(), JsValue> {
let implementation = if !enabled {
None
} else {
Some(match implementation.as_deref().unwrap_or("iroh-carrier") {
"external" => crate::client::TransportImplementation::External,
"iroh" | "iroh-carrier" => crate::client::TransportImplementation::IrohCarrier,
"auto" => crate::client::TransportImplementation::Auto,
_ => {
return Err(JsValue::from_str(
"WebRTC implementation must be external, iroh-carrier, or auto",
))
}
})
};
self.inner
.configure_wasm_webrtc_implementation(implementation, privacy_mode)
.await;
Ok(())
}
#[cfg(any(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
#[wasm_bindgen(js_name = __configureIrohRoutePolicy)]
pub async fn configure_iroh_route_policy(
&self,
relay_only: bool,
optimize_for: Option<String>,
route_priority: JsValue,
) -> Result<(), JsValue> {
let optimize_for = match optimize_for.as_deref().unwrap_or("balanced") {
"balanced" => crate::route_policy::TransportOptimization::Balanced,
"lowest-latency" => crate::route_policy::TransportOptimization::LowestLatency,
_ => {
return Err(JsValue::from_str(
"transport optimization must be balanced or lowest-latency",
))
}
};
let route_priority = if route_priority.is_null() || route_priority.is_undefined() {
Vec::new()
} else {
serde_wasm_bindgen::from_value(route_priority)
.map_err(|error| JsValue::from_str(&error.to_string()))?
};
self.inner
.configure_wasm_route_policy(relay_only, optimize_for, route_priority)
.await;
Ok(())
}
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
#[wasm_bindgen(js_name = __beginPreferredIrohCarrier)]
pub async fn begin_preferred_iroh_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
remote_supports_webrtc: bool,
remote_supports_webrtc_external: bool,
remote_supports_moq: bool,
remote_supports_moq_external: bool,
) -> Result<bool, JsValue> {
let capabilities = WasmRemoteCarrierCapabilities {
webrtc: remote_supports_webrtc,
webrtc_external: remote_supports_webrtc_external,
moq: remote_supports_moq,
moq_external: remote_supports_moq_external,
};
self.wasm_remote_carrier_capabilities
.borrow_mut()
.insert(connection_id.clone(), capabilities);
let ranked = self
.inner
.ranked_wasm_iroh_carriers(remote_supports_webrtc, remote_supports_moq)
.await;
let has_next = ranked.len() > 1;
match ranked.first() {
Some(crate::route_policy::KnownRoute::IrohWebRtc) => {
self.begin_iroh_webrtc_carrier(
connection_id,
remote_endpoint_id,
remote_supports_webrtc_external && !has_next,
)
.await
}
Some(crate::route_policy::KnownRoute::IrohMoq) => {
self.begin_iroh_moq_carrier(
connection_id,
remote_endpoint_id,
remote_supports_moq_external && !has_next,
)
.await
}
_ => Ok(false),
}
}
#[cfg(all(feature = "iroh-transport-webrtc", feature = "iroh-transport-moq"))]
#[wasm_bindgen(js_name = __advancePreferredIrohCarrier)]
pub async fn advance_preferred_iroh_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
failed_route: String,
) -> Result<bool, JsValue> {
let Some(capabilities) = self
.wasm_remote_carrier_capabilities
.borrow()
.get(&connection_id)
.copied()
else {
return Ok(false);
};
let failed = crate::route_policy::normalize_route(&failed_route)
.ok_or_else(|| JsValue::from_str("unknown failed carrier route"))?;
let ranked = self
.inner
.ranked_wasm_iroh_carriers(capabilities.webrtc, capabilities.moq)
.await;
let Some(next_index) = ranked
.iter()
.position(|route| *route == failed)
.map(|index| index + 1)
.filter(|index| *index < ranked.len())
else {
return Ok(false);
};
let next = ranked[next_index];
let allow_external = next_index + 1 == ranked.len();
match next {
crate::route_policy::KnownRoute::IrohWebRtc => {
self.begin_iroh_webrtc_carrier(
connection_id,
remote_endpoint_id,
capabilities.webrtc_external && allow_external,
)
.await
}
crate::route_policy::KnownRoute::IrohMoq => {
self.begin_iroh_moq_carrier(
connection_id,
remote_endpoint_id,
capabilities.moq_external && allow_external,
)
.await
}
_ => Ok(false),
}
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __configureIrohMoqCarrier)]
pub async fn configure_iroh_moq_carrier(
&self,
enabled: bool,
implementation: Option<String>,
relay_url: Option<String>,
) -> Result<(), JsValue> {
let implementation = if !enabled {
None
} else {
Some(match implementation.as_deref().unwrap_or("external") {
"external" => crate::client::TransportImplementation::External,
"iroh" | "iroh-carrier" => crate::client::TransportImplementation::IrohCarrier,
"auto" => crate::client::TransportImplementation::Auto,
_ => {
return Err(JsValue::from_str(
"MoQ implementation must be external, iroh-carrier, or auto",
))
}
})
};
let relay_url = relay_url.map(|value| value.trim().to_string());
if implementation.is_some()
&& implementation.is_some_and(|value| value.allows_iroh_carrier())
&& relay_url.as_deref().unwrap_or_default().is_empty()
{
return Err(JsValue::from_str(
"MoQ Iroh carrier requires an explicit relay URL",
));
}
self.inner
.configure_wasm_moq_implementation(implementation, relay_url)
.await;
Ok(())
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __beginIrohMoqCarrier)]
pub async fn begin_iroh_moq_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
remote_supports_external: bool,
) -> Result<bool, JsValue> {
self.wasm_moq_remote_external
.borrow_mut()
.insert(connection_id.clone(), remote_supports_external);
if !self.inner.is_iroh_moq_carrier_enabled().await {
return Ok(false);
}
let local_endpoint_id = self
.inner
.current_node_id()
.await
.ok_or_else(|| JsValue::from_str("local Iroh endpoint id is unavailable"))?;
if local_endpoint_id.as_str() <= remote_endpoint_id.as_str()
|| !matches!(
self.inner.iroh_path_kind(&remote_endpoint_id).await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
)
{
return Ok(false);
}
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
self.inner
.get_connection(endpoint_id)
.await
.ok_or_else(|| JsValue::from_str("MoQ carrier base is unavailable"))?;
let generation = self
.inner
.current_wasm_peer_data_generation(&connection_id, None)
.await
.ok_or_else(|| JsValue::from_str("MoQ carrier generation is unavailable"))?;
let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
crate::iroh_carrier_bootstrap::CarrierGenerationFence {
transport_stable_id: generation.transport_stable_id,
transport_generation: generation.transport_generation,
route_generation: generation.route_generation,
},
1,
)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let kind = crate::client::IrohPathKind::IrohMoq;
if !self
.inner
.reserve_wasm_carrier_upgrade(&connection_id, kind, &bootstrap.upgrade_id)
.await
{
return Ok(false);
}
let fallback_allowed =
self.inner.wasm_moq_external_fallback_allowed().await && remote_supports_external;
let previous = self.wasm_moq_carrier_attempts.borrow_mut().insert(
connection_id.clone(),
WasmMoqCarrierAttempt {
connection_id: connection_id.clone(),
remote_endpoint_id: endpoint_id,
bootstrap: bootstrap.clone(),
generation,
role: "initiator",
external_fallback_allowed: fallback_allowed,
inbound_authorization_expires_at_ms: None,
},
);
if let Some(previous) = previous {
self.wasm_moq_carrier_sessions
.borrow_mut()
.remove(&previous.bootstrap.upgrade_id);
}
let (publish_namespace, subscribe_namespace, track_name) = wasm_moq_carrier_namespaces(
&local_endpoint_id,
&remote_endpoint_id,
&bootstrap.carrier_session_id,
);
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "prepare-moq",
"connectionId": connection_id,
"remoteEndpointId": remote_endpoint_id,
"role": "initiator",
"upgradeId": bootstrap.upgrade_id,
"carrierSessionId": bootstrap.carrier_session_id,
"transportGeneration": generation.transport_generation.saturating_add(1),
"publishNamespace": publish_namespace,
"subscribeNamespace": subscribe_namespace,
"trackName": track_name,
}),
);
self.schedule_wasm_moq_carrier_watchdog(bootstrap);
Ok(true)
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __irohMoqCarrierPrepared)]
pub async fn iroh_moq_carrier_prepared(
&self,
connection_id: String,
upgrade_id: String,
) -> Result<(), JsValue> {
let attempt = self
.wasm_moq_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
.cloned()
.ok_or_else(|| JsValue::from_str("MoQ carrier attempt is stale"))?;
if !self
.inner
.wasm_carrier_upgrade_is_current(
&connection_id,
crate::client::IrohPathKind::IrohMoq,
&upgrade_id,
)
.await
{
return Err(JsValue::from_str("MoQ carrier attempt was retired"));
}
if attempt.role == "initiator" {
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "send-control",
"connectionId": connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": attempt.bootstrap,
}),
);
} else {
let ready = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
&attempt.bootstrap,
)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "send-control",
"connectionId": connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": ready,
}),
);
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "activate-moq",
"connectionId": attempt.connection_id,
"upgradeId": upgrade_id,
}),
);
}
Ok(())
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __irohMoqCarrierFailed)]
pub async fn iroh_moq_carrier_failed(
&self,
connection_id: String,
upgrade_id: String,
failure_code: String,
) {
let attempt = self
.wasm_moq_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
.cloned();
if let Some(attempt) = attempt {
let failure_code = match failure_code.as_str() {
"relay-failed" => "relay-failed",
"draft14-datagram-unavailable" => "draft14-datagram-unavailable",
"carrier-backpressure" => "carrier-backpressure",
_ => "browser-adapter-failed",
};
self.fail_wasm_moq_carrier_attempt(attempt, failure_code, true)
.await;
}
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __retireIrohMoqCarrier)]
pub async fn retire_iroh_moq_carrier(
&self,
connection_id: String,
terminal_reason: Option<String>,
) {
self.retire_iroh_moq_carrier_with_reason(connection_id, terminal_reason.as_deref())
.await;
}
#[cfg(feature = "iroh-transport-moq")]
async fn retire_iroh_moq_carrier_with_reason(
&self,
connection_id: String,
terminal_reason: Option<&str>,
) {
let attempt = self
.wasm_moq_carrier_attempts
.borrow_mut()
.remove(&connection_id);
if let Some(attempt) = attempt {
self.inner
.retire_wasm_carrier_upgrade(
&connection_id,
crate::client::IrohPathKind::IrohMoq,
&attempt.bootstrap.upgrade_id,
)
.await;
let carrier = self
.wasm_moq_carrier_sessions
.borrow_mut()
.remove(&attempt.bootstrap.upgrade_id);
if let (Some(reason), Some(session)) = (terminal_reason, carrier.as_ref()) {
let _ = session.send_terminal(reason).await;
}
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "retire-moq",
"connectionId": connection_id,
"upgradeId": attempt.bootstrap.upgrade_id,
"failureCode": "logical-connection-retired",
}),
);
}
self.wasm_moq_remote_external
.borrow_mut()
.remove(&connection_id);
#[cfg(feature = "iroh-transport-webrtc")]
self.wasm_remote_carrier_capabilities
.borrow_mut()
.remove(&connection_id);
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __beginIrohWebRtcCarrier)]
pub async fn begin_iroh_webrtc_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
remote_supports_external: bool,
) -> Result<bool, JsValue> {
self.begin_iroh_webrtc_carrier_attempt(
connection_id,
remote_endpoint_id,
remote_supports_external,
0,
)
.await
}
async fn begin_iroh_webrtc_carrier_attempt(
&self,
connection_id: String,
remote_endpoint_id: String,
remote_supports_external: bool,
retry_count: u8,
) -> Result<bool, JsValue> {
self.wasm_webrtc_remote_external
.borrow_mut()
.insert(connection_id.clone(), remote_supports_external);
if !self.inner.is_iroh_webrtc_carrier_enabled().await {
return Ok(false);
}
let local_endpoint_id = self
.inner
.current_node_id()
.await
.ok_or_else(|| JsValue::from_str("local Iroh endpoint id is unavailable"))?;
if local_endpoint_id.as_str() <= remote_endpoint_id.as_str()
|| !matches!(
self.inner.iroh_path_kind(&remote_endpoint_id).await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
)
{
return Ok(false);
}
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
self.inner
.get_connection(endpoint_id)
.await
.ok_or_else(|| JsValue::from_str("WebRTC carrier base is unavailable"))?;
let generation = self
.inner
.current_wasm_peer_data_generation(&connection_id, None)
.await
.ok_or_else(|| JsValue::from_str("WebRTC carrier generation is unavailable"))?;
let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc,
crate::iroh_carrier_bootstrap::CarrierGenerationFence {
transport_stable_id: generation.transport_stable_id,
transport_generation: generation.transport_generation,
route_generation: generation.route_generation,
},
1,
)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let kind = crate::client::IrohPathKind::IrohWebRtc;
if !self
.inner
.reserve_wasm_carrier_upgrade(&connection_id, kind, &bootstrap.upgrade_id)
.await
{
return Ok(false);
}
let fallback_allowed = self.inner.wasm_webrtc_external_fallback_allowed().await
&& remote_supports_external;
let previous = self.wasm_webrtc_carrier_attempts.borrow_mut().insert(
connection_id.clone(),
WasmWebRtcCarrierAttempt {
connection_id: connection_id.clone(),
remote_endpoint_id: endpoint_id,
bootstrap: bootstrap.clone(),
generation,
role: "initiator",
offer_started: false,
remote_ready: false,
completion_started: false,
retry_count,
external_fallback_allowed: fallback_allowed,
inbound_authorization_expires_at_ms: None,
},
);
if let Some(previous) = previous {
self.wasm_webrtc_carrier_sessions
.borrow_mut()
.remove(&previous.bootstrap.upgrade_id);
}
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "prepare-webrtc",
"connectionId": connection_id,
"remoteEndpointId": remote_endpoint_id,
"role": "initiator",
"upgradeId": bootstrap.upgrade_id,
"carrierSessionId": bootstrap.carrier_session_id,
"transportGeneration": generation.transport_generation.saturating_add(1),
}),
);
self.schedule_wasm_webrtc_carrier_watchdog(bootstrap);
Ok(true)
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __irohWebRtcCarrierPrepared)]
pub async fn iroh_webrtc_carrier_prepared(
&self,
connection_id: String,
upgrade_id: String,
) -> Result<(), JsValue> {
let attempt = self
.wasm_webrtc_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
.cloned()
.ok_or_else(|| JsValue::from_str("WebRTC carrier attempt is stale"))?;
let kind = crate::client::IrohPathKind::IrohWebRtc;
if !self
.inner
.wasm_carrier_upgrade_is_current(&connection_id, kind, &upgrade_id)
.await
{
return Err(JsValue::from_str("WebRTC carrier attempt was retired"));
}
if attempt.role == "initiator" {
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "send-control",
"connectionId": connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": attempt.bootstrap,
}),
);
} else {
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "send-control",
"connectionId": attempt.connection_id,
"remoteEndpointId": attempt.remote_endpoint_id.to_string(),
"envelope": {
"type": "#pluto-signal",
"content": {
"transport": "iroh-webrtc",
"type": "renegotiate",
"negotiationId": upgrade_id,
}
},
}),
);
}
Ok(())
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __irohWebRtcCarrierFailed)]
pub async fn iroh_webrtc_carrier_failed(
&self,
connection_id: String,
upgrade_id: String,
failure_code: String,
) {
let attempt = self
.wasm_webrtc_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
.cloned();
if let Some(attempt) = attempt {
let failure_code = match failure_code.as_str() {
"data-channel-failed" => "data-channel-failed",
"ice-failed" => "ice-failed",
"signaling-failed" => "signaling-failed",
"carrier-backpressure" => "carrier-backpressure",
_ => "browser-adapter-failed",
};
let should_retry = attempt.role == "initiator"
&& attempt.retry_count == 0
&& matches!(
failure_code,
"data-channel-failed" | "ice-failed" | "signaling-failed"
);
let remote_endpoint_id = attempt.remote_endpoint_id.to_string();
let remote_supports_external = self
.wasm_webrtc_remote_external
.borrow()
.get(&connection_id)
.copied()
.unwrap_or(false);
self.fail_wasm_webrtc_carrier_attempt(attempt, failure_code, true)
.await;
if should_retry {
gloo_timers::future::sleep(std::time::Duration::from_millis(500)).await;
let _ = self
.begin_iroh_webrtc_carrier_attempt(
connection_id,
remote_endpoint_id,
remote_supports_external,
1,
)
.await;
}
}
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __retireIrohWebRtcCarrier)]
pub async fn retire_iroh_webrtc_carrier(
&self,
connection_id: String,
terminal_reason: Option<String>,
) {
let attempt = self
.wasm_webrtc_carrier_attempts
.borrow_mut()
.remove(&connection_id);
if let Some(attempt) = attempt {
self.inner
.retire_wasm_carrier_upgrade(
&connection_id,
crate::client::IrohPathKind::IrohWebRtc,
&attempt.bootstrap.upgrade_id,
)
.await;
let carrier = self
.wasm_webrtc_carrier_sessions
.borrow_mut()
.remove(&attempt.bootstrap.upgrade_id);
if let (Some(reason), Some(session)) =
(terminal_reason.as_deref(), carrier.as_ref())
{
let _ = session.send_terminal(reason).await;
}
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "retire-webrtc",
"connectionId": connection_id,
"upgradeId": attempt.bootstrap.upgrade_id,
"failureCode": "logical-connection-retired",
}),
);
}
self.wasm_webrtc_remote_external
.borrow_mut()
.remove(&connection_id);
#[cfg(feature = "iroh-transport-moq")]
self.wasm_remote_carrier_capabilities
.borrow_mut()
.remove(&connection_id);
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __handleIrohCarrierControl)]
pub async fn handle_iroh_carrier_control(
&self,
connection_id: String,
remote_endpoint_id: String,
frame: JsValue,
) -> Result<bool, JsValue> {
let frame: serde_json::Value = serde_wasm_bindgen::from_value(frame)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
if frame.get("type").and_then(serde_json::Value::as_str) == Some("#pluto-signal")
&& frame
.get("content")
.and_then(|content| content.get("transport"))
.and_then(serde_json::Value::as_str)
== Some("iroh-webrtc")
{
let negotiation_id = frame
.get("content")
.and_then(|content| content.get("negotiationId"))
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
let current = self
.wasm_webrtc_carrier_attempts
.borrow()
.get(&connection_id)
.is_some_and(|attempt| {
attempt.bootstrap.upgrade_id == negotiation_id
&& attempt.remote_endpoint_id.to_string() == remote_endpoint_id
});
if current {
let signal_type = frame
.get("content")
.and_then(|content| content.get("type"))
.and_then(serde_json::Value::as_str);
let start_offer = if signal_type == Some("renegotiate") {
self.wasm_webrtc_carrier_attempts
.borrow_mut()
.get_mut(&connection_id)
.filter(|attempt| {
attempt.bootstrap.upgrade_id == negotiation_id
&& attempt.role == "initiator"
&& !attempt.offer_started
})
.map(|attempt| {
attempt.offer_started = true;
})
.is_some()
} else {
false
};
if start_offer {
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "start-webrtc",
"connectionId": connection_id,
"upgradeId": negotiation_id,
}),
);
return Ok(true);
}
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "apply-webrtc-signal",
"connectionId": connection_id,
"upgradeId": negotiation_id,
"signal": frame,
}),
);
}
return Ok(true);
}
let Some(bootstrap) =
crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(&frame)
else {
return Ok(false);
};
#[cfg(feature = "iroh-transport-moq")]
if bootstrap.carrier == crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14
{
return self
.handle_wasm_moq_bootstrap(connection_id, remote_endpoint_id, bootstrap)
.await;
}
if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc {
return Ok(false);
}
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let kind = crate::client::IrohPathKind::IrohWebRtc;
match bootstrap.action {
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
let local_endpoint_id = self
.inner
.current_node_id()
.await
.ok_or_else(|| JsValue::from_str("local endpoint id is unavailable"))?;
if local_endpoint_id.as_str() >= remote_endpoint_id.as_str()
|| !self.inner.is_iroh_webrtc_carrier_enabled().await
{
return Ok(true);
}
if !matches!(
self.inner.iroh_path_kind(&remote_endpoint_id).await,
crate::client::IrohPathKind::Relay | crate::client::IrohPathKind::Unknown
) {
return Ok(true);
}
self.inner
.get_connection(endpoint_id)
.await
.ok_or_else(|| JsValue::from_str("WebRTC carrier base is unavailable"))?;
let generation = self
.inner
.current_wasm_peer_data_generation(&connection_id, None)
.await
.ok_or_else(|| {
JsValue::from_str("WebRTC carrier generation is unavailable")
})?;
if !self
.inner
.reserve_wasm_carrier_upgrade(&connection_id, kind, &bootstrap.upgrade_id)
.await
{
return Ok(true);
}
let node = self
.inner
.iroh_node
.read()
.await
.as_ref()
.cloned()
.ok_or_else(|| JsValue::from_str("Iroh node is unavailable"))?;
let authorization_expiry = node
.authorize_pending_inbound_replacement(
endpoint_id,
crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
std::time::Duration::from_secs(45),
)
.await;
let fallback_allowed = self.inner.wasm_webrtc_external_fallback_allowed().await
&& self
.wasm_webrtc_remote_external
.borrow()
.get(&connection_id)
.copied()
.unwrap_or(false);
let previous = self.wasm_webrtc_carrier_attempts.borrow_mut().insert(
connection_id.clone(),
WasmWebRtcCarrierAttempt {
connection_id: connection_id.clone(),
remote_endpoint_id: endpoint_id,
bootstrap: bootstrap.clone(),
generation,
role: "responder",
offer_started: false,
remote_ready: false,
completion_started: false,
retry_count: 0,
external_fallback_allowed: fallback_allowed,
inbound_authorization_expires_at_ms: Some(authorization_expiry),
},
);
if let Some(previous) = previous {
self.wasm_webrtc_carrier_sessions
.borrow_mut()
.remove(&previous.bootstrap.upgrade_id);
}
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "prepare-webrtc",
"connectionId": connection_id,
"remoteEndpointId": remote_endpoint_id,
"role": "responder",
"upgradeId": bootstrap.upgrade_id,
"carrierSessionId": bootstrap.carrier_session_id,
"transportGeneration": generation.transport_generation.saturating_add(1),
}),
);
self.schedule_wasm_webrtc_carrier_watchdog(bootstrap);
}
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
let upgrade_id = {
let mut attempts = self.wasm_webrtc_carrier_attempts.borrow_mut();
attempts
.get_mut(&connection_id)
.filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
.map(|attempt| {
attempt.remote_ready = true;
attempt.bootstrap.upgrade_id.clone()
})
};
let attempt = upgrade_id.as_deref().and_then(|upgrade_id| {
self.take_ready_outbound_wasm_webrtc_carrier_attempt(
&connection_id,
upgrade_id,
)
});
if let Some(attempt) = attempt {
self.spawn_outbound_wasm_webrtc_carrier_completion(attempt);
}
}
crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
let attempt = self
.wasm_webrtc_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
.cloned();
if let Some(attempt) = attempt {
self.fail_wasm_webrtc_carrier_attempt(
attempt,
"peer-rejected-carrier",
false,
)
.await;
}
}
}
Ok(true)
}
#[cfg(all(feature = "iroh-transport-moq", not(feature = "iroh-transport-webrtc")))]
#[wasm_bindgen(js_name = __handleIrohCarrierControl)]
pub async fn handle_iroh_carrier_control_moq_only(
&self,
connection_id: String,
remote_endpoint_id: String,
frame: JsValue,
) -> Result<bool, JsValue> {
let frame: serde_json::Value = serde_wasm_bindgen::from_value(frame)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let Some(bootstrap) =
crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(&frame)
else {
return Ok(false);
};
if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14
{
return Ok(false);
}
self.handle_wasm_moq_bootstrap(connection_id, remote_endpoint_id, bootstrap)
.await
}
#[cfg(feature = "iroh-transport-webrtc")]
#[wasm_bindgen(js_name = __attachIrohWebRtcCarrier)]
pub async fn attach_iroh_webrtc_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
channel: web_sys::RtcDataChannel,
upgrade_id: String,
) -> Result<(), JsValue> {
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let attempt = self
.wasm_webrtc_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| {
attempt.bootstrap.upgrade_id == upgrade_id
&& attempt.remote_endpoint_id == endpoint_id
})
.cloned()
.ok_or_else(|| JsValue::from_str("WebRTC carrier attempt is stale"))?;
if !self
.inner
.wasm_carrier_upgrade_is_current(
&connection_id,
crate::client::IrohPathKind::IrohWebRtc,
&upgrade_id,
)
.await
{
return Err(JsValue::from_str("WebRTC carrier attempt was retired"));
}
let packet_session = self
.inner
.activate_iroh_packet_carrier(
crate::iroh_carrier_kind::IrohCarrierKind::WebRtc,
endpoint_id,
)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let expected = attempt
.bootstrap
.frame_expectation()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let application_key = self
.inner
.application_crypto_key_for_connection(Some(&connection_id))
.ok_or_else(|| {
JsValue::from_str("WebRTC carrier requires the admitted application crypto key")
})?;
let terminal_client = self.inner.clone();
let terminal_connection_id = connection_id.clone();
let terminal_generation = attempt.generation.transport_generation.saturating_add(1);
let on_terminal: Rc<dyn Fn(&'static str)> = Rc::new(move |reason| {
let client = terminal_client.clone();
let connection_id = terminal_connection_id.clone();
spawn_local(async move {
let _ = client
.close_current_iroh_carrier_generation_with_reason(
&connection_id,
endpoint_id,
crate::client::IrohPathKind::IrohWebRtc,
terminal_generation,
reason,
)
.await;
});
});
let carrier = crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession::attach(
channel,
packet_session,
expected,
application_key,
on_terminal,
)?;
self.wasm_webrtc_carrier_sessions
.borrow_mut()
.insert(upgrade_id.clone(), carrier);
if attempt.role == "responder" {
let ready = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
&attempt.bootstrap,
)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
emit_wasm_carrier_action(
&self.wasm_carrier_action_handler,
serde_json::json!({
"type": "send-control",
"connectionId": connection_id,
"remoteEndpointId": remote_endpoint_id,
"envelope": ready,
}),
);
self.spawn_inbound_wasm_webrtc_carrier_completion(attempt);
} else if let Some(attempt) =
self.take_ready_outbound_wasm_webrtc_carrier_attempt(&connection_id, &upgrade_id)
{
self.spawn_outbound_wasm_webrtc_carrier_completion(attempt);
}
Ok(())
}
#[cfg(feature = "iroh-transport-moq")]
#[wasm_bindgen(js_name = __attachIrohMoqCarrier)]
pub async fn attach_iroh_moq_carrier(
&self,
connection_id: String,
remote_endpoint_id: String,
datagrams: JsValue,
upgrade_id: String,
) -> Result<(), JsValue> {
let endpoint_id = remote_endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let attempt = self
.wasm_moq_carrier_attempts
.borrow()
.get(&connection_id)
.filter(|attempt| {
attempt.bootstrap.upgrade_id == upgrade_id
&& attempt.remote_endpoint_id == endpoint_id
})
.cloned()
.ok_or_else(|| JsValue::from_str("MoQ carrier attempt is stale"))?;
if !self
.inner
.wasm_carrier_upgrade_is_current(
&connection_id,
crate::client::IrohPathKind::IrohMoq,
&upgrade_id,
)
.await
{
return Err(JsValue::from_str("MoQ carrier attempt was retired"));
}
let packet_session = self
.inner
.activate_iroh_packet_carrier(
crate::iroh_carrier_kind::IrohCarrierKind::Moq,
endpoint_id,
)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let expected = attempt
.bootstrap
.frame_expectation()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let application_key = self
.inner
.application_crypto_key_for_connection(Some(&connection_id))
.ok_or_else(|| {
JsValue::from_str("MoQ carrier requires an installed application key")
})?;
let terminal_client = self.inner.clone();
let terminal_connection_id = connection_id.clone();
let terminal_generation = attempt.generation.transport_generation.saturating_add(1);
let on_terminal: Rc<dyn Fn(&'static str)> = Rc::new(move |reason| {
let client = terminal_client.clone();
let connection_id = terminal_connection_id.clone();
spawn_local(async move {
let _ = client
.close_current_iroh_carrier_generation_with_reason(
&connection_id,
endpoint_id,
crate::client::IrohPathKind::IrohMoq,
terminal_generation,
reason,
)
.await;
});
});
let carrier = crate::wasm_moq_carrier::WasmMoqCarrierSession::attach(
datagrams,
packet_session,
expected,
application_key,
on_terminal,
)?;
self.wasm_moq_carrier_sessions
.borrow_mut()
.insert(upgrade_id, carrier);
if attempt.role == "initiator" {
self.spawn_outbound_wasm_moq_carrier_completion(attempt);
} else {
self.spawn_inbound_wasm_moq_carrier_completion(attempt);
}
Ok(())
}
#[wasm_bindgen(js_name = setIdentityCredential)]
pub fn set_identity_credential(&self, credential: Option<String>) {
let has_credential = credential
.as_ref()
.map(|value| !value.is_empty())
.unwrap_or(false);
let credential_len = credential.as_ref().map(|value| value.len()).unwrap_or(0);
if let Ok(mut guard) = self.identity_credential.lock() {
*guard = credential.filter(|value| !value.is_empty());
}
let should_log = if let Ok(mut guard) = self.last_auth_log.lock() {
let next = (has_credential, credential_len);
if guard.as_ref() == Some(&next) {
false
} else {
*guard = Some(next);
true
}
} else {
true
};
if should_log {
web_sys::console::log_1(&JsValue::from_str(&format!(
"[OPENRTC][WASM-IDENTITY] credential updated present={} len={}",
has_credential, credential_len
)));
}
}
#[wasm_bindgen(js_name = clearIdentityCredential)]
pub fn clear_identity_credential(&self) {
self.set_identity_credential(None);
}
#[cfg(feature = "legacy-v1")]
#[deprecated(since = "2.0.0", note = "rollback-only: use setIdentityCredential")]
pub fn set_auth_token(&self, token: Option<String>) {
self.set_identity_credential(token);
}
#[wasm_bindgen(js_name = rankRoutes)]
pub fn rank_routes(
&self,
configured_priority: Vec<String>,
candidates: Vec<String>,
) -> Vec<String> {
crate::route_policy::rank_routes(&configured_priority, &candidates)
}
pub async fn init_iroh(&self, secret_key: Option<Vec<u8>>) -> Result<String, JsValue> {
let started_at = js_sys::Date::now();
web_sys::console::log_1(&JsValue::from_str(&format!(
"[OPENRTC][WASM-API] init_iroh called has_secret_key={} secret_key_len={}",
secret_key.as_ref().is_some(),
secret_key.as_ref().map(|k| k.len()).unwrap_or(0)
)));
match self.inner.init_iroh(secret_key, vec![]).await {
Ok(node_id) => {
self.inner.clone().start_wasm_accept_bridge();
let elapsed = js_sys::Date::now() - started_at;
web_sys::console::log_1(&JsValue::from_str(&format!(
"[OPENRTC][WASM-API] init_iroh success elapsed_ms={:.0} node_id={}",
elapsed, node_id
)));
Ok(node_id)
}
Err(err) => {
let elapsed = js_sys::Date::now() - started_at;
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OPENRTC][WASM-API] init_iroh failed elapsed_ms={:.0} error={}",
elapsed, err
)));
Err(JsValue::from_str(&err.to_string()))
}
}
}
#[wasm_bindgen(js_name = initIrohWithTestRelay)]
pub async fn init_iroh_with_test_relay(
&self,
secret_key: Option<Vec<u8>>,
test_relay_url: Option<String>,
) -> Result<String, JsValue> {
let node_id = self
.inner
.init_iroh_with_test_relay(secret_key, vec![], test_relay_url.as_deref())
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
self.inner.clone().start_wasm_accept_bridge();
Ok(node_id)
}
pub async fn iroh_secret_key(&self) -> Result<Vec<u8>, JsValue> {
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
Ok(node.secret_key())
} else {
Err(JsValue::from_str("Iroh node not initialized"))
}
}
pub async fn node_addr(&self) -> Result<String, JsValue> {
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
let addr = node
.node_addr()
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
serde_json::to_string(&addr).map_err(|e| JsValue::from_str(&e.to_string()))
} else {
Err(JsValue::from_str("Iroh node not initialized"))
}
}
pub async fn endpoint_ticket(&self) -> Result<String, JsValue> {
self.inner
.endpoint_ticket()
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn endpoint_ticket_with_token(
&self,
grant_scope: String,
max_connections: u32,
) -> Result<String, JsValue> {
self.inner
.endpoint_ticket_with_token(&grant_scope, max_connections)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub fn register_session_token(
&self,
token: String,
grant_scope: String,
max_connections: u32,
) {
self.inner
.register_session_token(token, grant_scope, max_connections);
}
pub fn register_session_token_with_expiry_ms(
&self,
token: String,
grant_scope: String,
max_connections: u32,
expires_at_ms: u64,
) {
self.inner.register_session_token_with_expiry_ms(
token,
grant_scope,
max_connections,
expires_at_ms,
);
}
#[wasm_bindgen(js_name = setConnectionApplicationCryptoRequired)]
pub fn set_connection_application_crypto_required(
&self,
connection_id: String,
) -> Result<(), JsValue> {
self.inner
.set_connection_application_crypto_required(&connection_id);
Ok(())
}
#[wasm_bindgen(js_name = setConnectionApplicationCryptoKey)]
pub async fn set_connection_application_crypto_key(
&self,
connection_id: String,
key: Vec<u8>,
) -> Result<(), JsValue> {
if key.len() != crate::application_crypto::APPLICATION_KEY_BYTES {
return Err(JsValue::from_str("application crypto key must be 32 bytes"));
}
let mut key_bytes = [0u8; crate::application_crypto::APPLICATION_KEY_BYTES];
key_bytes.copy_from_slice(&key);
self.inner
.set_connection_application_crypto_key(&connection_id, key_bytes);
self.inner
.emit_current_wasm_connection_state(&connection_id)
.await;
Ok(())
}
#[wasm_bindgen(js_name = clearConnectionApplicationCryptoKey)]
pub fn clear_connection_application_crypto_key(&self, connection_id: String) {
self.inner
.clear_connection_application_crypto_key(&connection_id);
}
pub fn validate_session_token(&self, token: String) -> Result<String, JsValue> {
self.inner
.validate_session_token(&token)
.map_err(|e| JsValue::from_str(&e))
}
pub async fn validate_session_token_for_connection(
&self,
token: String,
connection_id: String,
) -> Result<String, JsValue> {
self.inner
.validate_session_token_for_connection(&token, &connection_id)
.await
.map_err(|e| JsValue::from_str(&e))
}
pub async fn validate_session_token_for_connection_with_payload(
&self,
token: String,
connection_id: String,
token_payload: Option<String>,
) -> Result<String, JsValue> {
self.inner
.validate_session_token_for_connection_with_payload(
&token,
&connection_id,
token_payload.as_deref(),
)
.await
.map_err(|e| JsValue::from_str(&e))
}
pub async fn present_session_token_to_host(
&self,
endpoint_id: String,
token: String,
) -> Result<String, JsValue> {
self.present_session_token_to_host_with_payload(endpoint_id, token, None)
.await
}
pub async fn present_session_token_to_host_with_payload(
&self,
endpoint_id: String,
token: String,
token_payload: Option<String>,
) -> Result<String, JsValue> {
self.present_session_token_to_host_with_payload_and_device_id(
endpoint_id,
token,
token_payload,
None,
)
.await
}
pub async fn present_session_token_to_host_with_payload_and_device_id(
&self,
endpoint_id: String,
token: String,
token_payload: Option<String>,
device_id: Option<String>,
) -> Result<String, JsValue> {
crate::console_log!(
"[OpenRTC][session-admission][wasm-present] endpoint_id={} claimed_local_device_id={}",
endpoint_id,
device_id.as_deref().unwrap_or("<none>")
);
let endpoint_id_parsed: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
let local_node_id = self.inner.current_node_id().await.ok_or_else(|| {
JsValue::from_str("missing local node id for session-token presentation")
})?;
let connection_id =
crate::client::Client::deterministic_connection_id(&local_node_id, &endpoint_id);
let approval_scope = self
.inner
.present_and_accept_session_token_with_local_claim(
endpoint_id_parsed,
&connection_id,
&token,
token_payload.as_deref(),
None,
device_id,
)
.await
.map_err(|e| JsValue::from_str(&e))?;
self.inner
.emit_current_wasm_connection_state(&connection_id)
.await;
Ok(approval_scope)
}
pub async fn remote_session_admission_ready_for_ticket(
&self,
endpoint_ticket: String,
) -> Result<bool, JsValue> {
self.inner
.remote_session_admission_ready_for_ticket(&endpoint_ticket)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))
}
#[allow(clippy::too_many_arguments)]
pub async fn prepare_inline_reciprocal_session_admission(
&self,
endpoint_id: String,
expected_transport_stable_id: u64,
stream_instance_id: String,
presentation_id: String,
token: String,
token_payload: String,
device_id: String,
stream_contract: String,
) -> Result<bool, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|error| JsValue::from_str(&format!("{error}")))?;
let stream_contract = match stream_contract.trim() {
"one-shot-admission" => {
crate::native_protocol::SessionTokenStreamContract::OneShotAdmission
}
"persistent-control" => {
crate::native_protocol::SessionTokenStreamContract::PersistentControl
}
other => {
return Err(JsValue::from_str(&format!(
"unsupported reciprocal stream contract: {other}"
)))
}
};
self.inner
.prepare_inline_reciprocal_session_admission(
endpoint_id,
expected_transport_stable_id,
stream_instance_id.as_str(),
presentation_id.as_str(),
token.as_str(),
token_payload.as_str(),
device_id.as_str(),
stream_contract,
)
.await
.map_err(|error| JsValue::from_str(&error))?;
Ok(true)
}
pub async fn confirm_inline_reciprocal_session_admission(
&self,
endpoint_id: String,
expected_transport_stable_id: u64,
stream_instance_id: String,
presentation_id: String,
accepted: bool,
approval_scope: Option<String>,
) -> Result<bool, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|error| JsValue::from_str(&format!("{error}")))?;
self.inner
.confirm_inline_reciprocal_session_admission(
endpoint_id,
expected_transport_stable_id,
stream_instance_id.as_str(),
presentation_id.as_str(),
accepted,
approval_scope.as_deref(),
)
.await
.map_err(|error| JsValue::from_str(&error))?;
self.inner
.emit_current_wasm_connection_state(
&crate::client::Client::deterministic_connection_id(
&self.inner.current_node_id().await.ok_or_else(|| {
JsValue::from_str(
"missing local node id after reciprocal admission ACK",
)
})?,
&endpoint_id.to_string(),
),
)
.await;
Ok(true)
}
pub fn revoke_session_token(&self, token: String) -> Result<JsValue, JsValue> {
serde_wasm_bindgen::to_value(&self.inner.revoke_session_token(&token))
.map_err(|error| JsValue::from_str(&error.to_string()))
}
pub async fn revoke_tokens_by_scope(
&self,
grant_scope: String,
) -> Result<JsValue, JsValue> {
let affected = self.inner.begin_revoke_tokens_by_scope(&grant_scope);
for connection_id in &affected {
#[cfg(feature = "iroh-transport-webrtc")]
self.retire_iroh_webrtc_carrier(
connection_id.clone(),
Some(crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED.to_string()),
)
.await;
#[cfg(feature = "iroh-transport-moq")]
self.retire_iroh_moq_carrier_with_reason(
connection_id.clone(),
Some(crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED),
)
.await;
}
self.inner
.finish_revoke_tokens_by_scope(&grant_scope, &affected)
.await;
serde_wasm_bindgen::to_value(&affected).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub fn clear_session_tokens(&self) {
self.inner.clear_session_tokens();
}
pub fn endpoint_id_from_ticket(&self, ticket: String) -> Result<String, JsValue> {
let (iroh_ticket, _token_suffix) = split_compound_ticket(ticket.trim());
let parsed = EndpointTicket::from_str(iroh_ticket)
.map_err(|e| JsValue::from_str(&format!("Invalid endpoint ticket: {}", e)))?;
Ok(parsed.endpoint_addr().id.to_string())
}
pub async fn connect(&self, ticket: String) -> Result<JsReadableStream, JsValue> {
web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(
"[OPENRTC][WASM-API] WasmClient.connect() is deprecated; use connect_device() for managed product dials.",
));
let (iroh_ticket, _token_suffix) = split_compound_ticket(ticket.trim());
let parsed = EndpointTicket::from_str(iroh_ticket)
.map_err(|e| JsValue::from_str(&format!("Invalid endpoint ticket: {}", e)))?;
let endpoint_addr = parsed.endpoint_addr().clone();
let endpoint_id = endpoint_addr.id;
let stream = {
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
node.connect_addr(endpoint_id, endpoint_addr)
} else {
return Err(JsValue::from_str("Iroh node not initialized"));
}
};
Ok(into_js_readable_stream(stream))
}
pub async fn disconnect(&self, endpoint_id: String) -> Result<(), JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
node.disconnect(endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
} else {
Err(JsValue::from_str("Iroh node not initialized"))
}
}
pub async fn disconnect_transient(&self, endpoint_id: String) -> Result<(), JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
self.inner
.disconnect_with_reason(
endpoint_id,
crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
self.inner.wake_browser_auto_connect();
Ok(())
}
pub async fn is_connected(&self, endpoint_id: String) -> Result<bool, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
Ok(self.inner.is_connected(endpoint_id).await)
}
pub fn runtime_policy(&self) -> Result<JsValue, JsValue> {
serde_wasm_bindgen::to_value(&self.inner.runtime_policy_snapshot())
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn add_peer_scope(&self, id: String, scope: String) -> Result<JsValue, JsValue> {
let scopes = self.inner.add_peer_scope(&id, &scope).await;
serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn release_peer_scope(
&self,
id: String,
scope: Option<String>,
) -> Result<JsValue, JsValue> {
let scopes = self.inner.release_peer_scope(&id, scope.as_deref()).await;
serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn peer_scopes(&self, id: String) -> Result<JsValue, JsValue> {
let scopes = self.inner.peer_scopes(&id).await;
serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn same_peer(&self, left: String, right: String) -> Result<bool, JsValue> {
Ok(self.inner.same_peer(&left, &right).await)
}
pub async fn peer_snapshot(&self, id: String) -> Result<JsValue, JsValue> {
let snapshot = self.inner.peer_snapshot(&id).await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn peer_session(&self, id: String) -> Result<JsValue, JsValue> {
let snapshot = self.inner.peer_session(&id).await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn peer_sessions(&self) -> Result<JsValue, JsValue> {
let snapshots = self.inner.peer_sessions().await;
serde_wasm_bindgen::to_value(&snapshots).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn connection_state(&self, connection_id: String) -> Result<JsValue, JsValue> {
let snapshot = self.inner.connection_state(&connection_id).await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn connection_states(&self) -> Result<JsValue, JsValue> {
let snapshots = self.inner.connection_states().await;
serde_wasm_bindgen::to_value(&snapshots).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn wait_for_settled_peer(
&self,
id: String,
timeout_ms: Option<u32>,
) -> Result<JsValue, JsValue> {
let snapshot = self
.inner
.wait_for_settled_peer(&id, timeout_ms.map(|value| value as u64))
.await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn resolve_peer_connection_records(
&self,
id: String,
) -> Result<JsValue, JsValue> {
let records = self.inner.resolve_peer_connection_records(&id).await;
serde_wasm_bindgen::to_value(&records).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn list_managed_connections(&self) -> Result<JsValue, JsValue> {
let records = self.inner.list_managed_connections().await;
serde_wasm_bindgen::to_value(&records).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn bind_connection_device_id(
&self,
connection_id: String,
device_id: String,
) -> Result<JsValue, JsValue> {
let snapshot = self
.inner
.bind_connection_device_id(&connection_id, &device_id)
.await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn bind_node_device_id(
&self,
node_id: String,
device_id: String,
) -> Result<(), JsValue> {
self.inner.bind_node_device_id(&node_id, &device_id).await;
Ok(())
}
pub async fn reject_connection_admission(
&self,
connection_id: String,
reason: String,
) -> Result<JsValue, JsValue> {
self.inner
.reject_session_connection(&connection_id, &reason);
self.inner
.emit_current_wasm_connection_state(&connection_id)
.await;
let snapshot = self.inner.connection_state(&connection_id).await;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn report_managed_connection_settled(
&self,
connection_id: String,
settled: bool,
device_id: Option<String>,
transport_stable_id: Option<u64>,
transport_generation: Option<u64>,
route_generation: Option<u64>,
) -> Result<JsValue, JsValue> {
let snapshot = match (transport_stable_id, transport_generation, route_generation) {
(Some(transport_stable_id), Some(transport_generation), Some(route_generation)) => {
self.inner
.report_managed_connection_settled_for_transport(
&connection_id,
settled,
transport_stable_id,
transport_generation,
route_generation,
)
.await
}
_ => None,
};
let _ = device_id;
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn report_transport_status(
&self,
connection_id: String,
active_transport: String,
parallel_transport: Option<String>,
transport_stable_id: Option<u64>,
transport_generation: Option<u64>,
route_generation: Option<u64>,
) -> Result<JsValue, JsValue> {
let snapshot = match (transport_stable_id, transport_generation, route_generation) {
(Some(transport_stable_id), Some(transport_generation), Some(route_generation)) => {
self.inner
.report_transport_status_for_generation(
&connection_id,
&active_transport,
parallel_transport.as_deref(),
transport_stable_id,
transport_generation,
route_generation,
)
.await
}
_ => None,
};
serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn is_current_transport_stable_id(
&self,
endpoint_id: String,
transport_stable_id: u64,
) -> Result<bool, JsValue> {
let endpoint_id = endpoint_id
.parse::<iroh::EndpointId>()
.map_err(|error| JsValue::from_str(&error.to_string()))?;
Ok(self
.inner
.is_current_transport_stable_id(endpoint_id, transport_stable_id)
.await)
}
pub async fn open_bi(&self, endpoint_id: String) -> Result<BiStream, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
self.inner
.assert_raw_peer_stream_allowed(&endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
self.inner
.ensure_connection_manager_record_before_peer_stream(&endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
let (send, recv) = self
.inner
.open_bi_internal(endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(BiStream::from_parts(send, recv, endpoint_id.to_string()))
}
pub async fn open_native_main_control_bi(
&self,
endpoint_id: String,
) -> Result<BiStream, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
self.inner
.ensure_connection_manager_record_before_peer_stream(&endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
let (send, recv) = self
.inner
.open_bi_internal(endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(BiStream::from_parts(send, recv, endpoint_id.to_string()))
}
pub async fn send_native_main_control_frame(
&self,
endpoint_id: String,
frame: Vec<u8>,
) -> Result<(), JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|error| JsValue::from_str(&format!("{error}")))?;
self.inner
.ensure_connection_manager_record_before_peer_stream(&endpoint_id)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let (mut send, _recv) = self
.inner
.open_bi_internal(endpoint_id)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let label = b"signal";
send.write_all(&[0x00])
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
send.write_all(&(label.len() as u32).to_be_bytes())
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
send.write_all(label)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
send.write_all(&frame)
.await
.map_err(|error| JsValue::from_str(&error.to_string()))?;
crate::application_crypto_streams::PeerSendStream::plain(send)
.finish_and_wait_for_peer(std::time::Duration::from_secs(2))
.await
.map_err(|error| JsValue::from_str(&error.to_string()))
}
pub async fn open_peer_bi(
&self,
id: String,
timeout_ms: Option<u32>,
) -> Result<BiStream, JsValue> {
let (_connection_id, remote_node_id, send, recv) = self
.inner
.open_peer_bi(&id, timeout_ms.map(|value| value as u64))
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(BiStream::from_peer_parts(send, recv, remote_node_id))
}
pub async fn send_peer_application_frame(
&self,
id: String,
frame: Vec<u8>,
timeout_ms: Option<u32>,
) -> Result<(), JsValue> {
self.inner
.send_peer_application_frame(&id, &frame, timeout_ms.map(|value| value as u64))
.await
.map_err(|error| JsValue::from_str(&error.to_string()))
}
pub async fn open_peer_bi_explicit_file_sender(
&self,
id: String,
timeout_ms: Option<u32>,
) -> Result<PeerUniStream, JsValue> {
let (_connection_id, _remote_node_id, send) = self
.inner
.open_peer_bi_explicit_file_sender(&id, timeout_ms.map(|value| value as u64))
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(peer_uni_stream_from_send(send))
}
pub async fn open_peer_bi_transport_only(
&self,
id: String,
timeout_ms: Option<u32>,
) -> Result<BiStream, JsValue> {
let (_connection_id, remote_node_id, send, recv) = self
.inner
.open_peer_bi_transport_only(&id, timeout_ms.map(|value| value as u64))
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(BiStream::from_parts(send, recv, remote_node_id))
}
pub async fn open_peer_native_bi(
&self,
id: String,
label: String,
timeout_ms: Option<u32>,
) -> Result<BiStream, JsValue> {
if label != "drive-view" {
return Err(JsValue::from_str(
"unsupported native peer stream label; only drive-view is allowed",
));
}
let (_connection_id, remote_node_id, mut send, recv) = self
.inner
.open_peer_bi(&id, timeout_ms.map(|value| value as u64))
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
let envelope = crate::stream_metadata::encode_channel_envelope(&label, None)
.map_err(|e| JsValue::from_str(&e.to_string()))?;
send.write_all(&envelope)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(BiStream::from_peer_parts(send, recv, remote_node_id))
}
pub async fn open_uni(&self, endpoint_id: String) -> Result<PeerUniStream, JsValue> {
let endpoint_id: iroh::EndpointId = endpoint_id
.parse()
.map_err(|e| JsValue::from_str(&format!("{}", e)))?;
self.inner
.assert_raw_peer_stream_allowed(&endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
self.inner
.ensure_connection_manager_record_before_peer_stream(&endpoint_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
let send = node
.open_uni(endpoint_id.clone())
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(peer_uni_stream_from_send(
crate::application_crypto_streams::PeerSendStream::plain(send),
))
} else {
Err(JsValue::from_str("Iroh node not initialized"))
}
}
pub async fn open_peer_uni(
&self,
id: String,
timeout_ms: Option<u32>,
) -> Result<PeerUniStream, JsValue> {
let (_connection_id, _remote_node_id, send) = self
.inner
.open_peer_uni(&id, timeout_ms.map(|value| value as u64))
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
Ok(peer_uni_stream_from_send(send))
}
pub async fn send_peer(&self, id: String, data: Vec<u8>) -> Result<(), JsValue> {
self.inner
.send_peer(&id, &data)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn iroh_path_kind(&self, peer_id: String) -> String {
match self.inner.iroh_path_kind(&peer_id).await {
crate::client::IrohPathKind::DirectQuic => "direct-quic".to_string(),
crate::client::IrohPathKind::DirectLan => "direct-lan".to_string(),
crate::client::IrohPathKind::Relay => "relay".to_string(),
crate::client::IrohPathKind::Ble => "ble".to_string(),
crate::client::IrohPathKind::IrohWebRtc => "iroh-webrtc".to_string(),
crate::client::IrohPathKind::IrohMoq => "iroh-moq".to_string(),
crate::client::IrohPathKind::Unknown => "unknown".to_string(),
}
}
pub async fn iroh_transport_rtt_ms(&self, peer_id: String) -> Option<u32> {
self.inner
.iroh_transport_rtt_ms(&peer_id)
.await
.map(|value| value.min(u32::MAX as u64) as u32)
}
pub async fn incoming_streams(&self) -> Result<JsReadableStream, JsValue> {
let (node, stream) = {
let node_guard = self.inner.iroh_node.read().await;
if let Some(node) = node_guard.as_ref() {
(node.clone(), node.incoming_streams_stream())
} else {
return Err(JsValue::from_str("Iroh node not initialized"));
}
};
use futures::StreamExt;
let mapped_stream = stream.filter_map(move |incoming| {
let node = node.clone();
async move {
node.incoming_stream_is_current(&incoming)
.await
.then(|| crate::wasm_node::BiStream::incoming_to_js_value(incoming))
}
});
Ok(wasm_streams::ReadableStream::from_stream(mapped_stream).into_raw())
}
pub async fn update_presence(
&self,
user_id: String,
device_name: String,
ticket: String,
metadata: Option<String>,
ttl_ms: Option<u64>,
) -> Result<(), JsValue> {
self.inner
.update_presence_with_ttl(
&user_id,
&device_name,
&ticket,
ttl_ms.unwrap_or(300_000),
metadata.as_deref(),
)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn send_message(
&self,
target_id: String,
payload: String,
state: Option<String>,
reply_payload: Option<String>,
) -> Result<String, JsValue> {
self.inner
.send_message(
&target_id,
&payload,
state.as_deref(),
reply_payload.as_deref(),
)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn set_offline(&self, user_id: String) -> Result<(), JsValue> {
self.inner
.set_offline(&user_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn update_device(
&self,
user_id: String,
device_id: String,
device_name: Option<String>,
capabilities: Option<JsValue>,
metadata: Option<String>,
) -> Result<(), JsValue> {
let parsed_capabilities = match capabilities {
Some(value) if !value.is_null() && !value.is_undefined() => Some(
serde_wasm_bindgen::from_value::<crate::signaling::DeviceCapabilities>(value)
.map_err(|e| JsValue::from_str(&e.to_string()))?,
),
_ => None,
};
self.inner
.update_device(
&user_id,
&device_id,
device_name.as_deref(),
parsed_capabilities,
metadata.as_deref(),
)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn delete_device(
&self,
user_id: String,
device_id: String,
) -> Result<(), JsValue> {
self.inner
.delete_device(&user_id, &device_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub fn force_reconnect_snapshot(&self) {
self.inner.clone().force_reconnect_snapshot();
}
pub fn stop_presence_loop(&self) {
self.inner.stop_presence_loop();
}
pub fn stop_auto_connect(&self) {
self.inner.stop_auto_connect();
self.inner.stop_browser_auto_connect();
}
pub fn start_auto_connect(
&self,
user_id: String,
local_device_id: String,
) -> Result<(), JsValue> {
self.inner
.start_browser_auto_connect(user_id, local_device_id)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
pub fn submit_browser_desired_peers(
&self,
revision: u32,
peers_json: String,
) -> Result<bool, JsValue> {
self.inner
.submit_browser_desired_peers(u64::from(revision), &peers_json)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
pub fn wake_browser_auto_connect(&self) -> bool {
self.inner.wake_browser_auto_connect()
}
pub async fn set_auto_connect_excluded(&self, device_id: String, excluded: bool) {
if excluded {
self.inner.exclude_peer_and_publish(&device_id).await;
} else {
self.inner.unexclude_peer_and_publish(&device_id).await;
}
self.inner.wake_browser_auto_connect();
}
pub fn is_auto_connect_excluded(&self, device_id: String) -> bool {
self.inner.is_auto_connect_excluded(&device_id)
}
pub async fn disconnect_device(
&self,
device_id: String,
node_id_hint: Option<String>,
) -> Result<JsValue, JsValue> {
let retired = self
.inner
.disconnect_device(&device_id, node_id_hint.as_deref())
.await;
serde_wasm_bindgen::to_value(&retired).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub fn stop_auth_scoped_activity(&self) {
self.inner.stop_auth_scoped_activity();
self.inner.stop_browser_auto_connect();
}
pub fn start_presence_loop(
&self,
user_id: String,
device_name: String,
ticket: String,
metadata: Option<String>,
) {
self.inner
.clone()
.start_signaling_loop(user_id, device_name, ticket, metadata);
}
pub async fn search_devices(&self, user_id: String) -> Result<JsValue, JsValue> {
let devices = self
.inner
.search_devices(&user_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
serde_wasm_bindgen::to_value(&devices).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn devices_with_status(&self, user_id: String) -> Result<JsValue, JsValue> {
let devices = self
.inner
.devices_with_status(&user_id)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
serde_wasm_bindgen::to_value(&devices).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn connect_device(
&self,
device_id: Option<String>,
endpoint_ticket: String,
) -> Result<JsValue, JsValue> {
let result = self
.inner
.connect_device(device_id.as_deref(), &endpoint_ticket)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
serde_wasm_bindgen::to_value(&result).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn create_session(&self, session_json: String) -> Result<(), JsValue> {
let session: crate::signaling::SignalingSession =
serde_json::from_str(&session_json)
.map_err(|e| JsValue::from_str(&e.to_string()))?;
self.inner
.create_session(session)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn update_session(
&self,
session_id: String,
update_json: String,
) -> Result<(), JsValue> {
let update_data: serde_json::Value = serde_json::from_str(&update_json)
.map_err(|e| JsValue::from_str(&e.to_string()))?;
self.inner
.update_session(&session_id, update_data)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn create_room(
&self,
room_id: String,
user_id: String,
ticket_str: String,
my_node_id: String,
tag: String,
max_members: Option<u32>,
) -> Result<bool, JsValue> {
self.inner
.room
.create_room(
&room_id,
&user_id,
&ticket_str,
&my_node_id,
&tag,
max_members,
)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn join_room(
&self,
room_id: String,
user_id: String,
ticket_str: String,
my_node_id: String,
tag: String,
) -> Result<(), JsValue> {
self.inner
.room
.join_room(&room_id, &user_id, &ticket_str, &my_node_id, &tag)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn get_members(
&self,
room_id: String,
my_node_id: String,
tag: String,
) -> Result<String, JsValue> {
let members = self
.inner
.room
.get_members(&room_id, &my_node_id, &tag)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))?;
serde_json::to_string(&members).map_err(|e| JsValue::from_str(&e.to_string()))
}
pub async fn leave_room(
&self,
room_id: String,
my_node_id: String,
tag: String,
) -> Result<(), JsValue> {
self.inner
.room
.leave_room(&room_id, &my_node_id, &tag)
.await
.map_err(|e| JsValue::from_str(&e.to_string()))
}
}
#[cfg(feature = "iroh-protocols-wasm")]
#[wasm_bindgen]
impl WasmClient {
#[wasm_bindgen(js_name = __initPersistentIrohProtocols)]
pub async fn init_persistent_iroh_protocols(
&self,
replica_store: JsValue,
) -> Result<(), JsValue> {
let mut guard = self.persistent_protocols.lock().await;
if guard.is_some() {
return Ok(());
}
let node = self
.inner
.iroh_node
.read()
.await
.as_ref()
.cloned()
.ok_or_else(|| {
JsValue::from_str("OpenRTC endpoint must be initialized before protocols")
})?;
let store = crate::wasm_docs_persistence::JsReplicaStore::new(replica_store)
.map_err(|error| JsValue::from_str(&error.to_string()))?;
let actor = crate::wasm_docs_persistence::WasmPersistentDocsActor::hydrate(
store,
node.endpoint().clone(),
)
.await
.map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
node.install_standard_protocols(
actor.docs_protocol(),
actor.blobs_protocol(),
actor.gossip_protocol(),
)
.await
.map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
*guard = Some(actor);
Ok(())
}
#[wasm_bindgen(js_name = __importPersistentIrohAuthor)]
pub async fn import_persistent_iroh_author(
&self,
author_secret: Vec<u8>,
) -> Result<String, JsValue> {
let guard = self.persistent_protocols.lock().await;
let actor = persistent_protocols(&guard)?;
actor
.import_author(author_secret)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __importPersistentIrohNamespace)]
pub async fn import_persistent_iroh_namespace(
&self,
capability_kind: String,
capability: Vec<u8>,
generation: u64,
share_revision: u64,
) -> Result<String, JsValue> {
let mut guard = self.persistent_protocols.lock().await;
let actor = persistent_protocols_mut(&mut guard)?;
actor
.import_namespace(
capability_kind.as_str(),
capability,
generation,
share_revision,
)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __createPersistentIrohNamespace)]
pub async fn create_persistent_iroh_namespace(
&self,
generation: u64,
share_revision: u64,
) -> Result<JsValue, JsValue> {
let mut guard = self.persistent_protocols.lock().await;
let descriptor = persistent_protocols_mut(&mut guard)?
.create_namespace(generation, share_revision)
.await
.map_err(js_protocol_error)?;
serde_wasm_bindgen::to_value(&descriptor)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
#[wasm_bindgen(js_name = __importPersistentIrohTicket)]
pub async fn import_persistent_iroh_ticket(
&self,
ticket: String,
generation: u64,
share_revision: u64,
) -> Result<String, JsValue> {
let mut guard = self.persistent_protocols.lock().await;
persistent_protocols_mut(&mut guard)?
.import_ticket(&ticket, generation, share_revision)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __sharePersistentIrohNamespace)]
pub async fn share_persistent_iroh_namespace(
&self,
namespace_id: String,
writable: bool,
) -> Result<String, JsValue> {
let guard = self.persistent_protocols.lock().await;
persistent_protocols(&guard)?
.share(&namespace_id, writable)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __removePersistentIrohNamespace)]
pub async fn remove_persistent_iroh_namespace(
&self,
namespace_id: String,
generation: u64,
share_revision: u64,
) -> Result<(), JsValue> {
let mut guard = self.persistent_protocols.lock().await;
persistent_protocols_mut(&mut guard)?
.remove_namespace(&namespace_id, generation, share_revision)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __putPersistentIrohBytes)]
pub async fn put_persistent_iroh_bytes(
&self,
namespace_id: String,
key: Vec<u8>,
value: Vec<u8>,
) -> Result<JsValue, JsValue> {
let guard = self.persistent_protocols.lock().await;
let receipt = persistent_protocols(&guard)?
.set_bytes(&namespace_id, key, value)
.await
.map_err(js_protocol_error)?;
serde_wasm_bindgen::to_value(&receipt)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
#[wasm_bindgen(js_name = __setPersistentIrohHash)]
pub async fn set_persistent_iroh_hash(
&self,
namespace_id: String,
key: Vec<u8>,
content_hash: String,
content_length: u64,
) -> Result<String, JsValue> {
let guard = self.persistent_protocols.lock().await;
persistent_protocols(&guard)?
.set_hash(&namespace_id, key, &content_hash, content_length)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __deletePersistentIrohPrefix)]
pub async fn delete_persistent_iroh_prefix(
&self,
namespace_id: String,
prefix: Vec<u8>,
) -> Result<JsValue, JsValue> {
let guard = self.persistent_protocols.lock().await;
let receipt = persistent_protocols(&guard)?
.delete_prefix(&namespace_id, prefix)
.await
.map_err(js_protocol_error)?;
serde_wasm_bindgen::to_value(&receipt)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
#[wasm_bindgen(js_name = __queryPersistentIrohNamespace)]
pub async fn query_persistent_iroh_namespace(
&self,
namespace_id: String,
key_prefix: Vec<u8>,
) -> Result<JsValue, JsValue> {
let guard = self.persistent_protocols.lock().await;
let entries = persistent_protocols(&guard)?
.query(&namespace_id, key_prefix)
.await
.map_err(js_protocol_error)?;
serde_wasm_bindgen::to_value(&entries)
.map_err(|error| JsValue::from_str(&error.to_string()))
}
#[wasm_bindgen(js_name = __hydratePersistentIrohBlob)]
pub async fn hydrate_persistent_iroh_blob(
&self,
content_hash: String,
) -> Result<(), JsValue> {
let guard = self.persistent_protocols.lock().await;
persistent_protocols(&guard)?
.hydrate_blob(&content_hash)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __acknowledgePersistentIrohOutbox)]
pub async fn acknowledge_persistent_iroh_outbox(
&self,
operation_id: String,
) -> Result<(), JsValue> {
let guard = self.persistent_protocols.lock().await;
persistent_protocols(&guard)?
.acknowledge_outbox(&operation_id)
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __flushPersistentIrohProtocols)]
pub async fn flush_persistent_iroh_protocols(&self) -> Result<(), JsValue> {
let guard = self.persistent_protocols.lock().await;
persistent_protocols(&guard)?
.flush()
.await
.map_err(js_protocol_error)
}
#[wasm_bindgen(js_name = __shutdownPersistentIrohProtocols)]
pub async fn shutdown_persistent_iroh_protocols(&self) -> Result<(), JsValue> {
if let Some(actor) = self.persistent_protocols.lock().await.take() {
actor.shutdown().await.map_err(js_protocol_error)?;
}
Ok(())
}
}
#[cfg(feature = "iroh-protocols-wasm")]
fn persistent_protocols(
guard: &Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>,
) -> Result<&crate::wasm_docs_persistence::WasmPersistentDocsActor, JsValue> {
guard
.as_ref()
.ok_or_else(|| JsValue::from_str("persistent Iroh protocols are not initialized"))
}
#[cfg(feature = "iroh-protocols-wasm")]
fn persistent_protocols_mut(
guard: &mut Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>,
) -> Result<&mut crate::wasm_docs_persistence::WasmPersistentDocsActor, JsValue> {
guard
.as_mut()
.ok_or_else(|| JsValue::from_str("persistent Iroh protocols are not initialized"))
}
#[cfg(feature = "iroh-protocols-wasm")]
fn js_protocol_error(error: impl std::fmt::Display) -> JsValue {
JsValue::from_str(&format!("{error:#}"))
}
#[wasm_bindgen(start)]
pub fn start() {
console_error_panic_hook::set_once();
}
}