buzz/crates/buzz-relay/src/subscription.rs

//! Subscription registry with active WebSocket indexes for targeted fan-out.

use std::collections::{HashMap, HashSet};

use dashmap::DashMap;
use nostr::{Alphabet, Filter, Kind, SingleLetterTag};
use uuid::Uuid;

use buzz_core::{filter::filters_match, CommunityId, StoredEvent};

/// Connection identifier — a UUID assigned to each WebSocket connection.
pub type ConnId = Uuid;
/// Subscription identifier — the client-supplied string from a REQ message.
pub type SubId = String;
/// Stored subscription entry: filters paired with server-resolved community and optional channel scope.
pub type SubEntry = (Vec<Filter>, CommunityId, Option<Uuid>);

/// Index key combining a channel and event kind for O(1) fan-out lookups.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct IndexKey {
    /// The channel this key is scoped to.
    pub channel_id: Uuid,
    /// The Nostr event kind this key is scoped to.
    pub kind: Kind,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct GlobalPKindIndexKey {
    community_id: CommunityId,
    kind: Kind,
    p: String,
}

/// A removed subscription's server-resolved routing scope.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct RemovedSubscription {
    /// Server-resolved community this subscription belonged to.
    pub community_id: CommunityId,
    /// Tenant-local channel scope; `None` means the community-global topic.
    pub channel_id: Option<Uuid>,
}

/// Thread-safe registry of active subscriptions with targeted in-memory fan-out indexes.
#[derive(Debug, Default)]
pub struct SubscriptionRegistry {
    /// Maps conn_id → sub_id → (filters, community_id, channel_id).
    /// Storing scope alongside filters enables O(1) targeted index removal and
    /// gives lifecycle code the exact Redis topic to release.
    subs: DashMap<ConnId, HashMap<SubId, SubEntry>>,
    channel_kind_index: DashMap<(CommunityId, IndexKey), Vec<(ConnId, SubId)>>,
    /// Subscriptions with a channel_id but no kind filter — need to receive ALL kinds.
    channel_wildcard_index: DashMap<(CommunityId, Uuid), Vec<(ConnId, SubId)>>,
    /// Global subscriptions indexed by kind — avoids O(all_subs) scan for global events.
    global_kind_index: DashMap<(CommunityId, Kind), Vec<(ConnId, SubId)>>,
    /// Global subscriptions indexed by both kind and `#p` recipient.
    global_p_kind_index: DashMap<GlobalPKindIndexKey, Vec<(ConnId, SubId)>>,
    /// Global subscriptions with no kind filter — wildcard, receives all global events.
    global_wildcard_index: DashMap<CommunityId, Vec<(ConnId, SubId)>>,
}

impl SubscriptionRegistry {
    /// Creates a new empty registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Replaces any existing subscription with the same sub_id (NIP-01), scoped
    /// to the server-resolved community that owns the connection.
    pub fn register_scoped(
        &self,
        community_id: CommunityId,
        conn_id: ConnId,
        sub_id: SubId,
        filters: Vec<Filter>,
        channel_id: Option<Uuid>,
    ) -> Option<RemovedSubscription> {
        let removed = self.remove_subscription(conn_id, &sub_id);

        self.subs
            .entry(conn_id)
            .or_default()
            .insert(sub_id.clone(), (filters.clone(), community_id, channel_id));
        metrics::gauge!("buzz_subscriptions_active").increment(1.0);

        if let Some(ch_id) = channel_id {
            match extract_kinds_from_filters(&filters) {
                None => {
                    // At least one filter has no `kinds` constraint — wildcard,
                    // this sub wants all kinds in this channel.
                    self.channel_wildcard_index
                        .entry((community_id, ch_id))
                        .or_default()
                        .push((conn_id, sub_id.clone()));
                }
                Some(kinds) if kinds.is_empty() => {
                    // All filters had explicit empty kinds lists (`kinds: []`).
                    // Per NIP-01, `kinds: []` means "match no kinds" — this
                    // subscription will never receive any events. Do not index it
                    // anywhere; `filters_match` will reject all events at fan-out.
                }
                Some(kinds) => {
                    for kind in kinds {
                        let key = IndexKey {
                            channel_id: ch_id,
                            kind,
                        };
                        self.channel_kind_index
                            .entry((community_id, key))
                            .or_default()
                            .push((conn_id, sub_id.clone()));
                    }
                }
            }
        } else {
            // Global subscription. Fully p-constrained filters can use the
            // narrower (kind, #p) index; broader filters stay on the generic
            // kind/wildcard indexes.
            if let Some(keys) = extract_global_p_kind_index_keys(community_id, &filters) {
                for key in keys {
                    self.global_p_kind_index
                        .entry(key)
                        .or_default()
                        .push((conn_id, sub_id.clone()));
                }
            } else {
                match extract_kinds_from_filters(&filters) {
                    None => {
                        self.global_wildcard_index
                            .entry(community_id)
                            .or_default()
                            .push((conn_id, sub_id.clone()));
                    }
                    Some(kinds) if kinds.is_empty() => {}
                    Some(kinds) => {
                        for kind in kinds {
                            self.global_kind_index
                                .entry((community_id, kind))
                                .or_default()
                                .push((conn_id, sub_id.clone()));
                        }
                    }
                }
            }
        }

        removed
    }

    /// Test-only convenience wrapper preserving the original single-tenant test API.
    #[cfg(test)]
    pub fn register(
        &self,
        conn_id: ConnId,
        sub_id: SubId,
        filters: Vec<Filter>,
        channel_id: Option<Uuid>,
    ) {
        self.register_scoped(test_community(), conn_id, sub_id, filters, channel_id);
    }

    /// Remove a single subscription and clean up its index entries.
    pub fn remove_subscription(
        &self,
        conn_id: ConnId,
        sub_id: &str,
    ) -> Option<RemovedSubscription> {
        self.remove_subscription_inner(conn_id, sub_id, || {})
    }

