inferencelayer 0.2.8

//! Native WebRTC data-channel endpoint — **P2 (endpoint) + P3 (byte-pipe transport).**
//!
//! The browser half of the transport was de-risked first (`apps/lfm2-browser-worker/src/webrtc.rs`,
//! P0/P1): a wasm `RTCPeerConnection` brought up through the coordinator-relayed, non-trickle
//! offer/answer handshake. This module is the *native* mirror, so a V100/Mac/Linux worker speaks the
//! same data channels — the piece that lets any two peers (browser↔native, native↔native), even both
//! behind NAT, forward hidden states directly.
//!
//! It is a **feature-gated optional adapter** (`webrtc`), off by default (per the feature-flags
//! rule): the `webrtc-rs` stack (ICE/DTLS/SCTP) is large and native-only, so only a `--features
//! webrtc` build carries it. The default/`net`/`ws`/wasm builds are byte-for-byte unaffected.
//!
//! **P2 — endpoint.** [`NativeWebrtcPeer::make_offer`]/[`accept_offer`]/[`accept_answer`] run the
//! non-trickle handshake ([`RTCPeerConnection::gathering_complete_promise`] gathers every ICE
//! candidate before replying, so each relayed SDP is complete), built on Google STUN by default
//! ([`default_ice_servers`]). A process-global multi-threaded tokio runtime ([`rt`]) drives each
//! peer's background ICE/DTLS/SCTP tasks; the sync `serve_conn` `OP_SIGNAL` arm `block_on`s one step
//! per signaling op.
//!
//! **P3 — byte pipe.** Once the channel opens, [`NativeWebrtcPeer::open_pipe`] hands back a
//! [`WebrtcPipe`]: a blocking [`std::io::Read`] + [`std::io::Write`] over the data channel, the exact
//! shape of the WebSocket `WsPipe`. Writes `block_on` `dc.send` (one protocol frame = one SCTP
//! message); reads drain inbound messages (fed by the async `on_message` callback through an mpsc)
//! with the same partial-read buffering. Because it is just another `Conn`, the fleet's whole
//! length-prefixed protocol — forward-mode hidden-state hops, in-band control ops, the sink — rides
//! it unchanged and bitwise-identically to TCP/WS.
//!
//! [`accept_offer`]: NativeWebrtcPeer::accept_offer
//! [`accept_answer`]: NativeWebrtcPeer::accept_answer
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use bytes::Bytes;
use turn::auth::{AuthHandler, generate_auth_key};
use turn::relay::relay_static::RelayAddressGeneratorStatic;
use turn::server::Server;
use turn::server::config::{ConnConfig, ServerConfig};
use webrtc::api::APIBuilder;
use webrtc::data_channel::RTCDataChannel;
use webrtc::data_channel::data_channel_message::DataChannelMessage;
use webrtc::ice_transport::ice_server::RTCIceServer;
use webrtc::peer_connection::RTCPeerConnection;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;

/// How long [`NativeWebrtcPeer::open_pipe`] waits for the channel to open (offerer) or arrive
/// (answerer) before giving up — generous versus a real ICE/DTLS/SCTP bring-up (sub-second on a LAN).
const OPEN_TIMEOUT: Duration = Duration::from_secs(20);

/// Max bytes per SCTP message on a write. `webrtc-sctp`'s default max message size is 64 KiB; 16 KiB
/// leaves comfortable headroom, and larger protocol frames are simply split across messages (the
/// reader reassembles them — message boundaries don't matter to a length-prefixed byte stream).
const MAX_SCTP_MESSAGE: usize = 16 * 1024;

/// The process-global tokio runtime that drives every native WebRTC peer. It must outlive the peers
/// (its worker threads run their ICE/DTLS/SCTP tasks), so it is created once and never dropped.
/// Multi-threaded so those tasks make progress *between* the `block_on` signaling/forwarding steps.
fn rt() -> &'static tokio::runtime::Runtime {
    static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
    RT.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .expect("build webrtc tokio runtime")
    })
}

/// Google's public STUN — each peer discovers its server-reflexive (public) address so the pair works
/// across most NAT. The default when no TURN is configured.
const DEFAULT_STUN: &str = "stun:stun.l.google.com:19302";

/// One ICE server: a STUN server (no credentials) or a TURN server (with `username`/`credential`).
#[derive(Clone, Default, Debug)]
pub struct IceServer {
    pub urls: Vec<String>,
    pub username: String,
    pub credential: String,
}

