openrtc 0.2.1

//! Test Harness Binary for pluto-rtc
//!
//! A lightweight HTTP server that wraps `openrtc::Client` to enable
//! cross-environment parity testing. The vitest orchestrator spawns this
//! binary, then controls it via JSON HTTP calls.
//!
//! Usage: cargo run --bin test-harness -- --port 9100 --api-key pk_live_...
//!
//! Endpoints:
//!   GET  /status                    → { nodeId, ready }
//!   POST /auth                      → { token, userId } → { ok }
//!   POST /native/device_identity/init → { preferredName? } → { ok }
//!   POST /presence/update           → { userId, name, ticket, metadata? } → { ok }
//!   POST /presence/offline          → { userId } → { ok }
//!   POST /presence/auto_connect     → { userId, localDeviceId } → { ok }
//!   POST /presence/start_loop       → { userId, deviceName, ticket } → { ok }
//!   POST /devices/search            → { userId, excludeNodeId? } → { devices }
//!   POST /devices/status            → { userId } → { devices }
//!   POST /devices/delete            → { userId, deviceId } → { ok }
//!   POST /rooms/create              → { roomId?, userId, ticket, nodeId, maxMembers? } → { roomId }
//!   POST /rooms/join                → { roomId, userId, ticket, nodeId } → { members }
//!   POST /rooms/members             → { roomId, nodeId } → { members }
//!   POST /rooms/leave               → { roomId, nodeId } → { ok }
//!   POST /signaling/send            → { targetId, payload, state? } → { docId }
//!   POST /signaling/create_session  → { session } → { ok }
//!   POST /signaling/subscribe_sessions → { localDeviceId, waitMs? } → { sessions }
//!   POST /iroh/init                 → { secretKey?, routerMode? } → { nodeId }
//!   GET  /iroh/node_addr            → { nodeAddr }
//!   POST /iroh/node_addr_with_token → { scope, maxConnections? } → { nodeAddr }
//!   POST /session/revoke_by_scope   → { scope } → { affectedConnectionIds }
//!   POST /iroh/connect              → { ticket } → { ok }
//!   POST /iroh/connect_ticket       → { ticket, deviceId? } → { connection }
//!   POST /iroh/disconnect           → { endpointId } → { ok }
//!   POST /iroh/is_connected         → { endpointId } → { connected }
//!   POST /iroh/wait_connected       → { endpointId, timeoutMs? } → { connected }
//!   GET  /iroh/connections          → { connections }
//!   POST /iroh/peer_session         → { id } → { session }
//!   GET  /iroh/peer_sessions        → { sessions }
//!   POST /iroh/wait_settled         → { id, timeoutMs? } → { session }
//!   POST /iroh/open_bi              → { endpointId } → { streamId }
//!   POST /iroh/send                 → { streamId, data } → { ok }
//!   POST /iroh/recv                 → { streamId, timeoutMs? } → { data }
//!   POST /iroh/close_stream         → { streamId } → { ok }
//!   POST /iroh/heartbeat_ping       → { endpointId, timeoutMs? } → { ok, response?, rttMs }
//!   POST /iroh/send_explicit_data   → { endpointId, filename, mimeType?, transferId?, payloadBase64 } → { ok }
//!   POST /iroh/send_explicit_file   → { endpointId, filePath, transferId? } → { ok }
//!   POST /transport/send_moq_peer_data → { endpointId, payloadBase64 } → { ok }
//!   GET  /iroh/received_explicit    → { transfers }
//!   POST /iroh/received_explicit/clear → { ok }
//!   GET  /iroh/received_peer_data   → { messages }
//!   POST /iroh/received_peer_data/clear → { ok }
//!   POST /transport/config          → { transports } → { ok }
//!   GET  /transport/config          → { transports }
//!   POST /transport/webrtc_state    → { id } → { featureEnabled, configEnabled, connectionId?, state? }
//!   POST /transport/request_webrtc  → { id, remoteNodeId?, reason? } → { requested }
//!   POST /transport/moq_state       → { id } → { featureEnabled, configEnabled, connectionId?, state?, relayUrl? }
//!   POST /transport/request_moq     → { id, remoteNodeId?, reason? } → { requested }
//!   POST /transport/repair_moq      → { id, remoteNodeId?, reason? } → { requested }

use anyhow::Result;
use axum::{
    extract::State,
    http::StatusCode,
    routing::{get, post},
    Json, Router,
};
use base64::Engine as _;
use clap::Parser;
use openrtc::application_crypto_streams::{PeerRecvStream, PeerSendStream};
use openrtc::client::Client;
use openrtc::native_node::IncomingStreamType;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::RwLock;
use tower_http::cors::CorsLayer;

#[derive(Parser, Debug)]
#[command(
    name = "test-harness",
    about = "pluto-rtc cross-environment test harness"
)]
struct Cli {
    #[arg(short, long, default_value = "9100")]
    port: u16,

    #[arg(long, default_value = "pluto-rtc-prod")]
    project_id: String,

    #[arg(
        long,
        default_value = "pk_test_0000000000000000000000000000000000000000"
    )]
    api_key: String,
}

struct StreamHandle {
    send: openrtc::application_crypto_streams::PeerSendStream,
    recv: openrtc::application_crypto_streams::PeerRecvStream,
}

struct AppState {
    client: Arc<Client>,
    user_id: RwLock<Option<String>>,
    auth_token: Arc<std::sync::RwLock<Option<String>>>,
    native_device_base_dir: PathBuf,
    ready: RwLock<bool>,
    heartbeat_responder_started: RwLock<bool>,
    streams: RwLock<HashMap<String, StreamHandle>>,
    next_stream_id: RwLock<u64>,
    received_explicit: RwLock<Vec<ReceivedExplicitTransfer>>,
    received_peer_data: RwLock<Vec<ReceivedPeerData>>,
}

// --- Request/Response Types ---

#[derive(Deserialize)]
struct AuthRequest {
    token: String,
    #[serde(rename = "userId")]
    user_id: String,
}

#[derive(Deserialize)]
struct PresenceUpdateRequest {
    #[serde(rename = "userId")]
    user_id: String,
    name: String,
    ticket: String,
    metadata: Option<String>,
}

#[derive(Deserialize)]
struct InitNativeDeviceIdentityRequest {
    #[serde(rename = "preferredName")]
    preferred_name: Option<String>,
}

#[derive(Deserialize)]
struct PresenceOfflineRequest {
    #[serde(rename = "userId")]
    user_id: String,
}

#[derive(Deserialize)]
struct DeviceSearchRequest {
    #[serde(rename = "userId")]
    user_id: String,
    #[serde(rename = "excludeNodeId")]
    exclude_node_id: Option<String>,
}

#[derive(Deserialize)]
struct DeviceStatusRequest {
    #[serde(rename = "userId")]
    user_id: String,
}

#[derive(Deserialize)]
struct DeleteDeviceRequest {
    #[serde(rename = "userId")]
    user_id: String,
    #[serde(rename = "deviceId")]
    device_id: String,
}

#[derive(Deserialize)]
struct CreateRoomRequest {
    #[serde(rename = "roomId")]
    room_id: Option<String>,
    #[serde(rename = "userId")]
    user_id: String,
    ticket: String,
    #[serde(rename = "nodeId")]
    node_id: String,
    #[serde(rename = "maxMembers")]
    max_members: Option<u32>,
}

#[derive(Deserialize)]
struct JoinRoomRequest {
    #[serde(rename = "roomId")]
    room_id: String,
    #[serde(rename = "userId")]
    user_id: String,
    ticket: String,
    #[serde(rename = "nodeId")]
    node_id: String,
}

#[derive(Deserialize)]
struct RoomMembersRequest {
    #[serde(rename = "roomId")]
    room_id: String,
    #[serde(rename = "nodeId")]
    node_id: String,
}

#[derive(Serialize)]
struct RoomIdResponse {
    #[serde(rename = "roomId")]
    room_id: String,
}

#[derive(Serialize)]
struct RoomMembersResponse {
    members: Vec<
        openrtc::firebase::models::FirestoreDocument<openrtc::firebase::models::RoomMemberFields>,
    >,
}

#[derive(Deserialize)]
struct SignalingSendRequest {
    #[serde(rename = "targetId")]
    target_id: String,
    payload: String,
    state: Option<String>,
}

#[derive(Deserialize)]
struct IrohInitRequest {
    /// Optional secret key as array of bytes (e.g. [1, 2, 3, ...])
    #[serde(rename = "secretKey")]
    secret_key: Option<Vec<u8>>,
    /// Optional router mode:
    /// - "internal" (default): spawn pluto-rtc internal accept router
    /// - "custom": initialize endpoint without internal router (external router mode)
    #[serde(rename = "routerMode")]
    router_mode: Option<String>,
}

#[derive(Deserialize)]
struct ConnectRequest {
    ticket: String,
}

#[derive(Deserialize)]
struct ManagedConnectRequest {
    ticket: String,
    #[serde(rename = "deviceId")]
    device_id: Option<String>,
}