    fn remove_subscription_inner<F>(
        &self,
        conn_id: ConnId,
        sub_id: &str,
        after_remove: F,
    ) -> Option<RemovedSubscription>
    where
        F: FnOnce(),
    {
        let mut conn_subs = self.subs.get_mut(&conn_id)?;
        let (filters, community_id, channel_id) = conn_subs.remove(sub_id)?;

        after_remove();
        self.remove_from_index(conn_id, sub_id, &filters, community_id, channel_id);
        drop(conn_subs);

        metrics::gauge!("buzz_subscriptions_active").decrement(1.0);
        Some(RemovedSubscription {
            community_id,
            channel_id,
        })
    }

    /// Remove all subscriptions for a connection and clean up index entries.
    pub fn remove_connection(&self, conn_id: ConnId) -> Vec<RemovedSubscription> {
        let mut removed = Vec::new();
        if let Some((_, conn_subs)) = self.subs.remove(&conn_id) {
            let count = conn_subs.len();
            for (sub_id, (filters, community_id, channel_id)) in &conn_subs {
                self.remove_from_index(conn_id, sub_id, filters, *community_id, *channel_id);
                removed.push(RemovedSubscription {
                    community_id: *community_id,
                    channel_id: *channel_id,
                });
            }
            metrics::gauge!("buzz_subscriptions_active").decrement(count as f64);
        }
        removed
    }

    /// Remove all subscriptions on `conn_id` scoped to `channel_id` in one community.
    pub fn remove_channel_subscriptions_scoped(
        &self,
        community_id: CommunityId,
        conn_id: ConnId,
        channel_id: Uuid,
    ) -> Vec<(SubId, RemovedSubscription)> {
        let sub_ids: Vec<SubId> = self
            .subs
            .get(&conn_id)
            .map(|conn_subs| {
                conn_subs
                    .iter()
                    .filter_map(|(sub_id, (_, sub_community_id, sub_channel_id))| {
                        (*sub_community_id == community_id && *sub_channel_id == Some(channel_id))
                            .then_some(sub_id.clone())
                    })
                    .collect()
            })
            .unwrap_or_default();

        sub_ids
            .into_iter()
            .filter_map(|sub_id| {
                let removed = self.remove_subscription(conn_id, &sub_id)?;
                Some((sub_id, removed))
            })
            .collect()
    }

    /// Test-only convenience wrapper preserving the original single-tenant test API.
    #[cfg(test)]
    pub fn remove_channel_subscriptions(&self, conn_id: ConnId, channel_id: Uuid) -> Vec<SubId> {
        self.remove_channel_subscriptions_scoped(test_community(), conn_id, channel_id)
            .into_iter()
            .map(|(sub_id, _)| sub_id)
            .collect()
    }

    /// Return the distinct connection IDs holding any subscription scoped to
    /// `channel_id` (both kind-filtered and wildcard channel subscriptions) in
    /// one server-resolved community.
    pub fn channel_subscriber_conns_scoped(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
    ) -> Vec<ConnId> {
        let mut conns: HashSet<ConnId> = HashSet::new();
        for entry in self.channel_kind_index.iter() {
            if entry.key().0 == community_id && entry.key().1.channel_id == channel_id {
                conns.extend(entry.value().iter().map(|(conn_id, _)| *conn_id));
            }
        }
        if let Some(entry) = self.channel_wildcard_index.get(&(community_id, channel_id)) {
            conns.extend(entry.value().iter().map(|(conn_id, _)| *conn_id));
        }
        conns.into_iter().collect()
    }

    /// Test-only convenience wrapper preserving the original single-tenant test API.
    #[cfg(test)]
    pub fn channel_subscriber_conns(&self, channel_id: Uuid) -> Vec<ConnId> {
        self.channel_subscriber_conns_scoped(test_community(), channel_id)
    }

    /// Return all (conn_id, sub_id) pairs whose filters match the given event in
    /// one server-resolved community.
    #[tracing::instrument(skip_all)]
    pub fn fan_out_scoped(
        &self,
        community_id: CommunityId,
        event: &StoredEvent,
    ) -> Vec<(ConnId, SubId)> {
        let mut results = Vec::new();
        let mut seen = HashSet::new();

        if let Some(channel_id) = event.channel_id {
            let key = IndexKey {
                channel_id,
                kind: event.event.kind,
            };
            if let Some(candidates) = self
                .channel_kind_index
                .get(&(community_id, key))
                .map(|entry| entry.value().clone())
            {
                for (conn_id, sub_id) in candidates {
                    self.push_match(
                        conn_id,
                        &sub_id,
                        community_id,
                        event,
                        &mut results,
                        &mut seen,
                    );
                }
            }
            // Also check wildcard (channel-only, kindless) index.
            if let Some(wildcards) = self
                .channel_wildcard_index
                .get(&(community_id, channel_id))
                .map(|entry| entry.value().clone())
            {
                for (conn_id, sub_id) in wildcards {
                    self.push_match(
                        conn_id,
                        &sub_id,
                        community_id,
                        event,
                        &mut results,
                        &mut seen,
                    );
                }
            }
        } else {
            // Global event (channel_id = None) — use global indexes for sub-linear fan-out.
            // Channel-scoped subscriptions are never in these indexes, preserving the
            // scoping invariant without an explicit skip check.
            for p in event_p_tag_values(event) {
                let key = GlobalPKindIndexKey {
                    community_id,
                    kind: event.event.kind,
                    p,
                };
                if let Some(candidates) = self
                    .global_p_kind_index
                    .get(&key)
                    .map(|entry| entry.value().clone())
                {
                    for (conn_id, sub_id) in candidates {
                        self.push_match(
                            conn_id,
                            &sub_id,
                            community_id,
                            event,
                            &mut results,
                            &mut seen,
                        );
                    }
                }
            }
            if let Some(candidates) = self
                .global_kind_index
                .get(&(community_id, event.event.kind))
                .map(|entry| entry.value().clone())
            {
                for (conn_id, sub_id) in candidates {
                    self.push_match(
                        conn_id,
                        &sub_id,
                        community_id,
                        event,
                        &mut results,
                        &mut seen,
                    );
                }
            }
            // Also check global wildcard (kindless global subs).
            if let Some(wildcards) = self
                .global_wildcard_index
                .get(&community_id)
                .map(|entry| entry.value().clone())
            {
                for (conn_id, sub_id) in wildcards {
                    self.push_match(
                        conn_id,
                        &sub_id,
                        community_id,
                        event,
                        &mut results,
                        &mut seen,
                    );
                }
            }
        }

        // NOTE: The scoping invariant is symmetric:
        // - Global subscriptions (channel_id = None) do NOT receive channel-scoped events.
        // - Channel-scoped subscriptions do NOT receive global events.
        // This prevents both directions of information leakage: channel content
        // leaking to global subscribers, and global infrastructure events (like
        // membership notifications) leaking to channel subscribers.

        results
    }

