use crate::client::NativeTrustToken;
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, 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::{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 PROTOCOL_VERSION: u8 = 1;
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 = 30_000;
const MAX_RECONNECT_ATTEMPTS: u8 = 12;
const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30);
const MAX_EXCLUDED_PEERS: usize = 250;
type TokenProvider = Arc<dyn Fn() -> Option<String> + Send + Sync>;
type TrustTokenProvider = Arc<dyn Fn() -> Option<NativeTrustToken> + Send + Sync>;
#[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 {
error
.downcast_ref::<GatewayConnectError>()
.map(|classified| classified.retryable)
.unwrap_or(true)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NativeCoordinationEnvironment {
Staging,
Production,
}
pub struct NativeCoordinationGatewayOptions {
pub project_id: String,
pub endpoint: String,
pub control_plane_endpoint: String,
pub environment: NativeCoordinationEnvironment,
pub app_tag: String,
pub device_id: String,
pub platform_type: String,
pub auth_token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
pub native_trust_token_provider: Box<dyn Fn() -> Option<NativeTrustToken> + Send + Sync>,
}
#[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 GatewayCredential {
protocol_version: u8,
gateway_url: String,
route_key: String,
token: String,
expires_at_ms: u64,
}
#[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(rename = "auth.refreshed")]
AuthRefreshed {
#[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,
control_plane_endpoint: String,
app_tag: String,
device_id: String,
runtime_instance_id: String,
platform_type: String,
auth_token_provider: TokenProvider,
native_trust_token_provider: TrustTokenProvider,
http: reqwest::Client,
shared: Arc<SharedState>,
commands: mpsc::Sender<Command>,
}
impl NativeCoordinationGatewaySignaling {
pub fn new(options: NativeCoordinationGatewayOptions) -> Result<Arc<Self>> {
let project_id = required("project_id", options.project_id)?;
let endpoint = validate_endpoint("endpoint", options.endpoint, true)?;
let control_plane_endpoint = validate_endpoint(
"control_plane_endpoint",
options.control_plane_endpoint,
false,
)?;
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)?;
match options.environment {
NativeCoordinationEnvironment::Staging if project_id != "openrtc-platform-staging" => {
bail!("native coordination staging project is invalid")
}
NativeCoordinationEnvironment::Production if project_id != "pluto-rtc-prod" => {
bail!("native coordination production project is invalid")
}
_ => {}
}
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,
control_plane_endpoint,
app_tag,
device_id,
runtime_instance_id: format!("runtime:{}", Uuid::new_v4()),
platform_type,
auth_token_provider: Arc::from(options.auth_token_provider),
native_trust_token_provider: Arc::from(options.native_trust_token_provider),
http: reqwest::Client::builder()
.timeout(REQUEST_TIMEOUT)
.build()
.context("build native coordination HTTP client")?,
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,
) -> Result<GatewayCredential> {
let auth_token = (self.auth_token_provider)()
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| anyhow!("native coordination authentication token is unavailable"))?;
let trust = (self.native_trust_token_provider)()
.filter(|value| value.expires_at_ms > now_ms().saturating_add(30_000))
.ok_or_else(|| anyhow!("native coordination trust token is unavailable"))?;
let ticket_fingerprint = base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(Sha256::digest(desired.ticket.as_bytes()));
let body = serde_json::json!({
"data": {
"avenue": { "kind": "user", "id": desired.user_id },
"deviceId": self.device_id,
"runtimeInstanceId": self.runtime_instance_id,
"ticketFingerprint": ticket_fingerprint,
"purpose": purpose,
"idempotencyKey": random_id("credential")
}
});
let url = format!(
"{}/mintNativeCoordinationCredential",
self.control_plane_endpoint.trim_end_matches('/')
);
let response = self
.http
.post(url)
.bearer_auth(auth_token)
.header("X-Firebase-AppCheck", trust.token)
.json(&body)
.send()
.await
.context("mint native coordination credential")?;
let status = response.status();
let bytes = response
.bytes()
.await
.context("read native coordination credential response")?;
if !status.is_success() {
return Err(gateway_connect_error(
format!(
"native coordination credential denied with status {}",
status.as_u16()
),
status.as_u16() == 408
|| status.as_u16() == 425
|| status.as_u16() == 429
|| status.is_server_error(),
));
}
#[derive(Deserialize)]
struct CallableResponse {
result: Option<GatewayCredential>,
data: Option<GatewayCredential>,
}
let decoded: CallableResponse =
serde_json::from_slice(&bytes).context("decode native coordination credential")?;
let credential = decoded
.result
.or(decoded.data)
.ok_or_else(|| anyhow!("native coordination credential response is empty"))?;
if credential.protocol_version != 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: &GatewayCredential) -> 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!(
"{}/v1/connect/{}",
url.path().trim_end_matches('/'),
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,
}
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;
};
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)) => {
match refresh_authentication(&adapter, &mut active.socket).await {
Ok(expires_at_ms) => {
active.credential_expires_at_ms = expires_at_ms;
}
Err(error) => {
eprintln!(
"[openrtc][coordination-gateway][auth-refresh-failed] retryable={} error={}",
is_retryable_gateway_error(&error),
error,
);
*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()));
}
}
}
}
}
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).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": PROTOCOL_VERSION,
"type": "auth",
"token": credential.token,
"device": gateway_device_value(adapter, &desired).await
})
.to_string()
.into(),
))
.await?;
let ready = 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 { .. } => return Ok::<(), anyhow::Error>(()),
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")??;
let _ = ready;
Ok(ConnectedGateway {
socket,
credential_expires_at_ms: credential.expires_at_ms,
})
}
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": 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,
) -> Result<u64> {
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").await?;
socket
.send(Message::Text(
serde_json::json!({
"v": 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 } => {
return Ok(_expires_at_ms);
}
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 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": 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": 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": 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": 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": 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::Ack { .. }
| ServerFrame::Pong => {}
}
Ok(())
}
async fn presence_frame(
adapter: &NativeCoordinationGatewaySignaling,
desired: &DesiredPresence,
) -> serde_json::Value {
serde_json::json!({
"v": 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),
)
}
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<()> {
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());
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::*;
#[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");
}
#[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 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 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());
}
}