openrtc 0.2.0

#![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::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::iroh_connection_policy::{
    decide_inbound_install, decide_outbound_install, should_redial_without_precheck,
    ExistingConnectionState, IrohConnectionInstallDecision,
};

const READABLE_STREAM_CHUNK_BYTES: usize = 256 * 1024;

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

pub struct IncomingStream {
    pub endpoint_id: EndpointId,
    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()
}

#[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, f64>>>,
    connect_addr_waiters: Arc<RwLock<HashMap<EndpointId, Vec<Sender<ConnectEvent>>>>>,
    incoming_streams: async_channel::Sender<IncomingStream>,
    incoming_streams_receiver: async_channel::Receiver<IncomingStream>,
}

#[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,
    },
    Closed {
        endpoint_id: EndpointId,
        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, f64>>>,
    local_endpoint_id: EndpointId,
}

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>,
) -> 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,
                            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
                            )));
                        }
                        if let Err(e) = stream_sender.send(IncomingStream {
                            endpoint_id,
                            stream: IncomingStreamType::Uni(recv),
                        }).await {
                            web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                                "[pluto-rtc][wasm-node][{}] FAILED to forward uni 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_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, f64>>>,
) {
    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";

    pub 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, f64>>>,
        local_endpoint_id: EndpointId,
    ) -> Self {
        Self {
            event_sender,
            stream_sender,
            connections,
            connection_insert_at_ms,
            local_endpoint_id,
        }
    }

    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);

        // Sticky first-alive policy: if we already have a live connection
        // for this endpoint (either an earlier accept or our own completed
        // dial), close this duplicate accept and keep the existing one. Both
        // peers applying this rule converge on the same underlying QUIC
        // connection (the one that completed first on both sides), which
        // eliminates the "wrong transport" blackhole caused by symmetric
        // dialing. We only replace when the existing connection is already
        // closed (dead leg), so we never tear down a working transport.
        //
        // Zombie-replace exception: after a peer restart, the remote has new
        // local state and dials fresh (new stable_id), but our side still has
        // the old connection object which iroh's QUIC layer believes is alive
        // until keep-alive timeout (~27 s). If the stored connection is older
        // than ZOMBIE_REPLACE_AGE_MS it cannot be a symmetric-dial race
        // (those settle within a few RTTs), so we replace it eagerly rather
        // than wait for QUIC to discover the deadness.
        {
            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_inserted_at =
                    insert_times.get(&endpoint_id).copied().unwrap_or(now_ms);
                let previous_age_ms = (now_ms - previous_inserted_at).max(0.0) as u64;
                match decide_inbound_install(Some(ExistingConnectionState {
                    same_stable_id: previous.stable_id() == stable_id,
                    alive: previous.close_reason().is_none(),
                    age_ms: previous_age_ms,
                    prefer_fresh_duplicate: !local_prefers_outbound(
                        self.local_endpoint_id,
                        endpoint_id,
                    ),
                })) {
                    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::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(());
                    }
                    IrohConnectionInstallDecision::ReplaceExisting {
                        close_previous_reason,
                    } => {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][wasm-node][accept] replacing zombie stored connection (remote likely restarted) endpoint_id={} previous_stable_id={} incoming_stable_id={} previous_age_ms={}",
                            endpoint_id,
                            previous.stable_id(),
                            stable_id,
                            previous_age_ms,
                        )));
                        // Force-close the zombie so iroh drops its half of
                        // the QUIC connection immediately and the peer's
                        // read/write loops unwind without waiting for the
                        // 120 s idle timeout.
                        previous.close(0u8.into(), close_previous_reason.as_bytes());
                    }
                }
            }
            conns.insert(endpoint_id, connection.clone());
            insert_times.insert(endpoint_id, now_ms);
        }

        match self
            .event_sender
            .send(AcceptEvent::Accepted { endpoint_id })
        {
            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()).await;

        self.event_sender
            .send(AcceptEvent::Closed {
                endpoint_id,
                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
/// the sticky first-alive policy. If a live connection already exists (an
/// earlier accept from the peer's own dial, or a previous successful dial),
/// we close this duplicate dial and keep the existing one. Both sides
/// applying this rule converge on a single underlying QUIC connection.
/// Returns `true` if this connection was installed and should be tracked as
/// the active transport; `false` if we deferred to an existing live leg
/// (caller should close the freshly-dialed `connection` and drop 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, f64>>>,
    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_inserted_at = insert_times.get(&endpoint_id).copied().unwrap_or(now_ms);
        let previous_age_ms = (now_ms - previous_inserted_at).max(0.0) as u64;
        match decide_outbound_install(Some(ExistingConnectionState {
            same_stable_id: previous.stable_id() == stable_id,
            alive: previous.close_reason().is_none(),
            age_ms: previous_age_ms,
            prefer_fresh_duplicate: local_prefers_outbound(local_endpoint_id, endpoint_id),
        })) {
            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::KeepExisting { .. } => {
                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,
                )));
                return false;
            }
            IrohConnectionInstallDecision::ReplaceExisting {
                close_previous_reason,
            } => {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][{}] replacing zombie stored connection (local dial fresh) endpoint_id={} previous_stable_id={} outgoing_stable_id={} previous_age_ms={}",
                    source,
                    endpoint_id,
                    previous.stable_id(),
                    stable_id,
                    previous_age_ms,
                )));
                previous.close(0u8.into(), close_previous_reason.as_bytes());
            }
        }
    }
    conns.insert(endpoint_id, connection.clone());
    insert_times.insert(endpoint_id, now_ms);
    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, f64>>>,
    stream_sender: async_channel::Sender<IncomingStream>,
) -> Result<()> {
    // Sticky first-alive precheck. See `connect_addr` for the rationale —
    // dialing on top of a live connection causes the peer's
    // newest-inbound-wins policy to evict the existing good transport.
    {
        let conns = connections.read().await;
        if let Some(existing) = conns.get(&endpoint_id) {
            if !local_prefers_outbound(endpoint.id(), endpoint_id)
                && !should_redial_without_precheck(existing.close_reason().is_none())
            {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][connect] live connection already present; reusing without redial endpoint_id={} existing_stable_id={}",
                    endpoint_id,
                    existing.stable_id()
                )));
                drop(conns);
                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 live connection already exists for this endpoint (the sticky
        // first-alive rule). Close this duplicate dial so the remote drops
        // its half of this QUIC connection, but still notify the TS side
        // that the transport is connected — it will use the existing entry
        // in the connections map.
        connection.close(
            0u8.into(),
            crate::lifecycle_reason::REASON_DUPLICATE_DIAL_SUPERSEDED.as_bytes(),
        );
        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).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,
                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, f64>>>,
    stream_sender: async_channel::Sender<IncomingStream>,
) -> Result<()> {
    // Sticky first-alive precheck: if a live outbound or accepted connection
    // already exists for this endpoint, do NOT initiate a fresh QUIC dial.
    // The previous behaviour dialed first and closed the duplicate after the
    // fact, but `endpoint.connect()` always reaches the peer — and the peer's
    // newest-inbound-wins policy then closes the EXISTING good connection
    // with `replaced-by-new-inbound`, leaving both sides without a working
    // transport. Short-circuiting here keeps the existing live connection
    // authoritative and lets racing callers (e.g. WasmBridge.connectToDevice
    // → connect_device + Client.connect) converge on it.
    {
        let conns = connections.read().await;
        if let Some(existing) = conns.get(&endpoint_id) {
            if !local_prefers_outbound(endpoint.id(), endpoint_id)
                && !should_redial_without_precheck(existing.close_reason().is_none())
            {
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][wasm-node][connect_addr] live connection already present; reusing without redial endpoint_id={} existing_stable_id={}",
                    endpoint_id,
                    existing.stable_id()
                )));
                drop(conns);
                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 we also don't
        // want to keep the duplicate around — the sticky first-alive rule
        // has already locked in an authoritative transport.
        connection.close(
            0u8.into(),
            crate::lifecycle_reason::REASON_DUPLICATE_DIAL_SUPERSEDED.as_bytes(),
        );
        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).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,
                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 proto = PlutoniumProtocol::new(
            event_sender.clone(),
            stream_sender.clone(),
            connections.clone(),
            connection_insert_at_ms.clone(),
            endpoint.id(),
        );
        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,
        })
    }

    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();

        task::spawn(async move {
            let res = connect(
                &endpoint,
                endpoint_id,
                event_sender.clone(),
                Some(accept_events),
                connections,
                connection_insert_at_ms,
                stream_sender,
            )
            .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();

        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,
            )
            .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 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))
        }
    }

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

    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,
}

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