    /// Test-only convenience wrapper preserving the original single-tenant test API.
    #[cfg(test)]
    pub fn fan_out(&self, event: &StoredEvent) -> Vec<(ConnId, SubId)> {
        self.fan_out_scoped(test_community(), event)
    }

    /// Return the filters for a specific subscription, or `None` if not found.
    pub fn get_filters(&self, conn_id: ConnId, sub_id: &str) -> Option<Vec<Filter>> {
        self.subs
            .get(&conn_id)
            .and_then(|conn_subs| conn_subs.get(sub_id).map(|(filters, _, _)| filters.clone()))
    }

    /// Return the total number of active subscriptions across all connections.
    pub fn total_subscriptions(&self) -> usize {
        self.subs.iter().map(|e| e.value().len()).sum()
    }

    /// Return the total number of connections with at least one active subscription.
    pub fn total_connections(&self) -> usize {
        self.subs.len()
    }

    /// Snapshot the number of active subscriptions per community.
    ///
    /// Used by the usage poller to emit `buzz_community_subscriptions{community}`.
    /// Snapshotting avoids gauge drift from mismatched inc/dec across communities.
    pub fn per_community_subscriptions(&self) -> std::collections::HashMap<CommunityId, u64> {
        let mut counts: std::collections::HashMap<CommunityId, u64> =
            std::collections::HashMap::new();
        for conn_entry in self.subs.iter() {
            for (_, community_id, _) in conn_entry.value().values() {
                *counts.entry(*community_id).or_default() += 1;
            }
        }
        counts
    }

    fn push_match(
        &self,
        conn_id: ConnId,
        sub_id: &str,
        community_id: CommunityId,
        event: &StoredEvent,
        results: &mut Vec<(ConnId, SubId)>,
        seen: &mut HashSet<(ConnId, SubId)>,
    ) {
        if let Some(conn_subs) = self.subs.get(&conn_id) {
            if let Some((filters, sub_community_id, sub_channel_id)) = conn_subs.get(sub_id) {
                // Candidate snapshots can become stale while a same-ID replacement
                // moves the subscription. Re-check its authoritative scope before
                // matching so an old index entry cannot deliver across scopes.
                if *sub_community_id == community_id
                    && *sub_channel_id == event.channel_id
                    && filters_match(filters, event)
                {
                    let entry = (conn_id, sub_id.to_string());
                    if seen.insert(entry.clone()) {
                        results.push(entry);
                    }
                }
            }
        }
    }