impl IceServer {
    /// A credential-less STUN server.
    pub fn stun(url: impl Into<String>) -> Self {
        Self {
            urls: vec![url.into()],
            ..Default::default()
        }
    }
    /// A long-term-credential TURN server (the relay fallback for symmetric NAT).
    pub fn turn(
        url: impl Into<String>,
        username: impl Into<String>,
        credential: impl Into<String>,
    ) -> Self {
        Self {
            urls: vec![url.into()],
            username: username.into(),
            credential: credential.into(),
        }
    }
}

/// ICE configuration for a WebRTC peer: which STUN/TURN servers to use, and whether to force
/// **relay-only** (ignore host/srflx candidates, pair through TURN — the path symmetric NAT needs).
#[derive(Clone, Default, Debug)]
pub struct WebrtcConfig {
    pub ice_servers: Vec<IceServer>,
    pub relay_only: bool,
}

impl WebrtcConfig {
    /// No ICE servers, host candidates only (LAN/loopback) — the hermetic-test config: fast, no
    /// external network, no relay.
    pub fn hermetic() -> Self {
        Self::default()
    }

    /// Production config from the environment: Google STUN always, plus a TURN server when
    /// `OSFKB_WEBRTC_TURN_URL` (and optional `OSFKB_WEBRTC_TURN_USER`/`OSFKB_WEBRTC_TURN_PASS`) is set.
    /// STUN covers most NAT; TURN is the relay fallback for symmetric NAT. `OSFKB_WEBRTC_RELAY_ONLY=1`
    /// forces relay-only ICE (exclude host/srflx) — used to *prove* the media crossed the TURN relay.
    pub fn from_env() -> Self {
        let mut ice_servers = vec![IceServer::stun(DEFAULT_STUN)];
        if let Ok(url) = std::env::var("OSFKB_WEBRTC_TURN_URL")
            && !url.is_empty()
        {
            ice_servers.push(IceServer::turn(
                url,
                std::env::var("OSFKB_WEBRTC_TURN_USER").unwrap_or_default(),
                std::env::var("OSFKB_WEBRTC_TURN_PASS").unwrap_or_default(),
            ));
        }
        Self {
            ice_servers,
            relay_only: std::env::var("OSFKB_WEBRTC_RELAY_ONLY")
                .is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true")),
        }
    }
}

fn configuration(cfg: &WebrtcConfig) -> RTCConfiguration {
    RTCConfiguration {
        ice_servers: cfg
            .ice_servers
            .iter()
            .map(|s| RTCIceServer {
                urls: s.urls.clone(),
                username: s.username.clone(),
                credential: s.credential.clone(),
            })
            .collect(),
        ice_transport_policy: if cfg.relay_only {
            RTCIceTransportPolicy::Relay
        } else {
            RTCIceTransportPolicy::default()
        },
        ..Default::default()
    }
}

/// The inbound-byte sender, in a slot the connection-state handler can clear on
/// disconnect/failure/close so the reader ([`WebrtcPipe::read`]) sees EOF instead of blocking forever.
type ByteTxSlot = Arc<Mutex<Option<Sender<Vec<u8>>>>>;

/// Install a rustls process-level `CryptoProvider` exactly once. webrtc-rs's DTLS uses rustls 0.23,
/// which panics on first use if no provider is installed and its crypto features are ambiguous — the
/// case for the lean worker binary. Idempotent: a second call (or a provider already installed by the
/// wider process) is a harmless no-op.
fn ensure_crypto_provider() {
    static INSTALL: std::sync::Once = std::sync::Once::new();
    INSTALL.call_once(|| {
        let _ = rustls::crypto::ring::default_provider().install_default();
    });
}