#[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
    }
}

#[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
    }
}

// Helpers for stream conversion
pub fn peer_send_stream_to_writable(
    send: crate::application_crypto_streams::PeerSendStream,
) -> JsWritableStream {
    match send {
        crate::application_crypto_streams::PeerSendStream::Plain(inner) => {
            send_stream_to_writable(inner)
        }
        crate::application_crypto_streams::PeerSendStream::Encrypted(encrypted) => {
            let sink = futures::sink::unfold(encrypted, |mut encrypted, val: JsValue| async move {
                let bytes = js_sys::Uint8Array::new(&val).to_vec();
                encrypted
                    .write_all(&bytes)
                    .await
                    .map_err(|error| JsError::new(&error.to_string()))?;
                Ok(encrypted)
            });
            WritableStream::from_sink(sink).into_raw()
        }
    }
}

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(recv: RecvStream) -> JsReadableStream {
    // 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, JsValue, JsValue), JsValue> {
        let endpoint_id = incoming.endpoint_id.to_string();
        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,
                    protocol_hint,
                    JsValue::NULL,
                ))
            }
            IncomingStreamType::Uni(recv) => {
                // Web/mobile share flows can receive probe-style uni streams that the
                // TS client intentionally ignores. Building a JS ReadableStream wrapper
                // for these unhandled streams has triggered wasm-stream callback
                // lifecycle panics on mobile browsers. Drop the Rust recv stream and
                // surface a null payload so TS can treat it as intentionally dropped.
                drop(recv);
                Ok((
                    "uni".to_string(),
                    JsValue::NULL,
                    endpoint_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, 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, "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,
        }
    }

    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_stream_to_writable(send_stream);
        let recv = peer_recv_stream_to_readable(recv_stream);
        Self {
            recv,
            send,
            endpoint_id,
            application_crypto_wrapped,
        }
    }
}

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

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()
}