openrtc 1.0.4

use crate::firebase::device_registry::{
    offline_device_expires_at_ms, FirebaseDeviceRegistryContract, SharedDeviceRecord,
};
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;

const NATIVE_FIRESTORE_OPERATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(8);
const DURABLE_ROSTER_BURST_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(1);

#[derive(Clone)]
struct CachedDurableRoster {
    records: Vec<(String, SharedDeviceRecord)>,
    loaded_at: std::time::Instant,
}

impl CachedDurableRoster {
    fn is_usable(&self, listener_active: bool, now: std::time::Instant) -> bool {
        listener_active
            || now.saturating_duration_since(self.loaded_at) <= DURABLE_ROSTER_BURST_CACHE_TTL
    }
}

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>>>,
    durable_roster_cache:
        std::sync::Arc<tokio::sync::Mutex<std::collections::HashMap<String, CachedDurableRoster>>>,
    durable_roster_refresh_lock: std::sync::Arc<tokio::sync::Mutex<()>>,
    active_device_subscriptions:
        std::sync::Arc<tokio::sync::Mutex<std::collections::HashMap<String, usize>>>,
    rtdb: super::rtdb_presence::RtdbPresenceClient,
    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 {
        let rtdb_token_provider = token_provider.clone();
        let rtdb = super::rtdb_presence::RtdbPresenceClient::new(
            project_id,
            app_tag.clone(),
            std::sync::Arc::new(std::sync::Mutex::new(Box::new(move || {
                rtdb_token_provider()
            }))),
        );
        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)),
            durable_roster_cache: std::sync::Arc::new(tokio::sync::Mutex::new(
                std::collections::HashMap::new(),
            )),
            durable_roster_refresh_lock: std::sync::Arc::new(tokio::sync::Mutex::new(())),
            active_device_subscriptions: std::sync::Arc::new(tokio::sync::Mutex::new(
                std::collections::HashMap::new(),
            )),
            rtdb,
            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;
        self.durable_roster_cache.lock().await.clear();
    }

    pub(super) async fn invalidate_durable_roster(&self, user_id: &str) {
        self.durable_roster_cache.lock().await.remove(user_id);
    }

    async fn retain_device_subscription(&self, user_id: &str) {
        let mut subscriptions = self.active_device_subscriptions.lock().await;
        *subscriptions.entry(user_id.to_string()).or_default() += 1;
    }

    async fn release_device_subscription(&self, user_id: &str) {
        let mut subscriptions = self.active_device_subscriptions.lock().await;
        let remove = if let Some(count) = subscriptions.get_mut(user_id) {
            *count = count.saturating_sub(1);
            *count == 0
        } else {
            false
        };
        if remove {
            subscriptions.remove(user_id);
            drop(subscriptions);
            self.invalidate_durable_roster(user_id).await;
        }
    }

    async fn has_active_device_subscription(&self, user_id: &str) -> bool {
        self.active_device_subscriptions
            .lock()
            .await
            .get(user_id)
            .copied()
            .unwrap_or(0)
            > 0
    }

    async fn durable_roster_records(
        &self,
        user_id: &str,
        parent: &str,
    ) -> Result<(Vec<(String, SharedDeviceRecord)>, bool)> {
        let backend = self.clone();
        let parent = parent.to_string();
        self.durable_roster_records_with(user_id, move || async move {
            let parent_for_query = parent.clone();
            let docs = backend
                .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?;
            Ok(docs
                .into_iter()
                .map(|doc| {
                    let id = doc.name.split('/').last().unwrap_or("").to_string();
                    (id, Self::shared_device_record_from_fields(&doc.fields))
                })
                .collect())
        })
        .await
    }

    async fn durable_roster_records_with<F, Fut>(
        &self,
        user_id: &str,
        fetch: F,
    ) -> Result<(Vec<(String, SharedDeviceRecord)>, bool)>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<Vec<(String, SharedDeviceRecord)>>>,
    {
        let listener_active = self.has_active_device_subscription(user_id).await;
        let now = std::time::Instant::now();
        let cached = self.durable_roster_cache.lock().await.get(user_id).cloned();
        if let Some(cached) = cached.filter(|cached| cached.is_usable(listener_active, now)) {
            // The Firestore listener is the invalidation authority while a
            // device subscription is active. Before that listener is ready, a
            // short TTL coalesces concurrent startup reads without turning the
            // cache into a second durable-roster authority. RTDB still carries
            // hot liveness in both cases.
            return Ok((cached.records, true));
        }

        let _refresh_guard = self.durable_roster_refresh_lock.lock().await;
        let listener_active = self.has_active_device_subscription(user_id).await;
        let now = std::time::Instant::now();
        let refreshed = self.durable_roster_cache.lock().await.get(user_id).cloned();
        if let Some(cached) = refreshed.filter(|cached| cached.is_usable(listener_active, now)) {
            return Ok((cached.records, true));
        }

        let records = fetch().await?;
        self.durable_roster_cache.lock().await.insert(
            user_id.to_string(),
            CachedDurableRoster {
                records: records.clone(),
                loaded_at: std::time::Instant::now(),
            },
        );
        Ok((records, false))
    }

    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 = match tokio::time::timeout(NATIVE_FIRESTORE_OPERATION_TIMEOUT, self.get_db()).await
        {
            Ok(result) => result?,
            Err(_) => {
                let message = format!(
                    "[OPENRTC][SIGNALING][native] Firestore get_db timed out project_id={} app_tag={} timeout_ms={}",
                    self.project_id,
                    self.namespace.app_tag(),
                    NATIVE_FIRESTORE_OPERATION_TIMEOUT.as_millis(),
                );
                log_warn(&message);
                anyhow::bail!(message);
            }
        };

        match tokio::time::timeout(NATIVE_FIRESTORE_OPERATION_TIMEOUT, operation(db)).await {
            Ok(Ok(value)) => Ok(value),
            Ok(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 = match tokio::time::timeout(
                    NATIVE_FIRESTORE_OPERATION_TIMEOUT,
                    self.get_db(),
                )
                .await
                {
                    Ok(result) => result?,
                    Err(_) => {
                        let message = format!(
                                "[OPENRTC][SIGNALING][native] Firestore auth-retry get_db timed out project_id={} app_tag={} timeout_ms={}",
                                self.project_id,
                                self.namespace.app_tag(),
                                NATIVE_FIRESTORE_OPERATION_TIMEOUT.as_millis(),
                            );
                        log_warn(&message);
                        anyhow::bail!(message);
                    }
                };
                match tokio::time::timeout(
                    NATIVE_FIRESTORE_OPERATION_TIMEOUT,
                    operation(refreshed_db),
                )
                .await
                {
                    Ok(result) => result,
                    Err(_) => {
                        let message = format!(
                            "[OPENRTC][SIGNALING][native] Firestore auth-retry operation timed out project_id={} app_tag={} timeout_ms={}",
                            self.project_id,
                            self.namespace.app_tag(),
                            NATIVE_FIRESTORE_OPERATION_TIMEOUT.as_millis(),
                        );
                        log_warn(&message);
                        anyhow::bail!(message);
                    }
                }
            }
            Ok(Err(error)) => Err(error),
            Err(_) => {
                let message = format!(
                    "[OPENRTC][SIGNALING][native] Firestore operation timed out project_id={} app_tag={} timeout_ms={}",
                    self.project_id,
                    self.namespace.app_tag(),
                    NATIVE_FIRESTORE_OPERATION_TIMEOUT.as_millis(),
                );
                log_warn(&message);
                anyhow::bail!(message);
            }
        }
    }

    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(|| {
                    // Fallback keeps behavior safe even if exp can't be decoded.
                    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();

        // Try to return a cached authenticated DB first
        {
            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;
        // Check again after taking write lock
        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 {
            // DO NOT cache unauthenticated DB
            *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;
        }

        // Retention is avenue-aware: spaces drop heartbeat-stale records, but
        // user-scope devices stay in the roster with degraded presence until
        // the offline-retention window lapses. Only a record that leaves
        // discovery entirely is surfaced as `Removed`.
        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 {
        // Spaces are live-only rosters; the user-scope device roster is
        // persistent and surfaces stale devices with degraded presence so the
        // app layer can decide how to filter them.
        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 {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub app_tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub user_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub device_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub device_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub node_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub platform_type: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub capabilities: Option<crate::signaling::DeviceCapabilities>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub online: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ticket: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_seen_at: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub updated_at: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub created_at: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    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 {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub user_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub device_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub capabilities: Option<crate::signaling::DeviceCapabilities>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    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;

        // Strip compound ticket suffix for validation — the full compound
        // ticket (with token) is stored in the Firestore document.
        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 local_node_id_for_lookup = local_node_id.to_string();
        let (
            target_doc_id,
            existing_device_name,
            existing_created_at,
            should_set_created_at,
            duplicate_device_ids,
        ) = self
            .with_auth_retry(move |db| {
                let parent = parent_for_lookup.clone();
                let canonical_device_id = canonical_device_id_for_lookup.clone();
                let local_node_id = local_node_id_for_lookup.clone();
                async move {
                    let docs = db
                        .fluent()
                        .select()
                        .from("devices")
                        .parent(&parent)
                        .query()
                        .await?;

                    let mut exact_name = None;
                    let mut exact_created_at = None;
                    let mut inherited_name = None;
                    let mut inherited_created_at = None;
                    let mut exact_exists = false;
                    let mut duplicate_device_ids = Vec::new();
                    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");

                        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");
                        let is_exact = doc_id == canonical_device_id;
                        if is_exact {
                            exact_exists = true;
                            exact_name = existing_name;
                            exact_created_at = existing_created_at;
                            continue;
                        }

                        let is_same_native_node =
                            existing_node_id.as_deref() == Some(local_node_id.as_str());
                        let is_legacy_alias =
                            existing_device_id.as_deref() == Some(canonical_device_id.as_str());
                        if is_same_native_node || is_legacy_alias {
                            inherited_name = inherited_name.or(existing_name);
                            inherited_created_at = inherited_created_at.or(existing_created_at);
                            duplicate_device_ids.push(doc_id);
                        }
                    }

                    Ok((
                        canonical_device_id,
                        exact_name.or(inherited_name),
                        exact_created_at.or(inherited_created_at),
                        !exact_exists,
                        duplicate_device_ids,
                    ))
                }
            })
            .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",
        ];
        if metadata.is_some() {
            update_fields.push("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 update_fields_for_log = update_fields.join(",");
        let app_tag_for_log = self.namespace.app_tag().to_string();
        let user_id_for_log = user_id.to_string();
        let device_id_for_log = canonical_device_id.clone();
        let node_id_for_log = local_node_id.to_string();
        let online_for_log = is_online;
        let ticket_len_for_log = fields.ticket.as_deref().map(str::len).unwrap_or(0);
        let metadata_len_for_log = fields.metadata.as_deref().map(str::len).unwrap_or(0);
        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();
            let update_fields_for_log = update_fields_for_log.clone();
            let app_tag_for_log = app_tag_for_log.clone();
            let user_id_for_log = user_id_for_log.clone();
            let device_id_for_log = device_id_for_log.clone();
            let node_id_for_log = node_id_for_log.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(());
                    }

                    let final_update_result = db
                        .fluent()
                        .update()
                        .fields(update_fields)
                        .in_col("devices")
                        .document_id(&target_doc_id)
                        .parent(&parent)
                        .object(&fields)
                        .execute::<()>()
                        .await;
                    if let Err(error) = final_update_result {
                        log_warn(&format!(
                            "[OPENRTC][SIGNALING][native] update_presence write failed user_id={} app_tag={} device_id={} doc_id={} node_id={} online={} ticket_len={} metadata_len={} fields={} namespace=space error={}",
                            user_id_for_log,
                            app_tag_for_log,
                            device_id_for_log,
                            target_doc_id,
                            node_id_for_log,
                            online_for_log,
                            ticket_len_for_log,
                            metadata_len_for_log,
                            update_fields_for_log,
                            error
                        ));
                        return Err(error.into());
                    }
                } else {
                    if should_set_created_at {
                        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(());
                        }
                    }

                    let update_result = db
                            .fluent()
                            .update()
                            .fields(update_fields)
                            .in_col("devices")
                            .document_id(&target_doc_id)
                            .parent(&parent)
                            .object(&fields)
                            .execute::<()>()
                            .await;
                    if update_result.is_ok() {
                        return Ok(());
                    }

                    let update_error = update_result.err();
                    if !should_set_created_at {
                        let _ = db
                            .fluent()
                            .delete()
                            .from("devices")
                            .parent(&parent)
                            .document_id(&target_doc_id)
                            .execute()
                            .await;
                        let replace_result = db
                            .fluent()
                            .insert()
                            .into("devices")
                            .document_id(&target_doc_id)
                            .parent(&parent)
                            .object(&fields)
                            .execute::<DeviceFields>()
                            .await;
                        if replace_result.is_ok() {
                            return Ok(());
                        }
                    }

                    let error_text = update_error
                        .map(|error| error.to_string())
                        .unwrap_or_else(|| "unknown write failure".to_string());
                    log_warn(&format!(
                        "[OPENRTC][SIGNALING][native] update_presence write failed user_id={} app_tag={} device_id={} doc_id={} node_id={} online={} ticket_len={} metadata_len={} fields={} namespace=app-user create_attempted={} error={}",
                        user_id_for_log,
                        app_tag_for_log,
                        device_id_for_log,
                        target_doc_id,
                        node_id_for_log,
                        online_for_log,
                        ticket_len_for_log,
                        metadata_len_for_log,
                        update_fields_for_log,
                        should_set_created_at,
                        error_text
                    ));
                    anyhow::bail!(error_text);
                }
                Ok(())
            }
        })
        .await?;

        // A previous runtime could have published a frontend alias as a second
        // durable device for this same native node. Once the canonical write is
        // committed, remove only those same-node aliases and expire their hot
        // presence. Unrelated devices are never touched.
        for duplicate_device_id in duplicate_device_ids {
            let parent_for_delete = parent.clone();
            let duplicate_for_delete = duplicate_device_id.clone();
            self.with_auth_retry(move |db| {
                let parent = parent_for_delete.clone();
                let duplicate = duplicate_for_delete.clone();
                async move {
                    db.fluent()
                        .delete()
                        .from("devices")
                        .parent(&parent)
                        .document_id(&duplicate)
                        .execute()
                        .await?;
                    Ok(())
                }
            })
            .await?;
            self.rtdb
                .set_device_offline(user_id, &duplicate_device_id, Some(local_node_id))
                .await?;
            log_info(&format!(
                "[OPENRTC][SIGNALING][native] reconciled duplicate same-node device alias user_id={} canonical_device_id={} duplicate_device_id={} node_id={}",
                user_id, canonical_device_id, duplicate_device_id, local_node_id
            ));
        }

        self.invalidate_durable_roster(user_id).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?;
        self.invalidate_durable_roster(user_id).await;
        Ok(())
    }

    async fn update_live_presence(
        &self,
        user_id: &str,
        local_node_id: &str,
        ticket_str: &str,
        name: &str,
        metadata: Option<&str>,
    ) -> Result<()> {
        if self.is_app_backgrounded() {
            return Ok(());
        }

        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 RTDB live presence: {}", e)
        })?;

        let canonical_device_id = Self::extract_device_id(metadata, local_node_id);
        self.rtdb
            .publish(
                user_id,
                local_node_id,
                &canonical_device_id,
                ticket_str,
                name,
                metadata,
            )
            .await?;

        if signaling_verbose() {
            log_info(&format!(
                "[OPENRTC][SIGNALING][native][rtdb] update_live_presence success user_id={} app_tag={} device_id={} node_id={} rtdb_target={} ticket_len={} metadata_len={}",
                user_id,
                self.namespace.app_tag(),
                canonical_device_id,
                local_node_id,
                self.rtdb.target_label(),
                ticket_str.len(),
                metadata.map(str::len).unwrap_or(0),
            ));
        }

        Ok(())
    }

    async fn set_live_presence_offline(&self, user_id: &str, local_node_id: &str) -> Result<()> {
        self.rtdb.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<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?;

        self.invalidate_durable_roster(user_id).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?;

        self.invalidate_durable_roster(user_id).await;

        Ok(())
    }

    async fn set_excluded_peers(
        &self,
        user_id: &str,
        local_node_id: &str,
        excluded_peers: &[String],
    ) -> Result<()> {
        self.rtdb
            .set_excluded_peers(user_id, local_node_id, excluded_peers)
            .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);

        let (durable_records, durable_roster_cache_hit) =
            self.durable_roster_records(user_id, &parent).await?;

        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_millis() as i64;
        let mut filtered: Vec<Device> = durable_records
            .into_iter()
            .filter_map(|(id, shared)| {
                if shared.effective_app_tag() != Some(self.registry_app_tag()) {
                    return None;
                }

                // Avenue-aware retention lives in project_discovered_device:
                // spaces stay live-only, user-scope keeps stale devices with
                // degraded presence for the app layer to filter.
                self.project_discovered_device(id, shared, exclude_node_id, now_ms)
            })
            .collect();
        let firestore_count = filtered.len();
        let firestore_online_count = filtered.iter().filter(|device| device.online).count();
        let mut rtdb_presence_count: Option<usize> = None;
        let mut rtdb_overlay_stats = super::rtdb_presence::RtdbOverlayStats::default();

        // Browser/WebView clients keep the durable Firestore roster but publish
        // active liveness in RTDB. Native must merge the same RTDB overlay before
        // its online-only search path runs; otherwise active browser devices look
        // like retained offline records and native never dials them.
        match self.rtdb.read(user_id).await {
            Ok(presence) => {
                rtdb_presence_count = Some(presence.len());
                rtdb_overlay_stats = self.rtdb.overlay_devices(
                    &mut filtered,
                    &presence,
                    user_id,
                    exclude_node_id,
                    now_ms,
                );
            }
            Err(error) => {
                log_warn(&format!(
                    "[OPENRTC][SIGNALING][native][rtdb] presence overlay unavailable user_id={} app_tag={} error={}",
                    user_id,
                    self.namespace.app_tag(),
                    error
                ));
            }
        }

        if signaling_verbose() {
            let final_online_count = filtered.iter().filter(|device| device.online).count();
            log_info(&format!(
                "[OPENRTC][SIGNALING][native] list_devices user_id={} tag={} exclude_node_id={} count={} online={} firestore_count={} firestore_online={} durable_roster_cache_hit={} rtdb_target={} rtdb_count={} rtdb_matched={} rtdb_promoted={} rtdb_demoted={} rtdb_only_added={} elapsed_ms={}",
                user_id,
                self.namespace.app_tag(),
                exclude_node_id.unwrap_or("<none>"),
                filtered.len(),
                final_online_count,
                firestore_count,
                firestore_online_count,
                durable_roster_cache_hit,
                self.rtdb.target_label(),
                rtdb_presence_count
                    .map(|value| value.to_string())
                    .unwrap_or_else(|| "skipped".to_string()),
                rtdb_overlay_stats.matched,
                rtdb_overlay_stats.promoted_online,
                rtdb_overlay_stats.demoted_offline,
                rtdb_overlay_stats.rtdb_only_added,
                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 projection_user_id = user_id.to_string();

        listener
            .start(move |event| {
                let tx = tx_clone.clone();
                let projection_backend = projection_backend.clone();
                let projection_user_id = projection_user_id.clone();
                async move {
                    use gcloud_sdk::google::firestore::v1::listen_response::ResponseType;
                    match event {
                        ResponseType::DocumentChange(doc_change) => {
                            projection_backend
                                .invalidate_durable_roster(&projection_user_id)
                                .await;
                            if let Some(doc) = doc_change.document {
                                // Native watch events must use the same
                                // discovery projection as snapshot reads. A raw
                                // Firestore document can still say `online:
                                // true` after its TTL; forwarding it directly
                                // would let auto-connect dial stale peers that
                                // `list_devices()` correctly filters out.
                                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) => {
                            projection_backend
                                .invalidate_durable_roster(&projection_user_id)
                                .await;
                            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) => {
                            projection_backend
                                .invalidate_durable_roster(&projection_user_id)
                                .await;
                            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?;

        self.retain_device_subscription(user_id).await;
        let subscription_backend = self.clone();
        let subscription_user_id = user_id.to_string();
        tokio::spawn(async move {
            tx.closed().await;
            let mut l = listener;
            let _ = l.shutdown().await;
            subscription_backend
                .release_device_subscription(&subscription_user_id)
                .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 crate::firebase::device_registry::DEVICE_STALE_HEARTBEAT_MS;
    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() {
        // User-scope devices are persistent: a heartbeat-stale record stays in
        // the roster (presence degrades to idle/offline downstream) instead of
        // being surfaced as `Removed`. The app layer filters live vs stale.
        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());
    }

    #[tokio::test]
    async fn durable_roster_cache_invalidation_is_user_scoped_and_auth_safe() {
        let backend = backend();
        let record = SharedDeviceRecord {
            canonical_device_id: Some("device-1".to_string()),
            app_tag: Some("app_test".to_string()),
            user_id: Some("user-1".to_string()),
            device_name: "Device 1".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-1".to_string()),
            tag: Some("app_test".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket-1".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        };
        {
            let mut cache = backend.durable_roster_cache.lock().await;
            for user_id in ["user-1", "user-2"] {
                cache.insert(
                    user_id.to_string(),
                    CachedDurableRoster {
                        records: vec![("device-1".to_string(), record.clone())],
                        loaded_at: std::time::Instant::now(),
                    },
                );
            }
        }

        backend.invalidate_durable_roster("user-1").await;
        {
            let cache = backend.durable_roster_cache.lock().await;
            assert!(!cache.contains_key("user-1"));
            assert!(cache.contains_key("user-2"));
        }

        backend.invalidate_db_cache().await;
        assert!(backend.durable_roster_cache.lock().await.is_empty());

        backend.retain_device_subscription("user-1").await;
        backend.retain_device_subscription("user-1").await;
        assert!(backend.has_active_device_subscription("user-1").await);
        backend.release_device_subscription("user-1").await;
        assert!(backend.has_active_device_subscription("user-1").await);
        backend.release_device_subscription("user-1").await;
        assert!(!backend.has_active_device_subscription("user-1").await);
    }

    #[test]
    fn durable_roster_burst_cache_is_bounded_without_listener_authority() {
        let now = std::time::Instant::now();
        let fresh = CachedDurableRoster {
            records: vec![],
            loaded_at: now,
        };
        let expired = CachedDurableRoster {
            records: vec![],
            loaded_at: now - DURABLE_ROSTER_BURST_CACHE_TTL - std::time::Duration::from_millis(1),
        };

        assert!(fresh.is_usable(false, now));
        assert!(!expired.is_usable(false, now));
        assert!(expired.is_usable(true, now));
    }

    #[tokio::test]
    async fn native_discovery_operation_budget_is_one_startup_read_and_zero_idle_writes() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let backend = backend();
        let firestore_reads = std::sync::Arc::new(AtomicUsize::new(0));
        let firestore_writes = std::sync::Arc::new(AtomicUsize::new(0));

        let startup_reads = (0..32).map(|_| {
            let reads = firestore_reads.clone();
            backend.durable_roster_records_with("user-1", move || async move {
                reads.fetch_add(1, Ordering::SeqCst);
                tokio::task::yield_now().await;
                Ok(Vec::new())
            })
        });
        let startup_results = futures::future::join_all(startup_reads).await;
        assert!(startup_results.iter().all(Result::is_ok));
        assert_eq!(firestore_reads.load(Ordering::SeqCst), 1);
        assert_eq!(firestore_writes.load(Ordering::SeqCst), 0);

        // Once the durable listener owns invalidation, repeated auto-connect
        // snapshots and RTDB presence wakeups must reuse the same roster.
        backend.retain_device_subscription("user-1").await;
        for _ in 0..120 {
            let reads = firestore_reads.clone();
            let (_, cache_hit) = backend
                .durable_roster_records_with("user-1", move || async move {
                    reads.fetch_add(1, Ordering::SeqCst);
                    Ok(Vec::new())
                })
                .await
                .expect("idle roster read");
            assert!(cache_hit);
        }
        assert_eq!(firestore_reads.load(Ordering::SeqCst), 1);
        assert_eq!(firestore_writes.load(Ordering::SeqCst), 0);

        // A durable revision invalidation permits exactly one new coalesced
        // roster read. RTDB liveness updates never call this invalidator.
        backend.invalidate_durable_roster("user-1").await;
        let revision_reads = (0..16).map(|_| {
            let reads = firestore_reads.clone();
            backend.durable_roster_records_with("user-1", move || async move {
                reads.fetch_add(1, Ordering::SeqCst);
                tokio::task::yield_now().await;
                Ok(Vec::new())
            })
        });
        let revision_results = futures::future::join_all(revision_reads).await;
        assert!(revision_results.iter().all(Result::is_ok));
        assert_eq!(firestore_reads.load(Ordering::SeqCst), 2);
        assert_eq!(firestore_writes.load(Ordering::SeqCst), 0);
    }
}