Merge pubsub community-scoped (Redis topics buzz:{community}:channel:{id})

* commit '4af7348f670f190a56eb6266afe7147f4111559c':
  test(pubsub): pin tenant-scoped topic refcounts
  feat(pubsub): scope redis topics by community

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

crates/buzz-pubsub/src/cache_invalidation.rs

@@ -11,17 +11,48 @@
 //! universal delivery-enforcement point, so dropping the stale key is
 //! sufficient: the next read re-fetches authoritative state from the DB.
 
+use buzz_core::{CommunityId, TenantContext};
 use futures_util::StreamExt;
 use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast;
 use uuid::Uuid;
 
-/// Redis pub/sub channel for cache-invalidation messages. Distinct from the
-/// `buzz:channel:*` event topic so the two streams never interfere.
-pub const CACHE_INVALIDATION_CHANNEL: &str = "buzz:cache-invalidate";
+use crate::topic::BUZZ_PREFIX;
+
+/// Tenant-local Redis pub/sub channel suffix for cache-invalidation messages.
+pub const CACHE_INVALIDATION_SUFFIX: &str = "cache-invalidate";
+
+/// Pattern used by the subscriber to receive cache invalidations for all
+/// communities this pod may have cached locally.
+pub const CACHE_INVALIDATION_PATTERN: &str = "buzz:*:cache-invalidate";
+
+/// Redis pub/sub channel for cache-invalidation messages under `ctx`.
+pub fn cache_invalidation_channel(ctx: &TenantContext) -> String {
+    format!(
+        "{BUZZ_PREFIX}:{}:{CACHE_INVALIDATION_SUFFIX}",
+        ctx.community()
+    )
+}
+
+/// Parse a cache-invalidation Redis channel into its scoped community id.
+pub fn parse_cache_invalidation_channel(channel: &str) -> Option<CommunityId> {
+    let mut parts = channel.split(':');
+    if parts.next()? != BUZZ_PREFIX {
+        return None;
+    }
+    let community_id = Uuid::parse_str(parts.next()?).ok()?;
+    if parts.next()? != CACHE_INVALIDATION_SUFFIX {
+        return None;
+    }
+    if parts.next().is_some() {
+        return None;
+    }
+    Some(CommunityId::from_uuid(community_id))
+}
 
 /// A cache-key drop to apply on every pod. Each variant mirrors exactly one of
-/// the relay's local `invalidate_*` operations.
+/// the relay's local `invalidate_*` operations. The community is carried by
+/// [`ScopedCacheInvalidation`], not by the tenant-local operation.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(tag = "op")]
 pub enum CacheInvalidation {
@@ -47,19 +78,28 @@ pub enum CacheInvalidation {
     ChannelDeleted,
 }
 
+/// A cache invalidation received from a community-scoped Redis channel.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ScopedCacheInvalidation {
+    /// Community whose local cache key should be dropped.
+    pub community_id: CommunityId,
+    /// Tenant-local cache invalidation operation.
+    pub invalidation: CacheInvalidation,
+}
+
 /// Initial reconnect backoff (1 second).
 const BACKOFF_INITIAL_SECS: u64 = 1;
 /// Maximum reconnect backoff (30 seconds).
 const BACKOFF_MAX_SECS: u64 = 30;
 