async fn new_peer_connection(
    cfg: &WebrtcConfig,
    tx_slot: ByteTxSlot,
) -> Result<Arc<RTCPeerConnection>, String> {
    ensure_crypto_provider();
    let api = APIBuilder::new().build();
    let pc = api
        .new_peer_connection(configuration(cfg))
        .await
        .map_err(|e| format!("new peer connection: {e}"))?;
    // Operator-visible connection lifecycle AND EOF plumbing: on a terminal state, drop the byte
    // sender so a peer reading the channel unblocks with EOF (the WebRTC analogue of a socket close).
    pc.on_peer_connection_state_change(Box::new(move |state| {
        let tx_slot = Arc::clone(&tx_slot);
        Box::pin(async move {
            eprintln!("webrtc: peer connection state {state}");
            if matches!(
                state,
                RTCPeerConnectionState::Disconnected
                    | RTCPeerConnectionState::Failed
                    | RTCPeerConnectionState::Closed
            ) {
                *tx_slot.lock().expect("byte tx slot poisoned") = None;
            }
        })
    }));
    // Opt-in ICE diagnostics (`OSFKB_WEBRTC_ICE_DEBUG=1`): log every gathered local candidate (so we
    // can see whether a `typ relay` candidate is produced) and the ICE-transport state transitions.
    if std::env::var_os("OSFKB_WEBRTC_ICE_DEBUG").is_some() {
        pc.on_ice_candidate(Box::new(|cand| {
            Box::pin(async move {
                if let Some(c) = cand
                    && let Ok(init) = c.to_json()
                {
                    eprintln!("webrtc ICE local candidate: {}", init.candidate);
                }
            })
        }));
        pc.on_ice_connection_state_change(Box::new(|state| {
            Box::pin(async move { eprintln!("webrtc ICE connection state: {state}") })
        }));
    }
    Ok(Arc::new(pc))
}

/// Arm a data channel's inbound path: every message pushes its bytes to the byte sender (while it is
/// still live — the connection-state handler clears it on teardown, which EOFs the reader).
fn wire_inbound(dc: &RTCDataChannel, tx_slot: ByteTxSlot) {
    dc.on_message(Box::new(move |msg: DataChannelMessage| {
        let tx_slot = Arc::clone(&tx_slot);
        Box::pin(async move {
            if let Some(tx) = tx_slot.lock().expect("byte tx slot poisoned").as_ref() {
                let _ = tx.send(msg.data.to_vec());
            }
        })
    }));
}

/// A coordinator-signalled native WebRTC peer. After the handshake ([`make_offer`] →
/// [`accept_offer`] → [`accept_answer`]) opens the channel, [`open_pipe`] converts it into a
/// [`WebrtcPipe`] — the byte pipe the fleet protocol rides. The offerer owns the channel it created;
/// the answerer receives one via `on_data_channel`. The peer must live until `open_pipe` is called
/// (the runtime keeps the async events firing).
///
/// [`make_offer`]: NativeWebrtcPeer::make_offer
/// [`accept_offer`]: NativeWebrtcPeer::accept_offer
/// [`accept_answer`]: NativeWebrtcPeer::accept_answer
/// [`open_pipe`]: NativeWebrtcPeer::open_pipe
pub struct NativeWebrtcPeer {
    pc: Arc<RTCPeerConnection>,
    /// Offerer: the channel it created (`Some`). Answerer: `None` — its channel arrives via `incoming`.
    dc: Option<Arc<RTCDataChannel>>,
    /// Answerer only: the channel delivered by `on_data_channel`.
    incoming: Option<tokio::sync::mpsc::UnboundedReceiver<Arc<RTCDataChannel>>>,
    /// Inbound bytes from the channel's `on_message` (both roles).
    bytes_rx: Option<Receiver<Vec<u8>>>,
    /// Offerer only: fires when the channel opens (so `open_pipe` waits before the first send).
    open_rx: Option<tokio::sync::mpsc::UnboundedReceiver<()>>,
}

impl NativeWebrtcPeer {
    /// Offerer side. Create the data channel and return the full local SDP once ICE gathering
    /// completes (non-trickle).
    pub async fn make_offer(cfg: &WebrtcConfig) -> Result<(Self, String), String> {
        let (bytes_tx, bytes_rx) = std::sync::mpsc::channel();
        let tx_slot: ByteTxSlot = Arc::new(Mutex::new(Some(bytes_tx)));
        let pc = new_peer_connection(cfg, Arc::clone(&tx_slot)).await?;

        let (open_tx, open_rx) = tokio::sync::mpsc::unbounded_channel();
        let dc = pc
            .create_data_channel("lfm2", None)
            .await
            .map_err(|e| format!("create_data_channel: {e}"))?;
        wire_inbound(&dc, tx_slot);
        // Signal `open_pipe` when the SCTP stream is ready, so the first frame's `send` won't race it.
        dc.on_open(Box::new(move || {
            Box::pin(async move {
                let _ = open_tx.send(());
            })
        }));

        let offer = pc
            .create_offer(None)
            .await
            .map_err(|e| format!("create_offer: {e}"))?;
        let mut gather = pc.gathering_complete_promise().await;
        pc.set_local_description(offer)
            .await
            .map_err(|e| format!("set_local(offer): {e}"))?;
        let _ = gather.recv().await;
        let sdp = pc
            .local_description()
            .await
            .ok_or("no local description after gathering")?
            .sdp;
        Ok((
            Self {
                pc,
                dc: Some(dc),
                incoming: None,
                bytes_rx: Some(bytes_rx),
                open_rx: Some(open_rx),
            },
            sdp,
        ))
    }

