From 03f7d6d0ee5cd96febaae1b82a8de42e675dfe48 Mon Sep 17 00:00:00 2001 From: Prem Pillai <238277+cloud-on-prem@users.noreply.github.com> Date: Mon, 10 Aug 2026 20:23:44 +1000 Subject: [PATCH] feat(relay): add opt-in usage analytics webhook Signed-off-by: Prem Pillai <238277+cloud-on-prem@users.noreply.github.com> --- .env.example | 13 + crates/buzz-relay/src/api/bridge.rs | 141 +++++++++ crates/buzz-relay/src/config.rs | 104 +++++++ crates/buzz-relay/src/handlers/ingest.rs | 13 + crates/buzz-relay/src/lib.rs | 1 + crates/buzz-relay/src/state.rs | 23 ++ crates/buzz-relay/src/usage_analytics.rs | 349 +++++++++++++++++++++++ 7 files changed, 644 insertions(+) create mode 100644 crates/buzz-relay/src/usage_analytics.rs diff --git a/.env.example b/.env.example index 0f7bbba6f1..f3ccd0a09e 100644 --- a/.env.example +++ b/.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) # ----------------------------------------------------------------------------- diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453f..619b998188 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/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> { + bridge_handler_test_state_with_usage_analytics(None).await + } + + async fn bridge_handler_test_state_with_usage_analytics( + endpoint: Option, + ) -> Option> { 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>>, body: Bytes) -> StatusCode { + let _ = tx.send(body.to_vec()).await; + StatusCode::NO_CONTENT + } + + let (event_tx, mut event_rx) = mpsc::channel::>(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" + ); + } } diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 037c6b1dd3..868e0ad806 100644 --- a/crates/buzz-relay/src/config.rs +++ b/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, + /// Optional relay-hosted policy shown on join surfaces. Disabled when no /// documents or age attestation are configured. pub join_policy: Option, @@ -391,6 +394,36 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result { Ok(url) } +fn parse_usage_analytics_endpoint(raw: Option<&str>) -> Result, 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 { 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, 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(); diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 7aea2ed02e..2272e5a396 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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) diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 314adad92e..714ebf0876 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/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; diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 14a50df7b7..acae7e6f31 100644 --- a/crates/buzz-relay/src/state.rs +++ b/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>, + + /// Best-effort usage-event sender, absent when the webhook is disabled. + pub(crate) usage_analytics: Option, } 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, diff --git a/crates/buzz-relay/src/usage_analytics.rs b/crates/buzz-relay/src/usage_analytics.rs new file mode 100644 index 0000000000..9f35e9af6b --- /dev/null +++ b/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 { + 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, +} + +fn build_client() -> Result { + 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 { + let client = build_client()?; + let (tx, mut rx) = mpsc::channel::(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(rx: &mut mpsc::Receiver, 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>, _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>) -> StatusCode { + hits.fetch_add(1, Ordering::SeqCst); + StatusCode::NO_CONTENT + } + + async fn redirect(State(destination): State) -> 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); + } +}