openrtc 1.0.4

#![cfg(target_arch = "wasm32")]

use anyhow::Result;
use async_channel::Sender;
use futures::{future::Either, io::AsyncReadExt, FutureExt, SinkExt, Stream, StreamExt};
use iroh::{
    endpoint::{Connection, RecvStream, SendStream},
    protocol::{AcceptError, ProtocolHandler, Router},
    Endpoint, EndpointAddr, EndpointId, Watcher as _,
};
use n0_future::{boxed::BoxStream, task};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio::sync::{Mutex, RwLock};
use tokio_stream::wrappers::BroadcastStream;
use tokio_util::codec::{BytesCodec, FramedWrite};
use tokio_util::compat::TokioAsyncReadCompatExt;
use wasm_bindgen::{prelude::wasm_bindgen, JsError, JsValue};
use wasm_streams::{
    readable::sys::ReadableStream as JsReadableStream,
    writable::sys::WritableStream as JsWritableStream, ReadableStream, WritableStream,
};

use crate::heartbeat::{
    classify_incoming_uni, respond_to_ping, IncomingUniClassification, IrohConnectionProbe,
    IrohProbeRegistry, PrefixedRecvStream,
};
use crate::iroh_connection_policy::{
    decide_inbound_install, decide_outbound_install, should_start_outbound_dial,
    should_wait_for_canonical_inbound, ExistingConnectionState, IrohConnectionDirection,
    IrohConnectionInstallDecision, NONCANONICAL_OUTBOUND_DIAL_GRACE_MS,
};

const READABLE_STREAM_CHUNK_BYTES: usize = 256 * 1024;

// Define stream type
#[derive(Debug)]
pub enum IncomingStreamType {
    Bi(SendStream, RecvStream),
    Uni(PrefixedRecvStream),
}

pub struct IncomingStream {
    pub endpoint_id: EndpointId,
    /// Physical Iroh generation that accepted this stream. It must remain
    /// attached through export so queued ingress from a retired accept loop can
    /// be rejected before TypeScript admission or application delivery.
    pub transport_stable_id: u64,
    pub stream: IncomingStreamType,
}

/// Per-incoming-stream QUIC accept logging is extremely noisy (heartbeats and app
/// traffic each open uni streams). Enable only when debugging stream routing:
/// `globalThis.__OPENRTC_DEBUG_WASM_STREAMS__ = true` in the browser console.
fn wasm_debug_streams() -> bool {
    let global = js_sys::global();
    js_sys::Reflect::get(
        &global,
        &JsValue::from_str("__OPENRTC_DEBUG_WASM_STREAMS__"),
    )
    .ok()
    .map(|v| v == JsValue::TRUE)
    .unwrap_or(false)
}

fn is_manual_disconnect_close_reason(error: Option<&str>) -> bool {
    crate::lifecycle_reason::reason_is_manual_disconnect(error)
}

/// Wall-clock ms between low-rate WASM console summaries of accepted QUIC streams.
const STREAM_ACCEPT_SUMMARY_INTERVAL_MS: u64 = 30_000;

static WASM_ACCEPT_STREAM_UNI: AtomicU64 = AtomicU64::new(0);
static WASM_ACCEPT_STREAM_BI: AtomicU64 = AtomicU64::new(0);
static WASM_ACCEPT_STREAM_LAST_LOG_MS: AtomicU64 = AtomicU64::new(0);

fn record_stream_accepted(is_uni: bool) {
    if is_uni {
        WASM_ACCEPT_STREAM_UNI.fetch_add(1, Ordering::Relaxed);
    } else {
        WASM_ACCEPT_STREAM_BI.fetch_add(1, Ordering::Relaxed);
    }
    maybe_log_stream_accept_summary();
}

fn maybe_log_stream_accept_summary() {
    let now_ms = js_sys::Date::now() as u64;
    let prev_last = WASM_ACCEPT_STREAM_LAST_LOG_MS.load(Ordering::Relaxed);
    if now_ms.saturating_sub(prev_last) < STREAM_ACCEPT_SUMMARY_INTERVAL_MS {
        return;
    }
    let claimed = WASM_ACCEPT_STREAM_LAST_LOG_MS
        .fetch_update(Ordering::SeqCst, Ordering::Relaxed, |last| {
            if now_ms.saturating_sub(last) < STREAM_ACCEPT_SUMMARY_INTERVAL_MS {
                None
            } else {
                Some(now_ms)
            }
        })
        .is_ok();
    if !claimed {
        return;
    }
    let uni = WASM_ACCEPT_STREAM_UNI.swap(0, Ordering::Relaxed);
    let bi = WASM_ACCEPT_STREAM_BI.swap(0, Ordering::Relaxed);
    if uni == 0 && bi == 0 {
        return;
    }
    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
        "[pluto-rtc][wasm-node] accepted QUIC streams (since last summary, min {}s apart): uni={} bi={}",
        STREAM_ACCEPT_SUMMARY_INTERVAL_MS / 1000,
        uni,
        bi
    )));
}

fn local_prefers_outbound(local_endpoint_id: EndpointId, remote_endpoint_id: EndpointId) -> bool {
    local_endpoint_id.to_string() > remote_endpoint_id.to_string()
}

async fn outbound_dial_precheck(
    endpoint: &Endpoint,
    endpoint_id: EndpointId,
    connections: &Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: &Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
) -> bool {
    let prefers_outbound = local_prefers_outbound(endpoint.id(), endpoint_id);
    let existing = {
        let conns = connections.read().await;
        if let Some(existing) = conns.get(&endpoint_id) {
            let direction = connection_insert_at_ms
                .read()
                .await
                .get(&endpoint_id)
                .map(|metadata| metadata.direction)
                .unwrap_or(IrohConnectionDirection::Outbound);
            Some((existing.close_reason().is_none(), direction))
        } else {
            None
        }
    };

    if !should_start_outbound_dial(existing, prefers_outbound) {
        return true;
    }
    if !should_wait_for_canonical_inbound(existing, prefers_outbound) {
        return false;
    }

    gloo_timers::future::sleep(std::time::Duration::from_millis(
        NONCANONICAL_OUTBOUND_DIAL_GRACE_MS,
    ))
    .await;
    connections
        .read()
        .await
        .get(&endpoint_id)
        .is_some_and(|connection| connection.close_reason().is_none())
}

#[derive(Debug, Clone, Copy)]
struct IrohConnectionMetadata {
    inserted_at_ms: f64,
    direction: IrohConnectionDirection,
}

