openrtc 2.0.0-rc.16

//! Provider-neutral native coordination-gateway adapter.
//!
//! The adapter owns only authenticated gateway delivery: credential exchange,
//! WebSocket subscription, bounded reconnect, presence publication, and
//! signaling event projection. Rust `Client` remains the sole peer lifecycle
//! and transport authority.

use crate::native_v2::NativeControlPlaneHttpError;
use crate::signaling::{
    Device, DeviceCapabilities, DeviceEvent, SessionEvent, SignalingBackend, SignalingEnvelope,
    SignalingSession,
};
use anyhow::{anyhow, bail, Context, Result};
use async_trait::async_trait;
use base64::Engine as _;
use futures::{stream::BoxStream, Sink, SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::error::Error as StdError;
use std::fmt;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{broadcast, mpsc, oneshot, RwLock};
use tokio_tungstenite::tungstenite::{
    client::IntoClientRequest,
    http::{HeaderValue, Uri},
    Message,
};
use uuid::Uuid;

const WIRE_PROTOCOL_VERSION: u8 = 1;
const GRANT_PROTOCOL_VERSION: u8 = 2;
const GATEWAY_PROTOCOL: &str = "openrtc.v1";
const GATEWAY_AUTH_PROTOCOL_PREFIX: &str = "openrtc.auth.";
const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
const ACK_TIMEOUT: Duration = Duration::from_secs(15);
const AUTH_REFRESH_SKEW_MS: u64 = 5 * 60_000;
// Keep an otherwise idle native connection alive below common 30-60 minute
// NAT/proxy idle ceilings. Cloudflare handles protocol ping frames at the
// WebSocket edge without waking a hibernating Durable Object, so this does not
// create a presence write, logical usage event, or recurring DO execution.
const SOCKET_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(10 * 60);
const MAX_RECONNECT_ATTEMPTS: u8 = 12;
const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30);
const MAX_EXCLUDED_PEERS: usize = 250;
pub const OPENRTC_PRODUCTION_COORDINATION_GATEWAY: &str = "https://gateway.openrtc.app";

#[derive(Debug)]
struct GatewayConnectError {
    message: String,
    retryable: bool,
}

impl fmt::Display for GatewayConnectError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl StdError for GatewayConnectError {}

fn gateway_connect_error(message: impl Into<String>, retryable: bool) -> anyhow::Error {
    anyhow!(GatewayConnectError {
        message: message.into(),
        retryable,
    })
}

fn is_retryable_gateway_error(error: &anyhow::Error) -> bool {
    if let Some(classified) = error.downcast_ref::<GatewayConnectError>() {
        return classified.retryable;
    }
    if let Some(control_plane) = error
        .chain()
        .find_map(|source| source.downcast_ref::<NativeControlPlaneHttpError>())
    {
        return control_plane.is_retryable();
    }
    true
}

