//! Shared application state — Arc-wrapped, shared across all connections.

use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
use std::sync::Arc;
use std::time::Instant;

use axum::body::Bytes;
use axum::extract::ws::{Message as WsMessage, Utf8Bytes as WsUtf8Bytes};
use dashmap::DashMap;
use tokio::sync::mpsc;
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use buzz_audit::AuditService;
use buzz_auth::{AuthService, Nip98ReplayGuard};
use buzz_core::tenant::TenantContext;
use buzz_core::CommunityId;
use buzz_db::Db;
use buzz_media::MediaStorage;
use buzz_pubsub::cache_invalidation::CacheInvalidation;
use buzz_pubsub::conn_control::ConnControl;
use buzz_pubsub::rate_limiter::RedisRateLimiter;
use buzz_pubsub::{PubSubManager, RedisNip98ReplayGuard};
use buzz_search::SearchService;
use buzz_workflow::WorkflowEngine;
use deadpool_redis;

use crate::audio::AudioRoomManager;
use crate::config::Config;
use crate::connection::ConnectionSubscriptions;
use crate::subscription::SubscriptionRegistry;

pub(crate) type ScopedPubkeyKey = (CommunityId, [u8; 32]);
type SlidingWindowCounter = (u32, Instant);
type ScopedRateLimiter = DashMap<ScopedPubkeyKey, SlidingWindowCounter>;

/// Per-connection entry in the connection manager.
struct ConnEntry {
    tx: mpsc::Sender<WsMessage>,
    /// Control-frame sender, drained ahead of data and before cancel wins in
    /// the send loop. Used to deliver a ban-disconnect frame that must reach
    /// the client before the socket is closed (see [`ConnectionManager::disconnect_pubkey`]).
    ctrl_tx: mpsc::Sender<WsMessage>,
    cancel: CancellationToken,
    /// Community resolved from the connection host at handshake. This is the
    /// receiver-side tenant label fan-out must compare against the event label.
    community_id: CommunityId,
    /// Shared with `ConnectionState` — both direct sends and fan-out
    /// broadcasts track the same consecutive-full counter.
    backpressure_count: Arc<AtomicU8>,
    subscriptions: ConnectionSubscriptions,
    authenticated_pubkey: Arc<std::sync::RwLock<Option<Vec<u8>>>>,
    grace_limit: u8,
}

/// Community-scoped lifecycle registry shared by every long-lived socket type.
///
/// A handler registers before durable active-state revalidation. Archival after
/// registration cancels the token; archival before registration is observed by
/// the revalidation. The returned guard removes the entry on every exit path.
pub struct CommunityConnectionRegistry {
    connections: Arc<DashMap<Uuid, (CommunityId, CancellationToken)>>,
}

impl Default for CommunityConnectionRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl CommunityConnectionRegistry {
    /// Creates an empty lifecycle registry.
    pub fn new() -> Self {
        Self {
            connections: Arc::new(DashMap::new()),
        }
    }

    /// Registers one socket and returns a guard that deregisters it on drop.
    pub fn register(
        &self,
        connection_id: Uuid,
        community_id: CommunityId,
        cancel: CancellationToken,
    ) -> CommunityConnectionGuard {
        self.connections
            .insert(connection_id, (community_id, cancel));
        CommunityConnectionGuard {
            connection_id,
            connections: Arc::clone(&self.connections),
        }
    }

    /// Cancels every socket type currently bound to `community_id`.
    pub fn disconnect_community(&self, community_id: CommunityId) -> usize {
        let mut closed = 0;
        for entry in self.connections.iter() {
            if entry.value().0 == community_id {
                entry.value().1.cancel();
                closed += 1;
            }
        }
        closed
    }

    /// Returns the distinct communities with live sockets on this pod.
    pub fn bound_communities(&self) -> HashSet<CommunityId> {
        self.connections
            .iter()
            .map(|entry| entry.value().0)
            .collect()
    }
}

/// Removes a socket lifecycle registration on every handler exit path.
pub struct CommunityConnectionGuard {
    connection_id: Uuid,
    connections: Arc<DashMap<Uuid, (CommunityId, CancellationToken)>>,
}

impl Drop for CommunityConnectionGuard {
    fn drop(&mut self) {
        self.connections.remove(&self.connection_id);
    }
}

/// Registers a socket, durably revalidates its community, then runs it.
///
/// The ordering is the archival admission invariant: archive-before-query is
/// observed by the query, while archive-after-registration sees the token.
pub async fn run_registered_community_connection<Check, CheckFuture, Run, RunFuture>(
    registry: &CommunityConnectionRegistry,
    connection_id: Uuid,
    community_id: CommunityId,
    cancel: CancellationToken,
    check_active: Check,
    run: Run,
) where
    Check: FnOnce() -> CheckFuture,
    CheckFuture: Future<Output = Result<bool, buzz_db::DbError>>,
    Run: FnOnce() -> RunFuture,
    RunFuture: Future<Output = ()>,
{
    let _guard = registry.register(connection_id, community_id, cancel.clone());
    if !matches!(check_active().await, Ok(true)) {
        cancel.cancel();
        return;
    }
    if cancel.is_cancelled() {
        return;
    }
    run().await;
    cancel.cancel();
}

async fn revalidate_registered_communities<Check, CheckFuture>(
    registry: &CommunityConnectionRegistry,
    mut check_active: Check,
) -> (usize, Vec<(CommunityId, buzz_db::DbError)>)
where
    Check: FnMut(CommunityId) -> CheckFuture,
    CheckFuture: Future<Output = Result<bool, buzz_db::DbError>>,
{
    let communities = registry.bound_communities();
    let mut closed = 0;
    let mut failures = Vec::new();
    for community_id in communities {
        match check_active(community_id).await {
            Ok(false) => closed += registry.disconnect_community(community_id),
            Ok(true) => {}
            Err(error) => failures.push((community_id, error)),
        }
    }
    (closed, failures)
}

/// Tracks active Nostr WebSocket connections and provides message routing by connection ID.
pub struct ConnectionManager {
    connections: DashMap<Uuid, ConnEntry>,
}

impl ConnectionManager {
    /// Creates a new, empty connection manager.
    pub fn new() -> Self {
        Self {
            connections: DashMap::new(),
        }
    }

    /// Registers a connection with its outbound sender, cancellation token,
    /// server-resolved community, shared backpressure counter, mutable
    /// subscription map, and grace limit.
    // Each argument is a distinct per-connection attribute stored verbatim in
    // `ConnEntry`; a params struct would only relocate the same fields.
    #[allow(clippy::too_many_arguments)]
    pub fn register(
        &self,
        conn_id: Uuid,
        tx: mpsc::Sender<WsMessage>,
        ctrl_tx: mpsc::Sender<WsMessage>,
        cancel: CancellationToken,
        community_id: CommunityId,
        backpressure_count: Arc<AtomicU8>,
        subscriptions: ConnectionSubscriptions,
        grace_limit: u8,
    ) {
        self.connections.insert(
            conn_id,
            ConnEntry {
                tx,
                ctrl_tx,
                cancel,
                community_id,
                backpressure_count,
                subscriptions,
                authenticated_pubkey: Arc::new(std::sync::RwLock::new(None)),
                grace_limit,
            },
        );
    }

    /// Removes a connection from the registry.
    pub fn deregister(&self, conn_id: Uuid) {
        self.connections.remove(&conn_id);
    }

    /// Record the authenticated pubkey for a connection after NIP-42 succeeds.
    pub fn set_authenticated_pubkey(&self, conn_id: Uuid, pubkey_bytes: Vec<u8>) {
        if let Some(entry) = self.connections.get(&conn_id) {
            if let Ok(mut slot) = entry.authenticated_pubkey.write() {
                *slot = Some(pubkey_bytes);
            }
        }
    }

