relay: add author_type label to buzz_events_stored_total

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,

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

@@ -36,7 +36,7 @@ pub(crate) fn not_found(msg: &str) -> (StatusCode, Json<serde_json::Value>) {
 /// `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::*;

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

@@ -252,106 +252,25 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, 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"
+                    );
                 }
             }
 

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<ConnectionState>, state: Arc<A
     // 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);
+    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<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);
+                // buzz_events_stored_total is emitted inside ingest_event()
+                // (shared WS/HTTP seam), not here.
                 info!(
                     event_id = %result.event_id,
                     kind = kind_u32,

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<u8> {
+        self.pubkey().to_bytes().to_vec()
+    }
+
     /// Permission scopes for this auth context.
     pub fn scopes(&self) -> &[Scope] {
         match self {
@@ -1299,6 +1304,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<AppState>,
+    tenant: &TenantContext,
+    author_pubkey_bytes: Vec<u8>,
+) -> &'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 +1354,14 @@ pub async fn ingest_event(
     event: Event,
     auth: IngestAuth,
 ) -> Result<IngestResult, IngestError> {
+    // 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));
+    // 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(
         state.tracer.clone(),
@@ -1328,6 +1371,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
@@ -2856,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};

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<moka::sync::Cache<(CommunityId, Vec<u8>, Vec<u8>), 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<moka::sync::Cache<(CommunityId, Vec<u8>), 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