From 857d3b178b8987a6bfde8627a45df1f97d6b469d Mon Sep 17 00:00:00 2001 From: Jordan Mecom Date: Thu, 9 Jul 2026 10:45:06 -0400 Subject: [PATCH] Scope derived runtime state by community --- crates/buzz-relay/src/api/media.rs | 109 +++++- crates/buzz-relay/src/connection.rs | 7 +- crates/buzz-relay/src/handlers/event.rs | 83 +++-- .../buzz-relay/src/handlers/mesh_signaling.rs | 34 +- .../buzz-relay/src/handlers/side_effects.rs | 4 +- crates/buzz-relay/src/main.rs | 8 +- crates/buzz-relay/src/state.rs | 315 +++++++++++++++--- 7 files changed, 462 insertions(+), 98 deletions(-) diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 9cee7ab72b..28074611bd 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -42,15 +42,15 @@ const MEDIA_UPLOAD_RATE_WINDOW: Duration = Duration::from_secs(60); struct UploadPermit { _global: tokio::sync::OwnedSemaphorePermit, - in_flight: Arc>, - pubkey: [u8; 32], + in_flight: Arc>, + key: crate::state::ScopedPubkeyKey, } impl Drop for UploadPermit { fn drop(&mut self) { use dashmap::mapref::entry::Entry; - if let Entry::Occupied(mut entry) = self.in_flight.entry(self.pubkey) { + if let Entry::Occupied(mut entry) = self.in_flight.entry(self.key) { if *entry.get() <= 1 { entry.remove(); } else { @@ -60,8 +60,12 @@ impl Drop for UploadPermit { } } -fn upload_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool { - let key: [u8; 32] = pubkey.to_bytes(); +fn upload_rate_limited( + state: &AppState, + community_id: buzz_core::CommunityId, + pubkey: &nostr::PublicKey, +) -> bool { + let key = (community_id, pubkey.to_bytes()); let now = Instant::now(); let limit = state.config.media_uploads_per_minute; let mut entry = state @@ -83,6 +87,7 @@ fn upload_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool { fn acquire_upload_permit( state: &AppState, + community_id: buzz_core::CommunityId, pubkey: &nostr::PublicKey, ) -> Result { let global = state @@ -91,7 +96,7 @@ fn acquire_upload_permit( .try_acquire_owned() .map_err(|_| MediaError::UploadConcurrencyLimitReached)?; - let key: [u8; 32] = pubkey.to_bytes(); + let key = (community_id, pubkey.to_bytes()); let mut in_flight = state.media_uploads_in_flight.entry(key).or_insert(0); if *in_flight >= state.config.media_max_concurrent_uploads_per_pubkey { return Err(MediaError::UploadConcurrencyLimitReached); @@ -102,7 +107,7 @@ fn acquire_upload_permit( Ok(UploadPermit { _global: global, in_flight: Arc::clone(&state.media_uploads_in_flight), - pubkey: key, + key, }) } @@ -185,15 +190,16 @@ impl FromRequestParts> for AuthenticatedUpload { .await .map_err(|_| MediaError::RelayMembershipRequired)?; - if upload_rate_limited(state, &auth_event.pubkey) { + if upload_rate_limited(state, tenant.community(), &auth_event.pubkey) { metrics::counter!("buzz_media_upload_rejections_total", "reason" => "rate_limit") .increment(1); return Err(MediaError::UploadRateLimitExceeded); } - let upload_permit = acquire_upload_permit(state, &auth_event.pubkey).inspect_err(|_| { - metrics::counter!("buzz_media_upload_rejections_total", "reason" => "concurrency") - .increment(1); - })?; + let upload_permit = acquire_upload_permit(state, tenant.community(), &auth_event.pubkey) + .inspect_err(|_| { + metrics::counter!("buzz_media_upload_rejections_total", "reason" => "concurrency") + .increment(1); + })?; Ok(AuthenticatedUpload { auth_event, @@ -743,9 +749,88 @@ fn extract_blossom_auth(headers: &HeaderMap) -> Result #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; + + use uuid::Uuid; const VALID_HASH: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; + async fn test_state() -> Arc { + let mut config = crate::config::Config::from_env().expect("default config loads"); + config.require_relay_membership = false; + config.redis_url = "redis://127.0.0.1:1".to_string(); + config.media_uploads_per_minute = 1; + config.media_max_concurrent_uploads = 2; + config.media_max_concurrent_uploads_per_pubkey = 1; + + let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + nostr::Keys::generate(), + media_storage, + ); + Arc::new(state) + } + + #[tokio::test] + async fn upload_rate_limiter_is_scoped_by_community() { + let state = test_state().await; + let pubkey = nostr::Keys::generate().public_key(); + let community_a = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + + assert!(!upload_rate_limited(&state, community_a, &pubkey)); + assert!(upload_rate_limited(&state, community_a, &pubkey)); + assert!( + !upload_rate_limited(&state, community_b, &pubkey), + "A's exhausted upload budget must not rate-limit the same key in B" + ); + } + + #[tokio::test] + async fn upload_concurrency_limit_is_scoped_by_community() { + let state = test_state().await; + let pubkey = nostr::Keys::generate().public_key(); + let community_a = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + + let permit_a = + acquire_upload_permit(&state, community_a, &pubkey).expect("first A upload allowed"); + assert!(matches!( + acquire_upload_permit(&state, community_a, &pubkey), + Err(MediaError::UploadConcurrencyLimitReached) + )); + let permit_b = acquire_upload_permit(&state, community_b, &pubkey) + .expect("A's in-flight upload must not block B"); + + drop(permit_b); + drop(permit_a); + } + #[test] fn test_validate_media_path_bare_hash() { assert!(validate_media_path(VALID_HASH).is_ok()); diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index f6765553eb..0a72f76dc0 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -243,9 +243,10 @@ pub async fn handle_connection( } state.conn_manager.deregister(conn.conn_id); if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await { - let remaining = state - .conn_manager - .connection_ids_for_pubkey(auth_ctx.pubkey.to_bytes().as_slice()); + let remaining = state.conn_manager.connection_ids_for_pubkey_in_community( + conn.tenant.community(), + auth_ctx.pubkey.to_bytes().as_slice(), + ); if remaining.is_empty() { let _ = state .pubsub diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 9ab57bab86..31911e4c73 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -815,7 +815,11 @@ async fn handle_ephemeral_event( if event_kind_u32(&event) == KIND_MESH_CONNECT_REQUEST { // Per-requester rate limit shared with the HTTP door — see // `mesh_signaling::connect_request_rate_limited` for rationale. - if super::mesh_signaling::connect_request_rate_limited(&state, &auth_pubkey) { + if super::mesh_signaling::connect_request_rate_limited( + &state, + conn.tenant.community(), + &auth_pubkey, + ) { conn.send(RelayMessage::ok( event_id_hex, false, @@ -924,6 +928,32 @@ struct AgentObserverRoute { direction: AgentObserverDirection, } +/// Check + bump the per-agent observer telemetry limit (100/sec window). +/// +/// Observer frames are ephemeral, but the rejection is visible to the sender. +/// Scope the counter by community so an agent key active in one tenant does not +/// consume another tenant's logical rate budget. +fn observer_frame_rate_limited( + state: &AppState, + community_id: CommunityId, + agent_key: [u8; 32], +) -> bool { + let now = std::time::Instant::now(); + let mut entry = state + .observer_rate_limiter + .entry((community_id, agent_key)) + .or_insert((0, now)); + let (count, window_start) = entry.value_mut(); + if now.duration_since(*window_start).as_secs() >= 1 { + *count = 1; + *window_start = now; + false + } else { + *count += 1; + *count > 100 + } +} + /// Handle encrypted agent observer frames (kind 24200). /// /// These frames bypass storage and are routed as global ephemeral events. The @@ -1041,25 +1071,13 @@ async fn handle_agent_observer_event( // be starved by bursty telemetry from the agent. if matches!(route.direction, AgentObserverDirection::Telemetry) { let agent_key: [u8; 32] = agent_bytes.as_slice().try_into().unwrap_or([0u8; 32]); - let now = std::time::Instant::now(); - let mut entry = state - .observer_rate_limiter - .entry(agent_key) - .or_insert((0, now)); - let (count, window_start) = entry.value_mut(); - if now.duration_since(*window_start).as_secs() >= 1 { - *count = 1; - *window_start = now; - } else { - *count += 1; - if *count > 100 { - conn.send(RelayMessage::ok( - event_id_hex, - false, - "rate-limited: observer frame rate exceeded (100/sec per agent)", - )); - return; - } + if observer_frame_rate_limited(&state, conn.tenant.community(), agent_key) { + conn.send(RelayMessage::ok( + event_id_hex, + false, + "rate-limited: observer frame rate exceeded (100/sec per agent)", + )); + return; } } @@ -1304,6 +1322,31 @@ mod tests { assert!(err.contains("NIP-44")); } + #[tokio::test] + async fn observer_frame_rate_limiter_is_scoped_by_community() { + let state = fanout_access::test_state().await; + let agent_key = Keys::generate().public_key().to_bytes(); + let community_a = buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::from_u128(0xAAAA)); + let community_b = buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::from_u128(0xBBBB)); + + for _ in 0..100 { + assert!(!super::observer_frame_rate_limited( + &state, + community_a, + agent_key + )); + } + assert!(super::observer_frame_rate_limited( + &state, + community_a, + agent_key + )); + assert!( + !super::observer_frame_rate_limited(&state, community_b, agent_key), + "A's exhausted budget must not rate-limit the same agent key in B" + ); + } + mod pubsub_fanout { use std::collections::HashMap; use std::sync::atomic::AtomicU8; diff --git a/crates/buzz-relay/src/handlers/mesh_signaling.rs b/crates/buzz-relay/src/handlers/mesh_signaling.rs index aee9bb63a1..87793a2e00 100644 --- a/crates/buzz-relay/src/handlers/mesh_signaling.rs +++ b/crates/buzz-relay/src/handlers/mesh_signaling.rs @@ -38,9 +38,14 @@ use crate::state::AppState; /// bound the amplification. 20/sec is far above any real interactive use; a /// buggy desktop loop can't storm the relay. Shared by the WS door /// (`handlers::event`) and the HTTP door (`handle_mesh_event_http`) — one -/// limiter, two transports. -pub(crate) fn connect_request_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool { - let key: [u8; 32] = pubkey.to_bytes(); +/// limiter, two transports. The key includes community because the same Nostr +/// key can legitimately participate in more than one community. +pub(crate) fn connect_request_rate_limited( + state: &AppState, + community_id: CommunityId, + pubkey: &nostr::PublicKey, +) -> bool { + let key = (community_id, pubkey.to_bytes()); let now = std::time::Instant::now(); let mut entry = state .mesh_connect_rate_limiter @@ -93,7 +98,7 @@ pub async fn handle_mesh_event_http( handle_status_report(state, tenant, &pubkey_hex, event).await } k if k == KIND_MESH_CONNECT_REQUEST => { - if connect_request_rate_limited(state, auth_pubkey) { + if connect_request_rate_limited(state, tenant.community(), auth_pubkey) { return Err("rate-limited: mesh connect request rate exceeded (20/sec)".to_string()); } handle_connect_request(state, tenant, &pubkey_hex, event).await @@ -804,14 +809,31 @@ mod tests { let pubkey = nostr::Keys::generate().public_key(); for i in 0..20 { assert!( - !connect_request_rate_limited(&state, &pubkey), + !connect_request_rate_limited(&state, test_tenant().community(), &pubkey), "request {} should pass", i + 1 ); } assert!( - connect_request_rate_limited(&state, &pubkey), + connect_request_rate_limited(&state, test_tenant().community(), &pubkey), "21st request in the same window must be limited" ); } + + #[tokio::test] + async fn connect_request_rate_limiter_is_scoped_by_community() { + let state = test_state().await; + let pubkey = nostr::Keys::generate().public_key(); + let community_a = CommunityId::from_uuid(uuid::Uuid::from_u128(0xAAAA)); + let community_b = CommunityId::from_uuid(uuid::Uuid::from_u128(0xBBBB)); + + for _ in 0..20 { + assert!(!connect_request_rate_limited(&state, community_a, &pubkey)); + } + assert!(connect_request_rate_limited(&state, community_a, &pubkey)); + assert!( + !connect_request_rate_limited(&state, community_b, &pubkey), + "A's exhausted budget must not rate-limit the same key in B" + ); + } } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index a02324669f..c64ae28530 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -42,7 +42,9 @@ async fn evict_live_channel_subscriptions( channel_id: Uuid, target_pubkey: &[u8], ) { - let conn_ids = state.conn_manager.connection_ids_for_pubkey(target_pubkey); + let conn_ids = state + .conn_manager + .connection_ids_for_pubkey_in_community(tenant.community(), target_pubkey); for conn_id in conn_ids { evict_conn_channel_subscriptions(tenant, state, channel_id, conn_id).await; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 083739bf3b..e56e9dbffd 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -716,10 +716,10 @@ async fn main() -> anyhow::Result<()> { loop { match rx.recv().await { Ok(scoped) => { - // The local moka caches key on globally-unique UUIDs / - // pubkeys, so applying the tenant-local op by key is - // correct regardless of community; the `community_id` - // scope rides the Redis topic, not the moka key. + // The Redis topic carries the originating community, + // and the local moka keys carry that same label. Apply + // only the matching tenant-local drop; a mutation in A + // must not flush B's derived state. state_for_cache .apply_cache_invalidation(scoped.community_id, scoped.invalidation); } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index ce955524fc..299b599aec 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -31,6 +31,10 @@ use crate::config::Config; use crate::connection::ConnectionSubscriptions; use crate::subscription::SubscriptionRegistry; +pub(crate) type ScopedPubkeyKey = (CommunityId, [u8; 32]); +type SlidingWindowCounter = (u32, Instant); +type ScopedRateLimiter = DashMap; + /// Per-connection entry in the connection manager. struct ConnEntry { tx: mpsc::Sender, @@ -109,21 +113,31 @@ impl ConnectionManager { } } - /// Return all live connection IDs authenticated as `pubkey_bytes`. - pub fn connection_ids_for_pubkey(&self, pubkey_bytes: &[u8]) -> Vec { + /// Return live connection IDs authenticated as `pubkey_bytes` in one community. + /// + /// The same Nostr key may be connected to multiple communities at once. + /// Callers use this for tenant-visible cleanup such as presence clearing and + /// subscription eviction, so a connection in B must not keep A's derived + /// state alive. + pub fn connection_ids_for_pubkey_in_community( + &self, + community_id: CommunityId, + pubkey_bytes: &[u8], + ) -> Vec { self.connections .iter() .filter_map(|entry| { - let matches = entry - .authenticated_pubkey - .read() - .ok() - .and_then(|value| { - value - .as_ref() - .map(|stored| stored.as_slice() == pubkey_bytes) - }) - .unwrap_or(false); + let matches = entry.community_id == community_id + && entry + .authenticated_pubkey + .read() + .ok() + .and_then(|value| { + value + .as_ref() + .map(|stored| stored.as_slice() == pubkey_bytes) + }) + .unwrap_or(false); matches.then_some(*entry.key()) }) .collect() @@ -163,7 +177,7 @@ impl ConnectionManager { ) -> usize { let frame = crate::protocol::RelayMessage::ok(event_id, false, reason); let mut closed = 0usize; - for conn_id in self.connection_ids_for_pubkey(pubkey) { + for conn_id in self.connection_ids_for_pubkey_in_community(community, pubkey) { if let Some(entry) = self.connections.get(&conn_id) { if entry.community_id != community { continue; @@ -342,20 +356,20 @@ pub struct AppState { pub nip98_replay: Arc, /// Per-agent sliding-window rate limiter for observer frames (kind 24200). - /// Key: agent pubkey bytes (32). Value: (count, window_start). + /// Key: (community_id, agent pubkey bytes). Value: (count, window_start). /// 100 events/sec per agent — prevents relay/DB pressure from bursty telemetry. - pub observer_rate_limiter: Arc>, + pub observer_rate_limiter: Arc, /// Per-requester sliding-window rate limiter for mesh connect requests - /// (kind 24621). Key: requester pubkey bytes (32). Value: (count, + /// (kind 24621). Key: (community_id, requester pubkey bytes). Value: (count, /// window_start). Bounds the 1→2 call-me-now amplification: a member is /// trusted, but a buggy desktop loop shouldn't make the relay sign+fan /// unboundedly. 20/sec is far above any real interactive use. - pub mesh_connect_rate_limiter: Arc>, + pub mesh_connect_rate_limiter: Arc, /// Per-uploader sliding-window rate limiter for media upload starts. - /// Key: uploader pubkey bytes (32). Value: (count, window_start). - pub media_upload_rate_limiter: Arc>, - /// Current in-flight media uploads per uploader pubkey. - pub media_uploads_in_flight: Arc>, + /// Key: (community_id, uploader pubkey bytes). Value: (count, window_start). + pub media_upload_rate_limiter: Arc, + /// Current in-flight media uploads per (community, uploader pubkey). + pub media_uploads_in_flight: Arc>, /// Cache for observer agent-owner authorization (kind 24200). /// Key: (agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner. /// agent_owner_pubkey is immutable so a long TTL (5 min) is safe. @@ -468,18 +482,21 @@ impl AppState { moka::sync::Cache::builder() .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) + .support_invalidation_closures() .build(), ), accessible_channels_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) + .support_invalidation_closures() .build(), ), channel_visibility_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) + .support_invalidation_closures() .build(), ), audit_tx, @@ -575,13 +592,25 @@ impl AppState { /// Invalidate all users' accessible-channels cache (e.g. new open channel created). pub fn invalidate_all_accessible_channels(&self, tenant: &TenantContext) { - self.invalidate_all_accessible_channels_local(); + self.invalidate_all_accessible_channels_local(tenant.community()); self.spawn_cache_invalidation(tenant, CacheInvalidation::AccessibleAll); } /// Local-only accessible-channels drop. See [`invalidate_membership_local`]. - pub(crate) fn invalidate_all_accessible_channels_local(&self) { - self.accessible_channels_cache.invalidate_all(); + pub(crate) fn invalidate_all_accessible_channels_local(&self, community_id: CommunityId) { + if let Err(error) = self + .accessible_channels_cache + .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id) + { + // AppState enables invalidation closures at construction time. If + // that invariant ever regresses, prefer over-invalidating to + // serving stale access state. + tracing::error!( + ?error, + "community-scoped accessible-channel invalidation unavailable; falling back to full invalidation" + ); + self.accessible_channels_cache.invalidate_all(); + } } /// Invalidate the cached visibility for a single channel (e.g. after a flip). @@ -602,20 +631,50 @@ impl AppState { /// Invalidate all caches after a channel is deleted. /// - /// Channel deletion is a rare admin operation. We clear the entire membership - /// cache because moka doesn't support prefix-based invalidation on composite - /// keys, and stale `is_member=true` entries for a deleted channel would bypass - /// the DB's `deleted_at IS NULL` guard. + /// Channel deletion is a rare admin operation, but it is still tenant-local: + /// a deletion in A must not flush B's cache entries. Predicate invalidation + /// keeps the safety property that stale `is_member=true` entries for the + /// deleted channel are removed without turning the cache drop into a + /// cross-community signal. pub fn invalidate_channel_deleted(&self, tenant: &TenantContext) { - self.invalidate_channel_deleted_local(); + self.invalidate_channel_deleted_local(tenant.community()); self.spawn_cache_invalidation(tenant, CacheInvalidation::ChannelDeleted); } /// Local-only channel-deleted drop. See [`invalidate_membership_local`]. - pub(crate) fn invalidate_channel_deleted_local(&self) { - self.membership_cache.invalidate_all(); - self.accessible_channels_cache.invalidate_all(); - self.channel_visibility_cache.invalidate_all(); + pub(crate) fn invalidate_channel_deleted_local(&self, community_id: CommunityId) { + if let Err(error) = + self.membership_cache + .invalidate_entries_if(move |(entry_community, _, _), _| { + *entry_community == community_id + }) + { + tracing::error!( + ?error, + "community-scoped membership invalidation unavailable; falling back to full invalidation" + ); + self.membership_cache.invalidate_all(); + } + if let Err(error) = self + .accessible_channels_cache + .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id) + { + tracing::error!( + ?error, + "community-scoped accessible-channel invalidation unavailable; falling back to full invalidation" + ); + self.accessible_channels_cache.invalidate_all(); + } + if let Err(error) = self + .channel_visibility_cache + .invalidate_entries_if(move |(entry_community, _), _| *entry_community == community_id) + { + tracing::error!( + ?error, + "community-scoped visibility invalidation unavailable; falling back to full invalidation" + ); + self.channel_visibility_cache.invalidate_all(); + } } /// Fire-and-forget publish of a cache-key drop to all other pods. Failures @@ -646,13 +705,13 @@ impl AppState { self.invalidate_membership_local(community_id, channel_id, &pubkey); } CacheInvalidation::AccessibleAll => { - self.invalidate_all_accessible_channels_local(); + self.invalidate_all_accessible_channels_local(community_id); } CacheInvalidation::Visibility { channel_id } => { self.invalidate_channel_visibility_local(community_id, channel_id); } CacheInvalidation::ChannelDeleted => { - self.invalidate_channel_deleted_local(); + self.invalidate_channel_deleted_local(community_id); } } } @@ -873,6 +932,43 @@ mod tests { (mgr, conn_id, rx, ctrl_rx, cancel, bp) } + async fn test_state() -> Arc { + let mut config = crate::config::Config::from_env().expect("default config loads"); + config.require_relay_membership = false; + config.redis_url = "redis://127.0.0.1:1".to_string(); + let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); + let db = buzz_db::Db::from_pool(pool.clone()); + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("pubsub manager"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage"); + let (state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + nostr::Keys::generate(), + media_storage, + ); + Arc::new(state) + } + #[test] fn send_to_resets_grace_counter_on_success() { let (mgr, id, _rx, _ctrl_rx, _cancel, bp) = setup_conn(16); @@ -984,30 +1080,51 @@ mod tests { } #[tokio::test] - async fn tracks_connections_by_authenticated_pubkey() { + async fn tracks_connections_by_authenticated_pubkey_within_community() { let mgr = ConnectionManager::new(); - let conn_id = Uuid::new_v4(); - let (tx, _rx) = mpsc::channel(1); - let (ctrl_tx, _ctrl_rx) = mpsc::channel(1); - let cancel = CancellationToken::new(); - let bp = Arc::new(AtomicU8::new(0)); - let subscriptions = Arc::new(Mutex::new(HashMap::new())); + let community_a = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = buzz_core::tenant::CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + let conn_a = Uuid::new_v4(); + let conn_b = Uuid::new_v4(); + let (tx_a, _rx_a) = mpsc::channel(1); + let (ctrl_tx_a, _ctrl_rx_a) = mpsc::channel(1); + let (tx_b, _rx_b) = mpsc::channel(1); + let (ctrl_tx_b, _ctrl_rx_b) = mpsc::channel(1); mgr.register( - conn_id, - tx, - ctrl_tx, - cancel, - buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), - bp, - Arc::clone(&subscriptions), + conn_a, + tx_a, + ctrl_tx_a, + CancellationToken::new(), + community_a, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + mgr.register( + conn_b, + tx_b, + ctrl_tx_b, + CancellationToken::new(), + community_b, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), 3, ); let pubkey = vec![7u8; 32]; - mgr.set_authenticated_pubkey(conn_id, pubkey.clone()); + mgr.set_authenticated_pubkey(conn_a, pubkey.clone()); + mgr.set_authenticated_pubkey(conn_b, pubkey.clone()); - assert_eq!(mgr.connection_ids_for_pubkey(&pubkey), vec![conn_id]); - assert!(mgr.subscriptions_for(conn_id).is_some()); + assert_eq!( + mgr.connection_ids_for_pubkey_in_community(community_a, &pubkey), + vec![conn_a] + ); + assert_eq!( + mgr.connection_ids_for_pubkey_in_community(community_b, &pubkey), + vec![conn_b] + ); + assert!(mgr.subscriptions_for(conn_a).is_some()); + assert!(mgr.subscriptions_for(conn_b).is_some()); } #[tokio::test] @@ -1037,6 +1154,100 @@ mod tests { assert_eq!(mgr.pubkey_for_conn(Uuid::new_v4()), None); } + #[tokio::test] + async fn accessible_channel_invalidation_is_scoped_to_community() { + let state = test_state().await; + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + let pubkey = vec![7u8; 32]; + let channels_a = vec![Uuid::from_u128(1)]; + let channels_b = vec![Uuid::from_u128(2)]; + + state + .accessible_channels_cache + .insert((community_a, pubkey.clone()), channels_a); + state + .accessible_channels_cache + .insert((community_b, pubkey.clone()), channels_b.clone()); + + state.invalidate_all_accessible_channels_local(community_a); + + assert_eq!( + state + .accessible_channels_cache + .get(&(community_a, pubkey.clone())), + None + ); + assert_eq!( + state + .accessible_channels_cache + .get(&(community_b, pubkey.clone())), + Some(channels_b), + "A's cache drop must not evict B's accessible-channel entry" + ); + } + + #[tokio::test] + async fn channel_deleted_invalidation_is_scoped_to_community() { + let state = test_state().await; + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + let channel_id = Uuid::from_u128(1); + let pubkey = vec![7u8; 32]; + + for community in [community_a, community_b] { + state + .membership_cache + .insert((community, channel_id, pubkey.clone()), true); + state + .accessible_channels_cache + .insert((community, pubkey.clone()), vec![channel_id]); + state + .channel_visibility_cache + .insert((community, channel_id), "private".to_string()); + } + + state.invalidate_channel_deleted_local(community_a); + + assert_eq!( + state + .membership_cache + .get(&(community_a, channel_id, pubkey.clone())), + None + ); + assert_eq!( + state + .accessible_channels_cache + .get(&(community_a, pubkey.clone())), + None + ); + assert_eq!( + state + .channel_visibility_cache + .get(&(community_a, channel_id)), + None + ); + assert_eq!( + state + .membership_cache + .get(&(community_b, channel_id, pubkey.clone())), + Some(true) + ); + assert_eq!( + state + .accessible_channels_cache + .get(&(community_b, pubkey.clone())), + Some(vec![channel_id]) + ); + assert_eq!( + state + .channel_visibility_cache + .get(&(community_b, channel_id)), + Some("private".to_string()), + "A's channel deletion must not evict B's cache entries" + ); + } + #[tokio::test] async fn disconnect_pubkey_closes_matching_conns_with_reason() { let (mgr, id, _rx, mut ctrl_rx, cancel, _bp) = setup_conn(8);