bevy_symbios_multiuser 0.7.0

//! Custom matchbox signaller that speaks the Symbios relay's wire format and
//! optionally authenticates with an ATProto JWT.
//!
//! Implements the [`SignallerBuilder`] and [`Signaller`] traits from
//! `matchbox_socket`, bridging between the matchbox `PeerRequest`/`PeerEvent`
//! protocol and the relay's [`SignalEnvelope`]/[`SignalPayload`] wire format.
//!
//! The plugin **always** uses this signaller (via
//! [`signaller_with_token_source`] or [`signaller_anonymous`]) so that the
//! relay receives the expected `SignalEnvelope` JSON, rather than matchbox's
//! incompatible default format.
//! When a [`TokenSourceRes`] resource is present, the signaller uses the shared
//! token source for automatic refresh on reconnect. If no token source is
//! available, the signaller connects without authentication (anonymous mode).
//!
//! # Platform Support
//!
//! On native targets, the WebSocket connection uses `async-tungstenite` with
//! the JWT passed as an `Authorization: Bearer <token>` header during the
//! upgrade handshake.
//!
//! On WASM targets, the browser's `WebSocket` API (via `ws_stream_wasm`) is
//! used instead. Because the browser `WebSocket` constructor does not support
//! custom headers, the JWT is passed via the `Sec-WebSocket-Protocol`
//! subprotocol trick: the client sends `["access_token", "<jwt>"]` as the
//! requested subprotocols during the handshake. The relay extracts the token
//! from the second element.
//!
//! # Token Refresh
//!
//! Service auth tokens are short-lived. Use [`signaller_with_token_source`]:
//! it accepts a shared [`TokenSource`] that the application can update
//! externally (e.g. after calling [`crate::auth::get_service_auth`] with a
//! refreshed session), ensuring the signaller always uses the latest token
//! on reconnect.
//!
//! Note that the OAuth access token is DPoP-bound and **cannot** be verified
//! by a third-party relay that resolves DID documents. Always wrap a service
//! auth token from [`crate::auth::get_service_auth`] in the [`TokenSource`];
//! that token is signed by the user's `#atproto` key and verifies against
//! their published DID document.

use crate::protocol::{SignalEnvelope, SignalPayload};
#[allow(unused_imports)] // SinkExt is used by .send() inside async_trait impls
use futures_util::SinkExt;
use futures_util::StreamExt;
use matchbox_socket::async_trait::async_trait;
use matchbox_socket::{
    PeerEvent, PeerId, PeerRequest, PeerSignal, SignalingError, Signaller, SignallerBuilder,
};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use uuid::Uuid;

// ── Native-only imports ──────────────────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
use async_tungstenite::WebSocketStream;
#[cfg(not(target_arch = "wasm32"))]
use async_tungstenite::async_std::{ConnectStream, connect_async};
#[cfg(not(target_arch = "wasm32"))]
use async_tungstenite::tungstenite;

// ── WASM-only imports ────────────────────────────────────────────────────────

#[cfg(target_arch = "wasm32")]
use ws_stream_wasm::{WsMessage as WasmWsMessage, WsMeta, WsStream};

// ── Shared constants & helpers ───────────────────────────────────────────────

/// Namespace UUID for deterministic `PeerId` generation from DID strings
/// (`Uuid::NAMESPACE_X500` — `6ba7b814-9dad-11d1-80b4-00c04fd430c8`).
/// Using a fixed, well-known namespace ensures the same DID always maps to
/// the same `PeerId` regardless of when or where it is computed.
const DID_NAMESPACE: Uuid = Uuid::from_bytes([
    0x6b, 0xa7, 0xb8, 0x14, 0x9d, 0xad, 0x11, 0xd1, 0x80, 0xb4, 0x00, 0xc0, 0x4f, 0xd4, 0x30, 0xc8,
]);

/// Convert a session ID string to a [`PeerId`].
///
/// If the string is a valid UUID it is used directly. Otherwise (e.g. a DID),
/// a deterministic UUID v5 is derived so that the same session ID always
/// produces the same `PeerId`.
fn session_id_to_peer_id(session_id: &str) -> PeerId {
    match Uuid::parse_str(session_id) {
        Ok(uuid) => PeerId(uuid),
        Err(_) => PeerId(Uuid::new_v5(&DID_NAMESPACE, session_id.as_bytes())),
    }
}

/// Core of the WebRTC Offerer/Answerer arbitration, factored out of
/// [`SymbiosSignaller::should_offer`] as a free function so the rule is
/// unit-testable without a live signaller or socket.
///
/// Returns `true` iff the local peer — identified by `local_peer_id`, which is
/// the deterministic [`session_id_to_peer_id`] of the local session — is the
/// designated Offerer for the pair with `remote_session_id`. The order is a
/// strict total order over the two peers' deterministic session `PeerId`s, so
/// for any two distinct sessions exactly one side offers and the other answers,
/// independent of join order or timing. Both peers compute the same order from
/// the same two session ids, which is what prevents simultaneous-join glare.
fn local_is_offerer(local_peer_id: PeerId, remote_session_id: &str) -> bool {
    local_peer_id > session_id_to_peer_id(remote_session_id)
}

// ── Peer session map (PeerId → authenticated session_id / DID) ───────────────

/// Shared map from [`PeerId`] to the relay session identifier (typically the
/// authenticated DID) that the relay assigned to that peer.
///
/// The [`SymbiosSignaller`] mints a fresh random [`PeerId`] for every remote
/// peer it sees so that matchbox's per-peer WebRTC state machine cannot confuse
/// a reconnecting peer with stale state from the previous connection. That
/// choice breaks any attempt by the application to recover the real DID of a
/// peer directly from its `PeerId`, which in turn blocks DID-based identity
/// verification over the (unauthenticated) WebRTC data channel.
///
/// This map restores that binding: the signaller writes `(PeerId, session_id)`
/// pairs as peers join and removes them as peers leave, and the application
/// reads from it through [`PeerSessionMapRes`] to verify that a self-reported
/// identity payload really belongs to the peer the relay authenticated.
pub type PeerSessionMap = Arc<std::sync::RwLock<HashMap<PeerId, String>>>;

/// Bevy [`Resource`](bevy::prelude::Resource) wrapper around a
/// [`PeerSessionMap`].
///
/// The [`crate::plugin::SymbiosMultiuserPlugin`] inserts this resource
/// automatically and passes the inner [`PeerSessionMap`] to the signaller
/// builder so that the application and the signaller observe the same
/// `PeerId → session_id` view.
///
/// # Example — verifying a peer's DID claim
///
/// ```rust,ignore
/// use bevy::prelude::*;
/// use bevy_symbios_multiuser::prelude::*;
///
/// fn verify_identity(
///     map: Res<PeerSessionMapRes>,
///     peer_id: PeerId,
///     claimed_did: &str,
/// ) -> bool {
///     map.session_id(&peer_id).as_deref() == Some(claimed_did)
/// }
/// ```
#[derive(bevy::prelude::Resource, Clone)]
pub struct PeerSessionMapRes(pub PeerSessionMap);

impl Default for PeerSessionMapRes {
    fn default() -> Self {
        Self(Arc::new(std::sync::RwLock::new(HashMap::new())))
    }
}

