openrtc_tauri_plugin/

lib.rs

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

use base64::Engine as _;
use openrtc::application_crypto_streams::{PeerRecvStream, PeerSendStream};
use openrtc::native_node::IncomingStreamType;
use serde::{Deserialize, Serialize};
use tauri::{Emitter, Manager, Runtime};
use tokio::sync::Mutex;

const PLUGIN_NAME: &str = "openrtc-tauri-plugin";
const CONNECTION_STATE_CHANGED_EVENT: &str = "connection-state-changed";
const PEER_BI_STREAM_INCOMING_EVENT: &str = "openrtc://peer-bi-stream/incoming";
const PEER_BI_STREAM_CHUNK_EVENT: &str = "openrtc://peer-bi-stream/chunk";
const PEER_BI_STREAM_CLOSED_EVENT: &str = "openrtc://peer-bi-stream/closed";

pub type InstallFuture = std::pin::Pin<
    Box<
        dyn std::future::Future<Output = Result<Box<dyn std::any::Any + Send + Sync>, String>>
            + Send,
    >,
>;

#[derive(Debug, Clone)]
pub struct InstallContext {
    /// Stable host-owned storage used by every native endpoint initializer.
    /// Installers must derive their Iroh identity from this directory rather
    /// than creating a transport-specific identity.
    pub data_dir: PathBuf,
}

/// Host-supplied implementation for a native transport that cannot be bundled
/// in the publishable Tauri plugin. The OpenRTC constructor remains the sole
/// activation switch; registering an installer only declares compiled support.
pub trait TransportInstaller: Send + Sync {
    fn id(&self) -> &'static str;
    fn is_requested(&self, config: &openrtc::client::TransportConfig) -> bool;
    fn install(
        &self,
        client: Arc<openrtc::client::Client>,
        config: openrtc::client::TransportConfig,
        context: InstallContext,
    ) -> InstallFuture;
}

/// Host-owned secure-storage bridge for the OpenRTC 2.0 per-install device
/// proof key. OpenRTC never receives private key bytes: platform code stores
/// the Ed25519 key in Keychain/Keystore/Stronghold and exposes only public JWK
/// export plus signing.
pub trait DeviceKeySigner: Send + Sync {
    fn public_jwk(&self, app_tag: &str) -> Result<serde_json::Value, String>;
    fn sign(&self, app_tag: &str, challenge: &[u8]) -> Result<Vec<u8>, String>;
    fn delete(&self, app_tag: &str) -> Result<(), String>;
    fn read_secure_record(&self, _app_tag: &str, _key: &str) -> Result<Option<String>, String> {
        Ok(None)
    }
    fn write_secure_record(&self, _app_tag: &str, _key: &str, _value: &str) -> Result<(), String> {
        Err("OpenRTC 2.0 native certificate persistence requires a host secure store".to_string())
    }
    fn delete_secure_record(&self, _app_tag: &str, _key: &str) -> Result<(), String> {
        Ok(())
    }
}

struct InstalledNativeTransport {
    client_ptr: usize,
    _runtime: Box<dyn std::any::Any + Send + Sync>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OpenRtcTauriConfig {
    /// Public OpenRTC developer API key. The app identity is derived locally;
    /// consumers cannot retarget the native runtime with an app tag or
    /// provider project identifier.
    pub api_key: String,
    #[serde(default)]
    pub data_dir: Option<PathBuf>,
    #[serde(default)]
    pub transport_config: Option<openrtc::client::TransportConfig>,
}

impl Default for OpenRtcTauriConfig {
    fn default() -> Self {
        Self {
            api_key: String::new(),
            data_dir: None,
            transport_config: None,
        }
    }
}

impl OpenRtcTauriConfig {
    pub fn from_env() -> Self {
        let api_key = first_env(&[
            "VITE_OPENRTC_KEY",
            "VITE_OPENRTC_API_KEY",
            "VITE_PLUTO_OPENRTC_API_KEY",
            "OPENRTC_API_KEY",
        ]);
        Self {
            api_key: api_key.unwrap_or_default(),
            data_dir: None,
            transport_config: None,
        }
    }

    pub fn validated_api_key(&self) -> Result<&str, String> {
        openrtc::validate_api_key(&self.api_key)
            .map_err(|error| format!("invalid OpenRTC 2.0 public API key: {error}"))
    }

    pub fn app_tag(&self) -> Result<String, String> {
        self.validated_api_key().map(openrtc::app_tag_from_api_key)
    }
}

fn first_env(names: &[&str]) -> Option<String> {
    names.iter().find_map(|name| {
        std::env::var(name)
            .ok()
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
    })
}

#[derive(Default)]
struct TokenRelayState {
    identity_credential: RwLock<Option<String>>,
}

impl TokenRelayState {
    fn token_provider(self: &Arc<Self>) -> Box<dyn Fn() -> Option<String> + Send + Sync> {
        let relay = self.clone();
        Box::new(move || {
            relay
                .identity_credential
                .read()
                .ok()
                .and_then(|guard| guard.clone())
        })
    }