#[derive(Debug, Clone)]
pub struct IrohWasmNode {
    router: Router,
    accept_events: broadcast::Sender<AcceptEvent>,
    connections: Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
    connect_addr_waiters: Arc<RwLock<HashMap<EndpointId, Vec<Sender<ConnectEvent>>>>>,
    incoming_streams: async_channel::Sender<IncomingStream>,
    incoming_streams_receiver: async_channel::Receiver<IncomingStream>,
    probe_registry: IrohProbeRegistry,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum ConnectEvent {
    Connected,
    Closed { error: Option<String> },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum AcceptEvent {
    Accepted {
        endpoint_id: EndpointId,
        transport_stable_id: u64,
    },
    Closed {
        endpoint_id: EndpointId,
        transport_stable_id: u64,
        error: Option<String>,
        /// True when the local side initiated the close (e.g. `disconnect()`
        /// after auth rejection). The replacement-wait polling loop must NOT
        /// run for locally-closed connections — there is no replacement coming
        /// and polling burns the WASM event loop indefinitely.
        was_locally_closed: bool,
    },
}

#[derive(Debug, Clone)]
pub struct PlutoniumProtocol {
    event_sender: broadcast::Sender<AcceptEvent>,
    stream_sender: async_channel::Sender<IncomingStream>,
    connections: Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
    local_endpoint_id: EndpointId,
    probe_registry: IrohProbeRegistry,
}

struct WasmConnectionLoopOutcome {
    close_reason: Option<String>,
    was_locally_closed: bool,
}

async fn run_connection_loop(
    source: &'static str,
    connection: &Connection,
    stream_sender: async_channel::Sender<IncomingStream>,
    probe_registry: IrohProbeRegistry,
) -> WasmConnectionLoopOutcome {
    let endpoint_id = connection.remote_id();
    let stable_id = connection.stable_id();

    loop {
        tokio::select! {
            biased;
            res = connection.accept_bi() => {
                match res {
                    Ok((send, recv)) => {
                        record_stream_accepted(false);
                        if wasm_debug_streams() {
                            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                                "[pluto-rtc][wasm-node][{}] accept_bi got stream endpoint_id={}",
                                source,
                                endpoint_id
                            )));
                        }
                        if let Err(e) = stream_sender.send(IncomingStream {
                            endpoint_id,
                            transport_stable_id: stable_id as u64,
                            stream: IncomingStreamType::Bi(send, recv),
                        }).await {
                            web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                                "[pluto-rtc][wasm-node][{}] FAILED to forward bi stream endpoint_id={} error={}",
                                source,
                                endpoint_id,
                                e
                            )));
                        }
                    }
                    Err(e) => {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][wasm-node][{}] accept_bi error endpoint_id={} error={:?}",
                            source,
                            endpoint_id,
                            e
                        )));
                        break;
                    }
                }
            }
            res = connection.accept_uni() => {
                match res {
                    Ok(recv) => {
                        record_stream_accepted(true);
                        if wasm_debug_streams() {
                            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                                "[pluto-rtc][wasm-node][{}] accept_uni got stream endpoint_id={}",
                                source,
                                endpoint_id
                            )));
                        }
                        let connection = connection.clone();
                        let probe_registry = probe_registry.clone();
                        let application_streams = stream_sender.clone();
                        task::spawn(async move {
                            match classify_incoming_uni(recv).await {
                                IncomingUniClassification::Control { type_id, payload } => {
                            use crate::heartbeat::codec;
                            if type_id == codec::TYPE_PING {
                                if !respond_to_ping(&connection, &payload).await {
                                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                                        "[OpenRTC][iroh-probe] failed to send pong endpoint_id={} transport_stable_id={}",
                                        endpoint_id,
                                        stable_id,
                                    )));
                                }
                            } else if type_id == codec::TYPE_PONG {
                                if let Some(pong) = codec::decode_pong(&payload) {
                                    probe_registry
                                        .deliver_pong(
                                            &endpoint_id.to_string(),
                                            stable_id as u64,
                                            &pong,
                                        )
                                        .await;
                                }
                            } else if type_id == codec::TYPE_MANUAL_DISCONNECT {
                                connection.close(
                                    0u8.into(),
                                    crate::lifecycle_reason::REASON_MANUAL_DISCONNECT.as_bytes(),
                                );
                            }
                                }
                                IncomingUniClassification::Application(recv) => {
                                    let _ = application_streams
                                        .send(IncomingStream {
                                            endpoint_id,
                                            transport_stable_id: stable_id as u64,
                                            stream: IncomingStreamType::Uni(recv),
                                        })
                                        .await;
                                }
                                IncomingUniClassification::MalformedControl => {
                                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                                        "[OpenRTC][iroh-probe] malformed control stream endpoint_id={} transport_stable_id={}",
                                        endpoint_id,
                                        stable_id,
                                    )));
                                }
                            }
                        });
                    }
                    Err(e) => {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][wasm-node][{}] accept_uni error endpoint_id={} error={:?}",
                            source,
                            endpoint_id,
                            e
                        )));
                        break;
                    }
                }
            }
            _ = connection.closed() => {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][{}] connection.closed() fired endpoint_id={}",
                    source,
                    endpoint_id
                )));
                break;
            }
        }
    }

    let close_reason_native = connection.close_reason();
    let close_reason = close_reason_native
        .as_ref()
        .map(|reason| format!("{:?}", reason));
    let was_locally_closed = matches!(
        close_reason_native,
        Some(iroh::endpoint::ConnectionError::LocallyClosed)
    );
    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
        "[pluto-rtc][wasm-node][{}] connection loop exited endpoint_id={} stable_id={} close_reason={:?} was_locally_closed={}",
        source,
        endpoint_id,
        stable_id,
        close_reason_native,
        was_locally_closed
    )));

    WasmConnectionLoopOutcome {
        close_reason,
        was_locally_closed,
    }
}

async fn remove_connection_if_current(
    endpoint_id: EndpointId,
    stable_id: usize,
    connections: &Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: &Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
) {
    let mut conns = connections.write().await;
    let should_remove = conns
        .get(&endpoint_id)
        .map(|current| current.stable_id() == stable_id)
        .unwrap_or(false);
    if should_remove {
        conns.remove(&endpoint_id);
        connection_insert_at_ms.write().await.remove(&endpoint_id);
    }
}

impl PlutoniumProtocol {
    pub const ALPN: &[u8] = b"plutonium/p2p/1";

    fn new(
        event_sender: broadcast::Sender<AcceptEvent>,
        stream_sender: async_channel::Sender<IncomingStream>,
        connections: Arc<RwLock<HashMap<EndpointId, Connection>>>,
        connection_insert_at_ms: Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
        local_endpoint_id: EndpointId,
        probe_registry: IrohProbeRegistry,
    ) -> Self {
        Self {
            event_sender,
            stream_sender,
            connections,
            connection_insert_at_ms,
            local_endpoint_id,
            probe_registry,
        }
    }

