feat(relay): per-community usage metrics (#1723)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

crates/buzz-db/src/lib.rs

@@ -37,6 +37,8 @@ pub mod reaction;
 pub mod relay_members;
 /// Thread metadata persistence.
 pub mod thread;
+/// Per-community usage rollup queries for Prometheus gauges.
+pub mod usage;
 /// User profile persistence.
 pub mod user;
 /// Workflow, run, and approval persistence.
@@ -286,6 +288,64 @@ impl Db {
         }
     }
 
+    /// Return total number of communities on this relay.
+    pub async fn usage_community_count(&self) -> Result<i64> {
+        usage::community_count(&self.pool).await
+    }
+
+    /// Return per-community user counts split by human/agent.
+    pub async fn usage_user_counts(&self) -> Result<Vec<usage::CommunityUserCounts>> {
+        usage::user_counts(&self.pool).await
+    }
+
+    /// Return per-community channel counts by type.
+    pub async fn usage_channel_counts(&self) -> Result<Vec<usage::CommunityChannelCount>> {
+        usage::channel_counts(&self.pool).await
+    }
+
+    /// Return per-community kind=9 message counts.
+    pub async fn usage_message_counts(&self) -> Result<Vec<usage::CommunityMessageCount>> {
+        usage::message_counts(&self.pool).await
+    }
+
+    /// Return per-community relay-member counts by role.
+    pub async fn usage_relay_member_counts(&self) -> Result<Vec<usage::CommunityMemberCount>> {
+        usage::relay_member_counts(&self.pool).await
+    }
+
+    /// Return per-community workflow counts by status.
+    pub async fn usage_workflow_counts(&self) -> Result<Vec<usage::CommunityWorkflowCount>> {
+        usage::workflow_counts(&self.pool).await
+    }
+
+    /// Return per-community git-repo counts.
+    pub async fn usage_git_repo_counts(&self) -> Result<Vec<usage::CommunityGitRepoCount>> {
+        usage::git_repo_counts(&self.pool).await
+    }
+
+    /// Return per-community distinct active-user counts for a given SQL interval.
+    ///
+    /// `interval_sql` must be a trusted literal such as `"1 day"` or `"7 days"`.
+    pub async fn usage_active_user_counts(
+        &self,
+        interval_sql: &'static str,
+    ) -> Result<Vec<usage::CommunityActiveUsers>> {
+        usage::active_user_counts(&self.pool, interval_sql).await
+    }
+
+    /// Return per-community active-channel counts for a given SQL interval.
+    pub async fn usage_active_channel_counts(
+        &self,
+        interval_sql: &'static str,
+    ) -> Result<Vec<usage::CommunityActiveChannels>> {
+        usage::active_channel_counts(&self.pool, interval_sql).await
+    }
+
+    /// Return all community id → host mappings.
+    pub async fn usage_community_hosts(&self) -> Result<Vec<usage::CommunityHost>> {
+        usage::community_hosts(&self.pool).await
+    }
+
     /// Begin a database transaction for atomic multi-statement operations.
     ///
     /// Returns a `'static` transaction because `PgPool` is `Arc`-backed internally.
@@ -1154,7 +1214,11 @@ impl Db {
     }
 
     /// Ensure a user record exists (upsert).