    fn set(&self, identity_credential: Option<String>) {
        if let Ok(mut guard) = self.identity_credential.write() {
            *guard = normalize_token(identity_credential);
        }
    }
}

fn normalize_token(value: Option<String>) -> Option<String> {
    value
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

pub struct OpenRtcTauriState {
    client: RwLock<Arc<openrtc::client::Client>>,
    config: OpenRtcTauriConfig,
    token_relay: Arc<TokenRelayState>,
    data_dir: Option<PathBuf>,
    connection_state_forwarder: std::sync::Mutex<Option<tauri::async_runtime::JoinHandle<()>>>,
    presence_loop_active: Mutex<bool>,
    subscriptions: Mutex<HashMap<String, tokio::task::JoinHandle<()>>>,
    peer_bi_streams: Mutex<HashMap<String, PeerBiStreamHandle>>,
    managed_session_start_guard: Mutex<()>,
    managed_session_next_owner_epoch: AtomicU64,
    managed_session: Mutex<Option<ManagedSessionRecord>>,
    native_transport_installers: Vec<Arc<dyn TransportInstaller>>,
    installed_native_transports: Mutex<HashMap<&'static str, InstalledNativeTransport>>,
    native_device_key_signer: Option<Arc<dyn DeviceKeySigner>>,
}

impl OpenRtcTauriState {
    pub fn new(config: OpenRtcTauriConfig) -> Self {
        openrtc::ensure_rustls();

        let token_relay = Arc::new(TokenRelayState::default());
        let client = build_client(&config, &token_relay)
            .expect("OpenRTC Tauri 2.0 requires a valid public API key");
        let data_dir = config.data_dir.clone();

        Self {
            client: RwLock::new(client),
            config,
            token_relay,
            data_dir,
            connection_state_forwarder: std::sync::Mutex::new(None),
            presence_loop_active: Mutex::new(false),
            subscriptions: Mutex::new(HashMap::new()),
            peer_bi_streams: Mutex::new(HashMap::new()),
            managed_session_start_guard: Mutex::new(()),
            managed_session_next_owner_epoch: AtomicU64::new(0),
            managed_session: Mutex::new(None),
            native_transport_installers: Vec::new(),
            installed_native_transports: Mutex::new(HashMap::new()),
            native_device_key_signer: None,
        }
    }

    pub fn with_native_transport_installer(
        mut self,
        installer: Arc<dyn TransportInstaller>,
    ) -> Self {
        self.native_transport_installers.push(installer);
        self
    }

    pub fn with_native_device_key_signer(mut self, signer: Arc<dyn DeviceKeySigner>) -> Self {
        self.native_device_key_signer = Some(signer);
        self
    }

    fn device_public_key(&self, app_tag: &str) -> Result<serde_json::Value, String> {
        let app_tag = required_app_tag(app_tag)?;
        let signer = self.native_device_key_signer.as_ref().ok_or_else(|| {
            "OpenRTC 2.0 native device proof requires a host secure-storage signer".to_string()
        })?;
        let value = signer.public_jwk(app_tag)?;
        validate_public_device_jwk(&value)?;
        Ok(value)
    }

    fn sign_device_proof(&self, app_tag: &str, challenge: &str) -> Result<String, String> {
        let app_tag = required_app_tag(app_tag)?;
        if challenge.is_empty() || challenge.len() > 2_048 {
            return Err("OpenRTC device challenge is invalid".to_string());
        }
        let signer = self.native_device_key_signer.as_ref().ok_or_else(|| {
            "OpenRTC 2.0 native device proof requires a host secure-storage signer".to_string()
        })?;
        let signature = signer.sign(app_tag, challenge.as_bytes())?;
        if signature.len() != 64 {
            return Err("OpenRTC device signer returned an invalid Ed25519 signature".to_string());
        }
        Ok(base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(signature))
    }

    fn delete_device_key(&self, app_tag: &str) -> Result<(), String> {
        let app_tag = required_app_tag(app_tag)?;
        let signer = self.native_device_key_signer.as_ref().ok_or_else(|| {
            "OpenRTC 2.0 native device proof requires a host secure-storage signer".to_string()
        })?;
        signer.delete(app_tag)
    }

    fn read_secure_record(&self, app_tag: &str, key: &str) -> Result<Option<String>, String> {
        let app_tag = required_app_tag(app_tag)?;
        let key = required_secure_record_key(key)?;
        self.native_device_key_signer
            .as_ref()
            .ok_or_else(|| {
                "OpenRTC 2.0 native certificate persistence requires a host secure store"
                    .to_string()
            })?
            .read_secure_record(app_tag, key)
    }

    fn write_secure_record(&self, app_tag: &str, key: &str, value: &str) -> Result<(), String> {
        let app_tag = required_app_tag(app_tag)?;
        let key = required_secure_record_key(key)?;
        if value.is_empty() || value.len() > 16 * 1024 {
            return Err("OpenRTC secure record is invalid".to_string());
        }
        self.native_device_key_signer
            .as_ref()
            .ok_or_else(|| {
                "OpenRTC 2.0 native certificate persistence requires a host secure store"
                    .to_string()
            })?
            .write_secure_record(app_tag, key, value)
    }

    fn delete_secure_record(&self, app_tag: &str, key: &str) -> Result<(), String> {
        let app_tag = required_app_tag(app_tag)?;
        let key = required_secure_record_key(key)?;
        self.native_device_key_signer
            .as_ref()
            .ok_or_else(|| {
                "OpenRTC 2.0 native certificate persistence requires a host secure store"
                    .to_string()
            })?
            .delete_secure_record(app_tag, key)
    }

    pub fn client(&self) -> Arc<openrtc::client::Client> {
        self.client
            .read()
            .map(|guard| guard.clone())
            .expect("OpenRTC Tauri client state is poisoned")
    }

    pub fn set_identity_credential(&self, identity_credential: Option<String>) {
        self.token_relay.set(identity_credential);
    }

    fn allocate_managed_session_owner_epoch(&self) -> u64 {
        self.managed_session_next_owner_epoch
            .fetch_add(1, Ordering::SeqCst)
            + 1
    }

    fn replace_connection_state_forwarder<R: Runtime>(
        &self,
        app: tauri::AppHandle<R>,
        client: Arc<openrtc::client::Client>,
    ) {
        let next = tauri::async_runtime::spawn(async move {
            forward_connection_state_events(app, client).await;
        });
        if let Ok(mut guard) = self.connection_state_forwarder.lock() {
            if let Some(previous) = guard.replace(next) {
                previous.abort();
            }
        }
    }
}

fn build_client(
    config: &OpenRtcTauriConfig,
    token_relay: &Arc<TokenRelayState>,
) -> Result<Arc<openrtc::client::Client>, String> {
    let mut builder = openrtc::client::Client::builder(
        config.validated_api_key()?.to_string(),
        token_relay.token_provider(),
    )
    .map_err(|error| error.to_string())?;
    if let Some(transport_config) = config.transport_config.clone() {
        builder = builder.transport_config(transport_config);
    }
    Ok(Arc::new(builder.build()))
}

fn requested_local_device_id(value: Option<&str>) -> Option<String> {
    value
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

fn managed_session_device_id(
    _requested: Option<&str>,
    native_identity: &openrtc::native_device::NativeDeviceIdentity,
) -> String {
    // The persisted native identity is the only durable-device authority.
    // Frontend IDs are aliases/display hints and must never create a second
    // gateway device projection or presence lease for the same native node.
    native_identity.device_id.clone()
}

#[cfg(test)]
fn metadata_with_authoritative_device_id(
    metadata: Option<String>,
    device_id: &str,
) -> Option<String> {
    let device_id = device_id.trim();
    if device_id.is_empty() {
        return metadata;
    }

    let Some(raw_metadata) = metadata else {
        return Some(serde_json::json!({ "deviceId": device_id }).to_string());
    };

    match serde_json::from_str::<serde_json::Value>(&raw_metadata) {
        Ok(serde_json::Value::Object(mut map)) => {
            map.insert(
                "deviceId".to_string(),
                serde_json::Value::String(device_id.to_string()),
            );
            Some(serde_json::Value::Object(map).to_string())
        }
        _ => Some(
            serde_json::json!({
                "deviceId": device_id,
                "metadata": raw_metadata,
            })
            .to_string(),
        ),
    }
}

struct PeerBiStreamHandle {
    send: Arc<Mutex<Option<PeerSendStream>>>,
    recv: Option<PeerRecvStream>,
    read_task: Option<tokio::task::JoinHandle<()>>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct OpenBiResult {
    stream_id: String,
    connection_id: Option<String>,
    remote_node_id: String,
}

#[derive(Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct IncomingPeerBiStreamEvent {
    request_id: String,
    stream_id: String,
    remote_node_id: String,
    transport_stable_id: u64,
}

#[derive(Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct PeerBiStreamChunkEvent {
    stream_id: String,
    bytes: Vec<u8>,
}

#[derive(Serialize, Clone)]
#[serde(rename_all = "camelCase")]
struct PeerBiStreamClosedEvent {
    stream_id: String,
    error: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StartSessionResult {
    local_node_id: String,
    ticket_scope: Option<String>,
    ticket: Option<String>,
    presence_started: bool,
    auto_connect_started: bool,
    local_device: openrtc::native_device::NativeDeviceIdentity,
}

#[derive(Clone)]
struct ManagedSessionRecord {
    key: String,
    owner_client: Arc<openrtc::client::Client>,
    owner_epoch: u64,
    result: StartSessionResult,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SessionDisposition {
    Start,
    Reuse,
    Refresh,
    Replace,
}

fn managed_session_disposition(
    active_key: Option<&str>,
    active_client_matches: bool,
    active_presence_started: bool,
    active_auto_connect_started: bool,
    requested_key: &str,
    requested_presence: bool,
    requested_auto_connect: bool,
) -> SessionDisposition {
    let Some(active_key) = active_key else {
        return SessionDisposition::Start;
    };
    if !active_client_matches || active_key != requested_key {
        return SessionDisposition::Replace;
    }
    if (!requested_presence || active_presence_started)
        && (!requested_auto_connect || active_auto_connect_started)
    {
        SessionDisposition::Reuse
    } else {
        SessionDisposition::Refresh
    }
}

fn revokes_managed_session(scope: &str) -> bool {
    scope.trim() == "user-device"
}

pub fn init<R: Runtime>(config: OpenRtcTauriConfig) -> tauri::plugin::TauriPlugin<R> {
    init_with_state(OpenRtcTauriState::new(config))
}

pub fn init_with_state<R: Runtime>(state: OpenRtcTauriState) -> tauri::plugin::TauriPlugin<R> {
    tauri::plugin::Builder::new(PLUGIN_NAME)
        .setup(move |app, _api| {
            let client = state.client();
            let transport_config = state.config.transport_config.clone();
            // Optional native transports must decorate the endpoint before the
            // host app can initialize the shared Iroh client during its setup.
            // Managed-session startup is too late for custom transports because
            // Iroh endpoint hooks are immutable after bind/adoption.
            let install_context = InstallContext {
                data_dir: app_data_dir(app, &state).map_err(std::io::Error::other)?,
            };
            tauri::async_runtime::block_on(ensure_requested_native_transports(
                &state,
                &client,
                transport_config.as_ref(),
                &install_context,
            ))
            .map_err(std::io::Error::other)?;
            state.replace_connection_state_forwarder(app.clone(), client);
            app.manage(state);
            Ok(())
        })
        .invoke_handler(tauri::generate_handler![
            openrtc_set_identity_credential,
            openrtc_device_public_key,
            openrtc_sign_device_proof,
            openrtc_delete_device_key,
            openrtc_read_secure_record,
            openrtc_write_secure_record,
            openrtc_delete_secure_record,
            rtc_native_status,
            get_rtc_local_device_info,
            update_rtc_local_device_name,
            get_iroh_node_id,
            start_iroh_node,
            get_iroh_endpoint_ticket,
            register_session_token,
            get_endpoint_ticket_with_token,
            validate_session_token,
            revoke_session_tokens_by_scope,
            stop_rtc_presence_loop,
            start_rtc_managed_session,
            start_rtc_external_auto_connect,
            submit_rtc_desired_peers,
            stop_rtc_auto_connect,
            notify_rtc_network_change,
            connect_to_device,
            disconnect_device,
            set_auto_connect_excluded,
            set_rtc_external_auto_connect_excluded,
            resolve_rtc_peer_connection_records,
            resolve_rtc_peer_identity,
            get_rtc_peer_session,
            list_rtc_peer_sessions,
            list_rtc_managed_connections,
            wait_for_rtc_settled_peer,
            list_rtc_connection_states,
            get_rtc_connection_state,
            stop_rtc_subscription,
            start_incoming_peer_bi_streams,
            is_current_transport_stable_id,
            open_peer_bi_stream,
            open_peer_bi_transport_only_stream,
            open_peer_native_bi_stream,
            write_peer_bi_stream,
            start_peer_bi_stream_read,
            close_peer_bi_stream,
        ])
        .build()
}

pub fn init_from_env<R: Runtime>() -> tauri::plugin::TauriPlugin<R> {
    init(OpenRtcTauriConfig::from_env())
}

async fn forward_connection_state_events<R: Runtime>(
    app: tauri::AppHandle<R>,
    client: Arc<openrtc::client::Client>,
) {
    let mut rx = client.connection_state_updates();
    while let Ok(snapshot) = rx.recv().await {
        let _ = app.emit(CONNECTION_STATE_CHANGED_EVENT, snapshot);
    }
}

fn app_data_dir<R: Runtime>(
    app: &tauri::AppHandle<R>,
    state: &OpenRtcTauriState,
) -> Result<PathBuf, String> {
    state
        .data_dir
        .clone()
        .or_else(|| app.path().app_data_dir().ok())
        .ok_or_else(|| "failed to resolve OpenRTC app data directory".to_string())
}

async fn ensure_iroh_node(client: Arc<openrtc::client::Client>) -> Result<String, String> {
    if let Some(node_id) = client.current_node_id().await {
        return Ok(node_id);
    }
    client
        .init_iroh(None, Vec::new())
        .await
        .map_err(|error| format!("failed to initialize OpenRTC Iroh node: {error}"))
}

async fn ensure_requested_native_transports(
    state: &OpenRtcTauriState,
    client: &Arc<openrtc::client::Client>,
    transports: Option<&openrtc::client::TransportConfig>,
    context: &InstallContext,
) -> Result<(), String> {
    let Some(transports) = transports else {
        return Ok(());
    };
    let ble_requested = transports.ble.as_ref().is_some_and(|config| config.enabled);
    if ble_requested
        && !state
            .native_transport_installers
            .iter()
            .any(|installer| installer.is_requested(transports))
    {
        eprintln!(
            "[openrtc-tauri][transport] BLE requested but unavailable: this host did not register a BLE transport installer; continuing on the Iroh base route"
        );
        return Ok(());
    }
    let client_ptr = Arc::as_ptr(client) as usize;
    for installer in &state.native_transport_installers {
        if !installer.is_requested(transports) {
            continue;
        }
        let already_installed = state
            .installed_native_transports
            .lock()
            .await
            .get(installer.id())
            .is_some_and(|installed| installed.client_ptr == client_ptr);
        if already_installed {
            continue;
        }
        if client.current_node_id().await.is_some() {
            eprintln!(
                "[openrtc-tauri][transport] {} requested after native node startup; continuing on the Iroh base route",
                installer.id()
            );
            continue;
        }
        let runtime = match installer
            .install(client.clone(), transports.clone(), context.clone())
            .await
        {
            Ok(runtime) => runtime,
            Err(error) => {
                eprintln!(
                    "[openrtc-tauri][transport] {} unavailable: {}; continuing on the Iroh base route",
                    installer.id(),
                    error
                );
                continue;
            }
        };
        state.installed_native_transports.lock().await.insert(
            installer.id(),
            InstalledNativeTransport {
                client_ptr,
                _runtime: runtime,
            },
        );
    }
    Ok(())
}

async fn register_peer_bi_stream<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: &OpenRtcTauriState,
    connection_id: Option<String>,
    remote_node_id: String,
    send: PeerSendStream,
    recv: PeerRecvStream,
    start_reading: bool,
) -> OpenBiResult {
    let stream_id = uuid::Uuid::new_v4().to_string();
    let send = Arc::new(Mutex::new(Some(send)));
    let (recv, read_task) = if start_reading {
        (
            None,
            Some(spawn_peer_bi_stream_reader(
                app.clone(),
                stream_id.clone(),
                recv,
            )),
        )
    } else {
        // Incoming native streams are announced to JS before forwarding bytes so
        // the bridge can install per-stream listeners without losing the first
        // native-main label/session-token frame.
        (Some(recv), None)
    };

    state.peer_bi_streams.lock().await.insert(
        stream_id.clone(),
        PeerBiStreamHandle {
            send,
            recv,
            read_task,
        },
    );

    OpenBiResult {
        stream_id,
        connection_id,
        remote_node_id,
    }
}

fn spawn_peer_bi_stream_reader<R: Runtime>(
    app: tauri::AppHandle<R>,
    stream_id: String,
    mut recv: PeerRecvStream,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut chunk = vec![0_u8; 64 * 1024];
        let mut close_error: Option<String> = None;
        loop {
            match recv.read(&mut chunk).await {
                Ok(0) => break,
                Ok(n) => {
                    if std::env::var("OPENRTC_NATIVE_STREAM_TRACE").ok().as_deref() == Some("1") {
                        eprintln!(
                            "[openrtc-tauri][peer-bi-stream] stream_id={} phase=plaintext-chunk bytes={}",
                            stream_id, n
                        );
                    }
                    let _ = app.emit(
                        PEER_BI_STREAM_CHUNK_EVENT,
                        PeerBiStreamChunkEvent {
                            stream_id: stream_id.clone(),
                            bytes: chunk[..n].to_vec(),
                        },
                    );
                }
                Err(error) => {
                    if std::env::var("OPENRTC_NATIVE_STREAM_TRACE").ok().as_deref() == Some("1") {
                        eprintln!(
                            "[openrtc-tauri][peer-bi-stream] stream_id={} phase=read-error error={}",
                            stream_id, error
                        );
                    }
                    close_error = Some(error.to_string());
                    break;
                }
            }
        }
        let _ = app.emit(
            PEER_BI_STREAM_CLOSED_EVENT,
            PeerBiStreamClosedEvent {
                stream_id,
                error: close_error,
            },
        );
    })
}

async fn open_peer_bi_with<R: Runtime, F, Fut>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    peer_id: String,
    timeout_ms: Option<u64>,
    open: F,
) -> Result<OpenBiResult, String>
where
    F: FnOnce(Arc<openrtc::client::Client>, String, Option<u64>) -> Fut,
    Fut: std::future::Future<
        Output = anyhow::Result<(Option<String>, String, PeerSendStream, PeerRecvStream)>,
    >,
{
    let peer_id = peer_id.trim().to_string();
    if peer_id.is_empty() {
        return Err("peerId is required".to_string());
    }

    let (connection_id, remote_node_id, send, recv) = open(state.client(), peer_id, timeout_ms)
        .await
        .map_err(|error| format!("open peer bi stream failed: {error}"))?;
    Ok(register_peer_bi_stream(app, &state, connection_id, remote_node_id, send, recv, true).await)
}

