feat(relay): add opt-in usage analytics webhook

Signed-off-by: Prem Pillai <238277+cloud-on-prem@users.noreply.github.com>

.env.example

@@ -129,6 +129,19 @@ RUST_LOG=buzz_relay=debug,buzz_datastore=info,buzz_db=debug,buzz_auth=debug,buzz
 # OTLP tracing endpoint (optional — leave unset to disable)
 # OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
 
+# -----------------------------------------------------------------------------
+# Usage analytics webhook
+# -----------------------------------------------------------------------------
+# Disabled by default. When configured, the relay posts one small JSON event
+# after each newly persisted kind:9 message. The payload contains the author's
+# public key, the opaque community UUID, and relay acceptance time; it never
+# includes message content or tags.
+# The public key is a stable pseudonym and may be personal data.
+#
+# Remote endpoints must use HTTPS. HTTP is accepted only for localhost or a
+# loopback IP so the webhook can be exercised during local development.
+# BUZZ_USAGE_ANALYTICS_ENDPOINT=https://analytics.example.com/events
+
 # -----------------------------------------------------------------------------
 # ACP (Agent Communication Protocol — buzz-acp harness)
 # -----------------------------------------------------------------------------

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

@@ -3370,7 +3370,14 @@ mod tests {
     ///
     /// Returns `None` when local Postgres is not reachable.
     async fn bridge_handler_test_state() -> Option<Arc<crate::state::AppState>> {
+        bridge_handler_test_state_with_usage_analytics(None).await
+    }
+
+    async fn bridge_handler_test_state_with_usage_analytics(
+        endpoint: Option<url::Url>,
+    ) -> Option<Arc<crate::state::AppState>> {
         let mut config = crate::config::Config::from_env().ok()?;
+        config.usage_analytics_endpoint = endpoint;
         config.database_url = TEST_DB_URL.to_string();
         // Use the real local Redis so enforce_http_admission can pass.
         config.redis_url =
@@ -3766,4 +3773,138 @@ mod tests {
             "attribution line must carry the pubkey;\nlog:\n{log}"
         );
     }
+
+    /// Proves the usage hook runs only after a new kind:9 event is persisted.
+    #[tokio::test(flavor = "multi_thread")]
+    #[ignore = "requires Postgres + Redis"]
+    async fn usage_analytics_emits_new_kind_nine_once() {
+        use axum::{body::Bytes, extract::State, http::StatusCode, routing::post, Router};
+        use tokio::sync::mpsc;
+
+        async fn capture(State(tx): State<mpsc::Sender<Vec<u8>>>, body: Bytes) -> StatusCode {
+            let _ = tx.send(body.to_vec()).await;
+            StatusCode::NO_CONTENT
+        }
+
+        let (event_tx, mut event_rx) = mpsc::channel::<Vec<u8>>(8);
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind capture server");
+        let endpoint = format!(
+            "http://{}/events",
+            listener.local_addr().expect("capture address")
+        )
+        .parse()
+        .expect("capture URL");
+        tokio::spawn(async move {
+            axum::serve(
+                listener,
+                Router::new()
+                    .route("/events", post(capture))
+                    .with_state(event_tx),
+            )
+            .await
+            .expect("capture server")
+        });
+
+        let state = bridge_handler_test_state_with_usage_analytics(Some(endpoint))
+            .await
+            .expect("local Postgres and Redis must be running");
+        let host = format!("bridge-usage-{}.local", uuid::Uuid::new_v4().simple());
+        let community_id = state
+            .db
+            .ensure_configured_community(&host)
+            .await
+            .expect("ensure community")
+            .id
+            .to_string();
+
+        let submit = |event: nostr::Event, keys: &Keys, description: &'static str| {
+            let (state, host, pubkey) = (state.clone(), host.clone(), keys.public_key().to_hex());
+            async move {
+                let body = serde_json::to_vec(&event).expect("serialize event");
+                assert_eq!(
+                    post_events(state, &host, &pubkey, &body).await,
+                    axum::http::StatusCode::OK,
+                    "{description} must be accepted"
+                );
+            }
+        };
+
+        let channel = uuid::Uuid::new_v4().to_string();
+        let author_a = Keys::generate();
+        let author_b = Keys::generate();
+        let create_channel = EventBuilder::new(Kind::Custom(9007), "")
+            .tags([
+                Tag::parse(["h", &channel]).expect("h tag"),
+                Tag::parse(["name", "usage-proof"]).expect("name tag"),
+                Tag::parse(["channel_type", "stream"]).expect("channel type tag"),
+                Tag::parse(["visibility", "open"]).expect("visibility tag"),
+            ])
+            .sign_with_keys(&author_a)
+            .expect("sign channel creation");
+        submit(create_channel, &author_a, "channel creation").await;
+
+        let message = |keys: &Keys, content: &str| {
+            EventBuilder::new(Kind::Custom(9), content)
+                .tag(Tag::parse(["h", &channel]).expect("h tag"))
+                .sign_with_keys(keys)
+                .expect("sign message")
+        };
+        let first = message(&author_a, "private-a");
+        submit(first.clone(), &author_a, "first message").await;
+        submit(first, &author_a, "duplicate message").await;
+        submit(
+            message(&author_a, "private-a-follow-up"),
+            &author_a,
+            "second message from first author",
+        )
+        .await;
+        submit(
+            message(&author_b, "private-b"),
+            &author_b,
+            "message from second author",
+        )
+        .await;
+
+        let mut captured = Vec::new();
+        loop {
+            let body = tokio::time::timeout(std::time::Duration::from_secs(20), event_rx.recv())
+                .await
+                .expect("timed out waiting for usage event")
+                .expect("capture channel closed");
+            let payload: serde_json::Value =
+                serde_json::from_slice(&body).expect("usage event JSON");
+            let accepted_at_ms = payload["accepted_at_ms"]
+                .as_i64()
+                .expect("acceptance timestamp");
+            let actor_id = payload["actor_id"].as_str().expect("actor id").to_owned();
+            assert_eq!(
+                payload,
+                serde_json::json!({
+                    "schema_version": 1,
+                    "type": "message_sent",
+                    "accepted_at_ms": accepted_at_ms,
+                    "actor_id": actor_id,
+                    "community_id": community_id,
+                })
+            );
+
+            let reached_barrier = actor_id == author_b.public_key().to_hex();
+            captured.push(actor_id);
+            if reached_barrier {
+                break;
+            }
+        }
+
+        assert_eq!(
+            captured,
+            vec![
+                author_a.public_key().to_hex(),
+                author_a.public_key().to_hex(),
+                author_b.public_key().to_hex()
+            ],
+            "only the three new kind:9 messages should emit"
+        );
+    }
 }