    /// Return live connection IDs authenticated as `pubkey_bytes` in one community.
    ///
    /// The same Nostr key may be connected to multiple communities at once.
    /// Callers use this for tenant-visible cleanup such as presence clearing and
    /// subscription eviction, so a connection in B must not keep A's derived
    /// state alive.
    pub fn connection_ids_for_pubkey_in_community(
        &self,
        community_id: CommunityId,
        pubkey_bytes: &[u8],
    ) -> Vec<Uuid> {
        self.connections
            .iter()
            .filter_map(|entry| {
                let matches = entry.community_id == community_id
                    && entry
                        .authenticated_pubkey
                        .read()
                        .ok()
                        .and_then(|value| {
                            value
                                .as_ref()
                                .map(|stored| stored.as_slice() == pubkey_bytes)
                        })
                        .unwrap_or(false);
                matches.then_some(*entry.key())
            })
            .collect()
    }

    /// Return the authenticated pubkey recorded for a connection, if any.
    pub fn pubkey_for_conn(&self, conn_id: Uuid) -> Option<Vec<u8>> {
        self.connections
            .get(&conn_id)
            .and_then(|entry| entry.authenticated_pubkey.read().ok()?.clone())
    }

    /// Disconnect every live connection authenticated as `pubkey` **in
    /// `community`**, delivering a final `OK false` frame carrying `reason`
    /// before closing.
    ///
    /// Used for live ban enforcement (COMMUNITY_MODERATION_PLAN.md §0 decision
    /// 4): a ban must take effect immediately on existing sessions, not just at
    /// the next auth. The frame is sent on the control channel, which the send
    /// loop drains ahead of both queued data and the biased cancel branch, so
    /// the client learns *why* it was dropped. `event_id` labels the `OK` (the
    /// ban has no triggering client event, so a synthetic all-zero id is used).
    ///
    /// The `community` filter is the tenant fence: one pod holds sockets for
    /// many communities, and the same pubkey may be live in several. A ban in
    /// community A must close only A's sockets, never a session the member holds
    /// in community B ("authority stays inside the tenant fence").
    ///
    /// Returns the number of connections closed. This is the pod-local half of
    /// live enforcement; cross-pod fan-out publishes the same intent over Redis.
    pub fn disconnect_pubkey(
        &self,
        community: CommunityId,
        pubkey: &[u8],
        event_id: &str,
        reason: &str,
    ) -> usize {
        let frame = crate::protocol::RelayMessage::ok(event_id, false, reason);
        let mut closed = 0usize;
        for conn_id in self.connection_ids_for_pubkey_in_community(community, pubkey) {
            if let Some(entry) = self.connections.get(&conn_id) {
                if entry.community_id != community {
                    continue;
                }
                // Best-effort delivery: a full control buffer still gets the
                // close via cancel below, just without the reason frame.
                let _ = entry
                    .ctrl_tx
                    .try_send(WsMessage::Text(frame.clone().into()));
                entry.cancel.cancel();
                closed += 1;
            }
        }
        closed
    }

    /// Return the server-resolved community that the connection's host bound to.
    pub fn community_for_conn(&self, conn_id: Uuid) -> Option<CommunityId> {
        self.connections
            .get(&conn_id)
            .map(|entry| entry.community_id)
    }

    /// Return the subscription map for a connection, if it is still live.
    pub fn subscriptions_for(&self, conn_id: Uuid) -> Option<ConnectionSubscriptions> {
        self.connections
            .get(&conn_id)
            .map(|entry| Arc::clone(&entry.subscriptions))
    }

    /// Snapshot the number of live WebSocket connections per community.
    ///
    /// Returns a map from community UUID to connection count. Used by the
    /// usage poller; snapshotting avoids per-community gauge drift from
    /// mismatched inc/dec across async boundaries.
    pub fn per_community_ws_connections(&self) -> HashMap<CommunityId, u64> {
        let mut counts: HashMap<CommunityId, u64> = HashMap::new();
        for entry in self.connections.iter() {
            *counts.entry(entry.community_id).or_default() += 1;
        }
        counts
    }

    /// Snapshot the number of distinct authenticated pubkeys online per community.
    ///
    /// A pubkey connected to multiple pods will be counted once per pod — the
    /// dashboard sums across pods, so per-pod partial counts are correct.
    /// A pubkey connected twice on the same pod is counted once (distinct set).
    pub fn per_community_users_online(&self) -> HashMap<CommunityId, u64> {
        // community_id → set of pubkey bytes
        let mut seen: HashMap<CommunityId, HashSet<Vec<u8>>> = HashMap::new();
        for entry in self.connections.iter() {
            if let Ok(lock) = entry.authenticated_pubkey.read() {
                if let Some(pk) = lock.as_ref() {
                    seen.entry(entry.community_id)
                        .or_default()
                        .insert(pk.clone());
                }
            }
        }
        seen.into_iter()
            .map(|(cid, set)| (cid, set.len() as u64))
            .collect()
    }

    /// Return the authenticated pubkey for a connection, if any.
    pub fn pubkey_for(&self, conn_id: Uuid) -> Option<Vec<u8>> {
        self.connections
            .get(&conn_id)
            .and_then(|entry| entry.authenticated_pubkey.read().ok()?.clone())
    }

    /// Sends a text message to the given connection.
    ///
    /// Returns `false` if the connection is gone or the buffer is full.
    /// On sustained backpressure (>grace_limit consecutive full buffers),
    /// cancels the connection. Transient stalls get a warning only.
    pub fn send_to(&self, conn_id: Uuid, msg: String) -> bool {
        self.try_send_ws_message(conn_id, WsMessage::Text(msg.into()))
    }

    /// Sends an already-serialized UTF-8 text payload to the given connection.
    ///
    /// The shared `Bytes` payload is cloned into the outbound WS message without
    /// copying the frame body. Callers must only pass valid UTF-8 bytes.
    pub fn send_to_text_bytes(&self, conn_id: Uuid, msg: Arc<Bytes>) -> bool {
        let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
            .expect("relay fan-out frames are serialized UTF-8 JSON");
        self.try_send_ws_message(conn_id, WsMessage::Text(text))
    }

    fn try_send_ws_message(&self, conn_id: Uuid, msg: WsMessage) -> bool {
        if let Some(entry) = self.connections.get(&conn_id) {
            let conn = entry.value();
            match conn.tx.try_send(msg) {
                Ok(_) => {
                    conn.backpressure_count.store(0, Ordering::Relaxed);
                    true
                }
                Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                    let count = conn.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1;
                    if count >= conn.grace_limit {
                        tracing::warn!(conn_id = %conn_id, count, "fan-out: sustained backpressure — cancelling slow client");
                        metrics::counter!("buzz_ws_backpressure_disconnects_total").increment(1);
                        conn.cancel.cancel();
                    } else {
                        tracing::warn!(conn_id = %conn_id, count, grace = conn.grace_limit, "fan-out: send buffer full — grace {count}/{}", conn.grace_limit);
                    }
                    false
                }
                Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                    tracing::debug!(conn_id = %conn_id, "fan-out: send channel closed");
                    false
                }
            }
        } else {
            false
        }
    }
}

impl Default for ConnectionManager {
    fn default() -> Self {
        Self::new()
    }
}