    /// Removes a subscription from the channel_kind_index (or channel_wildcard_index) using
    /// targeted O(k) lookup where k = number of kinds in the filters, instead of O(n) full-scan.
    fn remove_from_index(
        &self,
        conn_id: ConnId,
        sub_id: &str,
        filters: &[Filter],
        community_id: CommunityId,
        channel_id: Option<Uuid>,
    ) {
        if let Some(ch_id) = channel_id {
            match extract_kinds_from_filters(filters) {
                // None = wildcard (at least one filter had no kinds constraint).
                None => {
                    // Was in wildcard index.
                    if let Some(mut entries) =
                        self.channel_wildcard_index.get_mut(&(community_id, ch_id))
                    {
                        entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                        if entries.is_empty() {
                            drop(entries);
                            self.channel_wildcard_index.remove(&(community_id, ch_id));
                        }
                    }
                }
                Some(kinds) if kinds.is_empty() => {
                    // `kinds: []` subscriptions are never indexed (they match nothing),
                    // so there is nothing to remove here.
                }
                Some(kinds) => {
                    // Was in kind-specific index.
                    for kind in kinds {
                        let key = IndexKey {
                            channel_id: ch_id,
                            kind,
                        };
                        if let Some(mut entries) = self
                            .channel_kind_index
                            .get_mut(&(community_id, key.clone()))
                        {
                            entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                            if entries.is_empty() {
                                drop(entries);
                                self.channel_kind_index.remove(&(community_id, key));
                            }
                        }
                    }
                }
            }
        } else {
            // Global subscription — remove from the same global index chosen at registration.
            if let Some(keys) = extract_global_p_kind_index_keys(community_id, filters) {
                for key in keys {
                    if let Some(mut entries) = self.global_p_kind_index.get_mut(&key) {
                        entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                        if entries.is_empty() {
                            drop(entries);
                            self.global_p_kind_index.remove(&key);
                        }
                    }
                }
            } else {
                match extract_kinds_from_filters(filters) {
                    None => {
                        if let Some(mut entries) = self.global_wildcard_index.get_mut(&community_id)
                        {
                            entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                            if entries.is_empty() {
                                drop(entries);
                                self.global_wildcard_index.remove(&community_id);
                            }
                        }
                    }
                    Some(kinds) if kinds.is_empty() => {}
                    Some(kinds) => {
                        for kind in kinds {
                            if let Some(mut entries) =
                                self.global_kind_index.get_mut(&(community_id, kind))
                            {
                                entries.retain(|(cid, sid)| !(*cid == conn_id && sid == sub_id));
                                if entries.is_empty() {
                                    drop(entries);
                                    self.global_kind_index.remove(&(community_id, kind));
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}

fn p_tag() -> SingleLetterTag {
    SingleLetterTag::lowercase(Alphabet::P)
}

fn extract_global_p_kind_index_keys(
    community_id: CommunityId,
    filters: &[Filter],
) -> Option<Vec<GlobalPKindIndexKey>> {
    let mut seen = HashSet::new();
    let mut keys = Vec::new();
    let p_tag = p_tag();

    for filter in filters {
        let kinds = filter.kinds.as_ref()?;
        if kinds.is_empty() {
            continue;
        }

        let p_values = filter.generic_tags.get(&p_tag)?;
        if p_values.is_empty() {
            return None;
        }

        for kind in kinds {
            for p in p_values {
                let key = GlobalPKindIndexKey {
                    community_id,
                    kind: *kind,
                    p: p.clone(),
                };
                if seen.insert(key.clone()) {
                    keys.push(key);
                }
            }
        }
    }

    Some(keys)
}

fn event_p_tag_values(event: &StoredEvent) -> Vec<String> {
    let mut seen = HashSet::new();
    let mut values = Vec::new();
    for tag in event.event.tags.iter() {
        if tag.kind().to_string() != "p" {
            continue;
        }
        if let Some(value) = tag.content() {
            let value = value.to_string();
            if seen.insert(value.clone()) {
                values.push(value);
            }
        }
    }
    values
}

/// Returns the union of all `kinds` across filters, or `None` if any filter
/// lacks a `kinds` array (meaning that filter matches all kinds — wildcard).
///
/// NIP-01 OR semantics: a subscription with multiple filters is satisfied when
/// *any* filter matches. If one filter has no `kinds` constraint it matches
/// every kind, making the whole subscription a wildcard regardless of the other
/// filters.
fn extract_kinds_from_filters(filters: &[Filter]) -> Option<Vec<Kind>> {
    let mut seen = std::collections::HashSet::new();
    let mut kinds = Vec::new();
    for f in filters {
        match &f.kinds {
            Some(filter_kinds) => {
                for k in filter_kinds {
                    if seen.insert(*k) {
                        kinds.push(*k);
                    }
                }
            }
            None => {
                // At least one filter has no kind constraint — the whole
                // subscription is a wildcard.
                return None;
            }
        }
    }
    Some(kinds)
}

#[cfg(test)]
fn test_community() -> CommunityId {
    CommunityId::from_uuid(Uuid::nil())
}

#[cfg(test)]
mod tests {
    use super::*;
    use buzz_core::StoredEvent;
    use chrono::Utc;
    use nostr::{EventBuilder, Keys, Kind, Tag};
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    fn make_stored_event(kind: Kind, channel_id: Option<Uuid>) -> StoredEvent {
        let keys = Keys::generate();
        let event = EventBuilder::new(kind, "test")
            .tags([])
            .sign_with_keys(&keys)
            .expect("sign");
        StoredEvent::with_received_at(event, Utc::now(), channel_id, true)
    }

    fn make_stored_event_with_p(kind: Kind, p: &str, channel_id: Option<Uuid>) -> StoredEvent {
        let keys = Keys::generate();
        let event = EventBuilder::new(kind, "test")
            .tags([Tag::parse(["p", p]).expect("valid p tag")])
            .sign_with_keys(&keys)
            .expect("sign");
        StoredEvent::with_received_at(event, Utc::now(), channel_id, true)
    }

    #[test]
    fn test_subscription_registry_register_and_fan_out() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();
        let sub_id = "sub1".to_string();

        let filters = vec![Filter::new().kind(Kind::TextNote)];
        registry.register(conn_id, sub_id.clone(), filters, Some(channel_id));

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event);
        assert_eq!(matches.len(), 1);
        assert_eq!(matches[0].0, conn_id);
        assert_eq!(matches[0].1, sub_id);
    }

    #[test]
    fn test_subscription_registry_remove() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();
        let sub_id = "sub1".to_string();

        let filters = vec![Filter::new().kind(Kind::TextNote)];
        registry.register(conn_id, sub_id.clone(), filters, Some(channel_id));

        registry.remove_subscription(conn_id, &sub_id);

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event);
        assert!(matches.is_empty());
    }

    #[test]
    fn test_subscription_removal_cannot_delete_replacement_index() {
        let registry = Arc::new(SubscriptionRegistry::new());
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();
        let sub_id = "same-id".to_string();
        let filters = vec![Filter::new().kind(Kind::TextNote)];
        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_id));

        let (removed_tx, removed_rx) = std::sync::mpsc::sync_channel(0);
        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
        let remove_registry = Arc::clone(&registry);
        let remove_sub_id = sub_id.clone();
        let remove = std::thread::spawn(move || {
            remove_registry.remove_subscription_inner(conn_id, &remove_sub_id, || {
                removed_tx.send(()).expect("signal authoritative removal");
                resume_rx.recv().expect("resume index cleanup");
            })
        });

        removed_rx.recv().expect("old subscription removed");

        let (registered_tx, registered_rx) = std::sync::mpsc::sync_channel(0);
        let register_registry = Arc::clone(&registry);
        let register_sub_id = sub_id.clone();
        let register = std::thread::spawn(move || {
            register_registry.register(conn_id, register_sub_id, filters, Some(channel_id));
            registered_tx
                .send(())
                .expect("signal replacement registration");
        });

        let replacement_finished_early = registered_rx
            .recv_timeout(Duration::from_millis(100))
            .is_ok();
        resume_tx.send(()).expect("resume old cleanup");
        remove.join().expect("removal thread completes");
        if !replacement_finished_early {
            registered_rx
                .recv_timeout(Duration::from_secs(1))
                .expect("replacement registration completes");
        }
        register.join().expect("registration thread completes");
        assert!(
            !replacement_finished_early,
            "replacement must wait until old index cleanup is complete"
        );

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        assert_eq!(
            registry.fan_out(&event),
            vec![(conn_id, sub_id)],
            "replacement must remain reachable through its index"
        );
    }

    #[test]
    fn test_stale_candidate_snapshot_does_not_cross_subscription_scope() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_a = Uuid::new_v4();
        let channel_b = Uuid::new_v4();
        let sub_id = "same-id".to_string();
        let filters = vec![Filter::new().kind(Kind::TextNote)];
        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_a));

        // Reproduce fan-out's unlocked candidate snapshot, then move the same
        // subscription ID before the authoritative subscription lookup.
        let key = IndexKey {
            channel_id: channel_a,
            kind: Kind::TextNote,
        };
        let candidates = registry
            .channel_kind_index
            .get(&(test_community(), key))
            .expect("channel A candidate exists")
            .value()
            .clone();
        registry.register(conn_id, sub_id, filters, Some(channel_b));