/// Relay an opaque OpenRTC-issued identity credential into the provider-neutral
/// Rust runtime. The host owns acquisition and renewal; this command never
/// interprets the value as a Firebase token and accepts no refresh token.
#[tauri::command]
async fn openrtc_set_identity_credential(
    state: tauri::State<'_, OpenRtcTauriState>,
    identity_credential: Option<String>,
) -> Result<(), String> {
    state.set_identity_credential(identity_credential);
    if *state.presence_loop_active.lock().await {
        state.client().request_presence_update();
    }
    Ok(())
}

fn required_app_tag(value: &str) -> Result<&str, String> {
    let value = value.trim();
    if value.len() < 5
        || value.len() > 80
        || !value.bytes().all(|byte| {
            byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':' | b'@')
        })
    {
        return Err("OpenRTC app tag is invalid".to_string());
    }
    Ok(value)
}

fn validate_public_device_jwk(value: &serde_json::Value) -> Result<(), String> {
    let object = value
        .as_object()
        .ok_or_else(|| "OpenRTC device public key is invalid".to_string())?;
    let x = object
        .get("x")
        .and_then(serde_json::Value::as_str)
        .unwrap_or_default();
    if object.get("kty").and_then(serde_json::Value::as_str) != Some("OKP")
        || object.get("crv").and_then(serde_json::Value::as_str) != Some("Ed25519")
        || object.contains_key("d")
        || x.len() != 43
        || !x
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
    {
        return Err("OpenRTC device public key must be a public Ed25519 JWK".to_string());
    }
    Ok(())
}

fn required_secure_record_key(value: &str) -> Result<&str, String> {
    let value = value.trim();
    if value.is_empty()
        || value.len() > 512
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':'))
    {
        return Err("OpenRTC secure record key is invalid".to_string());
    }
    Ok(value)
}