    async fn handle_connection(
        self,
        connection: Connection,
    ) -> std::result::Result<(), AcceptError> {
        let endpoint_id = connection.remote_id();
        let stable_id = connection.stable_id();
        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][accept] ENTER endpoint_id={} stable_id={}",
            endpoint_id, stable_id
        )));
        println!("[PlutoniumWasm] Accepting connection from: {}", endpoint_id);

        // Endpoint ordering chooses one canonical physical dial. Preserve an
        // existing canonical leg; replace a live noncanonical leg only when
        // this inbound accept is canonical for the local endpoint.
        {
            let now_ms = js_sys::Date::now();
            let mut conns = self.connections.write().await;
            let mut insert_times = self.connection_insert_at_ms.write().await;
            if let Some(previous) = conns.get(&endpoint_id).cloned() {
                let previous_metadata =
                    insert_times
                        .get(&endpoint_id)
                        .copied()
                        .unwrap_or(IrohConnectionMetadata {
                            inserted_at_ms: now_ms,
                            direction: IrohConnectionDirection::Inbound,
                        });
                let previous_age_ms = (now_ms - previous_metadata.inserted_at_ms).max(0.0) as u64;
                let prefers_outbound = local_prefers_outbound(self.local_endpoint_id, endpoint_id);
                match decide_inbound_install(
                    Some(ExistingConnectionState {
                        same_stable_id: previous.stable_id() == stable_id,
                        alive: previous.close_reason().is_none(),
                        direction: previous_metadata.direction,
                        age_ms: previous_age_ms,
                    }),
                    prefers_outbound,
                ) {
                    IrohConnectionInstallDecision::Install => {
                        if previous.stable_id() != stable_id {
                            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                                "[pluto-rtc][wasm-node][accept] replacing dead stored connection endpoint_id={} previous_stable_id={} incoming_stable_id={}",
                                endpoint_id,
                                previous.stable_id(),
                                stable_id
                            )));
                        }
                    }
                    IrohConnectionInstallDecision::ReplaceExisting {
                        close_existing_reason,
                    } => {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][wasm-node][accept] canonical inbound replacing noncanonical live connection endpoint_id={} existing_stable_id={} accept_stable_id={} previous_age_ms={}",
                            endpoint_id,
                            previous.stable_id(),
                            stable_id,
                            previous_age_ms,
                        )));
                        previous.close(0u8.into(), close_existing_reason.as_bytes());
                    }
                    IrohConnectionInstallDecision::KeepExisting { close_fresh_reason } => {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][wasm-node][accept] existing live connection present; closing duplicate accept endpoint_id={} existing_stable_id={} accept_stable_id={} previous_age_ms={}",
                            endpoint_id,
                            previous.stable_id(),
                            stable_id,
                            previous_age_ms,
                        )));
                        drop(insert_times);
                        drop(conns);
                        connection.close(0u8.into(), close_fresh_reason.as_bytes());
                        return Ok(());
                    }
                }
            }
            conns.insert(endpoint_id, connection.clone());
            insert_times.insert(
                endpoint_id,
                IrohConnectionMetadata {
                    inserted_at_ms: now_ms,
                    direction: IrohConnectionDirection::Inbound,
                },
            );
        }

        match self.event_sender.send(AcceptEvent::Accepted {
            endpoint_id,
            transport_stable_id: stable_id as u64,
        }) {
            Ok(receivers) => {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][accept] Accepted event dispatched endpoint_id={} receivers={}",
                    endpoint_id, receivers
                )));
            }
            Err(e) => {
                web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][accept] FAILED to send Accepted event endpoint_id={} error={}",
                    endpoint_id, e
                )));
            }
        }

        let outcome = run_connection_loop(
            "accept",
            &connection,
            self.stream_sender.clone(),
            self.probe_registry.clone(),
        )
        .await;

        self.event_sender
            .send(AcceptEvent::Closed {
                endpoint_id,
                transport_stable_id: stable_id as u64,
                error: outcome.close_reason,
                was_locally_closed: outcome.was_locally_closed,
            })
            .ok();

        remove_connection_if_current(
            endpoint_id,
            stable_id,
            &self.connections,
            &self.connection_insert_at_ms,
        )
        .await;

        Ok(())
    }
}

impl ProtocolHandler for PlutoniumProtocol {
    #[allow(refining_impl_trait)]
    fn accept(
        &self,
        connection: Connection,
    ) -> impl n0_future::Future<Output = std::result::Result<(), AcceptError>> + std::marker::Send
    {
        let proto = self.clone();
        async move { proto.handle_connection(connection).await }
    }
}

/// Install a freshly-dialed outbound connection into the shared map using
/// deterministic endpoint ordering. A canonical live leg is idempotent; a
/// canonical fresh leg may replace only a noncanonical live leg.
/// Returns `true` if this connection was installed and should be tracked as
/// the active transport; `false` if we deferred to an existing live leg
/// (the helper closes the freshly-dialed `connection`; the caller drops it).
async fn install_outbound_connection(
    endpoint_id: EndpointId,
    connection: &Connection,
    connections: &Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: &Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
    local_endpoint_id: EndpointId,
    source: &str,
) -> bool {
    let stable_id = connection.stable_id();
    let now_ms = js_sys::Date::now();
    let mut conns = connections.write().await;
    let mut insert_times = connection_insert_at_ms.write().await;
    if let Some(previous) = conns.get(&endpoint_id).cloned() {
        let previous_metadata =
            insert_times
                .get(&endpoint_id)
                .copied()
                .unwrap_or(IrohConnectionMetadata {
                    inserted_at_ms: now_ms,
                    direction: IrohConnectionDirection::Outbound,
                });
        let previous_age_ms = (now_ms - previous_metadata.inserted_at_ms).max(0.0) as u64;
        let prefers_outbound = local_prefers_outbound(local_endpoint_id, endpoint_id);
        match decide_outbound_install(
            Some(ExistingConnectionState {
                same_stable_id: previous.stable_id() == stable_id,
                alive: previous.close_reason().is_none(),
                direction: previous_metadata.direction,
                age_ms: previous_age_ms,
            }),
            prefers_outbound,
        ) {
            IrohConnectionInstallDecision::Install => {
                if previous.stable_id() != stable_id {
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][wasm-node][{}] replacing dead stored connection endpoint_id={} previous_stable_id={} outgoing_stable_id={}",
                        source,
                        endpoint_id,
                        previous.stable_id(),
                        stable_id
                    )));
                }
            }
            IrohConnectionInstallDecision::ReplaceExisting {
                close_existing_reason,
            } => {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][{}] canonical outbound replacing noncanonical live connection endpoint_id={} existing_stable_id={} dialed_stable_id={} previous_age_ms={}",
                    source,
                    endpoint_id,
                    previous.stable_id(),
                    stable_id,
                    previous_age_ms,
                )));
                previous.close(0u8.into(), close_existing_reason.as_bytes());
            }
            IrohConnectionInstallDecision::KeepExisting { close_fresh_reason } => {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][{}] existing live connection present; deferring to existing endpoint_id={} existing_stable_id={} dialed_stable_id={} previous_age_ms={}",
                    source,
                    endpoint_id,
                    previous.stable_id(),
                    stable_id,
                    previous_age_ms,
                )));
                connection.close(0u8.into(), close_fresh_reason.as_bytes());
                return false;
            }
        }
    }
    conns.insert(endpoint_id, connection.clone());
    insert_times.insert(
        endpoint_id,
        IrohConnectionMetadata {
            inserted_at_ms: now_ms,
            direction: IrohConnectionDirection::Outbound,
        },
    );
    true
}