-    pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result<()> {
+    ///
+    /// Returns `true` if a new row was inserted (first time), `false` if it
+    /// already existed. Callers use the `true` return to increment
+    /// `buzz_users_created_total`.
+    pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result<bool> {
         user::ensure_user(&self.pool, community_id, pubkey).await
     }
 

crates/buzz-db/src/usage.rs

@@ -0,0 +1,722 @@
+//! Per-community usage rollup queries for Prometheus gauges.
+//!
+//! Stock queries (`user_counts`, `channel_counts`, `relay_member_counts`,
+//! `workflow_counts`, `git_repo_counts`) use `GROUP BY community_id` against
+//! indexed columns — no per-community loops, no full-table scans.
+//!
+//! Event-derived queries (`message_counts`, `active_user_counts`,
+//! `active_channel_counts`) are exact aggregates over the `events` table.
+//! At scale these can become recurring partition scans; if that becomes a
+//! problem, move them to a maintained rollup table and drop the interval.
+//!
+//! Returned structs are plain data; the caller (relay poller) maps them
+//! to Prometheus labels and calls `metrics::gauge!(...).set(...)`.
+
+use crate::error::Result;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+/// Total number of communities registered on this relay.
+pub async fn community_count(pool: &PgPool) -> Result<i64> {
+    let row = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM communities")
+        .fetch_one(pool)
+        .await?;
+    Ok(row)
+}
+
+/// Per-community user counts split by human/agent.
+#[derive(Debug)]
+pub struct CommunityUserCounts {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Number of active human users (no `agent_owner_pubkey`).
+    pub human: i64,
+    /// Number of active agent users (`agent_owner_pubkey IS NOT NULL`).
+    pub agent: i64,
+}
+
+/// Return active (non-deactivated) user counts per community, split by type.
+///
+/// Agent discriminator: `agent_owner_pubkey IS NOT NULL`.
+pub async fn user_counts(pool: &PgPool) -> Result<Vec<CommunityUserCounts>> {
+    // Single GROUP BY query; two conditional SUMs avoid two round-trips.
+    let rows = sqlx::query_as::<_, (Uuid, i64, i64)>(
+        r#"
+        SELECT
+            community_id,
+            COUNT(*) FILTER (WHERE agent_owner_pubkey IS NULL)     AS human,
+            COUNT(*) FILTER (WHERE agent_owner_pubkey IS NOT NULL) AS agent
+        FROM users
+        WHERE deactivated_at IS NULL
+        GROUP BY community_id
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, human, agent)| CommunityUserCounts {
+            community_id,
+            human,
+            agent,
+        })
+        .collect())
+}
+
+/// Per-community channel counts by type.
+#[derive(Debug)]
+pub struct CommunityChannelCount {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Channel type string (e.g. `"stream"`, `"dm"`, `"forum"`, `"workflow"`).
+    pub channel_type: String,
+    /// Number of non-deleted channels of this type.
+    pub count: i64,
+}
+
+/// Return non-deleted channel counts per community per type.
+pub async fn channel_counts(pool: &PgPool) -> Result<Vec<CommunityChannelCount>> {
+    let rows = sqlx::query_as::<_, (Uuid, String, i64)>(
+        r#"
+        SELECT community_id, channel_type::text, COUNT(*) AS count
+        FROM channels
+        WHERE deleted_at IS NULL
+        GROUP BY community_id, channel_type
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(
+            |(community_id, channel_type, count)| CommunityChannelCount {
+                community_id,
+                channel_type,
+                count,
+            },
+        )
+        .collect())
+}
+
+/// Per-community message (kind=9) count.
+#[derive(Debug)]
+pub struct CommunityMessageCount {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Number of stored non-deleted kind=9 events.
+    pub count: i64,
+}
+
+/// Return non-deleted kind=9 event counts per community.
+pub async fn message_counts(pool: &PgPool) -> Result<Vec<CommunityMessageCount>> {
+    let rows = sqlx::query_as::<_, (Uuid, i64)>(
+        r#"
+        SELECT community_id, COUNT(*) AS count
+        FROM events
+        WHERE kind = 9 AND deleted_at IS NULL
+        GROUP BY community_id
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, count)| CommunityMessageCount {
+            community_id,
+            count,
+        })
+        .collect())
+}
+
+/// Per-community relay-member counts by role.
+#[derive(Debug)]
+pub struct CommunityMemberCount {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Role string (e.g. `"owner"`, `"admin"`, `"member"`).
+    pub role: String,
+    /// Number of members with this role.
+    pub count: i64,
+}
+
+/// Return relay-member counts per community per role.
+pub async fn relay_member_counts(pool: &PgPool) -> Result<Vec<CommunityMemberCount>> {
+    let rows = sqlx::query_as::<_, (Uuid, String, i64)>(
+        r#"
+        SELECT community_id, role::text, COUNT(*) AS count
+        FROM relay_members
+        GROUP BY community_id, role
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, role, count)| CommunityMemberCount {
+            community_id,
+            role,
+            count,
+        })
+        .collect())
+}
+
+/// Per-community workflow counts by status.
+#[derive(Debug)]
+pub struct CommunityWorkflowCount {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Workflow status string (e.g. `"active"`, `"inactive"`).
+    pub status: String,
+    /// Number of workflows in this status.
+    pub count: i64,
+}
+
+/// Return workflow counts per community per status.
+pub async fn workflow_counts(pool: &PgPool) -> Result<Vec<CommunityWorkflowCount>> {
+    let rows = sqlx::query_as::<_, (Uuid, String, i64)>(
+        r#"
+        SELECT community_id, status::text, COUNT(*) AS count
+        FROM workflows
+        GROUP BY community_id, status
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, status, count)| CommunityWorkflowCount {
+            community_id,
+            status,
+            count,
+        })
+        .collect())
+}
+
+/// Per-community git-repo count.
+#[derive(Debug)]
+pub struct CommunityGitRepoCount {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Number of git repos registered for this community.
+    pub count: i64,
+}
+
+/// Return git repo counts per community.
+pub async fn git_repo_counts(pool: &PgPool) -> Result<Vec<CommunityGitRepoCount>> {
+    let rows = sqlx::query_as::<_, (Uuid, i64)>(
+        r#"
+        SELECT community_id, COUNT(*) AS count
+        FROM git_repo_names
+        GROUP BY community_id
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, count)| CommunityGitRepoCount {
+            community_id,
+            count,
+        })
+        .collect())
+}
+
+/// Per-community active-user counts for a given window (e.g. 1d, 7d, 30d),
+/// split by human/agent.
+#[derive(Debug)]
+pub struct CommunityActiveUsers {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Distinct human pubkeys that published at least one event in the window.
+    /// A pubkey is human when its `users` row exists and `agent_owner_pubkey IS NULL`.
+    pub human: i64,
+    /// Distinct agent pubkeys that published at least one event in the window.
+    /// A pubkey is an agent when its `users` row exists and `agent_owner_pubkey IS NOT NULL`.
+    pub agent: i64,
+    /// Distinct pubkeys that published at least one event but have no `users` row.
+    /// Ingest does not guarantee a `users` row for every pubkey (profileless posters,
+    /// agents with missing rows). These are not classified and must not be folded into
+    /// `human` to avoid inflating the human count.
+    pub unknown: i64,
+}
+
+/// Return distinct-publisher counts for events in `[now - interval, now]`
+/// per community, split by human/agent/unknown.
+///
+/// `interval_sql` must be a trusted literal (e.g. `"1 day"`, `"7 days"`) —
+/// it is not user-controlled; callers are in the relay process.
+pub async fn active_user_counts(
+    pool: &PgPool,
+    interval_sql: &'static str,
+) -> Result<Vec<CommunityActiveUsers>> {
+    // LEFT JOIN users: pubkeys with no row have u.* = NULL.
+    // Three-way classification:
+    //   human   — row exists (u.pubkey IS NOT NULL) and agent_owner_pubkey IS NULL
+    //   agent   — row exists and agent_owner_pubkey IS NOT NULL
+    //   unknown — no row (u.pubkey IS NULL); not classified, reported separately
+    let sql = format!(
+        r#"
+        SELECT
+            e.community_id,
+            COUNT(DISTINCT e.pubkey)
+                FILTER (WHERE u.pubkey IS NOT NULL AND u.agent_owner_pubkey IS NULL)     AS human,
+            COUNT(DISTINCT e.pubkey)
+                FILTER (WHERE u.pubkey IS NOT NULL AND u.agent_owner_pubkey IS NOT NULL) AS agent,
+            COUNT(DISTINCT e.pubkey)
+                FILTER (WHERE u.pubkey IS NULL)                                          AS unknown
+        FROM events e
+        LEFT JOIN users u
+            ON u.community_id = e.community_id AND u.pubkey = e.pubkey
+        WHERE e.created_at >= NOW() - INTERVAL '{interval_sql}'
+          AND e.deleted_at IS NULL
+        GROUP BY e.community_id
+        "#
+    );
+    let rows = sqlx::query_as::<_, (Uuid, i64, i64, i64)>(sqlx::AssertSqlSafe(sql))
+        .fetch_all(pool)
+        .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(
+            |(community_id, human, agent, unknown)| CommunityActiveUsers {
+                community_id,
+                human,
+                agent,
+                unknown,
+            },
+        )
+        .collect())
+}
+
+/// Per-community active-channel counts for a given window.
+#[derive(Debug)]
+pub struct CommunityActiveChannels {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// Distinct channel IDs with ≥1 kind=9 message in the window.
+    pub count: i64,
+}
+
+/// Return distinct channel IDs with ≥1 kind=9 message in `[now - interval, now]`.
+pub async fn active_channel_counts(
+    pool: &PgPool,
+    interval_sql: &'static str,
+) -> Result<Vec<CommunityActiveChannels>> {
+    let sql = format!(
+        r#"
+        SELECT community_id, COUNT(DISTINCT channel_id) AS count
+        FROM events
+        WHERE kind = 9
+          AND channel_id IS NOT NULL
+          AND created_at >= NOW() - INTERVAL '{interval_sql}'
+          AND deleted_at IS NULL
+        GROUP BY community_id
+        "#
+    );
+    let rows = sqlx::query_as::<_, (Uuid, i64)>(sqlx::AssertSqlSafe(sql))
+        .fetch_all(pool)
+        .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(community_id, count)| CommunityActiveChannels {
+            community_id,
+            count,
+        })
+        .collect())
+}
+
+/// Mapping from community UUID to host string, used by the poller to resolve
+/// Prometheus label values.
+#[derive(Debug)]
+pub struct CommunityHost {
+    /// The UUID of the community.
+    pub id: Uuid,
+    /// The canonical host string for this community (used as the Prometheus label value).
+    pub host: String,
+}
+
+/// Fetch all community id → host mappings in one query.
+pub async fn community_hosts(pool: &PgPool) -> Result<Vec<CommunityHost>> {
+    let rows = sqlx::query_as::<_, (Uuid, String)>("SELECT id, host FROM communities")
+        .fetch_all(pool)
+        .await?;
+    Ok(rows
+        .into_iter()
+        .map(|(id, host)| CommunityHost { id, host })
+        .collect())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use buzz_core::CommunityId;
+    use nostr::Keys;
+    use sqlx::PgPool;
+
+    const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
+
+    async fn get_pool() -> PgPool {
+        PgPool::connect(TEST_DB_URL)
+            .await
+            .expect("connect to test DB")
+    }
+
+    fn random_pubkey() -> Vec<u8> {
+        Keys::generate().public_key().to_bytes().to_vec()
+    }
+
+    async fn make_community(pool: &PgPool) -> (Uuid, CommunityId, String) {
+        let id = uuid::Uuid::new_v4();
+        let host = format!("usage-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, CommunityId::from_uuid(id), host)
+    }
+
+    async fn insert_user(pool: &PgPool, community_id: Uuid, pubkey: &[u8], is_agent: bool) {
+        if is_agent {
+            let owner = random_pubkey();
+            // Insert owner first (FK constraint).
+            sqlx::query(
+                "INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+            )
+            .bind(community_id)
+            .bind(&owner)
+            .execute(pool)
+            .await
+            .expect("insert owner");
+            sqlx::query(
+                "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
+            )
+            .bind(community_id)
+            .bind(pubkey)
+            .bind(&owner)
+            .execute(pool)
+            .await
+            .expect("insert agent user");
+        } else {
+            sqlx::query(
+                "INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+            )
+            .bind(community_id)
+            .bind(pubkey)
+            .execute(pool)
+            .await
+            .expect("insert human user");
+        }
+    }
+
+    /// user_counts returns correct human/agent split and is scoped per community.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_user_counts_scoped_per_community() {
+        let pool = get_pool().await;
+        let (comm_a_uuid, _, _) = make_community(&pool).await;
+        let (comm_b_uuid, _, _) = make_community(&pool).await;
+
+        // Community A: insert 2 humans first, then 1 agent whose owner is one
+        // of those humans (reuses existing pubkey — no extra human row).
+        let human1 = random_pubkey();
+        let human2 = random_pubkey();
+        let agent_pk = random_pubkey();
+        insert_user(&pool, comm_a_uuid, &human1, false).await;
+        insert_user(&pool, comm_a_uuid, &human2, false).await;
+        // Insert agent with human1 as owner (human1 is already in users).
+        sqlx::query(
+            "INSERT INTO users (community_id, pubkey, agent_owner_pubkey)
+             VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
+        )
+        .bind(comm_a_uuid)
+        .bind(&agent_pk)
+        .bind(&human1)
+        .execute(&pool)
+        .await
+        .expect("insert agent user");
+
+        // Community B: 0 human, 1 agent (owner is a fresh human in comm_b).
+        let owner_b = random_pubkey();
+        insert_user(&pool, comm_b_uuid, &owner_b, false).await;
+        let agent_b = random_pubkey();
+        sqlx::query(
+            "INSERT INTO users (community_id, pubkey, agent_owner_pubkey)
+             VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
+        )
+        .bind(comm_b_uuid)
+        .bind(&agent_b)
+        .bind(&owner_b)
+        .execute(&pool)
+        .await
+        .expect("insert agent user b");
+
+        let counts = user_counts(&pool).await.expect("user_counts");
+
+        let a = counts.iter().find(|r| r.community_id == comm_a_uuid);
+        let b = counts.iter().find(|r| r.community_id == comm_b_uuid);
+
+        let a = a.expect("community A row");
+        assert_eq!(a.human, 2, "community A: 2 humans");
+        assert_eq!(a.agent, 1, "community A: 1 agent");
+
+        let b = b.expect("community B row");
+        assert_eq!(b.human, 1, "community B: 1 human (the agent owner)");
+        assert_eq!(b.agent, 1, "community B: 1 agent");
+    }
+
+    /// Deactivated users are excluded from user_counts.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_user_counts_excludes_deactivated() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+
+        let active_pk = random_pubkey();
+        let deactivated_pk = random_pubkey();
+
+        insert_user(&pool, comm_uuid, &active_pk, false).await;
+        insert_user(&pool, comm_uuid, &deactivated_pk, false).await;
+        // Deactivate the second user.
+        sqlx::query(
+            "UPDATE users SET deactivated_at = NOW() WHERE community_id = $1 AND pubkey = $2",
+        )
+        .bind(comm_uuid)
+        .bind(&deactivated_pk)
+        .execute(&pool)
+        .await
+        .expect("deactivate user");
+
+        let counts = user_counts(&pool).await.expect("user_counts");
+        let row = counts
+            .iter()
+            .find(|r| r.community_id == comm_uuid)
+            .expect("row");
+        assert_eq!(row.human, 1, "only active user counted");
+    }
+
+    /// channel_counts is scoped per community and excludes deleted channels.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_channel_counts_scoped_and_excludes_deleted() {
+        let pool = get_pool().await;
+        let (comm_uuid, comm_id, _) = make_community(&pool).await;
+        let owner = random_pubkey();
+        insert_user(&pool, comm_uuid, &owner, false).await;
+
+        // Insert a stream and a DM channel.
+        sqlx::query(
+            "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by)
+             VALUES ($1, $2, 'test-stream', 'stream', 'open', $3)",
+        )
+        .bind(uuid::Uuid::new_v4())
+        .bind(comm_uuid)
+        .bind(&owner)
+        .execute(&pool)
+        .await
+        .expect("insert stream channel");
+
+        let dm_id = uuid::Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by)
+             VALUES ($1, $2, 'test-dm', 'dm', 'private', $3)",
+        )
+        .bind(dm_id)
+        .bind(comm_uuid)
+        .bind(&owner)
+        .execute(&pool)
+        .await
+        .expect("insert dm channel");
+
+        // Soft-delete the DM.
+        sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1")
+            .bind(dm_id)
+            .execute(&pool)
+            .await
+            .expect("delete channel");
+
+        // Use comm_id to satisfy unused import warning.
+        let _ = comm_id;
+
+        let counts = channel_counts(&pool).await.expect("channel_counts");
+        let comm_counts: Vec<_> = counts
+            .iter()
+            .filter(|r| r.community_id == comm_uuid)
+            .collect();
+
+        // Only the stream channel should be counted.
+        assert_eq!(comm_counts.len(), 1);
+        assert_eq!(comm_counts[0].channel_type, "stream");
+        assert_eq!(comm_counts[0].count, 1);
+    }
+
+    /// community_hosts returns id → host mapping.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_community_hosts_returns_mapping() {
+        let pool = get_pool().await;
+        let (id, _, host) = make_community(&pool).await;
+
+        let hosts = community_hosts(&pool).await.expect("community_hosts");
+        let found = hosts.iter().find(|h| h.id == id);
+        assert!(found.is_some(), "inserted community not found");
+        assert_eq!(found.unwrap().host, host);
+    }
+
+    /// community_count reflects newly inserted communities.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_community_count_increases() {
+        let pool = get_pool().await;
+        let before = community_count(&pool).await.expect("count before");
+        make_community(&pool).await;
+        let after = community_count(&pool).await.expect("count after");
+        assert!(after > before, "count should increase after insert");
+    }
+
+    /// git_repo_counts queries git_repo_names (not git_repos) and is scoped per community.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_git_repo_counts_scoped_per_community() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let owner = random_pubkey();
+        insert_user(&pool, comm_uuid, &owner, false).await;
+        let owner_hex = hex::encode(&owner);
+
+        // Insert two repos for this community.
+        for repo_id in &["repo-alpha", "repo-beta"] {
+            sqlx::query(
+                "INSERT INTO git_repo_names (community_id, repo_id, owner_pubkey)
+                 VALUES ($1, $2, $3)
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(comm_uuid)
+            .bind(repo_id)
+            .bind(&owner_hex)
+            .execute(&pool)
+            .await
+            .expect("insert git repo");
+        }
+
+        let counts = git_repo_counts(&pool).await.expect("git_repo_counts");
+        let comm_counts: Vec<_> = counts
+            .iter()
+            .filter(|r| r.community_id == comm_uuid)
+            .collect();
+
+        assert_eq!(comm_counts.len(), 1, "one row per community");
+        assert_eq!(comm_counts[0].count, 2, "two repos");
+    }
+
+    /// active_user_counts classifies pubkeys with no users row as "unknown",
+    /// not "human" — the old LEFT JOIN treated NULL.agent_owner_pubkey as human.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_active_user_counts_unknown_bucket_for_profileless_poster() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+
+        // One known human (has a users row).
+        let human_pk = random_pubkey();
+        insert_user(&pool, comm_uuid, &human_pk, false).await;
+
+        // One profileless poster (no users row at all).
+        let profileless_pk = random_pubkey();
+
+        // Insert events for both pubkeys in this community.
+        let event_id1 = random_pubkey(); // 32-byte id
+        let event_id2 = random_pubkey();
+        let sig = vec![0u8; 64];
+        for (pk, eid) in [(&human_pk, &event_id1), (&profileless_pk, &event_id2)] {
+            sqlx::query(
+                "INSERT INTO events \
+                 (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \
+                 VALUES ($1, $2, $3, NOW(), 9, '[]', '', $4, NOW()) \
+                 ON CONFLICT DO NOTHING",
+            )
+            .bind(comm_uuid)
+            .bind(eid)
+            .bind(pk)
+            .bind(&sig)
+            .execute(&pool)
+            .await
+            .expect("insert event");
+        }
+
+        let counts = active_user_counts(&pool, "1 day")
+            .await
+            .expect("active_user_counts");
+        let row = counts.iter().find(|r| r.community_id == comm_uuid);
+        assert!(row.is_some(), "row for community must exist");
+        let row = row.unwrap();
+        assert_eq!(row.human, 1, "known human poster counts as human");
+        assert_eq!(row.agent, 0, "no agents");
+        assert_eq!(
+            row.unknown, 1,
+            "profileless poster must land in unknown, not human"
+        );
+    }
+
+    /// Regression: channel_counts returns no row for a community once all
+    /// channels of a type are soft-deleted.  The poller zero-fills from
+    /// host_map, so absence from this query is the correct "zero" signal.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_channel_counts_drops_to_zero_after_last_channel_deleted() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let owner = random_pubkey();
+        insert_user(&pool, comm_uuid, &owner, false).await;
+
+        // Insert one stream channel.
+        let ch_id = uuid::Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by)
+             VALUES ($1, $2, 'only-stream', 'stream', 'open', $3)",
+        )
+        .bind(ch_id)
+        .bind(comm_uuid)
+        .bind(&owner)
+        .execute(&pool)
+        .await
+        .expect("insert channel");
+
+        // Sanity: row present before deletion.
+        let before = channel_counts(&pool).await.expect("channel_counts before");
+        let before_row = before
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.channel_type == "stream");
+        assert_eq!(
+            before_row.map(|r| r.count),
+            Some(1),
+            "1 stream channel before deletion"
+        );
+
+        // Soft-delete the channel.
+        sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1")
+            .bind(ch_id)
+            .execute(&pool)
+            .await
+            .expect("soft-delete channel");
+
+        // After deletion: no row for this community+type — query returns nothing.
+        let after = channel_counts(&pool).await.expect("channel_counts after");
+        let after_row = after
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.channel_type == "stream");
+        assert!(
+            after_row.is_none(),
+            "no stream row after last channel deleted — poller will zero-fill"
+        );
+    }
+}