impl PeerSessionMapRes {
    /// Look up the session ID (typically the authenticated DID) that the relay
    /// assigned to `peer_id`, if the signaller currently knows that peer.
    ///
    /// Returns `None` for unknown peers, and for the brief window between
    /// matchbox surfacing a [`PeerState::Connected`] event and the signaller
    /// recording the underlying session ID. Callers that need a strict check
    /// should treat `None` as "not yet verified" rather than "verified absent".
    pub fn session_id(&self, peer_id: &matchbox_socket::PeerId) -> Option<String> {
        self.0
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(peer_id)
            .cloned()
    }
}

// ── SignallerBuilder ─────────────────────────────────────────────────────────

/// A shared, externally-refreshable token source.
///
/// The host application updates this (e.g. after an ATProto token refresh)
/// and the signaller reads the latest value on each reconnect attempt,
/// avoiding stale-JWT failures when short-lived access tokens expire
/// between the initial connection and a later reconnect.
///
/// # Safe-by-construction
///
/// The inner `RwLock` is **not** exposed. Callers can only observe the current
/// token via [`get`](Self::get) (which clones the inner `Option<String>` and
/// releases the guard) and replace it via [`set`](Self::set) (which acquires
/// the write guard only long enough to swap the value in place).
///
/// This structurally prevents the deadlock hazard of holding a lock guard
/// across an `.await`: since the guard never leaves the method body, a caller
/// refreshing the token over an async network request cannot accidentally
/// starve the signaller's reconnect path.
///
/// ```rust,ignore
/// let new_token = get_service_auth(...).await?;   // network call
/// token_source.set(Some(new_token));              // lock held only for swap
/// ```
#[derive(Clone, Default)]
pub struct TokenSource {
    inner: Arc<std::sync::RwLock<Option<String>>>,
}

impl TokenSource {
    /// Create a new token source with an optional initial value.
    pub fn new(initial: Option<String>) -> Self {
        Self {
            inner: Arc::new(std::sync::RwLock::new(initial)),
        }
    }

    /// Return a clone of the currently stored token, if any.
    ///
    /// The lock is released before this function returns, so the caller is
    /// free to `.await` on the result without blocking the signaller.
    pub fn get(&self) -> Option<String> {
        self.inner.read().unwrap_or_else(|e| e.into_inner()).clone()
    }

    /// Atomically replace the stored token.
    ///
    /// The write guard is acquired only for the duration of the swap; passing
    /// `None` clears the token (useful on logout).
    pub fn set(&self, token: Option<String>) {
        *self.inner.write().unwrap_or_else(|e| e.into_inner()) = token;
    }
}

impl std::fmt::Debug for TokenSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Deliberately do not render the token bytes — JWTs are sensitive and
        // often end up in tracing output by accident.
        f.debug_struct("TokenSource")
            .field("set", &self.get().is_some())
            .finish()
    }
}

/// Bevy [`Resource`](bevy::prelude::Resource) wrapper around a [`TokenSource`].
///
/// Insert this resource into the Bevy world to enable automatic token refresh
/// when using [`SymbiosMultiuserPlugin`](crate::plugin::SymbiosMultiuserPlugin).
/// The plugin reads the current token from this resource on every (re)connect
/// attempt, so reconnects after token expiry use the latest refreshed JWT.
///
/// # Example
///
/// ```rust,ignore
/// use bevy_symbios_multiuser::signaller::{TokenSource, TokenSourceRes};
///
/// // Use the service auth token from `get_service_auth`, not `access_jwt`.
/// // Relay servers verify service auth tokens via DID document resolution;
/// // `access_jwt` is signed by the PDS service key and cannot be verified
/// // by a third-party relay.
/// let source = TokenSource::new(Some(service_token));
/// app.insert_resource(TokenSourceRes(source.clone()));
///
/// // Later, after refreshing the session and calling get_service_auth again:
/// source.set(Some(new_service_token));
/// ```
#[derive(bevy::prelude::Resource, Clone, Debug, Default)]
pub struct TokenSourceRes(pub TokenSource);

// ── Signalling diagnostics (observability into the WebRTC handshake) ──────────

/// Shared, lock-free counters the signaller bumps as it drives the WebRTC
/// handshake, exposing the signalling layer that matchbox otherwise hides
/// behind a bare `Connected`/`Disconnected` peer state.
///
/// The application reads these through [`SignalDiagnosticsRes`]. The key
/// diagnostic is `last_peer_list_len >= 1` while **no** peer ever reaches
/// `Connected`: that is the fingerprint of a stalled handshake — the relay
/// reported a non-empty room yet no data channel formed — which a bare
/// connected-peer count cannot distinguish from genuinely being alone. A
/// classic offer glare additionally shows `offers_sent > 0` and
/// `offers_received > 0` with `answers_received == 0`.
///
/// All counters are cumulative for the lifetime of the shared handle (they
/// persist across reconnects because the plugin threads the same `Arc` into
/// every signaller it builds).
#[derive(Debug, Default)]
pub struct SignalDiagnostics {
    /// Number of `peer_list` welcome messages processed — one per successful
    /// (re)connect handshake.
    pub peer_lists_received: AtomicU64,
    /// Size of the most recent `peer_list`: how many peers were already in the
    /// room when we last joined.
    pub last_peer_list_len: AtomicU64,
    /// `NewPeer` events emitted to matchbox — the pairs this peer is the
    /// Offerer for (per the deterministic tie-break in [`local_is_offerer`]).
    pub offers_initiated: AtomicU64,
    /// SDP offers this peer sent over the signalling channel.
    pub offers_sent: AtomicU64,
    /// SDP offers this peer received (it answers these).
    pub offers_received: AtomicU64,
    /// SDP answers this peer sent, in response to a received offer.
    pub answers_sent: AtomicU64,
    /// SDP answers this peer received, completing an offer it initiated.
    pub answers_received: AtomicU64,
    /// Relay handshake rejections the signaller gave up on: an HTTP 4xx on the
    /// WebSocket upgrade (native — chiefly `401` from an expired/invalid
    /// service-auth token, but also `403`/`429`), or a wasm blind-retry
    /// exhaustion (the browser hides handshake status codes). Unlike the other
    /// counters this records a *failure to connect at all* — the socket never
    /// opens, so no `peer_list`/offer/answer follows. The dominant cause is a
    /// stale auth token; keep it fresh (re-issue via `get_service_auth` and
    /// update the [`TokenSource`]) to keep this at zero.
    pub auth_rejections: AtomicU64,
    /// HTTP status of the most recent relay rejection (e.g. `401`). `0` when
    /// unknown — including every wasm rejection, where the browser `WebSocket`
    /// API does not expose the handshake status code.
    pub last_reject_status: AtomicU64,
}

/// Cheap-to-clone shared handle to [`SignalDiagnostics`].
pub type SharedSignalDiagnostics = Arc<SignalDiagnostics>;