#[derive(Deserialize)]
struct NodeAddrWithTokenRequest {
    scope: String,
    #[serde(rename = "maxConnections")]
    max_connections: Option<u32>,
}

#[derive(Deserialize)]
struct RevokeTokensByScopeRequest {
    scope: String,
}

#[derive(Serialize)]
struct RevokeTokensResponse {
    #[serde(rename = "affectedConnectionIds")]
    affected_connection_ids: Vec<String>,
}

#[derive(Deserialize)]
struct AutoConnectRequest {
    #[serde(rename = "userId")]
    user_id: String,
    #[serde(rename = "localDeviceId")]
    local_device_id: String,
}

#[derive(Serialize)]
struct OkResponse {
    ok: bool,
}

#[derive(Serialize)]
struct DocIdResponse {
    #[serde(rename = "docId")]
    doc_id: String,
}

#[derive(Serialize)]
struct NodeIdResponse {
    #[serde(rename = "nodeId")]
    node_id: String,
}

#[derive(Serialize)]
struct DevicesResponse {
    devices: Vec<openrtc::signaling::Device>,
}

#[derive(Serialize)]
struct DeviceStatusResponse {
    devices: Vec<openrtc::client::DeviceStatusSnapshot>,
}

#[derive(Serialize)]
struct ConnectionsResponse {
    connections: Vec<openrtc::connection_manager::ConnectionRecord>,
}

#[derive(Serialize)]
struct ManagedConnectionsResponse {
    connections: Vec<openrtc::connection_manager::ConnectionRecord>,
}

#[derive(Serialize)]
struct PeerSessionResponse {
    session: Option<openrtc::client::PeerSessionSnapshot>,
}

#[derive(Serialize)]
struct PeerSessionsResponse {
    sessions: Vec<openrtc::client::PeerSessionSnapshot>,
}

#[derive(Serialize)]
struct NodeAddrResponse {
    #[serde(rename = "nodeAddr")]
    node_addr: String,
}

#[derive(Serialize)]
struct ManagedConnectResponse {
    connection: openrtc::client::ManagedConnectResult,
}

#[derive(Deserialize)]
struct DisconnectRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
}

#[derive(Deserialize)]
struct OpenBiRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
}

#[derive(Deserialize)]
struct PeerSessionRequest {
    id: String,
}

#[derive(Deserialize)]
struct WaitSettledRequest {
    id: String,
    #[serde(rename = "timeoutMs")]
    timeout_ms: Option<u64>,
}

#[derive(Deserialize)]
struct IsConnectedRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
}

#[derive(Deserialize)]
struct WaitConnectedRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
    #[serde(rename = "timeoutMs")]
    timeout_ms: Option<u64>,
}

#[derive(Serialize)]
struct StreamIdResponse {
    #[serde(rename = "streamId")]
    stream_id: String,
}

#[derive(Deserialize)]
struct SendOnStreamRequest {
    #[serde(rename = "streamId")]
    stream_id: String,
    data: String,
}

#[derive(Deserialize)]
struct RecvFromStreamRequest {
    #[serde(rename = "streamId")]
    stream_id: String,
    #[serde(rename = "timeoutMs")]
    timeout_ms: Option<u64>,
}

#[derive(Deserialize)]
struct CloseStreamRequest {
    #[serde(rename = "streamId")]
    stream_id: String,
}

#[derive(Serialize)]
struct RecvResponse {
    data: Option<String>,
}

#[derive(Serialize)]
struct IsConnectedResponse {
    connected: bool,
}

#[derive(Deserialize)]
struct HeartbeatPingRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
    #[serde(rename = "timeoutMs")]
    timeout_ms: Option<u64>,
}

#[derive(Deserialize)]
struct SendPeerDataRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
    #[serde(rename = "payloadBase64")]
    payload_base64: String,
}

#[derive(Serialize)]
struct HeartbeatPingResponse {
    ok: bool,
    response: Option<String>,
    #[serde(rename = "rttMs")]
    rtt_ms: u64,
}

#[derive(Debug, Deserialize)]
struct SendExplicitDataRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
    filename: String,
    #[serde(rename = "mimeType")]
    mime_type: Option<String>,
    #[serde(rename = "transferId")]
    transfer_id: Option<String>,
    #[serde(rename = "payloadBase64")]
    payload_base64: String,
}

#[derive(Debug, Deserialize)]
struct SendExplicitFileRequest {
    #[serde(rename = "peerId", alias = "endpointId", alias = "id")]
    peer_id: String,
    #[serde(rename = "filePath")]
    file_path: String,
    #[serde(rename = "transferId")]
    transfer_id: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
struct ReceivedExplicitTransfer {
    #[serde(rename = "remoteNodeId")]
    remote_node_id: String,
    #[serde(rename = "transferId")]
    transfer_id: String,
    filename: String,
    size: u64,
    #[serde(rename = "sha256Hex")]
    sha256_hex: String,
    #[serde(rename = "mimeType")]
    mime_type: String,
}

#[derive(Serialize)]
struct ReceivedExplicitResponse {
    transfers: Vec<ReceivedExplicitTransfer>,
}

#[derive(Deserialize)]
struct CreateSessionRequest {
    session: openrtc::signaling::SignalingSession,
}

#[derive(Deserialize)]
struct SubscribeSessionsRequest {
    #[serde(rename = "localDeviceId")]
    local_device_id: String,
    #[serde(rename = "waitMs")]
    wait_ms: Option<u64>,
}

#[derive(Serialize)]
struct SessionsResponse {
    sessions: Vec<openrtc::signaling::SignalingSession>,
}

#[derive(Deserialize)]
struct PresenceLoopRequest {
    #[serde(rename = "userId")]
    user_id: String,
    #[serde(rename = "deviceName")]
    device_name: String,
    ticket: String,
}

#[derive(Debug, Deserialize)]
struct TransportConfigRequest {
    transports: openrtc::client::TransportConfig,
}

#[derive(Serialize)]
struct TransportConfigResponse {
    transports: openrtc::client::TransportConfig,
}

#[derive(Debug, Clone, Serialize)]
struct ReceivedPeerData {
    #[serde(rename = "connectionId")]
    connection_id: String,
    #[serde(rename = "remoteNodeId")]
    remote_node_id: Option<String>,
    transport: String,
    #[serde(rename = "payloadBase64")]
    payload_base64: String,
    #[serde(rename = "receivedAtMs")]
    received_at_ms: u64,
}

#[derive(Serialize)]
struct ReceivedPeerDataResponse {
    messages: Vec<ReceivedPeerData>,
}

#[derive(Debug, Deserialize)]
struct TransportWebRtcStateRequest {
    id: String,
}

#[derive(Debug, Deserialize)]
struct TransportWebRtcRequest {
    id: String,
    #[serde(rename = "remoteNodeId")]
    remote_node_id: Option<String>,
    reason: Option<String>,
    role: Option<String>,
}

#[derive(Serialize)]
struct TransportWebRtcRequestResponse {
    requested: bool,
}

#[derive(Serialize)]
struct TransportWebRtcStateResponse {
    #[serde(rename = "featureEnabled")]
    feature_enabled: bool,
    #[serde(rename = "configEnabled")]
    config_enabled: bool,
    #[serde(rename = "connectionId")]
    connection_id: Option<String>,
    state: Option<String>,
}

#[derive(Debug, Deserialize)]
struct TransportMoqStateRequest {
    id: String,
}

#[derive(Debug, Deserialize)]
struct TransportMoqRequest {
    id: String,
    #[serde(rename = "remoteNodeId")]
    remote_node_id: Option<String>,
    reason: Option<String>,
}

#[derive(Serialize)]
struct TransportMoqRequestResponse {
    requested: bool,
}

#[derive(Serialize)]
struct TransportMoqStateResponse {
    #[serde(rename = "featureEnabled")]
    feature_enabled: bool,
    #[serde(rename = "configEnabled")]
    config_enabled: bool,
    #[serde(rename = "connectionId")]
    connection_id: Option<String>,
    state: Option<String>,
    #[serde(rename = "dataReady")]
    data_ready: bool,
    #[serde(rename = "relayUrl")]
    relay_url: Option<String>,
}

async fn write_peer_text_frame(send: &mut PeerSendStream, text: &str) -> anyhow::Result<()> {
    let bytes = text.as_bytes();
    let len = (bytes.len() as u32).to_be_bytes();
    send.write_all(&len).await?;
    send.write_all(bytes).await?;
    Ok(())
}

async fn read_exact_peer(recv: &mut PeerRecvStream, buf: &mut [u8]) -> anyhow::Result<()> {
    let mut offset = 0;
    while offset < buf.len() {
        let read = recv.read(&mut buf[offset..]).await?;
        if read == 0 {
            anyhow::bail!("peer stream ended before {} bytes were read", buf.len());
        }
        offset += read;
    }
    Ok(())
}

async fn read_peer_text_frame(recv: &mut PeerRecvStream) -> anyhow::Result<String> {
    let mut len_buf = [0u8; 4];
    read_exact_peer(recv, &mut len_buf).await?;
    let len = u32::from_be_bytes(len_buf) as usize;
    let mut data_buf = vec![0u8; len];
    read_exact_peer(recv, &mut data_buf).await?;
    Ok(String::from_utf8(data_buf)?)
}

async fn read_channel_envelope_after_magic(recv: &mut PeerRecvStream) -> anyhow::Result<String> {
    let mut header_rest = [0u8; openrtc::stream_metadata::CHANNEL_ENVELOPE_HEADER_BYTES - 1];
    read_exact_peer(recv, &mut header_rest).await?;
    if header_rest[0] != openrtc::stream_metadata::CHANNEL_ENVELOPE_VERSION {
        anyhow::bail!("unsupported channel envelope version {}", header_rest[0]);
    }

    let channel_len = u16::from_be_bytes([header_rest[1], header_rest[2]]) as usize;
    let metadata_len = u16::from_be_bytes([header_rest[3], header_rest[4]]) as usize;
    if channel_len == 0 {
        anyhow::bail!("empty channel envelope id");
    }

    let mut channel_bytes = vec![0u8; channel_len];
    read_exact_peer(recv, &mut channel_bytes).await?;
    if metadata_len > 0 {
        let mut metadata_bytes = vec![0u8; metadata_len];
        read_exact_peer(recv, &mut metadata_bytes).await?;
    }

    Ok(String::from_utf8_lossy(&channel_bytes).trim().to_string())
}

async fn handle_text_or_native_main_frame(
    state: &Arc<AppState>,
    send: &mut PeerSendStream,
    connection_id: &str,
    remote_node_id: &str,
    frame: &[u8],
) -> anyhow::Result<bool> {
    if let Ok(message) = std::str::from_utf8(frame) {
        if let Some(payload) = message.strip_prefix("__pluto_ping__:") {
            write_peer_text_frame(send, &format!("__pluto_pong__:{}", payload)).await?;
            return Ok(true);
        }
        if let Some(payload) = message.strip_prefix("__pluto_echo__:") {
            write_peer_text_frame(send, &format!("__pluto_echo_ack__:{}", payload)).await?;
            return Ok(true);
        }
    }

    if frame.first() == Some(&0x00) {
        state
            .client
            .inspect_incoming_native_main_frame(connection_id, Some(remote_node_id), None, frame)
            .await
            .map_err(|error| anyhow::anyhow!(error))?;
        return Ok(true);
    }

    Ok(false)
}

async fn read_exact_len_prefixed_bytes_with_first(
    recv: &mut PeerRecvStream,
    first: u8,
) -> anyhow::Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    len_buf[0] = first;
    read_exact_peer(recv, &mut len_buf[1..]).await?;
    let len = u32::from_be_bytes(len_buf) as usize;
    let mut data_buf = vec![0u8; len];
    read_exact_peer(recv, &mut data_buf).await?;
    Ok(data_buf)
}