        let event = make_stored_event(Kind::TextNote, Some(channel_a));
        let mut results = Vec::new();
        let mut seen = HashSet::new();
        for (candidate_conn_id, candidate_sub_id) in candidates {
            registry.push_match(
                candidate_conn_id,
                &candidate_sub_id,
                test_community(),
                &event,
                &mut results,
                &mut seen,
            );
        }

        assert!(
            results.is_empty(),
            "replacement on channel B received channel A event through stale snapshot"
        );
    }

    #[test]
    fn test_fan_out_concurrent_with_subscription_replacement_completes() {
        let registry = Arc::new(SubscriptionRegistry::new());
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();
        let sub_id = "sub1".to_string();
        let filters = vec![Filter::new().kind(Kind::TextNote)];
        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_id));
        let event = Arc::new(make_stored_event(Kind::TextNote, Some(channel_id)));
        let deadline = Instant::now() + Duration::from_secs(2);

        let fan_out_registry = Arc::clone(&registry);
        let fan_out_event = Arc::clone(&event);
        let fan_out = std::thread::spawn(move || {
            while Instant::now() < deadline {
                let _ = fan_out_registry.fan_out(&fan_out_event);
            }
        });

        let replace_registry = Arc::clone(&registry);
        let replace = std::thread::spawn(move || {
            while Instant::now() < deadline {
                replace_registry.register(
                    conn_id,
                    sub_id.clone(),
                    filters.clone(),
                    Some(channel_id),
                );
            }
        });

        fan_out.join().expect("fan-out thread completes");
        replace.join().expect("replacement thread completes");
    }

    #[test]
    fn test_subscription_registry_remove_connection() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );
        registry.register(
            conn_id,
            "sub2".to_string(),
            vec![Filter::new().kind(Kind::Metadata)],
            Some(channel_id),
        );

        assert_eq!(registry.total_subscriptions(), 2);

        registry.remove_connection(conn_id);

        assert_eq!(registry.total_subscriptions(), 0);
        assert_eq!(registry.total_connections(), 0);
    }

    #[test]
    fn test_subscription_registry_channel_kind_index() {
        let registry = SubscriptionRegistry::new();
        let channel_id = Uuid::new_v4();

        let mut conn_ids = Vec::new();
        for i in 0..3 {
            let conn_id = Uuid::new_v4();
            conn_ids.push(conn_id);
            registry.register(
                conn_id,
                format!("sub{i}"),
                vec![Filter::new().kind(Kind::TextNote)],
                Some(channel_id),
            );
        }

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event);
        assert_eq!(matches.len(), 3);

        let event_meta = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches_meta = registry.fan_out(&event_meta);
        assert!(matches_meta.is_empty());
    }

    #[test]
    fn test_subscription_registry_replace_existing() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new().kind(Kind::Metadata)],
            Some(channel_id),
        );

        let event1 = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches1 = registry.fan_out(&event1);
        assert!(matches1.is_empty());

        let event0 = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches0 = registry.fan_out(&event0);
        assert_eq!(matches0.len(), 1);
    }

    #[test]
    fn test_subscription_registry_no_channel_slow_path() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None, // no channel
        );

        let event = make_stored_event(Kind::TextNote, None);
        let matches = registry.fan_out(&event);
        assert_eq!(matches.len(), 1);
    }

    #[test]
    fn test_subscription_registry_get_filters() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let filters = vec![Filter::new().kind(Kind::TextNote)];

        registry.register(conn_id, "sub1".to_string(), filters.clone(), None);

        let retrieved = registry.get_filters(conn_id, "sub1");
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().len(), 1);

        let missing = registry.get_filters(conn_id, "nonexistent");
        assert!(missing.is_none());
    }

    #[test]
    fn test_remove_from_index_targeted_no_full_scan() {
        // Verify that removing a subscription only touches the relevant index keys.
        // We register subs for two different channels and two different kinds,
        // then remove one and confirm the other channel's index is untouched.
        let registry = SubscriptionRegistry::new();
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let channel_x = Uuid::new_v4();
        let channel_y = Uuid::new_v4();

        registry.register(
            conn_a,
            "sub_a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_x),
        );
        registry.register(
            conn_b,
            "sub_b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_y),
        );

        registry.remove_subscription(conn_a, "sub_a");

        let key_x = IndexKey {
            channel_id: channel_x,
            kind: Kind::TextNote,
        };
        assert!(registry
            .channel_kind_index
            .get(&(test_community(), key_x))
            .is_none());

        let key_y = IndexKey {
            channel_id: channel_y,
            kind: Kind::TextNote,
        };
        let entries = registry
            .channel_kind_index
            .get(&(test_community(), key_y))
            .expect("channel_y index intact");
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].0, conn_b);
    }

    #[test]
    fn test_kindless_channel_subscription_receives_all_kinds() {
        // A subscription with channel_id but NO kind filter should receive events
        // of any kind posted to that channel.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();
        let sub_id = "wildcard_sub".to_string();

        let filters = vec![Filter::new()]; // kindless — no .kind() constraint
        registry.register(conn_id, sub_id.clone(), filters, Some(channel_id));

        let event_text = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event_text);
        assert_eq!(matches.len(), 1, "kindless sub should receive TextNote");
        assert_eq!(matches[0].0, conn_id);
        assert_eq!(matches[0].1, sub_id);

        let event_meta = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches = registry.fan_out(&event_meta);
        assert_eq!(matches.len(), 1, "kindless sub should receive Metadata");

        let event_custom = make_stored_event(Kind::Custom(9999), Some(channel_id));
        let matches = registry.fan_out(&event_custom);
        assert_eq!(matches.len(), 1, "kindless sub should receive custom kind");

        let other_channel = Uuid::new_v4();
        let event_other = make_stored_event(Kind::TextNote, Some(other_channel));
        let matches = registry.fan_out(&event_other);
        assert!(
            matches.is_empty(),
            "kindless sub should not receive events from other channels"
        );
    }

    #[test]
    fn test_kindless_subscription_remove_cleans_wildcard_index() {
        // Verify that removing a kindless subscription cleans up the wildcard index.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        let filters = vec![Filter::new()]; // kindless
        registry.register(conn_id, "sub1".to_string(), filters, Some(channel_id));

        assert!(registry
            .channel_wildcard_index
            .get(&(test_community(), channel_id))
            .is_some());

        registry.remove_subscription(conn_id, "sub1");

        assert!(registry
            .channel_wildcard_index
            .get(&(test_community(), channel_id))
            .is_none());

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event);
        assert!(matches.is_empty());
    }

    #[test]
    fn test_kindless_and_kinded_subs_coexist() {
        // Both a kindless sub and a kind-specific sub in the same channel should
        // both receive events of the matching kind.
        let registry = SubscriptionRegistry::new();
        let conn_wildcard = Uuid::new_v4();
        let conn_kinded = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_wildcard,
            "sub_wildcard".to_string(),
            vec![Filter::new()],
            Some(channel_id),
        );

        registry.register(
            conn_kinded,
            "sub_kinded".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        let event_text = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event_text);
        assert_eq!(
            matches.len(),
            2,
            "both wildcard and kinded sub should match TextNote"
        );

        let event_meta = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches = registry.fan_out(&event_meta);
        assert_eq!(matches.len(), 1, "only wildcard sub should match Metadata");
        assert_eq!(matches[0].0, conn_wildcard);
    }

    #[test]
    fn test_kindless_subscription_replace() {
        // Replacing a kindless sub with a kinded sub should move it from wildcard
        // index to kind-specific index, and vice versa.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new()],
            Some(channel_id),
        );
        assert!(registry
            .channel_wildcard_index
            .get(&(test_community(), channel_id))
            .is_some());

        registry.register(
            conn_id,
            "sub1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        assert!(registry
            .channel_wildcard_index
            .get(&(test_community(), channel_id))
            .is_none());

        let key = IndexKey {
            channel_id,
            kind: Kind::TextNote,
        };
        assert!(registry
            .channel_kind_index
            .get(&(test_community(), key))
            .is_some());

        let event_meta = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches = registry.fan_out(&event_meta);
        assert!(matches.is_empty());

        let event_text = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event_text);
        assert_eq!(matches.len(), 1);
    }

    #[test]
    fn test_empty_kinds_array_matches_nothing() {
        // NIP-01: `kinds: []` means "match no kinds". A subscription with an
        // explicit empty kinds list should never receive any events — it should
        // NOT be treated as a wildcard (match-all).
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        let filter_empty_kinds = Filter::new().kinds(vec![] as Vec<Kind>);
        registry.register(
            conn_id,
            "sub_empty_kinds".to_string(),
            vec![filter_empty_kinds],
            Some(channel_id),
        );

        assert!(
            registry
                .channel_wildcard_index
                .get(&(test_community(), channel_id))
                .is_none(),
            "kinds:[] sub must NOT be in the wildcard index"
        );

        let key = IndexKey {
            channel_id,
            kind: Kind::TextNote,
        };
        assert!(
            registry
                .channel_kind_index
                .get(&(test_community(), key))
                .is_none(),
            "kinds:[] sub must NOT be in the kind-specific index"
        );

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&event);
        assert!(
            matches.is_empty(),
            "kinds:[] sub must not receive any events (got {:?})",
            matches
        );

        let event_meta = make_stored_event(Kind::Metadata, Some(channel_id));
        let matches = registry.fan_out(&event_meta);
        assert!(
            matches.is_empty(),
            "kinds:[] sub must not receive Metadata events"
        );
    }

    #[test]
    fn test_global_sub_does_not_receive_channel_events() {
        // Security regression test: a global subscription (channel_id = None) must
        // NOT receive events that are scoped to a channel. Doing so would bypass the
        // channel membership check and leak private channel content.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "global_sub".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None, // global — no channel scope
        );

        let channel_event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&channel_event);
        assert!(
            matches.is_empty(),
            "global sub must not receive channel-scoped events (got {:?})",
            matches
        );

        let global_event = make_stored_event(Kind::TextNote, None);
        let matches = registry.fan_out(&global_event);
        assert_eq!(
            matches.len(),
            1,
            "global sub should still receive non-channel events"
        );
        assert_eq!(matches[0].0, conn_id);
    }

    #[test]
    fn test_empty_kinds_array_remove_is_noop() {
        // Removing a kinds:[] subscription should not panic or corrupt the index.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        let filter_empty_kinds = Filter::new().kinds(vec![] as Vec<Kind>);
        registry.register(
            conn_id,
            "sub_empty".to_string(),
            vec![filter_empty_kinds],
            Some(channel_id),
        );

        registry.remove_subscription(conn_id, "sub_empty");

        assert!(registry
            .channel_wildcard_index
            .get(&(test_community(), channel_id))
            .is_none());
        let key = IndexKey {
            channel_id,
            kind: Kind::TextNote,
        };
        assert!(registry
            .channel_kind_index
            .get(&(test_community(), key))
            .is_none());
    }

    #[test]
    fn test_remove_channel_subscriptions_only_evicts_target_channel() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let channel_a = Uuid::new_v4();
        let channel_b = Uuid::new_v4();

        registry.register(
            conn_id,
            "sub-a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_a),
        );
        registry.register(
            conn_id,
            "sub-b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_b),
        );

        let removed = registry.remove_channel_subscriptions(conn_id, channel_a);
        assert_eq!(removed, vec!["sub-a".to_string()]);

        let event_a = make_stored_event(Kind::TextNote, Some(channel_a));
        assert!(registry.fan_out(&event_a).is_empty());

        let event_b = make_stored_event(Kind::TextNote, Some(channel_b));
        let matches_b = registry.fan_out(&event_b);
        assert_eq!(matches_b.len(), 1);
        assert_eq!(matches_b[0].1, "sub-b");
    }

    #[test]
    fn test_channel_subscriber_conns_dedupes_and_scopes_to_channel() {
        let registry = SubscriptionRegistry::new();
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let conn_other = Uuid::new_v4();
        let channel = Uuid::new_v4();
        let channel_other = Uuid::new_v4();

        // conn_a: a kinded + a wildcard sub on the channel — must dedupe to one entry.
        registry.register(
            conn_a,
            "kinded".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel),
        );
        registry.register(
            conn_a,
            "wildcard".to_string(),
            vec![Filter::new()],
            Some(channel),
        );
        // conn_b: kinded sub on the channel.
        registry.register(
            conn_b,
            "b".to_string(),
            vec![Filter::new().kind(Kind::Metadata)],
            Some(channel),
        );
        // conn_other: subscribed to a different channel — must be excluded.
        registry.register(
            conn_other,
            "other".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_other),
        );

        let mut conns = registry.channel_subscriber_conns(channel);
        conns.sort();
        let mut expected = vec![conn_a, conn_b];
        expected.sort();
        assert_eq!(conns, expected);
    }

    #[test]
    fn test_evict_all_subscribers_leaves_no_channel_subscriptions() {
        // Models the reaper's evict_all_channel_subscriptions: remove every
        // subscriber conn of a channel, leaving the channel with zero live subs
        // while an unrelated channel is untouched. This is the registry behavior
        // that lets an auto-archived channel stop fanning out without a storm.
        let registry = SubscriptionRegistry::new();
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let reaped = Uuid::new_v4();
        let survivor = Uuid::new_v4();

        registry.register(
            conn_a,
            "a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(reaped),
        );
        registry.register(
            conn_b,
            "b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(reaped),
        );
        registry.register(
            conn_a,
            "keep".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(survivor),
        );

        for conn_id in registry.channel_subscriber_conns(reaped) {
            registry.remove_channel_subscriptions(conn_id, reaped);
        }

        assert!(registry.channel_subscriber_conns(reaped).is_empty());
        assert!(registry
            .fan_out(&make_stored_event(Kind::TextNote, Some(reaped)))
            .is_empty());
        // Unrelated channel still fans out.
        let survivor_matches = registry.fan_out(&make_stored_event(Kind::TextNote, Some(survivor)));
        assert_eq!(survivor_matches.len(), 1);
        assert_eq!(survivor_matches[0].1, "keep");
    }

    #[test]
    fn test_global_kind_index_fan_out() {
        // Global subscriptions with explicit kinds should use the global_kind_index
        // for sub-linear fan-out instead of scanning all subs.
        let registry = SubscriptionRegistry::new();
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();

        registry.register(
            conn_a,
            "global_text".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );
        registry.register(
            conn_b,
            "global_meta".to_string(),
            vec![Filter::new().kind(Kind::Metadata)],
            None,
        );

        let event_text = make_stored_event(Kind::TextNote, None);
        let matches = registry.fan_out(&event_text);
        assert_eq!(matches.len(), 1);
        assert_eq!(matches[0].0, conn_a);

        let event_meta = make_stored_event(Kind::Metadata, None);
        let matches = registry.fan_out(&event_meta);
        assert_eq!(matches.len(), 1);
        assert_eq!(matches[0].0, conn_b);

        // Unrelated kind matches nobody.
        let event_custom = make_stored_event(Kind::Custom(9999), None);
        assert!(registry.fan_out(&event_custom).is_empty());
    }

    #[test]
    fn test_global_p_kind_index_fan_out_targets_matching_p() {
        let registry = SubscriptionRegistry::new();
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let kind = Kind::Custom(buzz_core::kind::KIND_AGENT_OBSERVER_FRAME as u16);
        let p_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        let p_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";

        registry.register(
            conn_a,
            "observer_a".to_string(),
            vec![Filter::new().kind(kind).custom_tags(p_tag(), [p_a])],
            None,
        );
        registry.register(
            conn_b,
            "observer_b".to_string(),
            vec![Filter::new().kind(kind).custom_tags(p_tag(), [p_b])],
            None,
        );

        assert!(
            registry
                .global_kind_index
                .get(&(test_community(), kind))
                .is_none(),
            "fully p-constrained global subscriptions should use the p-kind index"
        );

        let event = make_stored_event_with_p(kind, p_a, None);
        let matches = registry.fan_out(&event);
        assert_eq!(matches, vec![(conn_a, "observer_a".to_string())]);
    }

    #[test]
    fn test_global_p_kind_index_removal_cleanup() {
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();
        let kind = Kind::Custom(buzz_core::kind::KIND_AGENT_OBSERVER_FRAME as u16);
        let p = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        let filter = Filter::new().kind(kind).custom_tags(p_tag(), [p]);
        let key = GlobalPKindIndexKey {
            community_id: test_community(),
            kind,
            p: p.to_string(),
        };

        registry.register(conn_id, "observer".to_string(), vec![filter], None);
        assert!(registry.global_p_kind_index.get(&key).is_some());

        registry.remove_subscription(conn_id, "observer");
        assert!(registry.global_p_kind_index.get(&key).is_none());

        let event = make_stored_event_with_p(kind, p, None);
        assert!(registry.fan_out(&event).is_empty());
    }

    #[test]
    fn test_global_wildcard_index_fan_out() {
        // A global subscription with no kind filter should receive all global events.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();

        registry.register(
            conn_id,
            "global_wildcard".to_string(),
            vec![Filter::new()], // kindless
            None,
        );

        let event_text = make_stored_event(Kind::TextNote, None);
        let matches = registry.fan_out(&event_text);
        assert_eq!(matches.len(), 1);

        let event_meta = make_stored_event(Kind::Metadata, None);
        let matches = registry.fan_out(&event_meta);
        assert_eq!(matches.len(), 1);

        // Must NOT receive channel-scoped events.
        let channel_event = make_stored_event(Kind::TextNote, Some(Uuid::new_v4()));
        assert!(registry.fan_out(&channel_event).is_empty());
    }

    #[test]
    fn test_global_index_removal_cleanup() {
        // Removing a global subscription should clean up the global indexes.
        let registry = SubscriptionRegistry::new();
        let conn_id = Uuid::new_v4();

        // Kind-specific global sub.
        registry.register(
            conn_id,
            "g1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );
        assert!(registry
            .global_kind_index
            .get(&(test_community(), Kind::TextNote))
            .is_some());

        registry.remove_subscription(conn_id, "g1");
        assert!(registry
            .global_kind_index
            .get(&(test_community(), Kind::TextNote))
            .is_none());

        // Wildcard global sub.
        registry.register(conn_id, "g2".to_string(), vec![Filter::new()], None);
        assert!(registry
            .global_wildcard_index
            .get(&test_community())
            .is_some());

        registry.remove_subscription(conn_id, "g2");
        assert!(registry
            .global_wildcard_index
            .get(&test_community())
            .is_none());
    }

    #[test]
    fn test_global_and_channel_subs_are_isolated() {
        // Global subs must not see channel events; channel subs must not see global events.
        // This tests the invariant with the new global index in place.
        let registry = SubscriptionRegistry::new();
        let conn_global = Uuid::new_v4();
        let conn_channel = Uuid::new_v4();
        let channel_id = Uuid::new_v4();

        registry.register(
            conn_global,
            "global".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );
        registry.register(
            conn_channel,
            "channel".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        let global_event = make_stored_event(Kind::TextNote, None);
        let matches = registry.fan_out(&global_event);
        assert_eq!(matches.len(), 1);
        assert_eq!(
            matches[0].0, conn_global,
            "only global sub sees global event"
        );

        let channel_event = make_stored_event(Kind::TextNote, Some(channel_id));
        let matches = registry.fan_out(&channel_event);
        assert_eq!(matches.len(), 1);
        assert_eq!(
            matches[0].0, conn_channel,
            "only channel sub sees channel event"
        );
    }

    #[test]
    fn scoped_registry_does_not_cross_fanout_same_channel_and_kind() {
        let registry = SubscriptionRegistry::new();
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb));
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let channel_id = Uuid::from_u128(0xcccc);

        registry.register_scoped(
            community_a,
            conn_a,
            "a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );
        registry.register_scoped(
            community_b,
            conn_b,
            "b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        assert_eq!(
            registry.fan_out_scoped(community_a, &event),
            vec![(conn_a, "a".to_string())]
        );
        assert_eq!(
            registry.fan_out_scoped(community_b, &event),
            vec![(conn_b, "b".to_string())]
        );
    }

    #[test]
    fn scoped_global_registry_does_not_cross_fanout_same_kind() {
        let registry = SubscriptionRegistry::new();
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb));
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();

        registry.register_scoped(
            community_a,
            conn_a,
            "a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );
        registry.register_scoped(
            community_b,
            conn_b,
            "b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );

        let event = make_stored_event(Kind::TextNote, None);
        assert_eq!(
            registry.fan_out_scoped(community_a, &event),
            vec![(conn_a, "a".to_string())]
        );
        assert_eq!(
            registry.fan_out_scoped(community_b, &event),
            vec![(conn_b, "b".to_string())]
        );
    }

    #[test]
    fn scoped_remove_channel_subscriptions_keeps_same_channel_in_other_community() {
        let registry = SubscriptionRegistry::new();
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb));
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let channel_id = Uuid::from_u128(0xcccc);

        registry.register_scoped(
            community_a,
            conn_a,
            "a".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );
        registry.register_scoped(
            community_b,
            conn_b,
            "b".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel_id),
        );

        let removed = registry.remove_channel_subscriptions_scoped(community_a, conn_a, channel_id);
        assert_eq!(removed.len(), 1);

        let event = make_stored_event(Kind::TextNote, Some(channel_id));
        assert!(registry.fan_out_scoped(community_a, &event).is_empty());
        assert_eq!(
            registry.fan_out_scoped(community_b, &event),
            vec![(conn_b, "b".to_string())]
        );
    }

    #[test]
    fn per_community_subscriptions_snapshot_is_correctly_scoped() {
        // Verify that per_community_subscriptions() returns the correct
        // per-community counts and that removal keeps the snapshot accurate.
        let registry = SubscriptionRegistry::new();
        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb));
        let conn_a = Uuid::new_v4();
        let conn_b = Uuid::new_v4();
        let channel = Uuid::new_v4();

        // 2 subs in community A, 1 sub in community B.
        registry.register_scoped(
            community_a,
            conn_a,
            "a1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel),
        );
        registry.register_scoped(
            community_a,
            conn_a,
            "a2".to_string(),
            vec![Filter::new().kind(Kind::Metadata)],
            None,
        );
        registry.register_scoped(
            community_b,
            conn_b,
            "b1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            Some(channel),
        );

        let snap = registry.per_community_subscriptions();
        assert_eq!(snap.get(&community_a), Some(&2), "community A: 2 subs");
        assert_eq!(snap.get(&community_b), Some(&1), "community B: 1 sub");

        // Remove one sub from community A — snapshot should reflect it.
        registry.remove_subscription(conn_a, "a1");
        let snap2 = registry.per_community_subscriptions();
        assert_eq!(
            snap2.get(&community_a),
            Some(&1),
            "community A: 1 sub after removal"
        );
        assert_eq!(snap2.get(&community_b), Some(&1), "community B: unchanged");
    }

    #[test]
    fn per_community_subscriptions_drops_to_zero_when_all_subs_removed() {
        // Regression: when the last subscription for a community is removed,
        // per_community_subscriptions() must return no entry (not stale nonzero).
        // The usage poller emits one explicit zero for the previously observed
        // label before allowing its idle timeout to remove the series.
        let registry = SubscriptionRegistry::new();
        let community = CommunityId::from_uuid(Uuid::from_u128(0xcccc));
        let conn = Uuid::new_v4();

        registry.register_scoped(
            community,
            conn,
            "c1".to_string(),
            vec![Filter::new().kind(Kind::TextNote)],
            None,
        );

        // Verify nonzero before removal.
        let snap1 = registry.per_community_subscriptions();
        assert_eq!(snap1.get(&community), Some(&1), "1 sub before removal");

        // Remove the only sub — community should no longer appear in snapshot.
        registry.remove_subscription(conn, "c1");
        let snap2 = registry.per_community_subscriptions();
        assert!(
            snap2.get(&community).copied().unwrap_or(0) == 0,
            "community must have 0 subs after last removal; got {:?}",
            snap2.get(&community)
        );
    }
}