From f9e4bcf473e0e65d28e7fc5369e7a0d0f26b6a50 Mon Sep 17 00:00:00 2001 From: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:42:36 -0700 Subject: [PATCH 1/2] relay: add author_type label to buzz_events_stored_total Tag the stored-events counter with author_type (human|agent) so Datadog can chart agent vs human message activity. The label is derived from users.agent_owner_pubkey via a small moka cache (10k entries, 300s TTL); unknown pubkeys and lookup errors count as "human" so the label never adds a failure path to ingest. The counter now emits at the shared ingest_event() seam instead of the WebSocket handler, which also fixes a pre-existing undercount: events ingested through the HTTP bridge were never counted. author_type is a 2-value label, so it merely doubles the existing kind-series cardinality, consistent with the file's documented cardinality rules. Co-authored-by: Dawn (sprout agent) Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> --- crates/buzz-relay/src/handlers/event.rs | 9 ++-- crates/buzz-relay/src/handlers/ingest.rs | 52 ++++++++++++++++++++++++ crates/buzz-relay/src/state.rs | 12 ++++++ 3 files changed, 68 insertions(+), 5 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index edac325105..3d096f934f 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -32,7 +32,7 @@ fn reject(reason: &'static str) { } /// Bound the `kind` label to prevent cardinality explosion from arbitrary Nostr kinds. -fn bounded_kind_label(kind: u32) -> String { +pub(crate) fn bounded_kind_label(kind: u32) -> String { match kind { 0..=9 | 1059 | 1063 => kind.to_string(), 8000..=8003 | 9000..=9022 | 9030..=9036 => kind.to_string(), @@ -598,7 +598,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc kind_str.clone()).increment(1); + metrics::counter!("buzz_events_received_total", "kind" => kind_str).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. @@ -705,9 +705,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); + // buzz_events_stored_total is emitted inside ingest_event() + // (shared WS/HTTP seam), not here. info!( event_id = %result.event_id, kind = kind_u32, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 707bf37efd..7a25e8e556 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1299,6 +1299,36 @@ fn validate_event_reminder(event: &Event) -> Result<(), &'static str> { Ok(()) } +/// Resolve the `author_type` metric label (`"agent"` / `"human"`) for an +/// event author, from `users.agent_owner_pubkey IS NOT NULL` via a +/// per-community cache. Metric-labeling only — never used for authorization. +/// Unknown pubkeys and lookup errors count as "human" (the label must not +/// add a failure path to ingest). +async fn author_type_label( + state: &Arc, + tenant: &TenantContext, + author_pubkey_bytes: Vec, +) -> &'static str { + let key = (tenant.community(), author_pubkey_bytes); + let cached = state.author_type_cache.get(&key); + let is_agent = match cached { + Some(v) => v, + None => { + let v = match state.db.get_agent_channel_policy(key.0, &key.1).await { + Ok(Some((_, owner))) => owner.is_some(), + Ok(None) | Err(_) => false, + }; + state.author_type_cache.insert(key, v); + v + } + }; + if is_agent { + "agent" + } else { + "human" + } +} + /// Ingest a signed Nostr event through the full validation pipeline. /// /// Shared by WebSocket and HTTP transports. The caller constructs [`IngestAuth`] @@ -1319,6 +1349,12 @@ pub async fn ingest_event( event: Event, auth: IngestAuth, ) -> Result { + // Captured before `event` moves into the inner fn: the stored-events + // counter below is emitted at this shared seam so WebSocket and HTTP + // transports are counted identically. + let kind_label = super::event::bounded_kind_label(event_kind_u32(&event)); + let author_pubkey_bytes = event.pubkey.to_bytes().to_vec(); + let abstract_state = state_for_request(tenant, auth.pubkey()); let (_guard, tracer) = EmitGuard::arm( state.tracer.clone(), @@ -1328,6 +1364,22 @@ pub async fn ingest_event( let result = ingest_event_inner(state, &tracer, tenant, event, auth).await; + // Fleet-wide stored counter: kind + author_type only, no community tag + // (see the cardinality rationale on buzz_events_received_total — + // author_type is a 2-value label so it merely doubles the kind series). + // Emitted here rather than per-transport so HTTP bridge ingests count too. + if let Ok(r) = &result { + if r.accepted { + let author_type = author_type_label(state, tenant, author_pubkey_bytes).await; + metrics::counter!( + "buzz_events_stored_total", + "kind" => kind_label, + "author_type" => author_type + ) + .increment(1); + } + } + // Map terminal error variants onto the closed SanitizedReason // alphabet (spec line 778). The inner fn's success path emits // WriteInsert/WriteInsertGlobal/WriteDuplicate explicitly at its diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index b363be3d7e..64f1f0feb5 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -547,6 +547,12 @@ pub struct AppState { /// Prevents repeated DB lookups from bursty observer traffic. #[allow(clippy::type_complexity)] pub observer_owner_cache: Arc, Vec), bool>>, + /// Cache for the `author_type` metric label on the ingest path. + /// Key: (community_id, author pubkey bytes). Value: is_agent + /// (`users.agent_owner_pubkey IS NOT NULL`). The mapping is + /// first-write-wins and set during auth before an agent's first event, + /// so a short TTL only bounds staleness for the rare backfill race. + pub author_type_cache: Arc), bool>>, /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`] /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to @@ -718,6 +724,12 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(300)) .build(), ), + author_type_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(300)) + .build(), + ), // Default to NoopTracer: production builds pay zero cost. // Conformance tests overwrite this with a JsonlTracer after // construction (see test helpers in From 91f8b9f227dfb9678bb3200aaaf9c99f0ba42ef0 Mon Sep 17 00:00:00 2001 From: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Date: Tue, 21 Jul 2026 11:01:51 -0700 Subject: [PATCH 2/2] fix(relay): classify authenticated NIP-OA agents Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> --- crates/buzz-relay/src/api/bridge.rs | 14 ++- crates/buzz-relay/src/api/mod.rs | 67 ++++++++++++- crates/buzz-relay/src/handlers/auth.rs | 115 ++++------------------- crates/buzz-relay/src/handlers/ingest.rs | 27 +++++- 4 files changed, 121 insertions(+), 102 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 6f8628111d..ff4265077f 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -626,13 +626,23 @@ pub async fn submit_event( .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid event JSON: {e}")))?; // Enforce relay membership (with NIP-OA fallback via x-auth-tag header). let auth_tag = headers.get("x-auth-tag").and_then(|v| v.to_str().ok()); - super::relay_members::enforce_relay_membership( + let nip_oa_owner = super::relay_members::enforce_relay_membership( &state, tenant.community(), &pubkey_bytes, auth_tag, ) - .await?; + .await? + .or_else(|| { + if !state.config.require_relay_membership { + super::relay_members::extract_nip_oa_owner(&pubkey_bytes, auth_tag) + } else { + None + } + }); + if let Some(owner) = nip_oa_owner { + super::relay_members::materialize_nip_oa_owner(&state, &tenant, &pubkey, &owner).await; + } let auth = IngestAuth::Http { pubkey, diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index 6e83f55345..d9f829433b 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -36,7 +36,7 @@ pub(crate) fn not_found(msg: &str) -> (StatusCode, Json) { /// `git/transport.rs`, and `audio/handler.rs`. pub mod relay_members { use axum::{http::StatusCode, response::Json}; - use buzz_core::tenant::CommunityId; + use buzz_core::{tenant::CommunityId, TenantContext}; use tracing::{debug, info}; use crate::state::AppState; @@ -166,6 +166,71 @@ pub mod relay_members { } } + /// Persist a cryptographically verified NIP-OA agent→owner relationship. + /// + /// Both principals are ensured first because `agent_owner_pubkey` has a + /// community-scoped foreign key. The mapping is first-write-wins; an + /// existing mapping is accepted only when it names the same owner. + pub async fn materialize_nip_oa_owner( + state: &AppState, + tenant: &TenantContext, + agent: &nostr::PublicKey, + owner: &nostr::PublicKey, + ) -> bool { + for (role, pubkey) in [("agent", agent), ("owner", owner)] { + match state + .db + .ensure_user(tenant.community(), pubkey.as_bytes()) + .await + { + Ok(true) => { + metrics::counter!( + "buzz_users_created_total", + "community" => tenant.host().to_owned() + ) + .increment(1); + } + Ok(false) => {} + Err(e) => { + tracing::warn!(%role, error = %e, "ensure_user failed during NIP-OA backfill"); + return false; + } + } + } + + let materialized = match state + .db + .set_agent_owner(tenant.community(), agent.as_bytes(), owner.as_bytes()) + .await + { + Ok(true) => true, + Ok(false) => state + .db + .is_agent_owner(tenant.community(), agent.as_bytes(), owner.as_bytes()) + .await + .unwrap_or(false), + Err(e) => { + tracing::warn!(error = %e, "failed to backfill agent_owner_pubkey"); + false + } + }; + + if materialized { + state + .author_type_cache + .insert((tenant.community(), agent.to_bytes().to_vec()), true); + state.observer_owner_cache.insert( + ( + tenant.community(), + agent.to_bytes().to_vec(), + owner.to_bytes().to_vec(), + ), + true, + ); + } + materialized + } + #[cfg(test)] mod tests { use super::*; diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 481d9f4b50..127f1fc40e 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -252,106 +252,25 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } }); - // Stash NIP-OA owner on the auth context (session-scoped) only if - // the DB confirms this owner relationship (first-write-wins). + // Stash NIP-OA owner on the auth context only after the shared + // backfill confirms the first-write-wins relationship. 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). - match state - .db - .ensure_user(conn.tenant.community(), pubkey.as_bytes()) - .await - { - 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"); - } - } - match state - .db - .ensure_user(conn.tenant.community(), owner.as_bytes()) - .await + if crate::api::relay_members::materialize_nip_oa_owner( + &state, + &conn.tenant, + &pubkey, + &owner, + ) + .await { - 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 - // features (observer frames, channel policy) work for BYO agents. - // Returns Ok(true) if written, Ok(false) if already owned by someone else. - match state - .db - .set_agent_owner(conn.tenant.community(), pubkey.as_bytes(), owner.as_bytes()) - .await - { - Ok(true) => { - // Successfully materialized — this owner is authoritative. - auth_ctx.agent_owner_pubkey = Some(owner); - // Pre-warm the observer cache to avoid stale negatives. - let cache_key = ( - conn.tenant.community(), - pubkey.to_bytes().to_vec(), - owner.to_bytes().to_vec(), - ); - state.observer_owner_cache.insert(cache_key, true); - } - Ok(false) => { - // Agent already owned by someone else. Verify if this - // owner matches the existing DB record before trusting it. - match state - .db - .is_agent_owner( - conn.tenant.community(), - pubkey.as_bytes(), - owner.as_bytes(), - ) - .await - { - Ok(true) => { - auth_ctx.agent_owner_pubkey = Some(owner); - } - Ok(false) => { - warn!( - conn_id = %conn_id, - agent = %pubkey.to_hex(), - nip_oa_owner = %owner.to_hex(), - "NIP-OA owner differs from DB owner — session will not get owner fast-path" - ); - } - Err(e) => { - warn!( - conn_id = %conn_id, - error = %e, - "is_agent_owner check failed after set_agent_owner conflict" - ); - } - } - } - Err(e) => { - warn!( - conn_id = %conn_id, - agent = %pubkey.to_hex(), - owner = %owner.to_hex(), - error = %e, - "failed to backfill agent_owner_pubkey" - ); - } + auth_ctx.agent_owner_pubkey = Some(owner); + } else { + warn!( + conn_id = %conn_id, + agent = %pubkey.to_hex(), + nip_oa_owner = %owner.to_hex(), + "NIP-OA owner could not be materialized" + ); } } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 7a25e8e556..415a9e8baf 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -91,6 +91,11 @@ impl IngestAuth { } } + /// Pubkey used for principal-scoped accounting and policy lookups. + pub fn principal_pubkey_bytes(&self) -> Vec { + self.pubkey().to_bytes().to_vec() + } + /// Permission scopes for this auth context. pub fn scopes(&self) -> &[Scope] { match self { @@ -1353,7 +1358,9 @@ pub async fn ingest_event( // counter below is emitted at this shared seam so WebSocket and HTTP // transports are counted identically. let kind_label = super::event::bounded_kind_label(event_kind_u32(&event)); - let author_pubkey_bytes = event.pubkey.to_bytes().to_vec(); + // Classify the authenticated principal, not the event envelope signer: + // NIP-59 gift wraps deliberately use an unrelated ephemeral pubkey. + let author_pubkey_bytes = auth.principal_pubkey_bytes(); let abstract_state = state_for_request(tenant, auth.pubkey()); let (_guard, tracer) = EmitGuard::arm( @@ -2908,6 +2915,24 @@ mod tests { ); } + #[test] + fn accounting_uses_authenticated_principal_pubkey() { + let principal = nostr::Keys::generate(); + let envelope_signer = nostr::Keys::generate(); + let auth = IngestAuth::Nip42 { + pubkey: principal.public_key(), + scopes: vec![], + channel_ids: None, + conn_id: Uuid::new_v4(), + }; + + assert_ne!(principal.public_key(), envelope_signer.public_key()); + assert_eq!( + auth.principal_pubkey_bytes(), + principal.public_key().to_bytes().to_vec() + ); + } + #[test] fn ingest_auth_is_http_returns_true_for_http_variant() { use crate::handlers::ingest::{HttpAuthMethod, IngestAuth};