1#![allow(deprecated)]
2
3pub mod application_crypto;
4pub mod application_crypto_streams;
5#[cfg(not(target_arch = "wasm32"))]
6pub mod scoped_streams;
7pub mod broadcast;
8pub mod client;
9pub mod diagnostics;
10pub mod coordination;
11pub mod datagrams;
12pub mod explicit_transfer_crypto;
13pub(crate) mod generated;
14pub mod heartbeat;
15#[cfg(feature = "iroh-carrier-core")]
16pub mod iroh_carrier;
17#[cfg(feature = "iroh-carrier-core")]
18pub mod iroh_carrier_bootstrap;
19pub mod iroh_carrier_kind;
22#[cfg(feature = "iroh-carrier-core")]
23pub mod iroh_carrier_proof;
24pub(crate) mod iroh_connection_policy;
25pub mod key_agreement;
26pub mod lifecycle_reason;
27#[cfg(feature = "managed-group-encryption")]
28#[allow(
29 dead_code,
30 reason = "gated managed-room controller; consumed only by managed gateway adapters"
31)]
32pub(crate) mod managed_group_controller;
33#[cfg(feature = "managed-group-encryption")]
34#[allow(
35 dead_code,
36 reason = "gated managed-room crypto owner; enabled only after the tracked persistence and router milestones"
37)]
38pub(crate) mod managed_group_crypto;
39pub mod media;
40pub mod native_protocol;
41pub mod offline;
42#[cfg(feature = "iroh-carrier-core")]
43pub mod packet_carrier_transport;
44pub mod presence;
45pub(crate) mod presence_policy;
46pub mod protocol_config;
47pub mod route_policy;
48pub mod runtime_policy;
49pub mod service_errors;
50pub mod session_token;
51pub mod signaling;
52pub mod sparse_fanout;
53#[cfg(any(target_arch = "wasm32", test))]
54mod stream_dispatch;
55pub mod stream_metadata;
56pub(crate) mod transport_generation;
57pub(crate) mod transport_label;
58
59#[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
60pub mod local_discovery;
61
62#[cfg(test)]
66pub mod test_constants {
67 pub const TEST_PROJECT_ID: &str = "test-project";
68 pub const TEST_API_KEY: &str = "pk_test_0000000000000000000000000000000000000000";
69}
70
71pub const LIVE_PROJECT_ID: &str = "pluto-rtc-prod";
73
74pub fn validate_api_key(api_key: &str) -> anyhow::Result<&str> {
78 let trimmed = api_key.trim();
79 let valid_prefix = trimmed.starts_with("pk_live_") || trimmed.starts_with("pk_test_");
80 let suffix = trimmed.get(8..).unwrap_or_default();
81 if !valid_prefix || suffix.len() != 40 || !suffix.bytes().all(|byte| byte.is_ascii_hexdigit()) {
82 anyhow::bail!("OpenRTC 2.0 requires a public pk_live_ or pk_test_ API key");
83 }
84 Ok(trimmed)
85}
86
87pub fn app_tag_from_api_key(api_key: &str) -> String {
88 let trimmed = api_key.trim();
89 if trimmed.is_empty() {
90 return "app_anonymous".to_string();
91 }
92
93 let suffix_len = trimmed.len().min(16);
94 format!("app_{}", &trimmed[trimmed.len() - suffix_len..])
95}
96
97pub fn space_app_tag(api_key: &str, space_key: &str) -> String {
98 let input = format!("{}:{}", api_key.trim(), space_key.trim());
99 let digest = <sha2::Sha256 as sha2::Digest>::digest(input.as_bytes());
100 format!("space::{}", hex::encode(digest))
101}
102
103#[cfg(test)]
104mod constructor_contract_tests {
105 use super::*;
106
107 #[test]
108 fn rust_constructor_is_provider_neutral_and_side_effect_free() {
109 let api_key = test_constants::TEST_API_KEY;
110 let client = client::Client::new(api_key.to_string()).expect("valid public API key");
111
112 assert_eq!(client.app_tag(), app_tag_from_api_key(api_key));
113 assert!(client::Client::new("firebase-project-id".to_string()).is_err());
114 }
115}
116
117#[cfg(not(target_arch = "wasm32"))]
118pub fn ensure_rustls() {
119 if rustls::crypto::CryptoProvider::get_default().is_none() {
120 let _ = rustls::crypto::ring::default_provider().install_default();
121 }
122}
123
124#[cfg(not(target_arch = "wasm32"))]
125pub mod adapters;
126
127#[cfg(not(target_arch = "wasm32"))]
128pub(crate) mod native_coordination_gateway;
129
130#[cfg(not(target_arch = "wasm32"))]
131pub mod native;
132
133#[cfg(not(target_arch = "wasm32"))]
134mod native_key_store;
135
136pub use client::Client;
137pub mod ticket_mesh;
138#[cfg(not(target_arch = "wasm32"))]
139pub use native::ControlPlane;
140#[cfg(not(target_arch = "wasm32"))]
141pub use ticket_mesh::TicketMesh;
142pub use ticket_mesh::{TicketMeshError, TicketMeshErrorCode, TicketMeshOptions};
143
144pub mod connection_manager;
145
146pub mod protocol_registry;
147
148#[cfg(not(target_arch = "wasm32"))]
149pub mod runtime_manager;
150
151#[cfg(not(target_arch = "wasm32"))]
152pub mod transport;
153
154#[cfg(not(target_arch = "wasm32"))]
155pub use client::EndpointHandle;
156
157#[cfg(all(
158 not(target_arch = "wasm32"),
159 not(any(target_os = "ios", target_os = "android"))
160))]
161pub mod sso;
162
163#[cfg(not(target_arch = "wasm32"))]
164pub mod native_node;
165
166#[cfg(not(target_arch = "wasm32"))]
167pub mod native_device;
168
169#[cfg(all(
170 test,
171 not(target_arch = "wasm32"),
172 feature = "iroh-protocols-wasm",
173 any(feature = "transport-webrtc", feature = "transport-moq")
174))]
175mod native_carrier_protocol_test;
176#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
177pub mod native_moq_carrier;
178#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
179pub mod native_webrtc_carrier;
180
181#[cfg(feature = "iroh-protocols-wasm")]
182mod docs_ticket_sync;
183#[cfg(all(target_arch = "wasm32", feature = "iroh-protocols-wasm"))]
184mod wasm_docs_persistence;
185#[cfg(all(target_arch = "wasm32", feature = "iroh-protocols-wasm"))]
186mod wasm_indexeddb_blob_store;
187#[cfg(all(target_arch = "wasm32", feature = "transport-moq"))]
188pub mod wasm_moq_carrier;
189#[cfg(target_arch = "wasm32")]
190pub mod wasm_node;
191#[cfg(all(target_arch = "wasm32", feature = "transport-webrtc"))]
192pub mod wasm_webrtc_carrier;
193
194#[cfg(target_arch = "wasm32")]
195#[macro_export]
196macro_rules! console_log {
197 ($($t:tt)*) => (web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format_args!($($t)*).to_string())))
198}
199
200#[cfg(target_arch = "wasm32")]
202pub mod wasm_api {
203 fn session_admission_js_error(reason: &str) -> wasm_bindgen::JsValue {
207 let error = js_sys::Error::new(reason);
208 if reason.contains(crate::session_token::PEER_ADMISSION_LEASE_EXPIRED) {
209 let _ = js_sys::Reflect::set(
210 error.as_ref(),
211 &wasm_bindgen::JsValue::from_str("sessionAdmissionDisposition"),
212 &wasm_bindgen::JsValue::from_str("retry-local-policy"),
213 );
214 }
215 error.into()
216 }
217 use crate::client::Client;
218 use crate::session_token::split_ticket;
219 use crate::wasm_node::{peer_uni_stream_from_send, BiStream, PeerUniStream};
220 use iroh_tickets::endpoint::EndpointTicket;
221 use std::cell::RefCell;
222 use std::collections::{HashMap, HashSet};
223 use std::rc::Rc;
224 use std::str::FromStr;
225 use std::sync::{Arc, Mutex};
226 use wasm_bindgen::prelude::*;
227 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
228 use wasm_bindgen_futures::spawn_local;
229 use wasm_streams::readable::sys::ReadableStream as JsReadableStream;
230
231 async fn send_native_signal_control_frame(
232 inner: &Client,
233 endpoint_id: String,
234 frame: Vec<u8>,
235 ) -> Result<(), JsValue> {
236 let endpoint_id: iroh::EndpointId = endpoint_id
237 .parse()
238 .map_err(|error| JsValue::from_str(&format!("{error}")))?;
239 inner
240 .ensure_connection_manager_record_before_peer_stream(&endpoint_id)
241 .await
242 .map_err(|error| JsValue::from_str(&error.to_string()))?;
243 let (mut send, _recv) = inner
244 .open_bi_internal(endpoint_id)
245 .await
246 .map_err(|error| JsValue::from_str(&error.to_string()))?;
247 send.write_all(&[0x00])
248 .await
249 .map_err(|error| JsValue::from_str(&error.to_string()))?;
250 send.write_all(&(b"signal".len() as u32).to_be_bytes())
251 .await
252 .map_err(|error| JsValue::from_str(&error.to_string()))?;
253 send.write_all(b"signal")
254 .await
255 .map_err(|error| JsValue::from_str(&error.to_string()))?;
256 send.write_all(&frame)
257 .await
258 .map_err(|error| JsValue::from_str(&error.to_string()))?;
259 crate::application_crypto_streams::PeerSendStream::plain(send)
260 .finish_and_wait_for_peer(std::time::Duration::from_secs(2))
261 .await
262 .map_err(|error| JsValue::from_str(&error.to_string()))
263 }
264
265 #[cfg(feature = "transport-webrtc")]
266 #[derive(Debug, Clone)]
267 struct WasmWebRtcCarrierAttempt {
268 connection_id: String,
269 remote_endpoint_id: iroh::EndpointId,
270 bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
271 generation: crate::client::WasmPeerDataGeneration,
272 role: &'static str,
273 prepared: bool,
274 retry_sent: bool,
275 offer_started: bool,
276 remote_ready: bool,
277 completion_started: bool,
278 inbound_authorization_expires_at_ms: Option<f64>,
279 ice_refresh: crate::iroh_carrier_bootstrap::WebRtcIceRefreshGate,
280 ice_refresh_baseline:
281 Rc<RefCell<Option<crate::packet_carrier_transport::WebRtcIceRefreshPathLoan>>>,
282 }
283
284 #[cfg(feature = "transport-moq")]
285 #[derive(Debug, Clone)]
286 struct WasmMoqCarrierAttempt {
287 connection_id: String,
288 remote_endpoint_id: iroh::EndpointId,
289 bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
290 generation: crate::client::WasmPeerDataGeneration,
291 role: &'static str,
292 prepared: bool,
293 retry_sent: bool,
294 retry_count: u8,
295 inbound_authorization_expires_at_ms: Option<f64>,
296 }
297
298 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
299 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
300 struct WasmRemoteCarrierCapabilities {
301 webrtc: bool,
302 moq: bool,
303 }
304
305 #[wasm_bindgen]
306 pub struct WasmPeerDatagramPolicy {
307 inner: RefCell<crate::datagrams::PeerDatagramPolicy>,
308 }
309
310 #[wasm_bindgen]
311 impl WasmPeerDatagramPolicy {
312 #[wasm_bindgen(js_name = setIncomingMaxAge)]
313 pub fn set_incoming_max_age(&self, value: Option<u32>, now_ms: f64) {
314 self.inner
315 .borrow_mut()
316 .set_incoming_max_age_ms(value.map(u64::from), now_ms.max(0.0) as u64);
317 }
318
319 #[wasm_bindgen(js_name = setOutgoingMaxAge)]
320 pub fn set_outgoing_max_age(&self, value: Option<u32>) {
321 self.inner
322 .borrow_mut()
323 .set_outgoing_max_age_ms(value.map(u64::from));
324 }
325
326 #[wasm_bindgen(js_name = setIncomingMaxBufferedDatagrams)]
327 pub fn set_incoming_max_buffered_datagrams(&self, value: u32) -> Result<(), JsValue> {
328 self.inner
329 .borrow_mut()
330 .set_incoming_max_buffered(value as usize)
331 .map_err(|error| JsValue::from_str(&error.to_string()))
332 }
333
334 #[wasm_bindgen(js_name = setOutgoingMaxBufferedDatagrams)]
335 pub fn set_outgoing_max_buffered_datagrams(&self, value: u32) -> Result<(), JsValue> {
336 self.inner
337 .borrow_mut()
338 .set_outgoing_max_buffered(value as usize)
339 .map_err(|error| JsValue::from_str(&error.to_string()))
340 }
341
342 #[wasm_bindgen(js_name = pushIncoming)]
343 pub fn push_incoming(&self, payload: Vec<u8>, now_ms: f64) -> Result<(), JsValue> {
344 self.inner
345 .borrow_mut()
346 .push_incoming(payload, now_ms.max(0.0) as u64)
347 .map_err(|error| JsValue::from_str(&error.to_string()))
348 }
349
350 #[wasm_bindgen(js_name = popIncoming)]
351 pub fn pop_incoming(&self, now_ms: f64) -> Option<Vec<u8>> {
352 self.inner.borrow_mut().pop_incoming(now_ms.max(0.0) as u64)
353 }
354
355 #[wasm_bindgen(js_name = pushOutgoing)]
356 pub fn push_outgoing(&self, payload: Vec<u8>, now_ms: f64) -> Result<bool, JsValue> {
357 self.inner
358 .borrow_mut()
359 .push_outgoing(payload, now_ms.max(0.0) as u64)
360 .map_err(|error| JsValue::from_str(&error.to_string()))
361 }
362
363 #[wasm_bindgen(js_name = popOutgoing)]
364 pub fn pop_outgoing(&self, now_ms: f64) -> Result<JsValue, JsValue> {
365 let Some(outgoing) = self.inner.borrow_mut().pop_outgoing(now_ms.max(0.0) as u64)
366 else {
367 return Ok(JsValue::NULL);
368 };
369 let result = js_sys::Object::new();
370 js_sys::Reflect::set(
371 &result,
372 &JsValue::from_str("payload"),
373 &js_sys::Uint8Array::from(outgoing.payload.as_slice()),
374 )?;
375 js_sys::Reflect::set(
376 &result,
377 &JsValue::from_str("remainingMaxAgeMs"),
378 &outgoing
379 .remaining_max_age_ms
380 .map(|value| JsValue::from_f64(value as f64))
381 .unwrap_or(JsValue::NULL),
382 )?;
383 Ok(result.into())
384 }
385
386 #[wasm_bindgen(js_name = recordSent)]
387 pub fn record_sent(&self, bytes: usize) {
388 self.inner.borrow_mut().record_sent(bytes);
389 }
390
391 #[wasm_bindgen(js_name = recordExpiredOutgoing)]
392 pub fn record_expired_outgoing(&self) {
393 self.inner.borrow_mut().record_expired_outgoing();
394 }
395
396 #[wasm_bindgen(js_name = recordSendFailure)]
397 pub fn record_send_failure(&self) {
398 self.inner.borrow_mut().record_send_failure();
399 }
400
401 #[wasm_bindgen(js_name = getStats)]
402 pub fn get_stats(&self, now_ms: f64) -> Result<JsValue, JsValue> {
403 let stats = self.inner.borrow_mut().stats(now_ms.max(0.0) as u64);
404 serde::Serialize::serialize(&stats, &serde_wasm_bindgen::Serializer::json_compatible())
405 .map_err(|error| JsValue::from_str(&error.to_string()))
406 }
407
408 #[wasm_bindgen(js_name = closeIncoming)]
409 pub fn close_incoming(&self) {
410 self.inner.borrow_mut().close_incoming();
411 }
412
413 #[wasm_bindgen(js_name = closeOutgoing)]
414 pub fn close_outgoing(&self) {
415 self.inner.borrow_mut().close_outgoing();
416 }
417
418 pub fn close(&self) {
419 self.inner.borrow_mut().close();
420 }
421 }
422
423 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
424 fn emit_wasm_carrier_action(
425 handler: &Rc<RefCell<Option<js_sys::Function>>>,
426 action: serde_json::Value,
427 ) {
428 let Some(handler) = handler.borrow().as_ref().cloned() else {
429 return;
430 };
431 let Ok(value) = serde::Serialize::serialize(
437 &action,
438 &serde_wasm_bindgen::Serializer::json_compatible(),
439 ) else {
440 return;
441 };
442 if let Err(error) = handler.call1(&JsValue::UNDEFINED, &value) {
443 web_sys::console::error_2(
444 &JsValue::from_str("[OpenRTC][WASM carrier] action handler failed"),
445 &error,
446 );
447 }
448 }
449
450 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
451 fn wasm_carrier_failure_code(error: &anyhow::Error) -> &'static str {
452 let message = format!("{error:#}");
453 if message.contains("base generation changed before candidate acknowledgement")
454 || message.contains("incumbent generation changed")
455 || message.contains("replacement incumbent is stale")
456 {
457 "carrier-base-generation-stale"
458 } else if message.contains("authorization epoch changed") {
459 "carrier-authorization-stale"
460 } else if message.contains("stale before atomic commit")
461 || message.contains("became stale during atomic commit")
462 {
463 "carrier-logical-generation-stale"
464 } else if message.contains("attempt was retired") || message.contains("retired upgrade") {
465 "carrier-attempt-retired"
466 } else {
467 "carrier-proof-failed"
468 }
469 }
470
471 struct TicketMeshStartGuard<'a> {
472 starts: &'a RefCell<HashSet<String>>,
473 id: String,
474 }
475
476 impl Drop for TicketMeshStartGuard<'_> {
477 fn drop(&mut self) {
478 self.starts.borrow_mut().remove(&self.id);
479 }
480 }
481
482 #[wasm_bindgen]
483 pub struct WasmClient {
484 inner: Arc<Client>,
485 portable_media: RefCell<crate::media::PortableMediaSession>,
486 broadcast_sessions: RefCell<HashMap<String, crate::broadcast::BroadcastSession>>,
487 broadcast_signers: RefCell<HashMap<String, crate::broadcast::BroadcastPublisherSigner>>,
488 ticket_meshes:
489 RefCell<HashMap<String, Rc<crate::ticket_mesh::wasm_session::WasmTicketMesh>>>,
490 ticket_mesh_starts: RefCell<HashSet<String>>,
491 identity_credential: Arc<Mutex<Option<String>>>,
492 last_auth_log: Arc<Mutex<Option<(bool, usize)>>>,
493 #[cfg(feature = "iroh-protocols-wasm")]
494 persistent_protocols:
495 Rc<tokio::sync::Mutex<Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>>>,
496 #[cfg(feature = "managed-group-encryption")]
497 managed_group_controllers:
498 Rc<RefCell<HashMap<String, crate::managed_group_controller::ManagedGroupController>>>,
499 #[cfg(feature = "transport-webrtc")]
500 wasm_webrtc_carrier_sessions:
501 Rc<RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>>,
502 #[cfg(feature = "transport-webrtc")]
503 wasm_webrtc_carrier_attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
504 #[cfg(feature = "transport-webrtc")]
505 webrtc_availability: Rc<RefCell<crate::iroh_carrier_bootstrap::WebRtcAvailability>>,
506 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
507 wasm_carrier_action_handler: Rc<RefCell<Option<js_sys::Function>>>,
508 #[cfg(feature = "transport-moq")]
509 wasm_moq_carrier_sessions:
510 Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
511 #[cfg(feature = "transport-moq")]
512 wasm_moq_carrier_attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
513 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
514 wasm_remote_carrier_capabilities:
515 Rc<RefCell<HashMap<String, WasmRemoteCarrierCapabilities>>>,
516 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
522 wasm_carrier_peer_lifecycle: Rc<tokio::sync::Mutex<()>>,
523 }
524
525 struct WasmRevocationCarriers {
528 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
529 lifecycle: Rc<tokio::sync::Mutex<()>>,
530 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
531 handler: Rc<RefCell<Option<js_sys::Function>>>,
532 #[cfg(feature = "transport-webrtc")]
533 webrtc_attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
534 #[cfg(feature = "transport-webrtc")]
535 webrtc_sessions:
536 Rc<RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>>,
537 #[cfg(feature = "transport-moq")]
538 moq_attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
539 #[cfg(feature = "transport-moq")]
540 moq_sessions: Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
541 }
542
543 impl WasmClient {
544 fn revocation_carriers(&self) -> WasmRevocationCarriers {
545 WasmRevocationCarriers {
546 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
547 lifecycle: self.wasm_carrier_peer_lifecycle.clone(),
548 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
549 handler: self.wasm_carrier_action_handler.clone(),
550 #[cfg(feature = "transport-webrtc")]
551 webrtc_attempts: self.wasm_webrtc_carrier_attempts.clone(),
552 #[cfg(feature = "transport-webrtc")]
553 webrtc_sessions: self.wasm_webrtc_carrier_sessions.clone(),
554 #[cfg(feature = "transport-moq")]
555 moq_attempts: self.wasm_moq_carrier_attempts.clone(),
556 #[cfg(feature = "transport-moq")]
557 moq_sessions: self.wasm_moq_carrier_sessions.clone(),
558 }
559 }
560 }
561
562 async fn finish_wasm_revocation(
563 inner: Arc<Client>,
564 receipt: crate::session_token::RevocationReceipt,
565 scope: Option<String>,
566 carriers: WasmRevocationCarriers,
567 ) {
568 let retirements = inner.prepare_revoked_session_connections(&receipt).await;
569 for retirement in &retirements {
570 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
571 let lifecycle = carriers.lifecycle.lock().await;
572 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
573 let connection_id = &retirement.record.connection_id;
574 #[cfg(feature = "transport-webrtc")]
575 let webrtc = {
576 let mut attempts = carriers.webrtc_attempts.borrow_mut();
577 attempts.remove(connection_id).map(|attempt| {
581 let session = carriers
582 .webrtc_sessions
583 .borrow_mut()
584 .remove(&attempt.bootstrap.upgrade_id);
585 (attempt, session)
586 })
587 };
588 #[cfg(feature = "transport-moq")]
589 let moq = {
590 let mut attempts = carriers.moq_attempts.borrow_mut();
591 attempts.remove(connection_id).map(|attempt| {
595 let session = carriers
596 .moq_sessions
597 .borrow_mut()
598 .remove(&attempt.bootstrap.upgrade_id);
599 (attempt, session)
600 })
601 };
602 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
603 drop(lifecycle);
604 #[cfg(feature = "transport-webrtc")]
607 if let Some((attempt, session)) = webrtc {
608 if let Some(session) = session.as_ref() {
609 let _ = session
610 .send_terminal(crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED)
611 .await;
612 }
613 inner
614 .retire_wasm_carrier_upgrade(
615 connection_id,
616 crate::client::IrohPathKind::WebRtc,
617 &attempt.bootstrap.upgrade_id,
618 attempt.generation,
619 )
620 .await;
621 emit_wasm_carrier_action(
622 &carriers.handler,
623 serde_json::json!({
624 "type": "retire-webrtc", "connectionId": connection_id,
625 "upgradeId": attempt.bootstrap.upgrade_id,
626 "failureCode": "logical-connection-retired",
627 }),
628 );
629 }
630 #[cfg(feature = "transport-moq")]
631 if let Some((attempt, session)) = moq {
632 if let Some(session) = session.as_ref() {
633 let _ = session
634 .send_terminal(crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED)
635 .await;
636 }
637 inner
638 .retire_wasm_carrier_upgrade(
639 connection_id,
640 crate::client::IrohPathKind::Moq,
641 &attempt.bootstrap.upgrade_id,
642 attempt.generation,
643 )
644 .await;
645 emit_wasm_carrier_action(
646 &carriers.handler,
647 serde_json::json!({
648 "type": "retire-moq", "connectionId": connection_id,
649 "upgradeId": attempt.bootstrap.upgrade_id,
650 "failureCode": "logical-connection-retired",
651 }),
652 );
653 }
654 #[cfg(not(any(feature = "transport-webrtc", feature = "transport-moq")))]
655 let _ = (retirement, &carriers);
656 }
657 if let Some(scope) = scope {
658 inner
659 .finish_revoke_tokens_by_scope_with_retirements(&scope, &receipt, retirements)
660 .await;
661 } else {
662 inner.retire_revoked_session_connections(retirements).await;
663 }
664 }
665
666 #[cfg(feature = "transport-webrtc")]
670 #[derive(Clone)]
671 struct WasmWebRtcCarrierOwner {
672 inner: Arc<Client>,
673 wasm_webrtc_carrier_sessions:
674 Rc<RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>>,
675 wasm_webrtc_carrier_attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
676 wasm_carrier_action_handler: Rc<RefCell<Option<js_sys::Function>>>,
677 wasm_remote_carrier_capabilities:
678 Rc<RefCell<HashMap<String, WasmRemoteCarrierCapabilities>>>,
679 wasm_carrier_peer_lifecycle: Rc<tokio::sync::Mutex<()>>,
680 }
681
682 #[cfg(feature = "transport-webrtc")]
683 impl WasmClient {
684 fn webrtc_carrier_owner(&self) -> WasmWebRtcCarrierOwner {
685 WasmWebRtcCarrierOwner {
686 inner: self.inner.clone(),
687 wasm_webrtc_carrier_sessions: self.wasm_webrtc_carrier_sessions.clone(),
688 wasm_webrtc_carrier_attempts: self.wasm_webrtc_carrier_attempts.clone(),
689 wasm_carrier_action_handler: self.wasm_carrier_action_handler.clone(),
690 wasm_remote_carrier_capabilities: self.wasm_remote_carrier_capabilities.clone(),
691 wasm_carrier_peer_lifecycle: self.wasm_carrier_peer_lifecycle.clone(),
692 }
693 }
694 }
695
696 #[cfg(feature = "transport-webrtc")]
697 impl WasmWebRtcCarrierOwner {
698 async fn begin_attempt(
699 &self,
700 connection_id: String,
701 remote_endpoint_id: String,
702 retry_count: u8,
703 ) -> Result<bool, JsValue> {
704 if !self.inner.is_webrtc_carrier_enabled().await {
705 return Ok(false);
706 }
707 let local_endpoint_id = self
708 .inner
709 .current_node_id()
710 .await
711 .ok_or_else(|| JsValue::from_str("local Iroh endpoint id is unavailable"))?;
712 if local_endpoint_id.as_str() <= remote_endpoint_id.as_str()
713 || !crate::iroh_connection_policy::custom_carrier_base_allows(
714 self.inner.iroh_path_kind(&remote_endpoint_id).await,
715 crate::client::IrohPathKind::WebRtc,
716 false,
717 )
718 {
719 return Ok(false);
720 }
721 let endpoint_id = remote_endpoint_id
722 .parse::<iroh::EndpointId>()
723 .map_err(|error| JsValue::from_str(&error.to_string()))?;
724 self.inner
725 .get_connection(endpoint_id)
726 .await
727 .ok_or_else(|| JsValue::from_str("WebRTC carrier base is unavailable"))?;
728 let generation = self
729 .inner
730 .current_wasm_peer_data_generation(&connection_id, None)
731 .await
732 .ok_or_else(|| JsValue::from_str("WebRTC carrier generation is unavailable"))?;
733 if self
734 .inner
735 .session_admission_block_reason_for_transport(
736 &connection_id,
737 Some(generation.transport_stable_id),
738 )
739 .is_some()
740 {
741 return Ok(false);
742 }
743 let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
744 crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc,
745 crate::iroh_carrier_bootstrap::CarrierGenerationFence {
746 transport_stable_id: generation.transport_stable_id,
747 transport_generation: generation.transport_generation,
748 route_generation: generation.route_generation,
749 },
750 retry_count.saturating_add(1),
751 )
752 .map_err(|error| JsValue::from_str(&error.to_string()))?;
753 let kind = crate::client::IrohPathKind::WebRtc;
754 if !matches!(
755 self.inner
756 .reserve_wasm_carrier_upgrade(
757 &connection_id,
758 kind,
759 &bootstrap.upgrade_id,
760 generation,
761 )
762 .await,
763 crate::client::WasmCarrierUpgradeReservation::Reserved
764 ) {
765 let pending = self
766 .wasm_webrtc_carrier_attempts
767 .borrow_mut()
768 .get_mut(&connection_id)
769 .filter(|attempt| {
770 attempt.role == "initiator"
771 && attempt.generation == generation
772 && attempt.prepared
773 && !attempt.retry_sent
774 && !attempt.completion_started
775 })
776 .map(|attempt| {
777 attempt.retry_sent = true;
778 attempt.clone()
779 });
780 if let Some(pending) = pending {
781 emit_wasm_carrier_action(
785 &self.wasm_carrier_action_handler,
786 serde_json::json!({
787 "type": "send-control",
788 "connectionId": connection_id,
789 "remoteEndpointId": pending.remote_endpoint_id.to_string(),
790 "envelope": pending.bootstrap,
791 }),
792 );
793 return Ok(true);
794 }
795 return Ok(false);
796 }
797 let previous = self.wasm_webrtc_carrier_attempts.borrow_mut().insert(
798 connection_id.clone(),
799 WasmWebRtcCarrierAttempt {
800 connection_id: connection_id.clone(),
801 remote_endpoint_id: endpoint_id,
802 bootstrap: bootstrap.clone(),
803 generation,
804 role: "initiator",
805 prepared: false,
806 retry_sent: false,
807 offer_started: false,
808 remote_ready: false,
809 completion_started: false,
810 inbound_authorization_expires_at_ms: None,
811 ice_refresh: Default::default(),
812 ice_refresh_baseline: Rc::new(RefCell::new(None)),
813 },
814 );
815 if let Some(previous) = previous {
816 self.wasm_webrtc_carrier_sessions
817 .borrow_mut()
818 .remove(&previous.bootstrap.upgrade_id);
819 }
820 emit_wasm_carrier_action(
821 &self.wasm_carrier_action_handler,
822 serde_json::json!({
823 "type": "prepare-webrtc",
824 "connectionId": connection_id,
825 "remoteEndpointId": remote_endpoint_id,
826 "role": "initiator",
827 "upgradeId": bootstrap.upgrade_id,
828 "carrierSessionId": bootstrap.carrier_session_id,
829 "transportGeneration": generation.transport_generation.saturating_add(1),
830 }),
831 );
832 self.schedule_wasm_webrtc_carrier_watchdog(bootstrap);
833 Ok(true)
834 }
835
836 fn schedule_wasm_webrtc_carrier_watchdog(
837 &self,
838 bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
839 ) {
840 let owner = self.clone();
841 let inner = self.inner.clone();
842 let attempts = self.wasm_webrtc_carrier_attempts.clone();
843 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
844 spawn_local(async move {
845 let mut completion_phase_observed = false;
846 loop {
847 let timeout = if completion_phase_observed { 45 } else { 30 };
848 gloo_timers::future::sleep(std::time::Duration::from_secs(timeout)).await;
849 let _lifecycle = lifecycle.lock().await;
850 let attempt = attempts
851 .borrow()
852 .values()
853 .find(|attempt| attempt.bootstrap.upgrade_id == bootstrap.upgrade_id)
854 .cloned();
855 let Some(attempt) = attempt else {
856 return;
857 };
858 if !inner
859 .wasm_carrier_upgrade_is_current(
860 &attempt.connection_id,
861 crate::client::IrohPathKind::WebRtc,
862 &attempt.bootstrap.upgrade_id,
863 attempt.generation,
864 )
865 .await
866 {
867 return;
868 }
869 if attempt.completion_started && !completion_phase_observed {
870 completion_phase_observed = true;
877 continue;
878 }
879 fail_wasm_webrtc_attempt(owner.clone(), attempt, "carrier-timeout", true).await;
880 return;
881 }
882 });
883 }
884
885 fn schedule_retry(
886 &self,
887 attempt: WasmWebRtcCarrierAttempt,
888 generation: Option<crate::client::WasmPeerDataGeneration>,
889 policy_epoch: u64,
890 peer_policy_epoch: u64,
891 ) {
892 let owner = self.clone();
893 spawn_local(async move {
894 gloo_timers::future::sleep(crate::iroh_carrier_bootstrap::webrtc_retry_delay(
895 &attempt.bootstrap.upgrade_id,
896 ))
897 .await;
898 let _lifecycle = owner.wasm_carrier_peer_lifecycle.lock().await;
899 if generation.is_none()
900 || owner
901 .inner
902 .current_wasm_peer_data_generation(&attempt.connection_id, None)
903 .await
904 != generation
905 || owner.inner.current_iroh_carrier_policy_epoch() != policy_epoch
906 || owner
907 .inner
908 .current_iroh_carrier_peer_policy_epoch(&attempt.connection_id)
909 != peer_policy_epoch
910 || owner
911 .wasm_webrtc_carrier_attempts
912 .borrow()
913 .contains_key(&attempt.connection_id)
914 || !owner
915 .wasm_remote_carrier_capabilities
916 .borrow()
917 .get(&attempt.connection_id)
918 .is_some_and(|caps| caps.webrtc)
919 {
920 return;
921 }
922 if let Err(error) = Box::pin(owner.begin_attempt(
925 attempt.connection_id.clone(),
926 attempt.remote_endpoint_id.to_string(),
927 1,
928 ))
929 .await
930 {
931 web_sys::console::warn_1(&JsValue::from_str(&format!(
932 "[OpenRTC][WebRTC carrier] bounded retry remained pending connection_id={} error={error:?}",
933 attempt.connection_id,
934 )));
935 }
936 });
937 }
938 }
939
940 #[cfg(feature = "transport-webrtc")]
941 async fn fail_wasm_webrtc_attempt(
942 owner: WasmWebRtcCarrierOwner,
943 attempt: WasmWebRtcCarrierAttempt,
944 failure_code: &'static str,
945 notify_peer: bool,
946 ) {
947 let inner = owner.inner.clone();
948 let attempts = owner.wasm_webrtc_carrier_attempts.clone();
949 let sessions = owner.wasm_webrtc_carrier_sessions.clone();
950 let handler = owner.wasm_carrier_action_handler.clone();
951 let policy_epoch = inner.current_iroh_carrier_policy_epoch();
952 let peer_policy_epoch =
953 inner.current_iroh_carrier_peer_policy_epoch(&attempt.connection_id);
954 let kind = crate::client::IrohPathKind::WebRtc;
955 let retry_fence_current = matches!(
956 inner
957 .wasm_carrier_upgrade_fence(
958 &attempt.connection_id,
959 kind,
960 &attempt.bootstrap.upgrade_id,
961 attempt.generation,
962 )
963 .await,
964 crate::client::WasmCarrierUpgradeFence::Current(_)
965 | crate::client::WasmCarrierUpgradeFence::BaseGenerationChanged
966 );
967 if !inner
968 .retire_wasm_carrier_upgrade(
969 &attempt.connection_id,
970 kind,
971 &attempt.bootstrap.upgrade_id,
972 attempt.generation,
973 )
974 .await
975 {
976 return;
977 }
978 if attempts
979 .borrow()
980 .get(&attempt.connection_id)
981 .is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
982 {
983 attempts.borrow_mut().remove(&attempt.connection_id);
984 }
985 sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
986 if notify_peer {
987 if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
988 &attempt.bootstrap,
989 failure_code,
990 ) {
991 emit_wasm_carrier_action(
992 &handler,
993 serde_json::json!({
994 "type": "send-control",
995 "connectionId": attempt.connection_id,
996 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
997 "envelope": failed,
998 }),
999 );
1000 }
1001 }
1002 emit_wasm_carrier_action(
1003 &handler,
1004 serde_json::json!({
1005 "type": "retire-webrtc",
1006 "connectionId": attempt.connection_id,
1007 "upgradeId": attempt.bootstrap.upgrade_id,
1008 "failureCode": failure_code,
1009 }),
1010 );
1011 let retry_pending = retry_fence_current
1012 && attempt.role == "initiator"
1013 && crate::iroh_carrier_bootstrap::webrtc_retry_attempt(
1014 failure_code,
1015 attempt.bootstrap.attempt,
1016 )
1017 .is_some();
1018 if retry_pending {
1019 let generation = inner
1021 .current_wasm_peer_data_generation(&attempt.connection_id, None)
1022 .await;
1023 owner.schedule_retry(attempt, generation, policy_epoch, peer_policy_epoch);
1024 return;
1025 }
1026 if attempt.role == "initiator" && !retry_pending {
1027 emit_wasm_carrier_action(
1028 &handler,
1029 serde_json::json!({
1030 "type": "advance-carrier",
1031 "connectionId": attempt.connection_id,
1032 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1033 "failedRoute": "webrtc",
1034 }),
1035 );
1036 }
1037 }
1038
1039 #[cfg(feature = "transport-webrtc")]
1040 async fn retire_selected_wasm_webrtc_carrier(
1041 inner: Arc<Client>,
1042 attempts: Rc<RefCell<HashMap<String, WasmWebRtcCarrierAttempt>>>,
1043 sessions: Rc<
1044 RefCell<HashMap<String, crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession>>,
1045 >,
1046 handler: Rc<RefCell<Option<js_sys::Function>>>,
1047 attempt: &WasmWebRtcCarrierAttempt,
1048 failure_code: &'static str,
1049 lifecycle_reason: &'static str,
1050 ) -> bool {
1051 if !inner
1052 .close_current_iroh_carrier_generation_with_reason(
1053 &attempt.connection_id,
1054 attempt.remote_endpoint_id,
1055 crate::client::IrohPathKind::WebRtc,
1056 attempt.generation.transport_generation.saturating_add(1),
1057 lifecycle_reason,
1058 )
1059 .await
1060 {
1061 return false;
1062 }
1063 web_sys::console::warn_1(&JsValue::from_str(&format!(
1064 "[OpenRTC][WebRTC carrier] selected mechanism ended connection_id={} upgrade_id={} failure_code={failure_code}",
1065 attempt.connection_id, attempt.bootstrap.upgrade_id,
1066 )));
1067 if attempts
1068 .borrow()
1069 .get(&attempt.connection_id)
1070 .is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
1071 {
1072 attempts.borrow_mut().remove(&attempt.connection_id);
1073 }
1074 sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
1075 emit_wasm_carrier_action(
1076 &handler,
1077 serde_json::json!({
1078 "type": "retire-webrtc",
1079 "connectionId": attempt.connection_id,
1080 "upgradeId": attempt.bootstrap.upgrade_id,
1081 "failureCode": failure_code,
1082 }),
1083 );
1084 true
1085 }
1086
1087 #[cfg(feature = "transport-webrtc")]
1088 impl WasmClient {
1089 async fn fail_wasm_webrtc_carrier_attempt(
1090 &self,
1091 attempt: WasmWebRtcCarrierAttempt,
1092 failure_code: &'static str,
1093 notify_peer: bool,
1094 ) {
1095 fail_wasm_webrtc_attempt(
1096 self.webrtc_carrier_owner(),
1097 attempt,
1098 failure_code,
1099 notify_peer,
1100 )
1101 .await;
1102 }
1103
1104 fn spawn_outbound_wasm_webrtc_carrier_completion(&self, attempt: WasmWebRtcCarrierAttempt) {
1105 let inner = self.inner.clone();
1106 let handler = self.wasm_carrier_action_handler.clone();
1107 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
1108 let owner = self.webrtc_carrier_owner();
1109 spawn_local(async move {
1110 let kind = crate::client::IrohPathKind::WebRtc;
1111 let result = inner
1112 .complete_outbound_wasm_carrier_upgrade(
1113 &attempt.connection_id,
1114 &attempt.remote_endpoint_id.to_string(),
1115 &attempt.bootstrap.upgrade_id,
1116 attempt.generation,
1117 kind,
1118 )
1119 .await;
1120 let _lifecycle = lifecycle.lock().await;
1121 if let Err(error) = result {
1122 let failure_code = wasm_carrier_failure_code(&error);
1123 web_sys::console::error_1(&JsValue::from_str(&format!(
1124 "[OpenRTC][WebRTC carrier] outbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
1125 attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
1126 )));
1127 fail_wasm_webrtc_attempt(owner.clone(), attempt, failure_code, true).await;
1128 return;
1129 }
1130 if !owner
1134 .wasm_webrtc_carrier_attempts
1135 .borrow()
1136 .get(&attempt.connection_id)
1137 .is_some_and(|current| {
1138 current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id
1139 })
1140 || !inner
1141 .current_wasm_peer_data_generation(&attempt.connection_id, None)
1142 .await
1143 .is_some_and(|current| {
1144 current.transport_generation
1145 == attempt.generation.transport_generation.saturating_add(1)
1146 })
1147 {
1148 return;
1149 }
1150 emit_wasm_carrier_action(
1151 &handler,
1152 serde_json::json!({
1153 "type": "selected",
1154 "connectionId": attempt.connection_id,
1155 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1156 "upgradeId": attempt.bootstrap.upgrade_id,
1157 "family": "iroh",
1158 "carrier": "webrtc",
1159 "transportGeneration": attempt.generation.transport_generation.saturating_add(1),
1160 "routeGeneration": 0,
1161 }),
1162 );
1163 });
1164 }
1165
1166 fn take_ready_outbound_wasm_webrtc_carrier_attempt(
1167 &self,
1168 connection_id: &str,
1169 upgrade_id: &str,
1170 ) -> Option<WasmWebRtcCarrierAttempt> {
1171 if !self
1172 .wasm_webrtc_carrier_sessions
1173 .borrow()
1174 .contains_key(upgrade_id)
1175 {
1176 return None;
1177 }
1178 let mut attempts = self.wasm_webrtc_carrier_attempts.borrow_mut();
1179 let attempt = attempts.get_mut(connection_id).filter(|attempt| {
1180 attempt.bootstrap.upgrade_id == upgrade_id
1181 && attempt.role == "initiator"
1182 && attempt.remote_ready
1183 && !attempt.completion_started
1184 })?;
1185 attempt.completion_started = true;
1186 Some(attempt.clone())
1187 }
1188
1189 fn spawn_inbound_wasm_webrtc_carrier_completion(&self, attempt: WasmWebRtcCarrierAttempt) {
1190 let inner = self.inner.clone();
1191 let handler = self.wasm_carrier_action_handler.clone();
1192 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
1193 let owner = self.webrtc_carrier_owner();
1194 spawn_local(async move {
1195 let kind = crate::client::IrohPathKind::WebRtc;
1196 let node = inner.iroh_node.read().await.as_ref().cloned();
1197 let result = async {
1198 let node = node.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
1199 let authorization_expires_at_ms = attempt
1200 .inbound_authorization_expires_at_ms
1201 .ok_or_else(|| anyhow::anyhow!("browser WebRTC inbound authorization is missing"))?;
1202 let candidate = node
1203 .wait_for_inbound_replacement_candidate(
1204 attempt.remote_endpoint_id,
1205 crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
1206 authorization_expires_at_ms,
1207 std::time::Duration::from_secs(40),
1208 )
1209 .await?;
1210 let policy_epoch = inner
1211 .require_current_wasm_carrier_upgrade_fence(
1212 &attempt.connection_id,
1213 kind,
1214 &attempt.bootstrap.upgrade_id,
1215 attempt.generation,
1216 )
1217 .await?;
1218 let authorization_fence = inner
1219 .capture_wasm_iroh_carrier_authorization_fence(
1220 &attempt.connection_id,
1221 attempt.generation,
1222 kind,
1223 policy_epoch,
1224 )?;
1225 let proof = inner.wasm_candidate_proof_probe(
1226 &attempt.connection_id,
1227 &attempt.bootstrap.upgrade_id,
1228 attempt.bootstrap.base.transport_generation,
1229 attempt.bootstrap.base.route_generation,
1230 kind,
1231 crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
1232 )?;
1233 let (send, probe, ack) = inner
1234 .receive_inbound_wasm_carrier_candidate_proof(
1235 &attempt.connection_id,
1236 &candidate,
1237 &proof,
1238 )
1239 .await?;
1240 anyhow::ensure!(
1241 inner
1242 .current_wasm_peer_data_generation(&attempt.connection_id, None)
1243 .await
1244 == Some(attempt.generation),
1245 "browser WebRTC inbound carrier base generation changed before candidate acknowledgement"
1246 );
1247 inner
1248 .send_inbound_wasm_carrier_candidate_ack(send, &ack)
1249 .await?;
1250 let (commit_send, committed) = inner
1251 .receive_inbound_wasm_carrier_commit(
1252 &attempt.connection_id,
1253 &candidate,
1254 &probe,
1255 )
1256 .await?;
1257 let committed_replacement = inner
1258 .commit_proven_wasm_carrier_candidate(
1259 &attempt.connection_id,
1260 &attempt.bootstrap.upgrade_id,
1261 attempt.generation,
1262 kind,
1263 authorization_fence,
1264 candidate,
1265 )
1266 .await?;
1267 inner
1268 .send_inbound_wasm_carrier_candidate_ack(commit_send, &committed)
1269 .await?;
1270 let replacement_transport_stable_id = committed_replacement
1271 .logical_result()
1272 .transport_stable_id
1273 .ok_or_else(|| {
1274 anyhow::anyhow!(
1275 "browser WebRTC replacement has no stable ID"
1276 )
1277 })?;
1278 committed_replacement.finish(b"wasm-custom-transport-upgrade");
1279 inner
1280 .publish_committed_wasm_carrier_route(
1281 &attempt.connection_id,
1282 kind,
1283 replacement_transport_stable_id,
1284 )
1285 .await?;
1286 Ok::<(), anyhow::Error>(())
1287 }
1288 .await;
1289 if let Some(expires_at_ms) = attempt.inbound_authorization_expires_at_ms {
1290 if let Some(node) = inner.iroh_node.read().await.as_ref().cloned() {
1291 node.revoke_inbound_replacement_if_current(
1292 attempt.remote_endpoint_id,
1293 crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
1294 expires_at_ms,
1295 )
1296 .await;
1297 }
1298 }
1299 let _lifecycle = lifecycle.lock().await;
1300 if let Err(error) = result {
1301 let failure_code = wasm_carrier_failure_code(&error);
1302 web_sys::console::error_1(&JsValue::from_str(&format!(
1303 "[OpenRTC][WebRTC carrier] inbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
1304 attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
1305 )));
1306 fail_wasm_webrtc_attempt(owner.clone(), attempt, failure_code, true).await;
1307 return;
1308 }
1309 if !owner
1313 .wasm_webrtc_carrier_attempts
1314 .borrow()
1315 .get(&attempt.connection_id)
1316 .is_some_and(|current| {
1317 current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id
1318 })
1319 || !inner
1320 .current_wasm_peer_data_generation(&attempt.connection_id, None)
1321 .await
1322 .is_some_and(|current| {
1323 current.transport_generation
1324 == attempt.generation.transport_generation.saturating_add(1)
1325 })
1326 {
1327 return;
1328 }
1329 emit_wasm_carrier_action(
1330 &handler,
1331 serde_json::json!({
1332 "type": "selected",
1333 "connectionId": attempt.connection_id,
1334 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1335 "upgradeId": attempt.bootstrap.upgrade_id,
1336 "family": "iroh",
1337 "carrier": "webrtc",
1338 "transportGeneration": attempt.generation.transport_generation.saturating_add(1),
1339 "routeGeneration": 0,
1340 }),
1341 );
1342 });
1343 }
1344 }
1345
1346 #[cfg(feature = "transport-moq")]
1347 fn wasm_moq_carrier_namespaces(
1348 local_endpoint_id: &str,
1349 remote_endpoint_id: &str,
1350 carrier_session_id: &str,
1351 ) -> (String, String, &'static str) {
1352 let (first, second) = if local_endpoint_id <= remote_endpoint_id {
1353 (local_endpoint_id, remote_endpoint_id)
1354 } else {
1355 (remote_endpoint_id, local_endpoint_id)
1356 };
1357 let base = format!("openrtc/iroh-carrier/moq/{first}/{second}/{carrier_session_id}");
1358 (
1359 format!("{base}/from/{local_endpoint_id}"),
1360 format!("{base}/from/{remote_endpoint_id}"),
1361 "iroh-packets",
1362 )
1363 }
1364
1365 #[cfg(feature = "transport-moq")]
1366 async fn fail_wasm_moq_attempt(
1367 inner: Arc<Client>,
1368 attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
1369 sessions: Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
1370 handler: Rc<RefCell<Option<js_sys::Function>>>,
1371 attempt: WasmMoqCarrierAttempt,
1372 failure_code: &'static str,
1373 notify_peer: bool,
1374 ) {
1375 let kind = crate::client::IrohPathKind::Moq;
1376 if !inner
1377 .retire_wasm_carrier_upgrade(
1378 &attempt.connection_id,
1379 kind,
1380 &attempt.bootstrap.upgrade_id,
1381 attempt.generation,
1382 )
1383 .await
1384 {
1385 return;
1386 }
1387 if attempts
1388 .borrow()
1389 .get(&attempt.connection_id)
1390 .is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
1391 {
1392 attempts.borrow_mut().remove(&attempt.connection_id);
1393 }
1394 sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
1395 if notify_peer {
1396 if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
1397 &attempt.bootstrap,
1398 failure_code,
1399 ) {
1400 emit_wasm_carrier_action(
1401 &handler,
1402 serde_json::json!({
1403 "type": "send-control",
1404 "connectionId": attempt.connection_id,
1405 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1406 "envelope": failed,
1407 }),
1408 );
1409 }
1410 }
1411 emit_wasm_carrier_action(
1412 &handler,
1413 serde_json::json!({
1414 "type": "retire-moq",
1415 "connectionId": attempt.connection_id,
1416 "upgradeId": attempt.bootstrap.upgrade_id,
1417 "failureCode": failure_code,
1418 }),
1419 );
1420 let retry_pending = attempt.role == "initiator"
1421 && attempt.retry_count == 0
1422 && failure_code == "carrier-base-generation-stale";
1423 if attempt.role == "initiator" && !retry_pending {
1424 emit_wasm_carrier_action(
1425 &handler,
1426 serde_json::json!({
1427 "type": "advance-carrier",
1428 "connectionId": attempt.connection_id,
1429 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1430 "failedRoute": "moq",
1431 }),
1432 );
1433 }
1434 }
1435
1436 #[cfg(feature = "transport-moq")]
1437 async fn retire_selected_wasm_moq_carrier(
1438 inner: Arc<Client>,
1439 attempts: Rc<RefCell<HashMap<String, WasmMoqCarrierAttempt>>>,
1440 sessions: Rc<RefCell<HashMap<String, crate::wasm_moq_carrier::WasmMoqCarrierSession>>>,
1441 handler: Rc<RefCell<Option<js_sys::Function>>>,
1442 attempt: &WasmMoqCarrierAttempt,
1443 failure_code: &'static str,
1444 lifecycle_reason: &'static str,
1445 ) -> bool {
1446 if !inner
1447 .close_current_iroh_carrier_generation_with_reason(
1448 &attempt.connection_id,
1449 attempt.remote_endpoint_id,
1450 crate::client::IrohPathKind::Moq,
1451 attempt.generation.transport_generation.saturating_add(1),
1452 lifecycle_reason,
1453 )
1454 .await
1455 {
1456 return false;
1457 }
1458 web_sys::console::warn_1(&JsValue::from_str(&format!(
1459 "[OpenRTC][MoQ carrier] selected mechanism ended connection_id={} upgrade_id={} failure_code={failure_code}",
1460 attempt.connection_id, attempt.bootstrap.upgrade_id,
1461 )));
1462 if attempts
1463 .borrow()
1464 .get(&attempt.connection_id)
1465 .is_some_and(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
1466 {
1467 attempts.borrow_mut().remove(&attempt.connection_id);
1468 }
1469 sessions.borrow_mut().remove(&attempt.bootstrap.upgrade_id);
1470 emit_wasm_carrier_action(
1471 &handler,
1472 serde_json::json!({
1473 "type": "retire-moq",
1474 "connectionId": attempt.connection_id,
1475 "upgradeId": attempt.bootstrap.upgrade_id,
1476 "failureCode": failure_code,
1477 }),
1478 );
1479 true
1480 }
1481
1482 #[cfg(feature = "transport-moq")]
1483 impl WasmClient {
1484 fn schedule_wasm_moq_carrier_watchdog(
1485 &self,
1486 bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
1487 ) {
1488 let inner = self.inner.clone();
1489 let attempts = self.wasm_moq_carrier_attempts.clone();
1490 let sessions = self.wasm_moq_carrier_sessions.clone();
1491 let handler = self.wasm_carrier_action_handler.clone();
1492 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
1493 spawn_local(async move {
1494 gloo_timers::future::sleep(std::time::Duration::from_secs(30)).await;
1495 let _lifecycle = lifecycle.lock().await;
1496 let attempt = attempts
1497 .borrow()
1498 .values()
1499 .find(|attempt| attempt.bootstrap.upgrade_id == bootstrap.upgrade_id)
1500 .cloned();
1501 if let Some(attempt) = attempt {
1502 if !inner
1503 .wasm_carrier_upgrade_is_current(
1504 &attempt.connection_id,
1505 crate::client::IrohPathKind::Moq,
1506 &attempt.bootstrap.upgrade_id,
1507 attempt.generation,
1508 )
1509 .await
1510 {
1511 return;
1512 }
1513 fail_wasm_moq_attempt(
1514 inner,
1515 attempts,
1516 sessions,
1517 handler,
1518 attempt,
1519 "carrier-timeout",
1520 true,
1521 )
1522 .await;
1523 }
1524 });
1525 }
1526
1527 async fn fail_wasm_moq_carrier_attempt(
1528 &self,
1529 attempt: WasmMoqCarrierAttempt,
1530 failure_code: &'static str,
1531 notify_peer: bool,
1532 ) {
1533 fail_wasm_moq_attempt(
1534 self.inner.clone(),
1535 self.wasm_moq_carrier_attempts.clone(),
1536 self.wasm_moq_carrier_sessions.clone(),
1537 self.wasm_carrier_action_handler.clone(),
1538 attempt,
1539 failure_code,
1540 notify_peer,
1541 )
1542 .await;
1543 }
1544
1545 fn spawn_outbound_wasm_moq_carrier_completion(&self, attempt: WasmMoqCarrierAttempt) {
1546 let inner = self.inner.clone();
1547 let attempts = self.wasm_moq_carrier_attempts.clone();
1548 let sessions = self.wasm_moq_carrier_sessions.clone();
1549 let handler = self.wasm_carrier_action_handler.clone();
1550 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
1551 spawn_local(async move {
1552 let kind = crate::client::IrohPathKind::Moq;
1553 let result = inner
1554 .complete_outbound_wasm_carrier_upgrade(
1555 &attempt.connection_id,
1556 &attempt.remote_endpoint_id.to_string(),
1557 &attempt.bootstrap.upgrade_id,
1558 attempt.generation,
1559 kind,
1560 )
1561 .await;
1562 let _lifecycle = lifecycle.lock().await;
1563 if let Err(error) = result {
1564 let failure_code = wasm_carrier_failure_code(&error);
1565 web_sys::console::error_1(&JsValue::from_str(&format!(
1566 "[OpenRTC][MoQ carrier] outbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
1567 attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
1568 )));
1569 fail_wasm_moq_attempt(
1570 inner,
1571 attempts,
1572 sessions,
1573 handler,
1574 attempt,
1575 failure_code,
1576 true,
1577 )
1578 .await;
1579 return;
1580 }
1581 if !inner
1582 .retire_wasm_carrier_upgrade(
1583 &attempt.connection_id,
1584 kind,
1585 &attempt.bootstrap.upgrade_id,
1586 attempt.generation,
1587 )
1588 .await
1589 {
1590 return;
1591 }
1592 emit_wasm_carrier_action(
1593 &handler,
1594 serde_json::json!({
1595 "type": "selected",
1596 "connectionId": attempt.connection_id,
1597 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1598 "upgradeId": attempt.bootstrap.upgrade_id,
1599 "family": "iroh",
1600 "carrier": "moq",
1601 "transportGeneration": attempt.generation.transport_generation.saturating_add(1),
1602 "routeGeneration": 0,
1603 }),
1604 );
1605 });
1606 }
1607
1608 fn spawn_inbound_wasm_moq_carrier_completion(&self, attempt: WasmMoqCarrierAttempt) {
1609 let inner = self.inner.clone();
1610 let attempts = self.wasm_moq_carrier_attempts.clone();
1611 let sessions = self.wasm_moq_carrier_sessions.clone();
1612 let handler = self.wasm_carrier_action_handler.clone();
1613 let lifecycle = self.wasm_carrier_peer_lifecycle.clone();
1614 spawn_local(async move {
1615 let kind = crate::client::IrohPathKind::Moq;
1616 let node = inner.iroh_node.read().await.as_ref().cloned();
1617 let result = async {
1618 let node = node.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
1619 let authorization_expires_at_ms = attempt
1620 .inbound_authorization_expires_at_ms
1621 .ok_or_else(|| anyhow::anyhow!("browser MoQ inbound authorization is missing"))?;
1622 let candidate = node
1623 .wait_for_inbound_replacement_candidate(
1624 attempt.remote_endpoint_id,
1625 crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
1626 authorization_expires_at_ms,
1627 std::time::Duration::from_secs(40),
1628 )
1629 .await?;
1630 let policy_epoch = inner
1631 .require_current_wasm_carrier_upgrade_fence(
1632 &attempt.connection_id,
1633 kind,
1634 &attempt.bootstrap.upgrade_id,
1635 attempt.generation,
1636 )
1637 .await?;
1638 let authorization_fence = inner
1639 .capture_wasm_iroh_carrier_authorization_fence(
1640 &attempt.connection_id,
1641 attempt.generation,
1642 kind,
1643 policy_epoch,
1644 )?;
1645 let proof = inner.wasm_candidate_proof_probe(
1646 &attempt.connection_id,
1647 &attempt.bootstrap.upgrade_id,
1648 attempt.bootstrap.base.transport_generation,
1649 attempt.bootstrap.base.route_generation,
1650 kind,
1651 crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
1652 )?;
1653 let (send, probe, ack) = inner
1654 .receive_inbound_wasm_carrier_candidate_proof(
1655 &attempt.connection_id,
1656 &candidate,
1657 &proof,
1658 )
1659 .await?;
1660 anyhow::ensure!(
1661 inner
1662 .current_wasm_peer_data_generation(&attempt.connection_id, None)
1663 .await
1664 == Some(attempt.generation),
1665 "browser MoQ inbound carrier base generation changed before candidate acknowledgement"
1666 );
1667 inner
1668 .send_inbound_wasm_carrier_candidate_ack(send, &ack)
1669 .await?;
1670 let (commit_send, committed) = inner
1671 .receive_inbound_wasm_carrier_commit(
1672 &attempt.connection_id,
1673 &candidate,
1674 &probe,
1675 )
1676 .await?;
1677 let committed_replacement = inner
1678 .commit_proven_wasm_carrier_candidate(
1679 &attempt.connection_id,
1680 &attempt.bootstrap.upgrade_id,
1681 attempt.generation,
1682 kind,
1683 authorization_fence,
1684 candidate,
1685 )
1686 .await?;
1687 inner
1688 .send_inbound_wasm_carrier_candidate_ack(commit_send, &committed)
1689 .await?;
1690 let replacement_transport_stable_id = committed_replacement
1691 .logical_result()
1692 .transport_stable_id
1693 .ok_or_else(|| {
1694 anyhow::anyhow!("browser MoQ replacement has no stable ID")
1695 })?;
1696 committed_replacement.finish(b"wasm-custom-transport-upgrade");
1697 inner
1698 .publish_committed_wasm_carrier_route(
1699 &attempt.connection_id,
1700 kind,
1701 replacement_transport_stable_id,
1702 )
1703 .await?;
1704 Ok::<(), anyhow::Error>(())
1705 }
1706 .await;
1707 if let Some(expires_at_ms) = attempt.inbound_authorization_expires_at_ms {
1708 if let Some(node) = inner.iroh_node.read().await.as_ref().cloned() {
1709 node.revoke_inbound_replacement_if_current(
1710 attempt.remote_endpoint_id,
1711 crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
1712 expires_at_ms,
1713 )
1714 .await;
1715 }
1716 }
1717 let _lifecycle = lifecycle.lock().await;
1718 if let Err(error) = result {
1719 let failure_code = wasm_carrier_failure_code(&error);
1720 web_sys::console::error_1(&JsValue::from_str(&format!(
1721 "[OpenRTC][MoQ carrier] inbound candidate completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
1722 attempt.connection_id, attempt.bootstrap.upgrade_id, failure_code,
1723 )));
1724 fail_wasm_moq_attempt(
1725 inner,
1726 attempts,
1727 sessions,
1728 handler,
1729 attempt,
1730 failure_code,
1731 true,
1732 )
1733 .await;
1734 return;
1735 }
1736 if !inner
1737 .retire_wasm_carrier_upgrade(
1738 &attempt.connection_id,
1739 kind,
1740 &attempt.bootstrap.upgrade_id,
1741 attempt.generation,
1742 )
1743 .await
1744 {
1745 return;
1746 }
1747 emit_wasm_carrier_action(
1748 &handler,
1749 serde_json::json!({
1750 "type": "selected",
1751 "connectionId": attempt.connection_id,
1752 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
1753 "upgradeId": attempt.bootstrap.upgrade_id,
1754 "family": "iroh",
1755 "carrier": "moq",
1756 "transportGeneration": attempt.generation.transport_generation.saturating_add(1),
1757 "routeGeneration": 0,
1758 }),
1759 );
1760 });
1761 }
1762
1763 async fn handle_wasm_moq_bootstrap(
1764 &self,
1765 connection_id: String,
1766 remote_endpoint_id: String,
1767 bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
1768 ) -> Result<bool, JsValue> {
1769 let endpoint_id = remote_endpoint_id
1770 .parse::<iroh::EndpointId>()
1771 .map_err(|error| JsValue::from_str(&error.to_string()))?;
1772 let kind = crate::client::IrohPathKind::Moq;
1773 match bootstrap.action {
1774 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
1775 let local_endpoint_id = self
1776 .inner
1777 .current_node_id()
1778 .await
1779 .ok_or_else(|| JsValue::from_str("local endpoint id is unavailable"))?;
1780 if local_endpoint_id.as_str() >= remote_endpoint_id.as_str()
1781 || !self.inner.is_moq_carrier_enabled().await
1782 {
1783 return Ok(true);
1784 }
1785 if !crate::iroh_connection_policy::custom_carrier_base_allows(
1786 self.inner.iroh_path_kind(&remote_endpoint_id).await,
1787 crate::client::IrohPathKind::Moq,
1788 false,
1789 ) {
1790 return Ok(true);
1791 }
1792 let base_connection = self
1793 .inner
1794 .get_connection(endpoint_id)
1795 .await
1796 .ok_or_else(|| JsValue::from_str("MoQ carrier base is unavailable"))?;
1797 let generation = self
1798 .inner
1799 .current_wasm_peer_data_generation(&connection_id, None)
1800 .await
1801 .ok_or_else(|| {
1802 JsValue::from_str("MoQ carrier generation is unavailable")
1803 })?;
1804 if crate::transport_generation::for_connection(&base_connection)
1805 != generation.transport_stable_id
1806 {
1807 let failed =
1808 crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
1809 &bootstrap,
1810 "carrier-base-generation-stale",
1811 )
1812 .map_err(|error| JsValue::from_str(&error.to_string()))?;
1813 emit_wasm_carrier_action(
1814 &self.wasm_carrier_action_handler,
1815 serde_json::json!({
1816 "type": "send-control",
1817 "connectionId": connection_id,
1818 "remoteEndpointId": remote_endpoint_id,
1819 "envelope": failed,
1820 }),
1821 );
1822 return Ok(true);
1823 }
1824 if !matches!(
1825 self.inner
1826 .reserve_wasm_carrier_upgrade(
1827 &connection_id,
1828 kind,
1829 &bootstrap.upgrade_id,
1830 generation,
1831 )
1832 .await,
1833 crate::client::WasmCarrierUpgradeReservation::Reserved
1834 ) {
1835 return Ok(true);
1836 }
1837 let node = self
1838 .inner
1839 .iroh_node
1840 .read()
1841 .await
1842 .as_ref()
1843 .cloned()
1844 .ok_or_else(|| JsValue::from_str("Iroh node is unavailable"))?;
1845 let authorization_expiry = node
1846 .authorize_pending_inbound_replacement(
1847 endpoint_id,
1848 crate::iroh_carrier_kind::EXPERIMENTAL_MOQ_TRANSPORT_ID,
1849 std::time::Duration::from_secs(45),
1850 )
1851 .await;
1852 let previous = self.wasm_moq_carrier_attempts.borrow_mut().insert(
1853 connection_id.clone(),
1854 WasmMoqCarrierAttempt {
1855 connection_id: connection_id.clone(),
1856 remote_endpoint_id: endpoint_id,
1857 bootstrap: bootstrap.clone(),
1858 generation,
1859 role: "responder",
1860 prepared: false,
1861 retry_sent: false,
1862 retry_count: bootstrap.attempt.saturating_sub(1),
1863 inbound_authorization_expires_at_ms: Some(authorization_expiry),
1864 },
1865 );
1866 if let Some(previous) = previous {
1867 self.wasm_moq_carrier_sessions
1868 .borrow_mut()
1869 .remove(&previous.bootstrap.upgrade_id);
1870 }
1871 let (publish_namespace, subscribe_namespace, track_name) =
1872 wasm_moq_carrier_namespaces(
1873 &local_endpoint_id,
1874 &remote_endpoint_id,
1875 &bootstrap.carrier_session_id,
1876 );
1877 emit_wasm_carrier_action(
1878 &self.wasm_carrier_action_handler,
1879 serde_json::json!({
1880 "type": "prepare-moq",
1881 "connectionId": connection_id,
1882 "remoteEndpointId": remote_endpoint_id,
1883 "role": "responder",
1884 "upgradeId": bootstrap.upgrade_id,
1885 "carrierSessionId": bootstrap.carrier_session_id,
1886 "transportGeneration": generation.transport_generation.saturating_add(1),
1887 "publishNamespace": publish_namespace,
1888 "subscribeNamespace": subscribe_namespace,
1889 "trackName": track_name,
1890 }),
1891 );
1892 self.schedule_wasm_moq_carrier_watchdog(bootstrap);
1893 }
1894 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
1895 let attempt = self
1896 .wasm_moq_carrier_attempts
1897 .borrow()
1898 .get(&connection_id)
1899 .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
1900 .cloned();
1901 if let Some(attempt) = attempt {
1902 emit_wasm_carrier_action(
1903 &self.wasm_carrier_action_handler,
1904 serde_json::json!({
1905 "type": "activate-moq",
1906 "connectionId": attempt.connection_id,
1907 "upgradeId": attempt.bootstrap.upgrade_id,
1908 }),
1909 );
1910 }
1911 }
1912 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
1913 let attempt = self
1914 .wasm_moq_carrier_attempts
1915 .borrow()
1916 .get(&connection_id)
1917 .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
1918 .cloned();
1919 if let Some(attempt) = attempt {
1920 let failure_code = crate::client::wasm_peer_carrier_failure_code(
1921 bootstrap.failure_code.as_deref(),
1922 );
1923 if crate::client::should_rearm_wasm_carrier_event_retry(
1924 failure_code,
1925 attempt.role == "initiator",
1926 false,
1927 ) {
1928 if let Some(current) = self
1929 .wasm_moq_carrier_attempts
1930 .borrow_mut()
1931 .get_mut(&connection_id)
1932 .filter(|current| {
1933 current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id
1934 })
1935 {
1936 current.retry_sent = false;
1942 }
1943 return Ok(true);
1944 }
1945 let should_retry = attempt.role == "initiator"
1946 && attempt.retry_count == 0
1947 && failure_code == "carrier-base-generation-stale";
1948 let remote_endpoint_id = attempt.remote_endpoint_id.to_string();
1949 self.fail_wasm_moq_carrier_attempt(attempt, failure_code, false)
1950 .await;
1951 if should_retry {
1952 gloo_timers::future::sleep(std::time::Duration::from_millis(500)).await;
1956 let _ = self
1957 .begin_iroh_moq_carrier_attempt(
1958 connection_id,
1959 remote_endpoint_id,
1960 1,
1961 )
1962 .await;
1963 }
1964 }
1965 }
1966 }
1967 Ok(true)
1968 }
1969 }
1970
1971 #[wasm_bindgen]
1972 impl WasmClient {
1973 pub fn connection_state_owner_id(&self) -> String {
1975 self.inner.wasm_connection_state_owner_id.clone()
1976 }
1977
1978 #[wasm_bindgen(constructor)]
1982 pub fn new(api_key: String) -> Result<WasmClient, JsValue> {
1983 let api_key = crate::validate_api_key(&api_key)
1984 .map_err(|error| JsValue::from_str(&error.to_string()))?;
1985 let app_tag = crate::app_tag_from_api_key(api_key);
1986 let identity_credential: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
1987 let credential_state = identity_credential.clone();
1988 let identity_credential_provider = Box::new(move || -> Option<String> {
1989 credential_state.lock().ok().and_then(|guard| guard.clone())
1990 });
1991
1992 Ok(Self {
1993 inner: Arc::new(Client::new_provider_neutral(
1994 app_tag,
1995 identity_credential_provider,
1996 )),
1997 portable_media: RefCell::new(crate::media::PortableMediaSession::default()),
1998 broadcast_sessions: RefCell::new(HashMap::new()),
1999 broadcast_signers: RefCell::new(HashMap::new()),
2000 ticket_meshes: RefCell::new(HashMap::new()),
2001 ticket_mesh_starts: RefCell::new(HashSet::new()),
2002 identity_credential,
2003 last_auth_log: Arc::new(Mutex::new(None)),
2004 #[cfg(feature = "iroh-protocols-wasm")]
2005 persistent_protocols: Rc::new(tokio::sync::Mutex::new(None)),
2006 #[cfg(feature = "managed-group-encryption")]
2007 managed_group_controllers: Rc::new(RefCell::new(HashMap::new())),
2008 #[cfg(feature = "transport-webrtc")]
2009 wasm_webrtc_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
2010 #[cfg(feature = "transport-webrtc")]
2011 wasm_webrtc_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
2012 #[cfg(feature = "transport-webrtc")]
2013 webrtc_availability: Rc::new(RefCell::new(Default::default())),
2014 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2015 wasm_carrier_action_handler: Rc::new(RefCell::new(None)),
2016 #[cfg(feature = "transport-moq")]
2017 wasm_moq_carrier_sessions: Rc::new(RefCell::new(HashMap::new())),
2018 #[cfg(feature = "transport-moq")]
2019 wasm_moq_carrier_attempts: Rc::new(RefCell::new(HashMap::new())),
2020 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2021 wasm_remote_carrier_capabilities: Rc::new(RefCell::new(HashMap::new())),
2022 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2023 wasm_carrier_peer_lifecycle: Rc::new(tokio::sync::Mutex::new(())),
2024 })
2025 }
2026
2027 #[cfg(feature = "managed-group-encryption")]
2031 #[wasm_bindgen(js_name = __initManagedRoomGroup)]
2032 pub fn init_managed_room_group(
2033 &self,
2034 avenue_key: String,
2035 device_id: String,
2036 wrapping_key: Vec<u8>,
2037 sealed_state: Option<Vec<u8>>,
2038 ) -> Result<JsValue, JsValue> {
2039 let wrapping_key: [u8; 32] = wrapping_key
2040 .try_into()
2041 .map_err(|_| JsValue::from_str("managed room wrapping key must be 32 bytes"))?;
2042 let mut controllers = self.managed_group_controllers.borrow_mut();
2043 if !controllers.contains_key(&avenue_key) {
2044 let controller = crate::managed_group_controller::ManagedGroupController::new(
2045 device_id,
2046 wrapping_key,
2047 sealed_state.as_deref(),
2048 )
2049 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2050 controllers.insert(avenue_key.clone(), controller);
2051 }
2052 let controller = controllers
2053 .get(&avenue_key)
2054 .ok_or_else(|| JsValue::from_str("managed room group initialization failed"))?;
2055 let action = controller
2056 .publish_key_package()
2057 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2058 serde::Serialize::serialize(&action, &serde_wasm_bindgen::Serializer::json_compatible())
2059 .map_err(|error| JsValue::from_str(&error.to_string()))
2060 }
2061
2062 #[cfg(feature = "managed-group-encryption")]
2063 #[wasm_bindgen(js_name = __handleManagedRoomPreparePage)]
2064 pub fn handle_managed_room_prepare_page(
2065 &self,
2066 avenue_key: String,
2067 page: JsValue,
2068 ) -> Result<JsValue, JsValue> {
2069 let page = serde_wasm_bindgen::from_value(page)
2070 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2071 let mut controllers = self.managed_group_controllers.borrow_mut();
2072 let controller = controllers
2073 .get_mut(&avenue_key)
2074 .ok_or_else(|| JsValue::from_str("managed room group is not initialized"))?;
2075 let actions = controller
2076 .handle_prepare_page(page)
2077 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2078 serde::Serialize::serialize(
2079 &actions,
2080 &serde_wasm_bindgen::Serializer::json_compatible(),
2081 )
2082 .map_err(|error| JsValue::from_str(&error.to_string()))
2083 }
2084
2085 #[cfg(feature = "managed-group-encryption")]
2086 #[wasm_bindgen(js_name = __handleManagedRoomArtifactChunk)]
2087 pub fn handle_managed_room_artifact_chunk(
2088 &self,
2089 avenue_key: String,
2090 chunk: JsValue,
2091 ) -> Result<JsValue, JsValue> {
2092 let chunk = serde_wasm_bindgen::from_value(chunk)
2093 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2094 let mut controllers = self.managed_group_controllers.borrow_mut();
2095 let controller = controllers
2096 .get_mut(&avenue_key)
2097 .ok_or_else(|| JsValue::from_str("managed room group is not initialized"))?;
2098 let actions = controller
2099 .handle_artifact_chunk(chunk)
2100 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2101 serde::Serialize::serialize(
2102 &actions,
2103 &serde_wasm_bindgen::Serializer::json_compatible(),
2104 )
2105 .map_err(|error| JsValue::from_str(&error.to_string()))
2106 }
2107
2108 #[cfg(feature = "managed-group-encryption")]
2109 #[wasm_bindgen(js_name = __sealManagedRoomPayload)]
2110 #[allow(clippy::too_many_arguments)]
2111 pub fn seal_managed_room_payload(
2112 &self,
2113 avenue_key: String,
2114 architecture_epoch: u64,
2115 encryption_epoch: u64,
2116 message_id: String,
2117 channel: String,
2118 priority: u8,
2119 zone_id: Option<String>,
2120 payload: Vec<u8>,
2121 ) -> Result<JsValue, JsValue> {
2122 let protected = self
2123 .managed_group_controllers
2124 .borrow_mut()
2125 .get_mut(&avenue_key)
2126 .ok_or_else(|| JsValue::from_str("managed room group is not initialized"))?
2127 .seal_payload(
2128 architecture_epoch,
2129 encryption_epoch,
2130 &message_id,
2131 &channel,
2132 priority,
2133 zone_id.as_deref(),
2134 &payload,
2135 )
2136 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2137 serde_wasm_bindgen::to_value(&serde_json::json!({
2138 "payload": base64::Engine::encode(
2139 &base64::engine::general_purpose::URL_SAFE_NO_PAD,
2140 protected.data,
2141 ),
2142 "sealedState": base64::Engine::encode(
2143 &base64::engine::general_purpose::URL_SAFE_NO_PAD,
2144 protected.sealed_state,
2145 ),
2146 }))
2147 .map_err(|error| JsValue::from_str(&error.to_string()))
2148 }
2149
2150 #[cfg(feature = "managed-group-encryption")]
2151 #[wasm_bindgen(js_name = __openManagedRoomPayload)]
2152 #[allow(clippy::too_many_arguments)]
2153 pub fn open_managed_room_payload(
2154 &self,
2155 avenue_key: String,
2156 architecture_epoch: u64,
2157 encryption_epoch: u64,
2158 message_id: String,
2159 channel: String,
2160 priority: u8,
2161 zone_id: Option<String>,
2162 ciphertext: Vec<u8>,
2163 ) -> Result<JsValue, JsValue> {
2164 let protected = self
2165 .managed_group_controllers
2166 .borrow_mut()
2167 .get_mut(&avenue_key)
2168 .ok_or_else(|| JsValue::from_str("managed room group is not initialized"))?
2169 .open_payload(
2170 architecture_epoch,
2171 encryption_epoch,
2172 &message_id,
2173 &channel,
2174 priority,
2175 zone_id.as_deref(),
2176 &ciphertext,
2177 )
2178 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
2179 serde_wasm_bindgen::to_value(&serde_json::json!({
2180 "payload": base64::Engine::encode(
2181 &base64::engine::general_purpose::URL_SAFE_NO_PAD,
2182 protected.data,
2183 ),
2184 "sealedState": base64::Engine::encode(
2185 &base64::engine::general_purpose::URL_SAFE_NO_PAD,
2186 protected.sealed_state,
2187 ),
2188 }))
2189 .map_err(|error| JsValue::from_str(&error.to_string()))
2190 }
2191
2192 #[cfg(feature = "managed-group-encryption")]
2193 #[wasm_bindgen(js_name = __forgetManagedRoomGroup)]
2194 pub fn forget_managed_room_group(&self, avenue_key: String) {
2195 self.managed_group_controllers
2196 .borrow_mut()
2197 .remove(&avenue_key);
2198 }
2199
2200 #[wasm_bindgen(js_name = __prepareOpenRtcBroadcastGrant)]
2204 pub fn prepare_openrtc_broadcast_grant(
2205 &self,
2206 grant_token: String,
2207 issuer_public_key: Vec<u8>,
2208 now_ms: f64,
2209 ) -> Result<JsValue, JsValue> {
2210 let issuer_public_key: [u8; 32] = issuer_public_key
2211 .try_into()
2212 .map_err(|_| JsValue::from_str("broadcast issuer key must be 32 bytes"))?;
2213 let issuer = ed25519_dalek::VerifyingKey::from_bytes(&issuer_public_key)
2214 .map_err(|_| JsValue::from_str("broadcast issuer key is invalid"))?;
2215 if !now_ms.is_finite()
2216 || now_ms < 0.0
2217 || now_ms.fract() != 0.0
2218 || now_ms > crate::media::MAX_JAVASCRIPT_SAFE_INTEGER as f64
2219 {
2220 return Err(JsValue::from_str(
2221 "broadcast now must be a non-negative JavaScript-safe integer",
2222 ));
2223 }
2224 let challenge = self
2225 .inner
2226 .broadcasts()
2227 .prepare_grant_verification(&grant_token, &issuer, now_ms as u64)
2228 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2229 let object = js_sys::Object::new();
2230 js_sys::Reflect::set(
2231 &object,
2232 &JsValue::from_str("handle"),
2233 &JsValue::from_str(&challenge.handle),
2234 )?;
2235 js_sys::Reflect::set(
2236 &object,
2237 &JsValue::from_str("signingBytes"),
2238 &js_sys::Uint8Array::from(challenge.signing_bytes.as_slice()),
2239 )?;
2240 js_sys::Reflect::set(
2241 &object,
2242 &JsValue::from_str("expiresAtMs"),
2243 &JsValue::from_f64(challenge.expires_at_ms as f64),
2244 )?;
2245 Ok(object.into())
2246 }
2247
2248 #[wasm_bindgen(js_name = __openOpenRtcBroadcast)]
2252 pub fn open_openrtc_broadcast(
2253 &self,
2254 grant_token: String,
2255 issuer_public_key: Vec<u8>,
2256 binding_challenge_handle: String,
2257 binding_signature: Vec<u8>,
2258 publication_signer_handle: Option<String>,
2259 now_ms: f64,
2260 ) -> Result<JsValue, JsValue> {
2261 let issuer_public_key: [u8; 32] = issuer_public_key
2262 .try_into()
2263 .map_err(|_| JsValue::from_str("broadcast issuer key must be 32 bytes"))?;
2264 let issuer = ed25519_dalek::VerifyingKey::from_bytes(&issuer_public_key)
2265 .map_err(|_| JsValue::from_str("broadcast issuer key is invalid"))?;
2266 if !now_ms.is_finite()
2267 || now_ms < 0.0
2268 || now_ms.fract() != 0.0
2269 || now_ms > crate::media::MAX_JAVASCRIPT_SAFE_INTEGER as f64
2270 {
2271 return Err(JsValue::from_str(
2272 "broadcast now must be a non-negative JavaScript-safe integer",
2273 ));
2274 }
2275 let now_ms = now_ms as u64;
2276 let grant = self
2277 .inner
2278 .broadcasts()
2279 .complete_grant_verification(
2280 &grant_token,
2281 &issuer,
2282 &binding_challenge_handle,
2283 &binding_signature,
2284 now_ms,
2285 )
2286 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2287 let grant_generation = grant.grant_generation();
2288 let session = match publication_signer_handle {
2289 Some(handle) => {
2290 let signers = self.broadcast_signers.borrow();
2291 let signer = signers
2292 .get(&handle)
2293 .ok_or_else(|| JsValue::from_str("broadcast signer handle is invalid"))?;
2294 self.inner
2295 .broadcasts()
2296 .open_publisher(grant, signer, now_ms)
2297 }
2298 None => self.inner.broadcasts().open(grant, now_ms),
2299 }
2300 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2301 let handle_id = format!("{}:{grant_generation}", session.id());
2302 let result = serde_json::json!({
2303 "id": session.id(),
2304 "handleId": handle_id,
2305 "role": session.role(),
2306 "state": session.state(),
2307 "budget": session.budget(),
2308 "sourceSlots": session.source_slots(),
2309 });
2310 self.broadcast_sessions
2311 .borrow_mut()
2312 .insert(handle_id, session);
2313 serde::Serialize::serialize(&result, &serde_wasm_bindgen::Serializer::json_compatible())
2314 .map_err(|error| JsValue::from_str(&error.to_string()))
2315 }
2316
2317 #[wasm_bindgen(js_name = __createOpenRtcBroadcastPublisherSigner)]
2321 pub fn create_openrtc_broadcast_publisher_signer(&self) -> Result<JsValue, JsValue> {
2322 let signer = crate::broadcast::BroadcastPublisherSigner::generate()
2323 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2324 let mut handle_bytes = [0_u8; 16];
2325 getrandom::getrandom(&mut handle_bytes)
2326 .map_err(|_| JsValue::from_str("broadcast signer handle entropy failed"))?;
2327 let handle = hex::encode(handle_bytes);
2328 let public_key = signer.verifying_key();
2329 let object = js_sys::Object::new();
2330 js_sys::Reflect::set(
2331 &object,
2332 &JsValue::from_str("handle"),
2333 &JsValue::from_str(&handle),
2334 )?;
2335 js_sys::Reflect::set(
2336 &object,
2337 &JsValue::from_str("publicKey"),
2338 &js_sys::Uint8Array::from(public_key.as_slice()),
2339 )?;
2340 self.broadcast_signers.borrow_mut().insert(handle, signer);
2341 Ok(object.into())
2342 }
2343
2344 #[wasm_bindgen(js_name = __releaseOpenRtcBroadcastPublisherSigner)]
2345 pub fn release_openrtc_broadcast_publisher_signer(&self, handle: String) -> bool {
2346 self.broadcast_signers
2347 .borrow_mut()
2348 .remove(&handle)
2349 .is_some()
2350 }
2351
2352 #[wasm_bindgen(js_name = __takeOpenRtcBroadcastActions)]
2353 pub fn take_openrtc_broadcast_actions(
2354 &self,
2355 handle_id: String,
2356 max: usize,
2357 ) -> Result<JsValue, JsValue> {
2358 let sessions = self.broadcast_sessions.borrow();
2359 let session = sessions
2360 .get(&handle_id)
2361 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?;
2362 serde::Serialize::serialize(
2363 &session.take_actions(max),
2364 &serde_wasm_bindgen::Serializer::json_compatible(),
2365 )
2366 .map_err(|error| JsValue::from_str(&error.to_string()))
2367 }
2368
2369 #[wasm_bindgen(js_name = __getOpenRtcBroadcastStats)]
2370 pub fn get_openrtc_broadcast_stats(&self, handle_id: String) -> Result<JsValue, JsValue> {
2371 let sessions = self.broadcast_sessions.borrow();
2372 let stats = sessions
2373 .get(&handle_id)
2374 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2375 .stats();
2376 serde::Serialize::serialize(&stats, &serde_wasm_bindgen::Serializer::json_compatible())
2377 .map_err(|error| JsValue::from_str(&error.to_string()))
2378 }
2379
2380 #[wasm_bindgen(js_name = __observeOpenRtcBroadcastAdapter)]
2381 pub fn observe_openrtc_broadcast_adapter(
2382 &self,
2383 handle_id: String,
2384 observation: JsValue,
2385 ) -> Result<(), JsValue> {
2386 let observation = serde_wasm_bindgen::from_value(observation)
2387 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2388 self.broadcast_sessions
2389 .borrow()
2390 .get(&handle_id)
2391 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2392 .observe(observation)
2393 .map_err(|error| JsValue::from_str(&error.to_string()))
2394 }
2395
2396 #[wasm_bindgen(js_name = __closeOpenRtcBroadcast)]
2397 pub fn close_openrtc_broadcast(&self, handle_id: String) -> Result<(), JsValue> {
2398 let session = self
2399 .broadcast_sessions
2400 .borrow_mut()
2401 .remove(&handle_id)
2402 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?;
2403 session.close();
2404 Ok(())
2405 }
2406
2407 #[wasm_bindgen(js_name = __beginOpenRtcBroadcastPublication)]
2410 #[allow(clippy::too_many_arguments)]
2411 pub fn begin_openrtc_broadcast_publication(
2412 &self,
2413 handle_id: String,
2414 source_slot: String,
2415 publication_id: Option<String>,
2416 kind: String,
2417 codec: String,
2418 clock_rate: u32,
2419 coded_width: Option<u32>,
2420 coded_height: Option<u32>,
2421 channels: Option<u16>,
2422 ) -> Result<JsValue, JsValue> {
2423 let publication_id = match publication_id {
2424 Some(value) if !value.is_empty() => {
2425 value
2426 .parse()
2427 .map_err(|error: crate::media::MediaProtocolError| {
2428 JsValue::from_str(&error.to_string())
2429 })?
2430 }
2431 _ => {
2432 let mut bytes = [0_u8; 16];
2433 getrandom::getrandom(&mut bytes)
2434 .map_err(|_| JsValue::from_str("media publication entropy failed"))?;
2435 crate::media::PublicationId::from_bytes(bytes)
2436 }
2437 };
2438 let kind: crate::media::MediaKind =
2439 kind.parse()
2440 .map_err(|error: crate::media::MediaProtocolError| {
2441 JsValue::from_str(&error.to_string())
2442 })?;
2443 if !matches!(
2444 kind,
2445 crate::media::MediaKind::Audio | crate::media::MediaKind::Video
2446 ) {
2447 return Err(JsValue::from_str(
2448 "browser broadcast publications must be audio or video",
2449 ));
2450 }
2451 let codec: crate::media::MediaCodec =
2452 codec
2453 .parse()
2454 .map_err(|error: crate::media::MediaProtocolError| {
2455 JsValue::from_str(&error.to_string())
2456 })?;
2457 let publication = self
2458 .broadcast_sessions
2459 .borrow()
2460 .get(&handle_id)
2461 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2462 .begin_publication(
2463 &source_slot,
2464 crate::media::MediaPublicationConfig {
2465 publication_id,
2466 media_generation: 0,
2467 kind,
2468 codec,
2469 clock_rate,
2470 coded_width,
2471 coded_height,
2472 channels,
2473 },
2474 )
2475 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2476 serde::Serialize::serialize(
2477 &serde_json::json!({
2478 "publicationId": publication.publication_id.to_string(),
2479 "mediaGeneration": publication.media_generation,
2480 "control": Vec::<u8>::new(),
2481 }),
2482 &serde_wasm_bindgen::Serializer::json_compatible(),
2483 )
2484 .map_err(|error| JsValue::from_str(&error.to_string()))
2485 }
2486
2487 #[wasm_bindgen(js_name = __publishOpenRtcBroadcastSample)]
2490 #[allow(clippy::too_many_arguments)]
2491 pub fn publish_openrtc_broadcast_sample(
2492 &self,
2493 handle_id: String,
2494 publication_id: String,
2495 timestamp_us: f64,
2496 duration_us: u32,
2497 keyframe: bool,
2498 discardable: bool,
2499 payload: Vec<u8>,
2500 ) -> Result<(), JsValue> {
2501 if !timestamp_us.is_finite()
2502 || timestamp_us < 0.0
2503 || timestamp_us.fract() != 0.0
2504 || timestamp_us > crate::media::MAX_JAVASCRIPT_SAFE_INTEGER as f64
2505 {
2506 return Err(JsValue::from_str(
2507 "media timestamp must be a non-negative JavaScript-safe integer",
2508 ));
2509 }
2510 let publication_id =
2511 publication_id
2512 .parse()
2513 .map_err(|error: crate::media::MediaProtocolError| {
2514 JsValue::from_str(&error.to_string())
2515 })?;
2516 self.broadcast_sessions
2517 .borrow()
2518 .get(&handle_id)
2519 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2520 .publish(
2521 publication_id,
2522 crate::media::EncodedMediaSample {
2523 timestamp_us: timestamp_us as u64,
2524 duration_us,
2525 keyframe,
2526 discardable,
2527 payload,
2528 },
2529 )
2530 .map_err(|error| JsValue::from_str(&error.to_string()))
2531 }
2532
2533 #[wasm_bindgen(js_name = __retireOpenRtcBroadcastPublication)]
2534 pub fn retire_openrtc_broadcast_publication(
2535 &self,
2536 handle_id: String,
2537 publication_id: String,
2538 ) -> Result<(), JsValue> {
2539 let publication_id =
2540 publication_id
2541 .parse()
2542 .map_err(|error: crate::media::MediaProtocolError| {
2543 JsValue::from_str(&error.to_string())
2544 })?;
2545 self.broadcast_sessions
2546 .borrow()
2547 .get(&handle_id)
2548 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2549 .retire_publication(publication_id)
2550 .map_err(|error| JsValue::from_str(&error.to_string()))
2551 }
2552
2553 #[wasm_bindgen(js_name = __pauseOpenRtcBroadcastPublication)]
2554 pub fn pause_openrtc_broadcast_publication(
2555 &self,
2556 handle_id: String,
2557 publication_id: String,
2558 ) -> Result<(), JsValue> {
2559 let publication_id =
2560 publication_id
2561 .parse()
2562 .map_err(|error: crate::media::MediaProtocolError| {
2563 JsValue::from_str(&error.to_string())
2564 })?;
2565 self.broadcast_sessions
2566 .borrow()
2567 .get(&handle_id)
2568 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2569 .pause_publication(publication_id)
2570 .map_err(|error| JsValue::from_str(&error.to_string()))
2571 }
2572
2573 #[wasm_bindgen(js_name = __acceptOpenRtcBroadcastObject)]
2576 pub fn accept_openrtc_broadcast_object(
2577 &self,
2578 handle_id: String,
2579 source_slot: String,
2580 encoded: Vec<u8>,
2581 ) -> Result<JsValue, JsValue> {
2582 let accepted = self
2583 .broadcast_sessions
2584 .borrow()
2585 .get(&handle_id)
2586 .ok_or_else(|| JsValue::from_str("broadcast session is unavailable"))?
2587 .accept_media_for_source(&source_slot, &encoded)
2588 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2589 serde::Serialize::serialize(
2590 &accepted,
2591 &serde_wasm_bindgen::Serializer::json_compatible(),
2592 )
2593 .map_err(|error| JsValue::from_str(&error.to_string()))
2594 }
2595
2596 #[wasm_bindgen(js_name = __beginOpenRtcMediaPublication)]
2600 #[allow(clippy::too_many_arguments)]
2601 pub fn begin_openrtc_media_publication(
2602 &self,
2603 publication_id: Option<String>,
2604 kind: String,
2605 codec: String,
2606 clock_rate: u32,
2607 coded_width: Option<u32>,
2608 coded_height: Option<u32>,
2609 channels: Option<u16>,
2610 ) -> Result<JsValue, JsValue> {
2611 let publication_id = match publication_id {
2612 Some(value) if !value.is_empty() => {
2613 value
2614 .parse()
2615 .map_err(|error: crate::media::MediaProtocolError| {
2616 JsValue::from_str(&error.to_string())
2617 })?
2618 }
2619 _ => {
2620 let mut bytes = [0_u8; 16];
2621 getrandom::getrandom(&mut bytes)
2622 .map_err(|_| JsValue::from_str("media publication entropy failed"))?;
2623 crate::media::PublicationId::from_bytes(bytes)
2624 }
2625 };
2626 let kind: crate::media::MediaKind =
2627 kind.parse()
2628 .map_err(|error: crate::media::MediaProtocolError| {
2629 JsValue::from_str(&error.to_string())
2630 })?;
2631 if !matches!(
2632 kind,
2633 crate::media::MediaKind::Audio | crate::media::MediaKind::Video
2634 ) {
2635 return Err(JsValue::from_str(
2636 "browser media publications must be audio or video",
2637 ));
2638 }
2639 let codec: crate::media::MediaCodec =
2640 codec
2641 .parse()
2642 .map_err(|error: crate::media::MediaProtocolError| {
2643 JsValue::from_str(&error.to_string())
2644 })?;
2645 let publication = crate::media::MediaPublicationConfig {
2646 publication_id,
2647 media_generation: 0,
2648 kind,
2649 codec,
2650 clock_rate,
2651 coded_width,
2652 coded_height,
2653 channels,
2654 };
2655 let (publication, control) = self
2656 .portable_media
2657 .borrow_mut()
2658 .begin_publication(publication)
2659 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2660
2661 let object = js_sys::Object::new();
2662 js_sys::Reflect::set(
2663 &object,
2664 &JsValue::from_str("publicationId"),
2665 &JsValue::from_str(&publication_id.to_string()),
2666 )?;
2667 js_sys::Reflect::set(
2668 &object,
2669 &JsValue::from_str("mediaGeneration"),
2670 &JsValue::from_f64(publication.media_generation as f64),
2671 )?;
2672 js_sys::Reflect::set(
2673 &object,
2674 &JsValue::from_str("control"),
2675 &js_sys::Uint8Array::from(control.as_slice()),
2676 )?;
2677 Ok(object.into())
2678 }
2679
2680 #[wasm_bindgen(js_name = __encodeOpenRtcMediaSample)]
2683 pub fn encode_openrtc_media_sample(
2684 &self,
2685 publication_id: String,
2686 timestamp_us: f64,
2687 duration_us: u32,
2688 keyframe: bool,
2689 discardable: bool,
2690 payload: Vec<u8>,
2691 ) -> Result<Vec<u8>, JsValue> {
2692 fn safe_u64(value: f64, field: &str) -> Result<u64, JsValue> {
2693 if !value.is_finite()
2694 || value < 0.0
2695 || value.fract() != 0.0
2696 || value > crate::media::MAX_JAVASCRIPT_SAFE_INTEGER as f64
2697 {
2698 return Err(JsValue::from_str(&format!(
2699 "media {field} must be a non-negative JavaScript-safe integer"
2700 )));
2701 }
2702 Ok(value as u64)
2703 }
2704 let publication_id: crate::media::PublicationId =
2705 publication_id
2706 .parse()
2707 .map_err(|error: crate::media::MediaProtocolError| {
2708 JsValue::from_str(&error.to_string())
2709 })?;
2710 self.portable_media
2711 .borrow_mut()
2712 .encode_sample(
2713 publication_id,
2714 crate::media::EncodedMediaSample {
2715 timestamp_us: safe_u64(timestamp_us, "timestamp")?,
2716 duration_us,
2717 keyframe,
2718 discardable,
2719 payload,
2720 },
2721 )
2722 .map_err(|error| JsValue::from_str(&error.to_string()))
2723 }
2724
2725 #[wasm_bindgen(js_name = __pauseOpenRtcMediaPublication)]
2726 pub fn pause_openrtc_media_publication(
2727 &self,
2728 publication_id: String,
2729 ) -> Result<(), JsValue> {
2730 let publication_id: crate::media::PublicationId =
2731 publication_id
2732 .parse()
2733 .map_err(|error: crate::media::MediaProtocolError| {
2734 JsValue::from_str(&error.to_string())
2735 })?;
2736 self.portable_media
2737 .borrow_mut()
2738 .pause_publication(publication_id)
2739 .map_err(|error| JsValue::from_str(&error.to_string()))
2740 }
2741
2742 #[wasm_bindgen(js_name = __retireOpenRtcMediaPublication)]
2743 pub fn retire_openrtc_media_publication(
2744 &self,
2745 publication_id: String,
2746 ) -> Result<(), JsValue> {
2747 let publication_id: crate::media::PublicationId =
2748 publication_id
2749 .parse()
2750 .map_err(|error: crate::media::MediaProtocolError| {
2751 JsValue::from_str(&error.to_string())
2752 })?;
2753 self.portable_media
2754 .borrow_mut()
2755 .retire_publication(publication_id);
2756 Ok(())
2757 }
2758
2759 #[wasm_bindgen(js_name = __retireOpenRtcMediaReceiver)]
2760 pub fn retire_openrtc_media_receiver(
2761 &self,
2762 publication_id: String,
2763 media_generation: u32,
2764 ) -> Result<(), JsValue> {
2765 let publication_id: crate::media::PublicationId =
2766 publication_id
2767 .parse()
2768 .map_err(|error: crate::media::MediaProtocolError| {
2769 JsValue::from_str(&error.to_string())
2770 })?;
2771 self.portable_media
2772 .borrow_mut()
2773 .retire_receiver(publication_id, media_generation);
2774 Ok(())
2775 }
2776
2777 #[wasm_bindgen(js_name = __decodeOpenRtcMediaChunk)]
2781 pub fn decode_openrtc_media_chunk(&self, encoded: Vec<u8>) -> Result<JsValue, JsValue> {
2782 let chunk = self
2783 .portable_media
2784 .borrow_mut()
2785 .decode_chunk(&encoded)
2786 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2787 crate::media::ensure_javascript_safe_integer(chunk.sequence, "sequence")
2788 .and_then(|_| {
2789 crate::media::ensure_javascript_safe_integer(chunk.timestamp_us, "timestamp")
2790 })
2791 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2792
2793 let object = js_sys::Object::new();
2794 let set = |name: &str, value: JsValue| -> Result<(), JsValue> {
2795 js_sys::Reflect::set(&object, &JsValue::from_str(name), &value).map(|_| ())
2796 };
2797 set(
2798 "publicationId",
2799 JsValue::from_str(&chunk.publication_id.to_string()),
2800 )?;
2801 set(
2802 "mediaGeneration",
2803 JsValue::from_f64(chunk.media_generation as f64),
2804 )?;
2805 set("sequence", JsValue::from_f64(chunk.sequence as f64))?;
2806 set("timestampUs", JsValue::from_f64(chunk.timestamp_us as f64))?;
2807 set("durationUs", JsValue::from_f64(chunk.duration_us as f64))?;
2808 set(
2809 "kind",
2810 JsValue::from_str(match chunk.kind {
2811 crate::media::MediaKind::Audio => "audio",
2812 crate::media::MediaKind::Video => "video",
2813 crate::media::MediaKind::Screen => "screen",
2814 crate::media::MediaKind::Data => "data",
2815 }),
2816 )?;
2817 set(
2818 "codec",
2819 JsValue::from_str(match chunk.codec {
2820 crate::media::MediaCodec::Opus => "opus",
2821 crate::media::MediaCodec::H264 => "h264",
2822 crate::media::MediaCodec::Vp8 => "vp8",
2823 crate::media::MediaCodec::Vp9 => "vp9",
2824 crate::media::MediaCodec::Av1 => "av1",
2825 crate::media::MediaCodec::Pcm => "pcm",
2826 crate::media::MediaCodec::Opaque => "opaque",
2827 }),
2828 )?;
2829 set("keyframe", JsValue::from_bool(chunk.keyframe))?;
2830 set("discardable", JsValue::from_bool(chunk.discardable))?;
2831 set(
2832 "payload",
2833 js_sys::Uint8Array::from(chunk.payload.as_slice()).into(),
2834 )?;
2835 Ok(object.into())
2836 }
2837
2838 #[wasm_bindgen(js_name = __decodeOpenRtcMediaControl)]
2839 pub fn decode_openrtc_media_control(&self, encoded: Vec<u8>) -> Result<JsValue, JsValue> {
2840 let control = self
2841 .portable_media
2842 .borrow_mut()
2843 .decode_control(&encoded)
2844 .map_err(|error| JsValue::from_str(&error.to_string()))?;
2845 let portable = match control {
2850 crate::media::MediaControlFrame::Publish {
2851 publication_id,
2852 media_generation,
2853 kind,
2854 codec,
2855 clock_rate,
2856 coded_width,
2857 coded_height,
2858 channels,
2859 } => serde_json::json!({
2860 "type": "publish",
2861 "publicationId": publication_id.to_string(),
2862 "mediaGeneration": media_generation,
2863 "kind": kind,
2864 "codec": codec,
2865 "clockRate": clock_rate,
2866 "codedWidth": coded_width,
2867 "codedHeight": coded_height,
2868 "channels": channels,
2869 }),
2870 crate::media::MediaControlFrame::SetEnabled {
2871 publication_id,
2872 media_generation,
2873 enabled,
2874 } => serde_json::json!({
2875 "type": "set-enabled",
2876 "publicationId": publication_id.to_string(),
2877 "mediaGeneration": media_generation,
2878 "enabled": enabled,
2879 }),
2880 crate::media::MediaControlFrame::RequestKeyframe {
2881 publication_id,
2882 media_generation,
2883 } => serde_json::json!({
2884 "type": "request-keyframe",
2885 "publicationId": publication_id.to_string(),
2886 "mediaGeneration": media_generation,
2887 }),
2888 crate::media::MediaControlFrame::Stop {
2889 publication_id,
2890 media_generation,
2891 reason,
2892 } => serde_json::json!({
2893 "type": "stop",
2894 "publicationId": publication_id.to_string(),
2895 "mediaGeneration": media_generation,
2896 "reason": reason,
2897 }),
2898 };
2899 serde::Serialize::serialize(
2900 &portable,
2901 &serde_wasm_bindgen::Serializer::json_compatible(),
2902 )
2903 .map_err(|error| JsValue::from_str(&error.to_string()))
2904 }
2905
2906 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2910 #[wasm_bindgen(js_name = __setIrohCarrierActionHandler)]
2911 pub fn set_iroh_carrier_action_handler(&self, handler: Option<js_sys::Function>) {
2912 *self.wasm_carrier_action_handler.borrow_mut() = handler;
2913 }
2914
2915 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2919 #[wasm_bindgen(js_name = __getIrohCarrierReadiness)]
2920 pub async fn get_iroh_carrier_readiness(&self) -> Result<JsValue, JsValue> {
2921 let readiness = serde_json::json!({
2922 "webrtcConfigured": self.inner.is_webrtc_carrier_configured().await,
2923 "moqConfigured": self.inner.is_moq_carrier_configured().await,
2924 "webrtcAvailable": self.inner.is_webrtc_carrier_enabled().await,
2925 "moqAvailable": self.inner.is_moq_carrier_enabled().await,
2926 "actionHandlerInstalled": self.wasm_carrier_action_handler.borrow().is_some(),
2927 });
2928 serde::Serialize::serialize(
2929 &readiness,
2930 &serde_wasm_bindgen::Serializer::json_compatible(),
2931 )
2932 .map_err(|error| JsValue::from_str(&error.to_string()))
2933 }
2934
2935 #[cfg(feature = "transport-webrtc")]
2936 #[wasm_bindgen(js_name = __configureWebRtcCarrier)]
2937 pub async fn configure_iroh_webrtc_carrier(
2938 &self,
2939 enabled: bool,
2940 privacy_mode: bool,
2941 ) -> Result<(), JsValue> {
2942 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
2943 self.inner.set_webrtc_carrier(enabled, privacy_mode).await;
2944 Ok(())
2945 }
2946
2947 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2950 #[wasm_bindgen(js_name = __configureIrohRoutePolicy)]
2951 pub async fn configure_iroh_route_policy(
2952 &self,
2953 relay_only: bool,
2954 optimize_for: Option<String>,
2955 route_priority: JsValue,
2956 ) -> Result<(), JsValue> {
2957 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
2958 let optimize_for = match optimize_for.as_deref().unwrap_or("balanced") {
2959 "balanced" => crate::route_policy::RouteOptimization::Balanced,
2960 "lowest-latency" => crate::route_policy::RouteOptimization::LowestLatency,
2961 _ => {
2962 return Err(JsValue::from_str(
2963 "transport optimization must be balanced or lowest-latency",
2964 ))
2965 }
2966 };
2967 let route_priority = if route_priority.is_null() || route_priority.is_undefined() {
2968 Vec::new()
2969 } else {
2970 serde_wasm_bindgen::from_value(route_priority)
2971 .map_err(|error| JsValue::from_str(&error.to_string()))?
2972 };
2973 self.inner
2974 .configure_wasm_route_policy(relay_only, optimize_for, route_priority)
2975 .await;
2976 Ok(())
2977 }
2978
2979 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
2983 #[wasm_bindgen(js_name = __applyIrohCarrierPolicy)]
2984 pub async fn apply_iroh_carrier_policy(&self) -> Result<u32, JsValue> {
2985 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
2986 let capabilities: Vec<_> = self
2987 .wasm_remote_carrier_capabilities
2988 .borrow()
2989 .iter()
2990 .map(|(connection_id, capabilities)| (connection_id.clone(), *capabilities))
2991 .collect();
2992 let mut reconciled = 0u32;
2993 for (connection_id, capabilities) in capabilities {
2994 let Some(remote_endpoint_id) = self
2995 .inner
2996 .wasm_iroh_carrier_remote_endpoint_id(&connection_id)
2997 .await
2998 else {
2999 continue;
3000 };
3001 if self
3002 .begin_preferred_iroh_carrier_locked(
3003 connection_id,
3004 remote_endpoint_id,
3005 capabilities.webrtc,
3006 capabilities.moq,
3007 )
3008 .await?
3009 {
3010 reconciled = reconciled.saturating_add(1);
3011 }
3012 }
3013 Ok(reconciled)
3014 }
3015
3016 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3023 #[wasm_bindgen(js_name = __reconcileIrohCarriersAfterNetworkChange)]
3024 pub async fn reconcile_iroh_carriers_after_network_change(&self) -> u32 {
3025 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3026 let mut reconciled = 0u32;
3027 let mut recoveries = Vec::new();
3028 #[cfg(feature = "transport-webrtc")]
3029 {
3030 let attempts: Vec<_> = self
3031 .wasm_webrtc_carrier_attempts
3032 .borrow()
3033 .values()
3034 .cloned()
3035 .collect();
3036 let probed = futures::future::join_all(attempts.into_iter().map(|attempt| {
3037 let inner = self.inner.clone();
3038 async move {
3039 let needs_recovery = inner
3040 .wasm_carrier_generation_needs_network_recovery(
3041 &attempt.connection_id,
3042 attempt.remote_endpoint_id,
3043 crate::client::IrohPathKind::WebRtc,
3044 attempt.generation.transport_generation.saturating_add(1),
3045 )
3046 .await;
3047 (attempt, needs_recovery)
3048 }
3049 }))
3050 .await;
3051 for (attempt, needs_recovery) in probed {
3052 if !needs_recovery {
3053 continue;
3054 }
3055 if retire_selected_wasm_webrtc_carrier(
3056 self.inner.clone(),
3057 self.wasm_webrtc_carrier_attempts.clone(),
3058 self.wasm_webrtc_carrier_sessions.clone(),
3059 self.wasm_carrier_action_handler.clone(),
3060 &attempt,
3061 "network-change",
3062 crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
3063 )
3064 .await
3065 {
3066 reconciled = reconciled.saturating_add(1);
3067 recoveries.push((
3068 attempt.connection_id.clone(),
3069 attempt.remote_endpoint_id,
3070 attempt.generation.transport_generation.saturating_add(1),
3071 ));
3072 }
3073 }
3074 }
3075 #[cfg(feature = "transport-moq")]
3076 {
3077 let attempts: Vec<_> = self
3078 .wasm_moq_carrier_attempts
3079 .borrow()
3080 .values()
3081 .cloned()
3082 .collect();
3083 let probed = futures::future::join_all(attempts.into_iter().map(|attempt| {
3084 let inner = self.inner.clone();
3085 async move {
3086 let needs_recovery = inner
3087 .wasm_carrier_generation_needs_network_recovery(
3088 &attempt.connection_id,
3089 attempt.remote_endpoint_id,
3090 crate::client::IrohPathKind::Moq,
3091 attempt.generation.transport_generation.saturating_add(1),
3092 )
3093 .await;
3094 (attempt, needs_recovery)
3095 }
3096 }))
3097 .await;
3098 for (attempt, needs_recovery) in probed {
3099 if !needs_recovery {
3100 continue;
3101 }
3102 if retire_selected_wasm_moq_carrier(
3103 self.inner.clone(),
3104 self.wasm_moq_carrier_attempts.clone(),
3105 self.wasm_moq_carrier_sessions.clone(),
3106 self.wasm_carrier_action_handler.clone(),
3107 &attempt,
3108 "network-change",
3109 crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
3110 )
3111 .await
3112 {
3113 reconciled = reconciled.saturating_add(1);
3114 recoveries.push((
3115 attempt.connection_id.clone(),
3116 attempt.remote_endpoint_id,
3117 attempt.generation.transport_generation.saturating_add(1),
3118 ));
3119 }
3120 }
3121 }
3122 for (connection_id, remote_endpoint_id, transport_generation) in recoveries {
3123 let recovered = self
3124 .inner
3125 .recover_retired_wasm_carrier_generation(
3126 &connection_id,
3127 remote_endpoint_id,
3128 transport_generation,
3129 )
3130 .await;
3131 if !recovered {
3132 web_sys::console::warn_1(&JsValue::from_str(&format!(
3133 "[OpenRTC][Iroh carrier] base recovery remained pending connection_id={connection_id} transport_generation={transport_generation}"
3134 )));
3135 }
3136 }
3137 if reconciled > 0 {
3138 self.inner.wake_browser_auto_connect();
3145 }
3146 reconciled
3147 }
3148
3149 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3153 #[wasm_bindgen(js_name = __forgetIrohCarrierPeer)]
3154 pub async fn forget_iroh_carrier_peer(&self, connection_id: String) {
3155 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3156 self.forget_iroh_carrier_peer_locked(&connection_id);
3157 }
3158
3159 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3160 fn forget_iroh_carrier_peer_locked(&self, connection_id: &str) {
3161 self.wasm_remote_carrier_capabilities
3162 .borrow_mut()
3163 .remove(connection_id);
3164 self.inner
3165 .forget_iroh_carrier_peer_policy_epoch(connection_id);
3166 }
3167
3168 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3173 #[wasm_bindgen(js_name = __retireIrohCarrierPeer)]
3174 pub async fn retire_iroh_carrier_peer(
3175 &self,
3176 connection_id: String,
3177 terminal_reason: Option<String>,
3178 ) {
3179 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3180 #[cfg(feature = "transport-webrtc")]
3181 self.retire_iroh_webrtc_carrier_locked(
3182 connection_id.clone(),
3183 terminal_reason.as_deref(),
3184 )
3185 .await;
3186 #[cfg(feature = "transport-moq")]
3187 self.retire_iroh_moq_carrier_locked(connection_id.clone(), terminal_reason.as_deref())
3188 .await;
3189 self.forget_iroh_carrier_peer_locked(&connection_id);
3190 }
3191
3192 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3193 fn has_matching_in_flight_wasm_carrier_attempt(
3194 &self,
3195 connection_id: &str,
3196 capabilities: WasmRemoteCarrierCapabilities,
3197 ) -> bool {
3198 #[cfg(feature = "transport-webrtc")]
3199 if capabilities.webrtc
3200 && self
3201 .wasm_webrtc_carrier_attempts
3202 .borrow()
3203 .get(connection_id)
3204 .is_some_and(|attempt| {
3205 attempt.role == "responder" && !attempt.completion_started
3206 })
3207 {
3208 return true;
3209 }
3210 #[cfg(feature = "transport-moq")]
3211 if capabilities.moq
3212 && self
3213 .wasm_moq_carrier_attempts
3214 .borrow()
3215 .get(connection_id)
3216 .is_some_and(|attempt| attempt.role == "responder")
3217 {
3218 return true;
3219 }
3220 false
3221 }
3222
3223 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3224 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3227 #[wasm_bindgen(js_name = __beginPreferredIrohCarrier)]
3228 pub async fn begin_preferred_iroh_carrier(
3229 &self,
3230 connection_id: String,
3231 remote_endpoint_id: String,
3232 remote_supports_webrtc: bool,
3233 remote_supports_moq: bool,
3234 ) -> Result<bool, JsValue> {
3235 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3236 self.begin_preferred_iroh_carrier_locked(
3237 connection_id,
3238 remote_endpoint_id,
3239 remote_supports_webrtc,
3240 remote_supports_moq,
3241 )
3242 .await
3243 }
3244
3245 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3246 async fn begin_preferred_iroh_carrier_locked(
3247 &self,
3248 connection_id: String,
3249 remote_endpoint_id: String,
3250 remote_supports_webrtc: bool,
3251 remote_supports_moq: bool,
3252 ) -> Result<bool, JsValue> {
3253 let capabilities = WasmRemoteCarrierCapabilities {
3254 webrtc: remote_supports_webrtc,
3255 moq: remote_supports_moq,
3256 };
3257 let previous_capabilities = self
3258 .wasm_remote_carrier_capabilities
3259 .borrow()
3260 .get(&connection_id)
3261 .copied();
3262 if previous_capabilities != Some(capabilities) {
3263 let matching_first_attempt = previous_capabilities.is_none()
3268 && self
3269 .has_matching_in_flight_wasm_carrier_attempt(&connection_id, capabilities);
3270 if crate::client::should_bump_remote_carrier_peer_policy_epoch(
3271 previous_capabilities.is_some(),
3272 true,
3273 matching_first_attempt,
3274 ) {
3275 self.inner
3276 .bump_wasm_iroh_carrier_peer_policy_epoch(&connection_id)
3277 .await;
3278 }
3279 self.wasm_remote_carrier_capabilities
3280 .borrow_mut()
3281 .insert(connection_id.clone(), capabilities);
3282 }
3283 #[cfg(feature = "transport-webrtc")]
3284 {
3285 let refreshing = self
3286 .wasm_webrtc_carrier_attempts
3287 .borrow()
3288 .get(&connection_id)
3289 .is_some_and(|attempt| {
3290 let mut loan = attempt.ice_refresh_baseline.borrow_mut();
3291 if loan.as_ref().is_some_and(|loan| {
3292 loan.policy_epoch != self.inner.current_iroh_carrier_policy_epoch()
3293 || loan.peer_policy_epoch
3294 != self
3295 .inner
3296 .current_iroh_carrier_peer_policy_epoch(&connection_id)
3297 }) {
3298 loan.take();
3299 }
3300 loan.is_some()
3301 });
3302 if remote_supports_webrtc
3303 && refreshing
3304 && self.inner.is_webrtc_carrier_enabled().await
3305 {
3306 return Ok(false);
3307 }
3308 }
3309 if self
3310 .inner
3311 .reconcile_wasm_iroh_carrier_policy(
3312 &connection_id,
3313 remote_supports_webrtc,
3314 remote_supports_moq,
3315 )
3316 .await
3317 .map_err(|error| JsValue::from_str(&error.to_string()))?
3318 {
3319 return Ok(true);
3320 }
3321 let ranked = self
3322 .inner
3323 .ranked_wasm_iroh_carriers(remote_supports_webrtc, remote_supports_moq)
3324 .await;
3325 match ranked.first() {
3326 #[cfg(feature = "transport-webrtc")]
3327 Some(crate::route_policy::KnownRoute::WebRtc) => {
3328 self.begin_iroh_webrtc_carrier_attempt(connection_id, remote_endpoint_id, 0)
3329 .await
3330 }
3331 #[cfg(feature = "transport-moq")]
3332 Some(crate::route_policy::KnownRoute::Moq) => {
3333 self.begin_iroh_moq_carrier_attempt(connection_id, remote_endpoint_id, 0)
3334 .await
3335 }
3336 _ => Ok(false),
3337 }
3338 }
3339
3340 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3343 #[wasm_bindgen(js_name = __advancePreferredIrohCarrier)]
3344 pub async fn advance_preferred_iroh_carrier(
3345 &self,
3346 connection_id: String,
3347 remote_endpoint_id: String,
3348 failed_route: String,
3349 ) -> Result<bool, JsValue> {
3350 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3351 let Some(capabilities) = self
3352 .wasm_remote_carrier_capabilities
3353 .borrow()
3354 .get(&connection_id)
3355 .copied()
3356 else {
3357 return Ok(false);
3358 };
3359 let failed = crate::route_policy::normalize_route(&failed_route)
3360 .ok_or_else(|| JsValue::from_str("unknown failed carrier route"))?;
3361 let ranked = self
3362 .inner
3363 .ranked_wasm_iroh_carriers(capabilities.webrtc, capabilities.moq)
3364 .await;
3365 let Some(next_index) = ranked
3366 .iter()
3367 .position(|route| *route == failed)
3368 .map(|index| index + 1)
3369 .filter(|index| *index < ranked.len())
3370 else {
3371 return Ok(false);
3372 };
3373 let next = ranked[next_index];
3374 match next {
3375 #[cfg(feature = "transport-webrtc")]
3376 crate::route_policy::KnownRoute::WebRtc => {
3377 self.begin_iroh_webrtc_carrier_attempt(connection_id, remote_endpoint_id, 0)
3378 .await
3379 }
3380 #[cfg(feature = "transport-moq")]
3381 crate::route_policy::KnownRoute::Moq => {
3382 self.begin_iroh_moq_carrier_attempt(connection_id, remote_endpoint_id, 0)
3383 .await
3384 }
3385 _ => Ok(false),
3386 }
3387 }
3388
3389 #[cfg(feature = "transport-moq")]
3390 #[wasm_bindgen(js_name = __configureMoqCarrier)]
3391 pub async fn configure_iroh_moq_carrier(
3392 &self,
3393 enabled: bool,
3394 relay_url: Option<String>,
3395 ) -> Result<(), JsValue> {
3396 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3397 let relay_url = relay_url.map(|value| value.trim().to_string());
3398 if enabled && relay_url.as_deref().unwrap_or_default().is_empty() {
3399 return Err(JsValue::from_str(
3400 "MoQ Iroh carrier requires an explicit relay URL",
3401 ));
3402 }
3403 self.inner.set_moq_carrier(enabled, relay_url).await;
3404 Ok(())
3405 }
3406
3407 #[cfg(feature = "transport-moq")]
3410 #[wasm_bindgen(js_name = __beginMoqCarrier)]
3411 pub async fn begin_iroh_moq_carrier(
3412 &self,
3413 connection_id: String,
3414 remote_endpoint_id: String,
3415 ) -> Result<bool, JsValue> {
3416 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3417 self.begin_iroh_moq_carrier_attempt(connection_id, remote_endpoint_id, 0)
3418 .await
3419 }
3420
3421 #[cfg(feature = "transport-moq")]
3422 async fn begin_iroh_moq_carrier_attempt(
3423 &self,
3424 connection_id: String,
3425 remote_endpoint_id: String,
3426 retry_count: u8,
3427 ) -> Result<bool, JsValue> {
3428 if !self.inner.is_moq_carrier_enabled().await {
3429 return Ok(false);
3430 }
3431 let local_endpoint_id = self
3432 .inner
3433 .current_node_id()
3434 .await
3435 .ok_or_else(|| JsValue::from_str("local Iroh endpoint id is unavailable"))?;
3436 if local_endpoint_id.as_str() <= remote_endpoint_id.as_str()
3437 || !crate::iroh_connection_policy::custom_carrier_base_allows(
3438 self.inner.iroh_path_kind(&remote_endpoint_id).await,
3439 crate::client::IrohPathKind::Moq,
3440 false,
3441 )
3442 {
3443 return Ok(false);
3444 }
3445 let endpoint_id = remote_endpoint_id
3446 .parse::<iroh::EndpointId>()
3447 .map_err(|error| JsValue::from_str(&error.to_string()))?;
3448 self.inner
3449 .get_connection(endpoint_id)
3450 .await
3451 .ok_or_else(|| JsValue::from_str("MoQ carrier base is unavailable"))?;
3452 let generation = self
3453 .inner
3454 .current_wasm_peer_data_generation(&connection_id, None)
3455 .await
3456 .ok_or_else(|| JsValue::from_str("MoQ carrier generation is unavailable"))?;
3457 let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
3458 crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
3459 crate::iroh_carrier_bootstrap::CarrierGenerationFence {
3460 transport_stable_id: generation.transport_stable_id,
3461 transport_generation: generation.transport_generation,
3462 route_generation: generation.route_generation,
3463 },
3464 retry_count.saturating_add(1),
3465 )
3466 .map_err(|error| JsValue::from_str(&error.to_string()))?;
3467 let kind = crate::client::IrohPathKind::Moq;
3468 if !matches!(
3469 self.inner
3470 .reserve_wasm_carrier_upgrade(
3471 &connection_id,
3472 kind,
3473 &bootstrap.upgrade_id,
3474 generation,
3475 )
3476 .await,
3477 crate::client::WasmCarrierUpgradeReservation::Reserved
3478 ) {
3479 let pending = self
3480 .wasm_moq_carrier_attempts
3481 .borrow_mut()
3482 .get_mut(&connection_id)
3483 .filter(|attempt| {
3484 attempt.role == "initiator"
3485 && attempt.generation == generation
3486 && attempt.prepared
3487 && !attempt.retry_sent
3488 })
3489 .map(|attempt| {
3490 attempt.retry_sent = true;
3491 attempt.clone()
3492 });
3493 if let Some(pending) = pending {
3494 emit_wasm_carrier_action(
3500 &self.wasm_carrier_action_handler,
3501 serde_json::json!({
3502 "type": "send-control",
3503 "connectionId": connection_id,
3504 "remoteEndpointId": pending.remote_endpoint_id.to_string(),
3505 "envelope": pending.bootstrap,
3506 }),
3507 );
3508 return Ok(true);
3509 }
3510 return Ok(false);
3511 }
3512 let previous = self.wasm_moq_carrier_attempts.borrow_mut().insert(
3513 connection_id.clone(),
3514 WasmMoqCarrierAttempt {
3515 connection_id: connection_id.clone(),
3516 remote_endpoint_id: endpoint_id,
3517 bootstrap: bootstrap.clone(),
3518 generation,
3519 role: "initiator",
3520 prepared: false,
3521 retry_sent: false,
3522 retry_count,
3523 inbound_authorization_expires_at_ms: None,
3524 },
3525 );
3526 if let Some(previous) = previous {
3527 self.wasm_moq_carrier_sessions
3528 .borrow_mut()
3529 .remove(&previous.bootstrap.upgrade_id);
3530 }
3531 let (publish_namespace, subscribe_namespace, track_name) = wasm_moq_carrier_namespaces(
3532 &local_endpoint_id,
3533 &remote_endpoint_id,
3534 &bootstrap.carrier_session_id,
3535 );
3536 emit_wasm_carrier_action(
3537 &self.wasm_carrier_action_handler,
3538 serde_json::json!({
3539 "type": "prepare-moq",
3540 "connectionId": connection_id,
3541 "remoteEndpointId": remote_endpoint_id,
3542 "role": "initiator",
3543 "upgradeId": bootstrap.upgrade_id,
3544 "carrierSessionId": bootstrap.carrier_session_id,
3545 "transportGeneration": generation.transport_generation.saturating_add(1),
3546 "publishNamespace": publish_namespace,
3547 "subscribeNamespace": subscribe_namespace,
3548 "trackName": track_name,
3549 }),
3550 );
3551 self.schedule_wasm_moq_carrier_watchdog(bootstrap);
3552 Ok(true)
3553 }
3554
3555 #[cfg(feature = "transport-moq")]
3556 #[wasm_bindgen(js_name = __moqCarrierPrepared)]
3557 pub async fn iroh_moq_carrier_prepared(
3558 &self,
3559 connection_id: String,
3560 upgrade_id: String,
3561 ) -> Result<(), JsValue> {
3562 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3563 let attempt = {
3564 let mut attempts = self.wasm_moq_carrier_attempts.borrow_mut();
3565 let attempt = attempts
3566 .get_mut(&connection_id)
3567 .filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
3568 .ok_or_else(|| JsValue::from_str("MoQ carrier attempt is stale"))?;
3569 attempt.prepared = true;
3570 attempt.clone()
3571 };
3572 if !self
3573 .inner
3574 .wasm_carrier_upgrade_is_current(
3575 &connection_id,
3576 crate::client::IrohPathKind::Moq,
3577 &upgrade_id,
3578 attempt.generation,
3579 )
3580 .await
3581 {
3582 return Err(JsValue::from_str("MoQ carrier attempt was retired"));
3583 }
3584 if attempt.role == "initiator" {
3585 emit_wasm_carrier_action(
3586 &self.wasm_carrier_action_handler,
3587 serde_json::json!({
3588 "type": "send-control",
3589 "connectionId": connection_id,
3590 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
3591 "envelope": attempt.bootstrap,
3592 }),
3593 );
3594 } else {
3595 emit_wasm_carrier_action(
3601 &self.wasm_carrier_action_handler,
3602 serde_json::json!({
3603 "type": "activate-moq",
3604 "connectionId": attempt.connection_id,
3605 "upgradeId": upgrade_id,
3606 }),
3607 );
3608 }
3609 Ok(())
3610 }
3611
3612 #[cfg(feature = "transport-moq")]
3613 #[wasm_bindgen(js_name = __moqCarrierFailed)]
3614 pub async fn iroh_moq_carrier_failed(
3615 &self,
3616 connection_id: String,
3617 upgrade_id: String,
3618 failure_code: String,
3619 ) {
3620 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3621 let attempt = self
3622 .wasm_moq_carrier_attempts
3623 .borrow()
3624 .get(&connection_id)
3625 .filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
3626 .cloned();
3627 if let Some(attempt) = attempt {
3628 let failure_code = match failure_code.as_str() {
3629 "relay-failed" => "relay-failed",
3630 "draft14-datagram-unavailable" => "draft14-datagram-unavailable",
3631 "carrier-backpressure" => "carrier-backpressure",
3632 _ => "browser-adapter-failed",
3633 };
3634 if retire_selected_wasm_moq_carrier(
3635 self.inner.clone(),
3636 self.wasm_moq_carrier_attempts.clone(),
3637 self.wasm_moq_carrier_sessions.clone(),
3638 self.wasm_carrier_action_handler.clone(),
3639 &attempt,
3640 failure_code,
3641 crate::lifecycle_reason::REASON_IROH_CARRIER_FAILED,
3642 )
3643 .await
3644 {
3645 return;
3646 }
3647 self.fail_wasm_moq_carrier_attempt(attempt, failure_code, true)
3648 .await;
3649 }
3650 }
3651
3652 #[cfg(feature = "transport-moq")]
3653 #[wasm_bindgen(js_name = __retireMoqCarrier)]
3654 pub async fn retire_iroh_moq_carrier(
3655 &self,
3656 connection_id: String,
3657 terminal_reason: Option<String>,
3658 ) {
3659 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3660 self.retire_iroh_moq_carrier_locked(connection_id, terminal_reason.as_deref())
3661 .await;
3662 }
3663
3664 #[cfg(feature = "transport-moq")]
3665 async fn retire_iroh_moq_carrier_locked(
3666 &self,
3667 connection_id: String,
3668 terminal_reason: Option<&str>,
3669 ) {
3670 let attempt = self
3671 .wasm_moq_carrier_attempts
3672 .borrow_mut()
3673 .remove(&connection_id);
3674 if let Some(attempt) = attempt {
3675 self.inner
3676 .retire_wasm_carrier_upgrade(
3677 &connection_id,
3678 crate::client::IrohPathKind::Moq,
3679 &attempt.bootstrap.upgrade_id,
3680 attempt.generation,
3681 )
3682 .await;
3683 let carrier = self
3684 .wasm_moq_carrier_sessions
3685 .borrow_mut()
3686 .remove(&attempt.bootstrap.upgrade_id);
3687 if let (Some(reason), Some(session)) = (terminal_reason, carrier.as_ref()) {
3688 let _ = session.send_terminal(reason).await;
3689 }
3690 emit_wasm_carrier_action(
3691 &self.wasm_carrier_action_handler,
3692 serde_json::json!({
3693 "type": "retire-moq",
3694 "connectionId": connection_id,
3695 "upgradeId": attempt.bootstrap.upgrade_id,
3696 "failureCode": "logical-connection-retired",
3697 }),
3698 );
3699 }
3700 }
3701
3702 #[cfg(feature = "transport-webrtc")]
3705 #[wasm_bindgen(js_name = __beginWebRtcCarrier)]
3706 pub async fn begin_iroh_webrtc_carrier(
3707 &self,
3708 connection_id: String,
3709 remote_endpoint_id: String,
3710 ) -> Result<bool, JsValue> {
3711 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3712 self.begin_iroh_webrtc_carrier_attempt(connection_id, remote_endpoint_id, 0)
3713 .await
3714 }
3715
3716 #[cfg(feature = "transport-webrtc")]
3717 async fn begin_iroh_webrtc_carrier_attempt(
3718 &self,
3719 connection_id: String,
3720 remote_endpoint_id: String,
3721 retry_count: u8,
3722 ) -> Result<bool, JsValue> {
3723 self.webrtc_carrier_owner()
3724 .begin_attempt(connection_id, remote_endpoint_id, retry_count)
3725 .await
3726 }
3727
3728 #[cfg(feature = "transport-webrtc")]
3731 #[wasm_bindgen(js_name = __webRtcCarrierPrepared)]
3732 pub async fn iroh_webrtc_carrier_prepared(
3733 &self,
3734 connection_id: String,
3735 upgrade_id: String,
3736 ) -> Result<(), JsValue> {
3737 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3738 let attempt = {
3739 let mut attempts = self.wasm_webrtc_carrier_attempts.borrow_mut();
3740 let attempt = attempts
3741 .get_mut(&connection_id)
3742 .filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
3743 .ok_or_else(|| JsValue::from_str("WebRTC carrier attempt is stale"))?;
3744 attempt.prepared = true;
3745 attempt.clone()
3746 };
3747 let kind = crate::client::IrohPathKind::WebRtc;
3748 if !self
3749 .inner
3750 .wasm_carrier_upgrade_is_current(
3751 &connection_id,
3752 kind,
3753 &upgrade_id,
3754 attempt.generation,
3755 )
3756 .await
3757 {
3758 return Err(JsValue::from_str("WebRTC carrier attempt was retired"));
3759 }
3760 if attempt.role == "initiator" {
3761 emit_wasm_carrier_action(
3762 &self.wasm_carrier_action_handler,
3763 serde_json::json!({
3764 "type": "send-control",
3765 "connectionId": connection_id,
3766 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
3767 "envelope": attempt.bootstrap,
3768 }),
3769 );
3770 } else {
3771 emit_wasm_carrier_action(
3776 &self.wasm_carrier_action_handler,
3777 serde_json::json!({
3778 "type": "send-control",
3779 "connectionId": attempt.connection_id,
3780 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
3781 "envelope": {
3782 "type": "#pluto-signal",
3783 "content": {
3784 "transport": "webrtc",
3785 "type": "renegotiate",
3786 "negotiationId": upgrade_id,
3787 }
3788 },
3789 }),
3790 );
3791 }
3792 Ok(())
3793 }
3794
3795 #[cfg(feature = "transport-webrtc")]
3796 #[wasm_bindgen(js_name = __webRtcCarrierFailed)]
3797 pub async fn iroh_webrtc_carrier_failed(
3798 &self,
3799 connection_id: String,
3800 upgrade_id: String,
3801 failure_code: String,
3802 ) {
3803 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3804 let attempt = self
3805 .wasm_webrtc_carrier_attempts
3806 .borrow()
3807 .get(&connection_id)
3808 .filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
3809 .cloned();
3810 if let Some(attempt) = attempt {
3811 let failure_code = match failure_code.as_str() {
3812 "data-channel-failed" => "data-channel-failed",
3813 "ice-failed" => "ice-failed",
3814 "signaling-failed" => "signaling-failed",
3815 "carrier-backpressure" => "carrier-backpressure",
3816 "carrier-base-generation-stale" => "carrier-base-generation-stale",
3817 _ => "browser-adapter-failed",
3818 };
3819 if retire_selected_wasm_webrtc_carrier(
3820 self.inner.clone(),
3821 self.wasm_webrtc_carrier_attempts.clone(),
3822 self.wasm_webrtc_carrier_sessions.clone(),
3823 self.wasm_carrier_action_handler.clone(),
3824 &attempt,
3825 failure_code,
3826 crate::lifecycle_reason::REASON_IROH_CARRIER_FAILED,
3827 )
3828 .await
3829 {
3830 return;
3831 }
3832 self.fail_wasm_webrtc_carrier_attempt(attempt, failure_code, true)
3833 .await;
3834 }
3835 }
3836
3837 #[cfg(feature = "transport-webrtc")]
3838 #[wasm_bindgen(js_name = __retireWebRtcCarrier)]
3839 pub async fn retire_iroh_webrtc_carrier(
3840 &self,
3841 connection_id: String,
3842 terminal_reason: Option<String>,
3843 ) {
3844 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3845 self.retire_iroh_webrtc_carrier_locked(connection_id, terminal_reason.as_deref())
3846 .await;
3847 }
3848
3849 #[cfg(feature = "transport-webrtc")]
3850 async fn retire_iroh_webrtc_carrier_locked(
3851 &self,
3852 connection_id: String,
3853 terminal_reason: Option<&str>,
3854 ) {
3855 let attempt = self
3856 .wasm_webrtc_carrier_attempts
3857 .borrow_mut()
3858 .remove(&connection_id);
3859 if let Some(attempt) = attempt {
3860 self.inner
3861 .retire_wasm_carrier_upgrade(
3862 &connection_id,
3863 crate::client::IrohPathKind::WebRtc,
3864 &attempt.bootstrap.upgrade_id,
3865 attempt.generation,
3866 )
3867 .await;
3868 let carrier = self
3869 .wasm_webrtc_carrier_sessions
3870 .borrow_mut()
3871 .remove(&attempt.bootstrap.upgrade_id);
3872 if let (Some(reason), Some(session)) = (terminal_reason, carrier.as_ref()) {
3873 let _ = session.send_terminal(reason).await;
3874 }
3875 emit_wasm_carrier_action(
3876 &self.wasm_carrier_action_handler,
3877 serde_json::json!({
3878 "type": "retire-webrtc",
3879 "connectionId": connection_id,
3880 "upgradeId": attempt.bootstrap.upgrade_id,
3881 "failureCode": "logical-connection-retired",
3882 }),
3883 );
3884 }
3885 }
3886
3887 #[cfg(feature = "transport-webrtc")]
3891 #[wasm_bindgen(js_name = __observeWebRtcAvailability)]
3892 pub async fn observe_webrtc_availability(&self, observation: String) {
3893 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
3894 if !self.webrtc_availability.borrow_mut().observe(&observation) {
3895 return;
3896 }
3897 let attempts: Vec<_> = self
3898 .wasm_webrtc_carrier_attempts
3899 .borrow()
3900 .values()
3901 .cloned()
3902 .collect();
3903 for attempt in attempts {
3904 self.refresh_wasm_webrtc_ice_locked(&attempt, false).await;
3905 }
3906 }
3907
3908 #[cfg(feature = "transport-webrtc")]
3909 async fn refresh_wasm_webrtc_ice_locked(
3910 &self,
3911 attempt: &WasmWebRtcCarrierAttempt,
3912 peer_requested: bool,
3913 ) {
3914 if (peer_requested && attempt.role != "initiator")
3915 || !self
3916 .inner
3917 .wasm_webrtc_ice_refresh_is_current(
3918 &attempt.connection_id,
3919 attempt.remote_endpoint_id,
3920 attempt.generation.transport_generation.saturating_add(1),
3921 true,
3922 )
3923 .await
3924 {
3925 return;
3926 }
3927 let reserved = self
3928 .wasm_webrtc_carrier_attempts
3929 .borrow_mut()
3930 .get_mut(&attempt.connection_id)
3931 .filter(|current| current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id)
3932 .is_some_and(|current| current.ice_refresh.reserve(web_time::Instant::now()));
3933 if !reserved {
3934 return;
3935 }
3936 if attempt.role == "initiator" {
3937 let Some(refresh_id) = crate::packet_carrier_transport::new_ice_refresh_id() else {
3938 return;
3939 };
3940 if !self
3941 .prepare_wasm_ice_refresh_baseline(attempt, &refresh_id)
3942 .await
3943 {
3944 return;
3945 }
3946 emit_wasm_carrier_action(
3947 &self.wasm_carrier_action_handler,
3948 serde_json::json!({
3949 "type": "send-control",
3950 "connectionId": attempt.connection_id,
3951 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
3952 "envelope": {
3953 "type": "#pluto-signal", "content": {
3954 "transport": "webrtc", "type": "refresh-prepare", "negotiationId": attempt.bootstrap.upgrade_id, "refreshId": refresh_id,
3955 }
3956 },
3957 }),
3958 );
3959 } else {
3960 emit_wasm_carrier_action(
3961 &self.wasm_carrier_action_handler,
3962 serde_json::json!({
3963 "type": "send-control",
3964 "connectionId": attempt.connection_id,
3965 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
3966 "envelope": {
3967 "type": "#pluto-signal",
3968 "content": {
3969 "transport": "webrtc", "type": "refresh-ice",
3970 "negotiationId": attempt.bootstrap.upgrade_id,
3971 }
3972 },
3973 }),
3974 );
3975 }
3976 }
3977
3978 #[cfg(feature = "transport-webrtc")]
3979 async fn prepare_wasm_ice_refresh_baseline(
3980 &self,
3981 attempt: &WasmWebRtcCarrierAttempt,
3982 refresh_id: &str,
3983 ) -> bool {
3984 if attempt.ice_refresh_baseline.borrow().is_some() {
3985 return false;
3986 }
3987 let policy_epoch = self.inner.current_iroh_carrier_policy_epoch();
3988 let peer_policy_epoch = self
3989 .inner
3990 .current_iroh_carrier_peer_policy_epoch(&attempt.connection_id);
3991 let node = self.inner.iroh_node.read().await.as_ref().cloned();
3992 let Some(node) = node else { return false };
3993 let Some(baseline) = node
3994 .baseline_for_ice_refresh(attempt.remote_endpoint_id)
3995 .await
3996 else {
3997 return false;
3998 };
3999 if policy_epoch != self.inner.current_iroh_carrier_policy_epoch()
4000 || peer_policy_epoch
4001 != self
4002 .inner
4003 .current_iroh_carrier_peer_policy_epoch(&attempt.connection_id)
4004 || !self
4005 .inner
4006 .wasm_webrtc_ice_refresh_is_current(
4007 &attempt.connection_id,
4008 attempt.remote_endpoint_id,
4009 attempt.generation.transport_generation.saturating_add(1),
4010 false,
4011 )
4012 .await
4013 {
4014 return false;
4015 }
4016 let started = web_time::Instant::now();
4019 *attempt.ice_refresh_baseline.borrow_mut() =
4020 Some(crate::packet_carrier_transport::WebRtcIceRefreshPathLoan {
4021 _preference: baseline,
4022 since: started,
4023 refresh_id: refresh_id.to_owned(),
4024 policy_epoch,
4025 peer_policy_epoch,
4026 started: false,
4027 });
4028 let slot = Rc::downgrade(&attempt.ice_refresh_baseline);
4029 wasm_bindgen_futures::spawn_local(async move {
4030 gloo_timers::future::sleep(std::time::Duration::from_secs(30)).await;
4031 if let Some(slot) = slot.upgrade() {
4032 let mut slot = slot.borrow_mut();
4033 if slot.as_ref().is_some_and(|loan| loan.since == started) {
4034 slot.take();
4035 }
4036 }
4037 });
4038 true
4039 }
4040
4041 #[cfg(feature = "transport-webrtc")]
4042 #[wasm_bindgen(js_name = __webRtcCarrierIceRefreshed)]
4043 pub async fn webrtc_carrier_ice_refreshed(
4044 &self,
4045 connection_id: String,
4046 upgrade_id: String,
4047 refresh_id: String,
4048 ) {
4049 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
4050 if let Some(attempt) = self
4051 .wasm_webrtc_carrier_attempts
4052 .borrow()
4053 .get(&connection_id)
4054 .filter(|attempt| attempt.bootstrap.upgrade_id == upgrade_id)
4055 {
4056 let mut loan = attempt.ice_refresh_baseline.borrow_mut();
4057 if loan
4058 .as_ref()
4059 .is_some_and(|loan| loan.refresh_id == refresh_id)
4060 {
4061 loan.take();
4062 }
4063 }
4064 }
4065
4066 #[cfg(feature = "transport-webrtc")]
4069 #[wasm_bindgen(js_name = __handleIrohCarrierControl)]
4070 pub async fn handle_iroh_carrier_control(
4071 &self,
4072 connection_id: String,
4073 remote_endpoint_id: String,
4074 frame: JsValue,
4075 ) -> Result<bool, JsValue> {
4076 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
4077 let frame: serde_json::Value = serde_wasm_bindgen::from_value(frame)
4078 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4079 if frame.get("type").and_then(serde_json::Value::as_str) == Some("#pluto-signal")
4080 && frame
4081 .get("content")
4082 .and_then(|content| content.get("transport"))
4083 .and_then(serde_json::Value::as_str)
4084 == Some("webrtc")
4085 {
4086 let negotiation_id = frame
4087 .get("content")
4088 .and_then(|content| content.get("negotiationId"))
4089 .and_then(serde_json::Value::as_str)
4090 .unwrap_or_default();
4091 let current = self
4092 .wasm_webrtc_carrier_attempts
4093 .borrow()
4094 .get(&connection_id)
4095 .is_some_and(|attempt| {
4096 attempt.bootstrap.upgrade_id == negotiation_id
4097 && attempt.remote_endpoint_id.to_string() == remote_endpoint_id
4098 });
4099 if current {
4100 let signal_type = frame
4101 .get("content")
4102 .and_then(|content| content.get("type"))
4103 .and_then(serde_json::Value::as_str);
4104 if matches!(signal_type, Some("refresh-prepare" | "refresh-ready")) {
4105 let Some(refresh_id) = frame
4106 .get("content")
4107 .and_then(|content| content.get("refreshId"))
4108 .and_then(serde_json::Value::as_str)
4109 .filter(|id| {
4110 id.len() == 32 && id.bytes().all(|b| b.is_ascii_hexdigit())
4111 })
4112 else {
4113 return Ok(true);
4114 };
4115 let attempt = self
4116 .wasm_webrtc_carrier_attempts
4117 .borrow()
4118 .get(&connection_id)
4119 .cloned();
4120 if let Some(attempt) = attempt {
4121 if signal_type == Some("refresh-prepare") && attempt.role == "responder"
4122 {
4123 if attempt.ice_refresh_baseline.borrow().is_none()
4124 && (!self
4125 .inner
4126 .wasm_webrtc_ice_refresh_is_current(
4127 &connection_id,
4128 attempt.remote_endpoint_id,
4129 attempt
4130 .generation
4131 .transport_generation
4132 .saturating_add(1),
4133 true,
4134 )
4135 .await
4136 || !self
4137 .prepare_wasm_ice_refresh_baseline(&attempt, refresh_id)
4138 .await)
4139 {
4140 return Ok(true);
4141 }
4142 if !attempt
4143 .ice_refresh_baseline
4144 .borrow()
4145 .as_ref()
4146 .is_some_and(|loan| loan.refresh_id == refresh_id)
4147 {
4148 return Ok(true);
4149 }
4150 emit_wasm_carrier_action(
4151 &self.wasm_carrier_action_handler,
4152 serde_json::json!({
4153 "type": "send-control", "connectionId": connection_id, "remoteEndpointId": remote_endpoint_id,
4154 "envelope": { "type": "#pluto-signal", "content": {
4155 "transport": "webrtc", "type": "refresh-ready", "negotiationId": negotiation_id, "refreshId": refresh_id,
4156 } },
4157 }),
4158 );
4159 } else if signal_type == Some("refresh-ready")
4160 && attempt.role == "initiator"
4161 {
4162 if !self
4163 .inner
4164 .wasm_webrtc_ice_refresh_is_current(
4165 &connection_id,
4166 attempt.remote_endpoint_id,
4167 attempt.generation.transport_generation.saturating_add(1),
4168 false,
4169 )
4170 .await
4171 {
4172 return Ok(true);
4173 }
4174 let start = attempt
4175 .ice_refresh_baseline
4176 .borrow_mut()
4177 .as_mut()
4178 .filter(|loan| {
4179 loan.refresh_id == refresh_id
4180 && loan.policy_epoch
4181 == self.inner.current_iroh_carrier_policy_epoch()
4182 && loan.peer_policy_epoch
4183 == self
4184 .inner
4185 .current_iroh_carrier_peer_policy_epoch(
4186 &connection_id,
4187 )
4188 && !loan.started
4189 && loan.since.elapsed().as_secs() < 30
4190 })
4191 .map(|loan| {
4192 loan.started = true;
4193 })
4194 .is_some();
4195 if start {
4196 emit_wasm_carrier_action(
4197 &self.wasm_carrier_action_handler,
4198 serde_json::json!({
4199 "type": "restart-webrtc", "connectionId": connection_id, "upgradeId": negotiation_id, "refreshId": refresh_id,
4200 }),
4201 );
4202 }
4203 }
4204 }
4205 return Ok(true);
4206 }
4207 if signal_type == Some("refresh-ice") {
4208 let attempt = self
4209 .wasm_webrtc_carrier_attempts
4210 .borrow()
4211 .get(&connection_id)
4212 .cloned();
4213 if let Some(attempt) = attempt {
4214 self.refresh_wasm_webrtc_ice_locked(&attempt, true).await;
4215 }
4216 return Ok(true);
4217 }
4218 let start_offer = if signal_type == Some("renegotiate") {
4219 self.wasm_webrtc_carrier_attempts
4220 .borrow_mut()
4221 .get_mut(&connection_id)
4222 .filter(|attempt| {
4223 attempt.bootstrap.upgrade_id == negotiation_id
4224 && attempt.role == "initiator"
4225 && !attempt.offer_started
4226 })
4227 .map(|attempt| {
4228 attempt.offer_started = true;
4229 })
4230 .is_some()
4231 } else {
4232 false
4233 };
4234 if start_offer {
4235 emit_wasm_carrier_action(
4236 &self.wasm_carrier_action_handler,
4237 serde_json::json!({
4238 "type": "start-webrtc",
4239 "connectionId": connection_id,
4240 "upgradeId": negotiation_id,
4241 }),
4242 );
4243 return Ok(true);
4244 }
4245 let restart_offer = signal_type == Some("sdp")
4246 && frame
4247 .get("content")
4248 .and_then(|c| c.get("sdp"))
4249 .and_then(|s| s.get("type"))
4250 .and_then(serde_json::Value::as_str)
4251 == Some("offer");
4252 let attempt = self
4253 .wasm_webrtc_carrier_attempts
4254 .borrow()
4255 .get(&connection_id)
4256 .cloned();
4257 let refresh_id = if restart_offer {
4258 attempt.as_ref().and_then(|attempt| {
4259 let mut loan = attempt.ice_refresh_baseline.borrow_mut();
4260 loan.as_mut().map(|loan| {
4261 loan.started = true;
4262 loan.refresh_id.clone()
4263 })
4264 })
4265 } else {
4266 None
4267 };
4268 emit_wasm_carrier_action(
4269 &self.wasm_carrier_action_handler,
4270 serde_json::json!({
4271 "type": "apply-webrtc-signal",
4272 "connectionId": connection_id,
4273 "upgradeId": negotiation_id,
4274 "signal": frame,
4275 "refreshId": refresh_id,
4276 }),
4277 );
4278 }
4279 return Ok(true);
4280 }
4281
4282 let Some(bootstrap) =
4283 crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(&frame)
4284 else {
4285 return Ok(false);
4286 };
4287 #[cfg(feature = "transport-moq")]
4288 if bootstrap.carrier == crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14
4289 {
4290 return self
4291 .handle_wasm_moq_bootstrap(connection_id, remote_endpoint_id, bootstrap)
4292 .await;
4293 }
4294 if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc {
4295 return Ok(false);
4296 }
4297 let endpoint_id = remote_endpoint_id
4298 .parse::<iroh::EndpointId>()
4299 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4300 let kind = crate::client::IrohPathKind::WebRtc;
4301 match bootstrap.action {
4302 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
4303 let local_endpoint_id = self
4304 .inner
4305 .current_node_id()
4306 .await
4307 .ok_or_else(|| JsValue::from_str("local endpoint id is unavailable"))?;
4308 if local_endpoint_id.as_str() >= remote_endpoint_id.as_str()
4309 || !self.inner.is_webrtc_carrier_enabled().await
4310 {
4311 return Ok(true);
4312 }
4313 if !crate::iroh_connection_policy::custom_carrier_base_allows(
4314 self.inner.iroh_path_kind(&remote_endpoint_id).await,
4315 crate::client::IrohPathKind::WebRtc,
4316 false,
4317 ) {
4318 return Ok(true);
4319 }
4320 let base_connection = self
4321 .inner
4322 .get_connection(endpoint_id)
4323 .await
4324 .ok_or_else(|| JsValue::from_str("WebRTC carrier base is unavailable"))?;
4325 let generation = self
4326 .inner
4327 .current_wasm_peer_data_generation(&connection_id, None)
4328 .await
4329 .ok_or_else(|| {
4330 JsValue::from_str("WebRTC carrier generation is unavailable")
4331 })?;
4332 if crate::transport_generation::for_connection(&base_connection)
4333 != generation.transport_stable_id
4334 {
4335 let failed =
4340 crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
4341 &bootstrap,
4342 "carrier-base-generation-stale",
4343 )
4344 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4345 emit_wasm_carrier_action(
4346 &self.wasm_carrier_action_handler,
4347 serde_json::json!({
4348 "type": "send-control",
4349 "connectionId": connection_id,
4350 "remoteEndpointId": remote_endpoint_id,
4351 "envelope": failed,
4352 }),
4353 );
4354 return Ok(true);
4355 }
4356 if !matches!(
4357 self.inner
4358 .reserve_wasm_carrier_upgrade(
4359 &connection_id,
4360 kind,
4361 &bootstrap.upgrade_id,
4362 generation,
4363 )
4364 .await,
4365 crate::client::WasmCarrierUpgradeReservation::Reserved
4366 ) {
4367 return Ok(true);
4368 }
4369 let node = self
4370 .inner
4371 .iroh_node
4372 .read()
4373 .await
4374 .as_ref()
4375 .cloned()
4376 .ok_or_else(|| JsValue::from_str("Iroh node is unavailable"))?;
4377 let authorization_expiry = node
4378 .authorize_pending_inbound_replacement(
4379 endpoint_id,
4380 crate::iroh_carrier_kind::EXPERIMENTAL_WEBRTC_TRANSPORT_ID,
4381 std::time::Duration::from_secs(45),
4382 )
4383 .await;
4384 let previous = self.wasm_webrtc_carrier_attempts.borrow_mut().insert(
4385 connection_id.clone(),
4386 WasmWebRtcCarrierAttempt {
4387 connection_id: connection_id.clone(),
4388 remote_endpoint_id: endpoint_id,
4389 bootstrap: bootstrap.clone(),
4390 generation,
4391 role: "responder",
4392 prepared: false,
4393 retry_sent: false,
4394 offer_started: false,
4395 remote_ready: false,
4396 completion_started: false,
4397 inbound_authorization_expires_at_ms: Some(authorization_expiry),
4398 ice_refresh: Default::default(),
4399 ice_refresh_baseline: Rc::new(RefCell::new(None)),
4400 },
4401 );
4402 if let Some(previous) = previous {
4403 self.wasm_webrtc_carrier_sessions
4404 .borrow_mut()
4405 .remove(&previous.bootstrap.upgrade_id);
4406 }
4407 emit_wasm_carrier_action(
4408 &self.wasm_carrier_action_handler,
4409 serde_json::json!({
4410 "type": "prepare-webrtc",
4411 "connectionId": connection_id,
4412 "remoteEndpointId": remote_endpoint_id,
4413 "role": "responder",
4414 "upgradeId": bootstrap.upgrade_id,
4415 "carrierSessionId": bootstrap.carrier_session_id,
4416 "transportGeneration": generation.transport_generation.saturating_add(1),
4417 }),
4418 );
4419 self.webrtc_carrier_owner()
4420 .schedule_wasm_webrtc_carrier_watchdog(bootstrap);
4421 }
4422 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
4423 let upgrade_id = {
4424 let mut attempts = self.wasm_webrtc_carrier_attempts.borrow_mut();
4425 attempts
4426 .get_mut(&connection_id)
4427 .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
4428 .map(|attempt| {
4429 attempt.remote_ready = true;
4430 attempt.bootstrap.upgrade_id.clone()
4431 })
4432 };
4433 let attempt = upgrade_id.as_deref().and_then(|upgrade_id| {
4434 self.take_ready_outbound_wasm_webrtc_carrier_attempt(
4435 &connection_id,
4436 upgrade_id,
4437 )
4438 });
4439 if let Some(attempt) = attempt {
4440 self.spawn_outbound_wasm_webrtc_carrier_completion(attempt);
4441 }
4442 }
4443 crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
4444 let attempt = self
4445 .wasm_webrtc_carrier_attempts
4446 .borrow()
4447 .get(&connection_id)
4448 .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
4449 .cloned();
4450 if let Some(attempt) = attempt {
4451 let failure_code = crate::iroh_carrier_bootstrap::webrtc_peer_failure_code(
4452 bootstrap.failure_code.as_deref(),
4453 );
4454 if crate::client::should_rearm_wasm_carrier_event_retry(
4455 failure_code,
4456 attempt.role == "initiator",
4457 attempt.completion_started,
4458 ) {
4459 if let Some(current) = self
4460 .wasm_webrtc_carrier_attempts
4461 .borrow_mut()
4462 .get_mut(&connection_id)
4463 .filter(|current| {
4464 current.bootstrap.upgrade_id == attempt.bootstrap.upgrade_id
4465 && !current.completion_started
4466 })
4467 {
4468 current.retry_sent = false;
4469 }
4470 return Ok(true);
4471 }
4472 self.fail_wasm_webrtc_carrier_attempt(attempt, failure_code, false)
4473 .await;
4474 }
4475 }
4476 }
4477 Ok(true)
4478 }
4479
4480 #[cfg(all(feature = "transport-moq", not(feature = "transport-webrtc")))]
4481 #[wasm_bindgen(js_name = __handleIrohCarrierControl)]
4482 pub async fn handle_iroh_carrier_control_moq_only(
4483 &self,
4484 connection_id: String,
4485 remote_endpoint_id: String,
4486 frame: JsValue,
4487 ) -> Result<bool, JsValue> {
4488 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
4489 let frame: serde_json::Value = serde_wasm_bindgen::from_value(frame)
4490 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4491 let Some(bootstrap) =
4492 crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(&frame)
4493 else {
4494 return Ok(false);
4495 };
4496 if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14
4497 {
4498 return Ok(false);
4499 }
4500 self.handle_wasm_moq_bootstrap(connection_id, remote_endpoint_id, bootstrap)
4501 .await
4502 }
4503
4504 #[cfg(feature = "transport-webrtc")]
4508 #[wasm_bindgen(js_name = __attachWebRtcCarrier)]
4509 pub async fn attach_iroh_webrtc_carrier(
4510 &self,
4511 connection_id: String,
4512 remote_endpoint_id: String,
4513 channel: web_sys::RtcDataChannel,
4514 upgrade_id: String,
4515 ) -> Result<(), JsValue> {
4516 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
4517 let endpoint_id = remote_endpoint_id
4518 .parse::<iroh::EndpointId>()
4519 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4520 let attempt = self
4521 .wasm_webrtc_carrier_attempts
4522 .borrow()
4523 .get(&connection_id)
4524 .filter(|attempt| {
4525 attempt.bootstrap.upgrade_id == upgrade_id
4526 && attempt.remote_endpoint_id == endpoint_id
4527 })
4528 .cloned()
4529 .ok_or_else(|| JsValue::from_str("WebRTC carrier attempt is stale"))?;
4530 if !self
4531 .inner
4532 .wasm_carrier_upgrade_is_current(
4533 &connection_id,
4534 crate::client::IrohPathKind::WebRtc,
4535 &upgrade_id,
4536 attempt.generation,
4537 )
4538 .await
4539 {
4540 return Err(JsValue::from_str("WebRTC carrier attempt was retired"));
4541 }
4542 let packet_session = self
4543 .inner
4544 .activate_iroh_packet_carrier(
4545 crate::iroh_carrier_kind::IrohCarrierKind::WebRtc,
4546 endpoint_id,
4547 )
4548 .await
4549 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4550 let expected = attempt
4551 .bootstrap
4552 .frame_expectation()
4553 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4554 let application_key = self
4555 .inner
4556 .application_crypto_key_for_connection(Some(&connection_id))
4557 .ok_or_else(|| {
4558 JsValue::from_str("WebRTC carrier requires the admitted application crypto key")
4559 })?;
4560 let terminal_client = self.inner.clone();
4561 let terminal_connection_id = connection_id.clone();
4562 let terminal_upgrade_id = upgrade_id.clone();
4563 let terminal_generation = attempt.generation.transport_generation.saturating_add(1);
4564 let terminal_attempts = self.wasm_webrtc_carrier_attempts.clone();
4565 let terminal_sessions = self.wasm_webrtc_carrier_sessions.clone();
4566 let terminal_handler = self.wasm_carrier_action_handler.clone();
4567 let terminal_lifecycle = self.wasm_carrier_peer_lifecycle.clone();
4568 let on_terminal: Rc<dyn Fn(&'static str)> = Rc::new(move |reason| {
4569 let client = terminal_client.clone();
4570 let connection_id = terminal_connection_id.clone();
4571 let upgrade_id = terminal_upgrade_id.clone();
4572 let attempts = terminal_attempts.clone();
4573 let sessions = terminal_sessions.clone();
4574 let handler = terminal_handler.clone();
4575 let lifecycle = terminal_lifecycle.clone();
4576 spawn_local(async move {
4577 let _lifecycle = lifecycle.lock().await;
4578 if !client
4579 .close_current_iroh_carrier_generation_with_reason(
4580 &connection_id,
4581 endpoint_id,
4582 crate::client::IrohPathKind::WebRtc,
4583 terminal_generation,
4584 reason,
4585 )
4586 .await
4587 {
4588 return;
4589 }
4590 if attempts
4598 .borrow()
4599 .get(&connection_id)
4600 .is_some_and(|current| current.bootstrap.upgrade_id == upgrade_id)
4601 {
4602 attempts.borrow_mut().remove(&connection_id);
4603 }
4604 sessions.borrow_mut().remove(&upgrade_id);
4605 emit_wasm_carrier_action(
4606 &handler,
4607 serde_json::json!({
4608 "type": "retire-webrtc",
4609 "connectionId": connection_id,
4610 "upgradeId": upgrade_id,
4611 "failureCode": reason,
4612 }),
4613 );
4614 });
4615 });
4616 let carrier = crate::wasm_webrtc_carrier::WasmWebRtcCarrierSession::attach(
4617 channel,
4618 packet_session,
4619 expected,
4620 application_key,
4621 on_terminal,
4622 )?;
4623 self.wasm_webrtc_carrier_sessions
4624 .borrow_mut()
4625 .insert(upgrade_id.clone(), carrier);
4626 if attempt.role == "responder" {
4627 let ready = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
4628 &attempt.bootstrap,
4629 )
4630 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4631 emit_wasm_carrier_action(
4632 &self.wasm_carrier_action_handler,
4633 serde_json::json!({
4634 "type": "send-control",
4635 "connectionId": connection_id,
4636 "remoteEndpointId": remote_endpoint_id,
4637 "envelope": ready,
4638 }),
4639 );
4640 self.spawn_inbound_wasm_webrtc_carrier_completion(attempt);
4641 } else if let Some(attempt) =
4642 self.take_ready_outbound_wasm_webrtc_carrier_attempt(&connection_id, &upgrade_id)
4643 {
4644 self.spawn_outbound_wasm_webrtc_carrier_completion(attempt);
4645 }
4646 Ok(())
4647 }
4648
4649 #[cfg(feature = "transport-moq")]
4653 #[wasm_bindgen(js_name = __attachMoqCarrier)]
4654 pub async fn attach_iroh_moq_carrier(
4655 &self,
4656 connection_id: String,
4657 remote_endpoint_id: String,
4658 datagrams: JsValue,
4659 upgrade_id: String,
4660 ) -> Result<(), JsValue> {
4661 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
4662 let endpoint_id = remote_endpoint_id
4663 .parse::<iroh::EndpointId>()
4664 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4665 let attempt = self
4666 .wasm_moq_carrier_attempts
4667 .borrow()
4668 .get(&connection_id)
4669 .filter(|attempt| {
4670 attempt.bootstrap.upgrade_id == upgrade_id
4671 && attempt.remote_endpoint_id == endpoint_id
4672 })
4673 .cloned()
4674 .ok_or_else(|| JsValue::from_str("MoQ carrier attempt is stale"))?;
4675 if !self
4676 .inner
4677 .wasm_carrier_upgrade_is_current(
4678 &connection_id,
4679 crate::client::IrohPathKind::Moq,
4680 &upgrade_id,
4681 attempt.generation,
4682 )
4683 .await
4684 {
4685 return Err(JsValue::from_str("MoQ carrier attempt was retired"));
4686 }
4687 let packet_session = self
4688 .inner
4689 .activate_iroh_packet_carrier(
4690 crate::iroh_carrier_kind::IrohCarrierKind::Moq,
4691 endpoint_id,
4692 )
4693 .await
4694 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4695 let expected = attempt
4696 .bootstrap
4697 .frame_expectation()
4698 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4699 let application_key = self
4700 .inner
4701 .application_crypto_key_for_connection(Some(&connection_id))
4702 .ok_or_else(|| {
4703 JsValue::from_str("MoQ carrier requires an installed application key")
4704 })?;
4705 let terminal_client = self.inner.clone();
4706 let terminal_connection_id = connection_id.clone();
4707 let terminal_upgrade_id = upgrade_id.clone();
4708 let terminal_generation = attempt.generation.transport_generation.saturating_add(1);
4709 let terminal_attempts = self.wasm_moq_carrier_attempts.clone();
4710 let terminal_sessions = self.wasm_moq_carrier_sessions.clone();
4711 let terminal_handler = self.wasm_carrier_action_handler.clone();
4712 let terminal_lifecycle = self.wasm_carrier_peer_lifecycle.clone();
4713 let on_terminal: Rc<dyn Fn(&'static str)> = Rc::new(move |reason| {
4714 let client = terminal_client.clone();
4715 let connection_id = terminal_connection_id.clone();
4716 let upgrade_id = terminal_upgrade_id.clone();
4717 let attempts = terminal_attempts.clone();
4718 let sessions = terminal_sessions.clone();
4719 let handler = terminal_handler.clone();
4720 let lifecycle = terminal_lifecycle.clone();
4721 spawn_local(async move {
4722 let _lifecycle = lifecycle.lock().await;
4723 if !client
4724 .close_current_iroh_carrier_generation_with_reason(
4725 &connection_id,
4726 endpoint_id,
4727 crate::client::IrohPathKind::Moq,
4728 terminal_generation,
4729 reason,
4730 )
4731 .await
4732 {
4733 return;
4734 }
4735 if attempts
4736 .borrow()
4737 .get(&connection_id)
4738 .is_some_and(|current| current.bootstrap.upgrade_id == upgrade_id)
4739 {
4740 attempts.borrow_mut().remove(&connection_id);
4741 }
4742 sessions.borrow_mut().remove(&upgrade_id);
4743 emit_wasm_carrier_action(
4744 &handler,
4745 serde_json::json!({
4746 "type": "retire-moq",
4747 "connectionId": connection_id,
4748 "upgradeId": upgrade_id,
4749 "failureCode": reason,
4750 }),
4751 );
4752 });
4753 });
4754 let carrier = crate::wasm_moq_carrier::WasmMoqCarrierSession::attach(
4755 datagrams,
4756 packet_session,
4757 expected,
4758 application_key,
4759 on_terminal,
4760 )?;
4761 if attempt.role == "initiator" {
4762 carrier
4763 .wait_for_peer_data_bidirectional_readiness(std::time::Duration::from_secs(15))
4764 .await?;
4765 }
4766 self.wasm_moq_carrier_sessions
4767 .borrow_mut()
4768 .insert(upgrade_id.clone(), carrier);
4769 if attempt.role == "initiator" {
4770 self.spawn_outbound_wasm_moq_carrier_completion(attempt);
4771 } else {
4772 let ready = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
4773 &attempt.bootstrap,
4774 )
4775 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4776 emit_wasm_carrier_action(
4777 &self.wasm_carrier_action_handler,
4778 serde_json::json!({
4779 "type": "send-control",
4780 "connectionId": connection_id,
4781 "remoteEndpointId": attempt.remote_endpoint_id.to_string(),
4782 "envelope": ready,
4783 }),
4784 );
4785 self.spawn_inbound_wasm_moq_carrier_completion(attempt);
4786 }
4787 Ok(())
4788 }
4789
4790 #[wasm_bindgen(js_name = setIdentityCredential)]
4791 pub fn set_identity_credential(&self, credential: Option<String>) {
4792 let has_credential = credential
4793 .as_ref()
4794 .map(|value| !value.is_empty())
4795 .unwrap_or(false);
4796 let credential_len = credential.as_ref().map(|value| value.len()).unwrap_or(0);
4797
4798 if let Ok(mut guard) = self.identity_credential.lock() {
4799 *guard = credential.filter(|value| !value.is_empty());
4800 }
4801
4802 let should_log = if let Ok(mut guard) = self.last_auth_log.lock() {
4803 let next = (has_credential, credential_len);
4804 if guard.as_ref() == Some(&next) {
4805 false
4806 } else {
4807 *guard = Some(next);
4808 true
4809 }
4810 } else {
4811 true
4812 };
4813
4814 if should_log {
4815 web_sys::console::log_1(&JsValue::from_str(&format!(
4816 "[OPENRTC][WASM-IDENTITY] credential updated present={} len={}",
4817 has_credential, credential_len
4818 )));
4819 }
4820 }
4821
4822 #[wasm_bindgen(js_name = clearIdentityCredential)]
4828 pub fn clear_identity_credential(&self) {
4829 self.set_identity_credential(None);
4830 }
4831
4832 #[wasm_bindgen(js_name = rankRoutes)]
4835 pub fn rank_routes(
4836 &self,
4837 configured_priority: Vec<String>,
4838 candidates: Vec<String>,
4839 ) -> Vec<String> {
4840 crate::route_policy::rank_routes(&configured_priority, &candidates)
4841 }
4842
4843 #[wasm_bindgen(js_name = setRelay)]
4845 pub async fn set_relay(&self, enabled: bool) -> Result<(), JsValue> {
4846 self.inner
4847 .set_relay(enabled)
4848 .await
4849 .map_err(|error| JsValue::from_str(&error.to_string()))
4850 }
4851
4852 pub async fn init_iroh(&self, secret_key: Option<Vec<u8>>) -> Result<String, JsValue> {
4853 let started_at = js_sys::Date::now();
4854 web_sys::console::log_1(&JsValue::from_str(&format!(
4855 "[OPENRTC][WASM-API] init_iroh called has_secret_key={} secret_key_len={}",
4856 secret_key.as_ref().is_some(),
4857 secret_key.as_ref().map(|k| k.len()).unwrap_or(0)
4858 )));
4859 match self.inner.init_iroh(secret_key, vec![]).await {
4860 Ok(node_id) => {
4861 self.inner.clone().start_wasm_accept_bridge();
4862 let elapsed = js_sys::Date::now() - started_at;
4863 web_sys::console::log_1(&JsValue::from_str(&format!(
4864 "[OPENRTC][WASM-API] init_iroh success elapsed_ms={:.0} node_id={}",
4865 elapsed, node_id
4866 )));
4867 Ok(node_id)
4868 }
4869 Err(err) => {
4870 let elapsed = js_sys::Date::now() - started_at;
4871 web_sys::console::error_1(&JsValue::from_str(&format!(
4872 "[OPENRTC][WASM-API] init_iroh failed elapsed_ms={:.0} error={}",
4873 elapsed, err
4874 )));
4875 Err(JsValue::from_str(&err.to_string()))
4876 }
4877 }
4878 }
4879
4880 #[wasm_bindgen(js_name = initIrohWithTestRelay)]
4883 pub async fn init_iroh_with_test_relay(
4884 &self,
4885 secret_key: Option<Vec<u8>>,
4886 test_relay_url: Option<String>,
4887 ) -> Result<String, JsValue> {
4888 let node_id = self
4889 .inner
4890 .init_iroh_with_test_relay(secret_key, vec![], test_relay_url.as_deref())
4891 .await
4892 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4893 self.inner.clone().start_wasm_accept_bridge();
4897 Ok(node_id)
4898 }
4899
4900 pub async fn iroh_secret_key(&self) -> Result<Vec<u8>, JsValue> {
4901 let node_guard = self.inner.iroh_node.read().await;
4902 if let Some(node) = node_guard.as_ref() {
4903 Ok(node.secret_key())
4904 } else {
4905 Err(JsValue::from_str("Iroh node not initialized"))
4906 }
4907 }
4908
4909 pub async fn node_addr(&self) -> Result<String, JsValue> {
4910 let node_guard = self.inner.iroh_node.read().await;
4911 if let Some(node) = node_guard.as_ref() {
4912 let addr = node
4913 .node_addr()
4914 .await
4915 .map_err(|e| JsValue::from_str(&e.to_string()))?;
4916 serde_json::to_string(&addr).map_err(|e| JsValue::from_str(&e.to_string()))
4917 } else {
4918 Err(JsValue::from_str("Iroh node not initialized"))
4919 }
4920 }
4921
4922 pub async fn endpoint_ticket(&self) -> Result<String, JsValue> {
4923 self.inner
4924 .endpoint_ticket()
4925 .await
4926 .map_err(|e| JsValue::from_str(&e.to_string()))
4927 }
4928
4929 pub async fn endpoint_ticket_with_token(
4933 &self,
4934 grant_scope: String,
4935 max_connections: u32,
4936 ) -> Result<String, JsValue> {
4937 self.inner
4938 .endpoint_ticket_with_token(&grant_scope, max_connections)
4939 .await
4940 .map_err(|e| JsValue::from_str(&e.to_string()))
4941 }
4942
4943 pub async fn issue_ticket_invite(
4946 &self,
4947 id: String,
4948 max_peers: u32,
4949 mesh: bool,
4950 ) -> Result<String, JsValue> {
4951 let invite = if mesh {
4952 self.inner
4953 .issue_ticket_invite(&id, crate::ticket_mesh::TicketMeshOptions { max_peers })
4954 .await
4955 } else {
4956 self.inner.issue_direct_ticket_invite(&id, max_peers).await
4957 };
4958 invite
4959 .map(|invite| invite.encode())
4960 .map_err(|error| JsValue::from_str(&error.to_string()))
4961 }
4962
4963 pub async fn issue_ticket_mesh(
4967 &self,
4968 id: String,
4969 max_peers: u32,
4970 action_handler: js_sys::Function,
4971 ) -> Result<String, JsValue> {
4972 if self.ticket_meshes.borrow().contains_key(&id)
4973 || !self.ticket_mesh_starts.borrow_mut().insert(id.clone())
4974 {
4975 return Err(JsValue::from_str("ticket mesh already has an owner"));
4976 }
4977 let _start = TicketMeshStartGuard {
4978 starts: &self.ticket_mesh_starts,
4979 id: id.clone(),
4980 };
4981 let mesh = crate::ticket_mesh::wasm_session::WasmTicketMesh::issue(
4982 self.inner.clone(),
4983 &id,
4984 crate::ticket_mesh::TicketMeshOptions { max_peers },
4985 action_handler,
4986 )
4987 .await
4988 .map_err(|error| JsValue::from_str(&error.to_string()))?;
4989 let invitation = mesh.invite();
4990 self.ticket_meshes.borrow_mut().insert(id, Rc::new(mesh));
4991 Ok(invitation)
4992 }
4993
4994 pub async fn join_ticket_mesh(
4995 &self,
4996 invitation: String,
4997 action_handler: js_sys::Function,
4998 ) -> Result<String, JsValue> {
4999 let invite = crate::ticket_mesh::TicketInvite::parse(&invitation)
5000 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5001 let id = invite.id().to_string();
5002 if self.ticket_meshes.borrow().contains_key(&id)
5003 || !self.ticket_mesh_starts.borrow_mut().insert(id.clone())
5004 {
5005 return Err(JsValue::from_str("ticket mesh already has an owner"));
5006 }
5007 let _start = TicketMeshStartGuard {
5008 starts: &self.ticket_mesh_starts,
5009 id: id.clone(),
5010 };
5011 let mesh = crate::ticket_mesh::wasm_session::WasmTicketMesh::join(
5012 self.inner.clone(),
5013 &invitation,
5014 action_handler,
5015 )
5016 .await
5017 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5018 self.ticket_meshes
5019 .borrow_mut()
5020 .insert(id.clone(), Rc::new(mesh));
5021 Ok(id)
5022 }
5023
5024 pub async fn ticket_mesh_peers(&self, id: String) -> Result<JsValue, JsValue> {
5025 let mesh = self
5026 .ticket_meshes
5027 .borrow()
5028 .get(&id)
5029 .cloned()
5030 .ok_or_else(|| JsValue::from_str("ticket mesh is not active"))?;
5031 serde_wasm_bindgen::to_value(&mesh.peers().await)
5032 .map_err(|error| JsValue::from_str(&error.to_string()))
5033 }
5034
5035 pub fn ticket_mesh_issuer_node(&self, id: String) -> Result<String, JsValue> {
5036 let mesh = self
5037 .ticket_meshes
5038 .borrow()
5039 .get(&id)
5040 .cloned()
5041 .ok_or_else(|| JsValue::from_str("ticket mesh is not active"))?;
5042 mesh.issuer_node()
5043 .map_err(|error| JsValue::from_str(&error.to_string()))
5044 }
5045
5046 pub fn receive_ticket_mesh_frame(
5047 &self,
5048 id: String,
5049 connection_id: String,
5050 frame: String,
5051 ) -> Result<(), JsValue> {
5052 let mesh = self
5053 .ticket_meshes
5054 .borrow()
5055 .get(&id)
5056 .cloned()
5057 .ok_or_else(|| JsValue::from_str("ticket mesh is not active"))?;
5058 mesh.receive(connection_id, frame.into_bytes());
5059 Ok(())
5060 }
5061
5062 pub async fn close_ticket_mesh(&self, id: String) -> Result<(), JsValue> {
5063 if self.ticket_mesh_starts.borrow().contains(&id) {
5064 return Err(JsValue::from_str(
5065 "ticket mesh is still starting; await startup before closing",
5066 ));
5067 }
5068 let mesh = self.ticket_meshes.borrow().get(&id).cloned();
5069 if let Some(mesh) = mesh {
5070 mesh.close().await;
5071 let mut meshes = self.ticket_meshes.borrow_mut();
5072 if meshes
5073 .get(&id)
5074 .is_some_and(|current| Rc::ptr_eq(current, &mesh))
5075 {
5076 meshes.remove(&id);
5077 }
5078 }
5079 Ok(())
5080 }
5081
5082 pub async fn connect_ticket_invite(&self, invitation: String) -> Result<JsValue, JsValue> {
5085 let result = self
5086 .inner
5087 .connect_ticket_invite(&invitation)
5088 .await
5089 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5090 serde_wasm_bindgen::to_value(&result)
5091 .map_err(|error| JsValue::from_str(&error.to_string()))
5092 }
5093
5094 #[wasm_bindgen(js_name = __requireScopePeerAdmission)]
5096 pub fn require_scope_peer_admission(&self, scope: String) -> Result<(), JsValue> {
5097 self.inner
5098 .session_token_registry
5099 .require_scope_peer_admission(&scope)
5100 .map_err(|error| JsValue::from_str(&error))
5101 }
5102
5103 #[wasm_bindgen(js_name = __updateScopePeerAdmission)]
5106 pub fn update_scope_peer_admission(
5107 &self,
5108 scope: String,
5109 revision: f64,
5110 expires_at_ms: f64,
5111 peers_json: String,
5112 ) -> Result<(), JsValue> {
5113 let valid_integer = |value: f64| {
5114 value.is_finite()
5115 && value >= 0.0
5116 && value <= 9_007_199_254_740_991.0
5117 && value.fract() == 0.0
5118 };
5119 if !valid_integer(revision) || !valid_integer(expires_at_ms) || peers_json.len() > 4_000
5120 {
5121 return Err(JsValue::from_str("invalid peer admission lease"));
5122 }
5123 let peers = serde_json::from_str(&peers_json)
5124 .map_err(|_| JsValue::from_str("invalid peer admission peers"))?;
5125 self.inner
5126 .session_token_registry
5127 .update_scope_peer_admission(&scope, revision as u64, expires_at_ms as u64, peers)
5128 .map_err(|error| JsValue::from_str(&error))
5129 }
5130
5131 pub fn register_session_token(
5133 &self,
5134 token: String,
5135 grant_scope: String,
5136 max_connections: u32,
5137 ) {
5138 self.inner
5139 .register_session_token(token, grant_scope, max_connections);
5140 }
5141
5142 pub fn register_token_until(
5144 &self,
5145 token: String,
5146 grant_scope: String,
5147 max_connections: u32,
5148 expires_at_ms: u64,
5149 ) {
5150 self.inner
5151 .register_token_until(token, grant_scope, max_connections, expires_at_ms);
5152 }
5153
5154 #[wasm_bindgen(js_name = setConnectionApplicationCryptoRequired)]
5156 pub fn set_connection_application_crypto_required(
5157 &self,
5158 connection_id: String,
5159 ) -> Result<(), JsValue> {
5160 self.inner
5161 .set_connection_application_crypto_required(&connection_id);
5162 Ok(())
5163 }
5164
5165 #[wasm_bindgen(js_name = setConnectionApplicationCryptoKey)]
5167 pub async fn set_connection_application_crypto_key(
5168 &self,
5169 connection_id: String,
5170 key: Vec<u8>,
5171 ) -> Result<(), JsValue> {
5172 if key.len() != crate::application_crypto::APPLICATION_KEY_BYTES {
5173 return Err(JsValue::from_str("application crypto key must be 32 bytes"));
5174 }
5175 let mut key_bytes = [0u8; crate::application_crypto::APPLICATION_KEY_BYTES];
5176 key_bytes.copy_from_slice(&key);
5177 self.inner
5178 .set_connection_application_crypto_key(&connection_id, key_bytes);
5179 self.inner
5180 .emit_current_wasm_connection_state(&connection_id)
5181 .await;
5182 Ok(())
5183 }
5184
5185 #[wasm_bindgen(js_name = clearConnectionApplicationCryptoKey)]
5190 pub fn clear_connection_application_crypto_key(&self, connection_id: String) {
5191 self.inner
5192 .clear_connection_application_crypto_key(&connection_id);
5193 }
5194
5195 #[wasm_bindgen(js_name = ensureConnectionApplicationCrypto)]
5199 pub async fn ensure_connection_application_crypto(
5200 &self,
5201 connection_id: String,
5202 remote_node_id: String,
5203 timeout_ms: u64,
5204 send_frame: js_sys::Function,
5205 ) -> Result<(), JsValue> {
5206 let connection_id = connection_id.trim().to_string();
5207 if connection_id.is_empty() {
5208 return Err(JsValue::from_str("connection id is required"));
5209 }
5210 let remote_node_id = remote_node_id.trim().to_string();
5211 if remote_node_id.is_empty() {
5212 return Err(JsValue::from_str("remote node id is required"));
5213 }
5214 let remote_endpoint_id: iroh::EndpointId = remote_node_id.parse().map_err(|error| {
5215 JsValue::from_str(&format!("invalid remote endpoint id: {error}"))
5216 })?;
5217 self.inner
5223 .ensure_connection_manager_record_before_peer_stream(&remote_endpoint_id)
5224 .await
5225 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5226 let timeout_ms = timeout_ms.max(1);
5227 let initial = self
5228 .inner
5229 .connection_manager
5230 .get_by_connection_id(&connection_id)
5231 .await
5232 .ok_or_else(|| JsValue::from_str("connection is not registered"))?;
5233 let initial_endpoint = initial
5234 .endpoint_id
5235 .as_deref()
5236 .or(initial.node_id.as_deref())
5237 .map(ToOwned::to_owned)
5238 .ok_or_else(|| JsValue::from_str("connection has no endpoint"))?;
5239 let initial_transport_stable_id = initial.transport_stable_id;
5240 let initial_transport_generation = initial.transport_generation;
5241 let initial_route_generation = initial.route_generation;
5242 self.inner
5243 .set_connection_application_crypto_required(&connection_id);
5244
5245 let started_at_ms = js_sys::Date::now();
5246 let retry_delays_ms = [0_u64, 250, 750];
5247 for (attempt, delay_ms) in retry_delays_ms.into_iter().enumerate() {
5248 if delay_ms > 0 {
5249 let remaining_ms = timeout_ms
5250 .saturating_sub((js_sys::Date::now() - started_at_ms).max(0.0) as u64);
5251 if remaining_ms == 0 {
5252 break;
5253 }
5254 gloo_timers::future::sleep(std::time::Duration::from_millis(
5255 delay_ms.min(remaining_ms),
5256 ))
5257 .await;
5258 }
5259 self.inner
5260 .assert_application_crypto_generation(
5261 &connection_id,
5262 &initial_endpoint,
5263 initial_transport_stable_id,
5264 initial_transport_generation,
5265 initial_route_generation,
5266 )
5267 .await
5268 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5269 if self
5270 .inner
5271 .connection_application_crypto_key(&connection_id)
5272 .is_some()
5273 && self.inner.connection_application_crypto_is_confirmed(
5274 &connection_id,
5275 initial_transport_stable_id,
5276 )
5277 {
5278 let confirmed_transport_stable_id =
5279 initial_transport_stable_id.ok_or_else(|| {
5280 JsValue::from_str("confirmed connection has no stable ID")
5281 })?;
5282 if !self
5283 .inner
5284 .confirm_managed_connection_readiness_from_transport_proof(
5285 &connection_id,
5286 confirmed_transport_stable_id,
5287 )
5288 .await
5289 {
5290 return Err(JsValue::from_str(
5291 "confirmed application route belongs to a retired generation",
5292 ));
5293 }
5294 self.inner
5295 .emit_current_wasm_connection_state(&connection_id)
5296 .await;
5297 return Ok(());
5298 }
5299 let frame = self
5300 .inner
5301 .application_key_handshake_frame(&connection_id, "capability-update")
5302 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5303 let frame = serde::Serialize::serialize(
5304 &frame,
5305 &serde_wasm_bindgen::Serializer::json_compatible(),
5306 )
5307 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5308 let send_result = send_frame.call1(&JsValue::NULL, &frame)?;
5309 wasm_bindgen_futures::JsFuture::from(js_sys::Promise::resolve(&send_result))
5310 .await?;
5311
5312 let attempt_deadline_ms = ((attempt + 1) as u64 * timeout_ms / 3).max(1);
5313 loop {
5314 if self
5315 .inner
5316 .connection_application_crypto_key(&connection_id)
5317 .is_some()
5318 && self.inner.connection_application_crypto_is_confirmed(
5319 &connection_id,
5320 initial_transport_stable_id,
5321 )
5322 {
5323 let confirmed_transport_stable_id = initial_transport_stable_id
5324 .ok_or_else(|| {
5325 JsValue::from_str("confirmed connection has no stable ID")
5326 })?;
5327 if !self
5328 .inner
5329 .confirm_managed_connection_readiness_from_transport_proof(
5330 &connection_id,
5331 confirmed_transport_stable_id,
5332 )
5333 .await
5334 {
5335 return Err(JsValue::from_str(
5336 "confirmed application route belongs to a retired generation",
5337 ));
5338 }
5339 self.inner
5340 .emit_current_wasm_connection_state(&connection_id)
5341 .await;
5342 return Ok(());
5343 }
5344 let elapsed_ms = (js_sys::Date::now() - started_at_ms).max(0.0) as u64;
5345 if elapsed_ms >= attempt_deadline_ms || elapsed_ms >= timeout_ms {
5346 break;
5347 }
5348 self.inner
5349 .assert_application_crypto_generation(
5350 &connection_id,
5351 &initial_endpoint,
5352 initial_transport_stable_id,
5353 initial_transport_generation,
5354 initial_route_generation,
5355 )
5356 .await
5357 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5358 gloo_timers::future::sleep(std::time::Duration::from_millis(25)).await;
5359 }
5360 }
5361 let current_manager_transport_stable_id = self
5362 .inner
5363 .connection_manager
5364 .get_by_connection_id(&connection_id)
5365 .await
5366 .and_then(|record| record.transport_stable_id);
5367 let current_physical_transport_stable_id = self
5368 .inner
5369 .get_connection(remote_endpoint_id)
5370 .await
5371 .map(|connection| crate::transport_generation::for_connection(&connection));
5372 let confirmation = self
5373 .inner
5374 .connection_application_crypto_confirmation_snapshot(&connection_id);
5375 return Err(JsValue::from_str(&format!(
5376 "connection {} reciprocal application key agreement timed out after {}ms: frozen_stable_id={:?} current_manager_stable_id={:?} current_physical_stable_id={:?} confirmation={:?}",
5377 connection_id,
5378 timeout_ms,
5379 initial_transport_stable_id,
5380 current_manager_transport_stable_id,
5381 current_physical_transport_stable_id,
5382 confirmation,
5383 )));
5384 }
5385
5386 #[wasm_bindgen(js_name = handleConnectionApplicationCryptoHandshake)]
5387 pub async fn handle_connection_application_crypto_handshake(
5388 &self,
5389 connection_id: String,
5390 remote_node_id: String,
5391 transport_stable_id: Option<u64>,
5392 action: Option<String>,
5393 remote_public_key: Vec<u8>,
5394 ) -> Result<JsValue, JsValue> {
5395 if remote_public_key.len() != crate::key_agreement::PUBLIC_KEY_BYTES {
5396 return Err(JsValue::from_str(
5397 "application key agreement public key must be 32 bytes",
5398 ));
5399 }
5400 let mut remote_public = [0_u8; crate::key_agreement::PUBLIC_KEY_BYTES];
5401 remote_public.copy_from_slice(&remote_public_key);
5402 let outcome = self
5403 .inner
5404 .accept_application_key_handshake(
5405 &connection_id,
5406 &remote_node_id,
5407 transport_stable_id,
5408 action.as_deref(),
5409 remote_public,
5410 )
5411 .await
5412 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5413 let reply_frame = outcome
5414 .reply_action
5415 .map(|reply_action| {
5416 self.inner
5417 .application_key_handshake_frame(&connection_id, reply_action)
5418 })
5419 .transpose()
5420 .map_err(|error| JsValue::from_str(&error.to_string()))?;
5421 self.inner
5422 .emit_current_wasm_connection_state(&connection_id)
5423 .await;
5424 serde::Serialize::serialize(
5425 &reply_frame,
5426 &serde_wasm_bindgen::Serializer::json_compatible(),
5427 )
5428 .map_err(|error| JsValue::from_str(&error.to_string()))
5429 }
5430
5431 #[wasm_bindgen(js_name = connectionApplicationCryptoReady)]
5432 pub fn connection_application_crypto_ready(
5433 &self,
5434 connection_id: String,
5435 transport_stable_id: u64,
5436 ) -> bool {
5437 self.inner
5438 .connection_application_crypto_key(&connection_id)
5439 .is_some()
5440 && self.inner.connection_application_crypto_is_confirmed(
5441 &connection_id,
5442 Some(transport_stable_id),
5443 )
5444 }
5445
5446 #[wasm_bindgen(js_name = protectConnectionApplicationPayload)]
5447 pub fn protect_connection_application_payload(
5448 &self,
5449 connection_id: String,
5450 type_id: u8,
5451 payload: Vec<u8>,
5452 ) -> Result<Vec<u8>, JsValue> {
5453 self.inner
5454 .protect_outbound_application_payload_with_type(&connection_id, type_id, &payload)
5455 .map_err(|error| JsValue::from_str(&error.to_string()))
5456 }
5457
5458 #[wasm_bindgen(js_name = openConnectionApplicationPayload)]
5459 pub fn open_connection_application_payload(
5460 &self,
5461 connection_id: String,
5462 type_id: u8,
5463 payload: Vec<u8>,
5464 ) -> Result<Vec<u8>, JsValue> {
5465 self.inner
5466 .open_inbound_application_payload_with_type(&connection_id, type_id, &payload)
5467 .map_err(|error| JsValue::from_str(&error.to_string()))
5468 }
5469
5470 pub fn validate_session_token(&self, token: String) -> Result<String, JsValue> {
5474 self.inner
5475 .validate_session_token(&token)
5476 .map_err(|e| JsValue::from_str(&e))
5477 }
5478
5479 pub async fn validate_connection_token(
5481 &self,
5482 token: String,
5483 connection_id: String,
5484 token_payload: Option<String>,
5485 ) -> Result<String, JsValue> {
5486 self.inner
5487 .validate_connection_token(&token, &connection_id, token_payload.as_deref())
5488 .await
5489 .map_err(|e| session_admission_js_error(&e))
5490 }
5491
5492 pub async fn present_session_token(
5499 &self,
5500 endpoint_id: String,
5501 token: String,
5502 token_payload: Option<String>,
5503 device_id: Option<String>,
5504 ) -> Result<String, JsValue> {
5505 crate::console_log!(
5506 "[OpenRTC][session-admission][wasm-present] endpoint_id={} claimed_local_device_id={}",
5507 endpoint_id,
5508 device_id.as_deref().unwrap_or("<none>")
5509 );
5510 let endpoint_id_parsed: iroh::EndpointId = endpoint_id
5511 .parse()
5512 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
5513 let local_node_id = self.inner.current_node_id().await.ok_or_else(|| {
5514 JsValue::from_str("missing local node id for session-token presentation")
5515 })?;
5516 let connection_id =
5517 crate::client::Client::deterministic_connection_id(&local_node_id, &endpoint_id);
5518 let approval_scope = self
5519 .inner
5520 .present_and_accept_session_token_with_local_claim(
5521 endpoint_id_parsed,
5522 &connection_id,
5523 &token,
5524 token_payload.as_deref(),
5525 None,
5526 device_id,
5527 )
5528 .await
5529 .map_err(|e| session_admission_js_error(&e))?;
5530 self.inner
5531 .emit_current_wasm_connection_state(&connection_id)
5532 .await;
5533
5534 Ok(approval_scope)
5535 }
5536
5537 pub async fn is_remote_admitted(&self, endpoint_ticket: String) -> Result<bool, JsValue> {
5540 self.inner
5541 .is_remote_admitted(&endpoint_ticket)
5542 .await
5543 .map_err(|error| JsValue::from_str(&error.to_string()))
5544 }
5545
5546 #[allow(clippy::too_many_arguments)]
5549 pub async fn prepare_reciprocal_admission(
5550 &self,
5551 endpoint_id: String,
5552 expected_transport_stable_id: u64,
5553 stream_instance_id: String,
5554 presentation_id: String,
5555 token: String,
5556 token_payload: String,
5557 device_id: String,
5558 stream_contract: String,
5559 ) -> Result<bool, JsValue> {
5560 let endpoint_id: iroh::EndpointId = endpoint_id
5561 .parse()
5562 .map_err(|error| JsValue::from_str(&format!("{error}")))?;
5563 let stream_contract = match stream_contract.trim() {
5564 "one-shot-admission" => {
5565 crate::native_protocol::TokenStreamContract::OneShotAdmission
5566 }
5567 "persistent-control" => {
5568 crate::native_protocol::TokenStreamContract::PersistentControl
5569 }
5570 other => {
5571 return Err(JsValue::from_str(&format!(
5572 "unsupported reciprocal stream contract: {other}"
5573 )))
5574 }
5575 };
5576 self.inner
5577 .prepare_reciprocal_admission(
5578 endpoint_id,
5579 expected_transport_stable_id,
5580 stream_instance_id.as_str(),
5581 presentation_id.as_str(),
5582 token.as_str(),
5583 token_payload.as_str(),
5584 device_id.as_str(),
5585 stream_contract,
5586 )
5587 .await
5588 .map_err(|error| JsValue::from_str(&error))?;
5589 Ok(true)
5590 }
5591
5592 pub async fn confirm_reciprocal_admission(
5595 &self,
5596 endpoint_id: String,
5597 expected_transport_stable_id: u64,
5598 stream_instance_id: String,
5599 presentation_id: String,
5600 accepted: bool,
5601 approval_scope: Option<String>,
5602 ) -> Result<bool, JsValue> {
5603 let endpoint_id: iroh::EndpointId = endpoint_id
5604 .parse()
5605 .map_err(|error| JsValue::from_str(&format!("{error}")))?;
5606 self.inner
5607 .confirm_reciprocal_admission(
5608 endpoint_id,
5609 expected_transport_stable_id,
5610 stream_instance_id.as_str(),
5611 presentation_id.as_str(),
5612 accepted,
5613 approval_scope.as_deref(),
5614 )
5615 .await
5616 .map_err(|error| JsValue::from_str(&error))?;
5617 self.inner
5618 .emit_current_wasm_connection_state(
5619 &crate::client::Client::deterministic_connection_id(
5620 &self.inner.current_node_id().await.ok_or_else(|| {
5621 JsValue::from_str(
5622 "missing local node id after reciprocal admission ACK",
5623 )
5624 })?,
5625 &endpoint_id.to_string(),
5626 ),
5627 )
5628 .await;
5629 Ok(true)
5630 }
5631
5632 pub fn revoke_session_token(&self, token: String) -> Result<JsValue, JsValue> {
5634 let receipt = self.inner.begin_revoke_session_token(&token);
5635 let affected = receipt.connection_ids();
5636 let inner = self.inner.clone();
5637 let carriers = self.revocation_carriers();
5638 wasm_bindgen_futures::spawn_local(finish_wasm_revocation(
5639 inner, receipt, None, carriers,
5640 ));
5641 serde_wasm_bindgen::to_value(&affected)
5642 .map_err(|error| JsValue::from_str(&error.to_string()))
5643 }
5644
5645 pub async fn revoke_tokens_by_scope(
5647 &self,
5648 grant_scope: String,
5649 ) -> Result<JsValue, JsValue> {
5650 let receipt = self.inner.begin_revoke_tokens_by_scope(&grant_scope);
5651 let affected = receipt.connection_ids();
5652 finish_wasm_revocation(
5653 self.inner.clone(),
5654 receipt,
5655 Some(grant_scope),
5656 self.revocation_carriers(),
5657 )
5658 .await;
5659 serde_wasm_bindgen::to_value(&affected).map_err(|e| JsValue::from_str(&e.to_string()))
5660 }
5661
5662 pub fn clear_session_tokens(&self) {
5664 self.inner.clear_session_tokens();
5665 }
5666
5667 pub fn endpoint_id_from_ticket(&self, ticket: String) -> Result<String, JsValue> {
5668 let (iroh_ticket, _token_suffix) = split_ticket(ticket.trim());
5669 let parsed = EndpointTicket::from_str(iroh_ticket)
5670 .map_err(|e| JsValue::from_str(&format!("Invalid endpoint ticket: {}", e)))?;
5671 Ok(parsed.endpoint_addr().id.to_string())
5672 }
5673
5674 pub async fn disconnect(&self, endpoint_id: String) -> Result<(), JsValue> {
5675 let endpoint_id: iroh::EndpointId = endpoint_id
5676 .parse()
5677 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
5678 let node_guard = self.inner.iroh_node.read().await;
5679 if let Some(node) = node_guard.as_ref() {
5680 node.disconnect(endpoint_id)
5681 .await
5682 .map_err(|e| JsValue::from_str(&e.to_string()))
5683 } else {
5684 Err(JsValue::from_str("Iroh node not initialized"))
5685 }
5686 }
5687
5688 pub async fn disconnect_transient(&self, endpoint_id: String) -> Result<(), JsValue> {
5701 let endpoint_id: iroh::EndpointId = endpoint_id
5702 .parse()
5703 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
5704 self.inner
5705 .disconnect_with_reason(
5706 endpoint_id,
5707 crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
5708 )
5709 .await
5710 .map_err(|e| JsValue::from_str(&e.to_string()))?;
5711 self.inner.wake_browser_auto_connect();
5717 Ok(())
5718 }
5719
5720 pub async fn is_connected(&self, endpoint_id: String) -> Result<bool, JsValue> {
5721 let endpoint_id: iroh::EndpointId = endpoint_id
5722 .parse()
5723 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
5724 Ok(self.inner.is_connected(endpoint_id).await)
5725 }
5726
5727 pub fn runtime_policy(&self) -> Result<JsValue, JsValue> {
5728 serde_wasm_bindgen::to_value(&self.inner.policy_snapshot())
5729 .map_err(|e| JsValue::from_str(&e.to_string()))
5730 }
5731
5732 pub async fn add_peer_scope(&self, id: String, scope: String) -> Result<JsValue, JsValue> {
5733 let scopes = self.inner.add_peer_scope(&id, &scope).await;
5734 serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
5735 }
5736
5737 pub async fn release_peer_scope(
5738 &self,
5739 id: String,
5740 scope: Option<String>,
5741 ) -> Result<JsValue, JsValue> {
5742 let scopes = self.inner.release_peer_scope(&id, scope.as_deref()).await;
5743 serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
5744 }
5745
5746 pub async fn peer_scopes(&self, id: String) -> Result<JsValue, JsValue> {
5747 let scopes = self.inner.peer_scopes(&id).await;
5748 serde_wasm_bindgen::to_value(&scopes).map_err(|e| JsValue::from_str(&e.to_string()))
5749 }
5750
5751 pub async fn same_peer(&self, left: String, right: String) -> Result<bool, JsValue> {
5752 Ok(self.inner.same_peer(&left, &right).await)
5753 }
5754
5755 pub async fn peer_snapshot(&self, id: String) -> Result<JsValue, JsValue> {
5756 let snapshot = self.inner.peer_snapshot(&id).await;
5757 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5758 }
5759
5760 pub async fn peer_session(&self, id: String) -> Result<JsValue, JsValue> {
5764 let snapshot = self.inner.peer_session(&id).await;
5765 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5766 }
5767
5768 pub async fn peer_sessions(&self) -> Result<JsValue, JsValue> {
5769 let snapshots = self.inner.peer_sessions().await;
5770 serde_wasm_bindgen::to_value(&snapshots).map_err(|e| JsValue::from_str(&e.to_string()))
5771 }
5772
5773 pub async fn connection_state(&self, connection_id: String) -> Result<JsValue, JsValue> {
5774 let mut snapshot = self.inner.connection_state(&connection_id).await;
5775 #[cfg(feature = "transport-moq")]
5776 if let Some(value) = snapshot.as_mut() { self.enrich_moq_object_quantities(value); }
5777 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5778 }
5779
5780 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
5781 async fn revisit_preferred_iroh_carrier_after_settlement(&self, connection_id: &str) {
5782 let _lifecycle = self.wasm_carrier_peer_lifecycle.lock().await;
5783 let Some(record) = self
5784 .inner
5785 .connection_manager
5786 .get_by_connection_id(connection_id)
5787 .await
5788 else {
5789 return;
5790 };
5791 if matches!(
5794 record.active_transport.as_str(),
5795 crate::transport_label::WEBRTC | crate::transport_label::MOQ
5796 ) {
5797 return;
5798 }
5799 let Some(capabilities) = self
5800 .wasm_remote_carrier_capabilities
5801 .borrow()
5802 .get(connection_id)
5803 .copied()
5804 else {
5805 return;
5806 };
5807 let Some(remote_endpoint_id) = self
5808 .inner
5809 .wasm_iroh_carrier_remote_endpoint_id(connection_id)
5810 .await
5811 else {
5812 return;
5813 };
5814 if let Err(error) = self
5815 .begin_preferred_iroh_carrier_locked(
5816 connection_id.to_string(),
5817 remote_endpoint_id,
5818 capabilities.webrtc,
5819 capabilities.moq,
5820 )
5821 .await
5822 {
5823 web_sys::console::warn_1(&JsValue::from_str(&format!(
5824 "[OpenRTC][Iroh carrier] settled-generation revisit failed connection_id={} error={:?}",
5825 connection_id, error,
5826 )));
5827 }
5828 }
5829
5830 #[cfg(feature = "transport-moq")]
5831 fn enrich_moq_object_quantities(&self, snapshot: &mut crate::client::StateSnapshot) {
5832 snapshot.moq_object_quantities = None;
5833 if !snapshot.routable || snapshot.replacement_in_progress || snapshot.carrier.as_deref() != Some("moq") { return; }
5834 let attempts = self.wasm_moq_carrier_attempts.borrow();
5835 let Some(attempt) = attempts.get(&snapshot.connection_id)
5836 .filter(|attempt| attempt.generation.transport_generation.checked_add(1) == Some(snapshot.transport_generation)) else { return; };
5837 snapshot.moq_object_quantities = self.wasm_moq_carrier_sessions.borrow()
5838 .get(&attempt.bootstrap.upgrade_id).and_then(|session| session.object_quantities());
5839 }
5840
5841 pub async fn diagnostic_snapshot(&self) -> Result<JsValue, JsValue> {
5843 serde_wasm_bindgen::to_value(&self.inner.diagnostic_snapshot().await)
5844 .map_err(|_| JsValue::from_str("Unable to serialize diagnostic snapshot"))
5845 }
5846
5847 pub async fn connection_states(&self) -> Result<JsValue, JsValue> {
5848 let mut snapshots = self.inner.connection_states().await;
5849 #[cfg(feature = "transport-moq")]
5850 for snapshot in &mut snapshots { self.enrich_moq_object_quantities(snapshot); }
5851 serde_wasm_bindgen::to_value(&snapshots).map_err(|e| JsValue::from_str(&e.to_string()))
5852 }
5853
5854 pub async fn wait_for_peer(
5855 &self,
5856 id: String,
5857 timeout_ms: Option<u32>,
5858 ) -> Result<JsValue, JsValue> {
5859 let snapshot = self
5860 .inner
5861 .wait_for_peer(&id, timeout_ms.map(|value| value as u64))
5862 .await;
5863 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5864 }
5865
5866 pub async fn wait_for_settled_scope(
5867 &self,
5868 scope: String,
5869 timeout_ms: Option<u32>,
5870 ) -> Result<JsValue, JsValue> {
5871 let snapshot = self
5872 .inner
5873 .wait_for_settled_scope(&scope, timeout_ms.map(u64::from))
5874 .await;
5875 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5876 }
5877
5878 pub async fn resolve_peer_connection_records(
5879 &self,
5880 id: String,
5881 ) -> Result<JsValue, JsValue> {
5882 let records = self.inner.resolve_peer_connection_records(&id).await;
5883 serde_wasm_bindgen::to_value(&records).map_err(|e| JsValue::from_str(&e.to_string()))
5884 }
5885
5886 pub async fn list_managed_connections(&self) -> Result<JsValue, JsValue> {
5887 let records = self.inner.list_managed_connections().await;
5888 serde_wasm_bindgen::to_value(&records).map_err(|e| JsValue::from_str(&e.to_string()))
5889 }
5890
5891 pub async fn bind_connection_device_id(
5892 &self,
5893 connection_id: String,
5894 device_id: String,
5895 ) -> Result<JsValue, JsValue> {
5896 let snapshot = self
5897 .inner
5898 .bind_connection_device_id(&connection_id, &device_id)
5899 .await;
5900 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5901 }
5902
5903 pub async fn acknowledge_session_token_response(
5904 &self,
5905 connection_id: String,
5906 remote_node_id: String,
5907 transport_stable_id: u64,
5908 claimed_device_id: Option<String>,
5909 ) -> bool {
5910 self.inner
5911 .acknowledge_wasm_session_token_response(
5912 &connection_id,
5913 &remote_node_id,
5914 transport_stable_id,
5915 claimed_device_id.as_deref(),
5916 )
5917 .await
5918 }
5919
5920 pub async fn bind_node_device_id(
5921 &self,
5922 node_id: String,
5923 device_id: String,
5924 ) -> Result<(), JsValue> {
5925 self.inner.bind_node_device_id(&node_id, &device_id).await;
5926 Ok(())
5927 }
5928
5929 pub async fn reject_connection_admission(
5930 &self,
5931 connection_id: String,
5932 reason: String,
5933 ) -> Result<JsValue, JsValue> {
5934 self.inner
5935 .reject_session_connection(&connection_id, &reason);
5936 self.inner
5937 .emit_current_wasm_connection_state(&connection_id)
5938 .await;
5939 let snapshot = self.inner.connection_state(&connection_id).await;
5940 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5941 }
5942
5943 pub async fn report_managed_connection_settled(
5944 &self,
5945 connection_id: String,
5946 settled: bool,
5947 device_id: Option<String>,
5948 transport_stable_id: Option<u64>,
5949 transport_generation: Option<u64>,
5950 route_generation: Option<u64>,
5951 ) -> Result<JsValue, JsValue> {
5952 let snapshot = match (transport_stable_id, transport_generation, route_generation) {
5953 (Some(transport_stable_id), Some(transport_generation), Some(route_generation)) => {
5954 self.inner
5955 .report_managed_connection_settled_for_transport(
5956 &connection_id,
5957 settled,
5958 transport_stable_id,
5959 transport_generation,
5960 route_generation,
5961 )
5962 .await
5963 }
5964 _ => None,
5965 };
5966 let _ = device_id;
5967 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
5968 if settled && snapshot.is_some() {
5969 self.revisit_preferred_iroh_carrier_after_settlement(&connection_id)
5975 .await;
5976 }
5977 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
5978 }
5979
5980 pub async fn report_transport_status(
5981 &self,
5982 connection_id: String,
5983 active_transport: String,
5984 parallel_transport: Option<String>,
5985 transport_stable_id: Option<u64>,
5986 transport_generation: Option<u64>,
5987 route_generation: Option<u64>,
5988 ) -> Result<JsValue, JsValue> {
5989 let snapshot = match (transport_stable_id, transport_generation, route_generation) {
5990 (Some(transport_stable_id), Some(transport_generation), Some(route_generation)) => {
5991 self.inner
5992 .report_transport_status_for_generation(
5993 &connection_id,
5994 &active_transport,
5995 parallel_transport.as_deref(),
5996 transport_stable_id,
5997 transport_generation,
5998 route_generation,
5999 )
6000 .await
6001 }
6002 _ => None,
6003 };
6004 serde_wasm_bindgen::to_value(&snapshot).map_err(|e| JsValue::from_str(&e.to_string()))
6005 }
6006
6007 pub async fn is_current_transport_stable_id(
6008 &self,
6009 endpoint_id: String,
6010 transport_stable_id: u64,
6011 ) -> Result<bool, JsValue> {
6012 let endpoint_id = endpoint_id
6013 .parse::<iroh::EndpointId>()
6014 .map_err(|error| JsValue::from_str(&error.to_string()))?;
6015 Ok(self
6016 .inner
6017 .is_current_transport_stable_id(endpoint_id, transport_stable_id)
6018 .await)
6019 }
6020
6021 #[wasm_bindgen(js_name = isAuthenticatedReconnectCandidate)]
6025 pub async fn is_authenticated_reconnect_candidate(
6026 &self,
6027 endpoint_id: String,
6028 transport_stable_id: u64,
6029 ) -> Result<bool, JsValue> {
6030 let endpoint_id = endpoint_id
6031 .parse::<iroh::EndpointId>()
6032 .map_err(|error| JsValue::from_str(&error.to_string()))?;
6033 Ok(self
6034 .inner
6035 .incoming_stream_generation_is_authenticated_reconnect_candidate(
6036 endpoint_id,
6037 transport_stable_id,
6038 )
6039 .await)
6040 }
6041
6042 pub async fn open_bi(&self, endpoint_id: String) -> Result<BiStream, JsValue> {
6043 let endpoint_id: iroh::EndpointId = endpoint_id
6044 .parse()
6045 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
6046 self.inner
6047 .assert_raw_peer_stream_allowed(&endpoint_id)
6048 .await
6049 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6050 self.inner
6059 .ensure_connection_manager_record_before_peer_stream(&endpoint_id)
6060 .await
6061 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6062 let (send, recv) = self
6063 .inner
6064 .open_bi_internal(endpoint_id)
6065 .await
6066 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6067 Ok(BiStream::from_parts(send, recv, endpoint_id.to_string()))
6068 }
6069
6070 pub async fn open_native_main_control_bi(
6099 &self,
6100 endpoint_id: String,
6101 ) -> Result<BiStream, JsValue> {
6102 let endpoint_id: iroh::EndpointId = endpoint_id
6103 .parse()
6104 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
6105 self.inner
6106 .ensure_connection_manager_record_before_peer_stream(&endpoint_id)
6107 .await
6108 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6109 let (transport_stable_id, send, recv) = self
6110 .inner
6111 .open_bi_internal_with_transport_stable_id(endpoint_id)
6112 .await
6113 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6114 Ok(BiStream::from_parts_with_transport_stable_id(
6115 send,
6116 recv,
6117 endpoint_id.to_string(),
6118 transport_stable_id,
6119 ))
6120 }
6121
6122 pub async fn send_native_main_control_frame(
6127 &self,
6128 endpoint_id: String,
6129 frame: Vec<u8>,
6130 ) -> Result<(), JsValue> {
6131 send_native_signal_control_frame(&self.inner, endpoint_id, frame).await
6132 }
6133
6134 pub async fn open_peer_bi(
6135 &self,
6136 id: String,
6137 timeout_ms: Option<u32>,
6138 ) -> Result<BiStream, JsValue> {
6139 let (_connection_id, remote_node_id, send, recv) = self
6140 .inner
6141 .open_peer_bi(&id, timeout_ms.map(|value| value as u64))
6142 .await
6143 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6144 Ok(BiStream::from_peer_parts(send, recv, remote_node_id))
6145 }
6146
6147 pub async fn send_peer_application_frame(
6151 &self,
6152 id: String,
6153 frame: Vec<u8>,
6154 timeout_ms: Option<u32>,
6155 ) -> Result<(), JsValue> {
6156 self.inner
6157 .send_peer_application_frame(&id, &frame, timeout_ms.map(|value| value as u64))
6158 .await
6159 .map_err(|error| JsValue::from_str(&error.to_string()))
6160 }
6161
6162 pub async fn open_peer_bi_explicit_file_sender(
6168 &self,
6169 id: String,
6170 timeout_ms: Option<u32>,
6171 ) -> Result<PeerUniStream, JsValue> {
6172 let (_connection_id, _remote_node_id, send) = self
6173 .inner
6174 .open_peer_bi_explicit_file_sender(&id, timeout_ms.map(|value| value as u64))
6175 .await
6176 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6177 Ok(peer_uni_stream_from_send(send))
6178 }
6179
6180 pub async fn open_peer_bi_transport_only(
6184 &self,
6185 id: String,
6186 timeout_ms: Option<u32>,
6187 ) -> Result<BiStream, JsValue> {
6188 let (_connection_id, remote_node_id, send, recv) = self
6189 .inner
6190 .open_peer_bi_transport_only(&id, timeout_ms.map(|value| value as u64))
6191 .await
6192 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6193 Ok(BiStream::from_parts(send, recv, remote_node_id))
6194 }
6195
6196 pub async fn open_uni(&self, endpoint_id: String) -> Result<PeerUniStream, JsValue> {
6197 let endpoint_id: iroh::EndpointId = endpoint_id
6198 .parse()
6199 .map_err(|e| JsValue::from_str(&format!("{}", e)))?;
6200 self.inner
6201 .assert_raw_peer_stream_allowed(&endpoint_id)
6202 .await
6203 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6204 self.inner
6207 .ensure_connection_manager_record_before_peer_stream(&endpoint_id)
6208 .await
6209 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6210 let node_guard = self.inner.iroh_node.read().await;
6211 if let Some(node) = node_guard.as_ref() {
6212 let send = node
6213 .open_uni(endpoint_id.clone())
6214 .await
6215 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6216 Ok(peer_uni_stream_from_send(
6217 crate::application_crypto_streams::PeerSendStream::plain(send),
6218 ))
6219 } else {
6220 Err(JsValue::from_str("Iroh node not initialized"))
6221 }
6222 }
6223
6224 pub async fn open_peer_uni(
6225 &self,
6226 id: String,
6227 timeout_ms: Option<u32>,
6228 ) -> Result<PeerUniStream, JsValue> {
6229 let (_connection_id, _remote_node_id, send) = self
6230 .inner
6231 .open_peer_uni(&id, timeout_ms.map(|value| value as u64))
6232 .await
6233 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6234 Ok(peer_uni_stream_from_send(send))
6235 }
6236
6237 #[wasm_bindgen(js_name = sendPeerDatagram)]
6238 pub async fn send_peer_datagram(
6239 &self,
6240 id: String,
6241 payload: Vec<u8>,
6242 max_age_ms: Option<u32>,
6243 ) -> Result<bool, JsValue> {
6244 match max_age_ms {
6245 Some(max_age_ms) => {
6246 self.inner
6247 .send_peer_datagram_with_max_age(&id, &payload, u64::from(max_age_ms))
6248 .await
6249 }
6250 None => self
6251 .inner
6252 .send_peer_datagram(&id, &payload)
6253 .await
6254 .map(|()| true),
6255 }
6256 .map_err(|error| JsValue::from_str(&error.to_string()))
6257 }
6258
6259 #[wasm_bindgen(js_name = receivePeerDatagram)]
6260 pub async fn receive_peer_datagram(&self, id: String) -> Result<Vec<u8>, JsValue> {
6261 self.inner
6262 .receive_peer_datagram(&id)
6263 .await
6264 .map_err(|error| JsValue::from_str(&error.to_string()))
6265 }
6266
6267 #[wasm_bindgen(js_name = peerDatagramMaxSize)]
6268 pub fn peer_datagram_max_size(&self) -> usize {
6269 Client::MAX_PEER_DATAGRAM_BYTES
6270 }
6271
6272 #[wasm_bindgen(js_name = createPeerDatagramPolicy)]
6273 pub fn create_peer_datagram_policy(&self) -> WasmPeerDatagramPolicy {
6274 WasmPeerDatagramPolicy {
6275 inner: RefCell::new(crate::datagrams::PeerDatagramPolicy::new(
6276 Client::MAX_PEER_DATAGRAM_BYTES,
6277 )),
6278 }
6279 }
6280
6281 pub async fn send_peer(&self, id: String, data: Vec<u8>) -> Result<(), JsValue> {
6291 self.inner
6292 .send_peer(&id, &data)
6293 .await
6294 .map_err(|e| JsValue::from_str(&e.to_string()))
6295 }
6296
6297 pub async fn iroh_path_kind(&self, peer_id: String) -> String {
6301 match self.inner.iroh_path_kind(&peer_id).await {
6302 crate::client::IrohPathKind::DirectQuic => "direct-quic".to_string(),
6303 crate::client::IrohPathKind::DirectLan => "direct-lan".to_string(),
6304 crate::client::IrohPathKind::Relay => "relay".to_string(),
6305 crate::client::IrohPathKind::Ble => "ble".to_string(),
6306 crate::client::IrohPathKind::WebRtc => "webrtc".to_string(),
6307 crate::client::IrohPathKind::Moq => "moq".to_string(),
6308 crate::client::IrohPathKind::Unknown => "unknown".to_string(),
6309 }
6310 }
6311
6312 pub async fn iroh_transport_rtt_ms(&self, peer_id: String) -> Option<u32> {
6315 self.inner
6316 .iroh_transport_rtt_ms(&peer_id)
6317 .await
6318 .map(|value| value.min(u32::MAX as u64) as u32)
6319 }
6320
6321 pub async fn incoming_streams(&self) -> Result<JsReadableStream, JsValue> {
6322 let (stream_owner, stream) = {
6323 let node_guard = self.inner.iroh_node.read().await;
6324 if let Some(node) = node_guard.as_ref() {
6325 (self.inner.clone(), node.incoming_streams_stream())
6326 } else {
6327 return Err(JsValue::from_str("Iroh node not initialized"));
6328 }
6329 };
6330
6331 use futures::StreamExt;
6332 let mapped_stream = stream.filter_map(move |incoming| {
6333 let stream_owner = stream_owner.clone();
6334 async move {
6335 let owned = stream_owner
6336 .incoming_stream_generation_is_owned(
6337 incoming.endpoint_id,
6338 incoming.transport_stable_id,
6339 )
6340 .await;
6341 let admission_candidate = !owned
6342 && matches!(
6343 &incoming.stream,
6344 crate::wasm_node::IncomingStreamType::Bi(_, _)
6345 )
6346 && stream_owner
6347 .incoming_stream_generation_is_authenticated_reconnect_candidate(
6348 incoming.endpoint_id,
6349 incoming.transport_stable_id,
6350 )
6351 .await;
6352 (owned || admission_candidate).then(|| {
6353 crate::wasm_node::BiStream::incoming_to_js_value(
6354 incoming,
6355 admission_candidate,
6356 )
6357 })
6358 }
6359 });
6360
6361 Ok(wasm_streams::ReadableStream::from_stream(mapped_stream).into_raw())
6362 }
6363
6364 #[wasm_bindgen(js_name = inspectIncomingStreamPrefix)]
6368 pub fn inspect_incoming_stream_prefix(
6369 &self,
6370 bytes: Vec<u8>,
6371 reached_eof: bool,
6372 ) -> Result<JsValue, JsValue> {
6373 serde::Serialize::serialize(
6374 &crate::stream_metadata::inspect_prefix_for_binding(&bytes, reached_eof),
6375 &serde_wasm_bindgen::Serializer::json_compatible(),
6376 )
6377 .map_err(|error| JsValue::from_str(&error.to_string()))
6378 }
6379
6380 #[wasm_bindgen(js_name = decodeIncomingChannelEnvelope)]
6384 pub fn decode_incoming_channel_envelope(&self, bytes: Vec<u8>) -> Result<JsValue, JsValue> {
6385 serde::Serialize::serialize(
6386 &crate::stream_metadata::decode_prefix_for_binding(&bytes),
6387 &serde_wasm_bindgen::Serializer::json_compatible(),
6388 )
6389 .map_err(|error| JsValue::from_str(&error.to_string()))
6390 }
6391
6392 pub async fn update_presence(
6393 &self,
6394 user_id: String,
6395 device_name: String,
6396 ticket: String,
6397 metadata: Option<String>,
6398 ttl_ms: Option<u64>,
6399 ) -> Result<(), JsValue> {
6400 self.inner
6401 .update_presence_with_ttl(
6402 &user_id,
6403 &device_name,
6404 &ticket,
6405 ttl_ms.unwrap_or(300_000),
6406 metadata.as_deref(),
6407 )
6408 .await
6409 .map_err(|e| JsValue::from_str(&e.to_string()))
6410 }
6411
6412 pub async fn send_message(
6413 &self,
6414 target_id: String,
6415 payload: String,
6416 state: Option<String>,
6417 reply_payload: Option<String>,
6418 ) -> Result<String, JsValue> {
6419 self.inner
6420 .send_message(
6421 &target_id,
6422 &payload,
6423 state.as_deref(),
6424 reply_payload.as_deref(),
6425 )
6426 .await
6427 .map_err(|e| JsValue::from_str(&e.to_string()))
6428 }
6429
6430 pub async fn set_offline(&self, user_id: String) -> Result<(), JsValue> {
6431 self.inner
6432 .set_offline(&user_id)
6433 .await
6434 .map_err(|e| JsValue::from_str(&e.to_string()))
6435 }
6436
6437 pub async fn update_device(
6438 &self,
6439 user_id: String,
6440 device_id: String,
6441 device_name: Option<String>,
6442 capabilities: Option<JsValue>,
6443 metadata: Option<String>,
6444 ) -> Result<(), JsValue> {
6445 let parsed_capabilities = match capabilities {
6446 Some(value) if !value.is_null() && !value.is_undefined() => Some(
6447 serde_wasm_bindgen::from_value::<crate::signaling::DeviceCapabilities>(value)
6448 .map_err(|e| JsValue::from_str(&e.to_string()))?,
6449 ),
6450 _ => None,
6451 };
6452
6453 self.inner
6454 .update_device(
6455 &user_id,
6456 &device_id,
6457 device_name.as_deref(),
6458 parsed_capabilities,
6459 metadata.as_deref(),
6460 )
6461 .await
6462 .map_err(|e| JsValue::from_str(&e.to_string()))
6463 }
6464
6465 pub async fn delete_device(
6466 &self,
6467 user_id: String,
6468 device_id: String,
6469 ) -> Result<(), JsValue> {
6470 self.inner
6471 .delete_device(&user_id, &device_id)
6472 .await
6473 .map_err(|e| JsValue::from_str(&e.to_string()))
6474 }
6475
6476 pub fn force_reconnect_snapshot(&self) {
6477 self.inner.clone().force_reconnect_snapshot();
6478 }
6479
6480 pub fn stop_presence_loop(&self) {
6481 self.inner.stop_presence_loop();
6482 }
6483
6484 pub fn stop_auto_connect(&self) {
6485 self.inner.stop_auto_connect();
6486 self.inner.stop_browser_auto_connect();
6487 }
6488
6489 pub fn start_auto_connect(
6490 &self,
6491 user_id: String,
6492 local_device_id: String,
6493 ) -> Result<(), JsValue> {
6494 self.inner
6495 .start_browser_auto_connect(user_id, local_device_id)
6496 .map_err(|error| JsValue::from_str(&error.to_string()))
6497 }
6498
6499 pub async fn request_transfer_peer(&self, request_json: String) -> Result<String, JsValue> {
6500 if request_json.len() > 2048 { return Err(JsValue::from_str("transfer peer demand exceeds its input bound")); }
6501 let request = serde_json::from_str(&request_json).map_err(|error| JsValue::from_str(&error.to_string()))?;
6502 let status = self.inner.request_transfer_peer(request).await.map_err(|error| JsValue::from_str(&error.to_string()))?;
6503 serde_json::to_string(&status).map_err(|error| JsValue::from_str(&error.to_string()))
6504 }
6505
6506 pub async fn transfer_peer_demand_status(&self, id: String, generation: u64) -> Result<String, JsValue> {
6507 let status = self.inner.transfer_peer_demand_status(&id, generation).await.map_err(|error| JsValue::from_str(&error.to_string()))?;
6508 serde_json::to_string(&status).map_err(|error| JsValue::from_str(&error.to_string()))
6509 }
6510
6511 pub async fn release_transfer_peer(&self, id: String, generation: u64) -> Result<bool, JsValue> {
6512 self.inner.release_transfer_peer(&id, generation).await.map_err(|error| JsValue::from_str(&error.to_string()))
6513 }
6514
6515 pub async fn set_transfer_peer_budget(&self, budget: u32, generation: u64) -> Result<(), JsValue> {
6516 self.inner.set_transfer_peer_budget(budget as usize, generation).await.map_err(|error| JsValue::from_str(&error.to_string()))
6517 }
6518
6519 pub fn submit_browser_peer_intents(
6522 &self, revision: u32, peers_json: String,
6523 ) -> Result<bool, JsValue> {
6524 self.inner.submit_browser_peer_intents(u64::from(revision), &peers_json)
6525 .map_err(|error| JsValue::from_str(&error.to_string()))
6526 }
6527
6528 #[deprecated(note = "use submit_browser_desired_peers_with_sparse_fanout for sparse rooms")]
6530 pub fn submit_browser_desired_peers(
6531 &self,
6532 revision: u32,
6533 peers_json: String,
6534 ) -> Result<bool, JsValue> {
6535 self.inner
6536 .submit_browser_desired_peers(u64::from(revision), &peers_json)
6537 .map_err(|error| JsValue::from_str(&error.to_string()))
6538 }
6539
6540 pub async fn submit_browser_desired_peers_with_sparse_fanout(
6541 &self,
6542 revision: u32,
6543 peers_json: String,
6544 capability: Option<String>,
6545 local_device_id: Option<String>,
6546 local_device_key_x: Option<String>,
6547 sparse: Option<bool>,
6548 ) -> Result<bool, JsValue> {
6549 let peers_json = if sparse.unwrap_or(false) {
6550 let capability = capability
6551 .as_deref()
6552 .ok_or_else(|| JsValue::from_str("sparse fanout requires a capability"))?;
6553 let local_device_id = local_device_id
6554 .as_deref()
6555 .ok_or_else(|| JsValue::from_str("sparse fanout requires a local device id"))?;
6556 let local_device_key_x = local_device_key_x.as_deref().ok_or_else(|| {
6557 JsValue::from_str("sparse fanout requires a local device key")
6558 })?;
6559 self.inner
6560 .configure_sparse_fanout(
6561 capability,
6562 u64::from(revision),
6563 local_device_id,
6564 local_device_key_x,
6565 &peers_json,
6566 true,
6567 )
6568 .await
6569 .map_err(|error| JsValue::from_str(&error.to_string()))?
6570 .0
6571 } else {
6572 peers_json
6573 };
6574 self.inner
6575 .submit_browser_desired_peers(u64::from(revision), &peers_json)
6576 .map_err(|error| JsValue::from_str(&error.to_string()))
6577 }
6578
6579 #[wasm_bindgen(js_name = prepareSparseFanoutMessage)]
6580 pub async fn prepare_sparse_fanout_message(
6581 &self,
6582 capability: String,
6583 payload: Vec<u8>,
6584 ) -> Result<JsValue, JsValue> {
6585 let request = self
6586 .inner
6587 .prepare_sparse_fanout_message(&capability, &payload)
6588 .await
6589 .map_err(|error| JsValue::from_str(&error.to_string()))?;
6590 serde_wasm_bindgen::to_value(&request)
6591 .map_err(|error| JsValue::from_str(&error.to_string()))
6592 }
6593
6594 #[wasm_bindgen(js_name = finalizeSparseFanoutMessage)]
6595 pub async fn finalize_sparse_fanout_message(
6596 &self,
6597 request_id: String,
6598 signature: String,
6599 ) -> Result<Vec<u8>, JsValue> {
6600 self.inner
6601 .finalize_sparse_fanout_message(&request_id, &signature)
6602 .await
6603 .map_err(|error| JsValue::from_str(&error.to_string()))
6604 }
6605
6606 #[wasm_bindgen(js_name = acceptSparseFanoutMessage)]
6607 pub async fn accept_sparse_fanout_message(
6608 &self,
6609 capability: String,
6610 source_peer_id: String,
6611 encoded: Vec<u8>,
6612 ) -> Result<JsValue, JsValue> {
6613 let decision = self
6614 .inner
6615 .accept_sparse_fanout_message(&capability, &source_peer_id, &encoded)
6616 .await
6617 .map_err(|error| JsValue::from_str(&error.to_string()))?;
6618 serde_wasm_bindgen::to_value(&decision)
6619 .map_err(|error| JsValue::from_str(&error.to_string()))
6620 }
6621
6622 #[wasm_bindgen(js_name = sparseFanoutDiagnostics)]
6623 pub async fn sparse_fanout_diagnostics(
6624 &self,
6625 capability: String,
6626 ) -> Result<JsValue, JsValue> {
6627 serde_wasm_bindgen::to_value(&self.inner.sparse_fanout_diagnostics(&capability).await)
6628 .map_err(|error| JsValue::from_str(&error.to_string()))
6629 }
6630
6631 #[wasm_bindgen(js_name = recordSparseFanoutForwardQueueDrop)]
6632 pub async fn record_sparse_fanout_forward_queue_drop(
6633 &self,
6634 capability: String,
6635 count: u32,
6636 ) -> Result<(), JsValue> {
6637 self.inner
6638 .record_sparse_fanout_forward_queue_drop(&capability, u64::from(count))
6639 .await
6640 .map_err(|error| JsValue::from_str(&error.to_string()))
6641 }
6642
6643 pub fn wake_browser_auto_connect(&self) -> bool {
6644 self.inner.wake_browser_auto_connect()
6645 }
6646
6647 pub async fn set_auto_connect_excluded(&self, device_id: String, excluded: bool) {
6648 if excluded {
6649 self.inner.exclude_peer_and_publish(&device_id).await;
6650 } else {
6651 self.inner.unexclude_peer_and_publish(&device_id).await;
6652 }
6653 self.inner.wake_browser_auto_connect();
6654 }
6655
6656 pub fn is_auto_connect_excluded(&self, device_id: String) -> bool {
6657 self.inner.is_auto_connect_excluded(&device_id)
6658 }
6659
6660 pub async fn disconnect_device(
6661 &self,
6662 device_id: String,
6663 node_id_hint: Option<String>,
6664 ) -> Result<JsValue, JsValue> {
6665 let retired = self
6666 .inner
6667 .disconnect_device(&device_id, node_id_hint.as_deref())
6668 .await;
6669 serde_wasm_bindgen::to_value(&retired).map_err(|e| JsValue::from_str(&e.to_string()))
6670 }
6671
6672 pub fn stop_auth_scoped_activity(&self) {
6673 self.inner.stop_auth_scoped_activity();
6674 self.inner.stop_browser_auto_connect();
6675 }
6676
6677 pub fn start_presence_loop(
6678 &self,
6679 user_id: String,
6680 device_name: String,
6681 ticket: String,
6682 metadata: Option<String>,
6683 ) {
6684 self.inner
6685 .clone()
6686 .start_signaling_loop(user_id, device_name, ticket, metadata);
6687 }
6688
6689 pub async fn search_devices(&self, user_id: String) -> Result<JsValue, JsValue> {
6690 let devices = self
6691 .inner
6692 .search_devices(&user_id)
6693 .await
6694 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6695 serde_wasm_bindgen::to_value(&devices).map_err(|e| JsValue::from_str(&e.to_string()))
6696 }
6697
6698 pub async fn devices_with_status(&self, user_id: String) -> Result<JsValue, JsValue> {
6699 let devices = self
6700 .inner
6701 .devices_with_status(&user_id)
6702 .await
6703 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6704 serde_wasm_bindgen::to_value(&devices).map_err(|e| JsValue::from_str(&e.to_string()))
6705 }
6706
6707 pub async fn connect_device(
6708 &self,
6709 device_id: Option<String>,
6710 endpoint_ticket: String,
6711 timeout_ms: Option<u32>,
6712 ) -> Result<JsValue, JsValue> {
6713 let dial = self
6714 .inner
6715 .connect_device(device_id.as_deref(), &endpoint_ticket);
6716 let result = if let Some(timeout_ms) = timeout_ms {
6717 let deadline = gloo_timers::future::sleep(std::time::Duration::from_millis(
6718 u64::from(timeout_ms.clamp(1, 120_000)),
6719 ));
6720 futures::pin_mut!(dial, deadline);
6721 match futures::future::select(dial, deadline).await {
6722 futures::future::Either::Left((result, _)) => result,
6723 futures::future::Either::Right(_) => {
6724 return Err(JsValue::from_str("endpoint ticket dial timed out"));
6725 }
6726 }
6727 } else {
6728 dial.await
6729 }
6730 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6731 serde_wasm_bindgen::to_value(&result).map_err(|e| JsValue::from_str(&e.to_string()))
6732 }
6733
6734 pub async fn connect_device_for_route_adoption(
6736 &self,
6737 device_id: Option<String>,
6738 endpoint_ticket: String,
6739 ) -> Result<JsValue, JsValue> {
6740 let result = self
6741 .inner
6742 .connect_device_for_route_adoption(device_id.as_deref(), &endpoint_ticket)
6743 .await
6744 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6745 serde_wasm_bindgen::to_value(&result).map_err(|e| JsValue::from_str(&e.to_string()))
6746 }
6747
6748 pub async fn connect_known_device_with_token(
6749 &self,
6750 device_id: String,
6751 token: String,
6752 scope: String,
6753 max_connections: u32,
6754 expires_at_ms: Option<u64>,
6755 lookup_timeout_ms: Option<u32>,
6756 ) -> Result<JsValue, JsValue> {
6757 let result = self
6758 .inner
6759 .connect_known_device_with_token(
6760 &device_id,
6761 &token,
6762 &scope,
6763 max_connections,
6764 expires_at_ms,
6765 lookup_timeout_ms.map(u64::from),
6766 )
6767 .await
6768 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6769 serde_wasm_bindgen::to_value(&result).map_err(|e| JsValue::from_str(&e.to_string()))
6770 }
6771
6772 pub async fn observe_known_device_endpoint(
6773 &self,
6774 device_id: String,
6775 endpoint_ticket: String,
6776 ) -> Result<(), JsValue> {
6777 self.inner
6778 .observe_known_device_endpoint(&device_id, &endpoint_ticket)
6779 .await
6780 .map_err(|e| JsValue::from_str(&e.to_string()))
6781 }
6782
6783 pub async fn create_session(&self, session_json: String) -> Result<(), JsValue> {
6784 let session: crate::signaling::SignalingSession =
6785 serde_json::from_str(&session_json)
6786 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6787 self.inner
6788 .create_session(session)
6789 .await
6790 .map_err(|e| JsValue::from_str(&e.to_string()))
6791 }
6792
6793 pub async fn update_session(
6794 &self,
6795 session_id: String,
6796 update_json: String,
6797 ) -> Result<(), JsValue> {
6798 let update_data: serde_json::Value = serde_json::from_str(&update_json)
6799 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6800 self.inner
6801 .update_session(&session_id, update_data)
6802 .await
6803 .map_err(|e| JsValue::from_str(&e.to_string()))
6804 }
6805
6806 pub async fn create_room(
6809 &self,
6810 room_id: String,
6811 user_id: String,
6812 ticket_str: String,
6813 my_node_id: String,
6814 tag: String,
6815 max_members: Option<u32>,
6816 ) -> Result<bool, JsValue> {
6817 self.inner
6818 .room
6819 .create_room(
6820 &room_id,
6821 &user_id,
6822 &ticket_str,
6823 &my_node_id,
6824 &tag,
6825 max_members,
6826 )
6827 .await
6828 .map_err(|e| JsValue::from_str(&e.to_string()))
6829 }
6830
6831 pub async fn join_room(
6832 &self,
6833 room_id: String,
6834 user_id: String,
6835 ticket_str: String,
6836 my_node_id: String,
6837 tag: String,
6838 ) -> Result<(), JsValue> {
6839 self.inner
6840 .room
6841 .join_room(&room_id, &user_id, &ticket_str, &my_node_id, &tag)
6842 .await
6843 .map_err(|e| JsValue::from_str(&e.to_string()))
6844 }
6845
6846 pub async fn get_members(
6847 &self,
6848 room_id: String,
6849 my_node_id: String,
6850 tag: String,
6851 ) -> Result<String, JsValue> {
6852 let members = self
6853 .inner
6854 .room
6855 .get_members(&room_id, &my_node_id, &tag)
6856 .await
6857 .map_err(|e| JsValue::from_str(&e.to_string()))?;
6858 serde_json::to_string(&members).map_err(|e| JsValue::from_str(&e.to_string()))
6859 }
6860
6861 pub async fn leave_room(
6862 &self,
6863 room_id: String,
6864 my_node_id: String,
6865 tag: String,
6866 ) -> Result<(), JsValue> {
6867 self.inner
6868 .room
6869 .leave_room(&room_id, &my_node_id, &tag)
6870 .await
6871 .map_err(|e| JsValue::from_str(&e.to_string()))
6872 }
6873 }
6874
6875 #[cfg(feature = "iroh-protocols-wasm")]
6876 #[wasm_bindgen]
6877 impl WasmClient {
6878 #[wasm_bindgen(js_name = __initPersistentIrohProtocols)]
6882 pub async fn init_persistent_iroh_protocols(
6883 &self,
6884 replica_store: JsValue,
6885 ) -> Result<(), JsValue> {
6886 let mut guard = self.persistent_protocols.lock().await;
6887 if guard.is_some() {
6888 return Ok(());
6889 }
6890 let node = self
6891 .inner
6892 .iroh_node
6893 .read()
6894 .await
6895 .as_ref()
6896 .cloned()
6897 .ok_or_else(|| {
6898 JsValue::from_str("OpenRTC endpoint must be initialized before protocols")
6899 })?;
6900 let store = crate::wasm_docs_persistence::JsReplicaStore::new(replica_store)
6901 .map_err(|error| JsValue::from_str(&error.to_string()))?;
6902 let actor = crate::wasm_docs_persistence::WasmPersistentDocsActor::hydrate(
6903 store,
6904 node.endpoint().clone(),
6905 )
6906 .await
6907 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
6908 node.install_standard_protocols(
6909 actor.docs_protocol(),
6910 actor.blobs_protocol(),
6911 actor.gossip_protocol(),
6912 )
6913 .await
6914 .map_err(|error| JsValue::from_str(&format!("{error:#}")))?;
6915 *guard = Some(actor);
6916 Ok(())
6917 }
6918
6919 #[wasm_bindgen(js_name = __reconcilePersistentIrohProtocols)]
6923 pub async fn reconcile_persistent_iroh_protocols(&self) -> Result<u32, JsValue> {
6924 let guard = self.persistent_protocols.lock().await;
6925 let Some(actor) = guard.as_ref() else {
6926 return Ok(0);
6927 };
6928 actor.reconcile_sync().await.map_err(js_protocol_error)
6929 }
6930
6931 #[wasm_bindgen(js_name = __importPersistentIrohAuthor)]
6932 pub async fn import_persistent_iroh_author(
6933 &self,
6934 author_secret: Vec<u8>,
6935 ) -> Result<String, JsValue> {
6936 let guard = self.persistent_protocols.lock().await;
6937 let actor = persistent_protocols(&guard)?;
6938 actor
6939 .import_author(author_secret)
6940 .await
6941 .map_err(js_protocol_error)
6942 }
6943
6944 #[wasm_bindgen(js_name = __importPersistentIrohNamespace)]
6945 pub async fn import_persistent_iroh_namespace(
6946 &self,
6947 capability_kind: String,
6948 capability: Vec<u8>,
6949 generation: u64,
6950 share_revision: u64,
6951 ) -> Result<String, JsValue> {
6952 let mut guard = self.persistent_protocols.lock().await;
6953 let actor = persistent_protocols_mut(&mut guard)?;
6954 actor
6955 .import_namespace(
6956 capability_kind.as_str(),
6957 capability,
6958 generation,
6959 share_revision,
6960 )
6961 .await
6962 .map_err(js_protocol_error)
6963 }
6964
6965 #[wasm_bindgen(js_name = __createPersistentIrohNamespace)]
6966 pub async fn create_persistent_iroh_namespace(
6967 &self,
6968 generation: u64,
6969 share_revision: u64,
6970 ) -> Result<JsValue, JsValue> {
6971 let mut guard = self.persistent_protocols.lock().await;
6972 let descriptor = persistent_protocols_mut(&mut guard)?
6973 .create_namespace(generation, share_revision)
6974 .await
6975 .map_err(js_protocol_error)?;
6976 serde_wasm_bindgen::to_value(&descriptor)
6977 .map_err(|error| JsValue::from_str(&error.to_string()))
6978 }
6979
6980 #[wasm_bindgen(js_name = __importPersistentIrohTicket)]
6981 pub async fn import_persistent_iroh_ticket(
6982 &self,
6983 ticket: String,
6984 generation: u64,
6985 share_revision: u64,
6986 ) -> Result<String, JsValue> {
6987 let mut guard = self.persistent_protocols.lock().await;
6988 persistent_protocols_mut(&mut guard)?
6989 .import_ticket(&ticket, generation, share_revision)
6990 .await
6991 .map_err(js_protocol_error)
6992 }
6993
6994 #[wasm_bindgen(js_name = __sharePersistentIrohNamespace)]
6995 pub async fn share_persistent_iroh_namespace(
6996 &self,
6997 namespace_id: String,
6998 writable: bool,
6999 ) -> Result<String, JsValue> {
7000 let guard = self.persistent_protocols.lock().await;
7001 persistent_protocols(&guard)?
7002 .share(&namespace_id, writable)
7003 .await
7004 .map_err(js_protocol_error)
7005 }
7006
7007 #[wasm_bindgen(js_name = __removePersistentIrohNamespace)]
7008 pub async fn remove_persistent_iroh_namespace(
7009 &self,
7010 namespace_id: String,
7011 generation: u64,
7012 share_revision: u64,
7013 ) -> Result<(), JsValue> {
7014 let mut guard = self.persistent_protocols.lock().await;
7015 persistent_protocols_mut(&mut guard)?
7016 .remove_namespace(&namespace_id, generation, share_revision)
7017 .await
7018 .map_err(js_protocol_error)
7019 }
7020
7021 #[wasm_bindgen(js_name = __putPersistentIrohBytes)]
7022 pub async fn put_persistent_iroh_bytes(
7023 &self,
7024 namespace_id: String,
7025 key: Vec<u8>,
7026 value: Vec<u8>,
7027 ) -> Result<JsValue, JsValue> {
7028 let guard = self.persistent_protocols.lock().await;
7029 let receipt = persistent_protocols(&guard)?
7030 .set_bytes(&namespace_id, key, value)
7031 .await
7032 .map_err(js_protocol_error)?;
7033 serde_wasm_bindgen::to_value(&receipt)
7034 .map_err(|error| JsValue::from_str(&error.to_string()))
7035 }
7036
7037 #[wasm_bindgen(js_name = __setPersistentIrohHash)]
7038 pub async fn set_persistent_iroh_hash(
7039 &self,
7040 namespace_id: String,
7041 key: Vec<u8>,
7042 content_hash: String,
7043 content_length: u64,
7044 ) -> Result<String, JsValue> {
7045 let guard = self.persistent_protocols.lock().await;
7046 persistent_protocols(&guard)?
7047 .set_hash(&namespace_id, key, &content_hash, content_length)
7048 .await
7049 .map_err(js_protocol_error)
7050 }
7051
7052 #[wasm_bindgen(js_name = __deletePersistentIrohPrefix)]
7053 pub async fn delete_persistent_iroh_prefix(
7054 &self,
7055 namespace_id: String,
7056 prefix: Vec<u8>,
7057 ) -> Result<JsValue, JsValue> {
7058 let guard = self.persistent_protocols.lock().await;
7059 let receipt = persistent_protocols(&guard)?
7060 .delete_prefix(&namespace_id, prefix)
7061 .await
7062 .map_err(js_protocol_error)?;
7063 serde_wasm_bindgen::to_value(&receipt)
7064 .map_err(|error| JsValue::from_str(&error.to_string()))
7065 }
7066
7067 #[wasm_bindgen(js_name = __queryPersistentIrohNamespace)]
7068 pub async fn query_persistent_iroh_namespace(
7069 &self,
7070 namespace_id: String,
7071 key_prefix: Vec<u8>,
7072 ) -> Result<JsValue, JsValue> {
7073 let guard = self.persistent_protocols.lock().await;
7074 let entries = persistent_protocols(&guard)?
7075 .query(&namespace_id, key_prefix)
7076 .await
7077 .map_err(js_protocol_error)?;
7078 serde_wasm_bindgen::to_value(&entries)
7079 .map_err(|error| JsValue::from_str(&error.to_string()))
7080 }
7081
7082 #[wasm_bindgen(js_name = __hydratePersistentIrohBlob)]
7083 pub async fn hydrate_persistent_iroh_blob(
7084 &self,
7085 content_hash: String,
7086 ) -> Result<(), JsValue> {
7087 let guard = self.persistent_protocols.lock().await;
7088 persistent_protocols(&guard)?
7089 .hydrate_blob(&content_hash)
7090 .await
7091 .map_err(js_protocol_error)
7092 }
7093
7094 #[wasm_bindgen(js_name = __acknowledgePersistentIrohOutbox)]
7095 pub async fn acknowledge_persistent_iroh_outbox(
7096 &self,
7097 operation_id: String,
7098 ) -> Result<(), JsValue> {
7099 let guard = self.persistent_protocols.lock().await;
7100 persistent_protocols(&guard)?
7101 .acknowledge_outbox(&operation_id)
7102 .await
7103 .map_err(js_protocol_error)
7104 }
7105
7106 #[wasm_bindgen(js_name = __flushPersistentIrohProtocols)]
7107 pub async fn flush_persistent_iroh_protocols(&self) -> Result<(), JsValue> {
7108 let guard = self.persistent_protocols.lock().await;
7109 persistent_protocols(&guard)?
7110 .flush()
7111 .await
7112 .map_err(js_protocol_error)
7113 }
7114
7115 #[wasm_bindgen(js_name = __shutdownPersistentIrohProtocols)]
7116 pub async fn shutdown_persistent_iroh_protocols(&self) -> Result<(), JsValue> {
7117 if let Some(actor) = self.persistent_protocols.lock().await.take() {
7118 actor.shutdown().await.map_err(js_protocol_error)?;
7119 }
7120 Ok(())
7121 }
7122 }
7123
7124 #[cfg(feature = "iroh-protocols-wasm")]
7125 fn persistent_protocols(
7126 guard: &Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>,
7127 ) -> Result<&crate::wasm_docs_persistence::WasmPersistentDocsActor, JsValue> {
7128 guard
7129 .as_ref()
7130 .ok_or_else(|| JsValue::from_str("persistent Iroh protocols are not initialized"))
7131 }
7132
7133 #[cfg(feature = "iroh-protocols-wasm")]
7134 fn persistent_protocols_mut(
7135 guard: &mut Option<crate::wasm_docs_persistence::WasmPersistentDocsActor>,
7136 ) -> Result<&mut crate::wasm_docs_persistence::WasmPersistentDocsActor, JsValue> {
7137 guard
7138 .as_mut()
7139 .ok_or_else(|| JsValue::from_str("persistent Iroh protocols are not initialized"))
7140 }
7141
7142 #[cfg(feature = "iroh-protocols-wasm")]
7143 fn js_protocol_error(error: impl std::fmt::Display) -> JsValue {
7144 JsValue::from_str(&format!("{error:#}"))
7145 }
7146
7147 #[wasm_bindgen(start)]
7148 pub fn start() {
7149 console_error_panic_hook::set_once();
7150 }
7151}