async fn connect(
    endpoint: &Endpoint,
    endpoint_id: EndpointId,
    event_sender: Sender<ConnectEvent>,
    accept_events: Option<broadcast::Sender<AcceptEvent>>,
    connections: Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
    stream_sender: async_channel::Sender<IncomingStream>,
    probe_registry: IrohProbeRegistry,
) -> Result<()> {
    // Reuse a canonical live leg. If the stored leg is noncanonical and this
    // endpoint owns the canonical outbound direction, one repair dial is
    // allowed to converge both peers on the same physical QUIC connection.
    if outbound_dial_precheck(
        endpoint,
        endpoint_id,
        &connections,
        &connection_insert_at_ms,
    )
    .await
    {
        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect] live connection available after canonical-dial precheck; reusing endpoint_id={}",
            endpoint_id
        )));
        let _ = event_sender.send(ConnectEvent::Connected).await;
        return Ok(());
    }

    let connection = endpoint
        .connect(endpoint_id, PlutoniumProtocol::ALPN)
        .await?;
    let stable_id = connection.stable_id();

    let installed = install_outbound_connection(
        endpoint_id,
        &connection,
        &connections,
        &connection_insert_at_ms,
        endpoint.id(),
        "connect",
    )
    .await;

    if !installed {
        // A canonical live connection already exists for this endpoint. The
        // install helper closed this duplicate dial; still notify the TS side
        // that the transport is connected — it will use the existing entry
        // in the connections map.
        if let Err(e) = event_sender.send(ConnectEvent::Connected).await {
            web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][wasm-node][connect] event_sender.send(Connected) FAILED after superseded dial endpoint_id={} stable_id={} error={}",
                endpoint_id, stable_id, e
            )));
        }
        let _ = event_sender
            .send(ConnectEvent::Closed { error: None })
            .await;
        return Ok(());
    }

    if let Err(e) = event_sender.send(ConnectEvent::Connected).await {
        web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect] event_sender.send(Connected) FAILED (channel closed — TS side gave up?) endpoint_id={} stable_id={} error={}",
            endpoint_id, stable_id, e
        )));
        return Err(anyhow::anyhow!("event channel closed"));
    }

    let outcome = run_connection_loop("connect", &connection, stream_sender, probe_registry).await;

    if let Err(e) = event_sender
        .send(ConnectEvent::Closed {
            error: outcome.close_reason.clone(),
        })
        .await
    {
        web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect] event_sender.send(Closed) FAILED (TS already torn down) endpoint_id={} error={}",
            endpoint_id, e
        )));
    }

    if is_manual_disconnect_close_reason(outcome.close_reason.as_deref()) {
        if let Some(sender) = accept_events {
            let _ = sender.send(AcceptEvent::Closed {
                endpoint_id,
                transport_stable_id: stable_id as u64,
                error: outcome.close_reason.clone(),
                was_locally_closed: outcome.was_locally_closed,
            });
        }
    }

    // Keep the endpoint mapping visible until the close event has been
    // processed by the runtime bridge. This mirrors the accept-side ordering
    // and avoids a transient "no live transport" gap while duplicate-close
    // recovery is deciding whether to preserve or replace the connection.
    remove_connection_if_current(
        endpoint_id,
        stable_id,
        &connections,
        &connection_insert_at_ms,
    )
    .await;

    Ok(())
}

