diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index dd00ab1ff..0e747fbcf 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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 { + usage::community_count(&self.pool).await + } + + /// Return per-community user counts split by human/agent. + pub async fn usage_user_counts(&self) -> Result> { + usage::user_counts(&self.pool).await + } + + /// Return per-community channel counts by type. + pub async fn usage_channel_counts(&self) -> Result> { + usage::channel_counts(&self.pool).await + } + + /// Return per-community kind=9 message counts. + pub async fn usage_message_counts(&self) -> Result> { + usage::message_counts(&self.pool).await + } + + /// Return per-community relay-member counts by role. + pub async fn usage_relay_member_counts(&self) -> Result> { + usage::relay_member_counts(&self.pool).await + } + + /// Return per-community workflow counts by status. + pub async fn usage_workflow_counts(&self) -> Result> { + usage::workflow_counts(&self.pool).await + } + + /// Return per-community git-repo counts. + pub async fn usage_git_repo_counts(&self) -> Result> { + 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> { + 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> { + usage::active_channel_counts(&self.pool, interval_sql).await + } + + /// Return all community id → host mappings. + pub async fn usage_community_hosts(&self) -> Result> { + 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 { user::ensure_user(&self.pool, community_id, pubkey).await } diff --git a/crates/buzz-db/src/usage.rs b/crates/buzz-db/src/usage.rs new file mode 100644 index 000000000..f009dc6e0 --- /dev/null +++ b/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 { + 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> { + // 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> { + 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> { + 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> { + 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> { + 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> { + 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> { + // 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> { + 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> { + 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 { + 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" + ); + } +} diff --git a/crates/buzz-db/src/user.rs b/crates/buzz-db/src/user.rs index a4a5a9d7e..066fb5f5c 100644 --- a/crates/buzz-db/src/user.rs +++ b/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 { + 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. diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index c5498cc5e..b7b6db369 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/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(); diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 0a72f76dc..294c3b1b4 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/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 diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index a5d6be50c..dcb49df6a 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -257,19 +257,39 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, 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 diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index e3b19770b..bd3eff4de 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/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); } diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 31911e4c7..9e3bb05d8 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/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, state: Arc 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, state: Arc { 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, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index adfdaf2b2..69457e2df 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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); } } diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index 0e5b51ef2..8f57eea71 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/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. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index c64ae2853..33a95ee64 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/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. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e56e9dbff..df68ec270 100644 --- a/crates/buzz-relay/src/main.rs +++ b/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:` 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::().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::() % 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 299b599ae..8e6b96bb8 100644 --- a/crates/buzz-relay/src/state.rs +++ b/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 { + let mut counts: HashMap = 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 { + // community_id → set of pubkey bytes + let mut seen: HashMap>> = 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> { self.connections diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 419cf503c..9831e8eee 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/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 { + let mut counts: std::collections::HashMap = + 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) + ); + } }