openrtc 2.0.0

//! Browser WebTransport datagram pump for the Draft 14 MoQ Iroh carrier.

#![cfg(all(target_arch = "wasm32", feature = "iroh-transport-moq"))]

use std::{
    cell::{Cell, RefCell},
    rc::Rc,
};

use js_sys::{Function, Promise, Reflect, Uint8Array};
use wasm_bindgen::{JsCast, JsValue};
use wasm_bindgen_futures::{spawn_local, JsFuture};

use crate::{
    iroh_carrier::{
        segment_packet, CarrierControl, CarrierFrame, CarrierFrameExpectation, CarrierReassembler,
        CARRIER_CONTROL_PACKET_ID, CARRIER_HEADER_LEN,
    },
    packet_carrier_transport::PacketCarrierSession,
};

const FALLBACK_MOQ_OBJECT_PAYLOAD_CEILING: usize = 1_000;
// Keep the encoded object below the effective QUIC datagram ceiling across
// both Draft 14 relay hops. WebTransport reports its local hop ceiling, which
// can be larger than the relay's downstream hop and therefore is not by itself
// a safe end-to-end carrier MTU.
const MAX_MOQ_OBJECT_PAYLOAD_CEILING: usize = 1_000;
// Browser WebTransport object datagrams are unreliable and may discard a
// back-to-back fragment burst even after the relay's publisher FIFO accepts
// it. Pace only physical carrier fragments; this adds latency but no bytes or
// control-plane operations, and Iroh remains the retransmission owner.
const MOQ_PACKET_CARRIER_DATAGRAM_PACING: std::time::Duration =
    std::time::Duration::from_millis(10);

/// Retains browser datagram objects and bounded packet pumps. The admitted
/// Rust lifecycle actor remains responsible for eligibility and replacement.
pub struct WasmMoqCarrierSession {
    _datagrams: JsValue,
    writer: JsValue,
    closed: Rc<Cell<bool>>,
    _session: Rc<PacketCarrierSession>,
    expected: CarrierFrameExpectation,
    application_key: [u8; 32],
    terminal_acknowledged: Rc<Cell<bool>>,
}

impl std::fmt::Debug for WasmMoqCarrierSession {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("WasmMoqCarrierSession")
            .finish_non_exhaustive()
    }
}