#[tauri::command]
async fn openrtc_device_public_key(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
) -> Result<serde_json::Value, String> {
    state.device_public_key(&app_tag)
}

#[tauri::command]
async fn openrtc_sign_device_proof(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
    challenge: String,
) -> Result<String, String> {
    state.sign_device_proof(&app_tag, &challenge)
}

#[tauri::command]
async fn openrtc_delete_device_key(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
) -> Result<(), String> {
    state.delete_device_key(&app_tag)
}

#[tauri::command]
async fn openrtc_read_secure_record(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
    key: String,
) -> Result<Option<String>, String> {
    state.read_secure_record(&app_tag, &key)
}

#[tauri::command]
async fn openrtc_write_secure_record(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
    key: String,
    value: String,
) -> Result<(), String> {
    state.write_secure_record(&app_tag, &key, &value)
}

#[tauri::command]
async fn openrtc_delete_secure_record(
    state: tauri::State<'_, OpenRtcTauriState>,
    app_tag: String,
    key: String,
) -> Result<(), String> {
    state.delete_secure_record(&app_tag, &key)
}

#[tauri::command]
async fn rtc_native_status(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<openrtc::client::RuntimeStatus, String> {
    Ok(state.client().runtime_status().await)
}

#[tauri::command]
async fn get_rtc_local_device_info<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<openrtc::native_device::NativeDeviceIdentity, String> {
    let data_dir = app_data_dir(&app, &state)?;
    state
        .client()
        .init_native_device_identity(data_dir, None)
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn update_rtc_local_device_name<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    device_name: String,
) -> Result<openrtc::native_device::NativeDeviceIdentity, String> {
    let data_dir = app_data_dir(&app, &state)?;
    let _ = state
        .client()
        .init_native_device_identity(data_dir, None)
        .await
        .map_err(|error| error.to_string())?;
    state
        .client()
        .update_native_device_name(&device_name)
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn get_iroh_node_id(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<Option<String>, String> {
    Ok(state.client().current_node_id().await)
}

#[tauri::command]
async fn start_iroh_node(state: tauri::State<'_, OpenRtcTauriState>) -> Result<String, String> {
    ensure_iroh_node(state.client()).await
}

#[tauri::command]
async fn get_iroh_endpoint_ticket(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<String, String> {
    ensure_iroh_node(state.client()).await?;
    state
        .client()
        .endpoint_ticket()
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn register_session_token(
    state: tauri::State<'_, OpenRtcTauriState>,
    token: String,
    scope: String,
    max_connections: u32,
    expires_at_ms: Option<u64>,
) -> Result<(), String> {
    if let Some(expires_at_ms) = expires_at_ms {
        state
            .client()
            .register_token_until(token, scope, max_connections, expires_at_ms);
    } else {
        state
            .client()
            .register_session_token(token, scope, max_connections);
    }
    Ok(())
}

#[tauri::command]
async fn get_endpoint_ticket_with_token(
    state: tauri::State<'_, OpenRtcTauriState>,
    scope: String,
    max_connections: u32,
) -> Result<String, String> {
    ensure_iroh_node(state.client()).await?;
    state
        .client()
        .endpoint_ticket_with_token(&scope, max_connections)
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn validate_session_token(
    state: tauri::State<'_, OpenRtcTauriState>,
    token: String,
    connection_id: Option<String>,
) -> Result<String, String> {
    if let Some(connection_id) = connection_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        state
            .client()
            .validate_connection_token(&token, connection_id, None)
            .await
            .map_err(|error| error.to_string())
    } else {
        state.client().validate_session_token(&token)
    }
}

#[tauri::command]
async fn revoke_session_tokens_by_scope(
    state: tauri::State<'_, OpenRtcTauriState>,
    scope: String,
) -> Result<Vec<String>, String> {
    let affected = state.client().revoke_tokens_by_scope(&scope).await;
    if revokes_managed_session(&scope) {
        // The cached managed-session result contains the compound ticket whose
        // token was just revoked. Keeping that record would make the next
        // same-user/device start return a credential the host can no longer
        // validate. Invalidate only this lifecycle cache; the durable native
        // device identity remains intact so explicit reauthentication can
        // re-enroll the same installation with a freshly minted ticket.
        if let Some(previous) = state.managed_session.lock().await.take() {
            previous.owner_client.stop_presence_loop();
            previous.owner_client.stop_auto_connect();
            previous.owner_client.stop_external_auto_connect().await;
        }
        *state.presence_loop_active.lock().await = false;
    }
    Ok(affected)
}

#[tauri::command]
async fn stop_rtc_presence_loop(state: tauri::State<'_, OpenRtcTauriState>) -> Result<(), String> {
    state.client().stop_presence_loop();
    *state.presence_loop_active.lock().await = false;
    if let Some(active) = state.managed_session.lock().await.as_mut() {
        active.result.presence_started = false;
    }
    stop_subscription_by_id(&state, "internal-presence-loop").await;
    Ok(())
}

#[tauri::command]
async fn start_rtc_managed_session<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    user_id: String,
    device_name: Option<String>,
    local_device_id: Option<String>,
    metadata: Option<String>,
    transports: Option<openrtc::client::TransportConfig>,
    auto_connect: Option<bool>,
    presence: Option<bool>,
) -> Result<StartSessionResult, String> {
    let command_started = std::time::Instant::now();
    let _start_guard = state.managed_session_start_guard.lock().await;
    let client = state.client();
    let app_tag = client.app_tag().to_string();
    eprintln!(
        "[openrtc-tauri][managed-session] start user_id={} app_tag={} has_device_name={} has_local_device_id={} has_metadata={} auto_connect={} presence={}",
        user_id,
        app_tag,
        device_name
            .as_deref()
            .map(str::trim)
            .map(|value| !value.is_empty())
            .unwrap_or(false),
        local_device_id
            .as_deref()
            .map(str::trim)
            .map(|value| !value.is_empty())
            .unwrap_or(false),
        metadata
            .as_deref()
            .map(str::trim)
            .map(|value| !value.is_empty())
            .unwrap_or(false),
        auto_connect.unwrap_or(false),
        presence.unwrap_or(false)
    );
    let data_dir = app_data_dir(&app, &state)?;
    let install_context = InstallContext {
        data_dir: data_dir.clone(),
    };
    ensure_requested_native_transports(&state, &client, transports.as_ref(), &install_context)
        .await?;
    if let Some(transport_config) = transports {
        client
            .update_transport_config(transport_config)
            .await
            .map_err(|error| error.to_string())?;
    }

    let local_node_id = ensure_iroh_node(client.clone()).await?;
    let local_device = client
        .init_native_device_identity(data_dir, device_name.as_deref())
        .await
        .map_err(|error| error.to_string())?;

    let effective_local_device_id =
        managed_session_device_id(local_device_id.as_deref(), &local_device);
    if let Some(requested) = requested_local_device_id(local_device_id.as_deref()) {
        if requested != local_device.device_id {
            eprintln!(
                "[openrtc-tauri] requested localDeviceId={} is an alias; persisted native identity {} remains authoritative",
                requested, local_device.device_id
            );
        }
    }

    let should_start_presence = presence.unwrap_or(false);
    let should_start_auto_connect = auto_connect.unwrap_or(false);
    if should_start_presence || should_start_auto_connect {
        return Err(
            "native Firebase coordination has been removed; publish gateway presence and submit external desired peers instead"
                .to_string(),
        );
    }
    let managed_session_key = format!("{}:{}:{}", app_tag, user_id, effective_local_device_id);
    let (disposition, reused) = {
        let active = state.managed_session.lock().await;
        let disposition = managed_session_disposition(
            active.as_ref().map(|record| record.key.as_str()),
            active
                .as_ref()
                .is_some_and(|record| Arc::ptr_eq(&record.owner_client, &client)),
            active
                .as_ref()
                .is_some_and(|record| record.result.presence_started),
            active
                .as_ref()
                .is_some_and(|record| record.result.auto_connect_started),
            &managed_session_key,
            should_start_presence,
            should_start_auto_connect,
        );
        let reused = (disposition == SessionDisposition::Reuse).then(|| {
            active
                .as_ref()
                .expect("managed session reuse requires an active record")
                .result
                .clone()
        });
        (disposition, reused)
    };
    if let Some(reused) = reused {
        eprintln!(
            "[openrtc-tauri][managed-session] reused key={} elapsed_ms={}",
            managed_session_key,
            command_started.elapsed().as_millis()
        );
        return Ok(reused);
    }

    // A different app/user/device tuple replaces the previous lifecycle owner.
    // Stop old background work before publishing the new authoritative lease.
    if disposition == SessionDisposition::Replace {
        let previous = state
            .managed_session
            .lock()
            .await
            .take()
            .expect("managed session replacement requires an active record");
        previous.owner_client.stop_presence_loop();
        previous.owner_client.stop_auto_connect();
        previous.owner_client.stop_external_auto_connect().await;
        *state.presence_loop_active.lock().await = false;
    }
    let ticket_scope = Some("user-device".to_string());
    let ticket = client
        .endpoint_ticket_with_token("user-device", 0)
        .await
        .map_err(|error| error.to_string())?;

    let logical_local_device = local_device.clone();

    let owner_epoch = match disposition {
        SessionDisposition::Refresh => {
            state
                .managed_session
                .lock()
                .await
                .as_ref()
                .expect("managed session refresh requires an active record")
                .owner_epoch
        }
        SessionDisposition::Start | SessionDisposition::Replace => {
            state.allocate_managed_session_owner_epoch()
        }
        SessionDisposition::Reuse => unreachable!("reuse returns before session startup"),
    };

    eprintln!(
        "[openrtc-tauri][managed-session] ready user_id={} app_tag={} local_node_id={} local_device_id={} owner_epoch={} presence_started={} auto_connect_started={} ticket_scope={} elapsed_ms={}",
        user_id,
        app_tag,
        local_node_id,
        logical_local_device.device_id,
        owner_epoch,
        should_start_presence,
        should_start_auto_connect,
        ticket_scope.as_deref().unwrap_or("unrestricted"),
        command_started.elapsed().as_millis()
    );

    let result = StartSessionResult {
        local_node_id,
        ticket_scope,
        ticket: Some(ticket),
        presence_started: false,
        auto_connect_started: false,
        local_device: logical_local_device,
    };
    *state.managed_session.lock().await = Some(ManagedSessionRecord {
        key: managed_session_key,
        owner_client: client,
        owner_epoch,
        result: result.clone(),
    });
    Ok(result)
}

#[tauri::command]
async fn start_rtc_external_auto_connect(
    state: tauri::State<'_, OpenRtcTauriState>,
    user_id: String,
    local_device_id: String,
) -> Result<(), String> {
    state
        .client()
        .start_external_auto_connect(user_id, local_device_id)
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn submit_rtc_desired_peers(
    state: tauri::State<'_, OpenRtcTauriState>,
    revision: u64,
    peers_json: String,
) -> Result<bool, String> {
    state
        .client()
        .submit_external_desired_peers(revision, &peers_json)
        .await
        .map_err(|error| error.to_string())
}

#[tauri::command]
async fn stop_rtc_auto_connect(state: tauri::State<'_, OpenRtcTauriState>) -> Result<(), String> {
    state.client().stop_auto_connect();
    state.client().stop_external_auto_connect().await;
    if let Some(active) = state.managed_session.lock().await.as_mut() {
        active.result.auto_connect_started = false;
    }
    Ok(())
}

#[tauri::command]
async fn connect_to_device(
    state: tauri::State<'_, OpenRtcTauriState>,
    device_id: Option<String>,
    endpoint_ticket: String,
    timeout_ms: Option<u64>,
) -> Result<openrtc::client::ManagedConnectResult, String> {
    let client = state.client();
    let connect = client.connect_device(device_id.as_deref(), &endpoint_ticket);
    match timeout_ms {
        Some(timeout_ms) => {
            tokio::time::timeout(std::time::Duration::from_millis(timeout_ms.max(1)), connect)
                .await
                .map_err(|_| format!("connect_to_device timeout after {timeout_ms}ms"))?
                .map_err(|error| error.to_string())
        }
        None => connect.await.map_err(|error| error.to_string()),
    }
}

#[tauri::command]
async fn disconnect_device(
    state: tauri::State<'_, OpenRtcTauriState>,
    device_id: String,
    node_id_hint: Option<String>,
) -> Result<(), String> {
    state
        .client()
        .disconnect_device(&device_id, node_id_hint.as_deref())
        .await;
    Ok(())
}

#[tauri::command]
async fn set_auto_connect_excluded(
    state: tauri::State<'_, OpenRtcTauriState>,
    device_id: String,
    excluded: bool,
) -> Result<(), String> {
    if excluded {
        state.client().exclude_peer_and_publish(&device_id).await;
    } else {
        state.client().unexclude_peer_and_publish(&device_id).await;
    }
    Ok(())
}

#[tauri::command]
async fn set_rtc_external_auto_connect_excluded(
    state: tauri::State<'_, OpenRtcTauriState>,
    device_id: String,
    excluded: bool,
) -> Result<(), String> {
    state
        .client()
        .set_auto_connect_excluded(&device_id, excluded);
    state.client().wake_native_external_auto_connect().await;
    Ok(())
}

#[tauri::command]
async fn resolve_rtc_peer_connection_records(
    state: tauri::State<'_, OpenRtcTauriState>,
    id: String,
) -> Result<Vec<openrtc::connection_manager::ConnectionRecord>, String> {
    Ok(state.client().resolve_peer_connection_records(&id).await)
}

#[tauri::command]
async fn resolve_rtc_peer_identity(
    state: tauri::State<'_, OpenRtcTauriState>,
    id: String,
) -> Result<Option<openrtc::connection_manager::PeerSnapshot>, String> {
    Ok(state.client().peer_snapshot(&id).await)
}

#[tauri::command]
async fn get_rtc_peer_session(
    state: tauri::State<'_, OpenRtcTauriState>,
    id: String,
) -> Result<Option<openrtc::client::PeerSessionSnapshot>, String> {
    Ok(state.client().peer_session(&id).await)
}

#[tauri::command]
async fn list_rtc_peer_sessions(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<Vec<openrtc::client::PeerSessionSnapshot>, String> {
    Ok(state.client().peer_sessions().await)
}

#[tauri::command]
async fn list_rtc_managed_connections(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<Vec<openrtc::connection_manager::ConnectionRecord>, String> {
    Ok(state.client().list_managed_connections().await)
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
struct NotifyRtcNetworkChangeResult {
    retired_stale_connections: usize,
}

#[tauri::command]
async fn notify_rtc_network_change(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<NotifyRtcNetworkChangeResult, String> {
    let retired_stale_connections = state
        .client()
        .notify_network_change()
        .await
        .map_err(|error| error.to_string())?;
    Ok(NotifyRtcNetworkChangeResult {
        retired_stale_connections,
    })
}

#[tauri::command]
async fn wait_for_rtc_settled_peer(
    state: tauri::State<'_, OpenRtcTauriState>,
    id: String,
    timeout_ms: Option<u64>,
) -> Result<Option<openrtc::client::PeerSessionSnapshot>, String> {
    Ok(state.client().wait_for_peer(&id, timeout_ms).await)
}

#[tauri::command]
async fn list_rtc_connection_states(
    state: tauri::State<'_, OpenRtcTauriState>,
) -> Result<Vec<openrtc::client::StateSnapshot>, String> {
    Ok(state.client().connection_states().await)
}

#[tauri::command]
async fn get_rtc_connection_state(
    state: tauri::State<'_, OpenRtcTauriState>,
    connection_id: String,
) -> Result<Option<openrtc::client::StateSnapshot>, String> {
    Ok(state.client().connection_state(&connection_id).await)
}

#[tauri::command]
async fn stop_rtc_subscription(
    state: tauri::State<'_, OpenRtcTauriState>,
    request_id: String,
) -> Result<(), String> {
    stop_subscription_by_id(&state, &request_id).await;
    Ok(())
}

async fn stop_subscription_by_id(state: &OpenRtcTauriState, request_id: &str) {
    if let Some(handle) = state.subscriptions.lock().await.remove(request_id) {
        handle.abort();
    }
}

#[tauri::command]
async fn start_incoming_peer_bi_streams<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    request_id: Option<String>,
) -> Result<String, String> {
    let request_id = request_id
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

    stop_subscription_by_id(&state, &request_id).await;

    let incoming = state
        .client()
        .incoming_streams()
        .await
        .map_err(|error| format!("failed to subscribe to incoming peer streams: {error}"))?;
    let app_for_task = app.clone();
    let request_for_task = request_id.clone();
    let handle = tokio::spawn(async move {
        while let Ok(incoming_stream) = incoming.recv().await {
            let endpoint_id = incoming_stream.endpoint_id;
            let remote_node_id = endpoint_id.to_string();
            let transport_stable_id = incoming_stream.transport_stable_id;
            let recv_prefix = incoming_stream.recv_prefix;
            let IncomingStreamType::Bi(send, recv) = incoming_stream.stream else {
                continue;
            };
            if std::env::var("OPENRTC_NATIVE_STREAM_TRACE").ok().as_deref() == Some("1") {
                eprintln!(
                    "[openrtc-tauri][incoming-application-stream] remote_node_id={} transport_stable_id={} request_id={} phase=received",
                    remote_node_id,
                    transport_stable_id,
                    request_for_task
                );
            }
            let state_for_task = app_for_task.state::<OpenRtcTauriState>();
            let (send, recv) = match state_for_task
                .client()
                .wrap_bi_with_prefix(&endpoint_id, transport_stable_id, send, recv, &recv_prefix)
                .await
            {
                Ok(streams) => streams,
                Err(error) => {
                    eprintln!(
                        "[openrtc-tauri][incoming-application-stream] remote_node_id={} transport_stable_id={} phase=rejected error={}",
                        remote_node_id, transport_stable_id, error
                    );
                    continue;
                }
            };
            let result = register_peer_bi_stream(
                app_for_task.clone(),
                state_for_task.inner(),
                None,
                remote_node_id.clone(),
                send,
                recv,
                false,
            )
            .await;
            let _ = app_for_task.emit(
                PEER_BI_STREAM_INCOMING_EVENT,
                IncomingPeerBiStreamEvent {
                    request_id: request_for_task.clone(),
                    stream_id: result.stream_id,
                    remote_node_id: remote_node_id.clone(),
                    transport_stable_id,
                },
            );
            if std::env::var("OPENRTC_NATIVE_STREAM_TRACE").ok().as_deref() == Some("1") {
                eprintln!(
                    "[openrtc-tauri][incoming-application-stream] remote_node_id={} transport_stable_id={} request_id={} phase=emitted",
                    remote_node_id,
                    transport_stable_id,
                    request_for_task
                );
            }
        }
    });

    state
        .subscriptions
        .lock()
        .await
        .insert(request_id.clone(), handle);
    Ok(request_id)
}

#[tauri::command]
async fn is_current_transport_stable_id(
    state: tauri::State<'_, OpenRtcTauriState>,
    endpoint_id: String,
    transport_stable_id: u64,
) -> Result<bool, String> {
    let is_current = state
        .client()
        .is_current_transport_stable_id_str(&endpoint_id, transport_stable_id)
        .await
        .map_err(|error| error.to_string())?;
    if std::env::var("OPENRTC_NATIVE_STREAM_TRACE").ok().as_deref() == Some("1") {
        eprintln!(
            "[openrtc-tauri][incoming-application-stream] remote_node_id={} transport_stable_id={} current={} phase=validate",
            endpoint_id, transport_stable_id, is_current
        );
    }
    Ok(is_current)
}

#[tauri::command]
async fn open_peer_bi_stream<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    peer_id: String,
    timeout_ms: Option<u64>,
) -> Result<OpenBiResult, String> {
    open_peer_bi_with(app, state, peer_id, timeout_ms, |client, peer_id, timeout_ms| async move {
        client.open_peer_bi(&peer_id, timeout_ms).await
    })
    .await
}

#[tauri::command]
async fn open_peer_bi_transport_only_stream<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    peer_id: String,
    timeout_ms: Option<u64>,
) -> Result<OpenBiResult, String> {
    open_peer_bi_with(
        app,
        state,
        peer_id,
        timeout_ms,
        |client, peer_id, timeout_ms| async move {
            client
                .open_peer_bi_transport_only(&peer_id, timeout_ms)
                .await
                .map(|(connection_id, remote_node_id, send, recv)| {
                    (
                        connection_id,
                        remote_node_id,
                        PeerSendStream::plain(send),
                        PeerRecvStream::plain(recv),
                    )
                })
        },
    )
    .await
}

#[tauri::command]
async fn open_peer_native_bi_stream<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    peer_id: String,
    label: String,
    timeout_ms: Option<u64>,
) -> Result<OpenBiResult, String> {
    let channel_envelope = openrtc::stream_metadata::encode_envelope(&label, None)
        .map_err(|error| error.to_string())?;
    open_peer_bi_with(
        app,
        state,
        peer_id,
        timeout_ms,
        move |client, peer_id, timeout_ms| async move {
            let (connection_id, remote_node_id, mut send, recv) =
                client.open_peer_bi(&peer_id, timeout_ms).await?;
            send.write_all(&channel_envelope).await?;
            Ok((connection_id, remote_node_id, send, recv))
        },
    )
    .await
}

#[tauri::command]
async fn write_peer_bi_stream(
    state: tauri::State<'_, OpenRtcTauriState>,
    stream_id: String,
    bytes: Vec<u8>,
) -> Result<(), String> {
    let stream_id = stream_id.trim().to_string();
    if stream_id.is_empty() {
        return Err("streamId is required".to_string());
    }

    let send = {
        let streams = state.peer_bi_streams.lock().await;
        streams
            .get(&stream_id)
            .map(|handle| handle.send.clone())
            .ok_or_else(|| format!("peer bi stream not found: {stream_id}"))?
    };
    let result = send
        .lock()
        .await
        .as_mut()
        .ok_or_else(|| format!("peer bi stream already closed: {stream_id}"))?
        .write_all(&bytes)
        .await
        .map_err(|error| format!("write peer bi stream failed: {error}"));
    result
}

#[tauri::command]
async fn start_peer_bi_stream_read<R: Runtime>(
    app: tauri::AppHandle<R>,
    state: tauri::State<'_, OpenRtcTauriState>,
    stream_id: String,
) -> Result<(), String> {
    let stream_id = stream_id.trim().to_string();
    if stream_id.is_empty() {
        return Err("streamId is required".to_string());
    }

    let mut streams = state.peer_bi_streams.lock().await;
    let handle = streams
        .get_mut(&stream_id)
        .ok_or_else(|| format!("peer bi stream not found: {stream_id}"))?;
    if handle.read_task.is_some() {
        return Ok(());
    }
    let recv = handle
        .recv
        .take()
        .ok_or_else(|| format!("peer bi stream reader already consumed: {stream_id}"))?;
    handle.read_task = Some(spawn_peer_bi_stream_reader(app, stream_id, recv));
    Ok(())
}

#[tauri::command]
async fn close_peer_bi_stream(
    state: tauri::State<'_, OpenRtcTauriState>,
    stream_id: String,
) -> Result<(), String> {
    let stream_id = stream_id.trim().to_string();
    if stream_id.is_empty() {
        return Err("streamId is required".to_string());
    }

    let handle = {
        let mut streams = state.peer_bi_streams.lock().await;
        streams.remove(&stream_id)
    };
    if let Some(handle) = handle {
        // Hold the writer mutex before finishing so close waits for any
        // in-flight chunk write instead of skipping the flush when Arc clones exist.
        let mut send = handle.send.lock().await;
        if let Some(send) = send.take() {
            let _ = send.finish();
        }
        drop(send);
        if let Some(read_task) = handle.read_task {
            read_task.abort();
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap as StdHashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex as StdMutex;
    use tokio::sync::oneshot;

    struct FakeTransportInstaller {
        installs: Arc<AtomicUsize>,
    }

    #[derive(Default)]
    struct FakeDeviceKeySigner {
        records: StdMutex<StdHashMap<String, String>>,
    }

    impl DeviceKeySigner for FakeDeviceKeySigner {
        fn public_jwk(&self, _app_tag: &str) -> Result<serde_json::Value, String> {
            Ok(serde_json::json!({
                "kty": "OKP",
                "crv": "Ed25519",
                "x": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
            }))
        }

        fn sign(&self, _app_tag: &str, challenge: &[u8]) -> Result<Vec<u8>, String> {
            assert_eq!(challenge, b"openrtc:v2:test");
            Ok(vec![7; 64])
        }

        fn delete(&self, _app_tag: &str) -> Result<(), String> {
            Ok(())
        }

        fn read_secure_record(&self, app_tag: &str, key: &str) -> Result<Option<String>, String> {
            Ok(self
                .records
                .lock()
                .unwrap()
                .get(&format!("{app_tag}:{key}"))
                .cloned())
        }

        fn write_secure_record(&self, app_tag: &str, key: &str, value: &str) -> Result<(), String> {
            self.records
                .lock()
                .unwrap()
                .insert(format!("{app_tag}:{key}"), value.to_string());
            Ok(())
        }

        fn delete_secure_record(&self, app_tag: &str, key: &str) -> Result<(), String> {
            self.records
                .lock()
                .unwrap()
                .remove(&format!("{app_tag}:{key}"));
            Ok(())
        }
    }

    impl TransportInstaller for FakeTransportInstaller {
        fn id(&self) -> &'static str {
            "ble"
        }

        fn is_requested(&self, config: &openrtc::client::TransportConfig) -> bool {
            config.ble.as_ref().is_some_and(|ble| ble.enabled)
        }

        fn install(
            &self,
            _client: Arc<openrtc::client::Client>,
            _config: openrtc::client::TransportConfig,
            _context: InstallContext,
        ) -> InstallFuture {
            self.installs.fetch_add(1, Ordering::SeqCst);
            Box::pin(async { Ok(Box::new(()) as Box<dyn std::any::Any + Send + Sync>) })
        }
    }

    #[derive(Debug)]
    struct UnavailableTransportInstaller;

    impl TransportInstaller for UnavailableTransportInstaller {
        fn id(&self) -> &'static str {
            "ble"
        }

        fn is_requested(&self, config: &openrtc::client::TransportConfig) -> bool {
            config.ble.as_ref().is_some_and(|ble| ble.enabled)
        }

        fn install(
            &self,
            _client: Arc<openrtc::client::Client>,
            _config: openrtc::client::TransportConfig,
            _context: InstallContext,
        ) -> InstallFuture {
            Box::pin(async { Err("Bluetooth hardware unavailable".to_string()) })
        }
    }

    fn requested_ble_config() -> openrtc::client::TransportConfig {
        openrtc::client::TransportConfig {
            ble: Some(openrtc::client::BleConfig {
                enabled: true,
                ..openrtc::client::BleConfig::default()
            }),
            ..openrtc::client::TransportConfig::default()
        }
    }

    fn test_config() -> OpenRtcTauriConfig {
        OpenRtcTauriConfig {
            api_key: format!("pk_test_{}", "a".repeat(40)),
            ..OpenRtcTauriConfig::default()
        }
    }

    #[test]
    fn native_device_proof_uses_host_signer_without_exporting_private_key() {
        let state = OpenRtcTauriState::new(test_config())
            .with_native_device_key_signer(Arc::new(FakeDeviceKeySigner::default()));
        let public = state
            .device_public_key("app_native_test")
            .expect("public key");
        assert_eq!(public["kty"], "OKP");
        assert_eq!(public["crv"], "Ed25519");
        assert!(public.get("d").is_none());
        let signature = state
            .sign_device_proof("app_native_test", "openrtc:v2:test")
            .expect("signature");
        assert_eq!(
            base64::engine::general_purpose::URL_SAFE_NO_PAD
                .decode(signature)
                .expect("base64"),
            vec![7; 64]
        );
    }

    #[test]
    fn native_certificate_records_round_trip_through_host_secure_storage() {
        let state = OpenRtcTauriState::new(test_config())
            .with_native_device_key_signer(Arc::new(FakeDeviceKeySigner::default()));
        let app_tag = "app_native_test";
        let key = "openrtc:v2:device-session:abc:device-1";
        assert_eq!(state.read_secure_record(app_tag, key).unwrap(), None);
        state
            .write_secure_record(app_tag, key, "{\"token\":\"bound\"}")
            .unwrap();
        assert_eq!(
            state.read_secure_record(app_tag, key).unwrap().as_deref(),
            Some("{\"token\":\"bound\"}")
        );
        state.delete_secure_record(app_tag, key).unwrap();
        assert_eq!(state.read_secure_record(app_tag, key).unwrap(), None);
    }

    #[test]
    fn native_device_proof_fails_closed_without_secure_host_signer() {
        let state = OpenRtcTauriState::new(test_config());
        assert!(state
            .device_public_key("app_native_test")
            .expect_err("missing signer must fail")
            .contains("secure-storage signer"));
    }

    fn native_transport_install_context() -> InstallContext {
        InstallContext {
            data_dir: std::env::temp_dir().join("openrtc-tauri-native-transport-tests"),
        }
    }

    fn managed_session_test_result(user_id: &str) -> StartSessionResult {
        StartSessionResult {
            local_node_id: format!("node-{user_id}"),
            ticket_scope: Some("user-device".to_string()),
            ticket: None,
            presence_started: true,
            auto_connect_started: true,
            local_device: openrtc::native_device::NativeDeviceIdentity {
                device_id: format!("device-{user_id}"),
                device_name: "Test Device".to_string(),
                created_at_ms: 1,
                updated_at_ms: 1,
                name_source: None,
                system_info: None,
            },
        }
    }

    async fn simulate_managed_session_start(
        state: Arc<OpenRtcTauriState>,
        user_id: String,
        pause: Option<(oneshot::Sender<()>, oneshot::Receiver<()>)>,
        starts: Arc<AtomicUsize>,
        stopped_owners: Arc<Mutex<Vec<usize>>>,
    ) -> SessionDisposition {
        let _start_guard = state.managed_session_start_guard.lock().await;
        let client = state.client();
        let key = format!("{}:{user_id}:device-{user_id}", client.app_tag());
        let (disposition, active_epoch) = {
            let active = state.managed_session.lock().await;
            let disposition = managed_session_disposition(
                active.as_ref().map(|record| record.key.as_str()),
                active
                    .as_ref()
                    .is_some_and(|record| Arc::ptr_eq(&record.owner_client, &client)),
                active
                    .as_ref()
                    .is_some_and(|record| record.result.presence_started),
                active
                    .as_ref()
                    .is_some_and(|record| record.result.auto_connect_started),
                &key,
                true,
                true,
            );
            (
                disposition,
                active.as_ref().map(|record| record.owner_epoch),
            )
        };

        if disposition == SessionDisposition::Reuse {
            return disposition;
        }

        if disposition == SessionDisposition::Replace {
            let previous = state
                .managed_session
                .lock()
                .await
                .take()
                .expect("replacement must have a previous owner");
            stopped_owners
                .lock()
                .await
                .push(Arc::as_ptr(&previous.owner_client) as usize);
        }

        starts.fetch_add(1, Ordering::SeqCst);
        if let Some((entered, release)) = pause {
            entered.send(()).expect("start observer must be waiting");
            release.await.expect("start release must be sent");
        }

        let owner_epoch = match disposition {
            SessionDisposition::Refresh => {
                active_epoch.expect("refresh must preserve the active owner epoch")
            }
            SessionDisposition::Start | SessionDisposition::Replace => {
                state.allocate_managed_session_owner_epoch()
            }
            SessionDisposition::Reuse => unreachable!("reuse returned before startup"),
        };
        let result = managed_session_test_result(&user_id);
        *state.managed_session.lock().await = Some(ManagedSessionRecord {
            key,
            owner_client: client,
            owner_epoch,
            result,
        });
        disposition
    }

    #[tokio::test]
    async fn concurrent_same_key_start_reuses_one_owner_epoch() {
        let state = Arc::new(OpenRtcTauriState::new(test_config()));
        let starts = Arc::new(AtomicUsize::new(0));
        let stopped_owners = Arc::new(Mutex::new(Vec::new()));
        let (first_entered_tx, first_entered_rx) = oneshot::channel();
        let (first_release_tx, first_release_rx) = oneshot::channel();

        let first = tokio::spawn(simulate_managed_session_start(
            state.clone(),
            "same-user".to_string(),
            Some((first_entered_tx, first_release_rx)),
            starts.clone(),
            stopped_owners.clone(),
        ));
        first_entered_rx.await.expect("first start must pause");

        let (second_attempting_tx, second_attempting_rx) = oneshot::channel();
        let second_state = state.clone();
        let second_starts = starts.clone();
        let second_stopped_owners = stopped_owners.clone();
        let second = tokio::spawn(async move {
            second_attempting_tx.send(()).unwrap();
            simulate_managed_session_start(
                second_state,
                "same-user".to_string(),
                None,
                second_starts,
                second_stopped_owners,
            )
            .await
        });
        second_attempting_rx.await.unwrap();
        tokio::task::yield_now().await;
        assert_eq!(starts.load(Ordering::SeqCst), 1);

        first_release_tx.send(()).unwrap();
        assert_eq!(
            first.await.expect("first start task"),
            SessionDisposition::Start
        );
        assert_eq!(
            second.await.expect("second start task"),
            SessionDisposition::Reuse
        );
        assert_eq!(starts.load(Ordering::SeqCst), 1);

        let active = state
            .managed_session
            .lock()
            .await
            .clone()
            .expect("active owner");
        assert_eq!(active.owner_epoch, 1);
        assert!(Arc::ptr_eq(&active.owner_client, &state.client()));
        assert!(stopped_owners.lock().await.is_empty());
    }

    #[tokio::test]
    async fn concurrent_different_key_start_replaces_prior_owner_without_late_overwrite() {
        let state = Arc::new(OpenRtcTauriState::new(test_config()));
        let starts = Arc::new(AtomicUsize::new(0));
        let stopped_owners = Arc::new(Mutex::new(Vec::new()));
        let (first_entered_tx, first_entered_rx) = oneshot::channel();
        let (first_release_tx, first_release_rx) = oneshot::channel();

        let first = tokio::spawn(simulate_managed_session_start(
            state.clone(),
            "first-user".to_string(),
            Some((first_entered_tx, first_release_rx)),
            starts.clone(),
            stopped_owners.clone(),
        ));
        first_entered_rx.await.expect("first start must pause");
        let first_owner = state.client();

        let (second_attempting_tx, second_attempting_rx) = oneshot::channel();
        let second_state = state.clone();
        let second_starts = starts.clone();
        let second_stopped_owners = stopped_owners.clone();
        let second = tokio::spawn(async move {
            second_attempting_tx.send(()).unwrap();
            simulate_managed_session_start(
                second_state,
                "second-user".to_string(),
                None,
                second_starts,
                second_stopped_owners,
            )
            .await
        });
        second_attempting_rx.await.unwrap();
        tokio::task::yield_now().await;
        assert!(Arc::ptr_eq(&first_owner, &state.client()));

        first_release_tx.send(()).unwrap();
        assert_eq!(
            first.await.expect("first start task"),
            SessionDisposition::Start
        );
        assert_eq!(
            second.await.expect("replacement start task"),
            SessionDisposition::Replace
        );
        assert_eq!(starts.load(Ordering::SeqCst), 2);

        let active = state
            .managed_session
            .lock()
            .await
            .clone()
            .expect("active owner");
        assert_eq!(active.owner_epoch, 2);
        assert!(active.key.contains("second-user"));
        assert!(Arc::ptr_eq(&active.owner_client, &state.client()));
        assert_eq!(
            stopped_owners.lock().await.as_slice(),
            [Arc::as_ptr(&first_owner) as usize]
        );
        assert!(Arc::ptr_eq(&active.owner_client, &first_owner));
    }

    #[test]
    fn derives_app_tag_from_api_key_by_default() {
        let config = OpenRtcTauriConfig {
            api_key: format!("pk_test_{}", "b".repeat(24) + "1234567890abcdef"),
            ..OpenRtcTauriConfig::default()
        };

        assert_eq!(config.app_tag().unwrap(), "app_1234567890abcdef");
    }

    #[tokio::test]
    async fn requested_native_transport_without_installer_keeps_base_route_available() {
        let state = OpenRtcTauriState::new(test_config());
        let client = state.client();
        ensure_requested_native_transports(
            &state,
            &client,
            Some(&requested_ble_config()),
            &native_transport_install_context(),
        )
        .await
        .expect("an unavailable optional transport must not fail base Iroh startup");

        assert!(state.installed_native_transports.lock().await.is_empty());
    }

    #[tokio::test]
    async fn native_transport_installer_is_idempotent_for_one_client() {
        let installs = Arc::new(AtomicUsize::new(0));
        let state = OpenRtcTauriState::new(test_config()).with_native_transport_installer(
            Arc::new(FakeTransportInstaller {
                installs: installs.clone(),
            }),
        );
        let client = state.client();
        let config = requested_ble_config();

        ensure_requested_native_transports(
            &state,
            &client,
            Some(&config),
            &native_transport_install_context(),
        )
        .await
        .expect("first install");
        ensure_requested_native_transports(
            &state,
            &client,
            Some(&config),
            &native_transport_install_context(),
        )
        .await
        .expect("idempotent install");

        assert_eq!(installs.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn unavailable_native_transport_keeps_base_route_available() {
        let state = OpenRtcTauriState::new(test_config())
            .with_native_transport_installer(Arc::new(UnavailableTransportInstaller));
        let client = state.client();

        ensure_requested_native_transports(
            &state,
            &client,
            Some(&requested_ble_config()),
            &native_transport_install_context(),
        )
        .await
        .expect("optional transport installation failure must not fail base Iroh startup");

        assert!(state.installed_native_transports.lock().await.is_empty());
    }

    #[test]
    fn invalid_or_missing_api_key_fails_closed() {
        assert!(OpenRtcTauriConfig::default().validated_api_key().is_err());
        let mut config = OpenRtcTauriConfig::default();
        config.api_key = "pk_test_short".to_string();
        assert!(config.validated_api_key().is_err());
    }

    #[test]
    fn from_env_uses_api_key_and_ignores_legacy_namespace_selectors() {
        let previous_api_key = std::env::var("VITE_OPENRTC_API_KEY").ok();
        let previous_project = std::env::var("VITE_OPENRTC_PROJECT_ID").ok();
        let previous_app_tag = std::env::var("VITE_PLUTO_OPENRTC_APP_TAG").ok();
        let api_key = format!("pk_test_{}", "c".repeat(40));
        std::env::set_var("VITE_OPENRTC_API_KEY", &api_key);
        std::env::set_var("VITE_OPENRTC_PROJECT_ID", "pluto-rtc-prod");
        std::env::set_var("VITE_PLUTO_OPENRTC_APP_TAG", "app_from_vite_env");

        let config = OpenRtcTauriConfig::from_env();

        assert_eq!(config.api_key, api_key);
        assert_eq!(
            config.app_tag().unwrap(),
            openrtc::app_tag_from_api_key(&api_key)
        );

        match previous_api_key {
            Some(value) => std::env::set_var("VITE_OPENRTC_API_KEY", value),
            None => std::env::remove_var("VITE_OPENRTC_API_KEY"),
        }
        match previous_project {
            Some(value) => std::env::set_var("VITE_OPENRTC_PROJECT_ID", value),
            None => std::env::remove_var("VITE_OPENRTC_PROJECT_ID"),
        }
        match previous_app_tag {
            Some(value) => std::env::set_var("VITE_PLUTO_OPENRTC_APP_TAG", value),
            None => std::env::remove_var("VITE_PLUTO_OPENRTC_APP_TAG"),
        }
    }

    #[test]
    fn native_state_has_one_constructor_owned_app_identity() {
        let config = test_config();
        let expected = openrtc::app_tag_from_api_key(&config.api_key);
        let state = OpenRtcTauriState::new(config);
        let first = state.client();
        let second = state.client();

        assert_eq!(first.app_tag(), expected);
        assert!(Arc::ptr_eq(&first, &second));
    }

    #[test]
    fn managed_session_uses_persisted_native_device_id() {
        let identity = openrtc::native_device::NativeDeviceIdentity {
            device_id: "persisted-native-device".to_string(),
            device_name: "Mac".to_string(),
            created_at_ms: 1,
            updated_at_ms: 1,
            name_source: None,
            system_info: None,
        };

        assert_eq!(
            managed_session_device_id(Some(" desktop-e2e-native "), &identity),
            "persisted-native-device"
        );
        assert_eq!(
            managed_session_device_id(None, &identity),
            "persisted-native-device"
        );
    }

    #[test]
    fn repeated_managed_session_triggers_converge_on_one_owner() {
        let key = "app:user:persisted-native-device";
        let cases = [
            ("react-remount", true, true, true, true),
            ("hmr", true, true, true, true),
            ("auth-refresh", true, true, true, true),
            ("resume", true, true, true, true),
            ("alias-change", true, true, true, true),
        ];
        for (label, active_presence, active_auto, requested_presence, requested_auto) in cases {
            assert_eq!(
                managed_session_disposition(
                    Some(key),
                    true,
                    active_presence,
                    active_auto,
                    key,
                    requested_presence,
                    requested_auto,
                ),
                SessionDisposition::Reuse,
                "{label} must reuse the authoritative tuple"
            );
        }

        assert_eq!(
            managed_session_disposition(Some(key), true, false, true, key, true, true),
            SessionDisposition::Refresh,
            "failed presence startup must retry idempotently"
        );
        assert_eq!(
            managed_session_disposition(
                Some(key),
                true,
                true,
                true,
                "app:other-user:persisted-native-device",
                true,
                true,
            ),
            SessionDisposition::Replace,
            "user switching must replace the previous lifecycle owner"
        );
        assert_eq!(
            managed_session_disposition(Some(key), false, true, true, key, true, true),
            SessionDisposition::Replace,
            "the same tuple on a different client epoch must replace the previous owner"
        );
    }

    #[test]
    fn user_device_revocation_invalidates_the_cached_managed_ticket() {
        assert!(revokes_managed_session("user-device"));
        assert!(revokes_managed_session("  user-device  "));
        assert!(!revokes_managed_session("share:example"));
        assert!(!revokes_managed_session(""));
    }

    #[test]
    fn managed_session_metadata_uses_authoritative_device_id() {
        let raw = serde_json::json!({
            "deviceId": "persisted-native-device",
            "assistantDevice": {
                "deviceId": "desktop-e2e-native"
            }
        })
        .to_string();

        let metadata =
            metadata_with_authoritative_device_id(Some(raw), "desktop-e2e-native").unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&metadata).unwrap();

        assert_eq!(parsed["deviceId"], "desktop-e2e-native");
        assert_eq!(parsed["assistantDevice"]["deviceId"], "desktop-e2e-native");
    }
}