async fn connect_addr(
    endpoint: &Endpoint,
    endpoint_id: EndpointId,
    endpoint_addr: EndpointAddr,
    connect_addr_waiters: Arc<RwLock<HashMap<EndpointId, Vec<Sender<ConnectEvent>>>>>,
    accept_events: Option<broadcast::Sender<AcceptEvent>>,
    connections: Arc<RwLock<HashMap<EndpointId, Connection>>>,
    connection_insert_at_ms: Arc<RwLock<HashMap<EndpointId, IrohConnectionMetadata>>>,
    stream_sender: async_channel::Sender<IncomingStream>,
    probe_registry: IrohProbeRegistry,
) -> Result<()> {
    // Reuse a canonical live leg. A noncanonical accepted leg is repaired only
    // when this endpoint owns the canonical outbound direction.
    if outbound_dial_precheck(
        endpoint,
        endpoint_id,
        &connections,
        &connection_insert_at_ms,
    )
    .await
    {
        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect_addr] live connection available after canonical-dial precheck; reusing endpoint_id={}",
            endpoint_id
        )));
        publish_connect_addr_event(&connect_addr_waiters, endpoint_id, ConnectEvent::Connected)
            .await;
        return Ok(());
    }

    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
        "[pluto-rtc][wasm-node][connect_addr] dialing endpoint_id={} addr={:?}",
        endpoint_id, endpoint_addr
    )));
    let connection = match endpoint
        .connect(endpoint_addr, PlutoniumProtocol::ALPN)
        .await
    {
        Ok(conn) => {
            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][wasm-node][connect_addr] connected! endpoint_id={} stable_id={}",
                endpoint_id,
                conn.stable_id()
            )));
            conn
        }
        Err(e) => {
            web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][wasm-node][connect_addr] connect FAILED endpoint_id={} error={}",
                endpoint_id, e
            )));
            publish_connect_addr_event(
                &connect_addr_waiters,
                endpoint_id,
                ConnectEvent::Closed {
                    error: Some(e.to_string()),
                },
            )
            .await;
            return Err(e.into());
        }
    };
    let stable_id = connection.stable_id();

    let installed = install_outbound_connection(
        endpoint_id,
        &connection,
        &connections,
        &connection_insert_at_ms,
        endpoint.id(),
        "connect_addr",
    )
    .await;

    if !installed {
        // A live connection slipped in between the precheck and our
        // `endpoint.connect()` completing. Close this duplicate locally so
        // the remote drops its half of the QUIC connection promptly. We
        // can't avoid the brief peer-side accept here, but the deterministic
        // direction rule has already selected the authoritative transport.
        let delivered =
            publish_connect_addr_event(&connect_addr_waiters, endpoint_id, ConnectEvent::Connected)
                .await;
        if delivered == 0 {
            web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][wasm-node][connect_addr] no live waiters after superseded dial endpoint_id={} stable_id={}",
                endpoint_id, stable_id
            )));
        }
        publish_connect_addr_event(
            &connect_addr_waiters,
            endpoint_id,
            ConnectEvent::Closed { error: None },
        )
        .await;
        return Ok(());
    }

    let delivered =
        publish_connect_addr_event(&connect_addr_waiters, endpoint_id, ConnectEvent::Connected)
            .await;
    if delivered == 0 {
        web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect_addr] no live waiters after connect; closing unobserved transport endpoint_id={} stable_id={}",
            endpoint_id, stable_id
        )));
        // TS side has already torn down the receiver — closing the connection
        // so we don't leak a half-attached QUIC conn that nothing observes.
        connection.close(0u8.into(), b"ts-channel-closed");
        return Err(anyhow::anyhow!("event channel closed"));
    }

    let outcome =
        run_connection_loop("connect_addr", &connection, stream_sender, probe_registry).await;

    let delivered = publish_connect_addr_event(
        &connect_addr_waiters,
        endpoint_id,
        ConnectEvent::Closed {
            error: outcome.close_reason.clone(),
        },
    )
    .await;
    if delivered == 0 {
        web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
            "[pluto-rtc][wasm-node][connect_addr] no live waiters for Closed event endpoint_id={}",
            endpoint_id
        )));
    }

    // Only manual disconnects need the global accept-event bridge fallback.
    // Ordinary replacement churn (`replaced-by-new-inbound`,
    // `duplicate-dial-superseded`, etc.) belongs to the per-dial ConnectEvent
    // bridge; sending it through AcceptEvent makes the accept state machine
    // mark healthy replacement traffic as broken.
    if is_manual_disconnect_close_reason(outcome.close_reason.as_deref()) {
        if let Some(sender) = accept_events {
            let _ = sender.send(AcceptEvent::Closed {
                endpoint_id,
                transport_stable_id: stable_id as u64,
                error: outcome.close_reason.clone(),
                was_locally_closed: outcome.was_locally_closed,
            });
        }
    }

    // Keep the endpoint mapping visible until the close event has been
    // processed by the runtime bridge. This mirrors the accept-side ordering
    // and avoids a transient "no live transport" gap while duplicate-close
    // recovery is deciding whether to preserve or replace the connection.
    remove_connection_if_current(
        endpoint_id,
        stable_id,
        &connections,
        &connection_insert_at_ms,
    )
    .await;

    Ok(())
}

async fn publish_connect_addr_event(
    connect_addr_waiters: &Arc<RwLock<HashMap<EndpointId, Vec<Sender<ConnectEvent>>>>>,
    endpoint_id: EndpointId,
    event: ConnectEvent,
) -> usize {
    let waiters = {
        let waiters = connect_addr_waiters.read().await;
        waiters.get(&endpoint_id).cloned().unwrap_or_default()
    };

    let mut delivered = 0usize;
    for waiter in waiters {
        if waiter.send(event.clone()).await.is_ok() {
            delivered += 1;
        }
    }
    delivered
}

async fn clear_connect_addr_waiters(
    connect_addr_waiters: &Arc<RwLock<HashMap<EndpointId, Vec<Sender<ConnectEvent>>>>>,
    endpoint_id: EndpointId,
) {
    connect_addr_waiters.write().await.remove(&endpoint_id);
}

impl IrohWasmNode {
    pub async fn spawn_with_endpoint(endpoint: Endpoint) -> Result<Self> {
        let (event_sender, _event_receiver) = broadcast::channel(128);
        let (stream_sender, stream_receiver) = async_channel::bounded(64);
        let connections = Arc::new(RwLock::new(HashMap::new()));
        let connection_insert_at_ms = Arc::new(RwLock::new(HashMap::new()));
        let connect_addr_waiters = Arc::new(RwLock::new(HashMap::new()));
        let probe_registry = IrohProbeRegistry::new();

        let proto = PlutoniumProtocol::new(
            event_sender.clone(),
            stream_sender.clone(),
            connections.clone(),
            connection_insert_at_ms.clone(),
            endpoint.id(),
            probe_registry.clone(),
        );
        let router = Router::builder(endpoint)
            .accept(PlutoniumProtocol::ALPN, proto)
            .spawn();

        Ok(Self {
            router,
            accept_events: event_sender,
            connections,
            connection_insert_at_ms,
            connect_addr_waiters,
            incoming_streams: stream_sender,
            incoming_streams_receiver: stream_receiver,
            probe_registry,
        })
    }

    pub fn endpoint(&self) -> &Endpoint {
        self.router.endpoint()
    }

    pub async fn is_connected(&self, endpoint_id: EndpointId) -> bool {
        let conns = self.connections.read().await;
        if let Some(conn) = conns.get(&endpoint_id) {
            conn.close_reason().is_none()
        } else {
            false
        }
    }

    pub fn secret_key(&self) -> Vec<u8> {
        self.router.endpoint().secret_key().to_bytes().to_vec()
    }

