use crate::firebase::device_registry::{
offline_device_expires_at_ms, FirebaseDeviceRegistryContract,
SharedDeviceRecord, DEVICE_STALE_HEARTBEAT_MS, DEVICE_STALE_SWEEP_INTERVAL_SECS,
};
use crate::firebase::schema::AppNamespace;
use crate::logging::{env_flag_enabled, log_info, log_warn};
use crate::signaling::{Device, SignalingBackend, SignalingEnvelope};
use anyhow::Result;
use async_trait::async_trait;
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
use firestore::*;
use gcloud_sdk::ExternalJwtFunctionSource;
use gcloud_sdk::Token;
use gcloud_sdk::TokenSourceType;
use iroh_tickets::endpoint::EndpointTicket;
use std::future::Future;
use std::hash::{Hash, Hasher};
use std::str::FromStr;
fn signaling_verbose() -> bool {
env_flag_enabled("OPENRTC_SIGNALING_VERBOSE")
}
#[derive(Clone, Debug)]
pub struct LocalMemListenStateStorage {
tokens: std::sync::Arc<
tokio::sync::RwLock<
std::collections::HashMap<
firestore::FirestoreListenerTarget,
firestore::FirestoreListenerToken,
>,
>,
>,
}
impl LocalMemListenStateStorage {
pub fn new() -> Self {
Self {
tokens: std::sync::Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())),
}
}
}
#[async_trait::async_trait]
impl firestore::FirestoreResumeStateStorage for LocalMemListenStateStorage {
async fn read_resume_state(
&self,
target: &firestore::FirestoreListenerTarget,
) -> Result<
Option<firestore::FirestoreListenerTargetResumeType>,
Box<dyn std::error::Error + Send + Sync>,
> {
Ok(self
.tokens
.read()
.await
.get(target)
.cloned()
.map(firestore::FirestoreListenerTargetResumeType::Token))
}
async fn update_resume_token(
&self,
target: &firestore::FirestoreListenerTarget,
token: firestore::FirestoreListenerToken,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
self.tokens.write().await.insert(target.clone(), token);
Ok(())
}
}
#[derive(Clone)]
pub struct NativeFirestoreSignalingBackend {
project_id: String,
db: std::sync::Arc<tokio::sync::RwLock<Option<FirestoreDb>>>,
auth_fingerprint: std::sync::Arc<tokio::sync::RwLock<Option<u64>>>,
namespace: AppNamespace,
token_provider: std::sync::Arc<dyn Fn() -> Option<String> + Send + Sync>,
app_backgrounded_provider: std::sync::Arc<dyn Fn() -> bool + Send + Sync>,
}
impl NativeFirestoreSignalingBackend {
pub fn new(
project_id: &str,
app_tag: String,
token_provider: std::sync::Arc<dyn Fn() -> Option<String> + Send + Sync>,
app_backgrounded_provider: std::sync::Arc<dyn Fn() -> bool + Send + Sync>,
) -> Self {
Self {
project_id: project_id.to_string(),
db: std::sync::Arc::new(tokio::sync::RwLock::new(None)),
auth_fingerprint: std::sync::Arc::new(tokio::sync::RwLock::new(None)),
namespace: AppNamespace::new(app_tag),
token_provider,
app_backgrounded_provider,
}
}
fn is_app_backgrounded(&self) -> bool {
(self.app_backgrounded_provider)()
}
fn current_token_fingerprint(&self) -> Option<u64> {
let token = (self.token_provider)()?;
let trimmed = token.trim();
if trimmed.is_empty() {
return None;
}
let mut hasher = std::collections::hash_map::DefaultHasher::new();
trimmed.hash(&mut hasher);
Some(hasher.finish())
}
fn is_unauthenticated_error(error: &anyhow::Error) -> bool {
let text = error.to_string().to_ascii_lowercase();
text.contains("unauthenticated")
|| text.contains("missing or invalid authentication")
|| text.contains("request does not have valid authentication credentials")
|| text.contains("status code: 401")
}
async fn invalidate_db_cache(&self) {
{
let mut db_guard = self.db.write().await;
*db_guard = None;
}
let mut fp_guard = self.auth_fingerprint.write().await;
*fp_guard = None;
}
pub(super) async fn with_auth_retry<T, F, Fut>(&self, mut operation: F) -> Result<T>
where
F: FnMut(FirestoreDb) -> Fut,
Fut: Future<Output = Result<T>>,
{
let db = self.get_db().await?;
match operation(db).await {
Ok(value) => Ok(value),
Err(error) if Self::is_unauthenticated_error(&error) => {
if signaling_verbose() {
log_warn(&format!(
"[OPENRTC][SIGNALING][native] auth error detected; invalidating Firestore DB cache and retrying once"
));
}
self.invalidate_db_cache().await;
let refreshed_db = self.get_db().await?;
operation(refreshed_db).await
}
Err(error) => Err(error),
}
}
fn user_token_source(&self) -> Option<TokenSourceType> {
if !(self.token_provider)().is_some_and(|value| !value.trim().is_empty()) {
return None;
}
let token_provider = self.token_provider.clone();
let token_source = ExternalJwtFunctionSource::new(move || {
let token_provider = token_provider.clone();
async move {
let token = token_provider()
.filter(|value| !value.trim().is_empty())
.ok_or_else(|| gcloud_sdk::error::ErrorKind::TokenSource)?;
let expires_at = Self::parse_id_token_expiry_utc(&token).unwrap_or_else(|| {
chrono::Utc::now() + chrono::Duration::minutes(10)
});
Ok(Token::new("Bearer".to_string(), token.into(), expires_at))
}
});
Some(TokenSourceType::ExternalSource(Box::new(token_source)))
}
fn parse_id_token_expiry_utc(id_token: &str) -> Option<chrono::DateTime<chrono::Utc>> {
let mut parts = id_token.split('.');
let _header = parts.next()?;
let payload = parts.next()?;
let _signature = parts.next()?;
let payload_bytes = URL_SAFE_NO_PAD.decode(payload).ok()?;
let payload_json: serde_json::Value = serde_json::from_slice(&payload_bytes).ok()?;
let exp_seconds = payload_json.get("exp")?.as_i64()?;
chrono::DateTime::<chrono::Utc>::from_timestamp(exp_seconds, 0)
}
async fn get_db(&self) -> Result<FirestoreDb> {
let current_fp = self.current_token_fingerprint();
{
let guard = self.db.read().await;
let fp_guard = self.auth_fingerprint.read().await;
if let Some(db) = guard.as_ref() {
if *fp_guard == current_fp {
return Ok(db.clone());
}
}
}
let mut guard = self.db.write().await;
let mut fp_guard = self.auth_fingerprint.write().await;
if let Some(db) = guard.as_ref() {
if *fp_guard == current_fp {
return Ok(db.clone());
}
}
if let Some(token_source_type) = self.user_token_source() {
let options = FirestoreDbOptions::new(self.project_id.clone());
let db = FirestoreDb::with_options_token_source(
options,
gcloud_sdk::GCP_DEFAULT_SCOPES.clone(),
token_source_type,
)
.await?;
*guard = Some(db.clone());
*fp_guard = current_fp;
Ok(db)
} else {
*guard = None;
*fp_guard = None;
Ok(FirestoreDb::new(&self.project_id).await?)
}
}
fn app_parent(&self) -> String {
format!(
"projects/{}/databases/(default)/documents/{}",
self.project_id,
self.namespace.app_root()
)
}
pub(super) fn user_parent(&self, user_id: &str) -> String {
format!(
"projects/{}/databases/(default)/documents/{}",
self.project_id,
if self.namespace.is_space() {
self.namespace.app_root()
} else {
self.namespace.user_root(user_id)
}
)
}
fn extract_device_id(metadata: Option<&str>, fallback_node_id: &str) -> String {
let parsed = metadata
.and_then(|raw| serde_json::from_str::<serde_json::Value>(raw).ok())
.and_then(|value| value.as_object().cloned());
let candidate = parsed
.as_ref()
.and_then(|obj| {
obj.get("deviceId")
.or_else(|| obj.get("device_id"))
.and_then(|value| value.as_str())
})
.map(str::trim)
.filter(|value| !value.is_empty());
candidate
.map(|value| value.to_string())
.unwrap_or_else(|| fallback_node_id.to_string())
}
fn capabilities_value(
capabilities: &crate::signaling::DeviceCapabilities,
) -> crate::signaling::DeviceCapabilities {
crate::signaling::DeviceCapabilities {
can_host: capabilities.can_host,
can_sync: capabilities.can_sync,
read_only: capabilities.read_only,
}
}
pub(super) fn field_as_string(
fields: &std::collections::HashMap<String, gcloud_sdk::google::firestore::v1::Value>,
key: &str,
) -> Option<String> {
fields.get(key).and_then(|value| {
value
.value_type
.as_ref()
.and_then(|value_type| match value_type {
gcloud_sdk::google::firestore::v1::value::ValueType::StringValue(v) => {
Some(v.clone())
}
gcloud_sdk::google::firestore::v1::value::ValueType::IntegerValue(v) => {
Some(v.to_string())
}
_ => None,
})
})
}
pub(super) fn field_as_bool(
fields: &std::collections::HashMap<String, gcloud_sdk::google::firestore::v1::Value>,
key: &str,
) -> Option<bool> {
fields.get(key).and_then(|value| {
value
.value_type
.as_ref()
.and_then(|value_type| match value_type {
gcloud_sdk::google::firestore::v1::value::ValueType::BooleanValue(v) => {
Some(*v)
}
_ => None,
})
})
}
pub(super) fn field_as_millis(
fields: &std::collections::HashMap<String, gcloud_sdk::google::firestore::v1::Value>,
key: &str,
) -> Option<i64> {
fields.get(key).and_then(|value| {
value
.value_type
.as_ref()
.and_then(|value_type| match value_type {
gcloud_sdk::google::firestore::v1::value::ValueType::IntegerValue(v) => {
Some(*v)
}
gcloud_sdk::google::firestore::v1::value::ValueType::StringValue(v) => {
v.parse::<i64>().ok()
}
gcloud_sdk::google::firestore::v1::value::ValueType::TimestampValue(ts) => {
let millis = ts
.seconds
.saturating_mul(1_000)
.saturating_add((ts.nanos as i64) / 1_000_000);
Some(millis)
}
_ => None,
})
})
}
pub(super) fn app_tag(&self) -> &str {
self.namespace.app_tag()
}
pub(super) fn shared_device_record_from_fields(
fields: &std::collections::HashMap<String, gcloud_sdk::google::firestore::v1::Value>,
) -> SharedDeviceRecord {
let capabilities = fields.get("capabilities").and_then(|v| {
v.value_type.as_ref().and_then(|vt| match vt {
gcloud_sdk::google::firestore::v1::value::ValueType::MapValue(map_value) => {
let get_cap_bool = |key: &str| {
map_value
.fields
.get(key)
.and_then(|value| value.value_type.as_ref())
.and_then(|value_type| match value_type {
gcloud_sdk::google::firestore::v1::value::ValueType::BooleanValue(value) => {
Some(*value)
}
_ => None,
})
.unwrap_or(false)
};
Some(crate::signaling::DeviceCapabilities {
can_host: get_cap_bool("canHost"),
can_sync: get_cap_bool("canSync"),
read_only: get_cap_bool("readOnly"),
})
}
_ => None,
})
});
let excluded_peers = fields
.get("excludedPeers")
.and_then(|value| value.value_type.as_ref())
.and_then(|value_type| match value_type {
gcloud_sdk::google::firestore::v1::value::ValueType::ArrayValue(array) => Some(
array
.values
.iter()
.filter_map(|value| match value.value_type.as_ref()? {
gcloud_sdk::google::firestore::v1::value::ValueType::StringValue(
value,
) => Some(value.clone()),
_ => None,
})
.collect(),
),
_ => None,
})
.unwrap_or_default();
SharedDeviceRecord {
canonical_device_id: Self::field_as_string(fields, "deviceId"),
app_tag: Self::field_as_string(fields, "appTag"),
user_id: Self::field_as_string(fields, "userId"),
device_name: Self::field_as_string(fields, "deviceName").unwrap_or_default(),
platform_type: Self::field_as_string(fields, "platformType"),
capabilities,
session_id: Self::field_as_string(fields, "sessionId"),
node_id: Self::field_as_string(fields, "nodeId"),
tag: Self::field_as_string(fields, "tag"),
kind: Self::field_as_string(fields, "kind"),
metadata: Self::field_as_string(fields, "metadata"),
online: Self::field_as_bool(fields, "online").unwrap_or(false),
ticket: Self::field_as_string(fields, "ticket"),
last_seen_at: fields
.get("lastSeenAt")
.map(|_| {
Self::field_as_millis(fields, "lastSeenAt")
.map(|value| serde_json::Value::Number(value.into()))
.unwrap_or_else(|| {
Self::field_as_string(fields, "lastSeenAt")
.map(serde_json::Value::String)
.unwrap_or(serde_json::Value::Null)
})
})
.filter(|value| !value.is_null()),
expires_at: Self::field_as_millis(fields, "expiresAt")
.map(|value| serde_json::Value::Number(value.into()))
.or_else(|| {
Self::field_as_string(fields, "expiresAt").map(serde_json::Value::String)
}),
created_at: fields
.get("createdAt")
.map(|_| {
Self::field_as_millis(fields, "createdAt")
.map(|value| serde_json::Value::Number(value.into()))
.unwrap_or_else(|| {
Self::field_as_string(fields, "createdAt")
.map(serde_json::Value::String)
.unwrap_or(serde_json::Value::Null)
})
})
.filter(|value| !value.is_null()),
updated_at: fields
.get("updatedAt")
.map(|_| {
Self::field_as_millis(fields, "updatedAt")
.map(|value| serde_json::Value::Number(value.into()))
.unwrap_or_else(|| {
Self::field_as_string(fields, "updatedAt")
.map(serde_json::Value::String)
.unwrap_or(serde_json::Value::Null)
})
})
.filter(|value| !value.is_null()),
excluded_peers,
}
}
fn device_event_from_firestore_fields(
&self,
document_name: &str,
fields: &std::collections::HashMap<String, gcloud_sdk::google::firestore::v1::Value>,
exclude_node_id: Option<&str>,
now_ms: i64,
) -> Option<crate::signaling::DeviceEvent> {
let document_id = document_name
.split('/')
.last()
.unwrap_or(document_name)
.to_string();
let shared = Self::shared_device_record_from_fields(fields);
if shared.effective_app_tag() != Some(self.registry_app_tag()) {
return None;
}
if !crate::firebase::device_registry::is_device_record_discoverable(
&shared,
now_ms,
self.registry_retention(),
) {
return Some(crate::signaling::DeviceEvent::Removed {
device_id: shared.public_device_id(&document_id),
});
}
self.project_discovered_device(document_id, shared, exclude_node_id, now_ms)
.map(|device| crate::signaling::DeviceEvent::Modified { device })
}
async fn resolve_device_doc_id(
db: &FirestoreDb,
parent: &str,
device_id: &str,
) -> Result<Option<String>> {
let docs = db
.fluent()
.select()
.from("devices")
.parent(parent)
.query()
.await?;
for doc in docs {
let doc_id = doc.name.split('/').last().unwrap_or("").to_string();
let fields = doc.fields;
let canonical_device_id = Self::field_as_string(&fields, "deviceId");
let node_id = Self::field_as_string(&fields, "nodeId");
if doc_id == device_id
|| canonical_device_id.as_deref() == Some(device_id)
|| node_id.as_deref() == Some(device_id)
{
return Ok(Some(doc_id));
}
}
Ok(None)
}
}
impl FirebaseDeviceRegistryContract for NativeFirestoreSignalingBackend {
fn registry_app_tag(&self) -> &str {
self.namespace.app_tag()
}
fn registry_retention(&self) -> crate::firebase::device_registry::DiscoveryRetention {
if self.namespace.is_space() {
crate::firebase::device_registry::DiscoveryRetention::Ephemeral
} else {
crate::firebase::device_registry::DiscoveryRetention::Persistent
}
}
}
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub(super) struct DeviceFields {
pub app_tag: Option<String>,
pub user_id: Option<String>,
pub device_id: Option<String>,
pub device_name: Option<String>,
pub node_id: Option<String>,
pub platform_type: Option<String>,
pub session_id: Option<String>,
pub tag: Option<String>,
pub kind: Option<String>,
pub capabilities: Option<crate::signaling::DeviceCapabilities>,
pub metadata: Option<String>,
pub online: Option<bool>,
pub ticket: Option<String>,
pub last_seen_at: Option<i64>,
pub updated_at: Option<i64>,
pub created_at: Option<i64>,
pub expires_at: Option<i64>,
#[serde(default)]
pub excluded_peers: Vec<String>,
}
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
struct DeviceUpdateFields {
pub user_id: Option<String>,
pub device_name: Option<String>,
pub capabilities: Option<crate::signaling::DeviceCapabilities>,
pub metadata: Option<String>,
pub updated_at: Option<i64>,
}
#[async_trait]
impl SignalingBackend for NativeFirestoreSignalingBackend {
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 signaling_verbose() {
log_info(&format!(
"[OPENRTC][SIGNALING][native] update_presence start user_id={} node_id={} online={} tag={} ticket_len={} metadata_len={}",
user_id,
local_node_id,
is_online,
self.namespace.app_tag(),
ticket_str.len(),
metadata.map(|v| v.len()).unwrap_or(0),
));
}
let parent = self.user_parent(user_id);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
let ttl_i64 = ttl_ms.min(i64::MAX as u64) as i64;
let (iroh_ticket_part, _) = crate::session_token::split_compound_ticket(ticket_str.trim());
EndpointTicket::from_str(iroh_ticket_part)
.map_err(|e| anyhow::anyhow!("invalid endpoint ticket for presence: {}", e))?;
let ticket = Some(ticket_str.to_string());
let canonical_device_id = Self::extract_device_id(metadata, local_node_id);
let parent_for_lookup = parent.clone();
let canonical_device_id_for_lookup = canonical_device_id.clone();
let (target_doc_id, existing_device_name, existing_created_at, should_set_created_at) =
self.with_auth_retry(move |db| {
let parent = parent_for_lookup.clone();
let canonical_device_id = canonical_device_id_for_lookup.clone();
async move {
let docs = db
.fluent()
.select()
.from("devices")
.parent(&parent)
.query()
.await?;
for doc in docs {
let doc_id = doc.name.split('/').last().unwrap_or("").to_string();
let fields = doc.fields;
let existing_device_id = Self::field_as_string(&fields, "deviceId");
let existing_node_id = Self::field_as_string(&fields, "nodeId");
if doc_id == canonical_device_id
|| existing_device_id.as_deref() == Some(canonical_device_id.as_str())
|| existing_node_id.as_deref() == Some(canonical_device_id.as_str())
{
let existing_name = Self::field_as_string(&fields, "deviceName")
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty());
let existing_created_at = Self::field_as_millis(&fields, "createdAt");
return Ok((doc_id, existing_name, existing_created_at, false));
}
}
Ok((canonical_device_id, None, None, true))
}
})
.await?;
let should_set_device_name = existing_device_name.is_none();
let fields = DeviceFields {
app_tag: Some(self.namespace.app_tag().to_string()),
user_id: Some(user_id.to_string()),
device_id: Some(canonical_device_id.clone()),
device_name: existing_device_name.or_else(|| Some(name.to_string())),
node_id: Some(local_node_id.to_string()),
platform_type: Some(crate::native_device::default_platform_type().to_string()),
session_id: Some(canonical_device_id.clone()),
tag: Some(self.namespace.app_tag().to_string()),
kind: Some("device".to_string()),
capabilities: Some(crate::signaling::DeviceCapabilities {
can_host: false,
can_sync: false,
read_only: false,
}),
metadata: metadata.map(|m| m.to_string()),
online: Some(is_online),
ticket,
last_seen_at: Some(now),
updated_at: Some(now),
created_at: Some(existing_created_at.unwrap_or(now)),
expires_at: Some(now.saturating_add(ttl_i64)),
excluded_peers: vec![],
};
let parent_for_update = parent.clone();
let target_doc_id_for_update = target_doc_id.clone();
let fields_for_update = fields.clone();
let mut update_fields = vec![
"appTag",
"userId",
"deviceId",
"nodeId",
"platformType",
"sessionId",
"tag",
"kind",
"capabilities",
"online",
"lastSeenAt",
"updatedAt",
"expiresAt",
"ticket",
"metadata",
];
if should_set_device_name {
update_fields.push("deviceName");
}
if should_set_created_at {
update_fields.push("createdAt");
}
let update_fields_for_update = update_fields.clone();
let should_replace_space_doc = self.namespace.is_space();
self.with_auth_retry(move |db| {
let parent = parent_for_update.clone();
let target_doc_id = target_doc_id_for_update.clone();
let fields = fields_for_update.clone();
let update_fields = update_fields_for_update.clone();
async move {
if should_replace_space_doc {
let update_result = db
.fluent()
.update()
.fields(update_fields.clone())
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await;
if update_result.is_ok() {
return Ok(());
}
let _ = db
.fluent()
.delete()
.from("devices")
.parent(&parent)
.document_id(&target_doc_id)
.execute()
.await;
let insert_result = db
.fluent()
.insert()
.into("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<DeviceFields>()
.await;
if insert_result.is_ok() {
return Ok(());
}
db.fluent()
.update()
.fields(update_fields)
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
} else {
db.fluent()
.update()
.fields(update_fields)
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
}
Ok(())
}
})
.await?;
if signaling_verbose() {
log_info(&format!(
"[OPENRTC][SIGNALING][native] update_presence success user_id={} device_id={} node_id={} online={}",
user_id, target_doc_id, local_node_id, is_online
));
}
Ok(())
}
async fn set_offline(&self, user_id: &str, local_node_id: &str) -> Result<()> {
let parent = self.user_parent(user_id);
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
let parent_for_update = parent.clone();
let app_tag_for_update = self.namespace.app_tag().to_string();
let user_id_for_update = user_id.to_string();
let local_node_id_for_update = local_node_id.to_string();
self.with_auth_retry(move |db| {
let parent = parent_for_update.clone();
let app_tag = app_tag_for_update.clone();
let scoped_user_id = user_id_for_update.clone();
let local_node_id = local_node_id_for_update.clone();
async move {
let Some(target_doc_id) =
Self::resolve_device_doc_id(&db, &parent, &local_node_id).await?
else {
if signaling_verbose() {
log_info(&format!(
"[OPENRTC][SIGNALING][native] set_offline skipped user_id={} node_id={} reason=device-doc-missing",
scoped_user_id, local_node_id
));
}
return Ok(());
};
let fields = DeviceFields {
app_tag: Some(app_tag.clone()),
user_id: Some(scoped_user_id.clone()),
device_id: Some(target_doc_id.clone()),
device_name: None,
node_id: Some(local_node_id.clone()),
platform_type: None,
session_id: Some(target_doc_id.clone()),
tag: Some(app_tag),
kind: Some("device".to_string()),
capabilities: None,
metadata: None,
online: Some(false),
ticket: None,
last_seen_at: None,
updated_at: Some(now_ms),
created_at: None,
expires_at: Some(offline_device_expires_at_ms(now_ms)),
excluded_peers: vec![],
};
db.fluent()
.update()
.fields(vec![
"appTag",
"userId",
"deviceId",
"nodeId",
"sessionId",
"tag",
"kind",
"online",
"updatedAt",
"expiresAt",
])
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
Ok(())
}
})
.await?;
Ok(())
}
async fn update_device(
&self,
user_id: &str,
device_id: &str,
device_name: Option<&str>,
capabilities: Option<crate::signaling::DeviceCapabilities>,
metadata: Option<&str>,
) -> Result<()> {
let trimmed_name = device_name.map(str::trim);
if trimmed_name.is_some_and(|value| value.is_empty()) {
anyhow::bail!("device_name cannot be empty");
}
if trimmed_name.is_none() && capabilities.is_none() && metadata.is_none() {
return Ok(());
}
let parent = self.user_parent(user_id);
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
let parent_for_update = parent.clone();
let user_id_for_update = user_id.to_string();
let device_id_for_update = device_id.to_string();
let trimmed_name_for_update = trimmed_name.map(str::to_string);
let capabilities_for_update = capabilities.as_ref().map(Self::capabilities_value);
let metadata_for_update = metadata.map(str::to_string);
self.with_auth_retry(move |db| {
let parent = parent_for_update.clone();
let scoped_user_id = user_id_for_update.clone();
let requested_device_id = device_id_for_update.clone();
let next_name = trimmed_name_for_update.clone();
let next_capabilities = capabilities_for_update.clone();
let next_metadata = metadata_for_update.clone();
async move {
let Some(target_doc_id) =
Self::resolve_device_doc_id(&db, &parent, &requested_device_id).await?
else {
anyhow::bail!("device {} not found", requested_device_id);
};
let fields = DeviceUpdateFields {
user_id: Some(scoped_user_id),
device_name: next_name.clone(),
capabilities: next_capabilities,
metadata: next_metadata.clone(),
updated_at: Some(now_ms),
};
let mut update_mask = vec!["updatedAt", "userId"];
if next_name.is_some() {
update_mask.push("deviceName");
}
if fields.capabilities.is_some() {
update_mask.push("capabilities");
}
if next_metadata.is_some() {
update_mask.push("metadata");
}
db.fluent()
.update()
.fields(update_mask)
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
Ok(())
}
})
.await?;
Ok(())
}
async fn delete_device(&self, user_id: &str, device_id: &str) -> Result<()> {
let parent = self.user_parent(user_id);
let parent_for_delete = parent.clone();
let device_id_for_delete = device_id.to_string();
self.with_auth_retry(move |db| {
let parent = parent_for_delete.clone();
let requested_device_id = device_id_for_delete.clone();
async move {
let Some(target_doc_id) =
Self::resolve_device_doc_id(&db, &parent, &requested_device_id).await?
else {
anyhow::bail!("device {} not found", requested_device_id);
};
db.fluent()
.delete()
.from("devices")
.document_id(&target_doc_id)
.parent(&parent)
.execute()
.await?;
Ok(())
}
})
.await?;
Ok(())
}
async fn set_excluded_peers(
&self,
user_id: &str,
local_node_id: &str,
excluded_peers: &[String],
) -> Result<()> {
let parent = self.user_parent(user_id);
let local_node_id = local_node_id.to_string();
let excluded_peers = excluded_peers.to_vec();
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
self.with_auth_retry(move |db| {
let parent = parent.clone();
let local_node_id = local_node_id.clone();
let excluded_peers = excluded_peers.clone();
async move {
let target_doc_id =
match Self::resolve_device_doc_id(&db, &parent, &local_node_id).await? {
Some(id) => id,
None => return Ok(()), };
let fields = DeviceFields {
app_tag: None,
user_id: None,
device_id: None,
device_name: None,
node_id: None,
platform_type: None,
session_id: None,
tag: None,
kind: None,
capabilities: None,
metadata: None,
online: None,
ticket: None,
last_seen_at: None,
updated_at: Some(now),
created_at: None,
expires_at: None,
excluded_peers,
};
db.fluent()
.update()
.fields(vec!["excludedPeers", "updatedAt"])
.in_col("devices")
.document_id(&target_doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
Ok(())
}
})
.await
}
async fn search_devices(
&self,
user_id: &str,
exclude_node_id: Option<&str>,
) -> Result<Vec<Device>> {
self.list_devices(user_id, exclude_node_id)
.await
.map(|devices| devices.into_iter().filter(|device| device.online).collect())
}
async fn list_devices(
&self,
user_id: &str,
exclude_node_id: Option<&str>,
) -> Result<Vec<Device>> {
let search_started = std::time::Instant::now();
let parent = self.user_parent(user_id);
if !self.is_app_backgrounded() {
if let Err(error) = self
.sweep_stale_devices(user_id, DEVICE_STALE_HEARTBEAT_MS)
.await
{
if signaling_verbose() {
log_warn(&format!(
"[OPENRTC][SIGNALING][native] stale sweep failed during search_devices user_id={} error={}",
user_id, error
));
}
}
}
let parent_for_query = parent.clone();
let docs = self
.with_auth_retry(move |db| {
let parent = parent_for_query.clone();
async move {
let docs = db
.fluent()
.select()
.from("devices")
.parent(&parent)
.query()
.await?;
Ok(docs)
}
})
.await?;
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
let filtered: Vec<Device> = docs
.into_iter()
.filter_map(|doc| {
let id = doc.name.split('/').last().unwrap_or("").to_string();
let fields = doc.fields;
let shared = Self::shared_device_record_from_fields(&fields);
if shared.effective_app_tag() != Some(self.registry_app_tag()) {
return None;
}
self.project_discovered_device(id, shared, exclude_node_id, now_ms)
})
.collect();
if signaling_verbose() {
log_info(&format!(
"[OPENRTC][SIGNALING][native] list_devices user_id={} tag={} exclude_node_id={} count={} elapsed_ms={}",
user_id,
self.namespace.app_tag(),
exclude_node_id.unwrap_or("<none>"),
filtered.len(),
search_started.elapsed().as_millis(),
));
}
Ok(filtered)
}
async fn send_message(
&self,
sender_id: &str,
target_id: &str,
payload: &str,
state: Option<&str>,
reply_payload: Option<&str>,
) -> Result<String> {
let parent = self.app_parent();
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_millis() as i64;
const MESSAGE_TTL_MS: i64 = 24 * 60 * 60 * 1000;
let expires_at = now + MESSAGE_TTL_MS;
let auth_uid = (self.token_provider)()
.as_deref()
.and_then(super::auth_claims::firebase_uid_from_id_token);
let mut fields = SignalingEnvelope {
app_tag: Some(self.namespace.app_tag().to_string()),
sender_id: sender_id.to_string(),
target_id: target_id.to_string(),
payload: payload.to_string(),
state: state.map(|s| s.to_string()),
reply_payload: reply_payload.map(|r| r.to_string()),
timestamp: now,
sender_user_id: None,
target_user_id: None,
expires_at: Some(expires_at),
};
if !self.namespace.is_space() {
let uid = auth_uid.ok_or_else(|| {
anyhow::anyhow!(
"send_message requires an authenticated Firebase user for app-scoped signaling"
)
})?;
fields.sender_user_id = Some(uid.clone());
fields.target_user_id = Some(uid);
} else if let Some(uid) = auth_uid {
fields.sender_user_id = Some(uid.clone());
fields.target_user_id = Some(uid);
}
let doc_id = uuid::Uuid::new_v4().to_string();
let parent_for_insert = parent.clone();
let doc_id_for_insert = doc_id.clone();
let fields_for_insert = fields.clone();
self.with_auth_retry(move |db| {
let parent = parent_for_insert.clone();
let doc_id = doc_id_for_insert.clone();
let fields = fields_for_insert.clone();
async move {
db.fluent()
.insert()
.into("messages")
.document_id(&doc_id)
.parent(&parent)
.object(&fields)
.execute::<()>()
.await?;
Ok(())
}
})
.await?;
Ok(doc_id)
}
async fn subscribe_devices(
&self,
user_id: &str,
) -> Result<futures::stream::BoxStream<'static, Result<Vec<crate::signaling::DeviceEvent>>>>
{
let parent = self.user_parent(user_id);
let db = self.get_db().await?;
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let mut listener = db
.create_listener(LocalMemListenStateStorage::new())
.await?;
if let Err(error) = db
.fluent()
.select()
.from("devices")
.parent(&parent)
.listen()
.add_target(
firestore::FirestoreListenerTarget::try_from(1).unwrap(),
&mut listener,
)
{
let anyhow_error = anyhow::anyhow!(error.to_string());
if Self::is_unauthenticated_error(&anyhow_error) {
self.invalidate_db_cache().await;
}
return Err(anyhow_error);
}
let tx_clone = tx.clone();
let projection_backend = self.clone();
let cleanup_backend = self.clone();
let cleanup_user_id = user_id.to_string();
let cleanup_tx = tx.clone();
tokio::spawn(async move {
let mut cleanup_interval = tokio::time::interval(std::time::Duration::from_secs(
DEVICE_STALE_SWEEP_INTERVAL_SECS,
));
loop {
tokio::select! {
_ = cleanup_tx.closed() => break,
_ = cleanup_interval.tick() => {
if cleanup_backend.is_app_backgrounded() {
continue;
}
if let Err(error) = cleanup_backend
.sweep_stale_devices(&cleanup_user_id, DEVICE_STALE_HEARTBEAT_MS)
.await
{
if signaling_verbose() {
log_warn(&format!(
"[OPENRTC][SIGNALING][native] stale sweep failed in subscribe_devices user_id={} error={}",
cleanup_user_id, error
));
}
}
}
}
}
});
listener
.start(move |event| {
let tx = tx_clone.clone();
let projection_backend = projection_backend.clone();
async move {
use gcloud_sdk::google::firestore::v1::listen_response::ResponseType;
match event {
ResponseType::DocumentChange(doc_change) => {
if let Some(doc) = doc_change.document {
let now_ms = crate::firebase::now_millis_u64() as i64;
if let Some(event) = projection_backend
.device_event_from_firestore_fields(
&doc.name,
&doc.fields,
None,
now_ms,
)
{
let _ = tx.send(Ok(vec![event]));
}
}
}
ResponseType::DocumentDelete(doc_delete) => {
let _ = tx.send(Ok(vec![crate::signaling::DeviceEvent::Removed {
device_id: doc_delete
.document
.split('/')
.last()
.unwrap_or(&doc_delete.document)
.to_string(),
}]));
}
ResponseType::DocumentRemove(doc_remove) => {
let _ = tx.send(Ok(vec![crate::signaling::DeviceEvent::Removed {
device_id: doc_remove
.document
.split('/')
.last()
.unwrap_or(&doc_remove.document)
.to_string(),
}]));
}
_ => {}
}
Ok(())
}
})
.await?;
tokio::spawn(async move {
tx.closed().await;
let mut l = listener;
let _ = l.shutdown().await;
});
use futures::StreamExt;
use tokio_stream::wrappers::UnboundedReceiverStream;
Ok(UnboundedReceiverStream::new(rx).boxed())
}
async fn create_session(&self, session: crate::signaling::SignalingSession) -> Result<()> {
let parent = self.app_parent();
let parent_for_insert = parent.clone();
let session_for_insert = session.clone();
self.with_auth_retry(move |db| {
let parent = parent_for_insert.clone();
let session = session_for_insert.clone();
async move {
db.fluent()
.insert()
.into("sessions")
.document_id(&session.connection_id)
.parent(&parent)
.object(&session)
.execute::<()>()
.await?;
Ok(())
}
})
.await?;
Ok(())
}
async fn update_session(&self, session_id: &str, update_data: serde_json::Value) -> Result<()> {
let parent = self.app_parent();
#[derive(serde::Serialize, serde::Deserialize)]
struct PartialUpdate {
#[serde(flatten)]
data: serde_json::Value,
}
let parent_for_update = parent.clone();
let session_id_for_update = session_id.to_string();
self.with_auth_retry(move |db| {
let parent = parent_for_update.clone();
let session_id = session_id_for_update.clone();
let update_data = update_data.clone();
async move {
db.fluent()
.update()
.in_col("sessions")
.document_id(&session_id)
.parent(&parent)
.object(&PartialUpdate { data: update_data })
.execute::<()>()
.await?;
Ok(())
}
})
.await?;
Ok(())
}
async fn subscribe_sessions(
&self,
local_device_id: &str,
) -> Result<futures::stream::BoxStream<'static, Result<Vec<crate::signaling::SessionEvent>>>>
{
let parent = self.app_parent();
let db = self.get_db().await?;
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let mut listener = db
.create_listener(LocalMemListenStateStorage::new())
.await?;
if let Err(error) = db
.fluent()
.select()
.from("sessions")
.parent(&parent)
.filter(|q| q.for_all([q.field("targetDeviceId").equal(local_device_id)]))
.listen()
.add_target(
firestore::FirestoreListenerTarget::try_from(2).unwrap(),
&mut listener,
)
{
let anyhow_error = anyhow::anyhow!(error.to_string());
if Self::is_unauthenticated_error(&anyhow_error) {
self.invalidate_db_cache().await;
}
return Err(anyhow_error);
}
if let Err(error) = db
.fluent()
.select()
.from("sessions")
.parent(&parent)
.filter(|q| q.for_all([q.field("initiatorDeviceId").equal(local_device_id)]))
.listen()
.add_target(
firestore::FirestoreListenerTarget::try_from(3).unwrap(),
&mut listener,
)
{
let anyhow_error = anyhow::anyhow!(error.to_string());
if Self::is_unauthenticated_error(&anyhow_error) {
self.invalidate_db_cache().await;
}
return Err(anyhow_error);
}
let tx_clone = tx.clone();
listener
.start(move |event| {
let tx = tx_clone.clone();
async move {
use gcloud_sdk::google::firestore::v1::listen_response::ResponseType;
match event {
ResponseType::DocumentChange(doc_change) => {
if let Some(doc) = doc_change.document {
if let Ok(session) = firestore::FirestoreDb::deserialize_doc_to::<
crate::signaling::SignalingSession,
>(&doc)
{
let _ = tx.send(Ok(vec![
crate::signaling::SessionEvent::Modified { session },
]));
}
}
}
ResponseType::DocumentDelete(doc_delete) => {
let _ = tx.send(Ok(vec![crate::signaling::SessionEvent::Removed {
session_id: doc_delete
.document
.split('/')
.last()
.unwrap_or(&doc_delete.document)
.to_string(),
}]));
}
ResponseType::DocumentRemove(doc_remove) => {
let _ = tx.send(Ok(vec![crate::signaling::SessionEvent::Removed {
session_id: doc_remove
.document
.split('/')
.last()
.unwrap_or(&doc_remove.document)
.to_string(),
}]));
}
_ => {}
}
Ok(())
}
})
.await?;
tokio::spawn(async move {
tx.closed().await;
let mut l = listener;
let _ = l.shutdown().await;
});
use futures::StreamExt;
use tokio_stream::wrappers::UnboundedReceiverStream;
Ok(UnboundedReceiverStream::new(rx).boxed())
}
}
#[cfg(test)]
mod tests {
use super::*;
use gcloud_sdk::google::firestore::v1::{value::ValueType, Value};
use std::collections::HashMap;
fn backend() -> NativeFirestoreSignalingBackend {
NativeFirestoreSignalingBackend::new(
"project-test",
"app_test".to_string(),
std::sync::Arc::new(|| None),
std::sync::Arc::new(|| false),
)
}
fn string_value(value: &str) -> Value {
Value {
value_type: Some(ValueType::StringValue(value.to_string())),
}
}
fn bool_value(value: bool) -> Value {
Value {
value_type: Some(ValueType::BooleanValue(value)),
}
}
fn int_value(value: i64) -> Value {
Value {
value_type: Some(ValueType::IntegerValue(value)),
}
}
fn device_fields(now_ms: i64, updated_at: i64) -> HashMap<String, Value> {
HashMap::from([
("appTag".to_string(), string_value("app_test")),
("tag".to_string(), string_value("app_test")),
("userId".to_string(), string_value("user-1")),
("deviceId".to_string(), string_value("device-remote")),
("deviceName".to_string(), string_value("Remote")),
("nodeId".to_string(), string_value("node-remote")),
("platformType".to_string(), string_value("desktop")),
("online".to_string(), bool_value(true)),
("ticket".to_string(), string_value("ticket-remote")),
("updatedAt".to_string(), int_value(updated_at)),
("lastSeenAt".to_string(), int_value(updated_at)),
(
"expiresAt".to_string(),
int_value(now_ms + DEVICE_STALE_HEARTBEAT_MS),
),
])
}
#[test]
fn native_watch_projects_fresh_device_events_through_shared_registry() {
let now = 1_000_000;
let event = backend()
.device_event_from_firestore_fields(
"projects/p/databases/(default)/documents/apps/app_test/users/user-1/devices/doc-1",
&device_fields(now, now),
None,
now,
)
.expect("fresh device event");
match event {
crate::signaling::DeviceEvent::Modified { device } => {
assert_eq!(device.device_id, "device-remote");
assert_eq!(device.node_id.as_deref(), Some("node-remote"));
assert!(device.online);
}
other => panic!("expected modified device event, got {other:?}"),
}
}
#[test]
fn native_watch_retains_stale_online_user_scope_devices_as_modified() {
let now = 1_000_000;
let mut fields = device_fields(now, now - DEVICE_STALE_HEARTBEAT_MS - 1);
fields.remove("expiresAt");
let event = backend()
.device_event_from_firestore_fields(
"projects/p/databases/(default)/documents/apps/app_test/users/user-1/devices/doc-1",
&fields,
None,
now,
)
.expect("stale device retained event");
match event {
crate::signaling::DeviceEvent::Modified { device } => {
assert_eq!(device.device_id, "device-remote");
}
other => panic!("expected modified device event, got {other:?}"),
}
}
#[test]
fn native_watch_removes_user_scope_devices_beyond_offline_retention() {
let now = 100_000_000_000;
let mut fields = device_fields(
now,
now - crate::firebase::device_registry::DEVICE_OFFLINE_RETENTION_MS - 1,
);
fields.remove("expiresAt");
let event = backend()
.device_event_from_firestore_fields(
"projects/p/databases/(default)/documents/apps/app_test/users/user-1/devices/doc-1",
&fields,
None,
now,
)
.expect("beyond-retention removal event");
match event {
crate::signaling::DeviceEvent::Removed { device_id } => {
assert_eq!(device_id, "device-remote");
}
other => panic!("expected removed device event, got {other:?}"),
}
}
#[test]
fn native_watch_uses_shared_exclusion_filter() {
let now = 1_000_000;
let event = backend().device_event_from_firestore_fields(
"projects/p/databases/(default)/documents/apps/app_test/users/user-1/devices/doc-1",
&device_fields(now, now),
Some("node-remote"),
now,
);
assert!(event.is_none());
}
}