fn deterministic_connection_id(node_id_a: &str, node_id_b: &str) -> String {
    if node_id_a <= node_id_b {
        format!("{}-{}", node_id_a, node_id_b)
    } else {
        format!("{}-{}", node_id_b, node_id_a)
    }
}

async fn read_native_main_frame_after_main_label(
    recv: &mut PeerRecvStream,
) -> anyhow::Result<Vec<u8>> {
    let mut frame_len_buf = [0u8; 4];
    read_exact_peer(recv, &mut frame_len_buf).await?;
    let frame_len = u32::from_be_bytes(frame_len_buf) as usize;
    if frame_len > (1024 * 1024) {
        anyhow::bail!("native main frame too large: {}", frame_len);
    }

    let mut frame_buf = vec![0u8; frame_len];
    read_exact_peer(recv, &mut frame_buf).await?;
    Ok(frame_buf)
}

async fn write_native_main_response(
    send: &mut PeerSendStream,
    message: &openrtc::native_protocol::NativeMainMessage,
) -> anyhow::Result<()> {
    let serialized = serde_json::to_vec(message)?;
    let label = b"main";
    let label_len = (label.len() as u32).to_be_bytes();
    let frame_len = (serialized.len() as u32).to_be_bytes();

    let mut buf = Vec::with_capacity(1 + 4 + label.len() + 4 + serialized.len());
    buf.push(0x00);
    buf.extend_from_slice(&label_len);
    buf.extend_from_slice(label);
    buf.extend_from_slice(&frame_len);
    buf.extend_from_slice(&serialized);

    send.write_all(&buf).await?;
    Ok(())
}

/// Complete SDK-owned token admission in the same order as the production
/// response-writer path. The response must reach the dialer before lifecycle
/// side effects are allowed to replace a transport or restart WebRTC; otherwise
/// a refresh can close this stream while the dialer is waiting for byte zero.
async fn respond_to_session_token_presentation(
    state: &Arc<AppState>,
    send: &mut PeerSendStream,
    connection_id: &str,
    remote_node_id: &str,
    message: &openrtc::native_protocol::NativeMainMessage,
) -> anyhow::Result<()> {
    let token = message
        .presented_session_token()
        .ok_or_else(|| anyhow::anyhow!("session-token-missing-in-presentation"))?;
    let token_payload = message.presented_session_token_payload();
    let claimed_device_id = message.claimed_device_id();
    let was_already_admitted = state
        .client
        .session_token_registry
        .is_session_token_admitted_for_connection(connection_id);
    let response_guard = state
        .client
        .session_token_registry
        .begin_admission_response(connection_id);

    let scope = match state
        .client
        .validate_session_token_for_connection_with_payload(
            &token,
            connection_id,
            token_payload.as_deref(),
        )
        .await
    {
        Ok(scope) => scope,
        Err(error) => {
            let rejection = openrtc::native_protocol::NativeMainMessage::session_token_rejection(
                &error,
                connection_id,
            );
            write_native_main_response(send, &rejection).await?;
            drop(response_guard);
            return Err(anyhow::anyhow!(error));
        }
    };

    let approval = openrtc::native_protocol::NativeMainMessage::session_token_approval(
        Some(&scope),
        connection_id,
    );
    write_native_main_response(send, &approval).await?;
    drop(response_guard);

    // The production desktop runtime keeps this record in its managed registry.
    // The harness owns the equivalent registry setup so later native-main frames
    // resolve the same admitted connection, rather than creating a parallel path.
    state
        .client
        .connection_manager
        .upsert_pending(
            connection_id.to_string(),
            Some(remote_node_id.to_string()),
            None,
            Some(remote_node_id.to_string()),
        )
        .await;
    state
        .client
        .connection_manager
        .set_connected(connection_id, Some(remote_node_id.to_string()))
        .await;
    if let Some(device_id) = claimed_device_id.as_deref() {
        let _ = state
            .client
            .bind_session_admission_authoritative_device_id(connection_id, device_id)
            .await;
    }
    state
        .client
        .run_post_session_token_admission_side_effects(connection_id, !was_already_admitted)
        .await;

    eprintln!(
        "[test-harness][session-token] admission response written before side effects connection_id={} remote_node_id={} scope={}",
        connection_id,
        remote_node_id,
        scope
    );
    Ok(())
}

async fn read_text_frame_after_native_probe(
    recv: &mut PeerRecvStream,
    first: u8,
    probe: [u8; 8],
) -> anyhow::Result<String> {
    let len = u32::from_be_bytes([first, probe[0], probe[1], probe[2]]) as usize;
    if len == 0 {
        return Ok(String::new());
    }

    let already_buffered = &probe[3..];
    let buffered_len = already_buffered.len().min(len);
    let mut data_buf = Vec::with_capacity(len);
    data_buf.extend_from_slice(&already_buffered[..buffered_len]);

    if len > buffered_len {
        let mut remaining = vec![0u8; len - buffered_len];
        read_exact_peer(recv, &mut remaining).await?;
        data_buf.extend_from_slice(&remaining);
    }

    Ok(String::from_utf8(data_buf)?)
}

fn digest_sha256_hex(data: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(data);
    let digest = hasher.finalize();
    hex::encode(digest)
}

#[derive(Debug, Deserialize)]
struct ExplicitHeader {
    #[serde(rename = "transfer_id")]
    transfer_id: Option<String>,
    filename: Option<String>,
    size: Option<u64>,
    #[serde(rename = "mime_type")]
    mime_type: Option<String>,
}

