diff --git a/crates/buzz-pubsub/src/cache_invalidation.rs b/crates/buzz-pubsub/src/cache_invalidation.rs index 7ad675e1d..f48158c9c 100644 --- a/crates/buzz-pubsub/src/cache_invalidation.rs +++ b/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 { + 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, + broadcast_tx: broadcast::Sender, ) { 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, + broadcast_tx: &broadcast::Sender, ) -> 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 { diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index de6618a87..e8e00176d 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/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) -> 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, + subscription_rx: Mutex>>, broadcast_tx: broadcast::Sender, - cache_invalidation_tx: broadcast::Sender, + cache_invalidation_tx: broadcast::Sender, } 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::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 { 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) { - 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 { + pub fn subscribe_cache_invalidations(&self) -> broadcast::Receiver { 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 { 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 { - 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, PubSubError> { - presence::get_presence(&self.pool, pubkey).await + pub async fn get_presence( + &self, + ctx: &TenantContext, + pubkey: &PublicKey, + ) -> Result, 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, 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 { 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)); + } } diff --git a/crates/buzz-pubsub/src/presence.rs b/crates/buzz-pubsub/src/presence.rs index 65567b557..178ba7550 100644 --- a/crates/buzz-pubsub/src/presence.rs +++ b/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, PubSubError> { +pub async fn get_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); let value: Option = 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 Result, PubSubError> { if pubkeys.is_empty() { return Ok(HashMap::new()); } let mut conn = pool.get().await?; - let keys: Vec = pubkeys.iter().map(presence_key).collect(); + let keys: Vec = pubkeys + .iter() + .map(|pubkey| presence_key(ctx, pubkey)) + .collect(); let values: Vec> = 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(); } } diff --git a/crates/buzz-pubsub/src/publisher.rs b/crates/buzz-pubsub/src/publisher.rs index ea5d15166..8ad06cc56 100644 --- a/crates/buzz-pubsub/src/publisher.rs +++ b/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 { 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) diff --git a/crates/buzz-pubsub/src/subscriber.rs b/crates/buzz-pubsub/src/subscriber.rs index 120d9538e..88826ed99 100644 --- a/crates/buzz-pubsub/src/subscriber.rs +++ b/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>>; + +/// 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) { +/// 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, + desired_topics: DesiredTopics, + mut subscription_rx: mpsc::Receiver, +) { 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, + desired_topics: DesiredTopics, + subscription_rx: &mut mpsc::Receiver, ) -> 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 = { + 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); + } } diff --git a/crates/buzz-pubsub/src/topic.rs b/crates/buzz-pubsub/src/topic.rs new file mode 100644 index 000000000..07cabf9ec --- /dev/null +++ b/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 { + 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" + ); + } + } +}