crates/buzz-db/src/user.rs

@@ -35,8 +35,12 @@ pub struct UserSearchProfile {
 
 /// Ensure a user record exists for the given pubkey (upsert).
 /// Creates with minimal fields if not present; no-op if already exists.
-pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result<()> {
-    sqlx::query(
+///
+/// Returns `true` if a new row was inserted, `false` if the user already existed.
+/// The `true` case is the reliable signal for "user was just registered" — used
+/// by callers to increment `buzz_users_created_total`.
+pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result<bool> {
+    let result = sqlx::query(
         r#"
         INSERT INTO users (community_id, pubkey)
         VALUES ($1, $2)
@@ -47,7 +51,7 @@ pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]
     .bind(pubkey)
     .execute(pool)
     .await?;
-    Ok(())
+    Ok(result.rows_affected() == 1)
 }
 
 /// Get a single user record by pubkey.

crates/buzz-relay/src/api/media.rs

@@ -358,7 +358,12 @@ pub async fn upload_blob(
         }
         _ => "other",
     };
-    metrics::counter!("buzz_media_uploads_total", "mime" => mime_label.to_owned()).increment(1);
+    metrics::counter!(
+        "buzz_media_uploads_total",
+        "mime" => mime_label.to_owned(),
+        "community" => auth.tenant.host().to_owned()
+    )
+    .increment(1);
 
     // Audit via bounded channel — same pattern as event audit.
     let desc = descriptor.clone();

crates/buzz-relay/src/connection.rs

@@ -157,7 +157,11 @@ pub async fn handle_connection(
     });
 
     info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established");
-    metrics::counter!("buzz_ws_connections_total").increment(1);
+    metrics::counter!(
+        "buzz_ws_connections_total",
+        "community" => conn.tenant.host().to_owned()
+    )
+    .increment(1);
 
     let challenge_msg = RelayMessage::auth_challenge(&challenge);
     if tx

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

@@ -257,19 +257,39 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
             if let Some(owner) = nip_oa_owner {
                 // Ensure both agent and owner have users rows (BYO agents may not,
                 // and agent_owner_pubkey has a FK constraint to users.pubkey).
-                if let Err(e) = state
+                match state
                     .db
                     .ensure_user(conn.tenant.community(), pubkey.as_bytes())
                     .await
                 {
-                    warn!(conn_id = %conn_id, error = %e, "ensure_user(agent) failed during NIP-OA backfill");
+                    Ok(true) => {
+                        metrics::counter!(
+                            "buzz_users_created_total",
+                            "community" => conn.tenant.host().to_owned()
+                        )
+                        .increment(1);
+                    }
+                    Ok(false) => {}
+                    Err(e) => {
+                        warn!(conn_id = %conn_id, error = %e, "ensure_user(agent) failed during NIP-OA backfill");
+                    }
                 }
-                if let Err(e) = state
+                match state
                     .db
                     .ensure_user(conn.tenant.community(), owner.as_bytes())
                     .await
                 {
-                    warn!(conn_id = %conn_id, error = %e, "ensure_user(owner) failed during NIP-OA backfill");
+                    Ok(true) => {
+                        metrics::counter!(
+                            "buzz_users_created_total",
+                            "community" => conn.tenant.host().to_owned()
+                        )
+                        .increment(1);
+                    }
+                    Ok(false) => {}
+                    Err(e) => {
+                        warn!(conn_id = %conn_id, error = %e, "ensure_user(owner) failed during NIP-OA backfill");
+                    }
                 }
 
                 // Idempotent backfill: record agent→owner in DB so cross-connection

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

@@ -42,12 +42,22 @@ pub async fn handle_command(
     // Ensure the authenticated user exists in the users table (foreign key requirement).
     // The old REST handlers did this via extract_auth_context; command executor must do it explicitly.
     let pubkey_bytes = auth.pubkey().to_bytes().to_vec();
-    if let Err(e) = state
+    match state
         .db
         .ensure_user(tenant.community(), &pubkey_bytes)
         .await
     {
-        tracing::warn!("command_executor: ensure_user failed: {e}");
+        Ok(true) => {
+            metrics::counter!(
+                "buzz_users_created_total",
+                "community" => tenant.host().to_owned()
+            )
+            .increment(1);
+        }
+        Ok(false) => {}
+        Err(e) => {
+            tracing::warn!("command_executor: ensure_user failed: {e}");
+        }
     }
 
     let kind = event.kind.as_u16() as u32;
@@ -362,6 +372,13 @@ async fn handle_dm_open(
 
     // 5. Side effects if newly created (post-commit, best-effort)
     if was_created {
+        metrics::counter!(
+            "buzz_channels_created_total",
+            "community" => tenant.host().to_owned(),
+            "type" => "dm"
+        )
+        .increment(1);
+
         // Invalidate caches for all participants
         for pk in &all_bytes {
             state.invalidate_membership(tenant, channel.id, pk);
@@ -516,6 +533,13 @@ async fn handle_dm_add_member(
 
     // 7. Cache invalidation + notifications for new DM (post-commit, best-effort)
     if was_created {
+        metrics::counter!(
+            "buzz_channels_created_total",
+            "community" => tenant.host().to_owned(),
+            "type" => "dm"
+        )
+        .increment(1);
+
         for pk in &all_bytes {
             state.invalidate_membership(tenant, new_channel.id, pk);
         }

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

@@ -43,6 +43,7 @@ fn bounded_kind_label(kind: u32) -> String {
         41001 | 41010..=41012 => kind.to_string(),
         43001..=43006 => kind.to_string(),
         44100..=44101 => kind.to_string(),
+        44200 => kind.to_string(),
         45001..=45003 => kind.to_string(),
         46001..=46012 | 46020 | 46030..=46031 => kind.to_string(),
         48001 | 48100..=48103 | 48106 => kind.to_string(),
@@ -509,6 +510,7 @@ async fn dispatch_persistent_event_inner(
         let workflow_engine = Arc::clone(&state.workflow_engine);
         let workflow_event = stored_event.clone();
         let trigger_kind = kind_u32.to_string();
+        let workflow_community_host = tenant.host().to_owned();
         // The event was stored under `tenant.community()`; `StoredEvent` does
         // not carry the community, so pass it explicitly. The same channel UUID
         // can exist in another community — scoping the workflow lookup to this
@@ -522,8 +524,12 @@ async fn dispatch_persistent_event_inner(
             {
                 tracing::error!(event_id = ?workflow_event.event.id, "Workflow trigger failed: {e}");
             } else {
-                metrics::counter!("buzz_workflow_runs_total", "trigger" => trigger_kind)
-                    .increment(1);
+                metrics::counter!(
+                    "buzz_workflow_runs_total",
+                    "trigger" => trigger_kind,
+                    "community" => workflow_community_host
+                )
+                .increment(1);
             }
         });
     }
@@ -585,7 +591,19 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
         .record("kind", kind_u32);
 
     debug!(event_id = %event_id_hex, kind = kind_u32, "EVENT");
+    // Fleet-wide received counter: kind-only, no community tag.
+    // Rationale: bounded_kind_label passes through all 10k values in
+    // 20000..=29999 (client-controlled ephemeral range). Crossing kind ×
+    // community would produce up to millions of series. Keep kind fleet-wide.
     metrics::counter!("buzz_events_received_total", "kind" => kind_str.clone()).increment(1);
+    // Per-community volume counter: community-only, no kind tag.
+    // Use this for per-community throughput graphs; the fleet counter above
+    // for per-kind breakdowns.
+    metrics::counter!(
+        "buzz_community_events_received_total",
+        "community" => conn.tenant.host().to_owned()
+    )
+    .increment(1);
 
     let (conn_id, pubkey_bytes, auth_pubkey, scopes, channel_ids) = {
         let auth = conn.auth_state.read().await;
@@ -684,6 +702,8 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
     match super::ingest::ingest_event(&state, &conn.tenant, event, ingest_auth).await {
         Ok(result) => {
             if result.accepted {
+                // Fleet-wide stored counter: kind-only, no community tag.
+                // Same cardinality rationale as buzz_events_received_total above.
                 metrics::counter!("buzz_events_stored_total", "kind" => kind_str).increment(1);
                 info!(
                     event_id = %result.event_id,

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

@@ -1997,6 +1997,12 @@ async fn ingest_event_inner(
                 });
             }
             pre_created_channel = Some(client_uuid);
+            metrics::counter!(
+                "buzz_channels_created_total",
+                "community" => tenant.host().to_owned(),
+                "type" => channel_type.to_string()
+            )
+            .increment(1);
         }
     }
 

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

@@ -97,7 +97,7 @@ pub async fn send_moderation_notice(
     // 1. Create/reuse the two-party DM channel {relay mod key, recipient}.
     //    `open_dm` is participant-hash idempotent, so re-delivery to the same
     //    user reuses the one thread per (community, user).
-    let (dm_channel, _was_created) = state
+    let (dm_channel, was_created) = state
         .db
         .open_dm(
             tenant.community(),
@@ -107,6 +107,17 @@ pub async fn send_moderation_notice(
         .await?;
     let dm_channel_id = dm_channel.id;
 
+    // Count new DM creation; side-effect gates below intentionally do not
+    // gate on was_created (see comment at step 2).
+    if was_created {
+        metrics::counter!(
+            "buzz_channels_created_total",
+            "community" => tenant.host().to_owned(),
+            "type" => "dm"
+        )
+        .increment(1);
+    }
+
     // Resurface the moderation DM for the recipient. `open_dm` only clears
     // `hidden_at` for `created_by` (the relay key), so a user who hid the
     // "{host} Moderation" thread would never see a later ban/resolution notice.

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

@@ -1073,10 +1073,17 @@ async fn handle_agent_profile(
         .ok_or_else(|| anyhow::anyhow!("kind:10100 missing channel_add_policy field"))?;
 
     let pubkey_bytes = event.pubkey.to_bytes().to_vec();
-    state
+    if state
         .db
         .ensure_user(tenant.community(), &pubkey_bytes)
-        .await?;
+        .await?
+    {
+        metrics::counter!(
+            "buzz_users_created_total",
+            "community" => tenant.host().to_owned()
+        )
+        .increment(1);
+    }
     state
         .db
         .set_channel_add_policy(tenant.community(), &pubkey_bytes, policy)
@@ -1124,10 +1131,17 @@ async fn handle_kind0_profile(
 
     let pubkey_bytes = event.pubkey.to_bytes().to_vec();
 
-    state
+    if state
         .db
         .ensure_user(tenant.community(), &pubkey_bytes)
-        .await?;
+        .await?
+    {
+        metrics::counter!(
+            "buzz_users_created_total",
+            "community" => tenant.host().to_owned()
+        )
+        .increment(1);
+    }
 
     // Pass all fields as Some — empty string clears the field in the DB.
     // This ensures kind:0 is treated as absolute state, not a partial update.
@@ -1653,13 +1667,21 @@ async fn handle_create_group(
     // If the event has an h-tag UUID, ingest_event() already created the channel
     // via create_channel_with_id(). Fetch it rather than creating a duplicate.
     // If no h-tag, fall back to the original auto-UUID creation path.
+    //
+    // Double-count analysis (C5): the counter increments below do NOT
+    // double-count vs. ingest.rs. For the h-tag path, ingest increments on
+    // was_created=true and this handler only reaches create_channel() on a DB
+    // lookup Err — an error recovery path where ingest's channel is
+    // inaccessible, so the counter correctly records a new creation. For the
+    // no-h-tag path, ingest never creates the channel, so this is the sole
+    // increment.
     let channel = if let Some(client_uuid) = extract_h_tag_channel(event) {
         match state.db.get_channel(tenant.community(), client_uuid).await {
             Ok(ch) => ch,
             Err(_) => {
                 // Channel not found — shouldn't happen (ingest_event pre-created it),
                 // but fall back to creation to stay resilient.
-                state
+                let ch = state
                     .db
                     .create_channel(
                         tenant.community(),
@@ -1670,11 +1692,18 @@ async fn handle_create_group(
                         &actor_bytes,
                         ttl_seconds,
                     )
-                    .await?
+                    .await?;
+                metrics::counter!(
+                    "buzz_channels_created_total",
+                    "community" => tenant.host().to_owned(),
+                    "type" => channel_type.to_string()
+                )
+                .increment(1);
+                ch
             }
         }
     } else {
-        state
+        let ch = state
             .db
             .create_channel(
                 tenant.community(),
@@ -1685,7 +1714,14 @@ async fn handle_create_group(
                 &actor_bytes,
                 ttl_seconds,
             )
-            .await?
+            .await?;
+        metrics::counter!(
+            "buzz_channels_created_total",
+            "community" => tenant.host().to_owned(),
+            "type" => channel_type.to_string()
+        )
+        .increment(1);
+        ch
     };
 
     // Creator becomes owner — evict any stale negative membership lookup.

crates/buzz-relay/src/main.rs

@@ -1,11 +1,14 @@
+use std::collections::HashMap;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use tracing::{error, info, warn};
 use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+use uuid::Uuid;
 
 use buzz_audit::AuditService;
 use buzz_auth::AuthService;
+use buzz_core::CommunityId;
 use buzz_db::{Db, DbConfig};
 use buzz_pubsub::PubSubManager;
 use buzz_search::SearchService;
@@ -26,6 +29,48 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool {
     })
 }
 
+/// Controls how many per-community gauge series the usage poller emits.
+///
+/// Datadog cost is proportional to the number of unique time-series.  With ~25
+/// gauge label combinations per community, a relay hosting thousands of
+/// communities would incur five-figure monthly costs if every community always
+/// gets a full set of series.  This knob is the cost lever.
+///
+/// Fleet-wide totals (`buzz_total_*`) always emit regardless of mode.
+///
+/// Set via `BUZZ_USAGE_METRICS_PER_COMMUNITY`:
+///   - `all` — emit per-community series for every community (default)
+///   - `off` — suppress all per-community series; fleet totals only
+///
+/// A `top:<k>` mode (per-community series for the k most-active communities)
+/// is planned as a fast-follow once the series-lifecycle (gauge idle-timeout
+/// and stable tie-breaking across pods) is fully designed.
+#[derive(Debug, Clone)]
+enum PerCommunityMode {
+    All,
+    Off,
+}
+
+impl PerCommunityMode {
+    fn from_env() -> Self {
+        let raw = std::env::var("BUZZ_USAGE_METRICS_PER_COMMUNITY")
+            .unwrap_or_default()
+            .trim()
+            .to_ascii_lowercase();
+        match raw.as_str() {
+            "" | "all" => PerCommunityMode::All,
+            "off" => PerCommunityMode::Off,
+            other => {
+                warn!(
+                    value = other,
+                    "BUZZ_USAGE_METRICS_PER_COMMUNITY: unknown value — defaulting to all"
+                );
+                PerCommunityMode::All
+            }
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     // Install the ring CryptoProvider for rustls. Required before any rustls
@@ -806,6 +851,50 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    // Usage metrics: periodic background task polling per-community stats.
+    //
+    // DB-derived gauges (users, channels, messages, members, workflows, git
+    // repos, active users/channels) are SET from GROUP BY queries — one per
+    // tick. In-memory gauges (ws_connections, subscriptions, users_online)
+    // are snapshotted from live in-memory state. Both avoid inc/dec drift.
+    //
+    // Multi-pod semantics:
+    //   DB-derived: all pods export the same value → dashboard uses max()
+    //   In-memory:  each pod exports its partition → dashboard uses sum()
+    {
+        let usage_state = Arc::clone(&state);
+        let per_community_mode = PerCommunityMode::from_env();
+        let interval_secs = std::env::var("BUZZ_USAGE_METRICS_INTERVAL_SECS")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(300) // 300s default: adoption stocks don't need minute freshness;
+            // if event-table rollups (message counts, DAU/WAU/MAU) become
+            // slow at scale, move them to a maintained rollup table and drop
+            // the interval back to 60s.
+            .max(5); // 5s minimum: cheaper than pool poller's 1s minimum
+        tokio::spawn(async move {
+            // Jitter the first tick by a random fraction of the interval so
+            // that a rolling deploy with N pods doesn't hammer the DB
+            // simultaneously at boot. Each pod picks a start delay in
+            // [0, interval_secs) using true per-process randomness (PID-derived
+            // seeds are unsafe in containers where the relay is typically PID 1
+            // in every pod, which would make all pods compute the same delay).
+            let jitter_secs = rand::random::<u64>() % interval_secs;
+            tokio::time::sleep(std::time::Duration::from_secs(jitter_secs)).await;
+
+            let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
+            // Skip a tick rather than scheduling a burst of catch-up ticks if
+            // the system falls behind (e.g. the previous tick took > interval).
+            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            loop {
+                interval.tick().await;
+                if let Err(e) = run_usage_metrics_tick(&usage_state, &per_community_mode).await {
+                    error!(error = %e, "Usage metrics tick failed — skipping");
+                }
+            }
+        });
+    }
+
     serve(router, health_router, Arc::clone(&state)).await?;
 
     // Signal the audit worker to stop accepting, flush buffered entries, and
@@ -971,6 +1060,415 @@ fn reminder_to_event(reminder: &buzz_db::event::DueReminder) -> nostr::Event {
     serde_json::from_value(event_json).expect("valid event JSON from DB row")
 }
 
+/// Run one tick of the usage metrics poller.
+///
+/// Queries the DB for per-community stock counts and snapshots in-memory
+/// state for connection/subscription gauges. On any DB error, returns `Err`
+/// so the caller can log and skip the tick without crashing.
+///
+/// All DB-derived gauges use absolute SET (not increment), so they self-heal
+/// after a missed tick or a pod restart.
+async fn run_usage_metrics_tick(
+    state: &AppState,
+    per_community_mode: &PerCommunityMode,
+) -> anyhow::Result<()> {
+    // --- community id → host label map (one query, cached for this tick) ---
+    let hosts = state.db.usage_community_hosts().await?;
+    let host_map: HashMap<Uuid, String> = hosts.into_iter().map(|c| (c.id, c.host)).collect();
+
+    // --- Collect all DB results before emitting any metrics (C4) ---
+    //
+    // All `.await?` calls happen here. If any query fails the function returns
+    // early — no metrics are emitted for this tick — preventing a mixed
+    // fresh/stale snapshot where later gauges retain their last value while
+    // earlier ones are updated.
+
+    let community_total = state.db.usage_community_count().await?;
+    let user_rows = state.db.usage_user_counts().await?;
+    let channel_rows = state.db.usage_channel_counts().await?;
+    let message_rows = state.db.usage_message_counts().await?;
+    let relay_member_rows = state.db.usage_relay_member_counts().await?;
+    let workflow_rows = state.db.usage_workflow_counts().await?;
+    let git_repo_rows = state.db.usage_git_repo_counts().await?;
+    let active_users_1d = state.db.usage_active_user_counts("1 day").await?;
+    let active_users_7d = state.db.usage_active_user_counts("7 days").await?;
+    let active_users_30d = state.db.usage_active_user_counts("30 days").await?;
+    let active_channels_1d = state.db.usage_active_channel_counts("1 day").await?;
+    let active_channels_7d = state.db.usage_active_channel_counts("7 days").await?;
+
+    // In-memory snapshots are infallible — snapshot once before publish phase.
+    let conns_snapshot = state.conn_manager.per_community_ws_connections();
+    let online_snapshot = state.conn_manager.per_community_users_online();
+    let subs_snapshot = state.sub_registry.per_community_subscriptions();
+
+    // --- Determine which community IDs receive per-community series (K1) ---
+    //
+    // `active_set` is the subset of host_map IDs that get per-community gauges
+    // this tick. Fleet-wide totals (buzz_total_*) always emit regardless.
+    let active_set: std::collections::HashSet<Uuid> = match per_community_mode {
+        PerCommunityMode::All => host_map.keys().copied().collect(),
+        PerCommunityMode::Off => std::collections::HashSet::new(),
+    };
+
+    // --- Publish phase: emit all metrics now that every query succeeded ---
+
+    // --- A. Adoption stocks (DB-polled) ---
+
+    // buzz_communities_total (no tag — fleet-wide count)
+    metrics::gauge!("buzz_communities_total").set(community_total as f64);
+
+    // buzz_community_users{community, type:human|agent}
+    // Emit from host_map so communities that have zero users still get a 0
+    // rather than keeping the last nonzero value until process restart.
+    {
+        let rows: HashMap<Uuid, _> = user_rows.into_iter().map(|r| (r.community_id, r)).collect();
+        // Fleet totals (always emitted).
+        let (total_human, total_agent): (i64, i64) = rows
+            .values()
+            .fold((0, 0), |(h, a), r| (h + r.human, a + r.agent));
+        metrics::gauge!("buzz_total_users", "type" => "human").set(total_human as f64);
+        metrics::gauge!("buzz_total_users", "type" => "agent").set(total_agent as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let (human, agent) = rows.get(&id).map(|r| (r.human, r.agent)).unwrap_or((0, 0));
+            metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "human")
+                .set(human as f64);
+            metrics::gauge!("buzz_community_users", "community" => community.clone(), "type" => "agent")
+                .set(agent as f64);
+        }
+    }
+
+    // buzz_community_channels{community, type}
+    // Zero-fill across all (community, channel_type) pairs so a type that
+    // drops to zero emits 0 rather than retaining its last nonzero value.
+    {
+        const CHANNEL_TYPES: &[&str] = &["stream", "forum", "dm", "workflow"];
+        let rows: HashMap<(Uuid, &str), i64> = channel_rows
+            .into_iter()
+            .filter_map(|r| {
+                let matched = CHANNEL_TYPES
+                    .iter()
+                    .find(|&&t| t == r.channel_type.as_str())
+                    .map(|&t| ((r.community_id, t), r.count));
+                if matched.is_none() {
+                    warn!(
+                        channel_type = %r.channel_type,
+                        "usage_channel_counts: unrecognised channel_type — row skipped"
+                    );
+                }
+                matched
+            })
+            .collect();
+        // Fleet totals (always emitted).
+        for &ct in CHANNEL_TYPES {
+            let total: i64 = host_map
+                .keys()
+                .map(|id| rows.get(&(*id, ct)).copied().unwrap_or(0))
+                .sum();
+            metrics::gauge!("buzz_total_channels", "type" => ct).set(total as f64);
+        }
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            for &ct in CHANNEL_TYPES {
+                let count = rows.get(&(id, ct)).copied().unwrap_or(0);
+                metrics::gauge!(
+                    "buzz_community_channels",
+                    "community" => community.clone(),
+                    "type" => ct
+                )
+                .set(count as f64);
+            }
+        }
+    }
+
+    // buzz_community_messages{community}
+    // Emit 0 for communities with no messages so dashboards don't stale-read.
+    {
+        let rows: HashMap<Uuid, i64> = message_rows
+            .into_iter()
+            .map(|r| (r.community_id, r.count))
+            .collect();
+        // Fleet total (always emitted).
+        let total: i64 = rows.values().sum();
+        metrics::gauge!("buzz_total_messages").set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = rows.get(&id).copied().unwrap_or(0);
+            metrics::gauge!("buzz_community_messages", "community" => community.clone())
+                .set(count as f64);
+        }
+    }
+
+    // buzz_community_relay_members{community, role}
+    // Zero-fill across all (community, role) pairs; relay_members.role is a
+    // CHECK constraint over {'owner', 'admin', 'member'}.
+    {
+        const RELAY_ROLES: &[&str] = &["owner", "admin", "member"];
+        let rows: HashMap<(Uuid, &str), i64> = relay_member_rows
+            .into_iter()
+            .filter_map(|r| {
+                let matched = RELAY_ROLES
+                    .iter()
+                    .find(|&&role| role == r.role.as_str())
+                    .map(|&role| ((r.community_id, role), r.count));
+                if matched.is_none() {
+                    warn!(
+                        role = %r.role,
+                        "usage_relay_member_counts: unrecognised role — row skipped"
+                    );
+                }
+                matched
+            })
+            .collect();
+        // Fleet totals (always emitted).
+        for &role in RELAY_ROLES {
+            let total: i64 = host_map
+                .keys()
+                .map(|id| rows.get(&(*id, role)).copied().unwrap_or(0))
+                .sum();
+            metrics::gauge!("buzz_total_relay_members", "role" => role).set(total as f64);
+        }
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            for &role in RELAY_ROLES {
+                let count = rows.get(&(id, role)).copied().unwrap_or(0);
+                metrics::gauge!(
+                    "buzz_community_relay_members",
+                    "community" => community.clone(),
+                    "role" => role
+                )
+                .set(count as f64);
+            }
+        }
+    }
+
+    // buzz_community_workflows{community, status}
+    // Zero-fill across all (community, status) pairs; workflow_status is a
+    // DB enum: {'active', 'disabled', 'archived'}.
+    {
+        const WORKFLOW_STATUSES: &[&str] = &["active", "disabled", "archived"];
+        let rows: HashMap<(Uuid, &str), i64> = workflow_rows
+            .into_iter()
+            .filter_map(|r| {
+                let matched = WORKFLOW_STATUSES
+                    .iter()
+                    .find(|&&s| s == r.status.as_str())
+                    .map(|&s| ((r.community_id, s), r.count));
+                if matched.is_none() {
+                    warn!(
+                        status = %r.status,
+                        "usage_workflow_counts: unrecognised workflow status — row skipped"
+                    );
+                }
+                matched
+            })
+            .collect();
+        // Fleet totals (always emitted).
+        for &status in WORKFLOW_STATUSES {
+            let total: i64 = host_map
+                .keys()
+                .map(|id| rows.get(&(*id, status)).copied().unwrap_or(0))
+                .sum();
+            metrics::gauge!("buzz_total_workflows", "status" => status).set(total as f64);
+        }
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            for &status in WORKFLOW_STATUSES {
+                let count = rows.get(&(id, status)).copied().unwrap_or(0);
+                metrics::gauge!(
+                    "buzz_community_workflows",
+                    "community" => community.clone(),
+                    "status" => status
+                )
+                .set(count as f64);
+            }
+        }
+    }
+
+    // buzz_community_git_repos{community}
+    // Emit 0 for communities with no repos.
+    {
+        let rows: HashMap<Uuid, i64> = git_repo_rows
+            .into_iter()
+            .map(|r| (r.community_id, r.count))
+            .collect();
+        // Fleet total (always emitted).
+        let total: i64 = rows.values().sum();
+        metrics::gauge!("buzz_total_git_repos").set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = rows.get(&id).copied().unwrap_or(0);
+            metrics::gauge!("buzz_community_git_repos", "community" => community.clone())
+                .set(count as f64);
+        }
+    }
+
+    // --- C. Engagement — windowed DAU/WAU/MAU + active channels ---
+    // Emit 0 for window/type/community combos that had no activity; this
+    // ensures a community that was active last tick but quiet this tick reads
+    // 0 rather than retaining its last nonzero value.
+
+    for (data, label) in [
+        (active_users_1d, "1d"),
+        (active_users_7d, "7d"),
+        (active_users_30d, "30d"),
+    ] {
+        let rows: HashMap<Uuid, _> = data.into_iter().map(|r| (r.community_id, r)).collect();
+        // Fleet totals (always emitted).
+        let (total_human, total_agent, total_unknown): (i64, i64, i64) =
+            rows.values().fold((0, 0, 0), |(h, a, u), r| {
+                (h + r.human, a + r.agent, u + r.unknown)
+            });
+        metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "human")
+            .set(total_human as f64);
+        metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "agent")
+            .set(total_agent as f64);
+        metrics::gauge!("buzz_total_active_users", "window" => label, "type" => "unknown")
+            .set(total_unknown as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let (human, agent, unknown) = rows
+                .get(&id)
+                .map(|r| (r.human, r.agent, r.unknown))
+                .unwrap_or((0, 0, 0));
+            metrics::gauge!(
+                "buzz_community_active_users",
+                "community" => community.clone(),
+                "window" => label,
+                "type" => "human"
+            )
+            .set(human as f64);
+            metrics::gauge!(
+                "buzz_community_active_users",
+                "community" => community.clone(),
+                "window" => label,
+                "type" => "agent"
+            )
+            .set(agent as f64);
+            metrics::gauge!(
+                "buzz_community_active_users",
+                "community" => community.clone(),
+                "window" => label,
+                "type" => "unknown"
+            )
+            .set(unknown as f64);
+        }
+    }
+
+    for (data, label) in [(active_channels_1d, "1d"), (active_channels_7d, "7d")] {
+        let rows: HashMap<Uuid, i64> = data
+            .into_iter()
+            .map(|r| (r.community_id, r.count))
+            .collect();
+        // Fleet total (always emitted).
+        let total: i64 = rows.values().sum();
+        metrics::gauge!("buzz_total_active_channels", "window" => label).set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = rows.get(&id).copied().unwrap_or(0);
+            metrics::gauge!(
+                "buzz_community_active_channels",
+                "community" => community.clone(),
+                "window" => label
+            )
+            .set(count as f64);
+        }
+    }
+
+    // --- D. Realtime — in-memory snapshots ---
+    // All three gauges are derived from the in-memory conn/sub registries.
+    // We emit from host_map so communities that just dropped to zero
+    // (last connection closed, all subs removed) receive an explicit 0
+    // rather than keeping the stale last value.
+
+    // buzz_community_ws_connections{community}
+    {
+        // Fleet total (always emitted).
+        let total: u64 = conns_snapshot.values().sum();
+        metrics::gauge!("buzz_total_ws_connections").set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = conns_snapshot
+                .get(&CommunityId::from_uuid(id))
+                .copied()
+                .unwrap_or(0);
+            metrics::gauge!("buzz_community_ws_connections", "community" => community.clone())
+                .set(count as f64);
+        }
+    }
+
+    // buzz_community_users_online_pod{community}
+    // This is a pod-local distinct count — a pubkey connected to N pods is
+    // counted once per pod. Dashboard queries should sum across pods to get
+    // the fleet-wide total (with the caveat that multi-pod connections are
+    // counted N times). The metric name "_pod" suffix makes this explicit.
+    {
+        // Fleet total (always emitted; same pod-local caveat applies).
+        let total: u64 = online_snapshot.values().sum();
+        metrics::gauge!("buzz_total_users_online_pod").set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = online_snapshot
+                .get(&CommunityId::from_uuid(id))
+                .copied()
+                .unwrap_or(0);
+            metrics::gauge!("buzz_community_users_online_pod", "community" => community.clone())
+                .set(count as f64);
+        }
+    }
+
+    // buzz_community_subscriptions{community}
+    {
+        // Fleet total (always emitted).
+        let total: u64 = subs_snapshot.values().sum();
+        metrics::gauge!("buzz_total_subscriptions").set(total as f64);
+        // Per-community series (gated by active_set).
+        for (&id, community) in &host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let count = subs_snapshot
+                .get(&CommunityId::from_uuid(id))
+                .copied()
+                .unwrap_or(0);
+            metrics::gauge!("buzz_community_subscriptions", "community" => community.clone())
+                .set(count as f64);
+        }
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::buzz_auto_migrate_enabled;

crates/buzz-relay/src/state.rs

@@ -1,5 +1,6 @@
 //! Shared application state — Arc-wrapped, shared across all connections.
 
+use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
@@ -208,6 +209,41 @@ impl ConnectionManager {
             .map(|entry| Arc::clone(&entry.subscriptions))
     }
 
+    /// Snapshot the number of live WebSocket connections per community.
+    ///
+    /// Returns a map from community UUID to connection count. Used by the
+    /// usage poller; snapshotting avoids per-community gauge drift from
+    /// mismatched inc/dec across async boundaries.
+    pub fn per_community_ws_connections(&self) -> HashMap<CommunityId, u64> {
+        let mut counts: HashMap<CommunityId, u64> = HashMap::new();
+        for entry in self.connections.iter() {
+            *counts.entry(entry.community_id).or_default() += 1;
+        }
+        counts
+    }
+
+    /// Snapshot the number of distinct authenticated pubkeys online per community.
+    ///
+    /// A pubkey connected to multiple pods will be counted once per pod — the
+    /// dashboard sums across pods, so per-pod partial counts are correct.
+    /// A pubkey connected twice on the same pod is counted once (distinct set).
+    pub fn per_community_users_online(&self) -> HashMap<CommunityId, u64> {
+        // community_id → set of pubkey bytes
+        let mut seen: HashMap<CommunityId, HashSet<Vec<u8>>> = HashMap::new();
+        for entry in self.connections.iter() {
+            if let Ok(lock) = entry.authenticated_pubkey.read() {
+                if let Some(pk) = lock.as_ref() {
+                    seen.entry(entry.community_id)
+                        .or_default()
+                        .insert(pk.clone());
+                }
+            }
+        }
+        seen.into_iter()
+            .map(|(cid, set)| (cid, set.len() as u64))
+            .collect()
+    }
+
     /// Return the authenticated pubkey for a connection, if any.
     pub fn pubkey_for(&self, conn_id: Uuid) -> Option<Vec<u8>> {
         self.connections

crates/buzz-relay/src/subscription.rs

@@ -351,6 +351,21 @@ impl SubscriptionRegistry {
         self.subs.len()
     }
 
+    /// Snapshot the number of active subscriptions per community.
+    ///
+    /// Used by the usage poller to emit `buzz_community_subscriptions{community}`.
+    /// Snapshotting avoids gauge drift from mismatched inc/dec across communities.
+    pub fn per_community_subscriptions(&self) -> std::collections::HashMap<CommunityId, u64> {
+        let mut counts: std::collections::HashMap<CommunityId, u64> =
+            std::collections::HashMap::new();
+        for conn_entry in self.subs.iter() {
+            for (_, community_id, _) in conn_entry.value().values() {
+                *counts.entry(*community_id).or_default() += 1;
+            }
+        }
+        counts
+    }
+
     fn push_match(
         &self,
         conn_id: ConnId,
@@ -1463,4 +1478,84 @@ mod tests {
             vec![(conn_b, "b".to_string())]
         );
     }
+
+    #[test]
+    fn per_community_subscriptions_snapshot_is_correctly_scoped() {
+        // Verify that per_community_subscriptions() returns the correct
+        // per-community counts and that removal keeps the snapshot accurate.
+        let registry = SubscriptionRegistry::new();
+        let community_a = CommunityId::from_uuid(Uuid::from_u128(0xaaaa));
+        let community_b = CommunityId::from_uuid(Uuid::from_u128(0xbbbb));
+        let conn_a = Uuid::new_v4();
+        let conn_b = Uuid::new_v4();
+        let channel = Uuid::new_v4();
+
+        // 2 subs in community A, 1 sub in community B.
+        registry.register_scoped(
+            community_a,
+            conn_a,
+            "a1".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            Some(channel),
+        );
+        registry.register_scoped(
+            community_a,
+            conn_a,
+            "a2".to_string(),
+            vec![Filter::new().kind(Kind::Metadata)],
+            None,
+        );
+        registry.register_scoped(
+            community_b,
+            conn_b,
+            "b1".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            Some(channel),
+        );
+
+        let snap = registry.per_community_subscriptions();
+        assert_eq!(snap.get(&community_a), Some(&2), "community A: 2 subs");
+        assert_eq!(snap.get(&community_b), Some(&1), "community B: 1 sub");
+
+        // Remove one sub from community A — snapshot should reflect it.
+        registry.remove_subscription(conn_a, "a1");
+        let snap2 = registry.per_community_subscriptions();
+        assert_eq!(
+            snap2.get(&community_a),
+            Some(&1),
+            "community A: 1 sub after removal"
+        );
+        assert_eq!(snap2.get(&community_b), Some(&1), "community B: unchanged");
+    }
+
+    #[test]
+    fn per_community_subscriptions_drops_to_zero_when_all_subs_removed() {
+        // Regression: when the last subscription for a community is removed,
+        // per_community_subscriptions() must return no entry (not stale nonzero).
+        // The poller zero-fills from host_map, so absence == 0 is correct.
+        let registry = SubscriptionRegistry::new();
+        let community = CommunityId::from_uuid(Uuid::from_u128(0xcccc));
+        let conn = Uuid::new_v4();
+
+        registry.register_scoped(
+            community,
+            conn,
+            "c1".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            None,
+        );
+
+        // Verify nonzero before removal.
+        let snap1 = registry.per_community_subscriptions();
+        assert_eq!(snap1.get(&community), Some(&1), "1 sub before removal");
+
+        // Remove the only sub — community should no longer appear in snapshot.
+        registry.remove_subscription(conn, "c1");
+        let snap2 = registry.per_community_subscriptions();
+        assert!(
+            snap2.get(&community).copied().unwrap_or(0) == 0,
+            "community must have 0 subs after last removal; got {:?}",
+            snap2.get(&community)
+        );
+    }
 }