crates/buzz-relay/src/config.rs

@@ -277,6 +277,9 @@ pub struct Config {
     /// Hard timeout for one gateway delivery request.
     pub push_gateway_timeout: Duration,
 
+    /// Optional endpoint for privacy-minimal usage events.
+    pub(crate) usage_analytics_endpoint: Option<url::Url>,
+
     /// Optional relay-hosted policy shown on join surfaces. Disabled when no
     /// documents or age attestation are configured.
     pub join_policy: Option<JoinPolicyConfig>,
@@ -391,6 +394,36 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result<url::Url, ConfigError> {
     Ok(url)
 }
 
+fn parse_usage_analytics_endpoint(raw: Option<&str>) -> Result<Option<url::Url>, ConfigError> {
+    let Some(raw) = raw.map(str::trim).filter(|value| !value.is_empty()) else {
+        return Ok(None);
+    };
+    let url = url::Url::parse(raw).map_err(|_| {
+        ConfigError::InvalidValue(
+            "BUZZ_USAGE_ANALYTICS_ENDPOINT must be a valid absolute URL".to_string(),
+        )
+    })?;
+    let is_loopback = match url.host() {
+        Some(url::Host::Domain(host)) => host.eq_ignore_ascii_case("localhost"),
+        Some(url::Host::Ipv4(address)) => address.is_loopback(),
+        Some(url::Host::Ipv6(address)) => address.is_loopback(),
+        None => false,
+    };
+    if !(url.scheme() == "https" || (url.scheme() == "http" && is_loopback))
+        || url.host().is_none()
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(ConfigError::InvalidValue(
+            "BUZZ_USAGE_ANALYTICS_ENDPOINT must be an absolute HTTPS URL, or an HTTP loopback URL for local testing, without credentials, query, or fragment"
+                .to_string(),
+        ));
+    }
+    Ok(Some(url))
+}
+
 fn parse_bool(name: &str, default: bool) -> Result<bool, ConfigError> {
     match std::env::var(name) {
         Err(std::env::VarError::NotPresent) => Ok(default),
@@ -887,6 +920,15 @@ impl Config {
             Err(_) => 2_000,
         };
         let push_gateway_timeout = Duration::from_millis(push_gateway_timeout_millis);
+        let usage_analytics_endpoint = match std::env::var("BUZZ_USAGE_ANALYTICS_ENDPOINT") {
+            Ok(raw) => parse_usage_analytics_endpoint(Some(&raw))?,
+            Err(std::env::VarError::NotPresent) => None,
+            Err(std::env::VarError::NotUnicode(_)) => {
+                return Err(ConfigError::InvalidValue(
+                    "BUZZ_USAGE_ANALYTICS_ENDPOINT must be valid Unicode".to_string(),
+                ));
+            }
+        };
 
         const MAX_POLICY_MARKDOWN_BYTES: usize = 256 * 1024;
         let read_policy_markdown = |name: &str| -> Result<Option<String>, ConfigError> {
@@ -1037,6 +1079,7 @@ impl Config {
             push_executor_key_id,
             push_gateway_delivery_url,
             push_gateway_timeout,
+            usage_analytics_endpoint,
             join_policy,
             admin,
             web_dir,
@@ -1617,6 +1660,67 @@ mod tests {
         }
     }
 
+    #[test]
+    fn usage_analytics_endpoint_is_optional_and_accepts_secure_or_loopback_urls() {
+        assert!(parse_usage_analytics_endpoint(None)
+            .expect("unset endpoint")
+            .is_none());
+        assert!(parse_usage_analytics_endpoint(Some("   "))
+            .expect("blank endpoint")
+            .is_none());
+
+        for endpoint in [
+            "https://analytics.example.com/events",
+            "http://localhost:8080/events",
+            "http://127.0.0.1:8080/events",
+            "http://[::1]:8080/events",
+        ] {
+            assert_eq!(
+                parse_usage_analytics_endpoint(Some(endpoint))
+                    .expect("valid endpoint")
+                    .as_ref()
+                    .map(url::Url::as_str),
+                Some(endpoint)
+            );
+        }
+    }
+
+    #[test]
+    fn usage_analytics_endpoint_rejects_unsafe_or_ambiguous_urls() {
+        for endpoint in [
+            "http://analytics.example.com/events",
+            "ftp://analytics.example.com/events",
+            "analytics.example.com/events",
+            "/events",
+            "https://analytics.example.com/events?source=relay",
+            "https://analytics.example.com/events#fragment",
+        ] {
+            assert!(
+                parse_usage_analytics_endpoint(Some(endpoint)).is_err(),
+                "endpoint must be rejected: {endpoint}"
+            );
+        }
+
+        let username = "user";
+        let password = "secret";
+        let host = "analytics.example.com";
+        let mut username_endpoint = url::Url::parse("https://analytics.example.com/events")
+            .expect("base credential test URL");
+        username_endpoint
+            .set_username(username)
+            .expect("set test username");
+        assert!(parse_usage_analytics_endpoint(Some(username_endpoint.as_str())).is_err());
+
+        let mut credential_endpoint = username_endpoint;
+        credential_endpoint
+            .set_password(Some(password))
+            .expect("set test password");
+        let error = parse_usage_analytics_endpoint(Some(credential_endpoint.as_str()))
+            .expect_err("credentials must be rejected")
+            .to_string();
+        assert!(!error.contains(password) && !error.contains(host));
+    }
+
     #[test]
     fn invalid_push_gateway_timeout_is_not_silently_defaulted() {
         let _guard = ENV_MUTEX.lock().unwrap();

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

@@ -2942,6 +2942,19 @@ async fn ingest_event_inner(
         });
     }
 
+    // This runs after duplicate detection so each newly persisted kind:9
+    // message contributes at most one usage event. Enqueueing is best-effort
+    // and never changes message acceptance.
+    if let Some(sender) = state.usage_analytics.as_ref() {
+        if let Some(usage_event) = crate::usage_analytics::message_sent(
+            &stored_event.event,
+            &tenant.community().as_uuid().to_string(),
+            stored_event.received_at.timestamp_millis(),
+        ) {
+            let _ = sender.try_send(usage_event);
+        }
+    }
+
     if crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
         if let Err(e) =
             crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state)

crates/buzz-relay/src/lib.rs

@@ -3,6 +3,7 @@
 //! NIP-01 WebSocket relay for Buzz private team communication.
 
 mod admission;
+mod usage_analytics;
 
 /// REST API route handlers.
 pub mod api;

crates/buzz-relay/src/state.rs

@@ -717,6 +717,9 @@ pub struct AppState {
     /// byte-identically to a relay without the mesh. Access via
     /// [`AppState::mesh`].
     pub mesh: Arc<std::sync::OnceLock<crate::mesh_boot::MeshHandle>>,
+
+    /// Best-effort usage-event sender, absent when the webhook is disabled.
+    pub(crate) usage_analytics: Option<crate::usage_analytics::UsageAnalyticsSender>,
 }
 
 impl AppState {
@@ -781,6 +784,25 @@ impl AppState {
             tracing::warn!("audit log worker exited (expected on shutdown)");
         });
 
+        let usage_analytics = config
+            .usage_analytics_endpoint
+            .clone()
+            .and_then(|endpoint| {
+                match crate::usage_analytics::UsageAnalyticsSender::spawn(
+                    endpoint,
+                    crate::usage_analytics::QUEUE_CAPACITY,
+                ) {
+                    Ok(sender) => Some(sender),
+                    Err(error) => {
+                        tracing::warn!(
+                            error = %error.without_url(),
+                            "usage analytics disabled: HTTP client initialization failed"
+                        );
+                        None
+                    }
+                }
+            });
+
         let git_max_concurrent_ops = config.git_max_concurrent_ops;
         let media_max_concurrent_uploads = config.media_max_concurrent_uploads;
         let git_store = crate::api::git::store::GitStore::new(
@@ -890,6 +912,7 @@ impl AppState {
             // `crates/buzz-test-client` once those land).
             tracer: Arc::new(crate::conformance::NoopTracer),
             mesh: Arc::new(std::sync::OnceLock::new()),
+            usage_analytics,
         };
         (
             state,

crates/buzz-relay/src/usage_analytics.rs

@@ -0,0 +1,349 @@
+//! Best-effort usage-event delivery for relay operators.
+//!
+//! The exporter is deliberately small: it emits one privacy-minimal JSON
+//! object for each newly persisted kind:9 message. Delivery never blocks event
+//! ingestion; a bounded queue drops analytics work when its single worker
+//! cannot keep up.
+
+use std::time::Duration;
+
+use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE};
+use nostr::Event;
+use serde::Serialize;
+use tokio::sync::mpsc;
+use tracing::warn;
+
+const EVENT_TYPE_MESSAGE_SENT: &str = "message_sent";
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Fixed queue depth. Usage analytics must not add backpressure to ingestion.
+pub(crate) const QUEUE_CAPACITY: usize = 1024;
+
+/// Minimal usage event sent to the configured webhook.
+#[derive(Clone, Serialize)]
+pub(crate) struct UsageEvent {
+    schema_version: u8,
+    #[serde(rename = "type")]
+    event_type: &'static str,
+    accepted_at_ms: i64,
+    actor_id: String,
+    community_id: String,
+}
+
+/// Build an event for a kind:9 message, or `None` for every other kind.
+///
+/// This reads only the event kind and author public key. Content, tags, event
+/// identifiers, signatures, and channel or thread identifiers are not used.
+pub(crate) fn message_sent(
+    event: &Event,
+    community_id: &str,
+    accepted_at_ms: i64,
+) -> Option<UsageEvent> {
+    if event_kind_u32(event) != KIND_STREAM_MESSAGE {
+        return None;
+    }
+
+    Some(UsageEvent {
+        schema_version: 1,
+        event_type: EVENT_TYPE_MESSAGE_SENT,
+        accepted_at_ms,
+        actor_id: event.pubkey.to_hex(),
+        community_id: community_id.to_owned(),
+    })
+}
+
+/// Non-blocking handle to the single usage-analytics delivery worker.
+#[derive(Clone)]
+pub(crate) struct UsageAnalyticsSender {
+    tx: mpsc::Sender<UsageEvent>,
+}
+
+fn build_client() -> Result<reqwest::Client, reqwest::Error> {
+    reqwest::Client::builder()
+        .timeout(REQUEST_TIMEOUT)
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+}
+
+impl UsageAnalyticsSender {
+    /// Start one worker with a bounded queue.
+    pub(crate) fn spawn(endpoint: url::Url, capacity: usize) -> Result<Self, reqwest::Error> {
+        let client = build_client()?;
+        let (tx, mut rx) = mpsc::channel::<UsageEvent>(capacity);
+
+        tokio::spawn(async move {
+            while let Some(event) = rx.recv().await {
+                post(&client, &endpoint, &event).await;
+            }
+        });
+
+        Ok(Self { tx })
+    }
+
+    /// Enqueue without waiting. Returns `false` when the event was dropped.
+    pub(crate) fn try_send(&self, event: UsageEvent) -> bool {
+        match self.tx.try_send(event) {
+            Ok(()) => true,
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                metrics::counter!(
+                    "buzz_usage_analytics_events_dropped_total",
+                    "reason" => "queue_full"
+                )
+                .increment(1);
+                warn!("usage analytics event dropped: queue full");
+                false
+            }
+            Err(mpsc::error::TrySendError::Closed(_)) => {
+                metrics::counter!(
+                    "buzz_usage_analytics_events_dropped_total",
+                    "reason" => "worker_stopped"
+                )
+                .increment(1);
+                warn!("usage analytics event dropped: worker stopped");
+                false
+            }
+        }
+    }
+}
+
+async fn post(client: &reqwest::Client, endpoint: &url::Url, event: &UsageEvent) {
+    match client.post(endpoint.clone()).json(event).send().await {
+        Ok(response) if response.status().is_success() => {
+            metrics::counter!(
+                "buzz_usage_analytics_deliveries_total",
+                "outcome" => "accepted"
+            )
+            .increment(1);
+        }
+        Ok(response) => {
+            metrics::counter!(
+                "buzz_usage_analytics_deliveries_total",
+                "outcome" => "http_error"
+            )
+            .increment(1);
+            warn!(
+                status = response.status().as_u16(),
+                "usage analytics webhook rejected an event"
+            );
+        }
+        Err(error) => {
+            metrics::counter!(
+                "buzz_usage_analytics_deliveries_total",
+                "outcome" => "transport_error"
+            )
+            .increment(1);
+            warn!(
+                error = %error.without_url(),
+                "usage analytics webhook request failed"
+            );
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
+    use std::time::Duration;
+
+    use axum::{
+        body::Bytes, extract::State, http::StatusCode, response::Redirect,
+        routing::post as post_route, Router,
+    };
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+    use tokio::sync::Semaphore;
+
+    use super::*;
+
+    const ACCEPTED_AT_MS: i64 = 1_752_000_000_000;
+    const PRIVATE_CONTENT: &str = "private message for marker@example.invalid";
+    const CHANNEL_ID: &str = "3f2504e0-4f89-11d3-9a0c-0305e82c3301";
+    const THREAD_ID: &str = "aabbccddeeff00112233445566778899aabbccddeeff001122334455667788ff";
+    const PROFILE_NAME: &str = "Synthetic Profile";
+    const NIP05_IDENTIFIER: &str = "nip05-marker@example.invalid";
+    const IP_ADDRESS: &str = "203.0.113.42";
+    const DEVICE_ID: &str = "device-installation-abc123";
+    const HOSTNAME: &str = "member.example.net";
+
+    fn signed_event(keys: &Keys, kind: u32) -> Event {
+        EventBuilder::new(Kind::from(kind as u16), PRIVATE_CONTENT)
+            .tags([
+                Tag::parse(["h", CHANNEL_ID]).expect("valid h tag"),
+                Tag::parse(["e", THREAD_ID]).expect("valid e tag"),
+                Tag::parse(["profile", PROFILE_NAME]).expect("valid profile tag"),
+                Tag::parse(["nip05", NIP05_IDENTIFIER]).expect("valid NIP-05 tag"),
+                Tag::parse(["ip", IP_ADDRESS]).expect("valid IP tag"),
+                Tag::parse(["device", DEVICE_ID]).expect("valid device tag"),
+                Tag::parse(["hostname", HOSTNAME]).expect("valid hostname tag"),
+            ])
+            .sign_with_keys(keys)
+            .expect("sign event")
+    }
+
+    fn sample_event(keys: &Keys) -> UsageEvent {
+        message_sent(
+            &signed_event(keys, KIND_STREAM_MESSAGE),
+            "community-opaque-1",
+            ACCEPTED_AT_MS,
+        )
+        .expect("kind:9 produces usage event")
+    }
+
+    async fn serve(app: Router) -> url::Url {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind capture server");
+        let endpoint = format!(
+            "http://{}/events",
+            listener.local_addr().expect("capture address")
+        )
+        .parse()
+        .expect("capture URL");
+        tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
+        endpoint
+    }
+
+    async fn next<T>(rx: &mut mpsc::Receiver<T>, description: &str) -> T {
+        tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .unwrap_or_else(|_| panic!("timed out waiting for {description}"))
+            .unwrap_or_else(|| panic!("channel closed waiting for {description}"))
+    }
+
+    #[test]
+    fn payload_is_exact_and_excludes_message_details() {
+        let keys = Keys::generate();
+        let event = signed_event(&keys, KIND_STREAM_MESSAGE);
+        let usage = message_sent(&event, "community-opaque-1", ACCEPTED_AT_MS)
+            .expect("kind:9 produces usage event");
+
+        assert_eq!(
+            serde_json::to_value(&usage).expect("serialize"),
+            serde_json::json!({
+                "schema_version": 1,
+                "type": "message_sent",
+                "accepted_at_ms": 1_752_000_000_000i64,
+                "actor_id": keys.public_key().to_hex(),
+                "community_id": "community-opaque-1"
+            })
+        );
+
+        let wire = serde_json::to_string(&usage).expect("serialize");
+        for (label, forbidden) in [
+            ("message content", PRIVATE_CONTENT),
+            ("email address", "marker@example.invalid"),
+            ("channel id", CHANNEL_ID),
+            ("thread id", THREAD_ID),
+            ("profile name", PROFILE_NAME),
+            ("NIP-05 identifier", NIP05_IDENTIFIER),
+            ("IP address", IP_ADDRESS),
+            ("device id", DEVICE_ID),
+            ("hostname", HOSTNAME),
+            ("Nostr event id", event.id.to_hex().as_str()),
+            ("signature", event.sig.to_string().as_str()),
+        ] {
+            assert!(!wire.contains(forbidden), "{label} must not be exported");
+        }
+    }
+
+    #[test]
+    fn only_kind_nine_produces_an_event() {
+        use buzz_core::kind::{
+            KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_V2, KIND_SYSTEM_MESSAGE, KIND_TEXT_NOTE,
+        };
+
+        let keys = Keys::generate();
+        for kind in [
+            0,
+            KIND_TEXT_NOTE,
+            7,
+            8,
+            10,
+            KIND_STREAM_MESSAGE_V2,
+            KIND_STREAM_MESSAGE_EDIT,
+            KIND_SYSTEM_MESSAGE,
+        ] {
+            assert!(
+                message_sent(&signed_event(&keys, kind), "community", ACCEPTED_AT_MS).is_none(),
+                "kind:{kind} must not produce a usage event"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn full_queue_drops_without_blocking_and_worker_continues_after_failure() {
+        struct Gate {
+            arrived: mpsc::Sender<()>,
+            hits: AtomicUsize,
+            release: Semaphore,
+        }
+
+        async fn gated(State(gate): State<Arc<Gate>>, _body: Bytes) -> StatusCode {
+            let hit = gate.hits.fetch_add(1, Ordering::SeqCst);
+            let _ = gate.arrived.send(()).await;
+            if hit == 0 {
+                let _permit = gate.release.acquire().await.expect("gate open");
+                StatusCode::INTERNAL_SERVER_ERROR
+            } else {
+                StatusCode::NO_CONTENT
+            }
+        }
+
+        let (arrived_tx, mut arrived_rx) = mpsc::channel(4);
+        let gate = Arc::new(Gate {
+            arrived: arrived_tx,
+            hits: AtomicUsize::new(0),
+            release: Semaphore::new(0),
+        });
+        let endpoint = serve(
+            Router::new()
+                .route("/events", post_route(gated))
+                .with_state(gate.clone()),
+        )
+        .await;
+        let sender = UsageAnalyticsSender::spawn(endpoint, 1).expect("analytics client");
+        let event = sample_event(&Keys::generate());
+
+        assert!(sender.try_send(event.clone()));
+        next(&mut arrived_rx, "first delivery").await;
+        assert!(sender.try_send(event.clone()));
+        assert!(!sender.try_send(event));
+
+        gate.release.add_permits(1);
+        next(&mut arrived_rx, "delivery after HTTP failure").await;
+        assert_eq!(gate.hits.load(Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn client_refuses_redirects() {
+        async fn destination(State(hits): State<Arc<AtomicUsize>>) -> StatusCode {
+            hits.fetch_add(1, Ordering::SeqCst);
+            StatusCode::NO_CONTENT
+        }
+
+        async fn redirect(State(destination): State<String>) -> Redirect {
+            Redirect::temporary(&destination)
+        }
+
+        let hits = Arc::new(AtomicUsize::new(0));
+        let destination = serve(
+            Router::new()
+                .route("/events", post_route(destination))
+                .with_state(hits.clone()),
+        )
+        .await;
+        let endpoint = serve(
+            Router::new()
+                .route("/events", post_route(redirect))
+                .with_state(destination.to_string()),
+        )
+        .await;
+
+        let client = build_client().expect("analytics client");
+        post(&client, &endpoint, &sample_event(&Keys::generate())).await;
+
+        assert_eq!(hits.load(Ordering::SeqCst), 0);
+    }
+}