Scope relay runtime state by community

crates/buzz-relay/src/audio/handler.rs

@@ -253,7 +253,9 @@ async fn handle_audio_connection(
         return;
     }
 
-    let room = state.audio_rooms.get_or_create(channel_id);
+    let room = state
+        .audio_rooms
+        .get_or_create(tenant.community(), channel_id);
 
     // Re-check archived status after obtaining the room. This closes the
     // cross-boundary race: a joiner that passed ensure_membership before
@@ -271,12 +273,16 @@ async fn handle_audio_connection(
                         .into(),
                 ))
                 .await;
-            state.audio_rooms.cleanup_if_empty(channel_id);
+            state
+                .audio_rooms
+                .cleanup_if_empty(tenant.community(), channel_id);
             return;
         }
         Err(e) => {
             warn!(channel_id = %channel_id, "pre-join channel check failed (fail-closed): {e}");
-            state.audio_rooms.cleanup_if_empty(channel_id);
+            state
+                .audio_rooms
+                .cleanup_if_empty(tenant.community(), channel_id);
             return;
         }
         Ok(_) => {} // Channel exists and is not archived — proceed.
@@ -475,7 +481,9 @@ async fn handle_audio_connection(
                 room.clear_ended();
             }
             Ok(()) => {
-                state.audio_rooms.cleanup_if_empty(channel_id);
+                state
+                    .audio_rooms
+                    .cleanup_if_empty(tenant.community(), channel_id);
 
                 emit_participant_event(
                     &state,
@@ -489,7 +497,9 @@ async fn handle_audio_connection(
             }
         }
     } else {
-        state.audio_rooms.cleanup_if_empty(channel_id);
+        state
+            .audio_rooms
+            .cleanup_if_empty(tenant.community(), channel_id);
     }
 
     info!(

crates/buzz-relay/src/audio/room.rs

@@ -8,6 +8,7 @@
 //! Frames are opaque Opus bytes — the relay never decodes audio.
 //! `try_send` is used throughout: real-time audio tolerates drops, never queues.
 
+use buzz_core::CommunityId;
 use bytes::Bytes;
 use dashmap::DashMap;
 use std::sync::Arc;
@@ -87,8 +88,8 @@ struct AdmissionGuard {
     ///
     /// Pin is per-`Room`-instance and clears when the manager evicts the
     /// Room via [`AudioRoomManager::cleanup_if_empty`] — the next
-    /// `get_or_create` for the same channel id then constructs a fresh
-    /// `Room` with a fresh `AdmissionGuard` (and therefore `None` pin),
+    /// `get_or_create` for the same community-local channel then constructs a
+    /// fresh `Room` with a fresh `AdmissionGuard` (and therefore `None` pin),
     /// so a new generation of joiners can negotiate a new version.
     /// A momentarily-empty-but-not-yet-cleaned-up Room keeps its pin so
     /// reconnecting peers don't accidentally renegotiate mid-call. See
@@ -126,6 +127,8 @@ impl AdmissionGuard {
 
 /// A single audio room for one channel.
 pub struct Room {
+    /// Community this room belongs to.
+    pub community_id: CommunityId,
     /// Channel UUID this room belongs to.
     pub channel_id: Uuid,
     /// Connected peers keyed by peer UUID.
@@ -135,9 +138,10 @@ pub struct Room {
 }
 
 impl Room {
-    /// Create an empty room for the given channel.
-    pub fn new(channel_id: Uuid) -> Self {
+    /// Create an empty room for the given community-local channel.
+    pub fn new(community_id: CommunityId, channel_id: Uuid) -> Self {
         Self {
+            community_id,
             channel_id,
             peers: DashMap::new(),
             guard: std::sync::Mutex::new(AdmissionGuard::new()),
@@ -321,7 +325,7 @@ impl Room {
 
 /// Global registry of active audio rooms.
 pub struct AudioRoomManager {
-    rooms: DashMap<Uuid, Arc<Room>>,
+    rooms: DashMap<(CommunityId, Uuid), Arc<Room>>,
 }
 
 impl AudioRoomManager {
@@ -333,17 +337,21 @@ impl AudioRoomManager {
     }
 
     /// Get an existing room or create a new one.
-    pub fn get_or_create(&self, channel_id: Uuid) -> Arc<Room> {
+    ///
+    /// Channel UUIDs are only unique inside a community. The room key must
+    /// carry both labels so two tenants that legitimately reuse the same UUID
+    /// never share peer lists, protocol pins, or audio frames.
+    pub fn get_or_create(&self, community_id: CommunityId, channel_id: Uuid) -> Arc<Room> {
         self.rooms
-            .entry(channel_id)
-            .or_insert_with(|| Arc::new(Room::new(channel_id)))
+            .entry((community_id, channel_id))
+            .or_insert_with(|| Arc::new(Room::new(community_id, channel_id)))
             .clone()
     }
 
     /// Remove the room if it has no peers. Returns `true` if the room was removed.
-    pub fn cleanup_if_empty(&self, channel_id: Uuid) -> bool {
+    pub fn cleanup_if_empty(&self, community_id: CommunityId, channel_id: Uuid) -> bool {
         self.rooms
-            .remove_if(&channel_id, |_, room| room.is_empty())
+            .remove_if(&(community_id, channel_id), |_, room| room.is_empty())
             .is_some()
     }
 }
@@ -359,7 +367,7 @@ mod tests {
     use super::*;
 
     fn fresh_room() -> Room {
-        Room::new(Uuid::new_v4())
+        Room::new(CommunityId::from_uuid(Uuid::new_v4()), Uuid::new_v4())
     }
 
     /// First peer's `requested_version` becomes the room's pin; later peers
@@ -424,9 +432,10 @@ mod tests {
     #[test]
     fn manager_cleanup_resets_version_pin() {
         let manager = AudioRoomManager::new();
+        let community_id = CommunityId::from_uuid(Uuid::new_v4());
         let channel_id = Uuid::new_v4();
 
-        let room1 = manager.get_or_create(channel_id);
+        let room1 = manager.get_or_create(community_id, channel_id);
         let (peer_id, _, _, _) = room1
             .add_peer("alice".to_string(), 2)
             .expect("first peer admits");
@@ -435,16 +444,43 @@ mod tests {
             .remove_peer_and_check_ended(peer_id)
             .expect("peer existed");
         assert!(ended, "single-peer room should end on its last departure");
-        assert!(manager.cleanup_if_empty(channel_id));
+        assert!(manager.cleanup_if_empty(community_id, channel_id));
 
         // Next joiner with a different version on the same channel id gets a
         // brand-new room (no v=2 pin carried over from the prior generation).
-        let room2 = manager.get_or_create(channel_id);
+        let room2 = manager.get_or_create(community_id, channel_id);
         let _ = room2
             .add_peer("bob".to_string(), 1)
             .expect("fresh room must accept any version");
     }
 
+    #[test]
+    fn manager_isolates_same_channel_uuid_across_communities() {
+        let manager = AudioRoomManager::new();
+        let channel_id = Uuid::new_v4();
+        let community_a = CommunityId::from_uuid(Uuid::new_v4());
+        let community_b = CommunityId::from_uuid(Uuid::new_v4());
+
+        let room_a = manager.get_or_create(community_a, channel_id);
+        let room_b = manager.get_or_create(community_b, channel_id);
+
+        assert!(
+            !Arc::ptr_eq(&room_a, &room_b),
+            "same channel UUID in two communities must create distinct rooms"
+        );
+        assert_eq!(room_a.community_id, community_a);
+        assert_eq!(room_b.community_id, community_b);
+
+        room_a
+            .add_peer("alice".to_string(), 1)
+            .expect("A peer admits");
+        assert_eq!(room_a.peer_pubkeys(), vec![("alice".to_string(), 0)]);
+        assert!(
+            room_b.peer_pubkeys().is_empty(),
+            "A room peers must not appear in B's same-UUID room"
+        );
+    }
+
     /// Peer-index reuse: after a peer leaves, their index is released; a new
     /// peer joining the same (still-pinned) room reuses the freed index.
     /// Version pin must persist across this reuse — the room generation

crates/buzz-relay/src/handlers/auth.rs

@@ -284,7 +284,11 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
                         // Successfully materialized — this owner is authoritative.
                         auth_ctx.agent_owner_pubkey = Some(owner);
                         // Pre-warm the observer cache to avoid stale negatives.
-                        let cache_key = (pubkey.to_bytes().to_vec(), owner.to_bytes().to_vec());
+                        let cache_key = (
+                            conn.tenant.community(),
+                            pubkey.to_bytes().to_vec(),
+                            owner.to_bytes().to_vec(),
+                        );
                         state.observer_owner_cache.insert(cache_key, true);
                     }
                     Ok(false) => {

crates/buzz-relay/src/handlers/event.rs

@@ -997,7 +997,11 @@ async fn handle_agent_observer_event(
 
     let agent_bytes = route.agent.to_bytes().to_vec();
     let owner_bytes = route.owner.to_bytes().to_vec();
-    let cache_key = (agent_bytes.clone(), owner_bytes.clone());
+    let cache_key = (
+        conn.tenant.community(),
+        agent_bytes.clone(),
+        owner_bytes.clone(),
+    );
     let is_owner = if session_owner_match {
         true
     } else {
@@ -1153,6 +1157,8 @@ fn single_tag_content<'a>(event: &'a Event, tag_name: &str) -> Result<&'a str, S
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+    use std::sync::atomic::AtomicU8;
     use std::sync::Arc;
 
     use buzz_core::kind::{
@@ -1164,6 +1170,9 @@ mod tests {
         OBSERVER_FRAME_TELEMETRY,
     };
     use nostr::{EventBuilder, Keys, Kind, Tag};
+    use tokio::sync::{mpsc, Mutex, RwLock};
+    use tokio_util::sync::CancellationToken;
+    use uuid::Uuid;
 
     #[test]
     fn fanout_event_frame_matches_legacy_format_byte_for_byte() {
@@ -1304,6 +1313,95 @@ mod tests {
         assert!(err.contains("NIP-44"));
     }
 
+    #[tokio::test]
+    async fn observer_owner_cache_is_scoped_to_community() {
+        let state = fanout_access::test_state().await;
+        let agent = Keys::generate();
+        let owner = Keys::generate();
+        let agent_bytes = agent.public_key().to_bytes().to_vec();
+        let owner_bytes = owner.public_key().to_bytes().to_vec();
+        let community_a = buzz_core::CommunityId::from_uuid(Uuid::new_v4());
+        let community_b = buzz_core::CommunityId::from_uuid(Uuid::new_v4());
+
+        state.observer_owner_cache.insert(
+            (community_a, agent_bytes.clone(), owner_bytes.clone()),
+            true,
+        );
+        assert_eq!(
+            state.observer_owner_cache.get(&(
+                community_b,
+                agent_bytes.clone(),
+                owner_bytes.clone()
+            )),
+            None,
+            "A cached allow must not populate B's observer authorization key"
+        );
+        state.observer_owner_cache.insert(
+            (community_b, agent_bytes.clone(), owner_bytes.clone()),
+            false,
+        );
+
+        let encrypted = encrypt_observer_payload(
+            &agent,
+            &owner.public_key(),
+            &serde_json::json!({"type": "acp_read"}),
+        )
+        .expect("encrypt observer payload");
+        let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
+            .tags([
+                Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
+                Tag::parse([OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"),
+                Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
+            ])
+            .sign_with_keys(&agent)
+            .expect("sign event");
+
+        let (send_tx, mut send_rx) = mpsc::channel(1);
+        let (ctrl_tx, _ctrl_rx) = mpsc::channel(1);
+        let conn = Arc::new(crate::connection::ConnectionState {
+            conn_id: Uuid::new_v4(),
+            tenant: buzz_core::TenantContext::resolved(community_b, "b.example"),
+            remote_addr: "127.0.0.1:1234".parse().expect("socket addr"),
+            auth_state: RwLock::new(crate::connection::AuthState::Authenticated(
+                buzz_auth::AuthContext {
+                    pubkey: agent.public_key(),
+                    scopes: vec![],
+                    channel_ids: None,
+                    auth_method: buzz_auth::AuthMethod::Nip42,
+                    agent_owner_pubkey: None,
+                },
+            )),
+            subscriptions: Arc::new(Mutex::new(HashMap::new())),
+            send_tx,
+            ctrl_tx,
+            cancel: CancellationToken::new(),
+            backpressure_count: Arc::new(AtomicU8::new(0)),
+            grace_limit: 3,
+        });
+
+        super::handle_agent_observer_event(
+            event.clone(),
+            conn.conn_id,
+            &event.id.to_hex(),
+            conn,
+            state,
+        )
+        .await;
+
+        let axum::extract::ws::Message::Text(text) =
+            send_rx.try_recv().expect("observer rejection sent")
+        else {
+            panic!("expected text relay message");
+        };
+        let frame: serde_json::Value = serde_json::from_str(&text).expect("relay frame JSON");
+        assert_eq!(frame[0], "OK");
+        assert_eq!(frame[2], false);
+        assert_eq!(
+            frame[3],
+            "restricted: observer frame is not authorized for this agent owner"
+        );
+    }
+
     mod pubsub_fanout {
         use std::collections::HashMap;
         use std::sync::atomic::AtomicU8;

crates/buzz-relay/src/state.rs

@@ -357,11 +357,12 @@ pub struct AppState {
     /// Current in-flight media uploads per uploader pubkey.
     pub media_uploads_in_flight: Arc<DashMap<[u8; 32], u32>>,
     /// 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.
+    /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
+    /// `agent_owner_pubkey` is immutable inside one community, so a long TTL
+    /// (5 min) is safe once the community label is part of the key.
     /// Prevents repeated DB lookups from bursty observer traffic.
     #[allow(clippy::type_complexity)]
-    pub observer_owner_cache: Arc<moka::sync::Cache<(Vec<u8>, Vec<u8>), bool>>,
+    pub observer_owner_cache: Arc<moka::sync::Cache<(CommunityId, Vec<u8>, Vec<u8>), bool>>,
 
     /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`]
     /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to