/// Bevy [`Resource`](bevy::prelude::Resource) wrapper around
/// [`SharedSignalDiagnostics`].
///
/// [`crate::plugin::SymbiosMultiuserPlugin`] inserts this automatically and
/// threads the inner handle into every signaller it builds, so the counters
/// survive reconnects and the application and signaller observe the same view.
#[derive(bevy::prelude::Resource, Clone, Default)]
pub struct SignalDiagnosticsRes(pub SharedSignalDiagnostics);

/// A [`SignallerBuilder`] that injects a service auth JWT into the WebSocket
/// upgrade request's `Authorization` header.
///
/// Supports two token modes:
/// - **Refreshable**: a shared [`TokenSource`] that the application can update
///   externally (via [`signaller_with_token_source`]). On each reconnect the
///   builder reads the latest token, avoiding stale-JWT rejections.
/// - **Anonymous**: no JWT at all (via [`signaller_anonymous`]). Still speaks
///   the relay's `SignalEnvelope` wire format; connects without authentication.
#[derive(Debug, Clone)]
pub struct SymbiosSignallerBuilder {
    /// Shared, externally-refreshable token.
    token_source: Option<TokenSource>,
    /// Shared `PeerId → session_id` map updated by the signaller as peers
    /// join and leave. When present, the application can resolve the real
    /// (relay-authenticated) DID for any active peer.
    session_map: Option<PeerSessionMap>,
    /// Persistent WASM blind-retry state. Shared across every builder the
    /// plugin constructs so the failure count is not re-zeroed on ECS-level
    /// socket respawn. `None` falls back to per-builder local state, which
    /// is sufficient for one-shot programmatic use but not for the plugin's
    /// respawn loop. Native builds carry the field (gated with
    /// `#[allow(dead_code)]`) so the struct layout and public constructors
    /// stay uniform across targets — the field is simply never read.
    #[cfg(target_arch = "wasm32")]
    wasm_retry_state: Option<WasmBlindRetryStateHandle>,
    /// Shared signalling counters bumped by every signaller this builder
    /// produces. Threaded through so the counts survive reconnects. `None`
    /// disables diagnostics (the default for one-shot programmatic use).
    diagnostics: Option<SharedSignalDiagnostics>,
}

impl SymbiosSignallerBuilder {
    /// Return the current JWT from the refreshable [`TokenSource`], if any.
    fn current_token(&self) -> Option<String> {
        self.token_source.as_ref().and_then(|s| s.get())
    }

    /// Record a relay handshake rejection into the shared [`SignalDiagnostics`]
    /// (if one was threaded through): bump the rejection counter and store the
    /// HTTP status (`0` = unknown, e.g. every wasm rejection). Called from the
    /// two give-up points in [`new_signaller`](Self::new_signaller) so a stale
    /// token surfaces to the host application instead of only the log.
    fn record_reject(&self, status: u64) {
        if let Some(d) = &self.diagnostics {
            d.auth_rejections.fetch_add(1, Ordering::Relaxed);
            d.last_reject_status.store(status, Ordering::Relaxed);
        }
    }

    /// Stable, non-cryptographic 64-bit fingerprint of the current token used
    /// only by the WASM retry loop to detect "same token failed again". A hash
    /// is used so the token bytes themselves never sit in retained memory
    /// across reconnect attempts; the absence of a token hashes to a fixed
    /// value distinct from any real token.
    #[cfg(target_arch = "wasm32")]
    fn current_token_fingerprint(&self) -> u64 {
        use std::hash::{Hash, Hasher};
        let mut h = std::collections::hash_map::DefaultHasher::new();
        self.current_token().hash(&mut h);
        h.finish()
    }
}

/// Maximum number of consecutive WASM connection failures with the same auth
/// token before [`new_signaller`](SymbiosSignallerBuilder::new_signaller)
/// gives up.
///
/// On native targets, `try_connect` can read the HTTP status code from the
/// failed upgrade and immediately bail on 4xx (see [`is_http_client_error`]).
/// The browser `WebSocket` API deliberately hides HTTP status codes from
/// failed handshakes, so on WASM a 401 Unauthorized is indistinguishable from
/// a TCP reset, and an unbounded retry loop with an expired or invalid JWT
/// would burn client battery and relay bandwidth indefinitely.
///
/// As a substitute, the WASM path tracks the fingerprint of the token used
/// for each failed attempt. If the *same* token fails this many times in a
/// row the loop bails out, on the assumption that the relay is rejecting
/// the credential rather than dropping packets. A token rotation between
/// failed attempts (e.g. the host application called `get_service_auth`
/// again and swapped the [`TokenSource`]) resets the counter, so legitimate
/// refresh-and-retry flows are not penalised.
#[cfg(target_arch = "wasm32")]
const WASM_MAX_BLIND_RETRIES: u32 = 5;

/// Persistent state for the WASM blind-retry guard.
///
/// This state must outlive any single [`SymbiosSignallerBuilder`] — the Bevy
/// plugin constructs a fresh builder on every reconnect, so state kept in a
/// local variable inside [`SymbiosSignallerBuilder::new_signaller`] (or in a
/// non-`Arc` field on the builder) would be re-zeroed each time. That would
/// let an expired token retry 5 times, bail with `Err`, be respawned by the
/// ECS after the backoff window, and retry 5 times *again* — indefinitely.
///
/// Sharing an `Arc<Mutex<WasmBlindRetryState>>` across builder instances
/// (via [`WasmBlindRetryStateRes`]) threads the failure count through every
/// reconnect attempt so the guard can bail permanently once the same token
/// has actually exhausted its budget.
#[cfg(target_arch = "wasm32")]
#[derive(Debug, Default)]
pub struct WasmBlindRetryState {
    /// Number of consecutive failed connection attempts made with the same
    /// token fingerprint. Reset to 1 whenever the fingerprint changes
    /// (i.e. the host application refreshed the auth token between attempts).
    failures: u32,
    /// Fingerprint of the token used on the most recent failure, or `None`
    /// if no attempt has failed yet.
    last_failed_token: Option<u64>,
}

#[cfg(target_arch = "wasm32")]
/// Shared handle to the WASM blind-retry state, cheap to clone.
pub type WasmBlindRetryStateHandle = Arc<std::sync::Mutex<WasmBlindRetryState>>;

/// Bevy [`Resource`](bevy::prelude::Resource) that persists the WASM
/// blind-retry counter across reconnect attempts.
///
/// The plugin inserts this resource once at startup and passes the inner
/// handle to every signaller builder it constructs, ensuring the guard's
/// failure count is not reset each time the ECS respawns the socket after a
/// dead-message-loop teardown.
///
/// Only meaningful on `wasm32` targets; the native path fast-fails on HTTP
/// 4xx directly.
#[cfg(target_arch = "wasm32")]
#[derive(bevy::prelude::Resource, Clone, Default)]
pub struct WasmBlindRetryStateRes(pub WasmBlindRetryStateHandle);