async fn read_explicit_transfer(
    recv: &mut PeerRecvStream,
    remote_node_id: String,
) -> anyhow::Result<ReceivedExplicitTransfer> {
    const MAX_EXPLICIT_PAYLOAD_BYTES: usize = 64 * 1024 * 1024;

    let mut len_buf = [0u8; 4];
    read_exact_peer(recv, &mut len_buf).await?;
    let header_len = u32::from_be_bytes(len_buf) as usize;
    if header_len == 0 {
        anyhow::bail!("explicit transfer header length is zero");
    }
    if header_len > (1024 * 1024) {
        anyhow::bail!("explicit transfer header too large: {}", header_len);
    }

    let mut header_buf = vec![0u8; header_len];
    read_exact_peer(recv, &mut header_buf).await?;
    let header: ExplicitHeader = serde_json::from_slice(&header_buf)?;

    let mut payload = Vec::new();
    let mut buffer = vec![0u8; 16 * 1024];
    loop {
        let read = recv.read(&mut buffer).await?;
        if read == 0 {
            break;
        }
        if payload.len() + read > MAX_EXPLICIT_PAYLOAD_BYTES {
            anyhow::bail!(
                "explicit transfer payload exceeds {} bytes",
                MAX_EXPLICIT_PAYLOAD_BYTES
            );
        }
        payload.extend_from_slice(&buffer[..read]);
    }

    let transfer_id = header
        .transfer_id
        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
    let filename = header
        .filename
        .unwrap_or_else(|| "received.bin".to_string());
    let mime_type = header
        .mime_type
        .unwrap_or_else(|| "application/octet-stream".to_string());
    let size = header.size.unwrap_or(payload.len() as u64);

    Ok(ReceivedExplicitTransfer {
        remote_node_id,
        transfer_id,
        filename,
        size,
        sha256_hex: digest_sha256_hex(&payload),
        mime_type,
    })
}

async fn write_explicit_transfer(
    client: Arc<Client>,
    peer_id: String,
    filename: String,
    mime_type: String,
    transfer_id: Option<String>,
    payload: Vec<u8>,
) -> anyhow::Result<()> {
    let transfer_id = transfer_id.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
    let header = serde_json::json!({
        "transfer_id": transfer_id,
        "filename": filename,
        "size": payload.len() as u64,
        "mime_type": mime_type
    });
    let header_bytes = serde_json::to_vec(&header)?;
    let header_len = (header_bytes.len() as u32).to_be_bytes();

    let mut body = Vec::with_capacity(header_len.len() + header_bytes.len() + payload.len());
    body.extend_from_slice(&header_len);
    body.extend_from_slice(&header_bytes);
    body.extend_from_slice(&payload);

    let mut frame = Vec::with_capacity(1 + body.len());
    frame.push(0x02);
    frame.extend_from_slice(&body);

    println!(
        "[HARNESS-RUST] write_explicit_transfer starting to peer_id={}",
        peer_id
    );

    // Explicit-file transfer is a stream protocol with an authenticated receiver
    // on the base Iroh path. Native WebRTC currently has a generic datachannel
    // send path but no explicit-file receiver/ack contract, so a successful
    // datachannel write is not proof of delivery. Keep this harness on the same
    // production-faithful explicit stream until WebRTC explicit transfer grows a
    // real bidirectional protocol.
    let (_connection_id, remote_node_id, mut send) = match client
        .open_peer_bi_explicit_file_sender(&peer_id, Some(5_000))
        .await
    {
        Ok(res) => res,
        Err(e) => {
            println!(
                "[HARNESS-RUST] write_explicit_transfer failed to open explicit-file bi stream: {}",
                e
            );
            return Err(e);
        }
    };
    println!(
        "[HARNESS-RUST] write_explicit_transfer opened bi stream to remote_node_id={}",
        remote_node_id
    );

    send.write_all(&body).await?;
    send.finish()?;
    Ok(())
}

async fn resolve_request_endpoint_id(
    client: &Client,
    peer_id: &str,
) -> Result<iroh::EndpointId, (StatusCode, String)> {
    if let Ok(endpoint_id) = peer_id.parse::<iroh::EndpointId>() {
        return Ok(endpoint_id);
    }

    let snapshot = match client.wait_for_settled_peer(peer_id, Some(5_000)).await {
        Some(snapshot) => Some(snapshot),
        None => client.peer_session(peer_id).await,
    };

    let node_id = snapshot
        .as_ref()
        .and_then(|session| session.node_id.as_deref())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            (
                StatusCode::BAD_REQUEST,
                format!("No active peer transport for '{}'", peer_id),
            )
        })?;

    node_id.parse::<iroh::EndpointId>().map_err(|error| {
        (
            StatusCode::BAD_REQUEST,
            format!("Invalid endpoint ID for '{}': {}", peer_id, error),
        )
    })
}