    /// Answerer side. Accept the remote `offer_sdp` and return the full answer SDP once ICE gathering
    /// completes. The channel arrives asynchronously via `on_data_channel`.
    pub async fn accept_offer(
        offer_sdp: &str,
        cfg: &WebrtcConfig,
    ) -> Result<(Self, String), String> {
        let (bytes_tx, bytes_rx) = std::sync::mpsc::channel();
        let tx_slot: ByteTxSlot = Arc::new(Mutex::new(Some(bytes_tx)));
        let pc = new_peer_connection(cfg, Arc::clone(&tx_slot)).await?;

        let (dc_tx, dc_rx) = tokio::sync::mpsc::unbounded_channel();
        pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
            let tx_slot = Arc::clone(&tx_slot);
            let dc_tx = dc_tx.clone();
            Box::pin(async move {
                eprintln!("webrtc data channel received (answerer)");
                // Arm inbound bytes immediately; the reader never needs `open` — messages queue.
                wire_inbound(&dc, tx_slot);
                let _ = dc_tx.send(dc);
            })
        }));

        pc.set_remote_description(
            RTCSessionDescription::offer(offer_sdp.to_owned())
                .map_err(|e| format!("parse offer: {e}"))?,
        )
        .await
        .map_err(|e| format!("set_remote(offer): {e}"))?;
        let answer = pc
            .create_answer(None)
            .await
            .map_err(|e| format!("create_answer: {e}"))?;
        let mut gather = pc.gathering_complete_promise().await;
        pc.set_local_description(answer)
            .await
            .map_err(|e| format!("set_local(answer): {e}"))?;
        let _ = gather.recv().await;
        let sdp = pc
            .local_description()
            .await
            .ok_or("no local description after gathering")?
            .sdp;
        Ok((
            Self {
                pc,
                dc: None,
                incoming: Some(dc_rx),
                bytes_rx: Some(bytes_rx),
                open_rx: None,
            },
            sdp,
        ))
    }

    /// Offerer side, final step: apply the remote `answer_sdp`. ICE connectivity checks start and the
    /// channel opens shortly after.
    pub async fn accept_answer(&self, answer_sdp: &str) -> Result<(), String> {
        self.pc
            .set_remote_description(
                RTCSessionDescription::answer(answer_sdp.to_owned())
                    .map_err(|e| format!("parse answer: {e}"))?,
            )
            .await
            .map_err(|e| format!("set_remote(answer): {e}"))?;
        Ok(())
    }

    /// Consume the peer and return the open channel as a byte pipe. The offerer waits for its channel
    /// to open (so the first `send` succeeds); the answerer waits for the channel to arrive. The
    /// returned pipe owns the peer connection, so the channel lives exactly as long as the pipe.
    pub async fn open_pipe(mut self) -> Result<WebrtcPipe, String> {
        let dc = match self.dc.take() {
            Some(dc) => {
                // Offerer: wait for `on_open` so the SCTP stream is ready for the first frame.
                let mut open_rx = self.open_rx.take().ok_or("offerer missing open signal")?;
                match tokio::time::timeout(OPEN_TIMEOUT, open_rx.recv()).await {
                    Ok(Some(())) => dc,
                    _ => return Err("webrtc channel did not open in time".into()),
                }
            }
            None => {
                // Answerer: wait for `on_data_channel` to deliver the channel.
                let incoming = self.incoming.as_mut().ok_or("answerer missing incoming")?;
                match tokio::time::timeout(OPEN_TIMEOUT, incoming.recv()).await {
                    Ok(Some(dc)) => dc,
                    _ => return Err("inbound data channel never arrived".into()),
                }
            }
        };
        let bytes_rx = self.bytes_rx.take().ok_or("pipe already taken")?;
        Ok(WebrtcPipe {
            _pc: self.pc,
            dc,
            bytes_rx,
            buf: Vec::new(),
            pos: 0,
        })
    }
}