/// Shared application state, cloned cheaply via inner `Arc` fields.
#[derive(Clone)]
pub struct AppState {
    /// Relay configuration.
    pub config: Arc<Config>,
    /// Database connection pool.
    pub db: Db,
    /// Redis pool for readiness health checks.
    pub redis_pool: deadpool_redis::Pool,
    /// Audit event service.
    pub audit: Arc<AuditService>,
    /// Pub/sub manager for broadcasting events to subscribers.
    pub pubsub: Arc<PubSubManager>,
    /// Authentication service.
    pub auth: Arc<AuthService>,
    /// Full-text search service.
    pub search: Arc<SearchService>,
    /// Registry of active client subscriptions.
    pub sub_registry: Arc<SubscriptionRegistry>,
    /// Registry of active WebSocket connections.
    pub conn_manager: Arc<ConnectionManager>,
    /// Lifecycle cancellation for every long-lived socket, including huddle audio.
    pub community_connections: Arc<CommunityConnectionRegistry>,
    /// Stops only the periodic lifecycle revalidator during graceful shutdown.
    pub community_revalidator_cancel: CancellationToken,
    /// Test/telemetry counter for archive disconnect publication attempts.
    pub community_disconnect_publish_attempts: Arc<AtomicU64>,
    /// Semaphore limiting total concurrent connections.
    pub conn_semaphore: Arc<Semaphore>,
    /// Semaphore limiting concurrent message handler tasks.
    pub handler_semaphore: Arc<Semaphore>,
    /// Semaphore limiting concurrent git subprocess operations across
    /// the whole relay. Bounds resource use; **not** writer
    /// serialization — that's the CAS at the manifest pointer (spec
    /// §Push step 7, `Inv_NoFork`).
    pub git_semaphore: Arc<Semaphore>,
    /// Semaphore limiting concurrent media upload parsing/transcoding work.
    pub media_upload_semaphore: Arc<Semaphore>,

    /// Workflow engine for background processing.
    pub workflow_engine: Arc<WorkflowEngine>,
    /// Relay signing keypair — used to sign system messages (kind 40099).
    pub relay_keypair: nostr::Keys,

    /// Recently-published event IDs for local-echo deduplication, keyed by
    /// `(community_id, event_id)`. Events fanned out in-process are added here;
    /// the Redis subscriber consumer skips them to avoid double delivery.
    ///
    /// The community is part of the key because the same Nostr event id can
    /// legitimately exist in two communities (channel-less events, and
    /// same-channel-UUID/same-`h` events across tenants). Keying on the bare id
    /// would let a local publish in community A suppress delivery of a distinct
    /// event with the same id arriving via Redis for community B — a
    /// cross-community non-interference violation. Entries expire after 60
    /// seconds via moka's TTL eviction — bounded regardless of subscriber health.
    pub local_event_ids: Arc<moka::sync::Cache<(CommunityId, [u8; 32]), ()>>,
    /// Membership cache: (community_id, channel_id, pubkey_bytes) → is_member.
    /// Short TTL (10s) — membership changes are rare but must propagate.
    #[allow(clippy::type_complexity)]
    pub membership_cache: Arc<moka::sync::Cache<(CommunityId, Uuid, Vec<u8>), bool>>,
    /// Accessible channel IDs cache: (community_id, pubkey_bytes) → channel UUIDs.
    /// Short TTL (10s) — invalidated on membership or channel visibility changes.
    #[allow(clippy::type_complexity)]
    pub accessible_channels_cache: Arc<moka::sync::Cache<(CommunityId, Vec<u8>), Vec<Uuid>>>,
    /// Per-community channel visibility string, used to gate the private-channel fan-out
    /// access check so open channels stay zero-cost. Invalidated on a flip.
    pub channel_visibility_cache: Arc<moka::sync::Cache<(CommunityId, Uuid), String>>,

    /// Bounded channel for audit logging — backpressure instead of unbounded spawns.
    /// Uses .send().await (blocks caller if full) because audit entries must not be lost.
    pub audit_tx: mpsc::Sender<buzz_audit::NewAuditEntry>,
    /// Media storage client (S3/MinIO).
    pub media_storage: Arc<MediaStorage>,
    /// Git object-store backend (content-addressed packs/manifests plus
    /// CAS-guarded manifest pointer). This is the durable git source of truth;
    /// see `api::git::store` and `docs/git-on-object-storage.md`.
    pub git_store: crate::api::git::store::GitStore,
    /// Audio relay room manager — tracks active huddle audio rooms.
    pub audio_rooms: Arc<AudioRoomManager>,
    /// Set to `true` on SIGTERM — readiness probe returns 503.
    pub shutting_down: Arc<AtomicBool>,
    /// Process start time — used by `/_status` endpoint.
    pub started_at: Instant,
    /// Shared, community-scoped NIP-98 replay prevention.
    ///
    /// Correctness boundary for stateless workers: every pod must consult the
    /// same Redis `SET NX EX` seen-set, keyed by resolved community. Do not
    /// replace this with process-local caching; replay freshness must survive
    /// cross-pod routing.
    pub nip98_replay: Arc<dyn Nip98ReplayGuard>,
    /// Shared Redis-backed admission limits for ordinary HTTP and WebSocket work.
    pub admission_rate_limiter: Arc<RedisRateLimiter>,

    /// Per-agent sliding-window rate limiter for observer frames (kind 24200).
    /// Key: (community_id, agent pubkey bytes). Value: (count, window_start).
    /// 100 events/sec per agent — prevents relay/DB pressure from bursty telemetry.
    pub observer_rate_limiter: Arc<ScopedRateLimiter>,
    /// Per-uploader sliding-window rate limiter for media upload starts.
    /// Key: (community_id, uploader pubkey bytes). Value: (count, window_start).
    pub media_upload_rate_limiter: Arc<ScopedRateLimiter>,
    /// Per-claimer fixed-window rate limiter for invite claim attempts
    /// (`POST /api/invites/claim`). Entries expire after the claim window and
    /// the cache has a hard capacity because pre-membership callers can cheaply
    /// generate fresh Nostr keys.
    pub invite_claim_rate_limiter:
        Arc<moka::sync::Cache<ScopedPubkeyKey, Arc<std::sync::atomic::AtomicU32>>>,
    /// Current in-flight media uploads per (community, uploader pubkey).
    pub media_uploads_in_flight: Arc<DashMap<ScopedPubkeyKey, u32>>,
    /// Cache for observer agent-owner authorization (kind 24200).
    /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
    /// `agent_owner_pubkey` is immutable inside one community, so a long TTL
    /// (5 min) is safe once the community label is part of the key.
    /// Prevents repeated DB lookups from bursty observer traffic.
    #[allow(clippy::type_complexity)]
    pub observer_owner_cache: Arc<moka::sync::Cache<(CommunityId, Vec<u8>, Vec<u8>), bool>>,