#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl SignallerBuilder for SymbiosSignallerBuilder {
    async fn new_signaller(
        &self,
        mut attempts: Option<u16>,
        room_url: String,
    ) -> Result<Box<dyn Signaller>, SignalingError> {
        // WASM-only state for the blind-retry guard. The browser WebSocket
        // API hides HTTP status codes from failed handshakes, so we cannot
        // fast-fail on a 401 the way the native path does. Instead, we
        // count consecutive failures that occurred under the *same* token
        // fingerprint and bail once that count crosses [`WASM_MAX_BLIND_RETRIES`].
        // A token rotation between failures resets the counter so legitimate
        // refresh-and-retry flows are not penalised.
        //
        // The state lives behind `wasm_retry_state` when the plugin threaded
        // one in — that shared handle outlives a single `new_signaller` call,
        // so a doomed token that exhausts its budget stays exhausted across
        // ECS-level respawns. Callers constructing a one-shot builder without
        // a shared handle fall back to a local state that only guards within
        // this single `new_signaller` invocation.
        #[cfg(target_arch = "wasm32")]
        let wasm_state: WasmBlindRetryStateHandle = self
            .wasm_retry_state
            .clone()
            .unwrap_or_else(|| Arc::new(std::sync::Mutex::new(WasmBlindRetryState::default())));

        let signaller = 'connect: loop {
            let ws = match self.try_connect(&room_url).await {
                Ok(stream) => stream,
                Err(e) => {
                    // HTTP 4xx errors are permanent client-side rejections (e.g. 401
                    // Invalid JWT). Retrying won't help — surface them immediately.
                    if is_http_client_error(&e) {
                        self.record_reject(http_client_error_status(&e).unwrap_or(0));
                        return Err(e);
                    }
                    #[cfg(target_arch = "wasm32")]
                    {
                        let token_fp = self.current_token_fingerprint();
                        let failures = {
                            let mut guard = wasm_state.lock().unwrap_or_else(|e| e.into_inner());
                            if guard.last_failed_token == Some(token_fp) {
                                guard.failures = guard.failures.saturating_add(1);
                            } else {
                                guard.failures = 1;
                                guard.last_failed_token = Some(token_fp);
                            }
                            guard.failures
                        };
                        if failures >= WASM_MAX_BLIND_RETRIES {
                            // Status unknown: the browser WebSocket API hides the
                            // handshake HTTP code, so record it as `0`.
                            self.record_reject(0);
                            tracing::error!(
                                attempts = failures,
                                "WASM relay connection failed {WASM_MAX_BLIND_RETRIES} times in \
                                 a row with the same token; aborting (browser WebSocket API \
                                 hides HTTP status codes, so we cannot tell auth failures \
                                 apart from network errors — refresh the auth token to retry)"
                            );
                            return Err(SignalingError::UserImplementationError(
                                "wasm_blind_retry_exhausted: relay rejected the same token \
                                 repeatedly; refresh the auth token before reconnecting"
                                    .to_string(),
                            ));
                        }
                    }
                    if let Some(ref mut remaining) = attempts {
                        if *remaining <= 1 {
                            return Err(SignalingError::NegotiationFailed(Box::new(e)));
                        }
                        *remaining -= 1;
                        tracing::warn!(
                            attempts_remaining = *remaining,
                            "connection to relay failed, retrying in 3s"
                        );
                        futures_timer::Delay::new(Duration::from_secs(3)).await;
                        continue 'connect;
                    }
                    // Unlimited retries
                    tracing::warn!("connection to relay failed, retrying in 3s");
                    futures_timer::Delay::new(Duration::from_secs(3)).await;
                    continue 'connect;
                }
            };

            let mut signaller = SymbiosSignaller {
                ws,
                local_peer_id: PeerId(Uuid::nil()),
                session_to_peer: HashMap::new(),
                peer_to_session: HashMap::new(),
                pending_events: VecDeque::new(),
                session_map: self.session_map.clone(),
                diagnostics: self.diagnostics.clone(),
            };

            // Read the relay's welcome messages (session_id + peer_list).
            // If this fails, treat it like a connection failure and retry.
            match signaller.read_welcome().await {
                Ok(()) => {
                    // Welcome handshake succeeded — the relay accepted the
                    // current token. Clear the blind-retry state so a future
                    // doomed token gets a fresh budget instead of inheriting
                    // leftover failures from an earlier flaky session.
                    #[cfg(target_arch = "wasm32")]
                    {
                        let mut guard = wasm_state.lock().unwrap_or_else(|e| e.into_inner());
                        guard.failures = 0;
                        guard.last_failed_token = None;
                    }
                    break signaller;
                }
                Err(e) => {
                    if let Some(ref mut remaining) = attempts {
                        if *remaining <= 1 {
                            return Err(e);
                        }
                        *remaining -= 1;
                        tracing::warn!(
                            attempts_remaining = *remaining,
                            "welcome handshake failed, retrying in 3s"
                        );
                        futures_timer::Delay::new(Duration::from_secs(3)).await;
                        continue 'connect;
                    }
                    tracing::warn!("welcome handshake failed, retrying in 3s");
                    futures_timer::Delay::new(Duration::from_secs(3)).await;
                    continue 'connect;
                }
            }
        };

        Ok(Box::new(signaller))
    }
}

/// Returns `true` if the error was tagged by `try_connect` as an HTTP 4xx response.
///
/// 4xx errors are permanent client-side rejections (e.g. 401 Invalid JWT,
/// 403 Forbidden). The retry loop uses this to bail out immediately instead
/// of burning through all reconnect attempts on a deterministic failure.
fn is_http_client_error(e: &SignalingError) -> bool {
    matches!(e, SignalingError::UserImplementationError(s) if s.starts_with("http_client_error:"))
}

/// Extract the HTTP status code from an `http_client_error:{code}` error tag,
/// for diagnostics. Returns `None` for any other error shape.
fn http_client_error_status(e: &SignalingError) -> Option<u64> {
    match e {
        SignalingError::UserImplementationError(s) => s
            .strip_prefix("http_client_error:")
            .and_then(|code| code.trim().parse().ok()),
        _ => None,
    }
}

/// Maximum time to wait for a WebSocket handshake to complete.
/// Without this, a tarpitted or firewall-dropped TCP connection can hold the
/// future pending forever, bypassing the retry loop entirely.
#[cfg(not(target_arch = "wasm32"))]
const WS_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);

