feat(relay): push per-community storage usage to kgoose (#8714)

### Why?
To bill hosted communities for storage in Orb, kgoose needs
per-community storage usage. The relay already computes it in a daily
snapshot, but nothing delivered it anywhere.

## Summary
- Adds opt-in `BUZZ_STORAGE_USAGE_URL`. When it is set, one
leader-elected relay pod checks every 5 minutes for a new storage
snapshot and POSTs it to kgoose's `POST /v1/buzz/storage-usage/record`
(squareup/cash-server#144285) in consecutive batches of up to 50
communities until every community is sent.
- Each batch is `{measured_at, communities: [{community_host,
storage_bytes, storage_objects}]}`. Active communities with no stored
bytes are reported as zero, and re-sends are safe because kgoose bills
each community once per UTC day.
- Retries on 408, 429, 5xx, and transport errors. No migration;
documented in `docs/storage-accounting.md`. kgoose authenticates the
relay as the `buzz` mesh service, so the relay needs mesh egress to
kgoose-builderbot.

### Related issue
None found. Complements #7899 (snapshot history in the DB). Receiving
endpoint: squareup/cash-server#144285.

### Testing
Unit tests cover batching, retry and reject handling, zero-filling, and
a real HTTP delivery that checks the exact kgoose request body. A
Postgres-gated DB test covers the new reads. End to end, a local relay
delivered to a fake endpoint once per new snapshot.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Shu Hui <shu@squareup.com>
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

.env.example

@@ -162,6 +162,13 @@ BUZZ_PUSH_GATEWAY_DELIVERY_URL=https://push.buzz.xyz/v1/deliveries/apns
 # BUZZ_OPERATOR_LISTENER_TIMEOUT_MS=5000
 # BUZZ_OPERATOR_LISTENERS=<64-char hex pubkey>:https://listener.example/mentions
 
+# Optional endpoint that receives each new per-community storage snapshot
+# written by `buzz-admin storage-snapshot`, in NIP-98 relay-signed POST batches
+# shaped like kgoose's storage-usage record request. One relay pod checks for a
+# new snapshot every 5 minutes. Receivers must treat a repeated community and
+# measured_at as already recorded.
+# BUZZ_STORAGE_USAGE_URL=https://kgoose.example/cash-app/goose/v1/buzz/storage-usage/record
+
 # Optional relay-owned KLIPY key. When set, NIP-11 advertises GIF search and
 # authenticated desktop clients use this relay as the metadata/search proxy.
 # Keep the real value in your deployment's secret manager; never commit it.

crates/buzz-db/src/store/storage_accounting.rs

@@ -3,6 +3,7 @@
 use buzz_datastore_tracing::datastore_span;
 use chrono::{DateTime, Utc};
 use sqlx::postgres::PgConnection;
+use uuid::Uuid;
 
 use crate::{observability, Db, Result};
 
@@ -14,6 +15,17 @@ pub struct StorageAccountingLeader {
     connection: PgConnection,
 }
 
+/// A community that storage-usage delivery reports on.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StorageUsageCommunity {
+    /// The community UUID used by the snapshot's per-community map.
+    pub id: Uuid,
+    /// The canonical host stored on the community row.
+    pub host: String,
+    /// Whether the community is serving: not archived and not being deleted.
+    pub active: bool,
+}
+
 /// The newest complete snapshot stored by the worker.
 #[derive(Debug, Clone)]
 pub struct StoredStorageSnapshot {
@@ -100,6 +112,39 @@ impl Db {
             },
         ))
     }
+
+    /// Load only the newest snapshot's commit time, so a poller can detect a
+    /// new snapshot without decoding the full per-community payload.
+    #[datastore_span(name = "load_storage_accounting_completed_at", system = "postgresql")]
+    pub async fn load_storage_accounting_completed_at(&self) -> Result<Option<DateTime<Utc>>> {
+        let mut connection =
+            observability::acquire_writer(&self.pool, observability::WriterOperation::Maintenance)
+                .await?;
+        Ok(sqlx::query_scalar::<_, DateTime<Utc>>(
+            "SELECT completed_at FROM storage_accounting_snapshots WHERE singleton = TRUE",
+        )
+        .fetch_optional(&mut *connection)
+        .await?)
+    }
+
+    /// List every community with its host and whether it is still serving.
+    #[datastore_span(name = "storage_usage_communities", system = "postgresql")]
+    pub async fn storage_usage_communities(&self) -> Result<Vec<StorageUsageCommunity>> {
+        let mut connection =
+            observability::acquire_writer(&self.pool, observability::WriterOperation::Maintenance)
+                .await?;
+        let rows = sqlx::query_as::<_, (Uuid, String, bool)>(
+            "SELECT id, host, \
+               (archived_at IS NULL AND deleted_at IS NULL AND deletion_state = 'active') \
+             FROM communities",
+        )
+        .fetch_all(&mut *connection)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|(id, host, active)| StorageUsageCommunity { id, host, active })
+            .collect())
+    }
 }
 
 #[cfg(test)]
