From e55c2ceae0e7140eb6053636c4081ab299bdf5ba Mon Sep 17 00:00:00 2001
From: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>
Date: Fri, 26 Jun 2026 18:59:03 -0400
Subject: [PATCH] fix(db): require community scope for row lookups

Co-authored-by: Mari <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>
---
 crates/buzz-db/src/channel.rs | 331 +++++++++++++++++++++++++++-------
 crates/buzz-db/src/event.rs   | 141 +++++++++++++--
 crates/buzz-db/src/lib.rs     | 225 +++++++++++++++++------
 crates/buzz-db/src/thread.rs  | 180 +++++++++++++++---
 4 files changed, 708 insertions(+), 169 deletions(-)

diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs
index ef2a1fa8e..bdbe812d4 100644
--- a/crates/buzz-db/src/channel.rs
+++ b/crates/buzz-db/src/channel.rs
@@ -256,8 +256,12 @@ pub async fn create_channel_with_id(
     Ok((record, was_created))
 }
 
-/// Fetches a channel record by ID. Returns `ChannelNotFound` if missing or deleted.
-pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecord> {
+/// Fetches a channel record by `(community_id, id)`. Returns `ChannelNotFound` if missing or deleted.
+pub async fn get_channel(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<ChannelRecord> {
     let row = sqlx::query(
         r#"
         SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility,
@@ -267,9 +271,10 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
                topic, topic_set_by, topic_set_at,
                purpose, purpose_set_by, purpose_set_at,
                ttl_seconds, ttl_deadline
-        FROM channels WHERE id = $1 AND deleted_at IS NULL
+        FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .fetch_optional(pool)
     .await?
@@ -279,19 +284,34 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
 }
 
 /// Returns the canvas content for a channel, if any.
-pub async fn get_canvas(pool: &PgPool, channel_id: Uuid) -> Result<Option<String>> {
-    let row = sqlx::query("SELECT canvas FROM channels WHERE id = $1 AND deleted_at IS NULL")
-        .bind(channel_id)
-        .fetch_optional(pool)
-        .await?
-        .ok_or(DbError::ChannelNotFound(channel_id))?;
+pub async fn get_canvas(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<Option<String>> {
+    let row = sqlx::query(
+        "SELECT canvas FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or(DbError::ChannelNotFound(channel_id))?;
     Ok(row.try_get("canvas")?)
 }
 
 /// Sets or clears the canvas content for a channel.
-pub async fn set_canvas(pool: &PgPool, channel_id: Uuid, canvas: Option<&str>) -> Result<()> {
-    let rows = sqlx::query("UPDATE channels SET canvas = $1 WHERE id = $2 AND deleted_at IS NULL")
+pub async fn set_canvas(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    canvas: Option<&str>,
+) -> Result<()> {
+    let rows = sqlx::query(
+        "UPDATE channels SET canvas = $1 WHERE community_id = $2 AND id = $3 AND deleted_at IS NULL",
+    )
         .bind(canvas)
+        .bind(community_id.as_uuid())
         .bind(channel_id)
         .execute(pool)
         .await?;
@@ -329,7 +349,7 @@ pub async fn add_member(
 
     let mut tx = pool.begin().await?;
 
-    let channel = get_channel_tx(&mut tx, channel_id).await?;
+    let channel = get_channel_tx(&mut tx, community_id, channel_id).await?;
 
     let effective_role = if channel.visibility == "private" {
         let inviter = invited_by.ok_or_else(|| {
@@ -497,12 +517,18 @@ pub async fn remove_member(
 }
 
 /// Returns `true` if the given pubkey is an active member of the channel.
-pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result<bool> {
+pub async fn is_member(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    pubkey: &[u8],
+) -> Result<bool> {
     let row = sqlx::query(
         "SELECT COUNT(*) as cnt FROM channel_members cm \
-         JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \
-         WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL",
+         JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
+         WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .bind(pubkey)
     .fetch_one(pool)
@@ -514,17 +540,22 @@ pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result
 /// Returns all active members of the given channel.
 ///
 /// Returns an empty list if the channel has been soft-deleted.
-pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result<Vec<MemberRecord>> {
+pub async fn get_members(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<Vec<MemberRecord>> {
     let rows = sqlx::query(
         r#"
         SELECT cm.channel_id, cm.pubkey, cm.role::text AS role, cm.joined_at, cm.invited_by, cm.removed_at
         FROM channel_members cm
-        JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
-        WHERE cm.channel_id = $1 AND cm.removed_at IS NULL
+        JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
+        WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.removed_at IS NULL
         ORDER BY cm.joined_at ASC
         LIMIT 1000
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .fetch_all(pool)
     .await?;
@@ -538,7 +569,11 @@ pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result<Vec<MemberRe
 /// Returns a flat `Vec<MemberRecord>` ordered by `joined_at`; callers should
 /// group by `channel_id` if per-channel access is needed.
 /// Returns an empty vec immediately when `channel_ids` is empty.
-pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec<MemberRecord>> {
+pub async fn get_members_bulk(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_ids: &[Uuid],
+) -> Result<Vec<MemberRecord>> {
     if channel_ids.is_empty() {
         return Ok(Vec::new());
     }
@@ -546,11 +581,12 @@ pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec
         r#"
         SELECT cm.channel_id, cm.pubkey, cm.role::text AS role, cm.joined_at, cm.invited_by, cm.removed_at
         FROM channel_members cm
-        JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
-        WHERE cm.channel_id = ANY($1) AND cm.removed_at IS NULL
+        JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
+        WHERE cm.community_id = $1 AND cm.channel_id = ANY($2) AND cm.removed_at IS NULL
         ORDER BY cm.joined_at ASC
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(channel_ids)
     .fetch_all(pool)
     .await?;
@@ -561,20 +597,25 @@ pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec
 ///
 /// Includes channels where the pubkey is an active member AND all open channels.
 /// Open channels must be included in REQ filter resolution.
-pub async fn get_accessible_channel_ids(pool: &PgPool, pubkey: &[u8]) -> Result<Vec<Uuid>> {
+pub async fn get_accessible_channel_ids(
+    pool: &PgPool,
+    community_id: CommunityId,
+    pubkey: &[u8],
+) -> Result<Vec<Uuid>> {
     let rows = sqlx::query(
         r#"
         SELECT cm.channel_id
         FROM channel_members cm
-        JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
-        WHERE cm.pubkey = $1 AND cm.removed_at IS NULL
+        JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
+        WHERE cm.community_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL
         UNION
         SELECT id AS channel_id
         FROM channels
-        WHERE visibility = 'open' AND deleted_at IS NULL
+        WHERE community_id = $1 AND visibility = 'open' AND deleted_at IS NULL
         LIMIT 1000
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(pubkey)
     .fetch_all(pool)
     .await?;
@@ -587,8 +628,12 @@ pub async fn get_accessible_channel_ids(pool: &PgPool, pubkey: &[u8]) -> Result<
         .collect()
 }
 
-/// Lists channels, optionally filtered by visibility string.
-pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Vec<ChannelRecord>> {
+/// Lists channels in a community, optionally filtered by visibility string.
+pub async fn list_channels(
+    pool: &PgPool,
+    community_id: CommunityId,
+    visibility: Option<&str>,
+) -> Result<Vec<ChannelRecord>> {
     let rows = if let Some(vis) = visibility {
         sqlx::query(
             r#"
@@ -600,11 +645,12 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
                    purpose, purpose_set_by, purpose_set_at,
                    ttl_seconds, ttl_deadline
             FROM channels
-            WHERE deleted_at IS NULL AND visibility::text = $1
+            WHERE community_id = $1 AND deleted_at IS NULL AND visibility::text = $2
             ORDER BY created_at DESC
             LIMIT 1000
             "#,
         )
+        .bind(community_id.as_uuid())
         .bind(vis)
         .fetch_all(pool)
         .await?
@@ -619,11 +665,12 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
                    purpose, purpose_set_by, purpose_set_at,
                    ttl_seconds, ttl_deadline
             FROM channels
-            WHERE deleted_at IS NULL
+            WHERE community_id = $1 AND deleted_at IS NULL
             ORDER BY created_at DESC
             LIMIT 1000
             "#,
         )
+        .bind(community_id.as_uuid())
         .fetch_all(pool)
         .await?
     };
@@ -653,6 +700,7 @@ async fn get_active_role_tx(
 /// Transaction-aware variant of [`get_channel`].
 async fn get_channel_tx(
     tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
     channel_id: Uuid,
 ) -> Result<ChannelRecord> {
     let row = sqlx::query(
@@ -664,9 +712,10 @@ async fn get_channel_tx(
                topic, topic_set_by, topic_set_at,
                purpose, purpose_set_by, purpose_set_at,
                ttl_seconds, ttl_deadline
-        FROM channels WHERE id = $1 AND deleted_at IS NULL
+        FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .fetch_optional(&mut **tx)
     .await?
@@ -683,6 +732,15 @@ pub struct BotChannelEntry {
     pub id: String,
 }
 
+/// A channel archived by the ephemeral-channel reaper.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ReapedEphemeralChannel {
+    /// Community that owns the archived channel.
+    pub community_id: CommunityId,
+    /// Archived channel UUID.
+    pub channel_id: Uuid,
+}
+
 /// Bot member record — a user with role=bot, with their channel memberships aggregated.
 #[derive(Debug, Clone)]
 pub struct BotMemberRecord {
@@ -730,6 +788,7 @@ pub struct AccessibleChannel {
 /// that visibility value are returned. `None` returns all accessible channels.
 pub async fn get_accessible_channels(
     pool: &PgPool,
+    community_id: CommunityId,
     pubkey: &[u8],
     visibility_filter: Option<&str>,
     member_only: Option<bool>,
@@ -756,20 +815,22 @@ pub async fn get_accessible_channels(
                (cm.channel_id IS NOT NULL) AS is_member
         FROM channels c
         LEFT JOIN channel_members cm
-            ON c.id = cm.channel_id AND cm.pubkey = $1 AND cm.removed_at IS NULL
-        WHERE c.deleted_at IS NULL
+            ON c.community_id = cm.community_id AND c.id = cm.channel_id AND cm.pubkey = $2 AND cm.removed_at IS NULL
+        WHERE c.community_id = $1 AND c.deleted_at IS NULL
           {membership_clause}
           AND (c.channel_type != 'dm' OR cm.hidden_at IS NULL)
     "#
     );
 
     let sql = if visibility_filter.is_some() {
-        format!("{base}  AND c.visibility::text = $2\n        ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n        LIMIT 1000")
+        format!("{base}  AND c.visibility::text = $3\n        ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n        LIMIT 1000")
     } else {
         format!("{base}        ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n        LIMIT 1000")
     };
 
-    let query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(pubkey);
+    let query = sqlx::query(sqlx::AssertSqlSafe(sql))
+        .bind(community_id.as_uuid())
+        .bind(pubkey);
     let query = if let Some(vis) = visibility_filter {
         query.bind(vis)
     } else {
@@ -786,24 +847,28 @@ pub async fn get_accessible_channels(
         .collect()
 }
 
-/// Returns all bot-role members with their channel memberships.
+/// Returns all bot-role members with their channel memberships in one community.
 ///
 /// Channels are returned as a JSON array of `{name, id}` objects via `json_agg`,
 /// preserving the 1:1 name↔UUID pairing. No separate string_agg ordering issues.
 /// Members with no active channel memberships are excluded (INNER JOIN on channels).
-pub async fn get_bot_members(pool: &PgPool) -> Result<Vec<BotMemberRecord>> {
+pub async fn get_bot_members(
+    pool: &PgPool,
+    community_id: CommunityId,
+) -> Result<Vec<BotMemberRecord>> {
     let rows = sqlx::query(
         r#"
         SELECT cm.pubkey, u.display_name, u.agent_type, u.capabilities,
                COALESCE(json_agg(DISTINCT jsonb_build_object('name', c.name, 'id', c.id::text)), '[]') AS channels_json
         FROM channel_members cm
-        LEFT JOIN users u ON cm.pubkey = u.pubkey
-        JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
-        WHERE cm.role = 'bot' AND cm.removed_at IS NULL
+        LEFT JOIN users u ON cm.community_id = u.community_id AND cm.pubkey = u.pubkey
+        JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
+        WHERE cm.community_id = $1 AND cm.role = 'bot' AND cm.removed_at IS NULL
         GROUP BY cm.pubkey, u.display_name, u.agent_type, u.capabilities
         LIMIT 1000
         "#,
     )
+    .bind(community_id.as_uuid())
     .fetch_all(pool)
     .await?;
 
@@ -939,6 +1004,7 @@ pub struct ChannelUpdate {
 /// Returns the updated `ChannelRecord` on success.
 pub async fn update_channel(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_id: Uuid,
     updates: ChannelUpdate,
 ) -> Result<ChannelRecord> {
@@ -980,8 +1046,9 @@ pub async fn update_channel(
             None => set_parts.push("ttl_deadline = NULL".to_string()),
         }
     }
+    let channel_param_idx = param_idx + 1;
     let sql = format!(
-        "UPDATE channels SET {}, updated_at = NOW() WHERE id = ${param_idx} AND deleted_at IS NULL",
+        "UPDATE channels SET {}, updated_at = NOW() WHERE community_id = ${param_idx} AND id = ${channel_param_idx} AND deleted_at IS NULL",
         set_parts.join(", ")
     );
 
@@ -998,6 +1065,7 @@ pub async fn update_channel(
     if let Some(ref ttl) = updates.ttl_seconds {
         q = q.bind(*ttl);
     }
+    q = q.bind(community_id.as_uuid());
     q = q.bind(channel_id);
 
     let result = q.execute(pool).await?;
@@ -1005,17 +1073,24 @@ pub async fn update_channel(
         return Err(DbError::ChannelNotFound(channel_id));
     }
 
-    get_channel(pool, channel_id).await
+    get_channel(pool, community_id, channel_id).await
 }
 
 /// Sets the topic for a channel, recording who set it and when.
-pub async fn set_topic(pool: &PgPool, channel_id: Uuid, topic: &str, set_by: &[u8]) -> Result<()> {
+pub async fn set_topic(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    topic: &str,
+    set_by: &[u8],
+) -> Result<()> {
     let result = sqlx::query(
         "UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \
-         WHERE id = $3 AND deleted_at IS NULL",
+         WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
     )
     .bind(topic)
     .bind(set_by)
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .execute(pool)
     .await?;
@@ -1028,16 +1103,18 @@ pub async fn set_topic(pool: &PgPool, channel_id: Uuid, topic: &str, set_by: &[u
 /// Sets the purpose for a channel, recording who set it and when.
 pub async fn set_purpose(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_id: Uuid,
     purpose: &str,
     set_by: &[u8],
 ) -> Result<()> {
     let result = sqlx::query(
         "UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \
-         WHERE id = $3 AND deleted_at IS NULL",
+         WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
     )
     .bind(purpose)
     .bind(set_by)
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .execute(pool)
     .await?;
@@ -1051,9 +1128,16 @@ pub async fn set_purpose(
 ///
 /// Returns `AccessDenied` if the channel is already archived.
 /// Returns `ChannelNotFound` if the channel does not exist or is deleted.
-pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
+pub async fn archive_channel(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<()> {
     // First check: does the channel exist and what is its state?
-    let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL")
+    let row = sqlx::query(
+        "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+        .bind(community_id.as_uuid())
         .bind(channel_id)
         .fetch_optional(pool)
         .await?;
@@ -1072,8 +1156,9 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
 
     sqlx::query(
         "UPDATE channels SET archived_at = NOW() \
-         WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NULL",
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .execute(pool)
     .await?;
@@ -1085,9 +1170,16 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
 ///
 /// Returns `AccessDenied` if the channel is not currently archived.
 /// Returns `ChannelNotFound` if the channel does not exist or is deleted.
-pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
+pub async fn unarchive_channel(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<()> {
     // First check: does the channel exist and what is its state?
-    let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL")
+    let row = sqlx::query(
+        "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+        .bind(community_id.as_uuid())
         .bind(channel_id)
         .fetch_optional(pool)
         .await?;
@@ -1108,8 +1200,9 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
                  WHEN ttl_seconds IS NOT NULL THEN NOW() + (ttl_seconds || ' seconds')::interval \
                  ELSE ttl_deadline \
              END \
-         WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NOT NULL",
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .execute(pool)
     .await?;
@@ -1121,9 +1214,15 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
 ///
 /// Returns `Ok(true)` if the channel was deleted, `Ok(false)` if already
 /// deleted or not found.
-pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result<bool> {
-    let result =
-        sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
+pub async fn soft_delete_channel(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<bool> {
+    let result = sqlx::query(
+        "UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+            .bind(community_id.as_uuid())
             .bind(channel_id)
             .execute(pool)
             .await?;
@@ -1132,10 +1231,15 @@ pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result<bool
 }
 
 /// Returns the count of active (non-removed) members in a channel.
-pub async fn get_member_count(pool: &PgPool, channel_id: Uuid) -> Result<i64> {
+pub async fn get_member_count(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<i64> {
     let row = sqlx::query(
-        "SELECT COUNT(*) as cnt FROM channel_members WHERE channel_id = $1 AND removed_at IS NULL",
+        "SELECT COUNT(*) as cnt FROM channel_members WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .fetch_one(pool)
     .await?;
@@ -1148,6 +1252,7 @@ pub async fn get_member_count(pool: &PgPool, channel_id: Uuid) -> Result<i64> {
 /// Single query regardless of input size.
 pub async fn get_member_counts_bulk(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_ids: &[Uuid],
 ) -> Result<std::collections::HashMap<Uuid, i64>> {
     if channel_ids.is_empty() {
@@ -1156,8 +1261,10 @@ pub async fn get_member_counts_bulk(
 
     let mut qb: sqlx::QueryBuilder<sqlx::Postgres> = sqlx::QueryBuilder::new(
         "SELECT channel_id, COUNT(*) as cnt FROM channel_members \
-         WHERE removed_at IS NULL AND channel_id IN (",
+         WHERE community_id = ",
     );
+    qb.push_bind(community_id.as_uuid());
+    qb.push(" AND removed_at IS NULL AND channel_id IN (");
     let mut sep = qb.separated(", ");
     for id in channel_ids {
         sep.push_bind(*id);
@@ -1180,14 +1287,16 @@ pub async fn get_member_counts_bulk(
 /// Returns `None` if the pubkey is not an active member.
 pub async fn get_member_role(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_id: Uuid,
     pubkey: &[u8],
 ) -> Result<Option<String>> {
     let row = sqlx::query(
         "SELECT cm.role::text AS role FROM channel_members cm \
-         JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \
-         WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL",
+         JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
+         WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .bind(pubkey)
     .fetch_optional(pool)
@@ -1198,11 +1307,16 @@ pub async fn get_member_role(
 /// Bump the TTL deadline for an ephemeral channel after a new message.
 ///
 /// No-op for permanent channels or channels that are already archived/deleted.
-pub async fn bump_ttl_deadline(pool: &PgPool, channel_id: Uuid) -> Result<()> {
+pub async fn bump_ttl_deadline(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<()> {
     sqlx::query(
         "UPDATE channels SET ttl_deadline = NOW() + (ttl_seconds || ' seconds')::interval \
-         WHERE id = $1 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
+         WHERE community_id = $1 AND id = $2 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .execute(pool)
     .await?;
@@ -1211,25 +1325,29 @@ pub async fn bump_ttl_deadline(pool: &PgPool, channel_id: Uuid) -> Result<()> {
 
 /// Archive ephemeral channels whose TTL deadline has passed.
 ///
-/// Returns the list of channel IDs that were archived. Idempotent — the
+/// Returns the `(community_id, channel_id)` list that was archived. Idempotent — the
 /// `archived_at IS NULL` guard prevents double-archiving even if called
 /// concurrently from multiple relay pods.
-pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Uuid>> {
+pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<ReapedEphemeralChannel>> {
     let rows = sqlx::query(
         "UPDATE channels SET archived_at = NOW() \
          WHERE ttl_seconds IS NOT NULL \
            AND ttl_deadline < NOW() \
            AND archived_at IS NULL \
            AND deleted_at IS NULL \
-         RETURNING id",
+         RETURNING community_id, id",
     )
     .fetch_all(pool)
     .await?;
 
     rows.into_iter()
         .map(|row| {
-            let id: Uuid = row.try_get("id")?;
-            Ok(id)
+            let community_id: Uuid = row.try_get("community_id")?;
+            let channel_id: Uuid = row.try_get("id")?;
+            Ok(ReapedEphemeralChannel {
+                community_id: CommunityId::from_uuid(community_id),
+                channel_id,
+            })
         })
         .collect()
 }
@@ -1311,7 +1429,78 @@ mod tests {
         .await
         .expect("insert owner membership");
 
-        get_channel(pool, id).await
+        get_channel(pool, CommunityId::from_uuid(community_id), id).await
+    }
+
+    async fn insert_channel_with_id(
+        pool: &PgPool,
+        community_id: Uuid,
+        id: Uuid,
+        name: &str,
+        created_by: &[u8],
+    ) {
+        sqlx::query(
+            r#"
+            INSERT INTO channels
+                (id, community_id, name, channel_type, visibility, created_by)
+            VALUES
+                ($1, $2, $3, 'stream', 'open', $4)
+            "#,
+        )
+        .bind(id)
+        .bind(community_id)
+        .bind(name)
+        .bind(created_by)
+        .execute(pool)
+        .await
+        .expect("insert channel with fixed id");
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn get_channel_is_scoped_when_channel_uuid_collides_across_communities() {
+        let pool = setup_pool().await;
+        let community_a = make_test_community(&pool).await;
+        let community_b = make_test_community(&pool).await;
+        let channel_id = Uuid::new_v4();
+        let creator = random_pubkey();
+
+        insert_channel_with_id(
+            &pool,
+            community_a,
+            channel_id,
+            "community-a-channel",
+            &creator,
+        )
+        .await;
+        insert_channel_with_id(
+            &pool,
+            community_b,
+            channel_id,
+            "community-b-channel",
+            &creator,
+        )
+        .await;
+
+        let a = get_channel(&pool, CommunityId::from_uuid(community_a), channel_id)
+            .await
+            .expect("community A channel should resolve");
+        let b = get_channel(&pool, CommunityId::from_uuid(community_b), channel_id)
+            .await
+            .expect("community B channel should resolve");
+
+        assert_eq!(a.name, "community-a-channel");
+        assert_eq!(b.name, "community-b-channel");
+
+        let listed_a = list_channels(&pool, CommunityId::from_uuid(community_a), None)
+            .await
+            .expect("list community A channels");
+        assert!(listed_a
+            .iter()
+            .any(|row| row.id == channel_id && row.name == "community-a-channel"));
+        assert!(!listed_a
+            .iter()
+            .any(|row| row.id == channel_id && row.name == "community-b-channel"));
     }
 
     /// Agent owner (non-admin) can remove their own bot from a channel.
@@ -1382,7 +1571,7 @@ mod tests {
 
         // Verify the agent is no longer a member
         assert!(
-            !is_member(&pool, channel.id, &agent_pk)
+            !is_member(&pool, community, channel.id, &agent_pk)
                 .await
                 .expect("is_member check"),
             "agent should no longer be a member"
@@ -1424,11 +1613,11 @@ mod tests {
         .await
         .expect("expire and archive channel");
 
-        unarchive_channel(&pool, channel.id)
+        unarchive_channel(&pool, community, channel.id)
             .await
             .expect("unarchive expired ephemeral channel");
 
-        let channel = get_channel(&pool, channel.id)
+        let channel = get_channel(&pool, community, channel.id)
             .await
             .expect("reload channel");
         assert!(
@@ -1444,7 +1633,9 @@ mod tests {
             .await
             .expect("run reaper");
         assert!(
-            !reaped.contains(&channel.id),
+            !reaped
+                .iter()
+                .any(|row| row.community_id == community && row.channel_id == channel.id),
             "reaper should not immediately rearchive renewed channel"
         );
     }
diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs
index 29a23b51f..b8a3adcad 100644
--- a/crates/buzz-db/src/event.rs
+++ b/crates/buzz-db/src/event.rs
@@ -580,9 +580,15 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result<i64> {
 /// Returns `Ok(true)` if the event was deleted, `Ok(false)` if already deleted
 /// or not found. Callers are responsible for decrementing thread reply counts
 /// when the deleted event is a thread reply.
-pub async fn soft_delete_event(pool: &PgPool, event_id: &[u8]) -> Result<bool> {
-    let result =
-        sqlx::query("UPDATE events SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
+pub async fn soft_delete_event(
+    pool: &PgPool,
+    community_id: CommunityId,
+    event_id: &[u8],
+) -> Result<bool> {
+    let result = sqlx::query(
+        "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+            .bind(community_id.as_uuid())
             .bind(event_id)
             .execute(pool)
             .await?;
@@ -603,14 +609,16 @@ pub async fn soft_delete_event(pool: &PgPool, event_id: &[u8]) -> Result<bool> {
 /// (already deleted, or never existed).
 pub async fn soft_delete_by_coordinate(
     pool: &PgPool,
+    community_id: CommunityId,
     kind: i32,
     pubkey: &[u8],
     d_tag: &str,
 ) -> Result<bool> {
     let result = sqlx::query(
         "UPDATE events SET deleted_at = NOW() \
-         WHERE kind = $1 AND pubkey = $2 AND d_tag = $3 AND deleted_at IS NULL",
+         WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL",
     )
+    .bind(community_id.as_uuid())
     .bind(kind)
     .bind(pubkey)
     .bind(d_tag)
@@ -627,17 +635,20 @@ pub async fn soft_delete_by_coordinate(
 /// event was deleted this call.
 pub async fn soft_delete_event_and_update_thread(
     pool: &PgPool,
+    community_id: CommunityId,
     event_id: &[u8],
     parent_event_id: Option<&[u8]>,
     root_event_id: Option<&[u8]>,
 ) -> Result<bool> {
     let mut tx = pool.begin().await?;
 
-    let result =
-        sqlx::query("UPDATE events SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
-            .bind(event_id)
-            .execute(&mut *tx)
-            .await?;
+    let result = sqlx::query(
+        "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+    )
+    .bind(community_id.as_uuid())
+    .bind(event_id)
+    .execute(&mut *tx)
+    .await?;
 
     let deleted = result.rows_affected() > 0;
 
@@ -646,8 +657,9 @@ pub async fn soft_delete_event_and_update_thread(
             sqlx::query(
                 "UPDATE thread_metadata \
                  SET reply_count = GREATEST(reply_count - 1, 0) \
-                 WHERE event_id = $1",
+                 WHERE community_id = $1 AND event_id = $2",
             )
+            .bind(community_id.as_uuid())
             .bind(pid)
             .execute(&mut *tx)
             .await?;
@@ -656,8 +668,9 @@ pub async fn soft_delete_event_and_update_thread(
                 sqlx::query(
                     "UPDATE thread_metadata \
                      SET descendant_count = GREATEST(descendant_count - 1, 0) \
-                     WHERE event_id = $1",
+                     WHERE community_id = $1 AND event_id = $2",
                 )
+                .bind(community_id.as_uuid())
                 .bind(root_id)
                 .execute(&mut *tx)
                 .await?;
@@ -672,13 +685,15 @@ pub async fn soft_delete_event_and_update_thread(
 /// Returns the `created_at` timestamp of the most recent non-deleted event in a channel.
 pub async fn get_last_message_at(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_id: uuid::Uuid,
 ) -> Result<Option<DateTime<Utc>>> {
     let row = sqlx::query(
         "SELECT created_at FROM events \
-         WHERE channel_id = $1 AND deleted_at IS NULL \
+         WHERE community_id = $1 AND channel_id = $2 AND deleted_at IS NULL \
          ORDER BY created_at DESC LIMIT 1",
     )
+    .bind(community_id.as_uuid())
     .bind(channel_id)
     .fetch_optional(pool)
     .await?;
@@ -695,6 +710,7 @@ pub async fn get_last_message_at(
 /// Single query regardless of input size.
 pub async fn get_last_message_at_bulk(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_ids: &[uuid::Uuid],
 ) -> Result<std::collections::HashMap<uuid::Uuid, DateTime<Utc>>> {
     if channel_ids.is_empty() {
@@ -703,8 +719,10 @@ pub async fn get_last_message_at_bulk(
 
     let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
         "SELECT channel_id, MAX(created_at) as last_at FROM events \
-         WHERE deleted_at IS NULL AND channel_id IN (",
+         WHERE community_id = ",
     );
+    qb.push_bind(community_id.as_uuid());
+    qb.push(" AND deleted_at IS NULL AND channel_id IN (");
     let mut sep = qb.separated(", ");
     for id in channel_ids {
         sep.push_bind(*id);
@@ -727,11 +745,16 @@ pub async fn get_last_message_at_bulk(
 /// Returns `None` if the event does not exist or has been soft-deleted.
 /// Use [`get_event_by_id_including_deleted`] when you need to inspect
 /// tombstoned rows (e.g. audit, undelete).
-pub async fn get_event_by_id(pool: &PgPool, id_bytes: &[u8]) -> Result<Option<StoredEvent>> {
+pub async fn get_event_by_id(
+    pool: &PgPool,
+    community_id: CommunityId,
+    id_bytes: &[u8],
+) -> Result<Option<StoredEvent>> {
     let row = sqlx::query(
         "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
-         FROM events WHERE id = $1 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
+         FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
     )
+    .bind(community_id.as_uuid())
     .bind(id_bytes)
     .fetch_optional(pool)
     .await?;
@@ -750,16 +773,18 @@ pub async fn get_event_by_id(pool: &PgPool, id_bytes: &[u8]) -> Result<Option<St
 /// duplicate survivors where multiple live rows share the same timestamp.
 pub async fn get_latest_global_replaceable(
     pool: &PgPool,
+    community_id: CommunityId,
     kind: i32,
     pubkey_bytes: &[u8],
 ) -> Result<Option<StoredEvent>> {
     let row = sqlx::query(
         "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
          FROM events \
-         WHERE kind = $1 AND pubkey = $2 AND channel_id IS NULL AND deleted_at IS NULL \
+         WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id IS NULL AND deleted_at IS NULL \
          ORDER BY created_at DESC, id ASC \
          LIMIT 1",
     )
+    .bind(community_id.as_uuid())
     .bind(kind)
     .bind(pubkey_bytes)
     .fetch_optional(pool)
@@ -778,12 +803,14 @@ pub async fn get_latest_global_replaceable(
 /// audit trails, compliance queries).
 pub async fn get_event_by_id_including_deleted(
     pool: &PgPool,
+    community_id: CommunityId,
     id_bytes: &[u8],
 ) -> Result<Option<StoredEvent>> {
     let row = sqlx::query(
         "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
-         FROM events WHERE id = $1 ORDER BY created_at DESC LIMIT 1",
+         FROM events WHERE community_id = $1 AND id = $2 ORDER BY created_at DESC LIMIT 1",
     )
+    .bind(community_id.as_uuid())
     .bind(id_bytes)
     .fetch_optional(pool)
     .await?;
@@ -798,7 +825,11 @@ pub async fn get_event_by_id_including_deleted(
 ///
 /// Returns events in arbitrary order — callers reorder as needed.
 /// Uses a single `WHERE id IN (...)` query regardless of input size.
-pub async fn get_events_by_ids(pool: &PgPool, ids: &[&[u8]]) -> Result<Vec<StoredEvent>> {
+pub async fn get_events_by_ids(
+    pool: &PgPool,
+    community_id: CommunityId,
+    ids: &[&[u8]],
+) -> Result<Vec<StoredEvent>> {
     if ids.is_empty() {
         return Ok(vec![]);
     }
@@ -806,8 +837,10 @@ pub async fn get_events_by_ids(pool: &PgPool, ids: &[&[u8]]) -> Result<Vec<Store
 
     let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
         "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
-         FROM events WHERE deleted_at IS NULL AND id IN (",
+         FROM events WHERE community_id = ",
     );
+    qb.push_bind(community_id.as_uuid());
+    qb.push(" AND deleted_at IS NULL AND id IN (");
     let mut sep = qb.separated(", ");
     for id in ids {
         sep.push_bind(id.to_vec());
@@ -1169,6 +1202,76 @@ mod tests {
     use super::*;
     use nostr::{EventBuilder, Keys, Kind, Tag};
 
+    const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
+
+    async fn setup_pool() -> PgPool {
+        let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
+            .or_else(|_| std::env::var("DATABASE_URL"))
+            .unwrap_or_else(|_| TEST_DB_URL.to_owned());
+
+        PgPool::connect(&database_url)
+            .await
+            .expect("connect to test DB")
+    }
+
+    async fn make_test_community(pool: &PgPool) -> Uuid {
+        let id = Uuid::new_v4();
+        let host = format!("event-test-{}.example", id.simple());
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(id)
+            .bind(host)
+            .execute(pool)
+            .await
+            .expect("insert test community");
+        id
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() {
+        let pool = setup_pool().await;
+        let community_a = CommunityId::from_uuid(make_test_community(&pool).await);
+        let community_b = CommunityId::from_uuid(make_test_community(&pool).await);
+        let keys = Keys::generate();
+        let event = EventBuilder::new(Kind::Custom(9), "same signed event")
+            .sign_with_keys(&keys)
+            .expect("sign event");
+
+        insert_event(&pool, community_a, &event, None)
+            .await
+            .expect("insert in community A");
+        insert_event(&pool, community_b, &event, None)
+            .await
+            .expect("insert same event in community B");
+
+        sqlx::query("UPDATE events SET content = $1 WHERE community_id = $2 AND id = $3")
+            .bind("community-a-copy")
+            .bind(community_a.as_uuid())
+            .bind(event.id.as_bytes())
+            .execute(&pool)
+            .await
+            .expect("mark community A row");
+        sqlx::query("UPDATE events SET content = $1 WHERE community_id = $2 AND id = $3")
+            .bind("community-b-copy")
+            .bind(community_b.as_uuid())
+            .bind(event.id.as_bytes())
+            .execute(&pool)
+            .await
+            .expect("mark community B row");
+
+        let a = get_event_by_id(&pool, community_a, event.id.as_bytes())
+            .await
+            .expect("lookup community A")
+            .expect("community A row exists");
+        let b = get_event_by_id(&pool, community_b, event.id.as_bytes())
+            .await
+            .expect("lookup community B")
+            .expect("community B row exists");
+
+        assert_eq!(a.event.content, "community-a-copy");
+        assert_eq!(b.event.content, "community-b-copy");
+    }
+
     fn make_event_with_kind_and_tags(kind: u16, tags: Vec<Tag>) -> nostr::Event {
         let keys = Keys::generate();
         EventBuilder::new(Kind::Custom(kind), "test")
diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs
index 6b0a7199c..efd96f49d 100644
--- a/crates/buzz-db/src/lib.rs
+++ b/crates/buzz-db/src/lib.rs
@@ -345,52 +345,65 @@ impl Db {
     /// historical duplicate survivors correctly.
     pub async fn get_latest_global_replaceable(
         &self,
+        community_id: CommunityId,
         kind: i32,
         pubkey_bytes: &[u8],
     ) -> Result<Option<StoredEvent>> {
-        event::get_latest_global_replaceable(&self.pool, kind, pubkey_bytes).await
+        event::get_latest_global_replaceable(&self.pool, community_id, kind, pubkey_bytes).await
     }
 
     /// Fetches a single non-deleted event by its raw ID bytes.
     ///
     /// Returns `None` if the event does not exist or has been soft-deleted.
-    pub async fn get_event_by_id(&self, id_bytes: &[u8]) -> Result<Option<StoredEvent>> {
-        event::get_event_by_id(&self.pool, id_bytes).await
+    pub async fn get_event_by_id(
+        &self,
+        community_id: CommunityId,
+        id_bytes: &[u8],
+    ) -> Result<Option<StoredEvent>> {
+        event::get_event_by_id(&self.pool, community_id, id_bytes).await
     }
 
     /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**.
     pub async fn get_event_by_id_including_deleted(
         &self,
+        community_id: CommunityId,
         id_bytes: &[u8],
     ) -> Result<Option<StoredEvent>> {
-        event::get_event_by_id_including_deleted(&self.pool, id_bytes).await
+        event::get_event_by_id_including_deleted(&self.pool, community_id, id_bytes).await
     }
 
     /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted.
-    pub async fn soft_delete_event(&self, event_id: &[u8]) -> Result<bool> {
-        event::soft_delete_event(&self.pool, event_id).await
+    pub async fn soft_delete_event(
+        &self,
+        community_id: CommunityId,
+        event_id: &[u8],
+    ) -> Result<bool> {
+        event::soft_delete_event(&self.pool, community_id, event_id).await
     }
 
     /// Soft-delete the live row for an addressable coordinate `(kind, pubkey, d_tag)`.
     /// Used by NIP-09 a-tag deletion for parameterized-replaceable kinds.
     pub async fn soft_delete_by_coordinate(
         &self,
+        community_id: CommunityId,
         kind: i32,
         pubkey: &[u8],
         d_tag: &str,
     ) -> Result<bool> {
-        event::soft_delete_by_coordinate(&self.pool, kind, pubkey, d_tag).await
+        event::soft_delete_by_coordinate(&self.pool, community_id, kind, pubkey, d_tag).await
     }
 
     /// Atomically soft-delete an event and decrement thread reply counters.
     pub async fn soft_delete_event_and_update_thread(
         &self,
+        community_id: CommunityId,
         event_id: &[u8],
         parent_event_id: Option<&[u8]>,
         root_event_id: Option<&[u8]>,
     ) -> Result<bool> {
         event::soft_delete_event_and_update_thread(
             &self.pool,
+            community_id,
             event_id,
             parent_event_id,
             root_event_id,
@@ -399,21 +412,30 @@ impl Db {
     }
 
     /// Returns the most recent `created_at` for a channel.
-    pub async fn get_last_message_at(&self, channel_id: Uuid) -> Result<Option<DateTime<Utc>>> {
-        event::get_last_message_at(&self.pool, channel_id).await
+    pub async fn get_last_message_at(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<Option<DateTime<Utc>>> {
+        event::get_last_message_at(&self.pool, community_id, channel_id).await
     }
 
     /// Bulk-fetch the most recent `created_at` for a set of channel IDs.
     pub async fn get_last_message_at_bulk(
         &self,
+        community_id: CommunityId,
         channel_ids: &[Uuid],
     ) -> Result<std::collections::HashMap<Uuid, DateTime<Utc>>> {
-        event::get_last_message_at_bulk(&self.pool, channel_ids).await
+        event::get_last_message_at_bulk(&self.pool, community_id, channel_ids).await
     }
 
     /// Batch-fetch non-deleted events by their raw IDs.
-    pub async fn get_events_by_ids(&self, ids: &[&[u8]]) -> Result<Vec<StoredEvent>> {
-        event::get_events_by_ids(&self.pool, ids).await
+    pub async fn get_events_by_ids(
+        &self,
+        community_id: CommunityId,
+        ids: &[&[u8]],
+    ) -> Result<Vec<StoredEvent>> {
+        event::get_events_by_ids(&self.pool, community_id, ids).await
     }
 
     /// Atomically insert an event AND its thread metadata in a single transaction.
@@ -494,18 +516,31 @@ impl Db {
     }
 
     /// Fetches a channel record by ID.
-    pub async fn get_channel(&self, channel_id: Uuid) -> Result<channel::ChannelRecord> {
-        channel::get_channel(&self.pool, channel_id).await
+    pub async fn get_channel(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<channel::ChannelRecord> {
+        channel::get_channel(&self.pool, community_id, channel_id).await
     }
 
     /// Returns the canvas content for a channel, if any.
-    pub async fn get_canvas(&self, channel_id: Uuid) -> Result<Option<String>> {
-        channel::get_canvas(&self.pool, channel_id).await
+    pub async fn get_canvas(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<Option<String>> {
+        channel::get_canvas(&self.pool, community_id, channel_id).await
     }
 
     /// Sets or clears the canvas content for a channel.
-    pub async fn set_canvas(&self, channel_id: Uuid, canvas: Option<&str>) -> Result<()> {
-        channel::set_canvas(&self.pool, channel_id, canvas).await
+    pub async fn set_canvas(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        canvas: Option<&str>,
+    ) -> Result<()> {
+        channel::set_canvas(&self.pool, community_id, channel_id, canvas).await
     }
 
     /// Adds a member to a channel.
@@ -540,49 +575,75 @@ impl Db {
     }
 
     /// Returns `true` if the pubkey is an active member.
-    pub async fn is_member(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<bool> {
-        channel::is_member(&self.pool, channel_id, pubkey).await
+    pub async fn is_member(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        pubkey: &[u8],
+    ) -> Result<bool> {
+        channel::is_member(&self.pool, community_id, channel_id, pubkey).await
     }
 
     /// Returns all active members of a channel.
-    pub async fn get_members(&self, channel_id: Uuid) -> Result<Vec<channel::MemberRecord>> {
-        channel::get_members(&self.pool, channel_id).await
+    pub async fn get_members(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<Vec<channel::MemberRecord>> {
+        channel::get_members(&self.pool, community_id, channel_id).await
     }
 
     /// Returns active members for multiple channels in a single query.
     pub async fn get_members_bulk(
         &self,
+        community_id: CommunityId,
         channel_ids: &[Uuid],
     ) -> Result<Vec<channel::MemberRecord>> {
-        channel::get_members_bulk(&self.pool, channel_ids).await
+        channel::get_members_bulk(&self.pool, community_id, channel_ids).await
     }
 
     /// Get all channel IDs accessible to a pubkey.
-    pub async fn get_accessible_channel_ids(&self, pubkey: &[u8]) -> Result<Vec<Uuid>> {
-        channel::get_accessible_channel_ids(&self.pool, pubkey).await
+    pub async fn get_accessible_channel_ids(
+        &self,
+        community_id: CommunityId,
+        pubkey: &[u8],
+    ) -> Result<Vec<Uuid>> {
+        channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await
     }
 
     /// Lists channels, optionally filtered by visibility.
     pub async fn list_channels(
         &self,
+        community_id: CommunityId,
         visibility: Option<&str>,
     ) -> Result<Vec<channel::ChannelRecord>> {
-        channel::list_channels(&self.pool, visibility).await
+        channel::list_channels(&self.pool, community_id, visibility).await
     }
 
     /// Returns full channel records for all channels a user can access.
     pub async fn get_accessible_channels(
         &self,
+        community_id: CommunityId,
         pubkey: &[u8],
         visibility_filter: Option<&str>,
         member_only: Option<bool>,
     ) -> Result<Vec<channel::AccessibleChannel>> {
-        channel::get_accessible_channels(&self.pool, pubkey, visibility_filter, member_only).await
+        channel::get_accessible_channels(
+            &self.pool,
+            community_id,
+            pubkey,
+            visibility_filter,
+            member_only,
+        )
+        .await
     }
 
-    /// Returns all bot-role members with their aggregated channel names.
-    pub async fn get_bot_members(&self) -> Result<Vec<channel::BotMemberRecord>> {
-        channel::get_bot_members(&self.pool).await
+    /// Returns all bot-role members with their aggregated channel names in one community.
+    pub async fn get_bot_members(
+        &self,
+        community_id: CommunityId,
+    ) -> Result<Vec<channel::BotMemberRecord>> {
+        channel::get_bot_members(&self.pool, community_id).await
     }
 
     /// Bulk-fetch user records by pubkey.
@@ -593,62 +654,99 @@ impl Db {
     /// Updates a channel's name and/or description.
     pub async fn update_channel(
         &self,
+        community_id: CommunityId,
         channel_id: Uuid,
         updates: channel::ChannelUpdate,
     ) -> Result<channel::ChannelRecord> {
-        channel::update_channel(&self.pool, channel_id, updates).await
+        channel::update_channel(&self.pool, community_id, channel_id, updates).await
     }
 
     /// Sets the topic for a channel.
-    pub async fn set_topic(&self, channel_id: Uuid, topic: &str, set_by: &[u8]) -> Result<()> {
-        channel::set_topic(&self.pool, channel_id, topic, set_by).await
+    pub async fn set_topic(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        topic: &str,
+        set_by: &[u8],
+    ) -> Result<()> {
+        channel::set_topic(&self.pool, community_id, channel_id, topic, set_by).await
     }
 
     /// Sets the purpose for a channel.
-    pub async fn set_purpose(&self, channel_id: Uuid, purpose: &str, set_by: &[u8]) -> Result<()> {
-        channel::set_purpose(&self.pool, channel_id, purpose, set_by).await
+    pub async fn set_purpose(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        purpose: &str,
+        set_by: &[u8],
+    ) -> Result<()> {
+        channel::set_purpose(&self.pool, community_id, channel_id, purpose, set_by).await
     }
 
     /// Archives a channel.
-    pub async fn archive_channel(&self, channel_id: Uuid) -> Result<()> {
-        channel::archive_channel(&self.pool, channel_id).await
+    pub async fn archive_channel(&self, community_id: CommunityId, channel_id: Uuid) -> Result<()> {
+        channel::archive_channel(&self.pool, community_id, channel_id).await
     }
 
     /// Unarchives a channel.
-    pub async fn unarchive_channel(&self, channel_id: Uuid) -> Result<()> {
-        channel::unarchive_channel(&self.pool, channel_id).await
+    pub async fn unarchive_channel(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<()> {
+        channel::unarchive_channel(&self.pool, community_id, channel_id).await
     }
 
     /// Soft-delete a channel.
-    pub async fn soft_delete_channel(&self, channel_id: Uuid) -> Result<bool> {
-        channel::soft_delete_channel(&self.pool, channel_id).await
+    pub async fn soft_delete_channel(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<bool> {
+        channel::soft_delete_channel(&self.pool, community_id, channel_id).await
     }
 
     /// Returns the count of active members in a channel.
-    pub async fn get_member_count(&self, channel_id: Uuid) -> Result<i64> {
-        channel::get_member_count(&self.pool, channel_id).await
+    pub async fn get_member_count(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<i64> {
+        channel::get_member_count(&self.pool, community_id, channel_id).await
     }
 
     /// Bulk-fetch member counts for a set of channel IDs.
     pub async fn get_member_counts_bulk(
         &self,
+        community_id: CommunityId,
         channel_ids: &[Uuid],
     ) -> Result<std::collections::HashMap<Uuid, i64>> {
-        channel::get_member_counts_bulk(&self.pool, channel_ids).await
+        channel::get_member_counts_bulk(&self.pool, community_id, channel_ids).await
     }
 
     /// Get the active role of a pubkey in a channel.
-    pub async fn get_member_role(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<Option<String>> {
-        channel::get_member_role(&self.pool, channel_id, pubkey).await
+    pub async fn get_member_role(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        pubkey: &[u8],
+    ) -> Result<Option<String>> {
+        channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await
     }
 
     /// Bump the TTL deadline for an ephemeral channel after a new message.
-    pub async fn bump_ttl_deadline(&self, channel_id: Uuid) -> Result<()> {
-        channel::bump_ttl_deadline(&self.pool, channel_id).await
+    pub async fn bump_ttl_deadline(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+    ) -> Result<()> {
+        channel::bump_ttl_deadline(&self.pool, community_id, channel_id).await
     }
 
     /// Archive ephemeral channels whose TTL deadline has passed.
-    pub async fn reap_expired_ephemeral_channels(&self) -> Result<Vec<Uuid>> {
+    pub async fn reap_expired_ephemeral_channels(
+        &self,
+    ) -> Result<Vec<channel::ReapedEphemeralChannel>> {
         channel::reap_expired_ephemeral_channels(&self.pool).await
     }
 
@@ -845,6 +943,7 @@ impl Db {
     #[allow(clippy::too_many_arguments)]
     pub async fn insert_thread_metadata(
         &self,
+        community_id: CommunityId,
         event_id: &[u8],
         event_created_at: DateTime<Utc>,
         channel_id: Uuid,
@@ -857,6 +956,7 @@ impl Db {
     ) -> Result<()> {
         thread::insert_thread_metadata(
             &self.pool,
+            community_id,
             event_id,
             event_created_at,
             channel_id,
@@ -873,25 +973,36 @@ impl Db {
     /// Fetch replies under a root event.
     pub async fn get_thread_replies(
         &self,
+        community_id: CommunityId,
         root_event_id: &[u8],
         depth_limit: Option<u32>,
         limit: u32,
         cursor: Option<&[u8]>,
     ) -> Result<Vec<thread::ThreadReply>> {
-        thread::get_thread_replies(&self.pool, root_event_id, depth_limit, limit, cursor).await
+        thread::get_thread_replies(
+            &self.pool,
+            community_id,
+            root_event_id,
+            depth_limit,
+            limit,
+            cursor,
+        )
+        .await
     }
 
     /// Fetch aggregated thread stats.
     pub async fn get_thread_summary(
         &self,
+        community_id: CommunityId,
         event_id: &[u8],
     ) -> Result<Option<thread::ThreadSummary>> {
-        thread::get_thread_summary(&self.pool, event_id).await
+        thread::get_thread_summary(&self.pool, community_id, event_id).await
     }
 
     /// Top-level messages for a channel.
     pub async fn get_channel_messages_top_level(
         &self,
+        community_id: CommunityId,
         channel_id: Uuid,
         limit: u32,
         before_cursor: Option<DateTime<Utc>>,
@@ -900,6 +1011,7 @@ impl Db {
     ) -> Result<Vec<thread::TopLevelMessage>> {
         thread::get_channel_messages_top_level(
             &self.pool,
+            community_id,
             channel_id,
             limit,
             before_cursor,
@@ -912,18 +1024,21 @@ impl Db {
     /// Look up a single thread_metadata row by event_id.
     pub async fn get_thread_metadata_by_event(
         &self,
+        community_id: CommunityId,
         event_id: &[u8],
     ) -> Result<Option<thread::ThreadMetadataRecord>> {
-        thread::get_thread_metadata_by_event(&self.pool, event_id).await
+        thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await
     }
 
     /// Decrement reply counts.
     pub async fn decrement_reply_count(
         &self,
+        community_id: CommunityId,
         parent_event_id: &[u8],
         root_event_id: Option<&[u8]>,
     ) -> Result<()> {
-        thread::decrement_reply_count(&self.pool, parent_event_id, root_event_id).await
+        thread::decrement_reply_count(&self.pool, community_id, parent_event_id, root_event_id)
+            .await
     }
 
     /// Add (or re-activate) a reaction.
@@ -1673,13 +1788,15 @@ impl Db {
     /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey.
     pub async fn soft_delete_discovery_events(
         &self,
+        community_id: CommunityId,
         channel_id: Uuid,
         relay_pubkey: &[u8],
     ) -> Result<u64> {
         let result = sqlx::query(
             "UPDATE events SET deleted_at = NOW() \
-             WHERE channel_id = $1 AND pubkey = $2 AND deleted_at IS NULL AND kind IN (39000, 39001, 39002)",
+             WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND deleted_at IS NULL AND kind IN (39000, 39001, 39002)",
         )
+        .bind(community_id.as_uuid())
         .bind(channel_id)
         .bind(relay_pubkey)
         .execute(&self.pool)
diff --git a/crates/buzz-db/src/thread.rs b/crates/buzz-db/src/thread.rs
index 2c7b07912..8281ed9db 100644
--- a/crates/buzz-db/src/thread.rs
+++ b/crates/buzz-db/src/thread.rs
@@ -9,6 +9,8 @@ use chrono::{DateTime, Utc};
 use sqlx::{PgPool, Row};
 use uuid::Uuid;
 
+use buzz_core::CommunityId;
+
 use crate::{error::Result, event::row_to_stored_event};
 
 // -- Structs ------------------------------------------------------------------
@@ -110,6 +112,7 @@ pub struct ThreadMetadataRecord {
 #[allow(clippy::too_many_arguments)]
 pub async fn insert_thread_metadata(
     pool: &PgPool,
+    community_id: CommunityId,
     event_id: &[u8],
     event_created_at: DateTime<Utc>,
     channel_id: Uuid,
@@ -125,14 +128,15 @@ pub async fn insert_thread_metadata(
     let result = sqlx::query(
         r#"
         INSERT INTO thread_metadata
-            (event_created_at, event_id, channel_id,
+            (community_id, event_created_at, event_id, channel_id,
              parent_event_id, parent_event_created_at,
              root_event_id, root_event_created_at,
              depth, broadcast)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
         ON CONFLICT DO NOTHING
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(event_created_at)
     .bind(event_id)
     .bind(channel_id)
@@ -156,14 +160,15 @@ pub async fn insert_thread_metadata(
             sqlx::query(
                 r#"
                 INSERT INTO thread_metadata
-                    (event_created_at, event_id, channel_id,
+                    (community_id, event_created_at, event_id, channel_id,
                      parent_event_id, parent_event_created_at,
                      root_event_id, root_event_created_at,
                      depth, broadcast)
-                VALUES ($1, $2, $3, NULL, NULL, NULL, NULL, 0, false)
+                VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false)
                 ON CONFLICT DO NOTHING
                 "#,
             )
+            .bind(community_id.as_uuid())
             .bind(parent_ts)
             .bind(pid)
             .bind(channel_id)
@@ -185,6 +190,7 @@ pub async fn insert_thread_metadata(
                         ON CONFLICT DO NOTHING
                         "#,
                     )
+                    .bind(community_id.as_uuid())
                     .bind(root_ts)
                     .bind(root_id)
                     .bind(channel_id)
@@ -199,9 +205,10 @@ pub async fn insert_thread_metadata(
                 UPDATE thread_metadata
                 SET reply_count   = reply_count + 1,
                     last_reply_at = NOW()
-                WHERE event_id = $1
+                WHERE community_id = $1 AND event_id = $2
                 "#,
             )
+            .bind(community_id.as_uuid())
             .bind(pid)
             .execute(&mut *tx)
             .await?;
@@ -212,9 +219,10 @@ pub async fn insert_thread_metadata(
                     r#"
                     UPDATE thread_metadata
                     SET descendant_count = descendant_count + 1
-                    WHERE event_id = $1
+                    WHERE community_id = $1 AND event_id = $2
                     "#,
                 )
+                .bind(community_id.as_uuid())
                 .bind(root_id)
                 .execute(&mut *tx)
                 .await?;
@@ -239,6 +247,7 @@ pub async fn insert_thread_metadata(
 #[allow(dead_code)]
 pub async fn increment_reply_count(
     pool: &PgPool,
+    community_id: CommunityId,
     parent_event_id: &[u8],
     root_event_id: Option<&[u8]>,
 ) -> Result<()> {
@@ -248,9 +257,10 @@ pub async fn increment_reply_count(
         UPDATE thread_metadata
         SET reply_count  = reply_count + 1,
             last_reply_at = NOW()
-        WHERE event_id = $1
+        WHERE community_id = $1 AND event_id = $2
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(parent_event_id)
     .execute(pool)
     .await?;
@@ -261,9 +271,10 @@ pub async fn increment_reply_count(
             r#"
             UPDATE thread_metadata
             SET descendant_count = descendant_count + 1
-            WHERE event_id = $1
+            WHERE community_id = $1 AND event_id = $2
             "#,
         )
+        .bind(community_id.as_uuid())
         .bind(root_id)
         .execute(pool)
         .await?;
@@ -277,6 +288,7 @@ pub async fn increment_reply_count(
 /// root -- even when root == parent. Mirrors the increment logic exactly.
 pub async fn decrement_reply_count(
     pool: &PgPool,
+    community_id: CommunityId,
     parent_event_id: &[u8],
     root_event_id: Option<&[u8]>,
 ) -> Result<()> {
@@ -285,9 +297,10 @@ pub async fn decrement_reply_count(
         r#"
         UPDATE thread_metadata
         SET reply_count = GREATEST(reply_count - 1, 0)
-        WHERE event_id = $1
+        WHERE community_id = $1 AND event_id = $2
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(parent_event_id)
     .execute(pool)
     .await?;
@@ -298,9 +311,10 @@ pub async fn decrement_reply_count(
             r#"
             UPDATE thread_metadata
             SET descendant_count = GREATEST(descendant_count - 1, 0)
-            WHERE event_id = $1
+            WHERE community_id = $1 AND event_id = $2
             "#,
         )
+        .bind(community_id.as_uuid())
         .bind(root_id)
         .execute(pool)
         .await?;
@@ -320,6 +334,7 @@ pub async fn decrement_reply_count(
 /// - `limit` -- maximum rows returned (caller should cap this).
 pub async fn get_thread_replies(
     pool: &PgPool,
+    community_id: CommunityId,
     root_event_id: &[u8],
     depth_limit: Option<u32>,
     limit: u32,
@@ -336,7 +351,7 @@ pub async fn get_thread_replies(
 
     // Build the query dynamically based on optional filters.
     // Track the next positional parameter index.
-    let mut param_idx = 2u32; // $1 is root_event_id
+    let mut param_idx = 3u32; // $1 is community_id, $2 is root_event_id
     let mut sql = String::from(
         r#"
         SELECT
@@ -357,9 +372,11 @@ pub async fn get_thread_replies(
             tm.broadcast
         FROM thread_metadata tm
         JOIN events e
-            ON e.created_at = tm.event_created_at
+            ON e.community_id = tm.community_id
+           AND e.created_at = tm.event_created_at
            AND e.id         = tm.event_id
-        WHERE tm.root_event_id = $1
+        WHERE tm.community_id = $1
+          AND tm.root_event_id = $2
           AND e.deleted_at IS NULL
         "#,
     );
@@ -377,7 +394,9 @@ pub async fn get_thread_replies(
         " ORDER BY tm.event_created_at ASC LIMIT ${param_idx}"
     ));
 
-    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(root_event_id);
+    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
+        .bind(community_id.as_uuid())
+        .bind(root_event_id);
 
     if let Some(dl) = depth_limit {
         q = q.bind(dl as i32);
@@ -428,15 +447,20 @@ pub async fn get_thread_replies(
 }
 
 /// Fetch aggregated thread stats for a single event, plus up to 10 participant pubkeys.
-pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option<ThreadSummary>> {
+pub async fn get_thread_summary(
+    pool: &PgPool,
+    community_id: CommunityId,
+    event_id: &[u8],
+) -> Result<Option<ThreadSummary>> {
     let row = sqlx::query(
         r#"
         SELECT reply_count, descendant_count, last_reply_at
         FROM thread_metadata
-        WHERE event_id = $1
+        WHERE community_id = $1 AND event_id = $2
         LIMIT 1
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(event_id)
     .fetch_optional(pool)
     .await?;
@@ -457,9 +481,11 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
             SELECT DISTINCT e.pubkey, MAX(e.created_at) AS last_seen
             FROM thread_metadata tm
             JOIN events e
-                ON e.created_at = tm.event_created_at
+                ON e.community_id = tm.community_id
+               AND e.created_at = tm.event_created_at
                AND e.id         = tm.event_id
-            WHERE tm.root_event_id = $1
+            WHERE tm.community_id = $1
+              AND tm.root_event_id = $2
               AND e.deleted_at IS NULL
             GROUP BY e.pubkey
         ) sub
@@ -467,6 +493,7 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
         LIMIT 10
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(event_id)
     .fetch_all(pool)
     .await?;
@@ -500,13 +527,14 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
 /// polling (returns only messages created after the given timestamp).
 pub async fn get_channel_messages_top_level(
     pool: &PgPool,
+    community_id: CommunityId,
     channel_id: Uuid,
     limit: u32,
     before_cursor: Option<DateTime<Utc>>,
     since_cursor: Option<DateTime<Utc>>,
     kind_filter: Option<&[u32]>,
 ) -> Result<Vec<TopLevelMessage>> {
-    let mut param_idx = 2u32; // $1 is channel_id
+    let mut param_idx = 3u32; // $1 is community_id, $2 is channel_id
     let mut sql = String::from(
         r#"
         SELECT
@@ -519,9 +547,11 @@ pub async fn get_channel_messages_top_level(
             e.channel_id
         FROM events e
         LEFT JOIN thread_metadata tm
-            ON tm.event_created_at = e.created_at
+            ON tm.community_id = e.community_id
+           AND tm.event_created_at = e.created_at
            AND tm.event_id         = e.id
-        WHERE e.channel_id = $1
+        WHERE e.community_id = $1
+          AND e.channel_id = $2
           AND e.deleted_at IS NULL
           AND (
                 tm.depth IS NULL
@@ -561,7 +591,9 @@ pub async fn get_channel_messages_top_level(
         " ORDER BY e.created_at {order} LIMIT ${param_idx}"
     ));
 
-    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(channel_id);
+    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
+        .bind(community_id.as_uuid())
+        .bind(channel_id);
 
     if let Some(cursor) = before_cursor {
         q = q.bind(cursor);
@@ -604,6 +636,7 @@ pub async fn get_channel_messages_top_level(
 /// can be decremented.
 pub async fn get_thread_metadata_by_event(
     pool: &PgPool,
+    community_id: CommunityId,
     event_id: &[u8],
 ) -> Result<Option<ThreadMetadataRecord>> {
     let row = sqlx::query(
@@ -619,10 +652,11 @@ pub async fn get_thread_metadata_by_event(
             descendant_count,
             broadcast
         FROM thread_metadata
-        WHERE event_id = $1
+        WHERE community_id = $1 AND event_id = $2
         LIMIT 1
         "#,
     )
+    .bind(community_id.as_uuid())
     .bind(event_id)
     .fetch_optional(pool)
     .await?;
@@ -746,11 +780,105 @@ mod tests {
         .await
         .expect("insert owner membership");
 
-        crate::channel::get_channel(pool, id)
+        crate::channel::get_channel(pool, buzz_core::CommunityId::from_uuid(community_id), id)
             .await
             .map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id)))
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn get_thread_metadata_by_event_is_scoped_when_event_id_collides_across_communities() {
+        let pool = setup_pool().await;
+        let author = Keys::generate();
+        let channel_id = Uuid::new_v4();
+        let community_a = make_test_community(&pool).await;
+        let community_b = make_test_community(&pool).await;
+        let community_a = buzz_core::CommunityId::from_uuid(community_a);
+        let community_b = buzz_core::CommunityId::from_uuid(community_b);
+
+        crate::channel::create_channel_with_id(
+            &pool,
+            community_a,
+            channel_id,
+            &format!("thread-collision-a-{channel_id}"),
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            author.public_key().to_bytes().as_slice(),
+            None,
+        )
+        .await
+        .expect("create community A channel");
+        crate::channel::create_channel_with_id(
+            &pool,
+            community_b,
+            channel_id,
+            &format!("thread-collision-b-{channel_id}"),
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            author.public_key().to_bytes().as_slice(),
+            None,
+        )
+        .await
+        .expect("create community B channel");
+
+        let event = make_stream_event(&author, "same id in both communities");
+        let created_at = event_created_at(&event);
+        insert_event_with_thread_metadata(
+            &pool,
+            community_a,
+            &event,
+            Some(channel_id),
+            Some(ThreadMetadataParams {
+                event_id: event.id.as_bytes(),
+                event_created_at: created_at,
+                channel_id,
+                parent_event_id: None,
+                parent_event_created_at: None,
+                root_event_id: None,
+                root_event_created_at: None,
+                depth: 0,
+                broadcast: true,
+            }),
+        )
+        .await
+        .expect("insert community A metadata");
+        insert_event_with_thread_metadata(
+            &pool,
+            community_b,
+            &event,
+            Some(channel_id),
+            Some(ThreadMetadataParams {
+                event_id: event.id.as_bytes(),
+                event_created_at: created_at,
+                channel_id,
+                parent_event_id: None,
+                parent_event_created_at: None,
+                root_event_id: None,
+                root_event_created_at: None,
+                depth: 3,
+                broadcast: false,
+            }),
+        )
+        .await
+        .expect("insert community B metadata");
+
+        let a = get_thread_metadata_by_event(&pool, community_a, event.id.as_bytes())
+            .await
+            .expect("lookup community A metadata")
+            .expect("community A metadata exists");
+        let b = get_thread_metadata_by_event(&pool, community_b, event.id.as_bytes())
+            .await
+            .expect("lookup community B metadata")
+            .expect("community B metadata exists");
+
+        assert_eq!(a.depth, 0);
+        assert!(a.broadcast);
+        assert_eq!(b.depth, 3);
+        assert!(!b.broadcast);
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn get_thread_replies_reconstructs_stored_events() {
@@ -797,7 +925,7 @@ mod tests {
         .await
         .expect("insert reply event and metadata");
 
-        let replies = get_thread_replies(&pool, root.id.as_bytes(), Some(10), 10, None)
+        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
             .await
             .expect("fetch thread replies");
 
@@ -893,7 +1021,7 @@ mod tests {
             .rows_affected();
         assert_eq!(rows_changed, 1, "expected to corrupt exactly one row");
 
-        let replies = get_thread_replies(&pool, root.id.as_bytes(), Some(10), 10, None)
+        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
             .await
             .expect("fetch thread replies must succeed despite a corrupt row");