// ── Native connection ────────────────────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
impl SymbiosSignallerBuilder {
    async fn try_connect(
        &self,
        room_url: &str,
    ) -> Result<WebSocketStream<ConnectStream>, SignalingError> {
        use futures_util::future::Either;

        let token = self.current_token();
        let request = build_ws_request(room_url, token.as_deref())
            .map_err(|e| SignalingError::UserImplementationError(e.to_string()))?;

        let connect_fut = connect_async(request);
        let timeout_fut = futures_timer::Delay::new(WS_CONNECT_TIMEOUT);
        futures_util::pin_mut!(connect_fut);
        futures_util::pin_mut!(timeout_fut);

        match futures_util::future::select(connect_fut, timeout_fut).await {
            Either::Left((result, _)) => {
                let (stream, _) = result.map_err(|e| {
                    // Surface HTTP 4xx errors as a distinct variant so the retry loop
                    // can fast-fail without wasting reconnect attempts on auth failures.
                    if let tungstenite::Error::Http(ref resp) = e {
                        let code = resp.status().as_u16();
                        if resp.status().is_client_error() {
                            tracing::error!(
                                status = code,
                                "relay rejected connection (HTTP 4xx) — not retrying"
                            );
                            return SignalingError::UserImplementationError(format!(
                                "http_client_error:{code}"
                            ));
                        }
                    }
                    SignalingError::from(e)
                })?;
                Ok(stream)
            }
            Either::Right(_) => Err(SignalingError::UserImplementationError(
                "WebSocket connection timed out".to_string(),
            )),
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn build_ws_request(
    url: &str,
    access_jwt: Option<&str>,
) -> Result<tungstenite::http::Request<()>, tungstenite::Error> {
    // <-- Changed return type here
    use tungstenite::client::IntoClientRequest;

    // 1. Let tungstenite parse the URL and automatically generate all the
    // mandatory WebSocket headers (Upgrade, Connection, Sec-WebSocket-Key)
    let mut request = url.into_client_request()?;

    // 2. Inject our ATProto JWT into the pre-formatted request
    if let Some(token) = access_jwt {
        let header_value = format!("Bearer {token}")
            .parse::<tungstenite::http::HeaderValue>()
            .map_err(|e| tungstenite::Error::HttpFormat(e.into()))?;
        request.headers_mut().insert("Authorization", header_value);
    }

    Ok(request)
}

// ── WASM connection ──────────────────────────────────────────────────────────

#[cfg(target_arch = "wasm32")]
impl SymbiosSignallerBuilder {
    async fn try_connect(&self, room_url: &str) -> Result<WsStream, SignalingError> {
        // The browser WebSocket API does not support custom headers. We pass
        // the JWT via the Sec-WebSocket-Protocol header using the two-element
        // subprotocol trick: `["access_token", "<jwt>"]`. This avoids leaking
        // the token in URL query parameters (which are logged by proxies and
        // load balancers).
        let token = self.current_token();
        let protocols: Vec<&str> = match token.as_deref() {
            Some(t) => vec!["access_token", t],
            None => vec![],
        };

        let (_meta, stream) = WsMeta::connect(room_url, Some(protocols))
            .await
            .map_err(|e| SignalingError::UserImplementationError(e.to_string()))?;

        Ok(stream)
    }
}

// ── Signaller ────────────────────────────────────────────────────────────────

/// A [`Signaller`] that bridges between the matchbox protocol and the Symbios
/// relay's [`SignalEnvelope`] wire format.
pub struct SymbiosSignaller {
    #[cfg(not(target_arch = "wasm32"))]
    ws: WebSocketStream<ConnectStream>,
    #[cfg(target_arch = "wasm32")]
    ws: WsStream,
    local_peer_id: PeerId,
    session_to_peer: HashMap<String, PeerId>,
    peer_to_session: HashMap<PeerId, String>,
    pending_events: VecDeque<PeerEvent>,
    /// Shared `PeerId → session_id` view that the application reads through
    /// [`PeerSessionMapRes`]. `None` if no map was threaded through the
    /// builder (e.g. crate consumers that never need DID verification).
    session_map: Option<PeerSessionMap>,
    /// Shared signalling counters ([`SignalDiagnostics`]) the application reads
    /// through [`SignalDiagnosticsRes`]. `None` disables diagnostics.
    diagnostics: Option<SharedSignalDiagnostics>,
}

impl SymbiosSignaller {
    /// Increment one [`SignalDiagnostics`] counter, if a diagnostics sink was
    /// threaded through the builder. `select` picks the counter to bump.
    fn bump(&self, select: impl FnOnce(&SignalDiagnostics) -> &AtomicU64) {
        if let Some(d) = &self.diagnostics {
            select(d).fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Publish a peer → session binding to the shared map, if the application
    /// provided one. Called whenever the signaller learns a new peer.
    fn publish_peer(&self, peer: PeerId, session_id: &str) {
        if let Some(map) = &self.session_map {
            map.write()
                .unwrap_or_else(|e| e.into_inner())
                .insert(peer, session_id.to_owned());
        }
    }

    /// Remove a peer binding from the shared map when the relay reports the
    /// peer has left.
    fn unpublish_peer(&self, peer: &PeerId) {
        if let Some(map) = &self.session_map {
            map.write().unwrap_or_else(|e| e.into_inner()).remove(peer);
        }
    }
}

impl SymbiosSignaller {
    /// Read the relay's two initial messages (`session_id` and `peer_list`)
    /// and buffer the corresponding `PeerEvent`s.
    async fn read_welcome(&mut self) -> Result<(), SignalingError> {
        // 1. session_id message
        let session_msg = self.read_text().await?;
        let session_json: serde_json::Value = serde_json::from_str(&session_msg).map_err(|e| {
            SignalingError::UserImplementationError(format!("invalid session_id message: {e}"))
        })?;

        let session_id = session_json
            .get("id")
            .and_then(|v| v.as_str())
            .ok_or_else(|| {
                SignalingError::UserImplementationError("missing 'id' in session_id message".into())
            })?;

        self.local_peer_id = session_id_to_peer_id(session_id);
        self.track_session(session_id.to_owned(), self.local_peer_id);
        self.pending_events
            .push_back(PeerEvent::IdAssigned(self.local_peer_id));

        // 2. peer_list message
        let peer_list_msg = self.read_text().await?;
        let peer_list_json: serde_json::Value =
            serde_json::from_str(&peer_list_msg).map_err(|e| {
                SignalingError::UserImplementationError(format!("invalid peer_list message: {e}"))
            })?;

        // Register every peer already in the room, but emit `NewPeer` (which
        // makes matchbox create an SDP offer) only for the subset this peer is
        // the designated Offerer for — see [`should_offer`]. Emitting `NewPeer`
        // for every `peer_list` entry unconditionally is what caused
        // simultaneous-join glare: under a near-simultaneous join the relay
        // hands *both* peers a `peer_list` containing the other, so both would
        // become Offerers and deadlock (matchbox cannot recover from two
        // offers). The deterministic tie-break makes exactly one side offer;
        // the other registers the mapping only and lazily answers the incoming
        // offer via matchbox's `accept_handshake` path.
        let mut peer_list_len: u64 = 0;
        if let Some(peers) = peer_list_json.get("peers").and_then(|v| v.as_array()) {
            for peer_val in peers {
                if let Some(sid) = peer_val.as_str() {
                    peer_list_len += 1;
                    let is_new = !self.session_to_peer.contains_key(sid);
                    let should_offer = self.should_offer(sid);
                    let pid = self.get_or_create_peer_id(sid);
                    if is_new && should_offer {
                        self.pending_events.push_back(PeerEvent::NewPeer(pid));
                        self.bump(|d| &d.offers_initiated);
                    }
                }
            }
        }
        if let Some(d) = &self.diagnostics {
            d.peer_lists_received.fetch_add(1, Ordering::Relaxed);
            d.last_peer_list_len.store(peer_list_len, Ordering::Relaxed);
        }

        Ok(())
    }

    /// Decide whether *this* peer should be the WebRTC **Offerer** for the pair
    /// `(self, remote)`, using a deterministic total order over the two peers'
    /// relay session identifiers.
    ///
    /// matchbox requires exactly one side of every pair to be the Offerer: it
    /// delivers `NewPeer` to that side (which creates the SDP offer) and the
    /// other side answers. The relay cannot guarantee this alone — under a
    /// near-simultaneous join it inserts each joiner into the room map before
    /// computing its `peer_list`, so both peers receive a `peer_list`
    /// containing the other. A naive "whoever sees the other offers" rule then
    /// makes both offer, producing WebRTC "glare": each `offer_handshake` spins
    /// forever awaiting an Answer that never arrives (the peer sent an Offer
    /// instead), no data channel opens, and both clients see an empty room.
    ///
    /// The tie-break removes the timing dependency: the peer whose deterministic
    /// session `PeerId` compares greater is always the Offerer, regardless of
    /// join order or of which relay message (`peer_list` vs `PeerJoined`) first
    /// revealed the pairing. Both peers derive the same order from the same two
    /// session ids, so exactly one offers and the other answers.
    ///
    /// The comparison uses the *deterministic* [`session_id_to_peer_id`] UUID
    /// (`self.local_peer_id` is already that value for the local peer), **not**
    /// the random `PeerId` that [`get_or_create_peer_id`] mints for matchbox's
    /// per-connection state machine — that random value differs on each side
    /// and would not yield a consistent order.
    fn should_offer(&self, remote_session_id: &str) -> bool {
        local_is_offerer(self.local_peer_id, remote_session_id)
    }

    /// Look up or create a `PeerId` for the given session ID string.
    fn get_or_create_peer_id(&mut self, session_id: &str) -> PeerId {
        // If we already know this exact session string (from the current connection instance), reuse it.
        if let Some(&pid) = self.session_to_peer.get(session_id) {
            tracing::debug!(
                session = %session_id,
                peer = %pid,
                "peer ID lookup: reusing existing mapping",
            );
            return pid;
        }

        // MINT A FRESH RANDOM ID instead of using the deterministic UUID hash.
        // This entirely sidesteps the WebRTC glare on rapid reconnects because Matchbox
        // will see the reconnecting peer as a completely new, distinct `PeerId`.
        let pid = PeerId(uuid::Uuid::new_v4());
        tracing::debug!(
            session = %session_id,
            peer = %pid,
            known_peers = self.session_to_peer.len(),
            "peer ID lookup: minted new PeerId (first time seeing this session)",
        );

        self.track_session(session_id.to_owned(), pid);
        pid
    }

    /// Insert a bidirectional mapping between session ID and PeerId.
    ///
    /// Also publishes the mapping to the shared [`PeerSessionMap`] (if
    /// present) so the host application can verify identity claims from this
    /// peer. The only entry that is *not* exposed via the map is the local
    /// peer — the map is intended for identifying *remote* peers that signed
    /// messages over the data channel, which the local peer never needs to do
    /// against itself.
    fn track_session(&mut self, session_id: String, peer_id: PeerId) {
        self.session_to_peer.insert(session_id.clone(), peer_id);
        self.peer_to_session.insert(peer_id, session_id.clone());
        if peer_id != self.local_peer_id {
            self.publish_peer(peer_id, &session_id);
        }
    }

    /// Remove a peer from the ID maps and return its `PeerId`.
    fn remove_peer(&mut self, session_id: &str) -> PeerId {
        let pid = self
            .session_to_peer
            .remove(session_id)
            .unwrap_or_else(|| session_id_to_peer_id(session_id));
        self.peer_to_session.remove(&pid);
        self.unpublish_peer(&pid);
        pid
    }
}

/// Clear any peer bindings this signaller owns from the shared
/// [`PeerSessionMap`] when it is dropped.
///
/// Without this, every reconnect (new `SymbiosSignaller` constructed by the
/// builder) would leak entries keyed by its own freshly-minted random
/// `PeerId`s into the shared map. For long-lived games on flaky connections,
/// the map would grow unbounded.
impl Drop for SymbiosSignaller {
    fn drop(&mut self) {
        let Some(map) = self.session_map.as_ref() else {
            return;
        };
        let mut guard = map.write().unwrap_or_else(|e| e.into_inner());
        for pid in self.peer_to_session.keys() {
            guard.remove(pid);
        }
    }
}

// ── Platform-specific read_text / Signaller impl ─────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
impl SymbiosSignaller {
    /// Read the next text frame from the WebSocket (native).
    async fn read_text(&mut self) -> Result<String, SignalingError> {
        loop {
            match self.ws.next().await {
                Some(Ok(tungstenite::Message::Text(t))) => {
                    // `Utf8Bytes` wraps `tungstenite::Bytes` with a UTF-8
                    // invariant tungstenite already validated when it built the
                    // frame. Go through `Bytes -> Vec<u8>` so the unique-owner
                    // case (just handed to us from the frame reader) reuses
                    // the existing allocation rather than paying the alloc +
                    // memcpy of `.to_string()`; shared views fall back to a
                    // copy via `From<Bytes>` — same cost as before. We then
                    // re-validate with the checked decode: skipping it would
                    // turn any future tungstenite bug, API change, or
                    // misrouted binary frame into UB. The validation cost is
                    // negligible for typical signaling frames (< 1 KiB) and a
                    // failure is treated as a relay protocol violation,
                    // dropping the connection so the reconnect path picks up.
                    let bytes: tungstenite::Bytes = t.into();
                    let vec: Vec<u8> = bytes.into();
                    return String::from_utf8(vec).map_err(|e| {
                        tracing::warn!(error = %e, "relay sent non-UTF-8 text frame; closing socket");
                        SignalingError::UserImplementationError(format!(
                            "relay protocol violation: text frame contained invalid UTF-8 ({e})"
                        ))
                    });
                }
                Some(Ok(tungstenite::Message::Close(_))) | None => {
                    return Err(SignalingError::StreamExhausted);
                }
                Some(Ok(_)) => continue, // skip pings, binary, etc.
                Some(Err(e)) => return Err(SignalingError::from(e)),
            }
        }
    }
}

#[cfg(target_arch = "wasm32")]
impl SymbiosSignaller {
    /// Read the next text frame from the WebSocket (WASM).
    async fn read_text(&mut self) -> Result<String, SignalingError> {
        loop {
            match self.ws.next().await {
                Some(WasmWsMessage::Text(t)) => return Ok(t),
                Some(WasmWsMessage::Binary(_)) => continue,
                None => return Err(SignalingError::StreamExhausted),
            }
        }
    }
}

#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Signaller for SymbiosSignaller {
    async fn send(&mut self, request: PeerRequest) -> Result<(), SignalingError> {
        match request {
            PeerRequest::Signal { receiver, data } => {
                let target_session = match self.peer_to_session.get(&receiver) {
                    Some(s) => s.clone(),
                    None => {
                        // Peer disconnected between the signal being queued and
                        // sent — this is a normal race condition, not fatal.
                        tracing::debug!(%receiver, "dropping signal to unknown peer (likely disconnected)");
                        return Ok(());
                    }
                };

                let signal_kind = match &data {
                    PeerSignal::Offer(_) => "Offer",
                    PeerSignal::Answer(_) => "Answer",
                    PeerSignal::IceCandidate(_) => "IceCandidate",
                };
                tracing::debug!(
                    local = %self.local_peer_id,
                    %receiver,
                    target = %target_session,
                    signal = signal_kind,
                    "signaller TX",
                );

                let signal = match data {
                    PeerSignal::Offer(sdp) => {
                        self.bump(|d| &d.offers_sent);
                        SignalPayload::Offer(sdp)
                    }
                    PeerSignal::Answer(sdp) => {
                        self.bump(|d| &d.answers_sent);
                        SignalPayload::Answer(sdp)
                    }
                    PeerSignal::IceCandidate(c) => SignalPayload::IceCandidate(c),
                };

                let envelope = SignalEnvelope {
                    peer_id: target_session,
                    signal,
                };
                let json = serde_json::to_string(&envelope)
                    .map_err(|e| SignalingError::UserImplementationError(e.to_string()))?;

                self.send_text(json).await
            }
            PeerRequest::KeepAlive => self.send_ping().await,
        }
    }

    async fn next_message(&mut self) -> Result<PeerEvent, SignalingError> {
        // Drain buffered events first (from welcome handshake).
        if let Some(event) = self.pending_events.pop_front() {
            return Ok(event);
        }

        loop {
            let text = self.read_text().await?;

            // Try parsing as a SignalEnvelope (the normal case).
            if let Ok(envelope) = serde_json::from_str::<SignalEnvelope>(&text) {
                let sender_id = &envelope.peer_id;

                let signal_kind = match &envelope.signal {
                    SignalPayload::Offer(_) => "Offer",
                    SignalPayload::Answer(_) => "Answer",
                    SignalPayload::IceCandidate(_) => "IceCandidate",
                    SignalPayload::PeerJoined(_) => "PeerJoined",
                    SignalPayload::PeerLeft(_) => "PeerLeft",
                };
                tracing::debug!(
                    local = %self.local_peer_id,
                    sender_session = %sender_id,
                    signal = signal_kind,
                    "signaller RX",
                );

                return match envelope.signal {
                    SignalPayload::Offer(sdp) => {
                        self.bump(|d| &d.offers_received);
                        let pid = self.get_or_create_peer_id(sender_id);
                        tracing::debug!(
                            local = %self.local_peer_id,
                            sender_session = %sender_id,
                            mapped_peer = %pid,
                            "RX Offer → forwarding to matchbox as PeerSignal::Offer",
                        );
                        Ok(PeerEvent::Signal {
                            sender: pid,
                            data: PeerSignal::Offer(sdp),
                        })
                    }
                    SignalPayload::Answer(sdp) => {
                        self.bump(|d| &d.answers_received);
                        let pid = self.get_or_create_peer_id(sender_id);
                        tracing::debug!(
                            local = %self.local_peer_id,
                            sender_session = %sender_id,
                            mapped_peer = %pid,
                            "RX Answer → forwarding to matchbox as PeerSignal::Answer",
                        );
                        Ok(PeerEvent::Signal {
                            sender: pid,
                            data: PeerSignal::Answer(sdp),
                        })
                    }
                    SignalPayload::IceCandidate(c) => {
                        let pid = self.get_or_create_peer_id(sender_id);
                        Ok(PeerEvent::Signal {
                            sender: pid,
                            data: PeerSignal::IceCandidate(c),
                        })
                    }
                    SignalPayload::PeerJoined(ref id) => {
                        // Register the session→PeerId mapping so subsequent
                        // signals from this peer resolve correctly. Matchbox
                        // treats `NewPeer` as "you are the Offerer" and creates
                        // an SDP offer, so we emit it only when the deterministic
                        // tie-break ([`should_offer`]) designates us the Offerer
                        // for this pair *and* we have not already started
                        // offering to it. A near-simultaneous join can surface
                        // the same peer via both `peer_list` (in `read_welcome`)
                        // and this `PeerJoined`; the `is_new` guard prevents a
                        // duplicate offer that would clobber the in-flight
                        // handshake.
                        //
                        // If we are the Answerer, we register the mapping only
                        // and wait: when the Offerer's Offer arrives as a
                        // `PeerEvent::Signal` from a not-yet-handshaking PeerId,
                        // matchbox lazily creates the peer via `accept_handshake`
                        // (the `or_insert_with` path in `message_loop`).
                        let is_new = !self.session_to_peer.contains_key(id);
                        let should_offer = self.should_offer(id);
                        let pid = self.get_or_create_peer_id(id);
                        if is_new && should_offer {
                            self.bump(|d| &d.offers_initiated);
                            tracing::debug!(
                                local = %self.local_peer_id,
                                joined_session = %id,
                                mapped_peer = %pid,
                                "RX PeerJoined → tie-break makes us the Offerer → emitting NewPeer",
                            );
                            return Ok(PeerEvent::NewPeer(pid));
                        }
                        tracing::debug!(
                            local = %self.local_peer_id,
                            joined_session = %id,
                            mapped_peer = %pid,
                            already_known = !is_new,
                            "RX PeerJoined → tie-break makes us the Answerer → registered \
                             mapping, will accept incoming Offer lazily",
                        );
                        continue;
                    }
                    SignalPayload::PeerLeft(ref id) => {
                        let pid = self.remove_peer(id);
                        tracing::debug!(
                            local = %self.local_peer_id,
                            left_session = %id,
                            mapped_peer = %pid,
                            "RX PeerLeft → emitting PeerLeft",
                        );
                        Ok(PeerEvent::PeerLeft(pid))
                    }
                };
            }

            // Unknown message format — skip and read the next one.
            tracing::debug!(msg = %text, "ignoring unrecognized relay message");
        }
    }
}

// ── Platform-specific send helpers ───────────────────────────────────────────

#[cfg(not(target_arch = "wasm32"))]
impl SymbiosSignaller {
    async fn send_text(&mut self, text: String) -> Result<(), SignalingError> {
        self.ws
            .send(tungstenite::Message::Text(text.into()))
            .await
            .map_err(SignalingError::from)
    }

    async fn send_ping(&mut self) -> Result<(), SignalingError> {
        self.ws
            .send(tungstenite::Message::Ping(vec![].into()))
            .await
            .map_err(SignalingError::from)
    }
}

#[cfg(target_arch = "wasm32")]
impl SymbiosSignaller {
    async fn send_text(&mut self, text: String) -> Result<(), SignalingError> {
        self.ws
            .send(WasmWsMessage::Text(text))
            .await
            .map_err(|e| SignalingError::UserImplementationError(e.to_string()))
    }

    async fn send_ping(&mut self) -> Result<(), SignalingError> {
        // The browser handles WebSocket pings/pongs automatically — no-op.
        Ok(())
    }
}

// ── Public constructor ───────────────────────────────────────────────────────

/// Create a [`SymbiosSignallerBuilder`] backed by a shared [`TokenSource`].
///
/// On each reconnect attempt, the builder reads the latest token from the
/// source. The host application is responsible for updating the source when
/// tokens are refreshed (re-issue a service auth token via
/// [`crate::auth::get_service_auth`] after the underlying OAuth session
/// rotates its access token).
///
/// # Example
///
/// ```rust,ignore
/// use bevy_symbios_multiuser::signaller::{signaller_with_token_source, TokenSource};
/// use bevy_symbios_multiuser::auth::get_service_auth;
///
/// // `get_service_auth` returns a JWT signed by the user's `#atproto` key,
/// // which the relay can verify by resolving the user's DID document. This
/// // is the only safe token for third-party services — the OAuth access
/// // token is DPoP-bound and cannot be handed off.
/// let service_token = get_service_auth(&session, relay_did).await?;
/// let token_source = TokenSource::new(Some(service_token));
/// let builder = signaller_with_token_source(token_source.clone());
///
/// // Later, when the service token is near expiry, re-issue it:
/// let new_service_token = get_service_auth(&session, relay_did).await?;
/// token_source.set(Some(new_service_token));
/// ```
pub fn signaller_with_token_source(source: TokenSource) -> Arc<dyn SignallerBuilder> {
    Arc::new(SymbiosSignallerBuilder {
        token_source: Some(source),
        session_map: None,
        #[cfg(target_arch = "wasm32")]
        wasm_retry_state: None,
        diagnostics: None,
    })
}

/// Create an anonymous [`SymbiosSignallerBuilder`] that connects without
/// authentication but still speaks the relay's [`SignalEnvelope`] wire format.
///
/// This ensures anonymous clients are understood by the relay, rather than
/// falling back to matchbox's default signaller which uses an incompatible
/// JSON format.
pub fn signaller_anonymous() -> Arc<dyn SignallerBuilder> {
    Arc::new(SymbiosSignallerBuilder {
        token_source: None,
        session_map: None,
        #[cfg(target_arch = "wasm32")]
        wasm_retry_state: None,
        diagnostics: None,
    })
}

/// Create a [`SymbiosSignallerBuilder`] with a shared [`PeerSessionMap`].
///
/// The builder carries `session_map` forward to every [`SymbiosSignaller`] it
/// produces, so reconnects keep populating the same map and the application's
/// [`PeerSessionMapRes`] view stays consistent.
///
/// Anonymous (no-JWT) variant. For authenticated connections, use
/// [`signaller_with_token_source_and_map`] with a service auth token from
/// [`crate::auth::get_service_auth`].
pub fn signaller_anonymous_with_map(session_map: PeerSessionMap) -> Arc<dyn SignallerBuilder> {
    Arc::new(SymbiosSignallerBuilder {
        token_source: None,
        session_map: Some(session_map),
        #[cfg(target_arch = "wasm32")]
        wasm_retry_state: None,
        diagnostics: None,
    })
}

/// Refreshable-token variant of [`signaller_anonymous_with_map`].
pub fn signaller_with_token_source_and_map(
    source: TokenSource,
    session_map: PeerSessionMap,
) -> Arc<dyn SignallerBuilder> {
    Arc::new(SymbiosSignallerBuilder {
        token_source: Some(source),
        session_map: Some(session_map),
        #[cfg(target_arch = "wasm32")]
        wasm_retry_state: None,
        diagnostics: None,
    })
}

/// Fully-specified constructor used by the Bevy plugin's `open_socket` path.
///
/// The plugin manages three pieces of cross-reconnect state and needs to thread
/// them all through every signaller it builds:
/// - [`TokenSource`] for auth-token refresh on reconnect,
/// - [`PeerSessionMap`] so the DID ↔ `PeerId` binding survives the short window
///   where the old signaller is dropped and the new one has not yet replayed
///   `peer_list`,
/// - [`WasmBlindRetryStateHandle`] (wasm only) so the blind-retry counter is
///   not re-zeroed on every ECS respawn.
///
/// Callers outside the plugin should prefer [`signaller_with_token_source_and_map`]
/// or [`signaller_anonymous_with_map`] — they produce a builder whose WASM
/// retry state is local to a single `new_signaller` call, which is the correct
/// scope for one-shot programmatic use.
///
/// `diagnostics` threads a shared [`SignalDiagnostics`] sink through every
/// signaller the builder produces, so the host application can observe the
/// signalling layer (peer_list size, offers/answers) that matchbox otherwise
/// hides. Pass `None` to disable diagnostics.
pub fn signaller_full(
    token_source: Option<TokenSource>,
    session_map: Option<PeerSessionMap>,
    diagnostics: Option<SharedSignalDiagnostics>,
    #[cfg(target_arch = "wasm32")] wasm_retry_state: Option<WasmBlindRetryStateHandle>,
) -> Arc<dyn SignallerBuilder> {
    Arc::new(SymbiosSignallerBuilder {
        token_source,
        session_map,
        #[cfg(target_arch = "wasm32")]
        wasm_retry_state,
        diagnostics,
    })
}

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

    /// The core anti-glare invariant: the offerer arbitration is a strict total
    /// order, so for any two distinct sessions exactly one side is the Offerer.
    /// This is what prevents BOTH the simultaneous-join glare (both offering,
    /// the bug this fixes) and its inverse (neither offering — a silent
    /// no-connection). `^` asserts exactly-one, never both, never neither.
    #[test]
    fn exactly_one_peer_offers_per_pair() {
        let pairs = [
            ("did:plc:alice", "did:plc:bob"),
            ("did:web:example.com", "did:plc:zzz"),
            ("11111111-1111-1111-1111-111111111111", "did:plc:someone"),
            (
                "11111111-1111-1111-1111-111111111111",
                "22222222-2222-2222-2222-222222222222",
            ),
        ];
        for (a, b) in pairs {
            let a_offers = local_is_offerer(session_id_to_peer_id(a), b);
            let b_offers = local_is_offerer(session_id_to_peer_id(b), a);
            assert!(
                a_offers ^ b_offers,
                "exactly one of ({a}, {b}) must offer, got a_offers={a_offers} b_offers={b_offers}",
            );
        }
    }

    /// The decision does not depend on evaluation order or join timing — the
    /// whole point of the tie-break. The higher deterministic PeerId always
    /// offers, whichever side computes it.
    #[test]
    fn decision_is_order_independent_and_deterministic() {
        let a = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
        let b = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb";
        let first = local_is_offerer(session_id_to_peer_id(a), b);
        let again = local_is_offerer(session_id_to_peer_id(a), b);
        assert_eq!(first, again, "decision must be deterministic");
        let expected = session_id_to_peer_id(a) > session_id_to_peer_id(b);
        assert_eq!(
            first, expected,
            "higher deterministic PeerId must be the offerer"
        );
    }

    /// A peer never offers to itself (`>` is irreflexive), so a self-entry
    /// accidentally left in a peer_list can never spawn a pointless self-offer.
    #[test]
    fn a_peer_never_offers_to_itself() {
        let me = "did:plc:selfselfselfselfselfself";
        assert!(!local_is_offerer(session_id_to_peer_id(me), me));
    }

    /// `session_id_to_peer_id` is deterministic — the same session string maps
    /// to the same PeerId on both peers — which is what lets them agree on the
    /// order. A raw UUID session id is used verbatim; a DID is hashed via v5.
    #[test]
    fn session_id_to_peer_id_is_deterministic() {
        assert_eq!(
            session_id_to_peer_id("did:plc:example"),
            session_id_to_peer_id("did:plc:example"),
        );
        let u = "abcdef00-1234-5678-9abc-def012345678";
        assert_eq!(
            session_id_to_peer_id(u),
            PeerId(Uuid::parse_str(u).unwrap())
        );
    }
}