async fn start_heartbeat_responder_if_needed(state: Arc<AppState>) {
    {
        let mut started = state.heartbeat_responder_started.write().await;
        if *started {
            return;
        }
        *started = true;
    }

    let client = state.client.clone();
    tokio::spawn(async move {
        loop {
            let incoming = match client.incoming_streams().await {
                Ok(streams) => streams,
                Err(error) => {
                    eprintln!(
                        "[heartbeat_responder] incoming stream subscription failed: {}",
                        error
                    );
                    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                    continue;
                }
            };

            while let Ok(incoming_stream) = incoming.recv().await {
                if let IncomingStreamType::Bi(send, mut recv) = incoming_stream.stream {
                    let remote_node_id = incoming_stream.endpoint_id.to_string();
                    let state_for_stream = state.clone();
                    tokio::spawn(async move {
                        let local_node_id = state_for_stream
                            .client
                            .current_node_id()
                            .await
                            .unwrap_or_default();
                        let connection_id =
                            deterministic_connection_id(&local_node_id, &remote_node_id);
                        let application_key = state_for_stream
                            .client
                            .connection_application_crypto_key(&connection_id);

                        let mut raw_first = [0u8; 1];
                        if let Err(error) =
                            tokio::io::AsyncReadExt::read_exact(&mut recv, &mut raw_first).await
                        {
                            eprintln!(
                                "[native_stream_gate] failed reading incoming protocol byte remote_node_id={} connection_id={} error={}",
                                remote_node_id,
                                connection_id,
                                error
                            );
                            return;
                        }

                        if raw_first[0] == 0x02 {
                            if let Err(error) =
                                state_for_stream.client.ensure_native_stream_admitted(
                                    &connection_id,
                                    Some(&remote_node_id),
                                    None,
                                )
                            {
                                eprintln!(
                                    "[explicit_capture] rejecting unauthenticated explicit stream remote_node_id={} connection_id={} error={}",
                                    remote_node_id,
                                    connection_id,
                                    error
                                );
                                if let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>()
                                {
                                    let _ = state_for_stream.client.disconnect(endpoint_id).await;
                                }
                                return;
                            }

                            let mut explicit_recv = match application_key {
                                Some(key) => match PeerRecvStream::encrypted(recv, key) {
                                    Ok(stream) => stream,
                                    Err(error) => {
                                        eprintln!(
                                            "[explicit_capture] failed wrapping encrypted explicit stream remote_node_id={} connection_id={} error={}",
                                            remote_node_id,
                                            connection_id,
                                            error
                                        );
                                        return;
                                    }
                                },
                                None => PeerRecvStream::plain(recv),
                            };
                            match read_explicit_transfer(&mut explicit_recv, remote_node_id.clone())
                                .await
                            {
                                Ok(transfer) => {
                                    state_for_stream
                                        .received_explicit
                                        .write()
                                        .await
                                        .push(transfer);
                                }
                                Err(error) => {
                                    eprintln!(
                                        "[explicit_capture] failed parsing incoming explicit transfer: {}",
                                        error
                                    );
                                }
                            }
                            return;
                        }

                        let (mut send, mut recv, mut pending_first) = match application_key {
                            Some(key) => {
                                let send = PeerSendStream::encrypted(send, key);
                                let recv = match PeerRecvStream::encrypted_with_prefix(
                                    recv, key, &raw_first,
                                ) {
                                    Ok(stream) => stream,
                                    Err(error) => {
                                        eprintln!(
                                            "[native_stream_gate] failed wrapping incoming peer stream remote_node_id={} connection_id={} error={}",
                                            remote_node_id,
                                            connection_id,
                                            error
                                        );
                                        return;
                                    }
                                };
                                (send, recv, None)
                            }
                            None => (
                                PeerSendStream::plain(send),
                                PeerRecvStream::plain(recv),
                                Some(raw_first),
                            ),
                        };

                        loop {
                            let mut first = [0u8; 1];
                            if let Some(buffered_first) = pending_first.take() {
                                first = buffered_first;
                            } else if read_exact_peer(&mut recv, &mut first).await.is_err() {
                                break;
                            }

                            if first[0] == openrtc::stream_metadata::CHANNEL_ENVELOPE_MAGIC {
                                // Scoped-channel streams carry a small envelope before
                                // the app's own frames. The harness should accept that
                                // package contract instead of forcing tests onto raw
                                // unscoped streams.
                                match read_channel_envelope_after_magic(&mut recv).await {
                                    Ok(channel_id) => {
                                        println!(
                                            "[native_stream_gate] accepted channel envelope remote_node_id={} connection_id={} channel_id={}",
                                            remote_node_id, connection_id, channel_id
                                        );
                                        continue;
                                    }
                                    Err(error) => {
                                        eprintln!(
                                            "[native_stream_gate] failed parsing channel envelope remote_node_id={} connection_id={} error={}",
                                            remote_node_id, connection_id, error
                                        );
                                        break;
                                    }
                                }
                            }

                            if first[0] == 0x00 {
                                let mut probe = [0u8; 8];
                                if let Err(error) = read_exact_peer(&mut recv, &mut probe).await {
                                    eprintln!(
                                        "[native_stream_gate] failed probing leading frame remote_node_id={} connection_id={} error={}",
                                        remote_node_id,
                                        connection_id,
                                        error
                                    );
                                    break;
                                }

                                let inspected = if probe[..4] == [0, 0, 0, 4]
                                    && &probe[4..] == b"main"
                                {
                                    match read_native_main_frame_after_main_label(&mut recv).await {
                                        Ok(frame) => {
                                            let parsed =
                                                openrtc::native_protocol::parse_main_frame(&frame);
                                            if let openrtc::native_protocol::ParsedMainFrame::NativeMessage(
                                                message,
                                            ) = &parsed
                                            {
                                                if message.is_session_token_presentation() {
                                                    if let Err(error) = respond_to_session_token_presentation(
                                                        &state_for_stream,
                                                        &mut send,
                                                        &connection_id,
                                                        &remote_node_id,
                                                        message,
                                                    )
                                                    .await
                                                    {
                                                        eprintln!(
                                                            "[native_stream_gate] session-token presentation failed remote_node_id={} connection_id={} error={}",
                                                            remote_node_id,
                                                            connection_id,
                                                            error
                                                        );
                                                        if let Ok(endpoint_id) =
                                                            remote_node_id.parse::<iroh::EndpointId>()
                                                        {
                                                            let _ = state_for_stream
                                                                .client
                                                                .disconnect(endpoint_id)
                                                                .await;
                                                        }
                                                    }
                                                    continue;
                                                }
                                            }
                                            state_for_stream
                                                .client
                                                .inspect_incoming_native_main_frame(
                                                    &connection_id,
                                                    Some(&remote_node_id),
                                                    None,
                                                    &frame,
                                                )
                                                .await
                                        }
                                        Err(error) => Err(error.to_string()),
                                    }
                                } else {
                                    if let Err(error) =
                                        state_for_stream.client.ensure_native_stream_admitted(
                                            &connection_id,
                                            Some(&remote_node_id),
                                            None,
                                        )
                                    {
                                        Err(error)
                                    } else {
                                        let message = match read_text_frame_after_native_probe(
                                            &mut recv, first[0], probe,
                                        )
                                        .await
                                        {
                                            Ok(value) => value,
                                            Err(error) => {
                                                eprintln!(
                                                    "[native_stream_gate] failed parsing probed app frame remote_node_id={} connection_id={} error={}",
                                                    remote_node_id,
                                                    connection_id,
                                                    error
                                                );
                                                break;
                                            }
                                        };
                                        if let Err(error) = handle_text_or_native_main_frame(
                                            &state_for_stream,
                                            &mut send,
                                            &connection_id,
                                            &remote_node_id,
                                            message.as_bytes(),
                                        )
                                        .await
                                        {
                                            eprintln!(
                                                "[native_stream_gate] failed handling probed frame remote_node_id={} connection_id={} error={}",
                                                remote_node_id,
                                                connection_id,
                                                error
                                            );
                                            if let Ok(endpoint_id) =
                                                remote_node_id.parse::<iroh::EndpointId>()
                                            {
                                                let _ = state_for_stream
                                                    .client
                                                    .disconnect(endpoint_id)
                                                    .await;
                                            }
                                            break;
                                        }
                                        continue;
                                    }
                                };

                                if let Err(error) = inspected {
                                    eprintln!(
                                        "[native_stream_gate] rejecting native main frame remote_node_id={} connection_id={} error={}",
                                        remote_node_id,
                                        connection_id,
                                        error
                                    );

                                    if let Ok(endpoint_id) =
                                        remote_node_id.parse::<iroh::EndpointId>()
                                    {
                                        let _ =
                                            state_for_stream.client.disconnect(endpoint_id).await;
                                    }
                                    break;
                                }

                                continue;
                            }

                            if first[0] == 0x02 {
                                if let Err(error) =
                                    state_for_stream.client.ensure_native_stream_admitted(
                                        &connection_id,
                                        Some(&remote_node_id),
                                        None,
                                    )
                                {
                                    eprintln!(
                                        "[explicit_capture] rejecting unauthenticated explicit stream remote_node_id={} connection_id={} error={}",
                                        remote_node_id,
                                        connection_id,
                                        error
                                    );
                                    if let Ok(endpoint_id) =
                                        remote_node_id.parse::<iroh::EndpointId>()
                                    {
                                        let _ =
                                            state_for_stream.client.disconnect(endpoint_id).await;
                                    }
                                    break;
                                }
                                match read_explicit_transfer(&mut recv, remote_node_id.clone())
                                    .await
                                {
                                    Ok(transfer) => {
                                        state_for_stream
                                            .received_explicit
                                            .write()
                                            .await
                                            .push(transfer);
                                    }
                                    Err(error) => {
                                        eprintln!(
                                            "[explicit_capture] failed parsing incoming explicit transfer: {}",
                                            error
                                        );
                                    }
                                }
                                break;
                            }

                            if let Err(error) =
                                state_for_stream.client.ensure_native_stream_admitted(
                                    &connection_id,
                                    Some(&remote_node_id),
                                    None,
                                )
                            {
                                eprintln!(
                                    "[native_stream_gate] rejecting unauthenticated app stream remote_node_id={} connection_id={} error={}",
                                    remote_node_id,
                                    connection_id,
                                    error
                                );
                                if let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>()
                                {
                                    let _ = state_for_stream.client.disconnect(endpoint_id).await;
                                }
                                break;
                            }

                            let frame =
                                match read_exact_len_prefixed_bytes_with_first(&mut recv, first[0])
                                    .await
                                {
                                    Ok(value) => value,
                                    Err(_) => break,
                                };
                            if let Err(error) = handle_text_or_native_main_frame(
                                &state_for_stream,
                                &mut send,
                                &connection_id,
                                &remote_node_id,
                                &frame,
                            )
                            .await
                            {
                                eprintln!(
                                    "[native_stream_gate] failed handling app/native-main frame remote_node_id={} connection_id={} error={}",
                                    remote_node_id,
                                    connection_id,
                                    error
                                );
                                if let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>()
                                {
                                    let _ = state_for_stream.client.disconnect(endpoint_id).await;
                                }
                                break;
                            }
                        }
                    });
                }
            }

            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        }
    });
}

// --- Handlers ---

async fn handle_status(State(state): State<Arc<AppState>>) -> Json<openrtc::client::RuntimeStatus> {
    let mut status = state.client.runtime_status().await;
    status.ready = *state.ready.read().await;
    Json(status)
}