    pub async fn node_addr(&self) -> Result<iroh::EndpointAddr> {
        let endpoint = self.router.endpoint();
        // Ticket minting is a UI-facing operation for browser apps (share links,
        // QR codes, drive grants). Do not let a slow or stale relay-online future
        // monopolize that path while another dial/reconnect is active. The
        // watched address is iroh's current best view; wait briefly for relay
        // readiness, then use the current address so callers can fail/retry at
        // the application boundary instead of hanging the page for minutes.
        let online = endpoint.online().fuse();
        let deadline = gloo_timers::future::sleep(std::time::Duration::from_millis(1_500)).fuse();
        futures::pin_mut!(online, deadline);
        if matches!(
            futures::future::select(online, deadline).await,
            Either::Right(_)
        ) {
            web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(
                "[pluto-rtc][wasm-node] node_addr relay online wait timed out; using current watched address",
            ));
        }
        Ok(endpoint.watch_addr().get())
    }

    pub fn accept_events(&self) -> BoxStream<AcceptEvent> {
        let receiver = self.accept_events.subscribe();
        Box::pin(
            BroadcastStream::new(receiver).filter_map(|event| futures::future::ready(event.ok())),
        )
    }

    pub fn connect(&self, endpoint_id: EndpointId) -> impl Stream<Item = ConnectEvent> + Unpin {
        let (event_sender, event_receiver) = async_channel::bounded(16);
        let endpoint = self.router.endpoint().clone();
        let connections = self.connections.clone();
        let connection_insert_at_ms = self.connection_insert_at_ms.clone();
        let stream_sender = self.incoming_streams.clone();
        let accept_events = self.accept_events.clone();
        let probe_registry = self.probe_registry.clone();

        task::spawn(async move {
            let res = connect(
                &endpoint,
                endpoint_id,
                event_sender.clone(),
                Some(accept_events),
                connections,
                connection_insert_at_ms,
                stream_sender,
                probe_registry,
            )
            .await;
            if let Err(e) = res {
                let error = Some(e.to_string());
                event_sender.send(ConnectEvent::Closed { error }).await.ok();
            }
        });

        Box::pin(event_receiver)
    }

    pub fn connect_addr(
        &self,
        endpoint_id: EndpointId,
        endpoint_addr: EndpointAddr,
    ) -> impl Stream<Item = ConnectEvent> + Unpin {
        let (event_sender, event_receiver) = async_channel::bounded(16);
        let endpoint = self.router.endpoint().clone();
        let connections = self.connections.clone();
        let connection_insert_at_ms = self.connection_insert_at_ms.clone();
        let connect_addr_waiters = self.connect_addr_waiters.clone();
        let stream_sender = self.incoming_streams.clone();
        let accept_events = self.accept_events.clone();
        let probe_registry = self.probe_registry.clone();

        task::spawn(async move {
            {
                let mut waiters = connect_addr_waiters.write().await;
                if let Some(existing_waiters) = waiters.get_mut(&endpoint_id) {
                    existing_waiters.push(event_sender);
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][wasm-node] connect_addr joined in-flight dial endpoint_id={} waiter_count={}",
                        endpoint_id,
                        existing_waiters.len()
                    )));
                    return;
                }
                waiters.insert(endpoint_id, vec![event_sender.clone()]);
            }

            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][wasm-node] connect_addr task spawned endpoint_id={}",
                endpoint_id
            )));
            let res = connect_addr(
                &endpoint,
                endpoint_id,
                endpoint_addr,
                connect_addr_waiters.clone(),
                Some(accept_events),
                connections,
                connection_insert_at_ms,
                stream_sender,
                probe_registry,
            )
            .await;
            if let Err(e) = res {
                web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node] connect_addr task FAILED endpoint_id={} error={}",
                    endpoint_id, e
                )));
            }
            clear_connect_addr_waiters(&connect_addr_waiters, endpoint_id).await;
        });

        Box::pin(event_receiver)
    }

    pub async fn disconnect(&self, endpoint_id: EndpointId) -> Result<()> {
        self.disconnect_with_reason(
            endpoint_id,
            crate::lifecycle_reason::REASON_DISCONNECTED_BY_USER,
        )
        .await
    }

    pub async fn disconnect_with_reason(
        &self,
        endpoint_id: EndpointId,
        reason: &str,
    ) -> Result<()> {
        let connection = {
            let mut conns = self.connections.write().await;
            let removed = conns.remove(&endpoint_id);
            self.connection_insert_at_ms
                .write()
                .await
                .remove(&endpoint_id);
            removed
        };

        if let Some(conn) = connection {
            web_sys::console::info_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[PlutoRTC][teardown-trace] WasmNode::disconnect endpoint_id={endpoint_id} (QUIC reason: {reason})"
            )));
            conn.close(1u8.into(), reason.as_bytes());
        }
        Ok(())
    }

    pub async fn disconnect_with_reason_if_current(
        &self,
        endpoint_id: EndpointId,
        expected_transport_stable_id: u64,
        reason: &str,
    ) -> Result<bool> {
        let connection = {
            let mut connections = self.connections.write().await;
            let is_current = connections.get(&endpoint_id).is_some_and(|connection| {
                connection.stable_id() as u64 == expected_transport_stable_id
            });
            if !is_current {
                return Ok(false);
            }
            let removed = connections.remove(&endpoint_id);
            self.connection_insert_at_ms
                .write()
                .await
                .remove(&endpoint_id);
            removed
        };

        if let Some(connection) = connection {
            connection.close(1u8.into(), reason.as_bytes());
            return Ok(true);
        }
        Ok(false)
    }

    pub async fn open_bi(&self, endpoint_id: EndpointId) -> Result<(SendStream, RecvStream)> {
        let connection = {
            let conns = self.connections.read().await;
            conns.get(&endpoint_id).cloned()
        };

        if let Some(conn) = connection {
            let (send, recv) = conn.open_bi().await?;
            Ok((send, recv))
        } else {
            Err(anyhow::anyhow!("No active connection to {}", endpoint_id))
        }
    }

    pub async fn open_uni(&self, endpoint_id: EndpointId) -> Result<SendStream> {
        let connection = {
            let conns = self.connections.read().await;
            conns.get(&endpoint_id).cloned()
        };

        if let Some(conn) = connection {
            let send = conn.open_uni().await?;
            Ok(send)
        } else {
            Err(anyhow::anyhow!("No active connection to {}", endpoint_id))
        }
    }

    /// Perform an application-level round trip on the current physical
    /// connection generation. A replacement that wins during the probe is
    /// reported as current and left for the next health pass.
    pub async fn probe_connection(
        &self,
        endpoint_id: EndpointId,
        timeout: std::time::Duration,
    ) -> Option<IrohConnectionProbe> {
        let connection = self.connections.read().await.get(&endpoint_id).cloned()?;
        let probed_stable_id = connection.stable_id() as u64;
        let responsive = self
            .probe_registry
            .probe(
                &endpoint_id.to_string(),
                probed_stable_id,
                &connection,
                timeout,
            )
            .await;
        let current_stable_id = self
            .connections
            .read()
            .await
            .get(&endpoint_id)
            .map(|current| current.stable_id() as u64)?;
        if current_stable_id != probed_stable_id {
            return Some(IrohConnectionProbe {
                transport_stable_id: probed_stable_id,
                responsive: false,
            });
        }
        Some(IrohConnectionProbe {
            transport_stable_id: probed_stable_id,
            responsive,
        })
    }

    pub fn incoming_streams_stream(&self) -> impl Stream<Item = IncomingStream> {
        self.incoming_streams_receiver.clone()
    }

    pub async fn incoming_stream_is_current(&self, incoming: &IncomingStream) -> bool {
        self.connections
            .read()
            .await
            .get(&incoming.endpoint_id)
            .is_some_and(|connection| connection.stable_id() as u64 == incoming.transport_stable_id)
    }

    pub async fn active_endpoint_ids(&self) -> Vec<EndpointId> {
        let conns = self.connections.read().await;
        conns.keys().cloned().collect()
    }

    pub async fn get_connection(&self, endpoint_id: EndpointId) -> Option<Connection> {
        let conns = self.connections.read().await;
        conns.get(&endpoint_id).cloned()
    }

    pub async fn add_node_addr(
        &self,
        node_id: EndpointId,
        _relay_url: Option<String>,
        _direct_addresses: Vec<String>,
    ) -> Result<()> {
        let _ = self
            .router
            .endpoint()
            .connect(node_id, PlutoniumProtocol::ALPN)
            .await;
        Ok(())
    }
}