@@ -212,4 +257,84 @@ mod postgres_tests {
         assert_eq!(stored.max_objects, 200);
         assert_eq!(stored.code_sha, "b");
     }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn usage_delivery_reads_snapshot_time_and_community_states() {
+        let admin = PgPoolOptions::new()
+            .max_connections(1)
+            .connect(&crate::test_support::database_url())
+            .await
+            .expect("connect admin");
+        let (pool, name) = create_scratch_db(&admin).await;
+        let db = Db::from_pool(pool.clone());
+
+        assert_eq!(
+            db.load_storage_accounting_completed_at()
+                .await
+                .expect("load absent completed_at"),
+            None
+        );
+        let mut leader = db
+            .try_lock_storage_accounting()
+            .await
+            .expect("lock")
+            .expect("worker owns lock");
+        leader
+            .save_snapshot(&serde_json::json!({"version": 1}), 10, 100, "a")
+            .await
+            .expect("save snapshot");
+        let stored = db
+            .load_storage_accounting_snapshot()
+            .await
+            .expect("load snapshot")
+            .expect("snapshot exists");
+        assert_eq!(
+            db.load_storage_accounting_completed_at()
+                .await
+                .expect("load completed_at"),
+            Some(stored.completed_at)
+        );
+
+        let mut expected = Vec::new();
+        for (host, archived, state, active) in [
+            ("active.example", false, "active", true),
+            ("archived.example", true, "active", false),
+            ("quiescing.example", false, "quiescing", false),
+            ("tombstone.example", false, "tombstone", false),
+        ] {
+            let id = sqlx::query_scalar::<_, Uuid>(
+                "INSERT INTO communities (host, archived_at, deletion_state) \
+                 VALUES ($1, CASE WHEN $2 THEN now() END, $3) RETURNING id",
+            )
+            .bind(host)
+            .bind(archived)
+            .bind(state)
+            .fetch_one(&pool)
+            .await
+            .expect("insert community");
+            expected.push(StorageUsageCommunity {
+                id,
+                host: host.to_string(),
+                active,
+            });
+        }
+        let mut communities = db
+            .storage_usage_communities()
+            .await
+            .expect("list communities");
+        communities.retain(|community| expected.iter().any(|e| e.id == community.id));
+        communities.sort_by(|a, b| a.host.cmp(&b.host));
+        expected.sort_by(|a, b| a.host.cmp(&b.host));
+        assert_eq!(communities, expected);
+
+        drop(leader);
+        drop(db);
+        pool.close().await;
+        let _ = sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DROP DATABASE IF EXISTS {name} WITH (FORCE)"
+        )))
+        .execute(&admin)
+        .await;
+    }
 }

crates/buzz-relay/src/config.rs

@@ -292,6 +292,10 @@ pub struct Config {
     /// Configured operator-listener identities and their HTTP(S) delivery URLs.
     pub operator_listener_delivery_urls: HashMap<String, url::Url>,
 
+    /// Endpoint receiving each new per-community storage snapshot
+    /// (`BUZZ_STORAGE_USAGE_URL`). Unset disables storage-usage delivery.
+    pub storage_usage_delivery_url: Option<url::Url>,
+
     /// Allow NIP-OA owner attestation for relay membership.
     ///
     /// When `true` and `require_relay_membership` is also `true`, agents
@@ -926,6 +930,18 @@ impl Config {
                 )));
             }
         };