async fn handle_auth(
    State(state): State<Arc<AppState>>,
    Json(req): Json<AuthRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    {
        let mut token = state.auth_token.write().unwrap();
        *token = Some(req.token);
    }
    {
        let mut uid = state.user_id.write().await;
        *uid = Some(req.user_id);
    }
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_init_native_device_identity(
    State(state): State<Arc<AppState>>,
    Json(req): Json<InitNativeDeviceIdentityRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .init_native_device_identity(
            state.native_device_base_dir.clone(),
            req.preferred_name.as_deref(),
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_presence_update(
    State(state): State<Arc<AppState>>,
    Json(req): Json<PresenceUpdateRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    println!(
        "[presence_update] user_id={} name={} ticket={} metadata={}",
        req.user_id,
        req.name,
        req.ticket,
        req.metadata.as_deref().unwrap_or("<none>"),
    );

    if let Err(e) = state
        .client
        .update_presence(
            &req.user_id,
            &req.name,
            &req.ticket,
            req.metadata.as_deref(),
        )
        .await
    {
        eprintln!("[presence_update] update_presence error: {}", e);
        return Err((StatusCode::INTERNAL_SERVER_ERROR, e.to_string()));
    }

    // Immediately query devices for this user to observe what native sees.
    match state.client.search_devices_raw(&req.user_id).await {
        Ok(devices) => {
            let names: Vec<String> = devices.iter().map(|d| d.device_name.clone()).collect();
            println!(
                "[presence_update] search_devices_raw returned {} device(s): [{}]",
                devices.len(),
                names.join(", ")
            );
        }
        Err(e) => {
            eprintln!(
                "[presence_update] search_devices_raw error for user {}: {}",
                req.user_id, e
            );
        }
    }

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_presence_offline(
    State(state): State<Arc<AppState>>,
    Json(req): Json<PresenceOfflineRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .set_offline(&req.user_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_device_search(
    State(state): State<Arc<AppState>>,
    Json(req): Json<DeviceSearchRequest>,
) -> Result<Json<DevicesResponse>, (StatusCode, String)> {
    // Use search_devices_raw to avoid auto-excluding self in tests
    let devices = state
        .client
        .search_devices_raw(&req.user_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let devices = if let Some(exclude) = req.exclude_node_id {
        devices
            .into_iter()
            .filter(|d| d.node_id.as_deref() != Some(&exclude))
            .collect()
    } else {
        devices
    };

    Ok(Json(DevicesResponse { devices }))
}

async fn handle_device_status(
    State(state): State<Arc<AppState>>,
    Json(req): Json<DeviceStatusRequest>,
) -> Result<Json<DeviceStatusResponse>, (StatusCode, String)> {
    let devices = state
        .client
        .devices_with_status(&req.user_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(DeviceStatusResponse { devices }))
}

async fn handle_device_delete(
    State(state): State<Arc<AppState>>,
    Json(req): Json<DeleteDeviceRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .delete_device(&req.user_id, &req.device_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

fn normalize_room_id(value: Option<String>) -> String {
    let trimmed = value.unwrap_or_default().trim().to_uppercase();
    if trimmed.is_empty() {
        format!("ROOM-{}", uuid::Uuid::new_v4().simple())
            .chars()
            .take(18)
            .collect()
    } else {
        trimmed
    }
}

async fn handle_room_create(
    State(state): State<Arc<AppState>>,
    Json(req): Json<CreateRoomRequest>,
) -> Result<Json<RoomIdResponse>, (StatusCode, String)> {
    let room_id = normalize_room_id(req.room_id);
    state
        .client
        .room
        .create_room(
            &room_id,
            &req.user_id,
            &req.ticket,
            &req.node_id,
            state.client.app_tag(),
            req.max_members,
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(RoomIdResponse { room_id }))
}

async fn handle_room_join(
    State(state): State<Arc<AppState>>,
    Json(req): Json<JoinRoomRequest>,
) -> Result<Json<RoomMembersResponse>, (StatusCode, String)> {
    state
        .client
        .room
        .join_room(
            &req.room_id,
            &req.user_id,
            &req.ticket,
            &req.node_id,
            state.client.app_tag(),
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let members = state
        .client
        .room
        .get_members(&req.room_id, &req.node_id, state.client.app_tag())
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(RoomMembersResponse { members }))
}

async fn handle_room_members(
    State(state): State<Arc<AppState>>,
    Json(req): Json<RoomMembersRequest>,
) -> Result<Json<RoomMembersResponse>, (StatusCode, String)> {
    let members = state
        .client
        .room
        .get_members(&req.room_id, &req.node_id, state.client.app_tag())
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(RoomMembersResponse { members }))
}

async fn handle_room_leave(
    State(state): State<Arc<AppState>>,
    Json(req): Json<RoomMembersRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .room
        .leave_room(&req.room_id, &req.node_id, state.client.app_tag())
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_signaling_send(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SignalingSendRequest>,
) -> Result<Json<DocIdResponse>, (StatusCode, String)> {
    let doc_id = state
        .client
        .send_message(&req.target_id, &req.payload, req.state.as_deref(), None)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(DocIdResponse { doc_id }))
}

async fn handle_iroh_init(
    State(state): State<Arc<AppState>>,
    Json(req): Json<IrohInitRequest>,
) -> Result<Json<NodeIdResponse>, (StatusCode, String)> {
    let router_mode = req
        .router_mode
        .as_deref()
        .unwrap_or("internal")
        .trim()
        .to_lowercase();
    println!("[test-harness] /iroh/init router_mode={}", router_mode);

    let node_id = match router_mode.as_str() {
        "internal" | "" => state
            .client
            .init_iroh(req.secret_key, vec![])
            .await
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?,
        "custom" | "no_internal_router" => state
            .client
            .init_iroh_without_internal_router(req.secret_key, vec![])
            .await
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?,
        other => {
            return Err((
                StatusCode::BAD_REQUEST,
                format!(
                    "Invalid routerMode '{}'. Expected one of: internal, custom, no_internal_router",
                    other
                ),
            ));
        }
    };

    {
        let mut ready = state.ready.write().await;
        *ready = true;
    }

    start_heartbeat_responder_if_needed(state.clone()).await;

    Ok(Json(NodeIdResponse { node_id }))
}

async fn handle_iroh_connections(State(state): State<Arc<AppState>>) -> Json<ConnectionsResponse> {
    let connections = state.client.connection_manager.list_all().await;
    Json(ConnectionsResponse { connections })
}

async fn handle_managed_connections(
    State(state): State<Arc<AppState>>,
) -> Json<ManagedConnectionsResponse> {
    Json(ManagedConnectionsResponse {
        connections: state.client.list_managed_connections().await,
    })
}

async fn handle_peer_session(
    State(state): State<Arc<AppState>>,
    Json(req): Json<PeerSessionRequest>,
) -> Json<PeerSessionResponse> {
    Json(PeerSessionResponse {
        session: state.client.peer_session(&req.id).await,
    })
}

async fn handle_peer_sessions(State(state): State<Arc<AppState>>) -> Json<PeerSessionsResponse> {
    Json(PeerSessionsResponse {
        sessions: state.client.peer_sessions().await,
    })
}

async fn handle_wait_settled(
    State(state): State<Arc<AppState>>,
    Json(req): Json<WaitSettledRequest>,
) -> Json<PeerSessionResponse> {
    Json(PeerSessionResponse {
        session: state
            .client
            .wait_for_settled_peer(&req.id, req.timeout_ms)
            .await,
    })
}

async fn handle_iroh_connect(
    State(state): State<Arc<AppState>>,
    Json(req): Json<ConnectRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let parsed =
        iroh_tickets::endpoint::EndpointTicket::from_str(req.ticket.trim()).map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                format!("Invalid endpoint ticket: {}", e),
            )
        })?;
    let endpoint_addr = parsed.endpoint_addr().clone();
    let endpoint_id = endpoint_addr.id;

    state
        .client
        .ensure_connected_addr(endpoint_id, endpoint_addr)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_auto_connect(
    State(state): State<Arc<AppState>>,
    Json(req): Json<AutoConnectRequest>,
) -> Json<OkResponse> {
    let client = state.client.clone();
    client.start_auto_connect(req.user_id, req.local_device_id);
    Json(OkResponse { ok: true })
}

async fn handle_node_addr(
    State(state): State<Arc<AppState>>,
) -> Result<Json<NodeAddrResponse>, (StatusCode, String)> {
    let ticket = state
        .client
        .endpoint_ticket()
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(NodeAddrResponse { node_addr: ticket }))
}

async fn handle_node_addr_with_token(
    State(state): State<Arc<AppState>>,
    Json(req): Json<NodeAddrWithTokenRequest>,
) -> Result<Json<NodeAddrResponse>, (StatusCode, String)> {
    let ticket = state
        .client
        .endpoint_ticket_with_token(&req.scope, req.max_connections.unwrap_or(0))
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(NodeAddrResponse { node_addr: ticket }))
}

async fn handle_revoke_tokens_by_scope(
    State(state): State<Arc<AppState>>,
    Json(req): Json<RevokeTokensByScopeRequest>,
) -> Result<Json<RevokeTokensResponse>, (StatusCode, String)> {
    let affected = state.client.revoke_tokens_by_scope(req.scope.trim()).await;
    Ok(Json(RevokeTokensResponse {
        affected_connection_ids: affected,
    }))
}

async fn handle_connect_ticket(
    State(state): State<Arc<AppState>>,
    Json(req): Json<ManagedConnectRequest>,
) -> Result<Json<ManagedConnectResponse>, (StatusCode, String)> {
    let connection = state
        .client
        .connect_device(req.device_id.as_deref(), &req.ticket)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(ManagedConnectResponse { connection }))
}

async fn handle_disconnect(
    State(state): State<Arc<AppState>>,
    Json(req): Json<DisconnectRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let endpoint_id = resolve_request_endpoint_id(&state.client, &req.peer_id).await?;
    state
        .client
        .disconnect(endpoint_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_open_bi(
    State(state): State<Arc<AppState>>,
    Json(req): Json<OpenBiRequest>,
) -> Result<Json<StreamIdResponse>, (StatusCode, String)> {
    let (_connection_id, _remote_node_id, send, recv) = state
        .client
        .open_peer_bi(&req.peer_id, Some(5_000))
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let mut next = state.next_stream_id.write().await;
    let id = *next;
    *next += 1;
    let stream_id = format!("stream-{}", id);

    state
        .streams
        .write()
        .await
        .insert(stream_id.clone(), StreamHandle { send, recv });

    Ok(Json(StreamIdResponse { stream_id }))
}

async fn handle_is_connected(
    State(state): State<Arc<AppState>>,
    Json(req): Json<IsConnectedRequest>,
) -> Result<Json<IsConnectedResponse>, (StatusCode, String)> {
    let endpoint_id = resolve_request_endpoint_id(&state.client, &req.peer_id).await?;
    let connected = state.client.is_connected(endpoint_id).await;
    Ok(Json(IsConnectedResponse { connected }))
}

async fn handle_wait_connected(
    State(state): State<Arc<AppState>>,
    Json(req): Json<WaitConnectedRequest>,
) -> Result<Json<IsConnectedResponse>, (StatusCode, String)> {
    let endpoint_id = resolve_request_endpoint_id(&state.client, &req.peer_id).await?;
    let timeout_ms = req.timeout_ms.unwrap_or(5000);
    let deadline =
        tokio::time::Instant::now() + std::time::Duration::from_millis(timeout_ms.max(250));

    loop {
        if state.client.is_connected(endpoint_id).await {
            return Ok(Json(IsConnectedResponse { connected: true }));
        }
        if tokio::time::Instant::now() >= deadline {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    Ok(Json(IsConnectedResponse { connected: false }))
}

async fn handle_send_on_stream(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendOnStreamRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let mut streams = state.streams.write().await;
    let handle = streams.get_mut(&req.stream_id).ok_or_else(|| {
        (
            StatusCode::NOT_FOUND,
            format!("Stream {} not found", req.stream_id),
        )
    })?;

    let bytes = req.data.as_bytes();
    let len = (bytes.len() as u32).to_be_bytes();
    handle
        .send
        .write_all(&len)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    handle
        .send
        .write_all(bytes)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_recv_from_stream(
    State(state): State<Arc<AppState>>,
    Json(req): Json<RecvFromStreamRequest>,
) -> Result<Json<RecvResponse>, (StatusCode, String)> {
    let timeout = std::time::Duration::from_millis(req.timeout_ms.unwrap_or(5000));

    let result = tokio::time::timeout(timeout, async {
        let mut streams = state.streams.write().await;
        let handle = streams.get_mut(&req.stream_id).ok_or_else(|| {
            (
                StatusCode::NOT_FOUND,
                format!("Stream {} not found", req.stream_id),
            )
        })?;

        let mut len_buf = [0u8; 4];
        read_exact_peer(&mut handle.recv, &mut len_buf)
            .await
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
        let len = u32::from_be_bytes(len_buf) as usize;

        let mut data_buf = vec![0u8; len];
        read_exact_peer(&mut handle.recv, &mut data_buf)
            .await
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

        let text = String::from_utf8(data_buf)
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

        Ok::<_, (StatusCode, String)>(text)
    })
    .await;

    match result {
        Ok(Ok(data)) => Ok(Json(RecvResponse { data: Some(data) })),
        Ok(Err(e)) => Err(e),
        Err(_) => Ok(Json(RecvResponse { data: None })),
    }
}

async fn handle_close_stream(
    State(state): State<Arc<AppState>>,
    Json(req): Json<CloseStreamRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let handle = state
        .streams
        .write()
        .await
        .remove(&req.stream_id)
        .ok_or_else(|| {
            (
                StatusCode::NOT_FOUND,
                format!("Stream {} not found", req.stream_id),
            )
        })?;

    let _ = handle.send.finish();
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_heartbeat_ping(
    State(state): State<Arc<AppState>>,
    Json(req): Json<HeartbeatPingRequest>,
) -> Result<Json<HeartbeatPingResponse>, (StatusCode, String)> {
    let timeout = std::time::Duration::from_millis(req.timeout_ms.unwrap_or(5000));
    let nonce = uuid::Uuid::new_v4().to_string();
    let ping = format!("__pluto_ping__:{}", nonce);
    let expected_pong = format!("__pluto_pong__:{}", nonce);
    let started = std::time::Instant::now();

    let (_connection_id, _remote_node_id, mut send, mut recv) = state
        .client
        .open_peer_bi(&req.peer_id, req.timeout_ms)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    write_peer_text_frame(&mut send, &ping)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let response = match tokio::time::timeout(timeout, read_peer_text_frame(&mut recv)).await {
        Ok(Ok(value)) => Some(value),
        Ok(Err(error)) => {
            return Err((StatusCode::INTERNAL_SERVER_ERROR, error.to_string()));
        }
        Err(_) => None,
    };

    let ok = response.as_deref() == Some(expected_pong.as_str());
    Ok(Json(HeartbeatPingResponse {
        ok,
        response,
        rtt_ms: started.elapsed().as_millis().min(u64::MAX as u128) as u64,
    }))
}

async fn handle_send_peer_data(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendPeerDataRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let payload = base64::engine::general_purpose::STANDARD
        .decode(req.payload_base64.as_bytes())
        .map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                format!("Invalid payloadBase64: {}", e),
            )
        })?;

    state
        .client
        .send_peer(&req.peer_id, &payload)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_send_moq_peer_data(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendPeerDataRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let payload = base64::engine::general_purpose::STANDARD
        .decode(req.payload_base64.as_bytes())
        .map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                format!("Invalid payloadBase64: {}", e),
            )
        })?;

    state
        .client
        .send_peer_over_moq(&req.peer_id, &payload)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_send_explicit_data(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendExplicitDataRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let payload = base64::engine::general_purpose::STANDARD
        .decode(req.payload_base64.as_bytes())
        .map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                format!("Invalid payloadBase64: {}", e),
            )
        })?;
    let mime_type = req
        .mime_type
        .unwrap_or_else(|| "application/octet-stream".to_string());

    write_explicit_transfer(
        state.client.clone(),
        req.peer_id,
        req.filename,
        mime_type,
        req.transfer_id,
        payload,
    )
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_send_explicit_file(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendExplicitFileRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    let file_path = PathBuf::from(req.file_path);
    let filename = file_path
        .file_name()
        .map(|v| v.to_string_lossy().to_string())
        .ok_or_else(|| (StatusCode::BAD_REQUEST, "Invalid filePath".to_string()))?;
    let payload = tokio::fs::read(&file_path).await.map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Failed to read file: {}", e),
        )
    })?;

    write_explicit_transfer(
        state.client.clone(),
        req.peer_id,
        filename,
        "application/octet-stream".to_string(),
        req.transfer_id,
        payload,
    )
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(OkResponse { ok: true }))
}