-/// Subscribes to `buzz:cache-invalidate` and forwards drops to the broadcast.
+/// Subscribes to `buzz:*:cache-invalidate` and forwards scoped drops to the broadcast.
 ///
 /// Mirrors `subscriber::run_subscriber`: a reconnect loop with exponential
 /// backoff (1s → 2s → 4s → … → 30s max). Never returns — runs for the lifetime
 /// of the relay.
 pub async fn run_cache_invalidation_subscriber(
     redis_url: String,
-    broadcast_tx: broadcast::Sender<CacheInvalidation>,
+    broadcast_tx: broadcast::Sender<ScopedCacheInvalidation>,
 ) {
     let mut backoff_secs = BACKOFF_INITIAL_SECS;
 
@@ -87,19 +127,25 @@ pub async fn run_cache_invalidation_subscriber(
 
 async fn connect_and_subscribe(
     redis_url: &str,
-    broadcast_tx: &broadcast::Sender<CacheInvalidation>,
+    broadcast_tx: &broadcast::Sender<ScopedCacheInvalidation>,
 ) -> Result<(), redis::RedisError> {
     let client = redis::Client::open(redis_url)?;
     let mut conn = client.get_async_pubsub().await?;
 
-    conn.subscribe(CACHE_INVALIDATION_CHANNEL).await?;
+    conn.psubscribe(CACHE_INVALIDATION_PATTERN).await?;
 
     tracing::info!(
-        "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_CHANNEL}"
+        "Redis cache-invalidation subscriber connected — listening on {CACHE_INVALIDATION_PATTERN}"
     );
 
     let mut stream = conn.on_message();
     while let Some(msg) = stream.next().await {
+        let channel = msg.get_channel_name();
+        let Some(community_id) = parse_cache_invalidation_channel(channel) else {
+            tracing::warn!("Received cache-invalidation message on unexpected channel: {channel}");
+            continue;
+        };
+
         let payload: String = match msg.get_payload() {
             Ok(p) => p,
             Err(e) => {
@@ -116,7 +162,12 @@ async fn connect_and_subscribe(
             }
         };
 
-        if broadcast_tx.send(invalidation).is_err() {
+        let scoped = ScopedCacheInvalidation {
+            community_id,
+            invalidation,
+        };
+
+        if broadcast_tx.send(scoped).is_err() {
             tracing::trace!("No cache-invalidation receivers — message dropped");
         }
     }
@@ -128,6 +179,46 @@ async fn connect_and_subscribe(
 mod tests {
     use super::*;
 
+    fn ctx(id: u128, host: &str) -> TenantContext {
+        TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host)
+    }
+
+    #[test]
+    fn cache_invalidation_channel_is_community_scoped() {
+        let community_a = ctx(0xaaaa, "a.example");
+        let community_b = ctx(0xbbbb, "b.example");
+
+        assert_eq!(
+            cache_invalidation_channel(&community_a),
+            format!("buzz:{}:cache-invalidate", community_a.community())
+        );
+        assert_ne!(
+            cache_invalidation_channel(&community_a),
+            cache_invalidation_channel(&community_b)
+        );
+    }
+
+    #[test]
+    fn parses_cache_invalidation_channel() {
+        let community_id = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
+        let raw = format!("buzz:{community_id}:cache-invalidate");
+
+        assert_eq!(parse_cache_invalidation_channel(&raw), Some(community_id));
+    }
+
+    #[test]
+    fn rejects_bad_cache_invalidation_channels() {
+        for raw in [
+            "buzz:cache-invalidate",
+            "buzz:not-a-uuid:cache-invalidate",
+            "not-buzz:00000000-0000-0000-0000-00000000aaaa:cache-invalidate",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:cache-invalidate:extra",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:channel:00000000-0000-0000-0000-00000000bbbb",
+        ] {
+            assert_eq!(parse_cache_invalidation_channel(raw), None);
+        }
+    }
+
     #[test]
     fn membership_roundtrips_through_json() {
         let msg = CacheInvalidation::Membership {

crates/buzz-pubsub/src/lib.rs

@@ -10,7 +10,7 @@
 //!   ├── deadpool-redis pool → PUBLISH, SET, ZADD, etc.
 //!   │
 //!   └── dedicated redis::aio::PubSub connection (NOT from pool)
-//!         └── PSUBSCRIBE buzz:channel:*
+//!         └── dynamic SUBSCRIBE buzz:{community}:channel:{id} / buzz:{community}:global
 //!               └── run_subscriber() → broadcast::channel(4096) → N WS receivers
 //! ```
 //!
@@ -35,23 +35,31 @@ pub mod publisher;
 pub mod rate_limiter;
 /// Redis SUBSCRIBE for channel event delivery.
 pub mod subscriber;
+/// Community-scoped Redis event topics.
+pub mod topic;
 /// Typing indicator tracking in Redis.
 pub use error::PubSubError;
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
+use buzz_core::TenantContext;
 use nostr::PublicKey;
-use tokio::sync::broadcast;
-use uuid::Uuid;
+use tokio::sync::{broadcast, mpsc, Mutex};
 
-use crate::cache_invalidation::{CacheInvalidation, CACHE_INVALIDATION_CHANNEL};
+use crate::cache_invalidation::{
+    cache_invalidation_channel, CacheInvalidation, ScopedCacheInvalidation,
+};
+pub use crate::topic::{channel_key, global_key, EventTopic, EventTopicKey};
 
-/// A Nostr event received on a specific channel, broadcast to local subscribers.
+/// A Nostr event received on a scoped Redis event topic, broadcast to local subscribers.
 #[derive(Debug, Clone)]
 pub struct ChannelEvent {
-    /// Channel the event belongs to.
-    pub channel_id: Uuid,
+    /// Server-resolved community that scoped the Redis topic.
+    pub community_id: buzz_core::CommunityId,
+    /// Tenant-local routing scope for this event.
+    pub topic: EventTopic,
     /// The Nostr event payload.
     pub event: nostr::Event,
 }
@@ -61,15 +69,27 @@ pub struct ChannelEvent {
 pub struct PubSubConfig {
     /// Redis connection URL (e.g. `redis://127.0.0.1:6379`).
     pub redis_url: String,
+    /// Delay before unsubscribing after the last local interest is released.
+    pub unsubscribe_debounce: Duration,
 }
 
 impl PubSubConfig {
+    /// Default delay before unsubscribing after the last local interest is released.
+    pub const DEFAULT_UNSUBSCRIBE_DEBOUNCE: Duration = Duration::from_millis(500);
+
     /// Creates a new `PubSubConfig` with the given Redis URL.
     pub fn new(redis_url: impl Into<String>) -> Self {
         Self {
             redis_url: redis_url.into(),
+            unsubscribe_debounce: Self::DEFAULT_UNSUBSCRIBE_DEBOUNCE,
         }
     }
+
+    /// Override the unsubscribe debounce delay.
+    pub fn with_unsubscribe_debounce(mut self, debounce: Duration) -> Self {
+        self.unsubscribe_debounce = debounce;
+        self
+    }
 }
 
 /// Central pub/sub manager for a Buzz relay instance.
@@ -77,19 +97,38 @@ pub struct PubSubManager {
     pool: deadpool_redis::Pool,
     /// Redis URL used by the reconnect loop to re-establish pub/sub connections.
     redis_url: String,
+    /// Delay before unsubscribing after the last local interest is released.
+    unsubscribe_debounce: Duration,
+    /// Local desired topic refcounts; source of truth across Redis reconnects.
+    desired_topics: subscriber::DesiredTopics,
+    subscription_tx: mpsc::Sender<subscriber::SubscriptionCommand>,
+    subscription_rx: Mutex<Option<mpsc::Receiver<subscriber::SubscriptionCommand>>>,
     broadcast_tx: broadcast::Sender<ChannelEvent>,
-    cache_invalidation_tx: broadcast::Sender<CacheInvalidation>,
+    cache_invalidation_tx: broadcast::Sender<ScopedCacheInvalidation>,
 }
 
 impl PubSubManager {
     /// Creates a new `PubSubManager` connected to the given Redis URL.
     pub async fn new(redis_url: &str, pool: deadpool_redis::Pool) -> Result<Self, PubSubError> {
+        Self::with_config(PubSubConfig::new(redis_url), pool).await
+    }
+
+    /// Creates a new `PubSubManager` using explicit pub/sub configuration.
+    pub async fn with_config(
+        config: PubSubConfig,
+        pool: deadpool_redis::Pool,
+    ) -> Result<Self, PubSubError> {
         let (broadcast_tx, _) = broadcast::channel(4096);
         let (cache_invalidation_tx, _) = broadcast::channel(4096);
+        let (subscription_tx, subscription_rx) = mpsc::channel(4096);
 
         Ok(Self {
             pool,
-            redis_url: redis_url.to_string(),
+            redis_url: config.redis_url,
+            unsubscribe_debounce: config.unsubscribe_debounce,
+            desired_topics: Arc::new(Mutex::new(HashMap::new())),
+            subscription_tx,
+            subscription_rx: Mutex::new(Some(subscription_rx)),
             broadcast_tx,
             cache_invalidation_tx,
         })
@@ -100,7 +139,18 @@ impl PubSubManager {
     /// Runs forever — spawn this in a background task. The loop reconnects
     /// with exponential backoff on Redis disconnect (1s → 2s → 4s → … → 30s).
     pub async fn run_subscriber(self: Arc<Self>) {
-        subscriber::run_subscriber(self.redis_url.clone(), self.broadcast_tx.clone()).await;
+        let Some(subscription_rx) = self.subscription_rx.lock().await.take() else {
+            tracing::error!("Redis pub/sub subscriber already started");
+            return;
+        };
+
+        subscriber::run_subscriber(
+            self.redis_url.clone(),
+            self.broadcast_tx.clone(),
+            self.desired_topics.clone(),
+            subscription_rx,
+        )
+        .await;
     }
 
     /// Starts the cache-invalidation subscriber loop with automatic
@@ -118,8 +168,78 @@ impl PubSubManager {
         self.broadcast_tx.subscribe()
     }
 
+    /// Retain local interest in a scoped Redis event topic.
+    ///
+    /// The first retain for a topic asks the subscriber task to `SUBSCRIBE`.
+    /// Additional retains only increment the local desired refcount.
+    pub async fn retain_topic(&self, ctx: &TenantContext, topic: EventTopic) {
+        let topic_key = EventTopicKey::from_context(ctx, topic);
+        let should_subscribe = {
+            let mut desired = self.desired_topics.lock().await;
+            let count = desired.entry(topic_key).or_insert(0);
+            let was_zero = *count == 0;
+            *count += 1;
+            was_zero
+        };
+
+        if should_subscribe {
+            let _ = self
+                .subscription_tx
+                .send(subscriber::SubscriptionCommand::Subscribe(topic_key))
+                .await;
+        }
+    }
+
+    /// Release local interest in a scoped Redis event topic.
+    ///
+    /// When the last retain is released, unsubscribe is delayed by the configured
+    /// debounce. If another retain arrives during that delay, the pending
+    /// unsubscribe becomes a no-op.
+    pub async fn release_topic(&self, ctx: &TenantContext, topic: EventTopic) {
+        let topic_key = EventTopicKey::from_context(ctx, topic);
+        let became_zero = {
+            let mut desired = self.desired_topics.lock().await;
+            let Some(count) = desired.get_mut(&topic_key) else {
+                tracing::warn!(?topic_key, "release_topic called for unretained topic");
+                return;
+            };
+
+            *count -= 1;
+            if *count == 0 {
+                desired.remove(&topic_key);
+                true
+            } else {
+                false
+            }
+        };
+
+        if became_zero {
+            let tx = self.subscription_tx.clone();
+            let debounce = self.unsubscribe_debounce;
+            tokio::spawn(async move {
+                tokio::time::sleep(debounce).await;
+                let _ = tx
+                    .send(subscriber::SubscriptionCommand::UnsubscribeIfIdle(
+                        topic_key,
+                    ))
+                    .await;
+            });
+        }
+    }
+
+    /// Current local desired refcount for tests and metrics.
+    pub async fn topic_refcount(&self, ctx: &TenantContext, topic: EventTopic) -> usize {
+        let topic_key = EventTopicKey::from_context(ctx, topic);
+        self.desired_topics
+            .lock()
+            .await
+            .get(&topic_key)
+            .copied()
+            .unwrap_or(0)
+    }
+
     /// Returns a new broadcast receiver for cross-pod cache-invalidation drops.
-    pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver<CacheInvalidation> {
+    pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver<ScopedCacheInvalidation> {
         self.cache_invalidation_tx.subscribe()
     }
 
@@ -129,12 +249,13 @@ impl PubSubManager {
     /// DB confirmation, so callers may spawn this without awaiting delivery.
     pub async fn publish_cache_invalidation(
         &self,
+        ctx: &TenantContext,
         invalidation: &CacheInvalidation,
     ) -> Result<i64, PubSubError> {
         let mut conn = self.pool.get().await?;
         let payload = serde_json::to_string(invalidation)?;
         let subscriber_count: i64 = redis::cmd("PUBLISH")
-            .arg(CACHE_INVALIDATION_CHANNEL)
+            .arg(cache_invalidation_channel(ctx))
             .arg(&payload)
             .query_async(&mut conn)
             .await?;
@@ -144,9 +265,9 @@ impl PubSubManager {
     /// Publish an event to the Redis channel. Returns subscriber count.
     ///
     /// Routing note (NIP-ER author-private reminders): events are keyed by
-    /// `channel_id` (`buzz:channel:{id}`), and every relay node's subscriber
-    /// `PSUBSCRIBE buzz:channel:*` — so the channel key is a routing label, not
-    /// an isolation boundary; every node already receives every published event.
+    /// `buzz:{community}:channel:{id}` / `buzz:{community}:global`, and
+    /// relay nodes dynamically subscribe only to topics with local interest —
+    /// so the topic key is a routing label, not an isolation boundary.
     /// Author-private reminders (kind:30300, stored under the nil channel
     /// sentinel) are therefore NOT protected by per-author Redis routing, and
     /// adding it would be pointless: the reminder's author may be connected to
@@ -158,33 +279,48 @@ impl PubSubManager {
     /// domain; the ciphertext is NIP-44-encrypted to the author regardless.
     pub async fn publish_event(
         &self,
-        channel_id: Uuid,
+        ctx: &TenantContext,
+        topic: EventTopic,
         event: &nostr::Event,
     ) -> Result<i64, PubSubError> {
-        publisher::publish_event(&self.pool, channel_id, event).await
+        publisher::publish_event(&self.pool, ctx, topic, event).await
     }
 
     /// Set presence with 60s TTL. Call on connect and every 30s heartbeat.
-    pub async fn set_presence(&self, pubkey: &PublicKey, status: &str) -> Result<(), PubSubError> {
-        presence::set_presence(&self.pool, pubkey, status).await
+    pub async fn set_presence(
+        &self,
+        ctx: &TenantContext,
+        pubkey: &PublicKey,
+        status: &str,
+    ) -> Result<(), PubSubError> {
+        presence::set_presence(&self.pool, ctx, pubkey, status).await
     }
 
     /// Remove presence for `pubkey`. Call on clean disconnect.
-    pub async fn clear_presence(&self, pubkey: &PublicKey) -> Result<(), PubSubError> {
-        presence::clear_presence(&self.pool, pubkey).await
+    pub async fn clear_presence(
+        &self,
+        ctx: &TenantContext,
+        pubkey: &PublicKey,
+    ) -> Result<(), PubSubError> {
+        presence::clear_presence(&self.pool, ctx, pubkey).await
     }
 
     /// Returns the current presence status for `pubkey`, or `None` if not set.
-    pub async fn get_presence(&self, pubkey: &PublicKey) -> Result<Option<String>, PubSubError> {
-        presence::get_presence(&self.pool, pubkey).await
+    pub async fn get_presence(
+        &self,
+        ctx: &TenantContext,
+        pubkey: &PublicKey,
+    ) -> Result<Option<String>, PubSubError> {
+        presence::get_presence(&self.pool, ctx, pubkey).await
     }
 
     /// Returns presence statuses for multiple pubkeys as a `pubkey_hex → status` map.
     pub async fn get_presence_bulk(
         &self,
+        ctx: &TenantContext,
         pubkeys: &[PublicKey],
     ) -> Result<HashMap<String, String>, PubSubError> {
-        presence::get_presence_bulk(&self.pool, pubkeys).await
+        presence::get_presence_bulk(&self.pool, ctx, pubkeys).await
     }
 }
 
@@ -201,7 +337,9 @@ pub(crate) mod test_util {
 mod tests {
     use super::*;
     use crate::test_util::make_test_pool;
+    use buzz_core::{CommunityId, TenantContext};
     use nostr::{EventBuilder, Keys, Kind};
+    use uuid::Uuid;
 
     async fn make_manager() -> Arc<PubSubManager> {
         let pool = make_test_pool();
@@ -212,6 +350,10 @@ mod tests {
         )
     }
 
+    fn ctx(id: u128, host: &str) -> TenantContext {
+        TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host)
+    }
+
     #[tokio::test]
     #[ignore = "requires Redis"]
     async fn test_publish_and_subscribe_roundtrip() {
@@ -222,6 +364,7 @@ mod tests {
         tokio::spawn(async move { manager_clone.run_subscriber().await });
         tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
 
+        let ctx = ctx(0xaaaa, "a.example");
         let channel_id = Uuid::new_v4();
         let keys = Keys::generate();
         let event = EventBuilder::new(Kind::TextNote, "hello pubsub")
@@ -231,7 +374,11 @@ mod tests {
         let event_id = event.id;
 
         manager
-            .publish_event(channel_id, &event)
+            .retain_topic(&ctx, EventTopic::Channel(channel_id))
+            .await;
+
+        manager
+            .publish_event(&ctx, EventTopic::Channel(channel_id), &event)
             .await
             .expect("publish failed");
 
@@ -240,7 +387,8 @@ mod tests {
             .expect("timeout")
             .expect("channel closed");
 
-        assert_eq!(received.channel_id, channel_id);
+        assert_eq!(received.community_id, ctx.community());
+        assert_eq!(received.topic, EventTopic::Channel(channel_id));
         assert_eq!(received.event.id, event_id);
     }
 
@@ -261,8 +409,10 @@ mod tests {
             pubkey: pubkey.clone(),
         };
 
+        let ctx = ctx(0xaaaa, "a.example");
+
         manager
-            .publish_cache_invalidation(&sent)
+            .publish_cache_invalidation(&ctx, &sent)
             .await
             .expect("publish failed");
 
@@ -271,7 +421,13 @@ mod tests {
             .expect("timeout")
             .expect("channel closed");
 
-        assert_eq!(received, sent);
+        assert_eq!(
+            received,
+            ScopedCacheInvalidation {
+                community_id: ctx.community(),
+                invalidation: sent,
+            }
+        );
     }
 
     #[tokio::test]
@@ -279,19 +435,20 @@ mod tests {
     async fn test_presence_set_and_get() {
         let pool = make_test_pool();
         let pubkey = Keys::generate().public_key();
+        let ctx = ctx(0xaaaa, "a.example");
 
-        let status = presence::get_presence(&pool, &pubkey).await.unwrap();
+        let status = presence::get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert!(status.is_none());
 
-        presence::set_presence(&pool, &pubkey, "online")
+        presence::set_presence(&pool, &ctx, &pubkey, "online")
             .await
             .unwrap();
-        let status = presence::get_presence(&pool, &pubkey).await.unwrap();
+        let status = presence::get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert_eq!(status.as_deref(), Some("online"));
 
         let mut conn = pool.get().await.unwrap();
         let ttl: i64 = redis::cmd("TTL")
-            .arg(presence::presence_key(&pubkey))
+            .arg(presence::presence_key(&ctx, &pubkey))
             .query_async(&mut conn)
             .await
             .unwrap();
@@ -301,8 +458,130 @@ mod tests {
             presence::PRESENCE_TTL_SECS
         );
 
-        presence::clear_presence(&pool, &pubkey).await.unwrap();
-        let status = presence::get_presence(&pool, &pubkey).await.unwrap();
+        presence::clear_presence(&pool, &ctx, &pubkey)
+            .await
+            .unwrap();
+        let status = presence::get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert!(status.is_none());
     }
+
+    #[tokio::test]
+    #[ignore = "requires Redis"]
+    async fn same_channel_id_in_two_communities_release_one_keeps_other_live() {
+        let pool = make_test_pool();
+        let manager = Arc::new(
+            PubSubManager::with_config(
+                PubSubConfig::new("redis://127.0.0.1:6379")
+                    .with_unsubscribe_debounce(Duration::from_millis(25)),
+                pool,
+            )
+            .await
+            .expect("Failed to create PubSubManager"),
+        );
+        let mut rx = manager.subscribe_local();
+
+        let manager_clone = manager.clone();
+        tokio::spawn(async move { manager_clone.run_subscriber().await });
+        tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
+
+        let ctx_a = ctx(0xaaaa, "a.example");
+        let ctx_b = ctx(0xbbbb, "b.example");
+        let channel_id = Uuid::from_u128(0xcccc);
+        let topic = EventTopic::Channel(channel_id);
+
+        manager.retain_topic(&ctx_a, topic).await;
+        manager.retain_topic(&ctx_b, topic).await;
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+
+        assert_eq!(manager.topic_refcount(&ctx_a, topic).await, 1);
+        assert_eq!(manager.topic_refcount(&ctx_b, topic).await, 1);
+
+        let keys = Keys::generate();
+        let event_before_release = EventBuilder::new(Kind::TextNote, "before A release")
+            .tags([])
+            .sign_with_keys(&keys)
+            .expect("signing failed");
+
+        manager
+            .publish_event(&ctx_b, topic, &event_before_release)
+            .await
+            .expect("publish before release failed");
+
+        let received_before_release =
+            tokio::time::timeout(tokio::time::Duration::from_secs(2), rx.recv())
+                .await
+                .expect("timeout before release")
+                .expect("channel closed before release");
+        assert_eq!(received_before_release.community_id, ctx_b.community());
+        assert_eq!(received_before_release.topic, topic);
+        assert_eq!(received_before_release.event.id, event_before_release.id);
+
+        manager.release_topic(&ctx_a, topic).await;
+        assert_eq!(manager.topic_refcount(&ctx_a, topic).await, 0);
+        assert_eq!(manager.topic_refcount(&ctx_b, topic).await, 1);
+
+        // Wait past A's debounce. A buggy implementation that keyed active
+        // Redis subscriptions by channel id alone would unsubscribe B here too.
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+
+        let event_after_release = EventBuilder::new(Kind::TextNote, "after A release")
+            .tags([])
+            .sign_with_keys(&keys)
+            .expect("signing failed");
+
+        manager
+            .publish_event(&ctx_b, topic, &event_after_release)
+            .await
+            .expect("publish after release failed");
+
+        let received_after_release =
+            tokio::time::timeout(tokio::time::Duration::from_secs(2), rx.recv())
+                .await
+                .expect("timeout after release")
+                .expect("channel closed after release");
+        assert_eq!(received_after_release.community_id, ctx_b.community());
+        assert_eq!(received_after_release.topic, topic);
+        assert_eq!(received_after_release.event.id, event_after_release.id);
+
+        manager.release_topic(&ctx_b, topic).await;
+        assert_eq!(manager.topic_refcount(&ctx_b, topic).await, 0);
+    }
+
+    #[tokio::test]
+    async fn retain_release_refcounts_and_debounces_last_release() {
+        let pool = make_test_pool();
+        let manager = PubSubManager::with_config(
+            PubSubConfig::new("redis://127.0.0.1:6379")
+                .with_unsubscribe_debounce(Duration::from_millis(1)),
+            pool,
+        )
+        .await
+        .unwrap();
+        let ctx = ctx(0xaaaa, "a.example");
+        let topic = EventTopic::Channel(Uuid::from_u128(0xbbbb));
+
+        assert_eq!(manager.topic_refcount(&ctx, topic).await, 0);
+
+        manager.retain_topic(&ctx, topic).await;
+        manager.retain_topic(&ctx, topic).await;
+        assert_eq!(manager.topic_refcount(&ctx, topic).await, 2);
+
+        manager.release_topic(&ctx, topic).await;
+        assert_eq!(manager.topic_refcount(&ctx, topic).await, 1);
+
+        manager.release_topic(&ctx, topic).await;
+        assert_eq!(manager.topic_refcount(&ctx, topic).await, 0);
+    }
+
+    #[test]
+    fn config_defaults_debounce_but_allows_override() {
+        let config = PubSubConfig::new("redis://example");
+        assert_eq!(
+            config.unsubscribe_debounce,
+            PubSubConfig::DEFAULT_UNSUBSCRIBE_DEBOUNCE
+        );
+
+        let config = config.with_unsubscribe_debounce(Duration::from_millis(42));
+        assert_eq!(config.unsubscribe_debounce, Duration::from_millis(42));
+    }
 }

crates/buzz-pubsub/src/presence.rs

@@ -1,31 +1,38 @@
 //! Presence tracking — online/away status with TTL.
 //!
-//! Stored as `SET buzz:presence:{pubkey_hex} "online" EX 90`.
+//! Stored as `SET buzz:{community}:presence:{pubkey_hex} "online" EX 90`.
 //! TTL is 3x the 30s heartbeat interval so a single missed heartbeat doesn't
 //! cause presence flap. Clean disconnect deletes immediately.
 
+use buzz_core::TenantContext;
 use deadpool_redis::Pool;
 use nostr::PublicKey;
 use std::collections::HashMap;
 
 use crate::error::PubSubError;
+use crate::topic::BUZZ_PREFIX;
 
 /// 3x the 30s heartbeat — single missed heartbeat won't cause presence flap.
 pub const PRESENCE_TTL_SECS: u64 = 90;
 
-/// Returns the Redis key for the presence entry of `pubkey`.
-pub fn presence_key(pubkey: &PublicKey) -> String {
-    format!("buzz:presence:{}", pubkey.to_hex())
+/// Returns the Redis key for the presence entry of `pubkey` under `ctx`.
+pub fn presence_key(ctx: &TenantContext, pubkey: &PublicKey) -> String {
+    format!(
+        "{BUZZ_PREFIX}:{}:presence:{}",
+        ctx.community(),
+        pubkey.to_hex()
+    )
 }
 
 /// Sets presence status for `pubkey` with a [`PRESENCE_TTL_SECS`]-second TTL.
 pub async fn set_presence(
     pool: &Pool,
+    ctx: &TenantContext,
     pubkey: &PublicKey,
     status: &str,
 ) -> Result<(), PubSubError> {
     let mut conn = pool.get().await?;
-    let key = presence_key(pubkey);
+    let key = presence_key(ctx, pubkey);
     redis::cmd("SET")
         .arg(&key)
         .arg(status)
@@ -37,9 +44,13 @@ pub async fn set_presence(
 }
 
 /// Removes the presence entry for `pubkey`. Call on clean disconnect.
-pub async fn clear_presence(pool: &Pool, pubkey: &PublicKey) -> Result<(), PubSubError> {
+pub async fn clear_presence(
+    pool: &Pool,
+    ctx: &TenantContext,
+    pubkey: &PublicKey,
+) -> Result<(), PubSubError> {
     let mut conn = pool.get().await?;
-    let key = presence_key(pubkey);
+    let key = presence_key(ctx, pubkey);
     redis::cmd("DEL")
         .arg(&key)
         .query_async::<()>(&mut conn)
@@ -48,9 +59,13 @@ pub async fn clear_presence(pool: &Pool, pubkey: &PublicKey) -> Result<(), PubSu
 }
 
 /// Returns the current presence status for `pubkey`, or `None` if not set or expired.
-pub async fn get_presence(pool: &Pool, pubkey: &PublicKey) -> Result<Option<String>, PubSubError> {
+pub async fn get_presence(
+    pool: &Pool,
+    ctx: &TenantContext,
+    pubkey: &PublicKey,
+) -> Result<Option<String>, PubSubError> {
     let mut conn = pool.get().await?;
-    let key = presence_key(pubkey);
+    let key = presence_key(ctx, pubkey);
     let value: Option<String> = redis::cmd("GET").arg(&key).query_async(&mut conn).await?;
     Ok(value)
 }
@@ -58,13 +73,17 @@ pub async fn get_presence(pool: &Pool, pubkey: &PublicKey) -> Result<Option<Stri
 /// Returns `pubkey_hex → status` for all currently-set keys.
 pub async fn get_presence_bulk(
     pool: &Pool,
+    ctx: &TenantContext,
     pubkeys: &[PublicKey],
 ) -> Result<HashMap<String, String>, PubSubError> {
     if pubkeys.is_empty() {
         return Ok(HashMap::new());
     }
     let mut conn = pool.get().await?;
-    let keys: Vec<String> = pubkeys.iter().map(presence_key).collect();
+    let keys: Vec<String> = pubkeys
+        .iter()
+        .map(|pubkey| presence_key(ctx, pubkey))
+        .collect();
     let values: Vec<Option<String>> = redis::cmd("MGET").arg(&keys).query_async(&mut conn).await?;
     let result = pubkeys
         .iter()
@@ -78,41 +97,62 @@ pub async fn get_presence_bulk(
 mod tests {
     use super::*;
     use crate::test_util::make_test_pool;
+    use buzz_core::{CommunityId, TenantContext};
     use nostr::Keys;
+    use uuid::Uuid;
 
     fn make_pubkey() -> PublicKey {
         Keys::generate().public_key()
     }
 
+    fn ctx(id: u128, host: &str) -> TenantContext {
+        TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host)
+    }
+
     #[test]
     fn test_presence_key_format() {
         let pubkey = make_pubkey();
-        let key = presence_key(&pubkey);
-        assert!(key.starts_with("buzz:presence:"));
-        let hex_part = key.strip_prefix("buzz:presence:").unwrap();
+        let ctx = ctx(0xaaaa, "a.example");
+        let key = presence_key(&ctx, &pubkey);
+        let prefix = format!("buzz:{}:presence:", ctx.community());
+        assert!(key.starts_with(&prefix));
+        let hex_part = key.strip_prefix(&prefix).unwrap();
         assert_eq!(hex_part.len(), 64);
         assert!(hex_part.chars().all(|c| c.is_ascii_hexdigit()));
     }
 
+    #[test]
+    fn same_pubkey_in_two_communities_has_different_presence_keys() {
+        let pubkey = make_pubkey();
+        let community_a = ctx(0xaaaa, "a.example");
+        let community_b = ctx(0xbbbb, "b.example");
+
+        assert_ne!(
+            presence_key(&community_a, &pubkey),
+            presence_key(&community_b, &pubkey)
+        );
+    }
+
     #[tokio::test]
     #[ignore = "requires Redis"]
     async fn test_presence_set_and_get() {
         let pool = make_test_pool();
         let pubkey = make_pubkey();
+        let ctx = ctx(0xaaaa, "a.example");
 
-        let status = get_presence(&pool, &pubkey).await.unwrap();
+        let status = get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert!(status.is_none());
 
-        set_presence(&pool, &pubkey, "online").await.unwrap();
-        let status = get_presence(&pool, &pubkey).await.unwrap();
+        set_presence(&pool, &ctx, &pubkey, "online").await.unwrap();
+        let status = get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert_eq!(status.as_deref(), Some("online"));
 
-        set_presence(&pool, &pubkey, "away").await.unwrap();
-        let status = get_presence(&pool, &pubkey).await.unwrap();
+        set_presence(&pool, &ctx, &pubkey, "away").await.unwrap();
+        let status = get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert_eq!(status.as_deref(), Some("away"));
 
-        clear_presence(&pool, &pubkey).await.unwrap();
-        let status = get_presence(&pool, &pubkey).await.unwrap();
+        clear_presence(&pool, &ctx, &pubkey).await.unwrap();
+        let status = get_presence(&pool, &ctx, &pubkey).await.unwrap();
         assert!(status.is_none());
     }
 
@@ -123,11 +163,14 @@ mod tests {
         let pk1 = make_pubkey();
         let pk2 = make_pubkey();
         let pk3 = make_pubkey();
+        let ctx = ctx(0xaaaa, "a.example");
 
-        set_presence(&pool, &pk1, "online").await.unwrap();
-        set_presence(&pool, &pk2, "away").await.unwrap();
+        set_presence(&pool, &ctx, &pk1, "online").await.unwrap();
+        set_presence(&pool, &ctx, &pk2, "away").await.unwrap();
 
-        let result = get_presence_bulk(&pool, &[pk1, pk2, pk3]).await.unwrap();
+        let result = get_presence_bulk(&pool, &ctx, &[pk1, pk2, pk3])
+            .await
+            .unwrap();
 
         assert_eq!(
             result.get(&pk1.to_hex()).map(|s| s.as_str()),
@@ -136,8 +179,8 @@ mod tests {
         assert_eq!(result.get(&pk2.to_hex()).map(|s| s.as_str()), Some("away"));
         assert!(!result.contains_key(&pk3.to_hex()));
 
-        clear_presence(&pool, &pk1).await.unwrap();
-        clear_presence(&pool, &pk2).await.unwrap();
+        clear_presence(&pool, &ctx, &pk1).await.unwrap();
+        clear_presence(&pool, &ctx, &pk2).await.unwrap();
     }
 
     #[tokio::test]
@@ -145,12 +188,13 @@ mod tests {
     async fn test_presence_ttl() {
         let pool = make_test_pool();
         let pubkey = make_pubkey();
+        let ctx = ctx(0xaaaa, "a.example");
 
-        set_presence(&pool, &pubkey, "online").await.unwrap();
+        set_presence(&pool, &ctx, &pubkey, "online").await.unwrap();
 
         let mut conn = pool.get().await.unwrap();
         let ttl: i64 = redis::cmd("TTL")
-            .arg(presence_key(&pubkey))
+            .arg(presence_key(&ctx, &pubkey))
             .query_async(&mut conn)
             .await
             .unwrap();
@@ -160,6 +204,6 @@ mod tests {
             "TTL should be 1-{PRESENCE_TTL_SECS}s, got {ttl}"
         );
 
-        clear_presence(&pool, &pubkey).await.unwrap();
+        clear_presence(&pool, &ctx, &pubkey).await.unwrap();
     }
 }

crates/buzz-pubsub/src/publisher.rs

@@ -1,24 +1,32 @@
 //! Event publishing — PUBLISH to Redis via pool connection.
 
+use buzz_core::TenantContext;
 use deadpool_redis::Pool;
 use nostr::JsonUtil;
 use uuid::Uuid;
 
 use crate::error::PubSubError;
+use crate::topic::{self, EventTopic};
 
-/// Returns the Redis pub/sub channel key for `channel_id`.
-pub fn channel_key(channel_id: Uuid) -> String {
-    format!("buzz:channel:{}", channel_id)
+/// Returns the Redis pub/sub channel key for `channel_id` under `ctx`.
+pub fn channel_key(ctx: &TenantContext, channel_id: Uuid) -> String {
+    topic::channel_key(ctx, channel_id)
+}
+
+/// Returns the Redis pub/sub channel key for community-global events under `ctx`.
+pub fn global_key(ctx: &TenantContext) -> String {
+    topic::global_key(ctx)
 }
 
 /// Returns the number of subscribers that received the message.
 pub async fn publish_event(
     pool: &Pool,
-    channel_id: Uuid,
+    ctx: &TenantContext,
+    topic: EventTopic,
     event: &nostr::Event,
 ) -> Result<i64, PubSubError> {
     let mut conn = pool.get().await?;
-    let key = channel_key(channel_id);
+    let key = crate::topic::EventTopicKey::from_context(ctx, topic).redis_channel();
     let payload = event.as_json();
     let subscriber_count: i64 = redis::cmd("PUBLISH")
         .arg(&key)

crates/buzz-pubsub/src/subscriber.rs

@@ -1,10 +1,14 @@
 //! Redis pub/sub subscriber — fans out messages to local WS connections via broadcast.
 
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::Duration;
+
 use futures_util::StreamExt;
 use nostr::JsonUtil;
-use tokio::sync::broadcast;
-use uuid::Uuid;
+use tokio::sync::{broadcast, mpsc, Mutex};
 
+use crate::topic::EventTopicKey;
 use crate::ChannelEvent;
 
 /// Initial reconnect backoff (1 second).
@@ -12,16 +16,41 @@ const BACKOFF_INITIAL_SECS: u64 = 1;
 /// Maximum reconnect backoff (30 seconds).
 const BACKOFF_MAX_SECS: u64 = 30;
 
-/// Pattern-subscribes to `buzz:channel:*` and forwards events to broadcast.
+/// Local desired topic refcounts, keyed by fully scoped Redis topic.
+pub(crate) type DesiredTopics = Arc<Mutex<HashMap<EventTopicKey, usize>>>;
+
+/// Commands sent from relay subscription registration/removal to the Redis
+/// pub/sub task.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum SubscriptionCommand {
+    /// Ensure the topic is subscribed on the current Redis pub/sub connection.
+    Subscribe(EventTopicKey),
+    /// Unsubscribe only if the desired refcount is still zero when processed.
+    UnsubscribeIfIdle(EventTopicKey),
+}
+
+/// Runs a dynamically scoped subscriber and forwards events to broadcast.
 ///
-/// Runs a reconnect loop with exponential backoff (1s → 2s → 4s → … → 30s max).
-/// Logs `error!` on disconnect and `info!` on successful reconnect.
-/// Never returns — the task runs for the lifetime of the relay.
-pub async fn run_subscriber(redis_url: String, broadcast_tx: broadcast::Sender<ChannelEvent>) {
+/// The desired refcount map is the source of truth. On every reconnect, this
+/// task snapshots topics with count > 0 and subscribes to those exact Redis
+/// channels before processing messages.
+pub(crate) async fn run_subscriber(
+    redis_url: String,
+    broadcast_tx: broadcast::Sender<ChannelEvent>,
+    desired_topics: DesiredTopics,
+    mut subscription_rx: mpsc::Receiver<SubscriptionCommand>,
+) {
     let mut backoff_secs = BACKOFF_INITIAL_SECS;
 
     loop {
-        match connect_and_subscribe(&redis_url, &broadcast_tx).await {
+        match connect_and_subscribe(
+            &redis_url,
+            &broadcast_tx,
+            desired_topics.clone(),
+            &mut subscription_rx,
+        )
+        .await
+        {
             Ok(()) => {
                 // Stream ended cleanly (Redis returned None). The connection was
                 // established and ran successfully, so reset backoff to the initial
@@ -34,72 +63,143 @@ pub async fn run_subscriber(redis_url: String, broadcast_tx: broadcast::Sender<C
             }
         }
 
-        tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)).await;
+        tokio::time::sleep(Duration::from_secs(backoff_secs)).await;
         backoff_secs = (backoff_secs * 2).min(BACKOFF_MAX_SECS);
 
         tracing::info!("Attempting to reconnect to Redis pub/sub...");
     }
 }
 
-/// Establish a Redis pub/sub connection, subscribe, and run the fan-out loop
-/// until the stream ends or an error occurs.
-///
-/// Returns `Ok(())` if the stream ends cleanly (disconnect), `Err` on
-/// connection or subscription failure.
+/// Establish a Redis pub/sub connection, subscribe to the current desired-topic
+/// snapshot, and run the fan-out / command loop until the connection ends.
 async fn connect_and_subscribe(
     redis_url: &str,
     broadcast_tx: &broadcast::Sender<ChannelEvent>,
+    desired_topics: DesiredTopics,
+    subscription_rx: &mut mpsc::Receiver<SubscriptionCommand>,
 ) -> Result<(), redis::RedisError> {
     let client = redis::Client::open(redis_url)?;
-    let mut conn = client.get_async_pubsub().await?;
+    let conn = client.get_async_pubsub().await?;
+    let (mut sink, mut stream) = conn.split();
+    let mut active_topics = HashSet::new();
 
-    conn.psubscribe("buzz:channel:*").await?;
+    let initial_topics: Vec<EventTopicKey> = {
+        let desired = desired_topics.lock().await;
+        desired
+            .iter()
+            .filter_map(|(topic, count)| (*count > 0).then_some(*topic))
+            .collect()
+    };
 
-    tracing::info!("Redis pub/sub subscriber connected — listening on buzz:channel:*");
-
-    // Note: backoff is NOT reset here on connect. It resets in the outer loop
-    // only after this function returns Ok(()) — i.e., after the connection ran
-    // to completion (natural disconnect). A transient connect that immediately
-    // drops would not reset backoff.
-
-    let mut stream = conn.on_message();
-    while let Some(msg) = stream.next().await {
-        let payload: String = match msg.get_payload() {
-            Ok(p) => p,
-            Err(e) => {
-                tracing::warn!("Failed to get pub/sub message payload: {e}");
-                continue;
-            }
-        };
-
-        let channel_name = msg.get_channel_name();
-        let channel_id = channel_name
-            .strip_prefix("buzz:channel:")
-            .and_then(|s| Uuid::parse_str(s).ok());
-
-        let channel_id = match channel_id {
-            Some(id) => id,
-            None => {
-                tracing::warn!("Received pub/sub message on unexpected channel: {channel_name}");
-                continue;
-            }
-        };
-
-        let event = match nostr::Event::from_json(&payload) {
-            Ok(e) => e,
-            Err(e) => {
-                tracing::warn!("Failed to deserialize event from pub/sub: {e}");
-                continue;
-            }
-        };
-
-        let channel_event = ChannelEvent { channel_id, event };
-
-        if let Err(_e) = broadcast_tx.send(channel_event) {
-            tracing::trace!("No broadcast receivers for channel {channel_id} — message dropped");
-        }
+    for topic in initial_topics {
+        let channel = topic.redis_channel();
+        sink.subscribe(&channel).await?;
+        active_topics.insert(channel);
     }
 
-    // Stream returned None — Redis connection closed.
-    Ok(())
+    tracing::info!(
+        topic_count = active_topics.len(),
+        "Redis pub/sub subscriber connected with dynamic scoped subscriptions"
+    );
+
+    loop {
+        tokio::select! {
+            Some(command) = subscription_rx.recv() => {
+                match command {
+                    SubscriptionCommand::Subscribe(topic) => {
+                        let channel = topic.redis_channel();
+                        if active_topics.insert(channel.clone()) {
+                            sink.subscribe(&channel).await?;
+                        }
+                    }
+                    SubscriptionCommand::UnsubscribeIfIdle(topic) => {
+                        if desired_refcount(&desired_topics, topic).await == 0 {
+                            let channel = topic.redis_channel();
+                            if active_topics.remove(&channel) {
+                                sink.unsubscribe(&channel).await?;
+                            }
+                        }
+                    }
+                }
+            }
+            msg = stream.next() => {
+                let Some(msg) = msg else {
+                    // Stream returned None — Redis connection closed.
+                    return Ok(());
+                };
+
+                let payload: String = match msg.get_payload() {
+                    Ok(p) => p,
+                    Err(e) => {
+                        tracing::warn!("Failed to get pub/sub message payload: {e}");
+                        continue;
+                    }
+                };
+
+                let channel_name = msg.get_channel_name();
+                let topic_key = match EventTopicKey::parse_redis_channel(channel_name) {
+                    Ok(topic_key) => topic_key,
+                    Err(_) => {
+                        tracing::warn!("Received pub/sub message on unexpected channel: {channel_name}");
+                        continue;
+                    }
+                };
+
+                let event = match nostr::Event::from_json(&payload) {
+                    Ok(e) => e,
+                    Err(e) => {
+                        tracing::warn!("Failed to deserialize event from pub/sub: {e}");
+                        continue;
+                    }
+                };
+
+                let channel_event = ChannelEvent {
+                    community_id: topic_key.community_id,
+                    topic: topic_key.topic,
+                    event,
+                };
+
+                if let Err(_e) = broadcast_tx.send(channel_event) {
+                    tracing::trace!(topic = %channel_name, "No broadcast receivers for topic — message dropped");
+                }
+            }
+            else => {
+                // Command channel closed and stream ended; let the reconnect loop retry.
+                return Ok(());
+            }
+        }
+    }
+}
+
+async fn desired_refcount(desired_topics: &DesiredTopics, topic: EventTopicKey) -> usize {
+    desired_topics
+        .lock()
+        .await
+        .get(&topic)
+        .copied()
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use buzz_core::{CommunityId, TenantContext};
+    use uuid::Uuid;
+
+    fn topic(id: u128) -> EventTopicKey {
+        let ctx = TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), "test");
+        EventTopicKey::from_context(&ctx, crate::EventTopic::Global)
+    }
+
+    #[tokio::test]
+    async fn desired_refcount_returns_zero_for_absent_topic() {
+        let desired = Arc::new(Mutex::new(HashMap::new()));
+        assert_eq!(desired_refcount(&desired, topic(1)).await, 0);
+    }
+
+    #[tokio::test]
+    async fn desired_refcount_reads_present_topic() {
+        let desired = Arc::new(Mutex::new(HashMap::from([(topic(1), 3)])));
+        assert_eq!(desired_refcount(&desired, topic(1)).await, 3);
+    }
 }

crates/buzz-pubsub/src/topic.rs

@@ -0,0 +1,197 @@
+//! Community-scoped Redis event topics.
+//!
+//! Pub/sub topics are a routing/performance boundary, not an authorization
+//! boundary. Tenant identity still comes from [`TenantContext`] on publish /
+//! retain paths, and the relay re-checks access before local fan-out.
+
+use buzz_core::{CommunityId, TenantContext};
+use uuid::Uuid;
+
+use crate::error::PubSubError;
+
+/// Redis key prefix for Buzz-scoped pub/sub topics and keys.
+pub const BUZZ_PREFIX: &str = "buzz";
+
+/// A tenant-local event routing scope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum EventTopic {
+    /// Events for one exact channel id.
+    Channel(Uuid),
+    /// Community-global events that are not exact-channel routed.
+    Global,
+}
+
+/// A fully qualified event topic, including its server-resolved community.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct EventTopicKey {
+    /// Server-resolved community id.
+    pub community_id: CommunityId,
+    /// Tenant-local routing scope.
+    pub topic: EventTopic,
+}
+
+impl EventTopicKey {
+    /// Build a topic key from a resolved tenant context.
+    pub fn from_context(ctx: &TenantContext, topic: EventTopic) -> Self {
+        Self {
+            community_id: ctx.community(),
+            topic,
+        }
+    }
+
+    /// Redis pub/sub channel name for this topic.
+    pub fn redis_channel(&self) -> String {
+        match self.topic {
+            EventTopic::Channel(channel_id) => {
+                format!("{BUZZ_PREFIX}:{}:channel:{channel_id}", self.community_id)
+            }
+            EventTopic::Global => format!("{BUZZ_PREFIX}:{}:global", self.community_id),
+        }
+    }
+
+    /// Parse a Redis pub/sub channel name into a scoped event topic.
+    pub fn parse_redis_channel(channel: &str) -> Result<Self, PubSubError> {
+        let mut parts = channel.split(':');
+        let Some(prefix) = parts.next() else {
+            return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+        };
+        if prefix != BUZZ_PREFIX {
+            return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+        }
+
+        let Some(community) = parts.next() else {
+            return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+        };
+        let community_id = Uuid::parse_str(community)
+            .map(CommunityId::from_uuid)
+            .map_err(|_| PubSubError::InvalidChannelKey(channel.to_string()))?;
+
+        let Some(scope) = parts.next() else {
+            return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+        };
+
+        let topic = match scope {
+            "global" => {
+                if parts.next().is_some() {
+                    return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+                }
+                EventTopic::Global
+            }
+            "channel" => {
+                let Some(channel_id) = parts.next() else {
+                    return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+                };
+                if parts.next().is_some() {
+                    return Err(PubSubError::InvalidChannelKey(channel.to_string()));
+                }
+                EventTopic::Channel(
+                    Uuid::parse_str(channel_id)
+                        .map_err(|_| PubSubError::InvalidChannelKey(channel.to_string()))?,
+                )
+            }
+            _ => return Err(PubSubError::InvalidChannelKey(channel.to_string())),
+        };
+
+        Ok(Self {
+            community_id,
+            topic,
+        })
+    }
+}
+
+/// Redis channel for exact-channel events under `ctx`.
+pub fn channel_key(ctx: &TenantContext, channel_id: Uuid) -> String {
+    EventTopicKey::from_context(ctx, EventTopic::Channel(channel_id)).redis_channel()
+}
+
+/// Redis channel for community-global events under `ctx`.
+pub fn global_key(ctx: &TenantContext) -> String {
+    EventTopicKey::from_context(ctx, EventTopic::Global).redis_channel()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ctx(id: u128, host: &str) -> TenantContext {
+        TenantContext::resolved(CommunityId::from_uuid(Uuid::from_u128(id)), host)
+    }
+
+    #[test]
+    fn channel_key_includes_community_and_channel() {
+        let ctx = ctx(0xaaaa, "a.example");
+        let channel_id = Uuid::from_u128(0xbbbb);
+
+        assert_eq!(
+            channel_key(&ctx, channel_id),
+            format!("buzz:{}:channel:{channel_id}", ctx.community())
+        );
+    }
+
+    #[test]
+    fn global_key_includes_community() {
+        let ctx = ctx(0xaaaa, "a.example");
+
+        assert_eq!(global_key(&ctx), format!("buzz:{}:global", ctx.community()));
+    }
+
+    #[test]
+    fn same_channel_in_two_communities_has_different_topics() {
+        let community_a = ctx(0xaaaa, "a.example");
+        let community_b = ctx(0xbbbb, "b.example");
+        let channel_id = Uuid::from_u128(0xcccc);
+
+        assert_ne!(
+            channel_key(&community_a, channel_id),
+            channel_key(&community_b, channel_id)
+        );
+    }
+
+    #[test]
+    fn parses_channel_topic() {
+        let community_id = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
+        let channel_id = Uuid::from_u128(0xbbbb);
+        let raw = format!("buzz:{community_id}:channel:{channel_id}");
+
+        assert_eq!(
+            EventTopicKey::parse_redis_channel(&raw).unwrap(),
+            EventTopicKey {
+                community_id,
+                topic: EventTopic::Channel(channel_id),
+            }
+        );
+    }
+
+    #[test]
+    fn parses_global_topic() {
+        let community_id = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
+        let raw = format!("buzz:{community_id}:global");
+
+        assert_eq!(
+            EventTopicKey::parse_redis_channel(&raw).unwrap(),
+            EventTopicKey {
+                community_id,
+                topic: EventTopic::Global,
+            }
+        );
+    }
+
+    #[test]
+    fn rejects_malformed_or_wrong_prefix_topics() {
+        for raw in [
+            "",
+            "not-buzz:00000000-0000-0000-0000-00000000aaaa:global",
+            "buzz:not-a-uuid:global",
+            "buzz:00000000-0000-0000-0000-00000000aaaa",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:global:extra",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:channel",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:channel:not-a-uuid",
+            "buzz:00000000-0000-0000-0000-00000000aaaa:presence:abc",
+        ] {
+            assert!(
+                EventTopicKey::parse_redis_channel(raw).is_err(),
+                "expected {raw:?} to be rejected"
+            );
+        }
+    }
+}