fn terminal_refresh_error(
    lease_error: anyhow::Error,
    fallback_error: Option<anyhow::Error>,
) -> anyhow::Error {
    fallback_error.unwrap_or(lease_error)
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeCoordinationAvenue {
    pub kind: String,
    pub id: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NativeGatewayGrantRequest {
    pub avenue: NativeCoordinationAvenue,
    pub device_id: String,
    pub runtime_instance_id: String,
    pub ticket_fingerprint: String,
    pub purpose: String,
    pub refresh_grant: Option<String>,
}

#[async_trait]
pub trait NativeGatewayGrantProvider: Send + Sync {
    async fn grant(&self, request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant>;
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeGatewayGrant {
    pub protocol_version: u8,
    pub gateway_url: String,
    pub route_key: String,
    pub token: String,
    pub expires_at_ms: u64,
}

/// Configuration for a pure-native OpenRTC 2.0 consumer. Identity assertion,
/// device proof, attestation, and secure storage remain host responsibilities;
/// the provider returns only OpenRTC-issued, avenue-bound grants.
pub struct NativeCoordinationGatewayOptions {
    pub endpoint: String,
    pub app_tag: String,
    pub device_id: String,
    pub platform_type: String,
    pub avenue: NativeCoordinationAvenue,
    pub grant_provider: Arc<dyn NativeGatewayGrantProvider>,
}

/// Lightweight, network-idle entrypoint for pure-Rust OpenRTC 2.0 capability
/// activation. It mirrors the public SDK namespaces without putting consumer
/// authentication, attestation, or provider-specific types into the runtime.
pub struct NativeCapabilities {
    endpoint: String,
    app_tag: String,
    device_id: String,
    platform_type: String,
    grant_provider: Arc<dyn NativeGatewayGrantProvider>,
}

impl NativeCapabilities {
    /// Construct the production capability namespace from a public developer
    /// API key. No grant, socket, timer, or provider request occurs here.
    pub fn new(
        api_key: &str,
        device_id: impl Into<String>,
        platform_type: impl Into<String>,
        grant_provider: Arc<dyn NativeGatewayGrantProvider>,
    ) -> Result<Self> {
        let api_key = crate::validate_v2_public_api_key(api_key)?;
        Ok(Self {
            endpoint: OPENRTC_PRODUCTION_COORDINATION_GATEWAY.to_string(),
            app_tag: crate::app_tag_from_api_key(api_key),
            device_id: required("device_id", device_id.into())?,
            platform_type: required("platform_type", platform_type.into())?,
            grant_provider,
        })
    }

    #[cfg(any(test, feature = "testing-endpoints"))]
    pub fn with_testing_endpoint(mut self, endpoint: impl Into<String>) -> Result<Self> {
        self.endpoint = validate_endpoint("endpoint", endpoint.into(), true)?;
        Ok(self)
    }

    pub fn devices(&self, principal_id: impl Into<String>) -> Result<NativeCapabilityHandle> {
        self.open("user", principal_id)
    }

    pub fn join_space(&self, id: impl Into<String>) -> Result<NativeCapabilityHandle> {
        self.open("space", id)
    }

    pub fn join_room(&self, id: impl Into<String>) -> Result<NativeCapabilityHandle> {
        self.open("room", id)
    }

    pub fn issue_ticket(&self, id: impl Into<String>) -> Result<NativeCapabilityHandle> {
        self.open("session", id)
    }

    fn open(&self, kind: &'static str, id: impl Into<String>) -> Result<NativeCapabilityHandle> {
        NativeCapabilityHandle::new(NativeCoordinationGatewayOptions {
            endpoint: self.endpoint.clone(),
            app_tag: self.app_tag.clone(),
            device_id: self.device_id.clone(),
            platform_type: self.platform_type.clone(),
            avenue: NativeCoordinationAvenue {
                kind: kind.to_string(),
                id: id.into(),
            },
            grant_provider: self.grant_provider.clone(),
        })
    }
}

/// Disposable OpenRTC 2.0 native avenue.
///
/// A handle owns exactly one coordination adapter and therefore exactly one
/// devices, space, room, or ticket avenue. Constructing the root `Client` does
/// not create one of these handles and performs no network work. The socket is
/// opened only when the runtime publishes its initial presence through the
/// adapter. Closing the handle stops grant refresh, reconnect, and presence
/// ownership for that avenue without replacing the shared peer runtime.
pub struct NativeCapabilityHandle {
    kind: NativeCapabilityKind,
    id: String,
    signaling: Arc<NativeCoordinationGatewaySignaling>,
    closed: Arc<AtomicBool>,
}

pub(crate) struct NativeCapabilityCloser {
    signaling: Arc<NativeCoordinationGatewaySignaling>,
    closed: Arc<AtomicBool>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NativeCapabilityKind {
    Devices,
    Space,
    Room,
    Ticket,
}

impl NativeCapabilityKind {
    fn from_avenue_kind(value: &str) -> Result<Self> {
        match value {
            "user" => Ok(Self::Devices),
            "space" => Ok(Self::Space),
            "room" => Ok(Self::Room),
            "session" => Ok(Self::Ticket),
            _ => bail!("native coordination avenue kind is invalid"),
        }
    }
}

impl NativeCapabilityHandle {
    /// Create one inactive native capability handle. This allocates the local
    /// actor only; it does not request a grant or connect to the gateway.
    pub fn new(options: NativeCoordinationGatewayOptions) -> Result<Self> {
        let kind = NativeCapabilityKind::from_avenue_kind(&options.avenue.kind)?;
        let id = required("avenue.id", options.avenue.id.clone())?;
        let signaling = NativeCoordinationGatewaySignaling::new(options)?;
        Ok(Self {
            kind,
            id,
            signaling,
            closed: Arc::new(AtomicBool::new(false)),
        })
    }

    pub fn kind(&self) -> NativeCapabilityKind {
        self.kind
    }

    pub fn id(&self) -> &str {
        &self.id
    }

    pub fn signaling(&self) -> Arc<dyn SignalingBackend> {
        self.signaling.clone()
    }

    pub fn is_closed(&self) -> bool {
        self.closed.load(Ordering::Acquire)
    }

    pub async fn close(&self) {
        if !self.closed.swap(true, Ordering::AcqRel) {
            self.signaling.stop().await;
        }
    }

    pub(crate) fn closer(&self) -> NativeCapabilityCloser {
        NativeCapabilityCloser {
            signaling: self.signaling.clone(),
            closed: self.closed.clone(),
        }
    }
}

impl NativeCapabilityCloser {
    pub(crate) async fn close(&self) {
        if !self.closed.swap(true, Ordering::AcqRel) {
            self.signaling.stop().await;
        }
    }
}

impl Drop for NativeCapabilityHandle {
    fn drop(&mut self) {
        if !self.closed.swap(true, Ordering::AcqRel) {
            let _ = self.signaling.commands.try_send(Command::Stop);
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct DesiredPresence {
    user_id: String,
    local_node_id: String,
    ticket: String,
    device_name: String,
    metadata: Option<String>,
    ttl_ms: u64,
    online: bool,
}

#[derive(Debug, Clone, Default, Serialize)]
#[serde(rename_all = "camelCase")]
struct DevicePatch {
    #[serde(skip_serializing_if = "Option::is_none")]
    device_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    capabilities: Option<DeviceCapabilities>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    excluded_peers: Option<Vec<String>>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct OutboundGatewayDevice {
    device_id: String,
    runtime_instance_id: String,
    node_id: String,
    device_name: String,
    platform_type: String,
    ticket: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    metadata: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    capabilities: Option<DeviceCapabilities>,
    excluded_peers: Vec<String>,
    online: bool,
}

#[derive(Debug)]
enum Command {
    Publish {
        desired: DesiredPresence,
        reply: oneshot::Sender<Result<()>>,
    },
    Patch {
        patch: DevicePatch,
        reply: oneshot::Sender<Result<()>>,
    },
    Offline {
        reply: oneshot::Sender<Result<()>>,
    },
    Delete {
        user_id: String,
        device_id: String,
        reply: oneshot::Sender<Result<()>>,
    },
    SendSignal {
        target_device_id: String,
        payload: String,
        state: Option<String>,
        reply_payload: Option<String>,
        reply: oneshot::Sender<Result<String>>,
    },
    PutSession {
        session_id: String,
        session: serde_json::Value,
        expires_at_ms: i64,
        reply: oneshot::Sender<Result<()>>,
    },
    Stop,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct GatewayDevice {
    #[serde(default)]
    user_id: Option<String>,
    device_id: String,
    runtime_instance_id: String,
    node_id: String,
    device_name: String,
    platform_type: String,
    ticket: String,
    #[serde(default)]
    metadata: Option<String>,
    #[serde(default)]
    capabilities: Option<DeviceCapabilities>,
    #[serde(default)]
    excluded_peers: Vec<String>,
    online: bool,
    updated_at_ms: i64,
    expires_at_ms: i64,
}

#[derive(Debug, Deserialize)]
#[serde(tag = "type")]
enum ServerFrame {
    #[serde(rename = "ready")]
    Ready {
        #[serde(rename = "budgetRemainingMicrousd")]
        _budget_remaining_microusd: i64,
        #[serde(default, rename = "leaseRefreshMode")]
        lease_refresh_mode: Option<String>,
    },
    #[serde(rename = "auth.refreshed")]
    AuthRefreshed {
        #[serde(rename = "expiresAtMs")]
        _expires_at_ms: u64,
    },
    #[serde(rename = "lease.refreshed")]
    LeaseRefreshed {
        #[serde(rename = "idempotencyKey")]
        idempotency_key: String,
        #[serde(rename = "expiresAtMs")]
        expires_at_ms: u64,
    },
    #[serde(rename = "roster.snapshot")]
    RosterSnapshot { devices: Vec<GatewayDevice> },
    #[serde(rename = "presence.changed")]
    PresenceChanged {
        operation: String,
        device: GatewayDevice,
    },
    #[serde(rename = "session.changed")]
    SessionChanged {
        operation: String,
        #[serde(rename = "sessionId")]
        session_id: String,
        #[serde(default)]
        session: Option<serde_json::Value>,
    },
    #[serde(rename = "ack")]
    Ack {
        #[serde(rename = "idempotencyKey")]
        idempotency_key: String,
    },
    #[serde(rename = "error")]
    Error {
        code: String,
        message: String,
        #[serde(default, rename = "idempotencyKey")]
        idempotency_key: Option<String>,
        retryable: bool,
    },
    #[serde(rename = "signal.received")]
    SignalReceived {
        #[serde(rename = "signalId")]
        _signal_id: String,
        #[serde(rename = "senderDeviceId")]
        sender_device_id: String,
        payload: String,
        #[serde(default)]
        state: Option<String>,
        #[serde(default, rename = "replyPayload")]
        reply_payload: Option<String>,
        #[serde(rename = "createdAtMs")]
        _created_at_ms: i64,
    },
    #[serde(rename = "pong")]
    Pong,
}

struct SharedState {
    desired: RwLock<Option<DesiredPresence>>,
    applied_presence: RwLock<Option<DesiredPresence>>,
    staged_patch: RwLock<DevicePatch>,
    devices: RwLock<HashMap<String, Device>>,
    device_events: broadcast::Sender<Vec<DeviceEvent>>,
    session_events: broadcast::Sender<Vec<SessionEvent>>,
    pending_messages: Mutex<Vec<SignalingEnvelope>>,
}

pub struct NativeCoordinationGatewaySignaling {
    endpoint: String,
    app_tag: String,
    device_id: String,
    avenue: NativeCoordinationAvenue,
    runtime_instance_id: String,
    platform_type: String,
    grant_provider: Arc<dyn NativeGatewayGrantProvider>,
    shared: Arc<SharedState>,
    commands: mpsc::Sender<Command>,
}

impl NativeCoordinationGatewaySignaling {
    pub fn new(options: NativeCoordinationGatewayOptions) -> Result<Arc<Self>> {
        let endpoint = validate_endpoint("endpoint", options.endpoint, true)?;
        let app_tag = required("app_tag", options.app_tag)?;
        let device_id = required("device_id", options.device_id)?;
        let platform_type = required("platform_type", options.platform_type)?;
        let avenue = options.avenue;
        if !matches!(avenue.kind.as_str(), "user" | "space" | "room" | "session") {
            bail!("native coordination avenue kind is invalid");
        }
        let avenue_id = required("avenue.id", avenue.id)?;
        let (device_events, _) = broadcast::channel(64);
        let (session_events, _) = broadcast::channel(64);
        let shared = Arc::new(SharedState {
            desired: RwLock::new(None),
            applied_presence: RwLock::new(None),
            staged_patch: RwLock::new(DevicePatch::default()),
            devices: RwLock::new(HashMap::new()),
            device_events,
            session_events,
            pending_messages: Mutex::new(Vec::new()),
        });
        let (commands, receiver) = mpsc::channel(64);
        let adapter = Arc::new(Self {
            endpoint,
            app_tag,
            device_id,
            avenue: NativeCoordinationAvenue {
                kind: avenue.kind,
                id: avenue_id,
            },
            runtime_instance_id: format!("runtime:{}", Uuid::new_v4()),
            platform_type,
            grant_provider: options.grant_provider,
            shared,
            commands,
        });
        tokio::spawn(run_actor(adapter.clone(), receiver));
        Ok(adapter)
    }

    pub async fn stop(&self) {
        let _ = self.commands.send(Command::Stop).await;
    }

    async fn request_unit(
        &self,
        build: impl FnOnce(oneshot::Sender<Result<()>>) -> Command,
    ) -> Result<()> {
        let (reply, response) = oneshot::channel();
        self.commands
            .send(build(reply))
            .await
            .map_err(|_| anyhow!("coordination gateway actor stopped"))?;
        response
            .await
            .map_err(|_| anyhow!("coordination gateway actor dropped its reply"))?
    }

    async fn mint_credential(
        &self,
        desired: &DesiredPresence,
        purpose: &str,
        refresh_grant: Option<String>,
    ) -> Result<NativeGatewayGrant> {
        let ticket_fingerprint = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .encode(Sha256::digest(desired.ticket.as_bytes()));
        let credential = self
            .grant_provider
            .grant(NativeGatewayGrantRequest {
                avenue: self.avenue.clone(),
                device_id: self.device_id.clone(),
                runtime_instance_id: self.runtime_instance_id.clone(),
                ticket_fingerprint,
                purpose: purpose.to_string(),
                refresh_grant,
            })
            .await
            .context("obtain native coordination grant")?;
        if credential.protocol_version != GRANT_PROTOCOL_VERSION {
            return Err(gateway_connect_error(
                "unsupported native coordination protocol",
                false,
            ));
        }
        let configured = reqwest::Url::parse(&self.endpoint)?;
        let returned = reqwest::Url::parse(&credential.gateway_url)?;
        if normalized_origin_scheme(configured.scheme())
            != normalized_origin_scheme(returned.scheme())
            || configured.host_str() != returned.host_str()
            || configured.port_or_known_default() != returned.port_or_known_default()
        {
            return Err(gateway_connect_error(
                "native coordination credential returned an unexpected gateway origin",
                false,
            ));
        }
        Ok(credential)
    }

    fn gateway_url(&self, credential: &NativeGatewayGrant) -> Result<String> {
        let mut url = reqwest::Url::parse(&credential.gateway_url)?;
        match url.scheme() {
            "https" => url
                .set_scheme("wss")
                .map_err(|_| anyhow!("invalid gateway scheme"))?,
            "http" => url
                .set_scheme("ws")
                .map_err(|_| anyhow!("invalid gateway scheme"))?,
            "wss" | "ws" => {}
            _ => bail!("native coordination gateway must use HTTPS/WSS"),
        }
        if url.query().is_some() || url.fragment().is_some() || !url.username().is_empty() {
            bail!("native coordination gateway URL contains forbidden credentials or parameters");
        }
        let path = format!(
            "{}/v{}/connect/{}",
            url.path().trim_end_matches('/'),
            credential.protocol_version,
            credential.route_key
        );
        url.set_path(&path);
        Ok(url.to_string())
    }

    fn device_from_gateway(&self, value: GatewayDevice) -> Device {
        Device {
            app_tag: Some(self.app_tag.clone()),
            device_id: value.device_id,
            user_id: value.user_id,
            device_name: value.device_name,
            platform_type: Some(value.platform_type),
            capabilities: value.capabilities,
            session_id: Some(value.runtime_instance_id),
            node_id: Some(value.node_id),
            tag: None,
            kind: None,
            metadata: value.metadata,
            online: value.online,
            ticket: Some(value.ticket),
            last_seen_at: Some(serde_json::json!(value.updated_at_ms)),
            expires_at: Some(serde_json::json!(value.expires_at_ms)),
            created_at: None,
            updated_at: Some(serde_json::json!(value.updated_at_ms)),
            excluded_peers: value.excluded_peers,
        }
    }
}

type GatewaySocket =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

struct ConnectedGateway {
    socket: GatewaySocket,
    credential_expires_at_ms: u64,
    credential_token: String,
    lease_refresh_mode: LeaseRefreshMode,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LeaseRefreshMode {
    Off,
    Shadow,
    Active,
}

impl LeaseRefreshMode {
    fn from_wire(value: Option<&str>) -> Self {
        match value {
            Some("shadow") => Self::Shadow,
            Some("active") => Self::Active,
            _ => Self::Off,
        }
    }
}

async fn run_actor(
    adapter: Arc<NativeCoordinationGatewaySignaling>,
    mut commands: mpsc::Receiver<Command>,
) {
    let mut socket: Option<ConnectedGateway> = None;
    let mut reconnect_attempt = 0_u8;
    let mut retry_at: Option<tokio::time::Instant> = None;
    let mut pending_publish: Option<(DesiredPresence, oneshot::Sender<Result<()>>)> = None;
    let mut circuit_failure: Option<(DesiredPresence, String)> = None;
    loop {
        if socket.is_none() {
            if pending_publish
                .as_ref()
                .is_some_and(|(_, reply)| reply.is_closed())
            {
                pending_publish = None;
                retry_at = None;
                reconnect_attempt = 0;
                *adapter.shared.desired.write().await = None;
                *adapter.shared.applied_presence.write().await = None;
            }
            let retry_delay = retry_at
                .map(|deadline| deadline.saturating_duration_since(tokio::time::Instant::now()))
                .unwrap_or(Duration::from_secs(365 * 24 * 60 * 60));
            tokio::select! {
                command = commands.recv() => {
                    let Some(command) = command else { break; };
                    match command {
                        Command::Publish { desired, reply } => {
                            if let Some((blocked, message)) = circuit_failure.as_ref() {
                                if blocked == &desired {
                                    let _ = reply.send(Err(anyhow!(message.clone())));
                                    continue;
                                }
                            }
                            circuit_failure = None;
                            if let Some((_, previous_reply)) = pending_publish.take() {
                                let _ = previous_reply.send(Err(anyhow!(
                                    "native coordination publication was superseded"
                                )));
                            }
                            *adapter.shared.desired.write().await = Some(desired.clone());
                            *adapter.shared.applied_presence.write().await = None;
                            pending_publish = Some((desired, reply));
                            reconnect_attempt = 0;
                            retry_at = Some(tokio::time::Instant::now());
                        }
                        Command::Stop => {
                            if let Some((_, reply)) = pending_publish.take() {
                                let _ = reply.send(Err(anyhow!(
                                    "native coordination gateway stopped"
                                )));
                            }
                            break;
                        }
                        Command::Delete { user_id, device_id, reply } => {
                            let result = execute_control_delete(
                                &adapter,
                                &user_id,
                                &device_id,
                            )
                            .await;
                            let _ = reply.send(result);
                        }
                        other => reject_command(
                            other,
                            if retry_at.is_some() {
                                "native coordination gateway is reconnecting"
                            } else {
                                "native coordination gateway is not connected"
                            },
                        ),
                    }
                }
                _ = tokio::time::sleep(retry_delay), if retry_at.is_some() => {
                    retry_at = None;
                    let desired = adapter.shared.desired.read().await.clone();
                    let Some(desired) = desired else {
                        reconnect_attempt = 0;
                        continue;
                    };
                    // Gateway authentication validates and persists the full
                    // device projection. `ready` is therefore the initial
                    // publication acknowledgement; sending presence.upsert
                    // here would duplicate the write and customer charge.
                    let result = connect(&adapter).await;
                    match result {
                        Ok(connected) => {
                            reconnect_attempt = 0;
                            circuit_failure = None;
                            *adapter.shared.applied_presence.write().await = Some(desired.clone());
                            socket = Some(connected);
                            if let Some((published, reply)) = pending_publish.take() {
                                if published == desired {
                                    let _ = reply.send(Ok(()));
                                } else {
                                    let _ = reply.send(Err(anyhow!(
                                        "native coordination publication was superseded"
                                    )));
                                }
                            }
                        }
                        Err(error) => {
                            reconnect_attempt = reconnect_attempt.saturating_add(1);
                            let retryable = is_retryable_gateway_error(&error);
                            eprintln!(
                                "[openrtc][coordination-gateway][connect-retry] attempt={}/{} retryable={} error={}",
                                reconnect_attempt,
                                MAX_RECONNECT_ATTEMPTS,
                                retryable,
                                error
                            );
                            if !retryable || reconnect_attempt >= MAX_RECONNECT_ATTEMPTS {
                                retry_at = None;
                                let failure = format!(
                                    "native coordination gateway unavailable after {} attempts: {}",
                                    reconnect_attempt, error
                                );
                                circuit_failure = Some((desired, failure.clone()));
                                if let Some((_, reply)) = pending_publish.take() {
                                    let _ = reply.send(Err(anyhow!(failure)));
                                }
                            } else {
                                retry_at = Some(
                                    tokio::time::Instant::now()
                                        + reconnect_delay(reconnect_attempt),
                                );
                            }
                        }
                    }
                }
            }
            continue;
        }

        if let Some(active) = socket.as_mut() {
            tokio::select! {
                command = commands.recv() => {
                    let Some(command) = command else { break; };
                    if matches!(command, Command::Stop) {
                        let _ = active.socket.close(None).await;
                        break;
                    }
                    if let Some((blocked, message)) = circuit_failure.as_ref() {
                        if matches!(
                            &command,
                            Command::Publish { desired, .. } if desired == blocked
                        ) {
                            let Command::Publish { reply, .. } = command else {
                                unreachable!("only unchanged publication enters this branch")
                            };
                            let _ = reply.send(Err(anyhow!(message.clone())));
                            continue;
                        }
                    }
                    let credential_rebind = match &command {
                        Command::Publish { desired, .. } => adapter
                            .shared
                            .applied_presence
                            .read()
                            .await
                            .as_ref()
                            .is_some_and(|applied| applied.ticket != desired.ticket),
                        _ => false,
                    };
                    if credential_rebind {
                        let Command::Publish { desired, reply } = command else {
                            unreachable!("credential rebind is only set for publication")
                        };
                        *adapter.shared.desired.write().await = Some(desired.clone());
                        *adapter.shared.applied_presence.write().await = None;
                        pending_publish = Some((desired, reply));
                        let _ = active.socket.close(None).await;
                        socket = None;
                        reconnect_attempt = 0;
                        retry_at = Some(tokio::time::Instant::now());
                        continue;
                    }
                    let going_offline = matches!(&command, Command::Offline { .. });
                    let publishing = matches!(&command, Command::Publish { .. });
                    let deleting_local_device =
                        command_deletes_device(&command, &adapter.device_id);
                    let result = handle_command(&adapter, &mut active.socket, command).await;
                    if going_offline || (deleting_local_device && result.is_ok()) {
                        let _ = active.socket.close(None).await;
                        *adapter.shared.desired.write().await = None;
                        *adapter.shared.applied_presence.write().await = None;
                        socket = None;
                        reconnect_attempt = 0;
                        retry_at = None;
                        circuit_failure = None;
                    } else if let Err(error) = result {
                        if is_retryable_gateway_error(&error) {
                            *adapter.shared.applied_presence.write().await = None;
                            socket = None;
                            reconnect_attempt = 0;
                            retry_at = Some(tokio::time::Instant::now());
                        } else {
                            eprintln!(
                                "[openrtc][coordination-gateway][operation-rejected] retryable=false error={}",
                                error
                            );
                            if publishing {
                                if let Some(desired) = adapter.shared.desired.read().await.clone() {
                                    circuit_failure = Some((desired, error.to_string()));
                                }
                            }
                        }
                    }
                }
                _ = tokio::time::sleep(auth_refresh_delay(active.credential_expires_at_ms)) => {
                    let refresh = if active.lease_refresh_mode == LeaseRefreshMode::Active {
                        refresh_lease(&adapter, &mut active.socket).await
                    } else {
                        refresh_authentication(
                            &adapter,
                            &mut active.socket,
                            Some(active.credential_token.clone()),
                        ).await
                    };
                    match refresh {
                        Ok((expires_at_ms, credential_token)) => {
                            active.credential_expires_at_ms = expires_at_ms;
                            if let Some(credential_token) = credential_token {
                                active.credential_token = credential_token;
                            }
                        }
                        Err(error) => {
                            let terminal_error = if active.lease_refresh_mode == LeaseRefreshMode::Active {
                                eprintln!(
                                    "[openrtc][coordination-gateway][lease-refresh-failed] retryable={} error={}",
                                    is_retryable_gateway_error(&error),
                                    error,
                                );
                                match refresh_authentication(
                                    &adapter,
                                    &mut active.socket,
                                    None,
                                ).await {
                                    Ok((expires_at_ms, Some(token))) => {
                                        active.credential_expires_at_ms = expires_at_ms;
                                        active.credential_token = token;
                                        continue;
                                    }
                                    Ok(_) => terminal_refresh_error(error, None),
                                    Err(fallback) => {
                                        eprintln!(
                                            "[openrtc][coordination-gateway][lease-fallback-failed] retryable={} error={}",
                                            is_retryable_gateway_error(&fallback),
                                            fallback,
                                        );
                                        terminal_refresh_error(error, Some(fallback))
                                    }
                                }
                            } else {
                                error
                            };
                            eprintln!(
                                "[openrtc][coordination-gateway][auth-refresh-failed] retryable={} error={}",
                                is_retryable_gateway_error(&terminal_error), terminal_error,
                            );
                            *adapter.shared.applied_presence.write().await = None;
                            socket = None;
                            reconnect_attempt = 0;
                            if is_retryable_gateway_error(&terminal_error) {
                                retry_at = Some(tokio::time::Instant::now());
                            } else {
                                retry_at = None;
                                if let Some(desired) = adapter.shared.desired.read().await.clone() {
                                    circuit_failure = Some((desired, terminal_error.to_string()));
                                }
                            }
                        }
                    }
                }
                _ = tokio::time::sleep(SOCKET_KEEPALIVE_INTERVAL) => {
                    if let Err(error) = send_socket_keepalive(&mut active.socket).await {
                        eprintln!(
                            "[openrtc][coordination-gateway][keepalive-failed] retryable=true error={}",
                            error,
                        );
                        *adapter.shared.applied_presence.write().await = None;
                        socket = None;
                        reconnect_attempt = 0;
                        retry_at = Some(tokio::time::Instant::now());
                    }
                }
                incoming = active.socket.next() => {
                    match incoming {
                        Some(Ok(message)) => {
                            if let Err(error) = handle_message(&adapter, message).await {
                                *adapter.shared.applied_presence.write().await = None;
                                socket = None;
                                reconnect_attempt = 0;
                                if is_retryable_gateway_error(&error) {
                                    retry_at = Some(tokio::time::Instant::now());
                                } else {
                                    retry_at = None;
                                    if let Some(desired) = adapter.shared.desired.read().await.clone() {
                                        circuit_failure = Some((desired, error.to_string()));
                                    }
                                }
                            }
                        }
                        _ => {
                            *adapter.shared.applied_presence.write().await = None;
                            socket = None;
                            reconnect_attempt = 0;
                            retry_at = Some(tokio::time::Instant::now());
                        },
                    }
                }
            }
        }
    }
}

fn command_deletes_device(command: &Command, local_device_id: &str) -> bool {
    matches!(
        command,
        Command::Delete { device_id, .. } if device_id == local_device_id
    )
}

async fn connect(adapter: &NativeCoordinationGatewaySignaling) -> Result<ConnectedGateway> {
    let desired = adapter
        .shared
        .desired
        .read()
        .await
        .clone()
        .ok_or_else(|| anyhow!("native coordination presence is not configured"))?;
    connect_with_desired(adapter, &desired, "presence").await
}

async fn connect_with_desired(
    adapter: &NativeCoordinationGatewaySignaling,
    desired: &DesiredPresence,
    purpose: &str,
) -> Result<ConnectedGateway> {
    let credential = adapter.mint_credential(desired, purpose, None).await?;
    if credential.expires_at_ms <= now_ms().saturating_add(30_000) {
        bail!("native coordination credential expires too soon");
    }
    let gateway_url = adapter.gateway_url(&credential)?;
    let uri: Uri = gateway_url.parse().context("parse native gateway URI")?;
    let mut request = uri.into_client_request()?;
    request.headers_mut().insert(
        "Sec-WebSocket-Protocol",
        HeaderValue::from_str(&format!(
            "{}, {}{}",
            GATEWAY_PROTOCOL, GATEWAY_AUTH_PROTOCOL_PREFIX, credential.token
        ))?,
    );
    let (mut socket, _) =
        tokio::time::timeout(REQUEST_TIMEOUT, tokio_tungstenite::connect_async(request))
            .await
            .context("native coordination WebSocket connect timed out")??;
    socket
        .send(Message::Text(
            serde_json::json!({
                "v": WIRE_PROTOCOL_VERSION,
                "type": "auth",
                "token": credential.token,
                "device": gateway_device_value(adapter, &desired).await
            })
            .to_string()
            .into(),
        ))
        .await?;
    let lease_refresh_mode = tokio::time::timeout(ACK_TIMEOUT, async {
        while let Some(message) = socket.next().await {
            let message = message?;
            if let Some(frame) = decode_frame(message)? {
                match frame {
                    ServerFrame::Ready {
                        lease_refresh_mode, ..
                    } => {
                        return Ok::<LeaseRefreshMode, anyhow::Error>(LeaseRefreshMode::from_wire(
                            lease_refresh_mode.as_deref(),
                        ))
                    }
                    ServerFrame::Error {
                        code,
                        message,
                        retryable,
                        ..
                    } => {
                        return Err(gateway_connect_error(
                            format!("coordination gateway {code}: {message}"),
                            retryable,
                        ));
                    }
                    other => handle_frame(adapter, other).await?,
                }
            }
        }
        bail!("coordination gateway closed before ready")
    })
    .await
    .context("native coordination authentication timed out")??;
    Ok(ConnectedGateway {
        socket,
        credential_expires_at_ms: credential.expires_at_ms,
        credential_token: credential.token,
        lease_refresh_mode,
    })
}

async fn execute_control_delete(
    adapter: &NativeCoordinationGatewaySignaling,
    user_id: &str,
    target_device_id: &str,
) -> Result<()> {
    let desired = DesiredPresence {
        user_id: user_id.to_string(),
        local_node_id: adapter.device_id.clone(),
        ticket: format!("openrtc-device-control:{}", adapter.runtime_instance_id),
        device_name: adapter.device_id.clone(),
        metadata: None,
        ttl_ms: 0,
        online: false,
    };
    let mut connected = connect_with_desired(adapter, &desired, "device-control").await?;
    let result = execute_operation(
        adapter,
        &mut connected.socket,
        serde_json::json!({
            "v": WIRE_PROTOCOL_VERSION,
            "type": "device.delete",
            "idempotencyKey": random_id("device-delete"),
            "targetDeviceId": target_device_id,
        }),
    )
    .await;
    let _ = connected.socket.close(None).await;
    result
}

async fn refresh_authentication(
    adapter: &NativeCoordinationGatewaySignaling,
    socket: &mut GatewaySocket,
    current_grant: Option<String>,
) -> Result<(u64, Option<String>)> {
    let desired = adapter
        .shared
        .desired
        .read()
        .await
        .clone()
        .ok_or_else(|| anyhow!("native coordination presence is not configured"))?;
    let credential = adapter
        .mint_credential(&desired, "presence", current_grant)
        .await?;
    let refreshed_token = credential.token.clone();
    socket
        .send(Message::Text(
            serde_json::json!({
                "v": WIRE_PROTOCOL_VERSION,
                "type": "auth.refresh",
                "token": credential.token,
            })
            .to_string()
            .into(),
        ))
        .await?;
    tokio::time::timeout(ACK_TIMEOUT, async {
        while let Some(message) = socket.next().await {
            let message = message?;
            if let Some(frame) = decode_frame(message)? {
                match frame {
                    ServerFrame::AuthRefreshed { _expires_at_ms } => {
                        update_local_device_expiry(adapter, _expires_at_ms).await;
                        return Ok((_expires_at_ms, Some(refreshed_token)));
                    }
                    ServerFrame::Error {
                        code,
                        message,
                        retryable,
                        ..
                    } => {
                        return Err(gateway_connect_error(
                            format!("coordination gateway {code}: {message}"),
                            retryable,
                        ));
                    }
                    other => handle_frame(adapter, other).await?,
                }
            }
        }
        bail!("coordination gateway closed before authentication refresh")
    })
    .await
    .context("native coordination authentication refresh timed out")?
}

async fn refresh_lease(
    adapter: &NativeCoordinationGatewaySignaling,
    socket: &mut GatewaySocket,
) -> Result<(u64, Option<String>)> {
    let idempotency_key = random_id("lease");
    socket
        .send(Message::Text(
            serde_json::json!({
                "v": WIRE_PROTOCOL_VERSION,
                "type": "lease.refresh",
                "idempotencyKey": idempotency_key,
            })
            .to_string()
            .into(),
        ))
        .await?;
    tokio::time::timeout(ACK_TIMEOUT, async {
        while let Some(message) = socket.next().await {
            let message = message?;
            if let Some(frame) = decode_frame(message)? {
                match frame {
                    ServerFrame::LeaseRefreshed {
                        idempotency_key: response_key,
                        expires_at_ms,
                    } if response_key == idempotency_key => {
                        update_local_device_expiry(adapter, expires_at_ms).await;
                        return Ok((expires_at_ms, None));
                    }
                    ServerFrame::Error {
                        code,
                        message,
                        retryable,
                        ..
                    } => {
                        return Err(gateway_connect_error(
                            format!("coordination gateway {code}: {message}"),
                            retryable,
                        ));
                    }
                    other => handle_frame(adapter, other).await?,
                }
            }
        }
        bail!("coordination gateway closed before lease refresh acknowledgement")
    })
    .await
    .context("native coordination lease refresh timed out")?
}

async fn update_local_device_expiry(
    adapter: &NativeCoordinationGatewaySignaling,
    expires_at_ms: u64,
) {
    let event = {
        let mut devices = adapter.shared.devices.write().await;
        let Some(device) = devices.get_mut(&adapter.device_id) else {
            return;
        };
        let next_expiry = serde_json::json!(expires_at_ms);
        if device.expires_at.as_ref() == Some(&next_expiry) {
            return;
        }
        device.expires_at = Some(next_expiry);
        DeviceEvent::Modified {
            device: device.clone(),
        }
    };
    let _ = adapter.shared.device_events.send(vec![event]);
}

async fn handle_command(
    adapter: &NativeCoordinationGatewaySignaling,
    socket: &mut GatewaySocket,
    command: Command,
) -> Result<()> {
    match command {
        Command::Publish { desired, reply } => {
            *adapter.shared.desired.write().await = Some(desired.clone());
            if adapter.shared.applied_presence.read().await.as_ref() == Some(&desired) {
                let _ = reply.send(Ok(()));
                return Ok(());
            }
            let result =
                execute_operation(adapter, socket, presence_frame(adapter, &desired).await).await;
            if result.is_ok() {
                *adapter.shared.applied_presence.write().await = Some(desired);
            }
            complete_command(reply, result, "native presence publication")?;
        }
        Command::Patch { patch, reply } => {
            merge_patch(
                &mut *adapter.shared.staged_patch.write().await,
                patch.clone(),
            );
            let result = execute_operation(
                adapter,
                socket,
                serde_json::json!({
                    "v": WIRE_PROTOCOL_VERSION,
                    "type": "device.patch",
                    "idempotencyKey": random_id("device-patch"),
                    "patch": patch,
                }),
            )
            .await;
            complete_command(reply, result, "native device patch")?;
        }
        Command::Offline { reply } => {
            if let Some(desired) = adapter.shared.desired.write().await.as_mut() {
                desired.online = false;
            }
            let result = execute_operation(
                adapter,
                socket,
                serde_json::json!({
                    "v": WIRE_PROTOCOL_VERSION,
                    "type": "presence.offline",
                    "idempotencyKey": random_id("offline"),
                }),
            )
            .await;
            complete_command(reply, result, "native presence offline")?;
        }
        Command::Delete {
            device_id, reply, ..
        } => {
            let result = execute_operation(
                adapter,
                socket,
                serde_json::json!({
                    "v": WIRE_PROTOCOL_VERSION,
                    "type": "device.delete",
                    "idempotencyKey": random_id("device-delete"),
                    "targetDeviceId": device_id,
                }),
            )
            .await;
            complete_command(reply, result, "native device delete")?;
        }
        Command::SendSignal {
            target_device_id,
            payload,
            state,
            reply_payload,
            reply,
        } => {
            let idempotency_key = random_id("signal");
            let result = execute_operation(
                adapter,
                socket,
                serde_json::json!({
                    "v": WIRE_PROTOCOL_VERSION,
                    "type": "signal.send",
                    "idempotencyKey": idempotency_key,
                    "targetDeviceId": target_device_id,
                    "payload": payload,
                    "state": state,
                    "replyPayload": reply_payload,
                }),
            )
            .await
            .map(|_| idempotency_key);
            complete_command(reply, result, "native signal send")?;
        }
        Command::PutSession {
            session_id,
            session,
            expires_at_ms,
            reply,
        } => {
            let result = execute_operation(
                adapter,
                socket,
                serde_json::json!({
                    "v": WIRE_PROTOCOL_VERSION,
                    "type": "session.put",
                    "idempotencyKey": random_id("session-put"),
                    "sessionId": session_id,
                    "session": session,
                    "expiresAtMs": expires_at_ms,
                }),
            )
            .await;
            complete_command(reply, result, "native session update")?;
        }
        Command::Stop => {}
    }
    Ok(())
}

fn complete_command<T>(
    reply: oneshot::Sender<Result<T>>,
    result: Result<T>,
    operation: &str,
) -> Result<()> {
    match result {
        Ok(value) => {
            let _ = reply.send(Ok(value));
            Ok(())
        }
        Err(error) => {
            let retryable = is_retryable_gateway_error(&error);
            let message = error.to_string();
            let _ = reply.send(Err(anyhow!(message.clone())));
            Err(gateway_connect_error(
                format!("{operation} failed: {message}"),
                retryable,
            ))
        }
    }
}

async fn execute_operation(
    adapter: &NativeCoordinationGatewaySignaling,
    socket: &mut GatewaySocket,
    frame: serde_json::Value,
) -> Result<()> {
    let operation = frame
        .get("type")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("unknown")
        .to_string();
    let idempotency_key = frame
        .get("idempotencyKey")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| anyhow!("gateway operation is missing idempotencyKey"))?
        .to_string();
    socket
        .send(Message::Text(frame.to_string().into()))
        .await
        .context("send native gateway operation")?;
    tokio::time::timeout(ACK_TIMEOUT, async {
        while let Some(message) = socket.next().await {
            let message = message?;
            if let Some(frame) = decode_frame(message)? {
                match frame {
                    ServerFrame::Ack {
                        idempotency_key: ack,
                    } if ack == idempotency_key => return Ok(()),
                    ServerFrame::Error {
                        code,
                        message,
                        idempotency_key: Some(key),
                        retryable,
                    } if key == idempotency_key => {
                        return Err(gateway_connect_error(
                            format!("coordination gateway {code}: {message} operation={operation}"),
                            retryable,
                        ));
                    }
                    other => handle_frame(adapter, other).await?,
                }
            }
        }
        bail!("coordination gateway closed before acknowledgement")
    })
    .await
    .context("native coordination acknowledgement timed out")?
}

async fn handle_message(
    adapter: &NativeCoordinationGatewaySignaling,
    message: Message,
) -> Result<()> {
    if let Some(frame) = decode_frame(message)? {
        handle_frame(adapter, frame).await?;
    }
    Ok(())
}

fn decode_frame(message: Message) -> Result<Option<ServerFrame>> {
    match message {
        Message::Text(text) => Ok(Some(serde_json::from_str(&text)?)),
        Message::Binary(bytes) => Ok(Some(serde_json::from_slice(&bytes)?)),
        Message::Ping(_) | Message::Pong(_) => Ok(None),
        Message::Close(_) => bail!("coordination gateway closed"),
        _ => Ok(None),
    }
}

async fn handle_frame(
    adapter: &NativeCoordinationGatewaySignaling,
    frame: ServerFrame,
) -> Result<()> {
    match frame {
        ServerFrame::RosterSnapshot { devices } => {
            let mut mapped = HashMap::new();
            let mut events = Vec::with_capacity(devices.len());
            for raw in devices {
                let device = adapter.device_from_gateway(raw);
                mapped.insert(device.device_id.clone(), device.clone());
                events.push(DeviceEvent::Added { device });
            }
            *adapter.shared.devices.write().await = mapped;
            if !events.is_empty() {
                let _ = adapter.shared.device_events.send(events);
            }
        }
        ServerFrame::PresenceChanged { operation, device } => {
            let device = adapter.device_from_gateway(device);
            let event = if operation == "delete" {
                adapter
                    .shared
                    .devices
                    .write()
                    .await
                    .remove(&device.device_id);
                DeviceEvent::Removed {
                    device_id: device.device_id,
                }
            } else {
                let existed = adapter
                    .shared
                    .devices
                    .write()
                    .await
                    .insert(device.device_id.clone(), device.clone())
                    .is_some();
                if existed {
                    DeviceEvent::Modified { device }
                } else {
                    DeviceEvent::Added { device }
                }
            };
            let _ = adapter.shared.device_events.send(vec![event]);
        }
        ServerFrame::SessionChanged {
            operation,
            session_id,
            session,
        } => {
            let event = if operation == "delete" {
                SessionEvent::Removed { session_id }
            } else {
                let mut value = session.unwrap_or_else(|| serde_json::json!({}));
                value["connectionId"] = serde_json::Value::String(session_id);
                SessionEvent::Modified {
                    session: serde_json::from_value(value)?,
                }
            };
            let _ = adapter.shared.session_events.send(vec![event]);
        }
        ServerFrame::SignalReceived {
            sender_device_id,
            payload,
            state,
            reply_payload,
            ..
        } => {
            let mut pending = adapter
                .shared
                .pending_messages
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if pending.len() >= 1_000 {
                pending.remove(0);
            }
            pending.push(SignalingEnvelope {
                app_tag: Some(adapter.app_tag.clone()),
                sender_id: sender_device_id,
                target_id: adapter.device_id.clone(),
                payload,
                state,
                reply_payload,
                timestamp: now_ms() as i64,
                sender_user_id: None,
                target_user_id: None,
                expires_at: None,
            });
        }
        ServerFrame::Error {
            code,
            message,
            retryable,
            ..
        } => {
            return Err(gateway_connect_error(
                format!("coordination gateway {code}: {message}"),
                retryable,
            ))
        }
        ServerFrame::Ready { .. }
        | ServerFrame::AuthRefreshed { .. }
        | ServerFrame::LeaseRefreshed { .. }
        | ServerFrame::Ack { .. }
        | ServerFrame::Pong => {}
    }
    Ok(())
}

async fn presence_frame(
    adapter: &NativeCoordinationGatewaySignaling,
    desired: &DesiredPresence,
) -> serde_json::Value {
    serde_json::json!({
        "v": WIRE_PROTOCOL_VERSION,
        "type": "presence.upsert",
        "idempotencyKey": random_id("presence"),
        "ttlMs": desired.ttl_ms,
        "device": gateway_device_value(adapter, desired).await,
    })
}

async fn gateway_device_value(
    adapter: &NativeCoordinationGatewaySignaling,
    desired: &DesiredPresence,
) -> serde_json::Value {
    let patch = adapter.shared.staged_patch.read().await.clone();
    serde_json::to_value(OutboundGatewayDevice {
        device_id: adapter.device_id.clone(),
        runtime_instance_id: adapter.runtime_instance_id.clone(),
        node_id: desired.local_node_id.clone(),
        device_name: patch
            .device_name
            .unwrap_or_else(|| desired.device_name.clone()),
        platform_type: adapter.platform_type.clone(),
        ticket: desired.ticket.clone(),
        metadata: patch.metadata.or_else(|| desired.metadata.clone()),
        capabilities: patch.capabilities,
        excluded_peers: patch.excluded_peers.unwrap_or_default(),
        online: desired.online,
    })
    .expect("gateway device projection is serializable")
}

fn merge_patch(target: &mut DevicePatch, patch: DevicePatch) {
    if patch.device_name.is_some() {
        target.device_name = patch.device_name;
    }
    if patch.capabilities.is_some() {
        target.capabilities = patch.capabilities;
    }
    if patch.metadata.is_some() {
        target.metadata = patch.metadata;
    }
    if patch.excluded_peers.is_some() {
        target.excluded_peers = patch.excluded_peers;
    }
}

fn reject_command(command: Command, message: &str) {
    match command {
        Command::Publish { reply, .. }
        | Command::Patch { reply, .. }
        | Command::Offline { reply }
        | Command::Delete { reply, .. }
        | Command::PutSession { reply, .. } => {
            let _ = reply.send(Err(anyhow!(message.to_string())));
        }
        Command::SendSignal { reply, .. } => {
            let _ = reply.send(Err(anyhow!(message.to_string())));
        }
        Command::Stop => {}
    }
}

fn reconnect_delay(attempt: u8) -> Duration {
    let exponent = u32::from(attempt.saturating_sub(1).min(7));
    Duration::from_millis(
        (250_u64.saturating_mul(1_u64 << exponent)).min(MAX_RECONNECT_DELAY.as_millis() as u64),
    )
}

fn auth_refresh_delay(expires_at_ms: u64) -> Duration {
    Duration::from_millis(
        expires_at_ms
            .saturating_sub(now_ms())
            .saturating_sub(AUTH_REFRESH_SKEW_MS)
            .max(1_000),
    )
}

async fn send_socket_keepalive<S, E>(socket: &mut S) -> Result<()>
where
    S: Sink<Message, Error = E> + Unpin,
    E: StdError + Send + Sync + 'static,
{
    socket
        .send(Message::Ping(Vec::new().into()))
        .await
        .context("send native coordination WebSocket keepalive")
}

fn required(name: &str, value: String) -> Result<String> {
    let value = value.trim().to_string();
    if value.is_empty() {
        bail!("native coordination {name} is required");
    }
    Ok(value)
}

fn validate_endpoint(name: &str, value: String, allow_websocket: bool) -> Result<String> {
    let value = required(name, value)?;
    let url = reqwest::Url::parse(&value)?;
    let allowed = matches!(url.scheme(), "https" | "http")
        || (allow_websocket && matches!(url.scheme(), "wss" | "ws"));
    let secure = matches!(url.scheme(), "https" | "wss");
    let loopback = matches!(
        url.host_str(),
        Some("127.0.0.1") | Some("localhost") | Some("::1")
    );
    if !allowed
        || (!secure && !loopback)
        || !url.username().is_empty()
        || url.password().is_some()
        || url.query().is_some()
        || url.fragment().is_some()
    {
        bail!("native coordination {name} is invalid");
    }
    Ok(value.trim_end_matches('/').to_string())
}

fn normalized_origin_scheme(scheme: &str) -> &str {
    match scheme {
        "ws" => "http",
        "wss" => "https",
        other => other,
    }
}

fn random_id(prefix: &str) -> String {
    format!("{prefix}:{}", Uuid::new_v4())
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

#[async_trait]
impl SignalingBackend for NativeCoordinationGatewaySignaling {
    async fn update_presence(
        &self,
        user_id: &str,
        local_node_id: &str,
        ticket_str: &str,
        is_online: bool,
        name: &str,
        ttl_ms: u64,
        metadata: Option<&str>,
    ) -> Result<()> {
        // The native lifecycle exposes its historical durable-device and
        // live-lease projections as two concurrent backend calls. The gateway
        // owns both projections in one presence upsert, so retain the durable
        // fields locally and let the live call perform the single billed
        // operation. Explicit offline transitions use `set_offline`.
        if !is_online {
            merge_patch(
                &mut *self.shared.staged_patch.write().await,
                DevicePatch {
                    device_name: Some(name.to_string()),
                    metadata: metadata.map(str::to_string),
                    ..DevicePatch::default()
                },
            );
            return Ok(());
        }
        let local_node_id = iroh::EndpointId::from_str(local_node_id)
            .context("native coordination node ID is invalid")?
            .to_string();
        let desired = DesiredPresence {
            user_id: user_id.to_string(),
            local_node_id,
            ticket: ticket_str.to_string(),
            device_name: name.to_string(),
            metadata: metadata.map(str::to_string),
            ttl_ms,
            online: is_online,
        };
        *self.shared.desired.write().await = Some(desired.clone());
        self.request_unit(|reply| Command::Publish { desired, reply })
            .await
    }

    async fn set_offline(&self, _user_id: &str, _local_node_id: &str) -> Result<()> {
        self.request_unit(|reply| Command::Offline { reply }).await
    }

    async fn update_live_presence(
        &self,
        user_id: &str,
        local_node_id: &str,
        ticket_str: &str,
        name: &str,
        metadata: Option<&str>,
    ) -> Result<()> {
        self.update_presence(
            user_id,
            local_node_id,
            ticket_str,
            true,
            name,
            15 * 60_000,
            metadata,
        )
        .await
    }

    async fn set_live_presence_offline(&self, user_id: &str, local_node_id: &str) -> Result<()> {
        self.set_offline(user_id, local_node_id).await
    }

    async fn update_device(
        &self,
        _user_id: &str,
        device_id: &str,
        device_name: Option<&str>,
        capabilities: Option<DeviceCapabilities>,
        metadata: Option<&str>,
    ) -> Result<()> {
        if device_id != self.device_id {
            bail!("native coordination socket can update only its local device");
        }
        let patch = DevicePatch {
            device_name: device_name.map(str::to_string),
            capabilities,
            metadata: metadata.map(str::to_string),
            excluded_peers: None,
        };
        merge_patch(&mut *self.shared.staged_patch.write().await, patch.clone());
        // Initial durable projection and live lease are folded into the first
        // presence upsert. Avoid a second billed device.patch while the
        // concurrent native presence actor is still establishing the socket.
        if self.shared.applied_presence.read().await.is_none() {
            return Ok(());
        }
        self.request_unit(|reply| Command::Patch { patch, reply })
            .await
    }

    async fn delete_device(&self, _user_id: &str, device_id: &str) -> Result<()> {
        self.request_unit(|reply| Command::Delete {
            user_id: _user_id.to_string(),
            device_id: device_id.to_string(),
            reply,
        })
        .await
    }

    async fn set_excluded_peers(
        &self,
        _user_id: &str,
        _local_node_id: &str,
        excluded_peers: &[String],
    ) -> Result<()> {
        let mut normalized = excluded_peers
            .iter()
            .map(|value| value.trim().to_ascii_lowercase())
            .filter(|value| !value.is_empty())
            .collect::<Vec<_>>();
        normalized.sort();
        normalized.dedup();
        if normalized.len() > MAX_EXCLUDED_PEERS {
            bail!(
                "native coordination excluded-peer set exceeds {} devices",
                MAX_EXCLUDED_PEERS
            );
        }
        let patch = DevicePatch {
            excluded_peers: Some(normalized),
            ..DevicePatch::default()
        };
        merge_patch(&mut *self.shared.staged_patch.write().await, patch.clone());
        if self.shared.applied_presence.read().await.is_none() {
            return Ok(());
        }
        self.request_unit(|reply| Command::Patch { patch, reply })
            .await
    }

    async fn search_devices(
        &self,
        _user_id: &str,
        exclude_node_id: Option<&str>,
    ) -> Result<Vec<Device>> {
        self.list_devices("", exclude_node_id).await
    }

    async fn list_devices(
        &self,
        _user_id: &str,
        exclude_node_id: Option<&str>,
    ) -> Result<Vec<Device>> {
        let mut devices = self
            .shared
            .devices
            .read()
            .await
            .values()
            .filter(|device| exclude_node_id != device.node_id.as_deref())
            .cloned()
            .collect::<Vec<_>>();
        devices.sort_by(|left, right| left.device_id.cmp(&right.device_id));
        Ok(devices)
    }

    async fn send_message(
        &self,
        _sender_id: &str,
        target_id: &str,
        payload: &str,
        state: Option<&str>,
        reply_payload: Option<&str>,
    ) -> Result<String> {
        let (reply, response) = oneshot::channel();
        self.commands
            .send(Command::SendSignal {
                target_device_id: target_id.to_string(),
                payload: payload.to_string(),
                state: state.map(str::to_string),
                reply_payload: reply_payload.map(str::to_string),
                reply,
            })
            .await
            .map_err(|_| anyhow!("coordination gateway actor stopped"))?;
        response
            .await
            .map_err(|_| anyhow!("coordination gateway actor dropped its reply"))?
    }

    async fn subscribe_devices(
        &self,
        _user_id: &str,
    ) -> Result<BoxStream<'static, Result<Vec<DeviceEvent>>>> {
        let receiver = self.shared.device_events.subscribe();
        Ok(Box::pin(
            tokio_stream::wrappers::BroadcastStream::new(receiver)
                .filter_map(|event| async move { event.ok().map(Ok) }),
        ))
    }

    async fn create_session(&self, session: SignalingSession) -> Result<()> {
        let expires_at_ms = session
            .expires_at
            .unwrap_or_else(|| now_ms() as i64 + 15 * 60_000);
        let session_id = session.connection_id.clone();
        let value = serde_json::to_value(session)?;
        self.request_unit(|reply| Command::PutSession {
            session_id,
            session: value,
            expires_at_ms,
            reply,
        })
        .await
    }

    async fn update_session(&self, session_id: &str, update_data: serde_json::Value) -> Result<()> {
        let expires_at_ms = update_data
            .get("expiresAt")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or_else(|| now_ms() as i64 + 15 * 60_000);
        self.request_unit(|reply| Command::PutSession {
            session_id: session_id.to_string(),
            session: update_data,
            expires_at_ms,
            reply,
        })
        .await
    }

    async fn subscribe_sessions(
        &self,
        _local_device_id: &str,
    ) -> Result<BoxStream<'static, Result<Vec<SessionEvent>>>> {
        let receiver = self.shared.session_events.subscribe();
        Ok(Box::pin(
            tokio_stream::wrappers::BroadcastStream::new(receiver)
                .filter_map(|event| async move { event.ok().map(Ok) }),
        ))
    }
}

impl Drop for NativeCoordinationGatewaySignaling {
    fn drop(&mut self) {
        let _ = self.commands.try_send(Command::Stop);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[derive(Default)]
    struct RecordingGrantProvider {
        requests: Mutex<Vec<NativeGatewayGrantRequest>>,
    }

    #[async_trait]
    impl NativeGatewayGrantProvider for RecordingGrantProvider {
        async fn grant(&self, request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant> {
            self.requests.lock().expect("requests").push(request);
            Ok(NativeGatewayGrant {
                protocol_version: 2,
                gateway_url: "https://gateway.example.test".to_string(),
                route_key: "route-1".to_string(),
                token: "grant-2".to_string(),
                expires_at_ms: now_ms() + 3_600_000,
            })
        }
    }

    #[tokio::test]
    async fn v2_native_capability_handle_is_inactive_until_runtime_presence_and_closes_once() {
        let provider = Arc::new(RecordingGrantProvider::default());
        let capabilities = NativeCapabilities::new(
            crate::test_constants::TEST_API_KEY,
            "device-native",
            "native",
            provider.clone(),
        )
        .expect("capability namespace")
        .with_testing_endpoint("https://gateway.example.test")
        .expect("testing endpoint");
        let handle = capabilities.join_room("room-1").expect("capability handle");

        assert_eq!(handle.kind(), NativeCapabilityKind::Room);
        assert_eq!(handle.id(), "room-1");
        assert!(!handle.is_closed());
        assert!(provider.requests.lock().expect("requests").is_empty());

        handle.close().await;
        handle.close().await;
        assert!(handle.is_closed());
        assert!(provider.requests.lock().expect("requests").is_empty());
    }

    #[test]
    fn reconnect_is_bounded_and_capped() {
        assert_eq!(reconnect_delay(1), Duration::from_millis(250));
        assert_eq!(reconnect_delay(12), MAX_RECONNECT_DELAY);
        assert_eq!(MAX_RECONNECT_ATTEMPTS, 12);
    }

    #[test]
    fn endpoints_reject_query_credentials() {
        assert!(validate_endpoint(
            "endpoint",
            "https://gateway.example.test?token=secret".to_string(),
            true,
        )
        .is_err());
        assert!(validate_endpoint(
            "endpoint",
            "https://user:secret@gateway.example.test".to_string(),
            true,
        )
        .is_err());
        assert!(
            validate_endpoint("endpoint", "http://gateway.example.test".to_string(), true,)
                .is_err()
        );
        assert!(validate_endpoint("endpoint", "http://127.0.0.1:8787".to_string(), true,).is_ok());
        assert_eq!(normalized_origin_scheme("wss"), "https");
        assert_eq!(normalized_origin_scheme("ws"), "http");
    }

    #[tokio::test]
    async fn v2_grant_provider_owns_native_issue_and_in_place_refresh() {
        let provider = Arc::new(RecordingGrantProvider::default());
        let adapter = NativeCoordinationGatewaySignaling::new(NativeCoordinationGatewayOptions {
            endpoint: "https://gateway.example.test".to_string(),
            app_tag: "app_native_test".to_string(),
            device_id: "device-1".to_string(),
            platform_type: "desktop".to_string(),
            avenue: NativeCoordinationAvenue {
                kind: "user".to_string(),
                id: "principal-1".to_string(),
            },
            grant_provider: provider.clone(),
        })
        .expect("adapter");
        let desired = DesiredPresence {
            user_id: "principal-1".to_string(),
            local_node_id: "node-1".to_string(),
            ticket: "ticket-1".to_string(),
            device_name: "Native".to_string(),
            metadata: None,
            ttl_ms: 900_000,
            online: true,
        };
        let issued = adapter
            .mint_credential(&desired, "presence", None)
            .await
            .expect("issue");
        let refreshed = adapter
            .mint_credential(&desired, "presence", Some(issued.token.clone()))
            .await
            .expect("refresh");
        assert_eq!(issued.protocol_version, 2);
        assert_eq!(refreshed.protocol_version, 2);
        assert!(adapter
            .gateway_url(&issued)
            .expect("url")
            .contains("/v2/connect/route-1"));
        let requests = provider.requests.lock().expect("requests");
        assert_eq!(requests.len(), 2);
        assert_eq!(requests[0].refresh_grant, None);
        assert_eq!(requests[1].refresh_grant.as_deref(), Some("grant-2"));
        assert_eq!(requests[1].avenue.kind, "user");
        assert_eq!(requests[1].avenue.id, "principal-1");
    }

    #[test]
    fn patch_merge_preserves_unrelated_fields() {
        let mut target = DevicePatch {
            device_name: Some("old".into()),
            capabilities: Some(DeviceCapabilities {
                can_host: true,
                can_sync: true,
                read_only: false,
            }),
            metadata: None,
            excluded_peers: None,
        };
        merge_patch(
            &mut target,
            DevicePatch {
                excluded_peers: Some(vec!["peer".into()]),
                ..DevicePatch::default()
            },
        );
        assert_eq!(target.device_name.as_deref(), Some("old"));
        assert_eq!(target.excluded_peers, Some(vec!["peer".into()]));
    }

    #[test]
    fn permanent_gateway_errors_are_not_retried() {
        let permanent = gateway_connect_error("invalid payload", false);
        let transient = gateway_connect_error("provider unavailable", true);
        assert!(!is_retryable_gateway_error(&permanent));
        assert!(is_retryable_gateway_error(&transient));
    }

    #[test]
    fn terminal_control_plane_rejections_are_not_retried() {
        let revoked = anyhow::Error::new(NativeControlPlaneHttpError {
            status: 403,
            message: "device revoked".into(),
        })
        .context("obtain native coordination grant");
        let throttled = anyhow::Error::new(NativeControlPlaneHttpError {
            status: 429,
            message: "try later".into(),
        })
        .context("obtain native coordination grant");
        let unavailable = anyhow::Error::new(NativeControlPlaneHttpError {
            status: 503,
            message: "provider unavailable".into(),
        })
        .context("obtain native coordination grant");

        assert!(!is_retryable_gateway_error(&revoked));
        assert!(is_retryable_gateway_error(&throttled));
        assert!(is_retryable_gateway_error(&unavailable));
    }

    #[test]
    fn lease_fallback_error_owns_terminal_reconnect_classification() {
        let permanent_lease_error = gateway_connect_error("missing operation price", false);
        let transient_fallback_error = gateway_connect_error("control plane timeout", true);
        let terminal =
            terminal_refresh_error(permanent_lease_error, Some(transient_fallback_error));

        assert!(is_retryable_gateway_error(&terminal));
        assert_eq!(terminal.to_string(), "control plane timeout");
    }

    #[tokio::test]
    async fn native_keepalive_is_a_websocket_control_frame() {
        let (mut sender, mut receiver) = futures::channel::mpsc::channel(1);
        send_socket_keepalive(&mut sender).await.unwrap();
        assert!(
            matches!(receiver.next().await, Some(Message::Ping(payload)) if payload.is_empty())
        );
    }

    #[test]
    fn only_local_device_delete_terminates_the_gateway_presence_owner() {
        let (local_reply, _local_result) = oneshot::channel();
        let local = Command::Delete {
            user_id: "user-1".into(),
            device_id: "local-device".into(),
            reply: local_reply,
        };
        let (remote_reply, _remote_result) = oneshot::channel();
        let remote = Command::Delete {
            user_id: "user-1".into(),
            device_id: "remote-device".into(),
            reply: remote_reply,
        };
        assert!(command_deletes_device(&local, "local-device"));
        assert!(!command_deletes_device(&remote, "local-device"));
    }

    #[tokio::test]
    async fn healthy_lease_response_advances_the_local_device_projection() {
        let provider = Arc::new(RecordingGrantProvider::default());
        let adapter = NativeCoordinationGatewaySignaling::new(NativeCoordinationGatewayOptions {
            endpoint: "https://gateway.example.test".to_string(),
            app_tag: "app_native_test".to_string(),
            device_id: "device-1".to_string(),
            platform_type: "desktop".to_string(),
            avenue: NativeCoordinationAvenue {
                kind: "user".to_string(),
                id: "principal-1".to_string(),
            },
            grant_provider: provider,
        })
        .expect("adapter");
        adapter.shared.devices.write().await.insert(
            "device-1".to_string(),
            Device {
                app_tag: Some("app_native_test".to_string()),
                device_id: "device-1".to_string(),
                user_id: Some("principal-1".to_string()),
                device_name: "Native".to_string(),
                platform_type: Some("desktop".to_string()),
                capabilities: None,
                session_id: Some("runtime-1".to_string()),
                node_id: Some("00".repeat(32)),
                tag: None,
                kind: None,
                metadata: None,
                online: true,
                ticket: Some("ticket-1".to_string()),
                last_seen_at: None,
                expires_at: Some(serde_json::json!(1_000_u64)),
                created_at: None,
                updated_at: None,
                excluded_peers: Vec::new(),
            },
        );

        update_local_device_expiry(&adapter, 2_000).await;

        let devices = adapter.shared.devices.read().await;
        assert_eq!(
            devices
                .get("device-1")
                .and_then(|device| device.expires_at.clone()),
            Some(serde_json::json!(2_000_u64)),
        );
    }

    #[tokio::test]
    async fn command_completion_preserves_permanent_error_classification() {
        let (reply, result) = oneshot::channel();
        let actor_error = complete_command::<()>(
            reply,
            Err(gateway_connect_error("invalid payload", false)),
            "presence publication",
        )
        .expect_err("permanent operation must fail");

        assert!(!is_retryable_gateway_error(&actor_error));
        assert_eq!(
            result
                .await
                .expect("caller receives reply")
                .expect_err("caller receives operation failure")
                .to_string(),
            "invalid payload"
        );
    }

    #[test]
    fn optional_device_fields_are_omitted_instead_of_null() {
        let value = serde_json::to_value(OutboundGatewayDevice {
            device_id: "device:test".into(),
            runtime_instance_id: "runtime:test".into(),
            node_id: "00".repeat(32),
            device_name: "Test".into(),
            platform_type: "macos".into(),
            ticket: "ticket".into(),
            metadata: None,
            capabilities: None,
            excluded_peers: Vec::new(),
            online: true,
        })
        .unwrap();
        assert!(value.get("metadata").is_none());
        assert!(value.get("capabilities").is_none());
    }
}