impl WasmMoqCarrierSession {
    /// `datagrams` is a Draft 14 MoQ object-datagram duplex. It is deliberately
    /// not the raw `WebTransport.datagrams` object: the browser adapter must
    /// first negotiate the directed MoQ track and prove exact Draft 14 support.
    pub fn attach(
        datagrams: JsValue,
        session: PacketCarrierSession,
        expected: CarrierFrameExpectation,
        application_key: [u8; 32],
        on_terminal: Rc<dyn Fn(&'static str)>,
    ) -> Result<Self, JsValue> {
        let readable = Reflect::get(&datagrams, &JsValue::from_str("readable"))?;
        let writable = Reflect::get(&datagrams, &JsValue::from_str("writable"))?;
        let reader = method(&readable, "getReader")?.call0(&readable)?;
        let writer = method(&writable, "getWriter")?.call0(&writable)?;
        let message_ceiling = Reflect::get(&datagrams, &JsValue::from_str("maxDatagramSize"))
            .ok()
            .and_then(|value| value.as_f64())
            .filter(|value| value.is_finite() && *value >= 0.0)
            .map(|value| value.floor() as usize)
            .unwrap_or(FALLBACK_MOQ_OBJECT_PAYLOAD_CEILING)
            .min(MAX_MOQ_OBJECT_PAYLOAD_CEILING);
        if message_ceiling <= CARRIER_HEADER_LEN {
            return Err(JsValue::from_str(
                "MoQ object datagram payload ceiling is too small for carrier framing",
            ));
        }
        let session = Rc::new(session);
        let closed = Rc::new(Cell::new(false));

        let inbound_session = session.clone();
        let inbound_closed = closed.clone();
        let inbound_terminal = on_terminal.clone();
        let terminal_notified = Rc::new(Cell::new(false));
        let terminal_acknowledged = Rc::new(Cell::new(false));
        let inbound_terminal_notified = terminal_notified.clone();
        let inbound_terminal_acknowledged = terminal_acknowledged.clone();
        let inbound_reader = reader;
        let inbound_writer = writer.clone();
        let reassembler = Rc::new(RefCell::new(CarrierReassembler::default()));
        spawn_local(async move {
            while !inbound_closed.get() {
                let Ok(read_result) = call_promise(&inbound_reader, "read", None).await else {
                    break;
                };
                if Reflect::get(&read_result, &JsValue::from_str("done"))
                    .ok()
                    .and_then(|value| value.as_bool())
                    .unwrap_or(false)
                {
                    break;
                }
                let Ok(value) = Reflect::get(&read_result, &JsValue::from_str("value")) else {
                    break;
                };
                let bytes = Uint8Array::new(&value).to_vec();
                let frame = CarrierFrame::decode(&bytes, expected);
                let Ok(frame) = frame else {
                    continue;
                };
                match frame.terminal_control_kind(expected, &application_key) {
                    Ok(Some(CarrierControl::SessionTokenRevokedAck)) => {
                        inbound_terminal_acknowledged.set(true);
                        continue;
                    }
                    Ok(Some(control @ CarrierControl::SessionTokenRevoked)) => {
                        let ack = CarrierFrame::terminal_control(
                            expected,
                            &application_key,
                            CarrierControl::SessionTokenRevokedAck,
                        )
                        .encode();
                        if let Ok(ack) = ack {
                            let value = Uint8Array::from(ack.as_slice());
                            let _ =
                                call_promise(&inbound_writer, "write", Some(value.as_ref())).await;
                        }
                        if !inbound_terminal_notified.replace(true) {
                            inbound_session.close();
                            if let Some(reason) = control.lifecycle_reason() {
                                inbound_terminal(reason);
                            }
                        }
                        continue;
                    }
                    Err(_) if frame.header.packet_id == CARRIER_CONTROL_PACKET_ID => continue,
                    _ => {}
                }
                let now_ms = js_sys::Date::now().max(0.0) as u64;
                let Ok(Some(packet)) = reassembler.borrow_mut().push(frame, now_ms) else {
                    continue;
                };
                if inbound_session.deliver_inbound(packet).await.is_err() {
                    break;
                }
            }
            if !inbound_closed.get() && !inbound_terminal_notified.replace(true) {
                inbound_session.close();
                inbound_terminal("iroh-carrier-ended");
            }
        });

        let outbound_session = session.clone();
        let outbound_closed = closed.clone();
        let outbound_terminal = on_terminal;
        let outbound_terminal_notified = terminal_notified;
        let outbound_writer = writer.clone();
        spawn_local(async move {
            let mut packet_id = 0_u64;
            'packets: while !outbound_closed.get() {
                let Ok(packet) = outbound_session.recv_outbound().await else {
                    break;
                };
                packet_id = packet_id.wrapping_add(1);
                if packet_id == CARRIER_CONTROL_PACKET_ID {
                    packet_id = 0;
                }
                let Ok(frames) = segment_packet(&packet, expected, packet_id, message_ceiling)
                else {
                    break;
                };
                for frame in frames {
                    let Ok(encoded) = frame.encode() else {
                        break 'packets;
                    };
                    let value = Uint8Array::from(encoded.as_slice());
                    if call_promise(&outbound_writer, "write", Some(value.as_ref()))
                        .await
                        .is_err()
                    {
                        break 'packets;
                    }
                    gloo_timers::future::sleep(MOQ_PACKET_CARRIER_DATAGRAM_PACING).await;
                }
            }
            if !outbound_closed.get() && !outbound_terminal_notified.replace(true) {
                outbound_session.close();
                outbound_terminal("iroh-carrier-ended");
            }
        });

        Ok(Self {
            _datagrams: datagrams,
            writer,
            closed,
            _session: session,
            expected,
            application_key,
            terminal_acknowledged,
        })
    }

    /// Send the authenticated terminal marker until its authenticated ACK or
    /// the bounded local-attempt ceiling. This performs no gateway or
    /// control-plane operation.
    pub async fn send_terminal(&self, reason: &str) -> Result<(), JsValue> {
        let control = match reason {
            crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED => {
                CarrierControl::SessionTokenRevoked
            }
            _ => return Ok(()),
        };
        let encoded = CarrierFrame::terminal_control(self.expected, &self.application_key, control)
            .encode()
            .map_err(|error| JsValue::from_str(&error.to_string()))?;
        self.terminal_acknowledged.set(false);
        for _ in 0..12 {
            let value = Uint8Array::from(encoded.as_slice());
            call_promise(&self.writer, "write", Some(value.as_ref())).await?;
            gloo_timers::future::sleep(std::time::Duration::from_millis(25)).await;
            if self.terminal_acknowledged.get() {
                break;
            }
        }
        Ok(())
    }
}

fn method(target: &JsValue, name: &str) -> Result<Function, JsValue> {
    Reflect::get(target, &JsValue::from_str(name))?.dyn_into::<Function>()
}

async fn call_promise(
    target: &JsValue,
    name: &str,
    argument: Option<&JsValue>,
) -> Result<JsValue, JsValue> {
    let value = match argument {
        Some(argument) => method(target, name)?.call1(target, argument)?,
        None => method(target, name)?.call0(target)?,
    };
    JsFuture::from(value.dyn_into::<Promise>()?).await
}

impl Drop for WasmMoqCarrierSession {
    fn drop(&mut self) {
        self.closed.set(true);
        self._session.close();
        if let Ok(close) = method(&self.writer, "close") {
            let _ = close.call0(&self.writer);
        }
    }
}