#[wasm_bindgen]
pub struct BiStream {
    recv: JsReadableStream,
    send: JsWritableStream,
    endpoint_id: String,
    application_crypto_wrapped: bool,
    peer_send_state: Option<PeerSendState>,
}

#[wasm_bindgen]
pub struct PeerUniStream {
    writable: JsWritableStream,
    application_crypto_wrapped: bool,
    peer_send_state: PeerSendState,
}

#[wasm_bindgen]
impl PeerUniStream {
    #[wasm_bindgen(getter)]
    pub fn writable(&self) -> JsWritableStream {
        self.writable.clone()
    }

    #[wasm_bindgen(getter, js_name = applicationCryptoWrapped)]
    pub fn application_crypto_wrapped(&self) -> bool {
        self.application_crypto_wrapped
    }

    /// Finish the Rust-owned QUIC send half after JavaScript closes the writable.
    #[wasm_bindgen(js_name = finishSend)]
    pub async fn finish_send(&self) -> Result<(), JsValue> {
        finish_peer_send_state(self.peer_send_state.clone()).await
    }
}

#[wasm_bindgen]
impl BiStream {
    #[wasm_bindgen(getter)]
    pub fn recv(&self) -> JsReadableStream {
        self.recv.clone()
    }

    #[wasm_bindgen(getter)]
    pub fn send(&self) -> JsWritableStream {
        self.send.clone()
    }

    #[wasm_bindgen(getter)]
    pub fn endpoint_id(&self) -> String {
        self.endpoint_id.clone()
    }

    #[wasm_bindgen(getter, js_name = applicationCryptoWrapped)]
    pub fn application_crypto_wrapped(&self) -> bool {
        self.application_crypto_wrapped
    }

    /// Finish the Rust-owned QUIC send half after JavaScript closes the writable.
    ///
    /// Dropping an iroh send stream without `finish()` resets it. Keeping the
    /// stream in Rust until this method runs prevents short WASM writes from
    /// disappearing before the remote intake observes them.
    #[wasm_bindgen(js_name = finishSend)]
    pub async fn finish_send(&self) -> Result<(), JsValue> {
        let Some(state) = self.peer_send_state.clone() else {
            return Ok(());
        };
        finish_peer_send_state(state).await
    }
}

// Helpers for stream conversion
type PeerSendState = Arc<Mutex<Option<crate::application_crypto_streams::PeerSendStream>>>;

async fn finish_peer_send_state(state: PeerSendState) -> Result<(), JsValue> {
    let send = state
        .lock()
        .await
        .take()
        .ok_or_else(|| JsValue::from_str("peer send stream is already finished"))?;
    send.finish_and_wait_for_peer(std::time::Duration::from_secs(2))
        .await
        .map_err(|error| JsValue::from_str(&format!("finish peer send stream: {error}")))
}

pub fn peer_send_stream_to_writable(
    send: crate::application_crypto_streams::PeerSendStream,
) -> (JsWritableStream, PeerSendState) {
    let state = Arc::new(Mutex::new(Some(send)));
    let sink_state = state.clone();
    let sink = futures::sink::unfold(sink_state, |state, val: JsValue| async move {
        let bytes = js_sys::Uint8Array::new(&val).to_vec();
        {
            let mut guard = state.lock().await;
            let send = guard
                .as_mut()
                .ok_or_else(|| JsError::new("peer send stream is already finished"))?;
            send.write_all(&bytes)
                .await
                .map_err(|error| JsError::new(&error.to_string()))?;
        }
        Ok(state)
    });
    (WritableStream::from_sink(sink).into_raw(), state)
}

pub fn peer_recv_stream_to_readable(
    recv: crate::application_crypto_streams::PeerRecvStream,
) -> JsReadableStream {
    match recv {
        crate::application_crypto_streams::PeerRecvStream::Plain(inner) => {
            recv_stream_to_readable(inner)
        }
        crate::application_crypto_streams::PeerRecvStream::Encrypted(encrypted) => {
            let stream = futures::stream::unfold(encrypted, |mut encrypted| async move {
                let mut buffer = vec![0u8; READABLE_STREAM_CHUNK_BYTES];
                match encrypted.read(&mut buffer).await {
                    Ok(0) => None,
                    Ok(read_bytes) => {
                        buffer.truncate(read_bytes);
                        Some((
                            Ok(JsValue::from(js_sys::Uint8Array::from(buffer.as_slice()))),
                            encrypted,
                        ))
                    }
                    Err(error) => Some((
                        Err(JsValue::from(JsError::new(&error.to_string()))),
                        encrypted,
                    )),
                }
            });
            ReadableStream::from_stream(stream).into_raw()
        }
    }
}

pub fn send_stream_to_writable(send: SendStream) -> JsWritableStream {
    let writer = FramedWrite::new(send, BytesCodec::new());
    let writer =
        <FramedWrite<_, _> as SinkExt<bytes::Bytes>>::sink_map_err(writer, |e: std::io::Error| {
            JsError::new(&e.to_string())
        });
    let writer = writer.with(|val: JsValue| {
        let data = js_sys::Uint8Array::new(&val);
        let vec = data.to_vec();
        futures::future::ready(Ok(bytes::Bytes::from(vec)))
    });
    WritableStream::from_sink(writer).into_raw()
}