async fn handle_get_received_explicit(
    State(state): State<Arc<AppState>>,
) -> Json<ReceivedExplicitResponse> {
    let transfers = state.received_explicit.read().await.clone();
    Json(ReceivedExplicitResponse { transfers })
}

async fn handle_clear_received_explicit(State(state): State<Arc<AppState>>) -> Json<OkResponse> {
    state.received_explicit.write().await.clear();
    Json(OkResponse { ok: true })
}

async fn handle_received_peer_data(
    State(state): State<Arc<AppState>>,
) -> Json<ReceivedPeerDataResponse> {
    let messages = state.received_peer_data.read().await.clone();
    Json(ReceivedPeerDataResponse { messages })
}

async fn handle_clear_received_peer_data(State(state): State<Arc<AppState>>) -> Json<OkResponse> {
    state.received_peer_data.write().await.clear();
    Json(OkResponse { ok: true })
}

async fn handle_create_session(
    State(state): State<Arc<AppState>>,
    Json(req): Json<CreateSessionRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .create_session(req.session)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_subscribe_sessions(
    State(state): State<Arc<AppState>>,
    Json(req): Json<SubscribeSessionsRequest>,
) -> Result<Json<SessionsResponse>, (StatusCode, String)> {
    use futures::StreamExt;

    let wait = std::time::Duration::from_millis(req.wait_ms.unwrap_or(3000));
    let stream = state
        .client
        .subscribe_sessions(&req.local_device_id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let mut sessions = Vec::new();
    let deadline = tokio::time::Instant::now() + wait;
    tokio::pin!(stream);

    loop {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        match tokio::time::timeout(remaining, stream.next()).await {
            Ok(Some(Ok(events))) => {
                for event in events {
                    match event {
                        openrtc::signaling::SessionEvent::Added { session }
                        | openrtc::signaling::SessionEvent::Modified { session } => {
                            sessions.push(session);
                        }
                        _ => {}
                    }
                }
            }
            Ok(Some(Err(e))) => {
                eprintln!("[subscribe_sessions] Stream error: {}", e);
                break;
            }
            _ => break,
        }
    }

    Ok(Json(SessionsResponse { sessions }))
}

async fn handle_presence_loop(
    State(state): State<Arc<AppState>>,
    Json(req): Json<PresenceLoopRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .update_presence(&req.user_id, &req.device_name, &req.ticket, None)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_set_transport_config(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportConfigRequest>,
) -> Result<Json<OkResponse>, (StatusCode, String)> {
    state
        .client
        .update_transport_config(req.transports)
        .await
        .map_err(|error| (StatusCode::BAD_REQUEST, error.to_string()))?;
    Ok(Json(OkResponse { ok: true }))
}

async fn handle_get_transport_config(
    State(state): State<Arc<AppState>>,
) -> Json<TransportConfigResponse> {
    Json(TransportConfigResponse {
        transports: state.client.transport_config().await,
    })
}

#[cfg(feature = "transport-lan")]
async fn handle_local_peers(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    let peers = state.client.list_local_peers().await;
    Json(serde_json::json!({ "peers": peers }))
}

#[cfg(not(feature = "transport-lan"))]
async fn handle_local_peers(State(_state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    Json(serde_json::json!({ "peers": [] }))
}

async fn handle_transport_webrtc_state(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportWebRtcStateRequest>,
) -> Json<TransportWebRtcStateResponse> {
    let feature_enabled = openrtc::transport::native_webrtc_feature_enabled();
    let config_enabled = state.client.is_webrtc_transport_enabled().await;

    let (connection_id, state_name) = match state.client.native_webrtc_state_for_peer(&req.id).await
    {
        Some((connection_id, state)) => {
            let state_name = match state {
                openrtc::transport::NativeWebRTCState::Idle => "idle",
                openrtc::transport::NativeWebRTCState::Connecting => "connecting",
                openrtc::transport::NativeWebRTCState::Connected => "connected",
                openrtc::transport::NativeWebRTCState::Failed => "failed",
                openrtc::transport::NativeWebRTCState::Closed => "closed",
            }
            .to_string();
            (Some(connection_id), Some(state_name))
        }
        None => (None, None),
    };

    Json(TransportWebRtcStateResponse {
        feature_enabled,
        config_enabled,
        connection_id,
        state: state_name,
    })
}

async fn handle_transport_request_webrtc(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportWebRtcRequest>,
) -> Result<Json<TransportWebRtcRequestResponse>, (StatusCode, String)> {
    req.role
        .as_deref()
        .map(validate_native_webrtc_role)
        .transpose()
        .map_err(|message| (StatusCode::BAD_REQUEST, message))?;
    let requested = state
        .client
        .request_webrtc_upgrade(
            &req.id,
            req.remote_node_id.as_deref(),
            req.reason.as_deref(),
            req.role.as_deref(),
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(TransportWebRtcRequestResponse { requested }))
}

fn validate_native_webrtc_role(role: &str) -> Result<(), String> {
    match role.trim().to_ascii_lowercase().as_str() {
        "auto" | "initiator" | "responder" => Ok(()),
        other => Err(format!("invalid WebRTC role override: {}", other)),
    }
}

async fn handle_transport_moq_state(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportMoqStateRequest>,
) -> Json<TransportMoqStateResponse> {
    let feature_enabled = openrtc::transport::native_moq_feature_enabled();
    let transport_config = state.client.transport_config().await;
    let config_enabled = transport_config.moq.is_some();
    let relay_url = transport_config.moq.map(|config| config.relay_url);

    let (connection_id, state_name, data_ready) =
        match state.client.native_moq_state_detail_for_peer(&req.id).await {
            Some((connection_id, state, data_ready)) => {
                let state_name = match state {
                    openrtc::transport::NativeMoQState::Idle => "idle",
                    openrtc::transport::NativeMoQState::Connecting => "connecting",
                    openrtc::transport::NativeMoQState::Connected => "connected",
                    openrtc::transport::NativeMoQState::Failed => "failed",
                    openrtc::transport::NativeMoQState::Closed => "closed",
                }
                .to_string();
                (Some(connection_id), Some(state_name), data_ready)
            }
            None => (None, None, false),
        };

    Json(TransportMoqStateResponse {
        feature_enabled,
        config_enabled,
        connection_id,
        state: state_name,
        data_ready,
        relay_url,
    })
}

async fn handle_transport_request_moq(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportMoqRequest>,
) -> Result<Json<TransportMoqRequestResponse>, (StatusCode, String)> {
    let requested = state
        .client
        .request_moq_upgrade(
            &req.id,
            req.remote_node_id.as_deref(),
            req.reason.as_deref(),
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(TransportMoqRequestResponse { requested }))
}

async fn handle_transport_repair_moq(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransportMoqRequest>,
) -> Result<Json<TransportMoqRequestResponse>, (StatusCode, String)> {
    let requested = state
        .client
        .request_moq_repair(
            &req.id,
            req.remote_node_id.as_deref(),
            req.reason.as_deref(),
        )
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(TransportMoqRequestResponse { requested }))
}

// --- Main ---

#[tokio::main]
async fn main() -> Result<()> {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let cli = Cli::parse();

    // Create client with a token provider that reads from shared state
    let auth_token: Arc<std::sync::RwLock<Option<String>>> = Arc::new(std::sync::RwLock::new(None));
    let token_ref = auth_token.clone();

    let client = Client::new(
        cli.project_id,
        cli.api_key,
        Box::new(move || {
            // Sync read for the token — safe because we use std::sync::RwLock
            token_ref.read().unwrap().clone()
        }),
    );

    let state = Arc::new(AppState {
        client: Arc::new(client),
        user_id: RwLock::new(None),
        auth_token,
        native_device_base_dir: std::env::temp_dir()
            .join("openrtc-test-harness-device-identities")
            .join(format!("port-{}", cli.port)),
        ready: RwLock::new(false),
        heartbeat_responder_started: RwLock::new(false),
        streams: RwLock::new(HashMap::new()),
        next_stream_id: RwLock::new(0),
        received_explicit: RwLock::new(Vec::new()),
        received_peer_data: RwLock::new(Vec::new()),
    });

    {
        let state_ref = state.clone();
        let mut peer_data_rx = state.client.subscribe_native_peer_data();
        tokio::spawn(async move {
            loop {
                match peer_data_rx.recv().await {
                    Ok(event) => {
                        let mut messages = state_ref.received_peer_data.write().await;
                        messages.push(ReceivedPeerData {
                            connection_id: event.connection_id,
                            remote_node_id: event.remote_node_id,
                            transport: event.transport,
                            payload_base64: base64::engine::general_purpose::STANDARD
                                .encode(event.payload),
                            received_at_ms: std::time::SystemTime::now()
                                .duration_since(std::time::UNIX_EPOCH)
                                .map(|duration| {
                                    duration.as_millis().min(u128::from(u64::MAX)) as u64
                                })
                                .unwrap_or(0),
                        });
                        if messages.len() > 512 {
                            let overflow = messages.len().saturating_sub(512);
                            messages.drain(0..overflow);
                        }
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                }
            }
        });
    }

    let app = Router::new()
        .route("/status", get(handle_status))
        .route("/transport/config", post(handle_set_transport_config))
        .route("/transport/config", get(handle_get_transport_config))
        .route(
            "/transport/webrtc_state",
            post(handle_transport_webrtc_state),
        )
        .route(
            "/transport/request_webrtc",
            post(handle_transport_request_webrtc),
        )
        .route("/transport/moq_state", post(handle_transport_moq_state))
        .route("/transport/request_moq", post(handle_transport_request_moq))
        .route("/transport/repair_moq", post(handle_transport_repair_moq))
        .route(
            "/transport/send_moq_peer_data",
            post(handle_send_moq_peer_data),
        )
        .route("/auth", post(handle_auth))
        .route(
            "/native/device_identity/init",
            post(handle_init_native_device_identity),
        )
        .route("/presence/update", post(handle_presence_update))
        .route("/presence/offline", post(handle_presence_offline))
        .route("/presence/auto_connect", post(handle_auto_connect))
        .route("/presence/start_loop", post(handle_presence_loop))
        .route("/devices/search", post(handle_device_search))
        .route("/devices/status", post(handle_device_status))
        .route("/devices/delete", post(handle_device_delete))
        .route("/rooms/create", post(handle_room_create))
        .route("/rooms/join", post(handle_room_join))
        .route("/rooms/members", post(handle_room_members))
        .route("/rooms/leave", post(handle_room_leave))
        .route("/signaling/send", post(handle_signaling_send))
        .route("/signaling/create_session", post(handle_create_session))
        .route(
            "/signaling/subscribe_sessions",
            post(handle_subscribe_sessions),
        )
        .route("/iroh/init", post(handle_iroh_init))
        .route("/iroh/node_addr", get(handle_node_addr))
        .route(
            "/iroh/node_addr_with_token",
            post(handle_node_addr_with_token),
        )
        .route(
            "/session/revoke_by_scope",
            post(handle_revoke_tokens_by_scope),
        )
        .route("/iroh/connect", post(handle_iroh_connect))
        .route("/iroh/connect_ticket", post(handle_connect_ticket))
        .route("/iroh/disconnect", post(handle_disconnect))
        .route("/iroh/is_connected", post(handle_is_connected))
        .route("/iroh/wait_connected", post(handle_wait_connected))
        .route("/iroh/local_peers", get(handle_local_peers))
        .route("/iroh/connections", get(handle_iroh_connections))
        .route("/iroh/managed_connections", get(handle_managed_connections))
        .route("/iroh/peer_session", post(handle_peer_session))
        .route("/iroh/peer_sessions", get(handle_peer_sessions))
        .route("/iroh/wait_settled", post(handle_wait_settled))
        .route("/iroh/open_bi", post(handle_open_bi))
        .route("/iroh/send", post(handle_send_on_stream))
        .route("/iroh/recv", post(handle_recv_from_stream))
        .route("/iroh/close_stream", post(handle_close_stream))
        .route("/iroh/heartbeat_ping", post(handle_heartbeat_ping))
        .route("/iroh/send_peer_data", post(handle_send_peer_data))
        .route("/iroh/send_explicit_data", post(handle_send_explicit_data))
        .route("/iroh/send_explicit_file", post(handle_send_explicit_file))
        .route("/iroh/received_explicit", get(handle_get_received_explicit))
        .route(
            "/iroh/received_explicit/clear",
            post(handle_clear_received_explicit),
        )
        .route("/iroh/received_peer_data", get(handle_received_peer_data))
        .route(
            "/iroh/received_peer_data/clear",
            post(handle_clear_received_peer_data),
        )
        .layer(CorsLayer::permissive())
        .with_state(state);

    let addr = format!("127.0.0.1:{}", cli.port);
    println!("[test-harness] Listening on http://{}", addr);

    let listener = tokio::net::TcpListener::bind(&addr).await?;
    axum::serve(listener, app).await?;

    Ok(())
}