    /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`]
    /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to
    /// record traces for replay against `docs/spec/MultiTenantRelay.tla`.
    /// See `crates/buzz-conformance/` and `crate::conformance` for the
    /// schema, emitter helpers, and the independent checker.
    pub tracer: Arc<dyn buzz_conformance::Tracer>,

    /// Inter-relay mesh handle, set once by `main.rs` after `mesh_boot` (never
    /// a constructor parameter, so `AppState::new` call sites are untouched).
    /// `None`/unset ⇒ mesh-off / single-instance: consumers must behave
    /// byte-identically to a relay without the mesh. Access via
    /// [`AppState::mesh`].
    pub mesh: Arc<std::sync::OnceLock<crate::mesh_boot::MeshHandle>>,
}

impl AppState {
    /// Constructs `AppState` from its component services.
    ///
    /// Returns `(state, audit_shutdown)`. The caller should call
    /// `audit_shutdown.drain().await` during graceful shutdown so queued
    /// audit entries are flushed before the process exits.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        config: Config,
        db: Db,
        redis_pool: deadpool_redis::Pool,
        audit: AuditService,
        pubsub: Arc<PubSubManager>,
        auth: AuthService,
        search: SearchService,
        workflow_engine: Arc<WorkflowEngine>,
        relay_keypair: nostr::Keys,
        media_storage: MediaStorage,
    ) -> (Self, AuditShutdownHandle) {
        let max_connections = config.max_connections;
        let max_concurrent_handlers = config.max_concurrent_handlers;
        let search_arc = Arc::new(search);

        let audit_arc = Arc::new(audit);
        let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
        let audit_for_worker = Arc::clone(&audit_arc);
        let audit_cancel = CancellationToken::new();
        let audit_cancel_worker = audit_cancel.clone();
        let audit_worker_handle = tokio::spawn(async move {
            // Normal operation: process entries as they arrive.
            loop {
                tokio::select! {
                    entry = audit_rx.recv() => {
                        match entry {
                            Some(entry) => log_audit_entry(&audit_for_worker, entry).await,
                            None => break, // channel closed
                        }
                    }
                    _ = audit_cancel_worker.cancelled() => {
                        // Close the receiver: rejects future sends and lets us
                        // drain everything already buffered without a race.
                        audit_rx.close();
                        break;
                    }
                }
            }
            // Drain: recv() returns buffered entries, then None once empty.
            let mut drained = 0u32;
            while let Some(entry) = audit_rx.recv().await {
                log_audit_entry(&audit_for_worker, entry).await;
                drained += 1;
            }
            if drained > 0 {
                tracing::info!(drained, "audit worker flushed remaining entries");
            }
            tracing::warn!("audit log worker exited (expected on shutdown)");
        });

        let git_max_concurrent_ops = config.git_max_concurrent_ops;
        let media_max_concurrent_uploads = config.media_max_concurrent_uploads;
        let git_store = crate::api::git::store::GitStore::new(
            &config.media.s3_endpoint,
            &config.media.s3_access_key,
            &config.media.s3_secret_key,
            &config.media.s3_bucket,
            &config.media.s3_region,
        )
        .expect("media storage was already constructed with this S3 config");
        let nip98_replay: Arc<dyn Nip98ReplayGuard> =
            Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
        let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
        let state = Self {
            config: Arc::new(config),
            db,
            redis_pool,
            audit: audit_arc,
            pubsub,
            auth: Arc::new(auth),
            search: search_arc,
            sub_registry: Arc::new(SubscriptionRegistry::new()),
            conn_manager: Arc::new(ConnectionManager::new()),
            community_connections: Arc::new(CommunityConnectionRegistry::new()),
            community_revalidator_cancel: CancellationToken::new(),
            community_disconnect_publish_attempts: Arc::new(AtomicU64::new(0)),
            conn_semaphore: Arc::new(Semaphore::new(max_connections)),
            handler_semaphore: Arc::new(Semaphore::new(max_concurrent_handlers)),
            git_semaphore: Arc::new(Semaphore::new(git_max_concurrent_ops)),
            media_upload_semaphore: Arc::new(Semaphore::new(media_max_concurrent_uploads)),
            workflow_engine,
            relay_keypair,

            local_event_ids: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(10_000)
                    .time_to_live(std::time::Duration::from_secs(60))
                    .build(),
            ),
            membership_cache: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(10_000)
                    .time_to_live(std::time::Duration::from_secs(10))
                    .support_invalidation_closures()
                    .build(),
            ),
            accessible_channels_cache: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(10_000)
                    .time_to_live(std::time::Duration::from_secs(10))
                    .support_invalidation_closures()
                    .build(),
            ),
            channel_visibility_cache: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(10_000)
                    .time_to_live(std::time::Duration::from_secs(10))
                    .support_invalidation_closures()
                    .build(),
            ),
            audit_tx,
            media_storage: Arc::new(media_storage),
            git_store,
            audio_rooms: Arc::new(AudioRoomManager::new()),
            shutting_down: Arc::new(AtomicBool::new(false)),
            started_at: Instant::now(),
            nip98_replay,
            admission_rate_limiter,
            observer_rate_limiter: Arc::new(DashMap::new()),
            media_upload_rate_limiter: Arc::new(DashMap::new()),
            invite_claim_rate_limiter: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(crate::api::invites::CLAIM_RATE_CACHE_CAPACITY)
                    .time_to_live(crate::api::invites::CLAIM_RATE_WINDOW)
                    .build(),
            ),
            media_uploads_in_flight: Arc::new(DashMap::new()),
            observer_owner_cache: Arc::new(
                moka::sync::Cache::builder()
                    .max_capacity(1_000)
                    .time_to_live(std::time::Duration::from_secs(300))
                    .build(),
            ),
            // Default to NoopTracer: production builds pay zero cost.
            // Conformance tests overwrite this with a JsonlTracer after
            // construction (see test helpers in
            // `crates/buzz-test-client` once those land).
            tracer: Arc::new(crate::conformance::NoopTracer),
            mesh: Arc::new(std::sync::OnceLock::new()),
        };
        (
            state,
            AuditShutdownHandle {
                cancel: audit_cancel,
                handle: audit_worker_handle,
            },
        )
    }

    /// Inter-relay mesh handle. `None` ⇒ mesh-off / single-instance: callers
    /// must no-op to today's behavior. Set once by `main.rs` after boot.
    pub fn mesh(&self) -> Option<&crate::mesh_boot::MeshHandle> {
        self.mesh.get()
    }

    /// Record an event ID as locally-published for dedup, scoped to the
    /// community it was fanned out in. Called before Redis publish so the
    /// multi-node consumer can skip the echo for *this* community only — a
    /// same-id event in another community is a distinct delivery and must not
    /// be suppressed.
    pub fn mark_local_event(&self, community: CommunityId, event_id: &nostr::EventId) {
        self.local_event_ids
            .insert((community, event_id.to_bytes()), ());
    }

    /// Check channel membership with a 10-second cache. Falls back to DB on miss.
    pub async fn is_member_cached(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        pubkey: &[u8],
    ) -> Result<bool, buzz_db::DbError> {
        let key = (community_id, channel_id, pubkey.to_vec());
        if let Some(cached) = self.membership_cache.get(&key) {
            metrics::counter!("buzz_membership_cache_hits_total").increment(1);
            return Ok(cached);
        }
        metrics::counter!("buzz_membership_cache_misses_total").increment(1);
        let result = self.db.is_member(community_id, channel_id, pubkey).await?;
        self.membership_cache.insert(key, result);
        Ok(result)
    }

    /// Invalidate caches after a membership change (add/remove member).
    ///
    /// Drops the local moka entries AND fire-and-forget publishes the same drop
    /// to every other pod over Redis (see [`apply_cache_invalidation`]). The
    /// publish is spawned, not awaited: the local drop is already done, and a
    /// dropped publish is backstopped by the REQ denial-path DB confirmation.
    pub fn invalidate_membership(&self, tenant: &TenantContext, channel_id: Uuid, pubkey: &[u8]) {
        self.invalidate_membership_local(tenant.community(), channel_id, pubkey);
        self.spawn_cache_invalidation(
            tenant,
            CacheInvalidation::Membership {
                channel_id,
                pubkey: pubkey.to_vec(),
            },
        );
    }

    /// Local-only membership drop. The cross-pod consumer calls this directly so
    /// applying a received drop never re-publishes it.
    pub(crate) fn invalidate_membership_local(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        pubkey: &[u8],
    ) {
        self.membership_cache
            .invalidate(&(community_id, channel_id, pubkey.to_vec()));
        self.accessible_channels_cache
            .invalidate(&(community_id, pubkey.to_vec()));
    }

    /// Invalidate all users' accessible-channels cache (e.g. new open channel created).
    pub fn invalidate_all_accessible_channels(&self, tenant: &TenantContext) {
        self.invalidate_all_accessible_channels_local(tenant.community());
        self.spawn_cache_invalidation(tenant, CacheInvalidation::AccessibleAll);
    }

    /// Local-only accessible-channels drop. See [`invalidate_membership_local`].
    pub(crate) fn invalidate_all_accessible_channels_local(&self, community_id: CommunityId) {
        if let Err(error) = self
            .accessible_channels_cache
            .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id)
        {
            // AppState enables invalidation closures at construction time. If
            // that invariant ever regresses, prefer over-invalidating to
            // serving stale access state.
            tracing::error!(
                ?error,
                "community-scoped accessible-channel invalidation unavailable; falling back to full invalidation"
            );
            self.accessible_channels_cache.invalidate_all();
        }
    }

    /// Invalidate the cached visibility for a single channel (e.g. after a flip).
    pub fn invalidate_channel_visibility(&self, tenant: &TenantContext, channel_id: Uuid) {
        self.invalidate_channel_visibility_local(tenant.community(), channel_id);
        self.spawn_cache_invalidation(tenant, CacheInvalidation::Visibility { channel_id });
    }

    /// Local-only visibility drop. See [`invalidate_membership_local`].
    pub(crate) fn invalidate_channel_visibility_local(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
    ) {
        self.channel_visibility_cache
            .invalidate(&(community_id, channel_id));
    }

    /// Invalidate all caches after a channel is deleted.
    ///
    /// Channel deletion is a rare admin operation, but it is still tenant-local:
    /// a deletion in A must not flush B's cache entries. Predicate invalidation
    /// keeps the safety property that stale `is_member=true` entries for the
    /// deleted channel are removed without turning the cache drop into a
    /// cross-community signal.
    pub fn invalidate_channel_deleted(&self, tenant: &TenantContext) {
        self.invalidate_channel_deleted_local(tenant.community());
        self.spawn_cache_invalidation(tenant, CacheInvalidation::ChannelDeleted);
    }

    /// Local-only channel-deleted drop. See [`invalidate_membership_local`].
    pub(crate) fn invalidate_channel_deleted_local(&self, community_id: CommunityId) {
        if let Err(error) =
            self.membership_cache
                .invalidate_entries_if(move |(entry_community, _, _), _| {
                    *entry_community == community_id
                })
        {
            tracing::error!(
                ?error,
                "community-scoped membership invalidation unavailable; falling back to full invalidation"
            );
            self.membership_cache.invalidate_all();
        }
        if let Err(error) = self
            .accessible_channels_cache
            .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id)
        {
            tracing::error!(
                ?error,
                "community-scoped accessible-channel invalidation unavailable; falling back to full invalidation"
            );
            self.accessible_channels_cache.invalidate_all();
        }
        if let Err(error) = self
            .channel_visibility_cache
            .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id)
        {
            tracing::error!(
                ?error,
                "community-scoped visibility invalidation unavailable; falling back to full invalidation"
            );
            self.channel_visibility_cache.invalidate_all();
        }
    }

    /// Fire-and-forget publish of a cache-key drop to all other pods. Failures
    /// are logged and swallowed — the REQ denial-path DB confirmation is the
    /// backstop, so a missed publish degrades to a <=10s TTL wait, never a leak.
    fn spawn_cache_invalidation(&self, tenant: &TenantContext, invalidation: CacheInvalidation) {
        let pubsub = Arc::clone(&self.pubsub);
        let tenant = tenant.clone();
        tokio::spawn(async move {
            if let Err(e) = pubsub
                .publish_cache_invalidation(&tenant, &invalidation)
                .await
            {
                tracing::warn!("Failed to publish cache invalidation {invalidation:?}: {e}");
            }
        });
    }

    /// Apply a cache-key drop received from another pod. Calls the local-only
    /// drop variants so a received drop is never re-published (no fan-out loop).
    pub fn apply_cache_invalidation(
        &self,
        community_id: CommunityId,
        invalidation: CacheInvalidation,
    ) {
        match invalidation {
            CacheInvalidation::Membership { channel_id, pubkey } => {
                self.invalidate_membership_local(community_id, channel_id, &pubkey);
            }
            CacheInvalidation::AccessibleAll => {
                self.invalidate_all_accessible_channels_local(community_id);
            }
            CacheInvalidation::Visibility { channel_id } => {
                self.invalidate_channel_visibility_local(community_id, channel_id);
            }
            CacheInvalidation::ChannelDeleted => {
                self.invalidate_channel_deleted_local(community_id);
            }
        }
    }

    /// Enforce a live ban cluster-wide: close this pod's sockets for `pubkey`
    /// now (fenced to `tenant`'s community) and fan the same disconnect out to
    /// every other pod over the conn-control Redis channel.
    ///
    /// This is the single entry point for live ban enforcement (decision 4:
    /// "a ban takes effect immediately, everywhere, including live sessions").
    /// Callers must not invoke the pod-local `conn_manager.disconnect_pubkey`
    /// directly — doing so closes sockets only on the pod that processed the
    /// ban and silently drops the cluster-wide half. Pairing both halves here
    /// makes that mistake unrepresentable.
    ///
    /// Returns the number of sockets closed on *this* pod only — remote pods
    /// close asynchronously and do not report back, so callers must not treat
    /// the count as cluster-wide truth. The cross-pod publish is fire-and-forget
    /// (mirrors [`Self::spawn_cache_invalidation`]): the DB ban row is the
    /// durable backstop, so a dropped publish still refuses the banned member's
    /// next auth and next write.
    pub fn disconnect_pubkey_clusterwide(
        &self,
        tenant: &TenantContext,
        pubkey: &[u8],
        event_id: &str,
        reason: &str,
    ) -> usize {
        let closed =
            self.conn_manager
                .disconnect_pubkey(tenant.community(), pubkey, event_id, reason);

        // The banning pod re-receives its own publish through the subscriber and
        // no-ops (its local sockets are already closed above) — intentional; do
        // not add origin-suppression, it buys nothing.
        let pubsub = Arc::clone(&self.pubsub);
        let tenant = tenant.clone();
        let command = ConnControl::DisconnectPubkey {
            pubkey: pubkey.to_vec(),
            event_id: event_id.to_string(),
            reason: reason.to_string(),
        };
        // This pre-existing ban path may remain fire-and-forget because the
        // durable ban row rejects the member again at auth. Community archival
        // is different: its API awaits publication and live sockets also have a
        // periodic durable-state revalidation backstop below.
        tokio::spawn(async move {
            if let Err(e) = pubsub.publish_conn_control(&tenant, &command).await {
                tracing::warn!("Failed to publish conn-control disconnect: {e}");
            }
        });

        closed
    }

    /// Disconnect a community locally and publish the command to every relay pod.
    ///
    /// Publication is awaited so the archive API can distinguish durable state
    /// from propagation completion and offer a retryable response on failure.
    pub async fn disconnect_community_clusterwide(
        &self,
        tenant: &TenantContext,
    ) -> Result<usize, buzz_pubsub::PubSubError> {
        let closed = self
            .community_connections
            .disconnect_community(tenant.community());
        self.community_disconnect_publish_attempts
            .fetch_add(1, Ordering::Relaxed);
        self.pubsub
            .publish_conn_control(tenant, &ConnControl::DisconnectCommunity)
            .await?;
        Ok(closed)
    }

    /// Revalidate all communities with live sockets and cancel inactive ones.
    ///
    /// This is the durable backstop for Redis pub/sub's lossy offline-subscriber
    /// semantics: a pod that missed a successful publish eventually observes the
    /// archived row directly.
    pub async fn revalidate_live_communities(&self) -> usize {
        let (closed, failures) =
            revalidate_registered_communities(&self.community_connections, |community_id| {
                self.db.is_community_active(community_id)
            })
            .await;
        for (community_id, error) in failures {
            tracing::warn!(%community_id, %error, "community lifecycle revalidation failed; retaining its sockets until next tick");
        }
        closed
    }

    /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss.
    pub async fn get_accessible_channel_ids_cached(
        &self,
        community_id: CommunityId,
        pubkey: &[u8],
    ) -> Result<Vec<Uuid>, buzz_db::DbError> {
        let key = (community_id, pubkey.to_vec());
        if let Some(cached) = self.accessible_channels_cache.get(&key) {
            metrics::counter!("buzz_accessible_channels_cache_hits_total").increment(1);
            return Ok(cached);
        }
        metrics::counter!("buzz_accessible_channels_cache_misses_total").increment(1);
        let result = self
            .db
            .get_accessible_channel_ids(community_id, pubkey)
            .await?;
        self.accessible_channels_cache.insert(key, result.clone());
        Ok(result)
    }

    /// Channel visibility string. Caches only `private` (10s); never caches a
    /// non-private value.
    ///
    /// The fan-out access gate fails open on a non-private result, so a stale
    /// cached `open` on another node would mask the filter for the whole TTL
    /// after an open->private flip (no cross-node cache invalidation). Caching
    /// only `private` keeps the cache fail-safe: the worst stale entry is an
    /// over-restrictive `private` (drops non-members on a now-open channel for
    /// <=10s), never a leak.
    ///
    /// `prefetched` lets a caller that already holds the channel row for this
    /// request (ingest's once-per-request fetch, E1 §4.8) reuse it instead of
    /// re-SELECTing. The gate is unchanged: a cached `private` still wins over
    /// the prefetched row (the cache is fail-safe by design), and a `private`
    /// read from the row still populates the cache. With `Some(row)` this
    /// method performs no DB I/O and cannot error.
    pub async fn channel_visibility_cached(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        prefetched: Option<&buzz_db::channel::ChannelRecord>,
    ) -> Result<String, buzz_db::DbError> {
        if let Some(cached) = self
            .channel_visibility_cache
            .get(&(community_id, channel_id))
        {
            return Ok(cached);
        }
        let visibility = match prefetched {
            Some(row) => row.visibility.clone(),
            None => {
                self.db
                    .get_channel(community_id, channel_id)
                    .await?
                    .visibility
            }
        };
        if visibility == "private" {
            self.channel_visibility_cache
                .insert((community_id, channel_id), visibility.clone());
        }
        Ok(visibility)
    }
}

/// A channel-visibility read resolved at ingest and threaded through to
/// fan-out within the same request (E1 phase-2, §4.8 phase-2 addendum).
///
/// The community and channel ids the visibility was resolved under travel
/// with the value so it can never be consulted for a different channel or
/// community's fan-out (channel UUIDs collide across communities —
/// `Inv_LabelPropagation`). Consumers must treat a missing/mismatched bundle
/// as "no threaded visibility" and fall back to a fresh fail-closed lookup —
/// never as "assume open".
#[derive(Debug, Clone)]
pub struct ThreadedChannelVisibility {
    /// Community the visibility was resolved under (server-resolved tenant).
    pub community_id: CommunityId,
    /// Channel the visibility was resolved for.
    pub channel_id: Uuid,
    /// The visibility string read at ingest (`"open"` / `"private"` / ...).
    pub visibility: String,
}

/// Handle for graceful audit worker shutdown.
///
/// Signals the worker to stop accepting new entries, drain its buffer,
/// and exit. Independent of `Arc<AppState>` lifetime — works even when
/// background tasks (reaper, pubsub, health) still hold state clones.
pub struct AuditShutdownHandle {
    cancel: CancellationToken,
    handle: JoinHandle<()>,
}

impl AuditShutdownHandle {
    /// Signal the audit worker to drain and wait up to `timeout` for it to finish.
    pub async fn drain(self, timeout: std::time::Duration) {
        self.cancel.cancel();
        match tokio::time::timeout(timeout, self.handle).await {
            Ok(Ok(())) => tracing::info!("Audit worker drained cleanly"),
            Ok(Err(e)) => tracing::error!("Audit worker panicked: {e}"),
            Err(_) => tracing::error!(
                ?timeout,
                "Audit worker did not drain in time — exiting anyway"
            ),
        }
    }
}

/// Log a single audit entry with metrics. Extracted so the normal loop
/// and the post-cancel drain share the same logic.
async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::NewAuditEntry) {
    let t = std::time::Instant::now();
    if let Err(e) = audit.log(entry).await {
        metrics::counter!("buzz_audit_log_errors_total").increment(1);
        tracing::error!("Audit log failed: {e}");
    } else {
        metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64());
    }
}

impl std::fmt::Debug for AppState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AppState")
            .field("relay_url", &self.config.relay_url)
            .field("max_connections", &self.config.max_connections)
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::connection::{AuthState, ConnectionState};
    use std::collections::HashMap;
    use tokio::sync::{Mutex, RwLock};

    /// Helper: create a ConnectionManager with one registered connection.
    /// Returns (manager, conn_id, receiver, ctrl_receiver, cancel,
    /// shared_backpressure_count).
    fn setup_conn(
        buffer_size: usize,
    ) -> (
        ConnectionManager,
        Uuid,
        mpsc::Receiver<WsMessage>,
        mpsc::Receiver<WsMessage>,
        CancellationToken,
        Arc<AtomicU8>,
    ) {
        let mgr = ConnectionManager::new();
        let conn_id = Uuid::new_v4();
        let (tx, rx) = mpsc::channel(buffer_size);
        let (ctrl_tx, ctrl_rx) = mpsc::channel(buffer_size);
        let cancel = CancellationToken::new();
        let bp = Arc::new(AtomicU8::new(0));
        mgr.register(
            conn_id,
            tx,
            ctrl_tx,
            cancel.clone(),
            buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
            Arc::clone(&bp),
            Arc::new(Mutex::new(HashMap::new())),
            3,
        );
        (mgr, conn_id, rx, ctrl_rx, cancel, bp)
    }

    async fn test_state() -> Arc<AppState> {
        let mut config = crate::config::Config::from_env().expect("default config loads");
        config.require_relay_membership = false;
        config.redis_url = "redis://127.0.0.1:1".to_string();
        let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool");
        let db = buzz_db::Db::from_pool(pool.clone());
        let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
            .create_pool(Some(deadpool_redis::Runtime::Tokio1))
            .expect("redis pool");
        let pubsub = Arc::new(
            buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
                .await
                .expect("pubsub manager"),
        );
        let audit = buzz_audit::AuditService::new(pool.clone());
        let auth = buzz_auth::AuthService::new(config.auth.clone());
        let search = buzz_search::SearchService::new(pool.clone());
        let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
            db.clone(),
            buzz_workflow::WorkflowConfig::default(),
        ));
        let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
        let (state, _audit_shutdown) = AppState::new(
            config,
            db,
            redis_pool,
            audit,
            pubsub,
            auth,
            search,
            workflow_engine,
            nostr::Keys::generate(),
            media_storage,
        );
        Arc::new(state)
    }

    #[test]
    fn send_to_resets_grace_counter_on_success() {
        let (mgr, id, _rx, _ctrl_rx, _cancel, bp) = setup_conn(16);
        // Simulate prior backpressure.
        bp.store(2, Ordering::Relaxed);
        assert!(mgr.send_to(id, "hello".into()));
        assert_eq!(
            bp.load(Ordering::Relaxed),
            0,
            "successful send should reset counter"
        );
    }

    #[test]
    fn send_to_increments_grace_counter_on_full() {
        // Buffer size 1 — fill it, then the next send is Full.
        let (mgr, id, _rx, _ctrl_rx, cancel, bp) = setup_conn(1);
        assert!(mgr.send_to(id, "fill".into()));
        // Buffer is now full.
        assert!(!mgr.send_to(id, "overflow-1".into()));
        assert_eq!(bp.load(Ordering::Relaxed), 1, "first overflow → count=1");
        assert!(
            !cancel.is_cancelled(),
            "should not cancel on first overflow"
        );

        assert!(!mgr.send_to(id, "overflow-2".into()));
        assert_eq!(bp.load(Ordering::Relaxed), 2);
        assert!(
            !cancel.is_cancelled(),
            "should not cancel on second overflow"
        );
    }

    #[test]
    fn send_to_cancels_after_grace_limit() {
        let (mgr, id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(1);
        assert!(mgr.send_to(id, "fill".into()));
        // Exhaust grace: 3 consecutive Full events (matches grace_limit=3 from setup_conn).
        for _ in 0..3u8 {
            mgr.send_to(id, "overflow".into());
        }
        assert!(
            cancel.is_cancelled(),
            "should cancel after grace_limit overflows"
        );
    }

    #[test]
    fn shared_counter_between_direct_and_fanout() {
        // Verify that ConnectionState::send() and ConnectionManager::send_to()
        // share the same backpressure counter via Arc<AtomicU8>.
        let conn_id = Uuid::new_v4();
        let (tx, _rx) = mpsc::channel(1);
        let (ctrl_tx, _ctrl_rx) = mpsc::channel(8);
        let cancel = CancellationToken::new();
        let bp = Arc::new(AtomicU8::new(0));

        let conn = ConnectionState {
            conn_id,
            tenant: buzz_core::tenant::TenantContext::resolved(
                buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
                "test.local".to_string(),
            ),
            remote_addr: "127.0.0.1:1234".parse().unwrap(),
            auth_state: RwLock::new(AuthState::Failed),
            subscriptions: Arc::new(Mutex::new(HashMap::new())),
            send_tx: tx.clone(),
            ctrl_tx,
            cancel: cancel.clone(),
            backpressure_count: Arc::clone(&bp),
            grace_limit: 3,
        };

        let mgr = ConnectionManager::new();
        mgr.register(
            conn_id,
            tx,
            conn.ctrl_tx.clone(),
            cancel.clone(),
            buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
            Arc::clone(&bp),
            Arc::clone(&conn.subscriptions),
            3,
        );

        // Fill the buffer via direct send.
        assert!(conn.send("fill".into()));
        // Overflow via fan-out.
        assert!(!mgr.send_to(conn_id, "overflow-fanout".into()));
        assert_eq!(
            bp.load(Ordering::Relaxed),
            1,
            "fan-out overflow increments shared counter"
        );
        // Overflow via direct send.
        assert!(!conn.send("overflow-direct".into()));
        assert_eq!(
            bp.load(Ordering::Relaxed),
            2,
            "direct overflow increments same counter"
        );
        // One more fan-out overflow → should cancel (3 consecutive).
        mgr.send_to(conn_id, "overflow-final".into());
        assert!(
            cancel.is_cancelled(),
            "shared counter reached limit via mixed path"
        );
    }

    #[tokio::test]
    async fn tracks_connections_by_authenticated_pubkey_within_community() {
        let mgr = ConnectionManager::new();
        let community_a = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xAAAA));
        let community_b = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xBBBB));
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let (tx_a, _rx_a) = mpsc::channel(1);
        let (ctrl_tx_a, _ctrl_rx_a) = mpsc::channel(1);
        let (tx_b, _rx_b) = mpsc::channel(1);
        let (ctrl_tx_b, _ctrl_rx_b) = mpsc::channel(1);
        mgr.register(
            conn_a,
            tx_a,
            ctrl_tx_a,
            CancellationToken::new(),
            community_a,
            Arc::new(AtomicU8::new(0)),
            Arc::new(Mutex::new(HashMap::new())),
            3,
        );
        mgr.register(
            conn_b,
            tx_b,
            ctrl_tx_b,
            CancellationToken::new(),
            community_b,
            Arc::new(AtomicU8::new(0)),
            Arc::new(Mutex::new(HashMap::new())),
            3,
        );

        let pubkey = vec![7u8; 32];
        mgr.set_authenticated_pubkey(conn_a, pubkey.clone());
        mgr.set_authenticated_pubkey(conn_b, pubkey.clone());

        assert_eq!(
            mgr.connection_ids_for_pubkey_in_community(community_a, &pubkey),
            vec![conn_a]
        );
        assert_eq!(
            mgr.connection_ids_for_pubkey_in_community(community_b, &pubkey),
            vec![conn_b]
        );
        assert!(mgr.subscriptions_for(conn_a).is_some());
        assert!(mgr.subscriptions_for(conn_b).is_some());
    }

    #[tokio::test]
    async fn pubkey_for_conn_returns_authenticated_pubkey() {
        let mgr = ConnectionManager::new();
        let conn_id = Uuid::new_v4();
        let (tx, _rx) = mpsc::channel(1);
        let (ctrl_tx, _ctrl_rx) = mpsc::channel(1);
        let cancel = CancellationToken::new();
        let bp = Arc::new(AtomicU8::new(0));
        let subscriptions = Arc::new(Mutex::new(HashMap::new()));
        mgr.register(
            conn_id,
            tx,
            ctrl_tx,
            cancel,
            buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
            bp,
            subscriptions,
            3,
        );

        assert_eq!(mgr.pubkey_for_conn(conn_id), None);
        let pubkey = vec![9u8; 32];
        mgr.set_authenticated_pubkey(conn_id, pubkey.clone());
        assert_eq!(mgr.pubkey_for_conn(conn_id), Some(pubkey));
        assert_eq!(mgr.pubkey_for_conn(Uuid::new_v4()), None);
    }

    #[tokio::test]
    async fn accessible_channel_invalidation_is_scoped_to_community() {
        let state = test_state().await;
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xAAAA));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xBBBB));
        let pubkey = vec![7u8; 32];
        let channels_a = vec![Uuid::from_u128(1)];
        let channels_b = vec![Uuid::from_u128(2)];

        state
            .accessible_channels_cache
            .insert((community_a, pubkey.clone()), channels_a);
        state
            .accessible_channels_cache
            .insert((community_b, pubkey.clone()), channels_b.clone());

        state.invalidate_all_accessible_channels_local(community_a);

        assert_eq!(
            state
                .accessible_channels_cache
                .get(&(community_a, pubkey.clone())),
            None
        );
        assert_eq!(
            state
                .accessible_channels_cache
                .get(&(community_b, pubkey.clone())),
            Some(channels_b),
            "A's cache drop must not evict B's accessible-channel entry"
        );
    }

    #[tokio::test]
    async fn channel_deleted_invalidation_is_scoped_to_community() {
        let state = test_state().await;
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xAAAA));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xBBBB));
        let channel_id = Uuid::from_u128(1);
        let pubkey = vec![7u8; 32];

        for community in [community_a, community_b] {
            state
                .membership_cache
                .insert((community, channel_id, pubkey.clone()), true);
            state
                .accessible_channels_cache
                .insert((community, pubkey.clone()), vec![channel_id]);
            state
                .channel_visibility_cache
                .insert((community, channel_id), "private".to_string());
        }

        state.invalidate_channel_deleted_local(community_a);

        assert_eq!(
            state
                .membership_cache
                .get(&(community_a, channel_id, pubkey.clone())),
            None
        );
        assert_eq!(
            state
                .accessible_channels_cache
                .get(&(community_a, pubkey.clone())),
            None
        );
        assert_eq!(
            state
                .channel_visibility_cache
                .get(&(community_a, channel_id)),
            None
        );
        assert_eq!(
            state
                .membership_cache
                .get(&(community_b, channel_id, pubkey.clone())),
            Some(true)
        );
        assert_eq!(
            state
                .accessible_channels_cache
                .get(&(community_b, pubkey.clone())),
            Some(vec![channel_id])
        );
        assert_eq!(
            state
                .channel_visibility_cache
                .get(&(community_b, channel_id)),
            Some("private".to_string()),
            "A's channel deletion must not evict B's cache entries"
        );
    }

    #[test]
    fn community_lifecycle_disconnect_covers_socket_types_and_preserves_tenant_fence() {
        let registry = CommunityConnectionRegistry::new();
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xb));
        let ordinary_a = CancellationToken::new();
        let audio_a = CancellationToken::new();
        let ordinary_b = CancellationToken::new();
        let _ordinary_a_guard = registry.register(Uuid::new_v4(), community_a, ordinary_a.clone());
        let _audio_a_guard = registry.register(Uuid::new_v4(), community_a, audio_a.clone());
        let _ordinary_b_guard = registry.register(Uuid::new_v4(), community_b, ordinary_b.clone());

        assert_eq!(registry.disconnect_community(community_a), 2);
        assert!(ordinary_a.is_cancelled());
        assert!(audio_a.is_cancelled());
        assert!(!ordinary_b.is_cancelled());
    }

    #[tokio::test]
    async fn register_then_revalidate_closes_both_archive_race_orderings() {
        let registry = CommunityConnectionRegistry::new();
        let community = CommunityId::from_uuid(Uuid::from_u128(0xa));

        // Archive wins before durable revalidation: the check observes inactive
        // and the socket body never starts.
        let cancel_before = CancellationToken::new();
        let started_before = Arc::new(AtomicBool::new(false));
        let started_before_run = Arc::clone(&started_before);
        run_registered_community_connection(
            &registry,
            Uuid::new_v4(),
            community,
            cancel_before.clone(),
            || async { Ok(false) },
            move || async move { started_before_run.store(true, Ordering::SeqCst) },
        )
        .await;
        assert!(cancel_before.is_cancelled());
        assert!(!started_before.load(Ordering::SeqCst));

        // Archive wins after registration but while revalidation is paused: its
        // sweep sees the token, and even an active query result cannot start the
        // socket body afterward.
        let cancel_during = CancellationToken::new();
        let registered = Arc::new(tokio::sync::Notify::new());
        let resume = Arc::new(tokio::sync::Notify::new());
        let registered_check = Arc::clone(&registered);
        let resume_check = Arc::clone(&resume);
        let started_during = Arc::new(AtomicBool::new(false));
        let started_during_run = Arc::clone(&started_during);
        let future = run_registered_community_connection(
            &registry,
            Uuid::new_v4(),
            community,
            cancel_during.clone(),
            move || async move {
                registered_check.notify_one();
                resume_check.notified().await;
                Ok(true)
            },
            move || async move { started_during_run.store(true, Ordering::SeqCst) },
        );
        tokio::pin!(future);
        tokio::select! {
            _ = registered.notified() => {}
            _ = &mut future => panic!("revalidation should be paused"),
        }
        assert_eq!(registry.disconnect_community(community), 1);
        resume.notify_one();
        future.await;
        assert!(cancel_during.is_cancelled());
        assert!(!started_during.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn revalidation_continues_after_one_community_lookup_failure() {
        let registry = CommunityConnectionRegistry::new();
        let archived_a = CommunityId::from_uuid(Uuid::from_u128(0xa));
        let failed = CommunityId::from_uuid(Uuid::from_u128(0xb));
        let archived_c = CommunityId::from_uuid(Uuid::from_u128(0xc));
        let cancel_a = CancellationToken::new();
        let cancel_failed = CancellationToken::new();
        let cancel_c = CancellationToken::new();
        let _guard_a = registry.register(Uuid::new_v4(), archived_a, cancel_a.clone());
        let _guard_failed = registry.register(Uuid::new_v4(), failed, cancel_failed.clone());
        let _guard_c = registry.register(Uuid::new_v4(), archived_c, cancel_c.clone());

        let (closed, failures) =
            revalidate_registered_communities(&registry, |community| async move {
                if community == failed {
                    Err(buzz_db::DbError::InvalidData(
                        "injected lookup failure".into(),
                    ))
                } else {
                    Ok(false)
                }
            })
            .await;

        assert_eq!(closed, 2);
        assert!(cancel_a.is_cancelled());
        assert!(!cancel_failed.is_cancelled());
        assert!(cancel_c.is_cancelled());
        assert_eq!(failures.len(), 1);
        assert_eq!(failures[0].0, failed);
        assert_eq!(
            registry.bound_communities(),
            HashSet::from([archived_a, failed, archived_c])
        );
    }

    #[test]
    fn community_lifecycle_guard_deregisters_on_early_return() {
        let registry = CommunityConnectionRegistry::new();
        let community = CommunityId::from_uuid(Uuid::from_u128(0xa));
        let cancel = CancellationToken::new();
        let guard = registry.register(Uuid::new_v4(), community, cancel.clone());
        assert_eq!(registry.bound_communities(), HashSet::from([community]));

        drop(guard);

        assert!(registry.bound_communities().is_empty());
        assert_eq!(registry.disconnect_community(community), 0);
        assert!(!cancel.is_cancelled());
    }

    #[tokio::test]
    async fn disconnect_pubkey_closes_matching_conns_with_reason() {
        let (mgr, id, _rx, mut ctrl_rx, cancel, _bp) = setup_conn(8);
        let pubkey = vec![3u8; 32];
        mgr.set_authenticated_pubkey(id, pubkey.clone());

        // setup_conn registers the connection under the nil community.
        let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil());
        let closed = mgr.disconnect_pubkey(
            community,
            &pubkey,
            "0".repeat(64).as_str(),
            "blocked: banned",
        );

        assert_eq!(closed, 1, "the one matching connection is closed");
        assert!(
            cancel.is_cancelled(),
            "connection is cancelled (socket close)"
        );
        // The reason frame is queued on the control channel ahead of the close.
        let frame = ctrl_rx.try_recv().expect("reason frame delivered");
        match frame {
            WsMessage::Text(t) => {
                assert!(t.as_str().contains("blocked: banned"), "carries the reason");
                assert!(t.as_str().contains("false"), "is an OK false frame");
            }
            other => panic!("expected text frame, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn disconnect_pubkey_ignores_non_matching_conns() {
        let (mgr, id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8);
        mgr.set_authenticated_pubkey(id, vec![1u8; 32]);

        let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil());
        let closed = mgr.disconnect_pubkey(
            community,
            &[2u8; 32],
            "0".repeat(64).as_str(),
            "blocked: banned",
        );

        assert_eq!(closed, 0, "no connection matches a different pubkey");
        assert!(!cancel.is_cancelled(), "unrelated connection stays live");
    }

    #[tokio::test]
    async fn disconnect_pubkey_is_fenced_to_the_banning_community() {
        // Same pubkey, two live sockets in two different communities on one pod.
        // A ban in community A must close only A's socket, never B's — the
        // tenant fence on live-disconnect fan-out (B1).
        let mgr = ConnectionManager::new();
        let pubkey = vec![7u8; 32];

        let community_a = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xa));
        let community_b = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xb));

        let register = |community| {
            let conn_id = Uuid::new_v4();
            let (tx, _rx) = mpsc::channel(8);
            let (ctrl_tx, _ctrl_rx) = mpsc::channel(8);
            let cancel = CancellationToken::new();
            mgr.register(
                conn_id,
                tx,
                ctrl_tx,
                cancel.clone(),
                community,
                Arc::new(AtomicU8::new(0)),
                Arc::new(Mutex::new(HashMap::new())),
                3,
            );
            mgr.set_authenticated_pubkey(conn_id, pubkey.clone());
            cancel
        };

        let cancel_a = register(community_a);
        let cancel_b = register(community_b);

        let closed = mgr.disconnect_pubkey(
            community_a,
            &pubkey,
            "0".repeat(64).as_str(),
            "blocked: banned",
        );

        assert_eq!(closed, 1, "only the community-A socket is closed");
        assert!(cancel_a.is_cancelled(), "community-A session is closed");
        assert!(
            !cancel_b.is_cancelled(),
            "community-B session stays live — ban does not cross the tenant fence"
        );
    }
}