fn recv_stream_to_readable<R>(recv: R) -> JsReadableStream
where
    R: tokio::io::AsyncRead + Unpin + 'static,
{
    // Use 256KB buffer to match sender chunk size and avoid QUIC flow-control stalls.
    // The previous 1KB buffer caused the receiver to consume data too slowly, filling
    // the sender's flow-control window and deadlocking large transfers.
    //
    // Note: We use unfolding + `from_stream` instead of `from_async_read` to avoid
    // a known bug in `wasm_streams` 0.4.0 where `from_async_read` can panic with
    // `Option::unwrap_throw` on a None value when the stream is cancelled or drops
    // on certain mobile environments (like WebKit/Safari).
    let recv = recv.compat();
    let stream = futures::stream::unfold(recv, |mut recv| async move {
        let mut buffer = vec![0u8; READABLE_STREAM_CHUNK_BYTES];
        match recv.read(&mut buffer).await {
            Ok(0) => None,
            Ok(read_bytes) => {
                buffer.truncate(read_bytes);
                Some((
                    Ok(JsValue::from(js_sys::Uint8Array::from(buffer.as_slice()))),
                    recv,
                ))
            }
            Err(error) => Some((Err(JsValue::from(JsError::new(&error.to_string()))), recv)),
        }
    });

    ReadableStream::from_stream(stream).into_raw()
}

impl BiStream {
    fn set_object_property(
        object: &js_sys::Object,
        key: &str,
        value: &JsValue,
    ) -> std::result::Result<(), JsValue> {
        let did_set = js_sys::Reflect::set(object, &JsValue::from_str(key), value)?;
        if did_set {
            Ok(())
        } else {
            Err(JsValue::from_str(&format!(
                "Failed to set incoming stream property `{}`",
                key
            )))
        }
    }

    fn incoming_stream_value(
        incoming: IncomingStream,
    ) -> std::result::Result<(String, JsValue, String, u64, JsValue, JsValue), JsValue> {
        let endpoint_id = incoming.endpoint_id.to_string();
        let transport_stable_id = incoming.transport_stable_id;
        match incoming.stream {
            IncomingStreamType::Bi(send, recv) => {
                // `incoming_streams` is an ordered source. Do not read from an incoming
                // stream to classify it here: a quiet stream would otherwise prevent every
                // later control/application stream from reaching the JS dispatcher. The
                // dispatcher classifies each stream independently after delivery.
                let stream = js_sys::Object::new();
                let send = send_stream_to_writable(send);
                let recv = recv_stream_to_readable(recv);
                Self::set_object_property(&stream, "send", &JsValue::from(send))?;
                Self::set_object_property(&stream, "recv", &JsValue::from(recv))?;
                Self::set_object_property(
                    &stream,
                    "endpoint_id",
                    &JsValue::from_str(&endpoint_id),
                )?;
                let protocol_hint = serde_wasm_bindgen::to_value(
                    &crate::stream_metadata::IncomingProtocolHint::Unknown,
                )
                .map_err(|error| JsValue::from_str(&error.to_string()))?;
                Ok((
                    "bi".to_string(),
                    JsValue::from(stream),
                    endpoint_id,
                    transport_stable_id,
                    protocol_hint,
                    JsValue::NULL,
                ))
            }
            IncomingStreamType::Uni(recv) => {
                // Use the same cancellation-safe unfolding bridge as incoming bi
                // streams. `recv_stream_to_readable` deliberately avoids
                // wasm_streams::from_async_read, whose one-shot cancellation callback
                // can panic during WebKit/mobile teardown.
                let recv = recv_stream_to_readable(recv);
                Ok((
                    "uni".to_string(),
                    JsValue::from(recv),
                    endpoint_id,
                    transport_stable_id,
                    serde_wasm_bindgen::to_value(
                        &crate::stream_metadata::IncomingProtocolHint::Unknown,
                    )
                    .map_err(|error| JsValue::from_str(&error.to_string()))?,
                    JsValue::NULL,
                ))
            }
        }
    }

    pub fn incoming_to_js_value(incoming: IncomingStream) -> std::result::Result<JsValue, JsValue> {
        let (ty, stream, endpoint_id, transport_stable_id, protocol_hint, channel) =
            Self::incoming_stream_value(incoming)?;

        let object = js_sys::Object::new();
        Self::set_object_property(&object, "type", &JsValue::from_str(&ty))?;
        Self::set_object_property(&object, "stream", &stream)?;
        Self::set_object_property(&object, "endpointId", &JsValue::from_str(&endpoint_id))?;
        Self::set_object_property(
            &object,
            "transportStableId",
            &JsValue::from_f64(transport_stable_id as f64),
        )?;
        Self::set_object_property(&object, "protocolHint", &protocol_hint)?;
        Self::set_object_property(&object, "channel", &channel)?;

        Ok(JsValue::from(object))
    }

    pub fn from_parts(
        send_stream: SendStream,
        recv_stream: RecvStream,
        endpoint_id: String,
    ) -> Self {
        let send = send_stream_to_writable(send_stream);
        let recv = recv_stream_to_readable(recv_stream);
        Self {
            recv,
            send,
            endpoint_id,
            application_crypto_wrapped: false,
            peer_send_state: None,
        }
    }

    pub fn from_peer_parts(
        send_stream: crate::application_crypto_streams::PeerSendStream,
        recv_stream: crate::application_crypto_streams::PeerRecvStream,
        endpoint_id: String,
    ) -> Self {
        let application_crypto_wrapped = send_stream.is_encrypted() || recv_stream.is_encrypted();
        let (send, peer_send_state) = peer_send_stream_to_writable(send_stream);
        let recv = peer_recv_stream_to_readable(recv_stream);
        Self {
            recv,
            send,
            endpoint_id,
            application_crypto_wrapped,
            peer_send_state: Some(peer_send_state),
        }
    }
}

pub fn peer_uni_stream_from_send(
    send: crate::application_crypto_streams::PeerSendStream,
) -> PeerUniStream {
    let application_crypto_wrapped = send.is_encrypted();
    let (writable, peer_send_state) = peer_send_stream_to_writable(send);
    PeerUniStream {
        writable,
        application_crypto_wrapped,
        peer_send_state,
    }
}

pub fn to_js_err(err: impl Into<anyhow::Error>) -> JsError {
    let err: anyhow::Error = err.into();
    JsError::new(&err.to_string())
}

pub fn into_js_readable_stream<T: Serialize>(
    stream: impl Stream<Item = T> + 'static,
) -> wasm_streams::readable::sys::ReadableStream {
    let stream = stream.map(|event| Ok(serde_wasm_bindgen::to_value(&event).unwrap()));
    ReadableStream::from_stream(stream).into_raw()
}