+        let storage_usage_delivery_url = match std::env::var("BUZZ_STORAGE_USAGE_URL") {
+            Ok(raw) => Some(
+                crate::storage_usage_delivery::parse_delivery_url(&raw)
+                    .map_err(ConfigError::InvalidValue)?,
+            ),
+            Err(std::env::VarError::NotPresent) => None,
+            Err(error) => {
+                return Err(ConfigError::InvalidValue(format!(
+                    "BUZZ_STORAGE_USAGE_URL must be valid UTF-8: {error}"
+                )));
+            }
+        };
         if !operator_listener_delivery_urls.is_empty() && relay_operator_api_origin.is_none() {
             error!(
                 "BUZZ_OPERATOR_LISTENERS is set but RELAY_OPERATOR_API_ORIGIN is not — operator-listener registration requests will reject every request until RELAY_OPERATOR_API_ORIGIN is set"
@@ -1459,6 +1475,7 @@ impl Config {
             relay_operator_api_origin,
             relay_operator_pubkeys,
             operator_listener_delivery_urls,
+            storage_usage_delivery_url,
             allow_nip_oa_auth,
             klipy,
             media,

crates/buzz-relay/src/lib.rs

@@ -68,6 +68,8 @@ pub mod startup_steps;
 /// Shared application state.
 pub mod state;
 pub mod storage_sweep;
+/// Push each new per-community storage snapshot to an operator endpoint.
+pub mod storage_usage_delivery;
 /// Subscription registry with (channel, kind) fan-out index.
 pub mod subscription;
 /// OpenTelemetry tracing initialisation (tracer provider + OTLP exporter).

crates/buzz-relay/src/main.rs

@@ -1314,6 +1314,14 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
     }
     info!("operator-listener registration cleanup started");
 
+    if let Some(url) = state.config.storage_usage_delivery_url.clone() {
+        tokio::spawn(buzz_relay::storage_usage_delivery::run_delivery_worker(
+            Arc::clone(&state),
+            url,
+        ));
+        info!("storage-usage delivery worker started");
+    }
+
     // Admin outbox delivery worker — drives `relay_admin_outbox` rows.
     // Uses DB-level leases (held_by / lease_expires_at) so multiple pods can
     // run the worker concurrently without double-delivery.

crates/buzz-relay/src/storage_usage_delivery.rs

@@ -0,0 +1,407 @@
+//! Push each new per-community storage snapshot to an operator endpoint.
+//!
+//! `buzz-admin storage-snapshot` replaces one durable snapshot row per run.
+//! When `BUZZ_STORAGE_USAGE_URL` is set, one relay pod (a session advisory
+//! lock) checks that row every few minutes and, when its `completed_at` is
+//! new, POSTs the per-community totals in batches, in the request shape of
+//! kgoose's `POST /v1/buzz/storage-usage/record`. Each batch is NIP-98 signed
+//! with the relay key, as operator-listener mention notifications are.
+//!
+//! The snapshot row is the retry record: it stays until the next worker run
+//! replaces it, and an undelivered snapshot is re-sent on every check. Which
+//! snapshot was delivered is remembered only in memory, so a restart or
+//! leader change re-sends the latest one. Receivers must treat a repeated
+//! community and `measured_at` as already recorded.
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use chrono::{DateTime, SecondsFormat, Utc};
+use reqwest::StatusCode;
+use serde::Serialize;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use buzz_db::storage_accounting::{StorageUsageCommunity, StoredStorageSnapshot};
+use buzz_media::BucketSnapshot;
+
+use crate::{nip98::nip98_header, state::AppState};
+
+/// Deployment-global advisory lock electing the single delivering pod.
+const STORAGE_USAGE_DELIVERY_LOCK_KEY: i64 = 0x4255_5A5A_5553_4744;
+const CHECK_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+/// Communities per POST. One run sends every batch in order, so this bounds
+/// each request body, not how many communities a snapshot reports.
+const BATCH_SIZE: usize = 50;
+
+/// Parse `BUZZ_STORAGE_USAGE_URL`: an HTTP(S) URL without credentials, query,
+/// or fragment, the same shape operator-listener endpoints must have.
+pub fn parse_delivery_url(raw: &str) -> Result<url::Url, String> {
+    let url = url::Url::parse(raw.trim())
+        .map_err(|e| format!("BUZZ_STORAGE_USAGE_URL is not a valid URL: {e}"))?;
+    if !matches!(url.scheme(), "http" | "https")
+        || url.host().is_none()
+        || !url.username().is_empty()
+        || url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+    {
+        return Err(
+            "BUZZ_STORAGE_USAGE_URL must be an HTTP(S) URL without credentials, query, or fragment"
+                .to_string(),
+        );
+    }
+    Ok(url)
+}
+
+/// One POST body: kgoose's `RecordBuzzStorageUsageRequest`. Batches are
+/// independent requests sharing `measured_at`, the snapshot's RFC 3339
+/// completion time.
+#[derive(Debug, Serialize)]
+struct RecordStorageUsageRequest<'a> {
+    measured_at: &'a str,
+    communities: &'a [CommunityUsage],
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+struct CommunityUsage {
+    community_host: String,
+    storage_bytes: u64,
+    storage_objects: u64,
+}
+
+/// Every snapshot community with its bytes, plus an explicit zero for each
+/// active community the snapshot does not mention, so a receiver keeping the
+/// latest value per community never keeps a stale nonzero one. A snapshot
+/// community with no community row has no host to report and is skipped.
+/// Hosts are unique, so sorting by host keeps batch boundaries stable across
+/// re-sends.
+fn usage_records(
+    snapshot: &BucketSnapshot,
+    communities: &[StorageUsageCommunity],
+) -> Vec<CommunityUsage> {
+    let hosts: HashMap<Uuid, &str> = communities
+        .iter()
+        .map(|community| (community.id, community.host.as_str()))
+        .collect();
+    let mut unmatched = 0usize;
+    let mut records: Vec<CommunityUsage> = snapshot
+        .per_community
+        .iter()
+        .filter_map(|(id, storage)| {
+            let Some(host) = hosts.get(id) else {
+                unmatched += 1;
+                return None;
+            };
+            Some(CommunityUsage {
+                community_host: (*host).to_string(),
+                storage_bytes: storage.bytes,
+                storage_objects: storage.objects,
+            })
+        })
+        .collect();
+    records.extend(
+        communities
+            .iter()
+            .filter(|community| {
+                community.active && !snapshot.per_community.contains_key(&community.id)
+            })
+            .map(|community| CommunityUsage {
+                community_host: community.host.clone(),
+                storage_bytes: 0,
+                storage_objects: 0,
+            }),
+    );
+    if unmatched > 0 {
+        warn!(
+            unmatched,
+            "storage snapshot communities without a community row were not reported"
+        );
+    }
+    records.sort_by(|a, b| a.community_host.cmp(&b.community_host));
+    records
+}
+
+#[async_trait::async_trait]
+trait UsageSource: Sync {
+    async fn completed_at(&self) -> Result<Option<DateTime<Utc>>, String>;
+    async fn snapshot(&self) -> Result<Option<StoredStorageSnapshot>, String>;
+    async fn communities(&self) -> Result<Vec<StorageUsageCommunity>, String>;
+}
+
+struct DbUsageSource<'a> {
+    db: &'a buzz_db::Db,
+}
+
+#[async_trait::async_trait]
+impl UsageSource for DbUsageSource<'_> {
+    async fn completed_at(&self) -> Result<Option<DateTime<Utc>>, String> {
+        self.db
+            .load_storage_accounting_completed_at()
+            .await
+            .map_err(|error| error.to_string())
+    }
+
+    async fn snapshot(&self) -> Result<Option<StoredStorageSnapshot>, String> {
+        self.db
+            .load_storage_accounting_snapshot()
+            .await
+            .map_err(|error| error.to_string())
+    }
+
+    async fn communities(&self) -> Result<Vec<StorageUsageCommunity>, String> {
+        self.db
+            .storage_usage_communities()
+            .await
+            .map_err(|error| error.to_string())
+    }
+}
+
+#[async_trait::async_trait]
+trait DeliveryTransport: Sync {
+    async fn post(
+        &self,
+        url: &url::Url,
+        authorization: &str,
+        body: Vec<u8>,
+    ) -> Result<StatusCode, String>;
+}
+
+struct ReqwestDeliveryTransport {
+    http: reqwest::Client,
+}
+
+#[async_trait::async_trait]
+impl DeliveryTransport for ReqwestDeliveryTransport {
+    async fn post(
+        &self,
+        url: &url::Url,
+        authorization: &str,
+        body: Vec<u8>,
+    ) -> Result<StatusCode, String> {
+        self.http
+            .post(url.clone())
+            .header("Authorization", authorization)
+            .header("Content-Type", "application/json")
+            .body(body)
+            .send()
+            .await
+            .map(|response| response.status())
+            .map_err(|error| error.to_string())
+    }
+}
+
+fn build_delivery_http_client() -> Result<reqwest::Client, reqwest::Error> {
+    reqwest::Client::builder()
+        .timeout(REQUEST_TIMEOUT)
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+}
+
+/// Result of one check against the stored snapshot.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum CheckOutcome {
+    /// No snapshot exists, or the stored one was already handled.
+    Unchanged,
+    /// Every batch was accepted.
+    Delivered,
+    /// The endpoint refused a batch with a non-retryable status; this
+    /// snapshot is abandoned and the next worker run is delivered normally.
+    Rejected,
+    /// A retryable failure; the same snapshot is re-sent on the next check.
+    Retry,
+}
+
+impl CheckOutcome {
+    fn metric_label(self) -> Option<&'static str> {
+        match self {
+            Self::Unchanged => None,
+            Self::Delivered => Some("accepted"),
+            Self::Rejected => Some("rejected"),
+            Self::Retry => Some("retry"),
+        }
+    }
+}
+
+/// What this pod has handled, kept across checks while it leads.
+#[derive(Debug, Default)]
+struct DeliveryProgress {
+    /// `completed_at` of the last snapshot delivered or abandoned. Set only
+    /// on a terminal outcome.
+    handled: Option<DateTime<Utc>>,
+    /// `completed_at` of the last snapshot whose every batch was accepted.
+    delivered: Option<DateTime<Utc>>,
+}
+
+/// One leader check: deliver the stored snapshot if it is new, then record
+/// the outcome. The delivered-snapshot gauge is re-published on every check,
+/// including unchanged and failed ones, so it outlives the exporter's gauge
+/// idle timeout between daily snapshots and a freshness alert always has a
+/// sample to fire on.
+async fn check_once<S: UsageSource, T: DeliveryTransport>(
+    source: &S,
+    transport: &T,
+    url: &url::Url,
+    relay_keypair: &nostr::Keys,
+    progress: &mut DeliveryProgress,
+) -> CheckOutcome {
+    let outcome = deliver_new_snapshot(source, transport, url, relay_keypair, progress).await;
+    if let Some(label) = outcome.metric_label() {
+        metrics::counter!("buzz_storage_usage_deliveries_total", "outcome" => label).increment(1);
+    }
+    if let Some(delivered) = progress.delivered {
+        metrics::gauge!("buzz_storage_usage_delivered_snapshot_timestamp_seconds")
+            .set(delivered.timestamp() as f64);
+    }
+    outcome
+}
+
+/// Deliver the stored snapshot if it differs from `progress.handled`.
+async fn deliver_new_snapshot<S: UsageSource, T: DeliveryTransport>(
+    source: &S,
+    transport: &T,
+    url: &url::Url,
+    relay_keypair: &nostr::Keys,
+    progress: &mut DeliveryProgress,
+) -> CheckOutcome {
+    let latest = match source.completed_at().await {
+        Ok(Some(latest)) => latest,
+        Ok(None) => return CheckOutcome::Unchanged,
+        Err(error) => {
+            warn!(%error, "storage-usage delivery could not read the snapshot time");
+            return CheckOutcome::Retry;
+        }
+    };
+    if progress.handled == Some(latest) {
+        return CheckOutcome::Unchanged;
+    }
+    let stored = match source.snapshot().await {
+        Ok(Some(stored)) => stored,
+        Ok(None) => return CheckOutcome::Unchanged,
+        Err(error) => {
+            warn!(%error, "storage-usage delivery could not load the snapshot");
+            return CheckOutcome::Retry;
+        }
+    };
+    let snapshot: BucketSnapshot = match serde_json::from_value(stored.snapshot) {
+        Ok(snapshot) => snapshot,
+        Err(error) => {
+            // Re-reading the same row cannot succeed; wait for the next run.
+            error!(%error, "stored storage snapshot does not decode; skipping it");
+            progress.handled = Some(stored.completed_at);
+            return CheckOutcome::Rejected;
+        }
+    };
+    let communities = match source.communities().await {
+        Ok(communities) => communities,
+        Err(error) => {
+            warn!(%error, "storage-usage delivery could not list communities");
+            return CheckOutcome::Retry;
+        }
+    };
+
+    let measured_at = stored
+        .completed_at
+        .to_rfc3339_opts(SecondsFormat::Secs, true);
+    let records = usage_records(&snapshot, &communities);
+    for (index, communities) in records.chunks(BATCH_SIZE).enumerate() {
+        let body = match serde_json::to_vec(&RecordStorageUsageRequest {
+            measured_at: &measured_at,
+            communities,
+        }) {
+            Ok(body) => body,
+            Err(error) => {
+                error!(%error, "storage-usage batch encoding failed");
+                progress.handled = Some(stored.completed_at);
+                return CheckOutcome::Rejected;
+            }
+        };
+        let auth = match nip98_header(relay_keypair, url.as_str(), &body) {
+            Ok(auth) => auth,
+            Err(error) => {
+                error!(%error, "storage-usage batch signing failed");
+                return CheckOutcome::Retry;
+            }
+        };
+        match transport.post(url, &auth, body).await {
+            Ok(status) if status.is_success() => {}
+            Ok(status)
+                if status == StatusCode::REQUEST_TIMEOUT
+                    || status == StatusCode::TOO_MANY_REQUESTS
+                    || status.is_server_error() =>
+            {
+                warn!(%status, batch = index, "storage-usage delivery failed; retrying next check");
+                return CheckOutcome::Retry;
+            }
+            Ok(status) => {
+                error!(%status, batch = index, "storage-usage delivery rejected; skipping this snapshot");
+                progress.handled = Some(stored.completed_at);
+                return CheckOutcome::Rejected;
+            }
+            Err(error) => {
+                warn!(%error, batch = index, "storage-usage delivery failed; retrying next check");
+                return CheckOutcome::Retry;
+            }
+        }
+    }
+    progress.handled = Some(stored.completed_at);
+    progress.delivered = Some(stored.completed_at);
+    info!(
+        %measured_at,
+        communities = records.len(),
+        "storage-usage snapshot delivered"
+    );
+    CheckOutcome::Delivered
+}
+
+/// Check for a new snapshot every few minutes while holding the delivery lock.
+pub async fn run_delivery_worker(state: Arc<AppState>, url: url::Url) {
+    let transport = match build_delivery_http_client() {
+        Ok(http) => ReqwestDeliveryTransport { http },
+        Err(error) => {
+            error!(%error, "storage-usage HTTP client initialization failed");
+            return;
+        }
+    };
+    let source = DbUsageSource { db: &state.db };
+    let mut leader: Option<buzz_db::SessionLeader> = None;
+    let mut progress = DeliveryProgress::default();
+    let mut interval = tokio::time::interval(CHECK_INTERVAL);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        interval.tick().await;
+        if let Some(guard) = leader.as_mut() {
+            if !guard.is_live().await {
+                warn!("storage-usage delivery lock session lost; re-electing");
+                leader = None;
+            }
+        }
+        if leader.is_none() {
+            match state
+                .db
+                .try_lock_session_leader(STORAGE_USAGE_DELIVERY_LOCK_KEY)
+                .await
+            {
+                Ok(acquired) => leader = acquired,
+                Err(error) => {
+                    warn!(%error, "storage-usage delivery lock attempt failed");
+                    continue;
+                }
+            }
+        }
+        if leader.is_none() {
+            continue;
+        }
+        check_once(
+            &source,
+            &transport,
+            &url,
+            &state.relay_keypair,
+            &mut progress,
+        )
+        .await;
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/buzz-relay/src/storage_usage_delivery/tests.rs

@@ -0,0 +1,560 @@
+use std::collections::VecDeque;
+use std::sync::Mutex;
+
+use buzz_media::CommunityStorage;
+use nostr::Keys;
+use tokio::{
+    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
+    net::{TcpListener, TcpStream},
+};
+
+use super::*;
+
+fn id(byte: u8) -> Uuid {
+    Uuid::from_bytes([byte; 16])
+}
+
+fn community(byte: u8, active: bool) -> StorageUsageCommunity {
+    StorageUsageCommunity {
+        id: id(byte),
+        host: format!("c{byte}.example"),
+        active,
+    }
+}
+
+fn snapshot_of(entries: &[(u8, u64, u64)]) -> BucketSnapshot {
+    BucketSnapshot {
+        per_community: entries
+            .iter()
+            .map(|(byte, bytes, objects)| {
+                (
+                    id(*byte),
+                    CommunityStorage {
+                        bytes: *bytes,
+                        objects: *objects,
+                    },
+                )
+            })
+            .collect(),
+        ..Default::default()
+    }
+}
+
+fn stored(snapshot: &BucketSnapshot, completed_at: i64) -> StoredStorageSnapshot {
+    StoredStorageSnapshot {
+        snapshot: serde_json::to_value(snapshot).expect("encode snapshot"),
+        completed_at: DateTime::from_timestamp(completed_at, 0).expect("timestamp"),
+        duration_ms: 1,
+        max_objects: 10,
+        code_sha: "test".to_string(),
+    }
+}
+
+struct MockSource {
+    stored: Mutex<Option<StoredStorageSnapshot>>,
+    communities: Vec<StorageUsageCommunity>,
+}
+
+impl MockSource {
+    fn new(stored: Option<StoredStorageSnapshot>, communities: Vec<StorageUsageCommunity>) -> Self {
+        Self {
+            stored: Mutex::new(stored),
+            communities,
+        }
+    }
+
+    fn replace(&self, stored: StoredStorageSnapshot) {
+        *self.stored.lock().expect("source lock") = Some(stored);
+    }
+}
+
+#[async_trait::async_trait]
+impl UsageSource for MockSource {
+    async fn completed_at(&self) -> Result<Option<DateTime<Utc>>, String> {
+        Ok(self
+            .stored
+            .lock()
+            .expect("source lock")
+            .as_ref()
+            .map(|stored| stored.completed_at))
+    }
+
+    async fn snapshot(&self) -> Result<Option<StoredStorageSnapshot>, String> {
+        Ok(self.stored.lock().expect("source lock").clone())
+    }
+
+    async fn communities(&self) -> Result<Vec<StorageUsageCommunity>, String> {
+        Ok(self.communities.clone())
+    }
+}
+
+/// Answers each POST with the next queued response, then 204.
+#[derive(Default)]
+struct MockTransport {
+    responses: Mutex<VecDeque<Result<StatusCode, String>>>,
+    bodies: Mutex<Vec<serde_json::Value>>,
+}
+
+impl MockTransport {
+    fn answering(responses: impl IntoIterator<Item = Result<StatusCode, String>>) -> Self {
+        Self {
+            responses: Mutex::new(responses.into_iter().collect()),
+            bodies: Mutex::default(),
+        }
+    }
+
+    fn take_bodies(&self) -> Vec<serde_json::Value> {
+        std::mem::take(&mut *self.bodies.lock().expect("bodies lock"))
+    }
+}
+
+#[async_trait::async_trait]
+impl DeliveryTransport for MockTransport {
+    async fn post(
+        &self,
+        _url: &url::Url,
+        _authorization: &str,
+        body: Vec<u8>,
+    ) -> Result<StatusCode, String> {
+        self.bodies
+            .lock()
+            .expect("bodies lock")
+            .push(serde_json::from_slice(&body).expect("batch JSON"));
+        self.responses
+            .lock()
+            .expect("responses lock")
+            .pop_front()
+            .unwrap_or(Ok(StatusCode::NO_CONTENT))
+    }
+}
+
+fn endpoint() -> url::Url {
+    url::Url::parse("https://usage.example/storage").expect("endpoint")
+}
+
+/// A snapshot that spans three batches: 2 * BATCH_SIZE + 1 communities.
+fn three_batch_source(completed_at: i64) -> MockSource {
+    let count = 2 * BATCH_SIZE + 1;
+    let mut per_community = HashMap::new();
+    let mut communities = Vec::new();
+    for index in 0..count {
+        let id = Uuid::from_u128(index as u128 + 1);
+        per_community.insert(
+            id,
+            CommunityStorage {
+                bytes: 10,
+                objects: 1,
+            },
+        );
+        communities.push(StorageUsageCommunity {
+            id,
+            host: format!("c{index}.example"),
+            active: true,
+        });
+    }
+    let snapshot = BucketSnapshot {
+        per_community,
+        ..Default::default()
+    };
+    MockSource::new(Some(stored(&snapshot, completed_at)), communities)
+}
+
+/// Each body's `measured_at` and community count.
+fn batch_shape(bodies: &[serde_json::Value]) -> Vec<(&str, usize)> {
+    bodies
+        .iter()
+        .map(|body| {
+            (
+                body["measured_at"].as_str().expect("measured_at"),
+                body["communities"].as_array().expect("communities").len(),
+            )
+        })
+        .collect()
+}
+
+/// RFC 3339 form of the 1_700_000_000 test snapshot time.
+const MEASURED_AT: &str = "2023-11-14T22:13:20Z";
+
+#[test]
+fn delivery_url_accepts_plain_http_endpoints_only() {
+    for valid in [
+        "https://usage.example/storage",
+        "http://kgoose:8080/storage",
+        " https://usage.example/storage ",
+    ] {
+        assert!(parse_delivery_url(valid).is_ok(), "{valid}");
+    }
+    for invalid in [
+        "",
+        "not a url",
+        "ftp://usage.example/storage",
+        "https://user:pass@usage.example/storage",
+        "https://usage.example/storage?token=secret",
+        "https://usage.example/storage#fragment",
+    ] {
+        assert!(parse_delivery_url(invalid).is_err(), "{invalid:?}");
+    }
+}
+
+#[test]
+fn records_cover_snapshot_communities_and_zero_absent_active_ones() {
+    let snapshot = snapshot_of(&[
+        (1, 100, 2), // active, stored
+        (2, 50, 1),  // archived but still storing bytes
+        (5, 7, 1),   // no community row: no host to report
+    ]);
+    let communities = vec![
+        community(1, true),
+        community(2, false),
+        community(3, true),  // active, nothing stored: explicit zero
+        community(4, false), // inactive, nothing stored: omitted
+    ];
+
+    let records = usage_records(&snapshot, &communities);
+
+    let record = |host: &str, bytes: u64, objects: u64| CommunityUsage {
+        community_host: host.to_string(),
+        storage_bytes: bytes,
+        storage_objects: objects,
+    };
+    assert_eq!(
+        records,
+        vec![
+            record("c1.example", 100, 2),
+            record("c2.example", 50, 1),
+            record("c3.example", 0, 0),
+        ]
+    );
+}
+
+#[tokio::test]
+async fn new_snapshot_is_delivered_once_in_bounded_batches() {
+    let source = three_batch_source(1_700_000_000);
+    let transport = MockTransport::default();
+    let mut progress = DeliveryProgress::default();
+
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+
+    assert_eq!(outcome, CheckOutcome::Delivered);
+    let bodies = transport.take_bodies();
+    assert_eq!(
+        batch_shape(&bodies),
+        vec![
+            (MEASURED_AT, BATCH_SIZE),
+            (MEASURED_AT, BATCH_SIZE),
+            (MEASURED_AT, 1)
+        ]
+    );
+
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+    assert_eq!(outcome, CheckOutcome::Unchanged);
+    assert!(
+        transport.take_bodies().is_empty(),
+        "delivered snapshot re-sent"
+    );
+
+    source.replace(stored(&snapshot_of(&[(1, 1, 1)]), 1_700_086_400));
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+    assert_eq!(outcome, CheckOutcome::Delivered);
+    let bodies = transport.take_bodies();
+    assert_eq!(bodies.len(), 3, "every active community is reported");
+    assert_eq!(bodies[0]["measured_at"], "2023-11-15T22:13:20Z");
+}
+
+#[tokio::test]
+async fn retryable_failures_resend_the_whole_snapshot_next_check() {
+    for failure in [
+        Ok(StatusCode::REQUEST_TIMEOUT),
+        Ok(StatusCode::TOO_MANY_REQUESTS),
+        Ok(StatusCode::SERVICE_UNAVAILABLE),
+        Err("connection refused".to_string()),
+    ] {
+        let source = three_batch_source(1_700_000_000);
+        let transport = MockTransport::answering([Ok(StatusCode::OK), failure.clone()]);
+        let mut progress = DeliveryProgress::default();
+
+        let outcome = check_once(
+            &source,
+            &transport,
+            &endpoint(),
+            &Keys::generate(),
+            &mut progress,
+        )
+        .await;
+        assert_eq!(outcome, CheckOutcome::Retry, "{failure:?}");
+        assert_eq!(progress.handled, None, "{failure:?}");
+        assert_eq!(transport.take_bodies().len(), 2, "{failure:?}");
+
+        let outcome = check_once(
+            &source,
+            &transport,
+            &endpoint(),
+            &Keys::generate(),
+            &mut progress,
+        )
+        .await;
+        assert_eq!(outcome, CheckOutcome::Delivered, "{failure:?}");
+        assert_eq!(
+            batch_shape(&transport.take_bodies()),
+            vec![
+                (MEASURED_AT, BATCH_SIZE),
+                (MEASURED_AT, BATCH_SIZE),
+                (MEASURED_AT, 1)
+            ],
+            "{failure:?}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn client_error_abandons_only_that_snapshot() {
+    let source = three_batch_source(1_700_000_000);
+    let transport = MockTransport::answering([Ok(StatusCode::BAD_REQUEST)]);
+    let mut progress = DeliveryProgress::default();
+
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+    assert_eq!(outcome, CheckOutcome::Rejected);
+    assert_eq!(transport.take_bodies().len(), 1, "stopped after rejection");
+
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+    assert_eq!(outcome, CheckOutcome::Unchanged);
+    assert!(
+        transport.take_bodies().is_empty(),
+        "rejected snapshot re-sent"
+    );
+
+    source.replace(stored(&snapshot_of(&[(1, 1, 1)]), 1_700_086_400));
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+    assert_eq!(outcome, CheckOutcome::Delivered);
+}
+
+#[tokio::test]
+async fn missing_snapshot_sends_nothing() {
+    let source = MockSource::new(None, vec![community(1, true)]);
+    let transport = MockTransport::default();
+    let mut progress = DeliveryProgress::default();
+
+    let outcome = check_once(
+        &source,
+        &transport,
+        &endpoint(),
+        &Keys::generate(),
+        &mut progress,
+    )
+    .await;
+
+    assert_eq!(outcome, CheckOutcome::Unchanged);
+    assert!(transport.take_bodies().is_empty());
+}
+
+/// Runs one real `check_once` against a fresh recorder and returns the
+/// delivered-snapshot gauge that check published, if any.
+fn check_publishing(
+    runtime: &tokio::runtime::Runtime,
+    source: &MockSource,
+    transport: &MockTransport,
+    progress: &mut DeliveryProgress,
+) -> (CheckOutcome, Option<f64>) {
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    let outcome = metrics::with_local_recorder(&recorder, || {
+        runtime.block_on(check_once(
+            source,
+            transport,
+            &endpoint(),
+            &Keys::generate(),
+            progress,
+        ))
+    });
+    let gauge = snapshotter
+        .snapshot()
+        .into_vec()
+        .into_iter()
+        .find_map(|(key, _, _, value)| match value {
+            DebugValue::Gauge(value)
+                if key.key().name()
+                    == "buzz_storage_usage_delivered_snapshot_timestamp_seconds" =>
+            {
+                Some(value.into_inner())
+            }
+            _ => None,
+        });
+    (outcome, gauge)
+}
+
+#[test]
+fn every_check_republishes_the_last_delivered_snapshot_time() {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("runtime");
+    let source = MockSource::new(
+        Some(stored(&snapshot_of(&[(1, 100, 2)]), 1_700_000_000)),
+        vec![community(1, true)],
+    );
+    let transport = MockTransport::default();
+    let mut progress = DeliveryProgress::default();
+
+    assert_eq!(
+        check_publishing(&runtime, &source, &transport, &mut progress),
+        (CheckOutcome::Delivered, Some(1_700_000_000.0))
+    );
+    // Between daily snapshots the gauge must keep its sample, or the
+    // exporter's idle timeout drops it and a freshness alert cannot fire.
+    assert_eq!(
+        check_publishing(&runtime, &source, &transport, &mut progress),
+        (CheckOutcome::Unchanged, Some(1_700_000_000.0))
+    );
+    // A failing new snapshot keeps reporting the old delivery time.
+    source.replace(stored(&snapshot_of(&[(1, 1, 1)]), 1_700_086_400));
+    let failing = MockTransport::answering([Ok(StatusCode::SERVICE_UNAVAILABLE)]);
+    assert_eq!(
+        check_publishing(&runtime, &source, &failing, &mut progress),
+        (CheckOutcome::Retry, Some(1_700_000_000.0))
+    );
+
+    // Nothing delivered yet: no sample, rather than a misleading zero.
+    let mut fresh = DeliveryProgress::default();
+    let failing = MockTransport::answering([Ok(StatusCode::SERVICE_UNAVAILABLE)]);
+    assert_eq!(
+        check_publishing(&runtime, &source, &failing, &mut fresh),
+        (CheckOutcome::Retry, None)
+    );
+}
+
+async fn read_http_request(stream: &mut TcpStream) -> (String, Vec<u8>, String) {
+    let mut reader = BufReader::new(stream);
+    let mut request_line = String::new();
+    reader
+        .read_line(&mut request_line)
+        .await
+        .expect("read request line");
+    let mut content_length = 0;
+    let mut authorization = String::new();
+    loop {
+        let mut line = String::new();
+        reader.read_line(&mut line).await.expect("read header");
+        if line == "\r\n" || line.is_empty() {
+            break;
+        }
+        if let Some((name, value)) = line.split_once(':') {
+            if name.eq_ignore_ascii_case("content-length") {
+                content_length = value.trim().parse().expect("valid content length");
+            } else if name.eq_ignore_ascii_case("authorization") {
+                authorization = value.trim().to_owned();
+            }
+        }
+    }
+    let mut body = vec![0; content_length];
+    reader
+        .read_exact(&mut body)
+        .await
+        .expect("read request body");
+    (request_line, body, authorization)
+}
+
+#[tokio::test]
+async fn http_delivery_is_signed_by_the_relay_key() {
+    use nostr::nips::nip98::{verify_auth_header, HttpMethod};
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("HTTP listener");
+    let url = url::Url::parse(&format!(
+        "http://{}/storage",
+        listener.local_addr().expect("listener address")
+    ))
+    .expect("HTTP endpoint");
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.expect("accept delivery");
+        let request = read_http_request(&mut stream).await;
+        stream
+            .write_all(b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n")
+            .await
+            .expect("write success response");
+        request
+    });
+    let relay_keypair = Keys::generate();
+    let source = MockSource::new(
+        Some(stored(&snapshot_of(&[(1, 100, 2)]), 1_700_000_000)),
+        vec![community(1, true)],
+    );
+    let transport = ReqwestDeliveryTransport {
+        http: build_delivery_http_client().expect("delivery client"),
+    };
+    let mut progress = DeliveryProgress::default();
+
+    let outcome = check_once(&source, &transport, &url, &relay_keypair, &mut progress).await;
+
+    assert_eq!(outcome, CheckOutcome::Delivered);
+    let (request_line, body, authorization) = tokio::time::timeout(Duration::from_secs(2), server)
+        .await
+        .expect("HTTP server completed")
+        .expect("HTTP server task");
+    assert!(request_line.starts_with("POST /storage HTTP/1.1"));
+    let signer = verify_auth_header(
+        &authorization,
+        &url,
+        HttpMethod::POST,
+        nostr::Timestamp::now(),
+        Some(&body),
+    )
+    .expect("valid signature binding the endpoint and body");
+    assert_eq!(signer, relay_keypair.public_key());
+    let payload: serde_json::Value = serde_json::from_slice(&body).expect("batch JSON");
+    assert_eq!(
+        payload,
+        serde_json::json!({
+            "measured_at": MEASURED_AT,
+            "communities": [{
+                "community_host": "c1.example",
+                "storage_bytes": 100,
+                "storage_objects": 2,
+            }],
+        })
+    );
+}

deploy/charts/buzz/values.yaml

@@ -373,6 +373,8 @@ s3:
 
 # Isolated S3 accounting worker. Disabled by default. The relay is ready to
 # consume snapshots whenever an operator enables a worker in this environment.
+# To push each new snapshot's per-community totals to a usage endpoint, set
+# BUZZ_STORAGE_USAGE_URL through relay.extraEnv (see .env.example).
 storageAccounting:
   enabled: false
   schedule: "0 3 * * *"

docs/storage-accounting.md

@@ -54,3 +54,55 @@ The reader preserves leader-only emission, community scope filtering, and
 cleanup of series for renamed or removed communities. It does not need S3
 listing permission. The worker still needs that permission and its database
 configuration.
+
+## Usage delivery
+
+Set `BUZZ_STORAGE_USAGE_URL` to push each new snapshot's per-community totals
+to an operator endpoint. kgoose's `POST {prefix}/v1/buzz/storage-usage/record`
+receives them and bills sponsored communities in Orb. Unset, nothing is sent.
+Every five minutes, one relay pod (elected by a PostgreSQL session advisory
+lock) compares the stored snapshot's `completed_at` with the last one it
+handled and, when it is new, POSTs every community in consecutive batches of
+at most 50:
+
+```json
+{
+  "measured_at": "2026-10-30T03:04:12Z",
+  "communities": [
+    {
+      "community_host": "acme.communities.buzz.xyz",
+      "storage_bytes": 123456789,
+      "storage_objects": 412
+    }
+  ]
+}
+```
+
+- `measured_at` is the snapshot's completion time in RFC 3339, the same for
+  every batch of a snapshot. Each batch is an independent request.
+- Each request carries a NIP-98 `Authorization` header signed by the relay
+  key, binding the URL and body, as operator-listener mention notifications
+  do. kgoose authenticates the relay by its `buzz` mesh service identity
+  instead, so the relay needs mesh egress to kgoose-builderbot.
+- `storage_bytes` and `storage_objects` are the logical per-community values
+  described above: a blob bound to several communities counts in full for
+  each.
+- Every snapshot community with a community row is sent, including archived
+  or deleting ones that still hold bytes. An active community absent from the
+  snapshot is sent with zero, so a receiver keeping the latest value never
+  retains a stale one. A snapshot community with no community row has no host
+  and is skipped with a warning.
+- A 2xx accepts a batch. A 408, 429, 5xx, or transport error stops the run and
+  re-sends the whole snapshot on the next check. Any other status abandons
+  that snapshot; the next worker run is delivered normally.
+- Which snapshot was handled is kept in memory, so a restart or leader change
+  re-sends the latest snapshot. Receivers MUST treat a repeated community and
+  `measured_at` as already recorded; kgoose bills only a community's first
+  report for each UTC day.
+- A snapshot replaced before it was delivered is never sent. Alert on
+  `time() - buzz_storage_usage_delivered_snapshot_timestamp_seconds` alongside
+  snapshot age; `buzz_storage_usage_deliveries_total{outcome}` counts
+  `accepted`, `retry`, and `rejected` checks. The leader re-publishes the
+  delivered timestamp on every check, so it survives the gauge idle timeout
+  between daily snapshots. After a restart or leader change it has no sample
+  until the next delivery succeeds, so treat a missing series as failing too.