/// A WebRTC data channel presented as a blocking byte stream — the transport-agnostic `Conn` shape
/// (mirrors the WebSocket `WsPipe`). One protocol message is one SCTP `send` (write-through, the
/// WebRTC analogue of TCP_NODELAY); reads drain inbound messages in arrival order. `send`/`recv` are
/// driven by the module runtime, so **these methods must be called from a non-runtime thread** —
/// which every `serve_conn` thread is.
pub struct WebrtcPipe {
    /// Keeps the peer connection (hence the channel + its callbacks) alive for the pipe's lifetime.
    _pc: Arc<RTCPeerConnection>,
    dc: Arc<RTCDataChannel>,
    bytes_rx: Receiver<Vec<u8>>,
    buf: Vec<u8>,
    pos: usize,
}

impl std::io::Read for WebrtcPipe {
    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
        while self.pos >= self.buf.len() {
            match self.bytes_rx.recv() {
                Ok(b) => {
                    self.buf = b;
                    self.pos = 0;
                }
                // All senders dropped (peer disconnected/failed/closed) ⇒ end of stream.
                Err(_) => return Ok(0),
            }
        }
        let n = out.len().min(self.buf.len() - self.pos);
        out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]);
        self.pos += n;
        Ok(n)
    }
}

impl std::io::Write for WebrtcPipe {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        // SCTP caps a single message at `webrtc-sctp`'s `DEFAULT_MAX_MESSAGE_SIZE` (64 KiB), so a
        // protocol frame larger than that (a micro-batched hidden hop, a shipped chunk) is split into
        // sub-limit messages. The reader ([`WebrtcPipe::read`]) reassembles across messages — the
        // protocol is a length-prefixed byte STREAM, so SCTP message boundaries are invisible to it.
        for chunk in buf.chunks(MAX_SCTP_MESSAGE) {
            let data = Bytes::copy_from_slice(chunk);
            rt().block_on(self.dc.send(&data))
                .map_err(std::io::Error::other)?;
        }
        Ok(buf.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

/// Establish a local (no signaling server) WebRTC data channel and return **both** ends as byte
/// pipes. The native analog of the browser P0 loopback: it runs the full offer/answer/ICE dance
/// locally and proves the [`WebrtcPipe`] end-to-end without a coordinator. `WebrtcConfig::hermetic()`
/// ⇒ host candidates only; a config with a TURN server + `relay_only` forces the pair through that
/// relay (the symmetric-NAT path).
pub fn loopback_pipes(cfg: &WebrtcConfig) -> Result<(WebrtcPipe, WebrtcPipe), String> {
    rt().block_on(async {
        let (offerer, offer) = NativeWebrtcPeer::make_offer(cfg).await?;
        let (answerer, answer) = NativeWebrtcPeer::accept_offer(&offer, cfg).await?;
        offerer.accept_answer(&answer).await?;
        // Open both ends concurrently: the offerer waits for `on_open`, the answerer for the channel.
        let (op, ap) = tokio::join!(offerer.open_pipe(), answerer.open_pipe());
        Ok((op?, ap?))
    })
}

// ---- in-process TURN server (dev/test utility for the relay-fallback path) -----------------------

/// Long-term-credential auth for [`LocalTurnServer`]: one fixed user whose key was precomputed.
struct StaticTurnAuth {
    user: String,
    key: Vec<u8>,
}

impl AuthHandler for StaticTurnAuth {
    fn auth_handle(
        &self,
        username: &str,
        _realm: &str,
        _src_addr: std::net::SocketAddr,
    ) -> Result<Vec<u8>, turn::Error> {
        if username == self.user {
            Ok(self.key.clone())
        } else {
            Err(turn::Error::ErrFakeErr)
        }
    }
}

/// A TURN server running in-process on the module runtime — a dev/test utility for exercising the
/// relay fallback (symmetric NAT) without a public TURN server. Keep it alive for the peers' lifetime;
/// [`close`](LocalTurnServer::close) shuts it down.
pub struct LocalTurnServer {
    server: Server,
    /// `turn:host:port` URL a peer uses to reach this relay (with `username`/`credential`).
    pub url: String,
    pub username: String,
    pub credential: String,
}

impl LocalTurnServer {
    /// The [`IceServer`] a peer should be configured with to relay through this server.
    pub fn ice_server(&self) -> IceServer {
        IceServer::turn(&self.url, &self.username, &self.credential)
    }
    /// Shut the server down (its UDP listener + allocation tasks).
    pub fn close(self) {
        let _ = rt().block_on(self.server.close());
    }
}

/// Start an in-process TURN server bound to loopback and return a handle. Long-term-credential auth
/// (fixed user/pass), relay address `127.0.0.1`. Used by the relay-fallback proof: two peers built
/// `relay_only` with this server's [`IceServer`] can only pair by relaying through it.
pub fn spawn_local_turn_server() -> Result<LocalTurnServer, String> {
    rt().block_on(async {
        let (user, pass, realm) = ("lfm2", "lfm2-turn-pass", "lfm2");
        let socket = Arc::new(
            tokio::net::UdpSocket::bind("127.0.0.1:0")
                .await
                .map_err(|e| format!("turn bind: {e}"))?,
        );
        let port = socket
            .local_addr()
            .map_err(|e| format!("turn addr: {e}"))?
            .port();
        let key = generate_auth_key(user, realm, pass);
        let server = Server::new(ServerConfig {
            conn_configs: vec![ConnConfig {
                conn: socket,
                relay_addr_generator: Box::new(RelayAddressGeneratorStatic {
                    relay_address: std::net::IpAddr::from([127, 0, 0, 1]),
                    address: "0.0.0.0".to_owned(),
                    net: Arc::new(webrtc_util::vnet::net::Net::new(None)),
                }),
            }],
            realm: realm.to_owned(),
            auth_handler: Arc::new(StaticTurnAuth {
                user: user.to_owned(),
                key,
            }),
            channel_bind_timeout: Duration::from_secs(0),
            alloc_close_notify: None,
        })
        .await
        .map_err(|e| format!("turn server: {e}"))?;
        Ok(LocalTurnServer {
            server,
            url: format!("turn:127.0.0.1:{port}"),
            username: user.to_owned(),
            credential: pass.to_owned(),
        })
    })
}

// ---- synchronous bridge for the `serve_conn` OP_SIGNAL arm (each step drives one async op) --------

/// `OP_SIGNAL`/`SIG_OFFER`: make an offer, store the peer, return its SDP (status 0) or the error.
pub fn do_offer(peer: &mut Option<NativeWebrtcPeer>) -> (u32, Vec<u8>) {
    match rt().block_on(NativeWebrtcPeer::make_offer(&WebrtcConfig::from_env())) {
        Ok((p, sdp)) => {
            *peer = Some(p);
            eprintln!("webrtc: made offer");
            (0, sdp.into_bytes())
        }
        Err(e) => (1, e.into_bytes()),
    }
}

/// `OP_SIGNAL`/`SIG_ANSWER`: accept the relayed `offer`, store the peer, return its answer SDP.
pub fn do_answer(peer: &mut Option<NativeWebrtcPeer>, offer: &[u8]) -> (u32, Vec<u8>) {
    let offer = String::from_utf8_lossy(offer).into_owned();
    match rt().block_on(NativeWebrtcPeer::accept_offer(
        &offer,
        &WebrtcConfig::from_env(),
    )) {
        Ok((p, sdp)) => {
            *peer = Some(p);
            eprintln!("webrtc: made answer");
            (0, sdp.into_bytes())
        }
        Err(e) => (1, e.into_bytes()),
    }
}

/// `OP_SIGNAL`/`SIG_FINISH`: apply the relayed `answer` to the pending offerer peer (channel opens).
pub fn do_finish(peer: &Option<NativeWebrtcPeer>, answer: &[u8]) -> (u32, Vec<u8>) {
    let answer = String::from_utf8_lossy(answer).into_owned();
    match peer.as_ref() {
        Some(p) => match rt().block_on(p.accept_answer(&answer)) {
            Ok(()) => {
                eprintln!("webrtc: applied answer — channel establishing");
                (0, Vec::new())
            }
            Err(e) => (1, e.into_bytes()),
        },
        None => (1, b"no pending webrtc peer for finish".to_vec()),
    }
}

/// Take the pending peer and open its byte pipe (blocking on the module runtime). The serving side
/// of `OP_RTC_SERVE`/`OP_RTC_CHAIN`: after this, the caller owns the channel as a [`WebrtcPipe`].
pub fn open_pipe_blocking(peer: Option<NativeWebrtcPeer>) -> Result<WebrtcPipe, String> {
    let peer = peer.ok_or("no pending webrtc peer to open")?;
    rt().block_on(peer.open_pipe())
}