feat(relay): add per-user storage metrics from audit log

Add logical-attribution storage observability, grouped by
(community_id, actor_pubkey): a partial index on audit_log for
media_uploaded actions, per-user/per-community/fleet byte and object
gauges in the usage poller, and a real-time upload-bytes counter in
the media handler.

Dedupe by (community_id, actor_pubkey, object_id) in a subquery so
idempotent re-uploads of the same blob are charged once, not once per
audit row. Fail safe on malformed historical detail->>'size' values
(non-numeric, negative, out-of-range, or exceeding bigint bounds)
by validating with a digits-only regex and a lexicographic bound check
against i64::MAX, then attributing 0 bytes instead of aborting the
whole poller query. The bound check uses ltrim + length comparison
and works on all PostgreSQL versions without requiring pg_input_is_valid
(PG 16+).

Storage aggregation is overflow-safe end to end: the per-user SUM
accumulates in numeric (SUM(bigint) returns numeric for wider precision)
and is clamped into [0, i64::MAX] before the final cast to bigint,
so an extreme historical aggregate degrades to a saturated reading
instead of erroring the whole query. Rust-side community/fleet rollups
use saturating addition for the same reason — individually valid
per-user values can still overflow i64 in aggregate, and these are
observability gauges, not a billing ledger.

The concurrent index build (migration 0021) is retry-safe via two
layers: a startup repair guard drops any indisvalid=false leftover
from a failed prior build, and IF NOT EXISTS skips creation when a
valid unrecorded index already exists from a bookkeeping race. Both
the repair guard and migrator run under a session-scoped advisory
lock on a single pinned connection — pg_advisory_lock is session-scoped,
so lock/guards/migrator/unlock must all execute on the same PG session;
if unlock fails, dropping the connection closes the session and releases
the lock automatically. The repair probe is schema-scoped (joins
pg_namespace, restricts to current_schema()) so a same-named invalid
index in another schema cannot interfere.

Named *_objects rather than *_uploads to stay distinct from the new
real-time buzz_media_upload_bytes_total event counter added in the
media handler. Per-user cardinality (two custom series per
community,pubkey pair) is intentional; BUZZ_USAGE_METRICS_PER_COMMUNITY=off
remains the emergency kill switch.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

crates/buzz-db/src/lib.rs

@@ -450,6 +450,11 @@ impl Db {
         usage::active_channel_counts(&self.pool, interval_sql).await
     }
 
+    /// Return per-user storage byte totals and object counts (logical attribution).
+    pub async fn usage_storage_byte_counts(&self) -> Result<Vec<usage::UserStorageCounts>> {
+        usage::storage_byte_counts(&self.pool).await
+    }
+
     /// Return all community id → host mappings.
     pub async fn usage_community_hosts(&self) -> Result<Vec<usage::CommunityHost>> {
         usage::community_hosts(&self.pool).await

crates/buzz-db/src/migration.rs

@@ -11,9 +11,84 @@ use crate::Result;
 static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
 
 /// Run all pending Buzz database migrations.
+///
+/// Wraps the sqlx migrator in a session-scoped advisory lock so that
+/// concurrent replicas (rolling deploy) serialize startup instead of
+/// racing each other's concurrent index builds or repair guards.
+///
+/// All operations (lock, pre-migration guards, migrator, unlock) run on
+/// a single acquired connection — `pg_advisory_lock` is session-scoped,
+/// so the lock-holding session must be the one that runs the critical
+/// section and releases it. The connection is marked `close_on_drop`,
+/// so if unlock fails or the future is cancelled, dropping the
+/// connection closes the PG session (rather than returning a locked
+/// session to the pool) and releases the lock automatically.
 pub async fn run_migrations(pool: &PgPool) -> Result<()> {
-    reject_legacy_nip_rs_cardinality_ambiguity(pool).await?;
-    MIGRATOR.run(pool).await?;
+    const MIGRATION_ADVISORY_LOCK_KEY: i64 = 0x42757a7a4d696772; // "BuzzMigr"
+
+    let mut conn = pool.acquire().await?;
+    conn.close_on_drop();
+
+    sqlx::query("SELECT pg_advisory_lock($1)")
+        .bind(MIGRATION_ADVISORY_LOCK_KEY)
+        .execute(&mut *conn)
+        .await?;
+
+    let result = async {
+        reject_legacy_nip_rs_cardinality_ambiguity(&mut conn).await?;
+        repair_invalid_media_index(&mut conn).await?;
+        MIGRATOR.run(&mut *conn).await?;
+        Ok(())
+    }
+    .await;
+
+    let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
+        .bind(MIGRATION_ADVISORY_LOCK_KEY)
+        .execute(&mut *conn)
+        .await;
+
+    result
+}
+
+/// Drop `idx_audit_log_media_uploads` if it exists but is invalid
+/// (`indisvalid = false`), so migration 21's `CREATE INDEX CONCURRENTLY
+/// IF NOT EXISTS` builds a fresh valid index instead of skipping over
+/// the broken leftover.
+///
+/// This handles the case where a prior concurrent build failed (e.g. OOM,
+/// process kill, unique violation on a UNIQUE variant) and left an invalid
+/// index behind. Without this guard, `IF NOT EXISTS` would skip creation,
+/// SQLx would record version 21, and the relay would run permanently
+/// without the required access path.
+///
+/// The probe joins `pg_index.indrelid = to_regclass('audit_log')` so it
+/// finds the invalid index attached to whichever `audit_log` the
+/// session's `search_path` resolves, regardless of schema. The DROP
+/// statement is built server-side via `quote_ident(schema.index)` so it
+/// targets the correct namespace without hardcoding `public`.
+///
+/// Multi-replica safety: the caller holds a session-scoped advisory lock
+/// on the same connection, so a second replica starting simultaneously
+/// cannot see replica A's in-progress concurrent build as invalid and
+/// drop it mid-build.
+async fn repair_invalid_media_index(conn: &mut sqlx::PgConnection) -> Result<()> {
+    let drop_stmt: Option<String> = sqlx::query_scalar(
+        "SELECT 'DROP INDEX IF EXISTS ' || quote_ident(n.nspname) || '.' || quote_ident(c.relname) \
+         FROM pg_index i \
+         JOIN pg_class c ON c.oid = i.indexrelid \
+         JOIN pg_namespace n ON n.oid = c.relnamespace \
+         WHERE c.relname = 'idx_audit_log_media_uploads' \
+           AND i.indrelid = to_regclass('audit_log') \
+           AND NOT i.indisvalid",
+    )
+    .fetch_optional(&mut *conn)
+    .await?;
+
+    if let Some(stmt) = drop_stmt {
+        sqlx::query(sqlx::AssertSqlSafe(stmt))
+            .execute(&mut *conn)
+            .await?;
+    }
     Ok(())
 }
 
@@ -21,17 +96,17 @@ pub async fn run_migrations(pool: &PgPool) -> Result<()> {
 /// enforcement. A populated database still on 0001-0006 must not let 0007
 /// irreversibly purge duplicate-tag history. Fail before sqlx starts its
 /// migration transaction so an operator can inspect and repair those rows.
-async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> {
+async fn reject_legacy_nip_rs_cardinality_ambiguity(conn: &mut sqlx::PgConnection) -> Result<()> {
     let migrations_table: Option<String> =
         sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations')::text")
-            .fetch_one(pool)
+            .fetch_one(&mut *conn)
             .await?;
     if migrations_table.is_none() {
         return Ok(());
     }
     let applied: Option<i64> =
         sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations WHERE success")
-            .fetch_one(pool)
+            .fetch_one(&mut *conn)
             .await?;
     if applied.is_none_or(|version| version >= 7) {
         return Ok(());
@@ -75,7 +150,7 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()>
                )\
          )",
     )
-    .fetch_one(pool)
+    .fetch_one(&mut *conn)
     .await?;
 
     if ambiguous {
@@ -549,7 +624,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 20);
+        assert_eq!(migrations.len(), 21);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -824,6 +899,23 @@ mod tests {
             .sql
             .as_str()
             .contains("join_policy_acceptances"));
+
+        // Per-user storage attribution needs a non-transactional CONCURRENTLY
+        // index build on audit_log; it must not lock out concurrent writers
+        // during migration on a populated relay.
+        assert_eq!(migrations[20].version, 21);
+        assert!(
+            migrations[20].no_tx,
+            "migration 21 builds an index CONCURRENTLY and must run outside a transaction"
+        );
+        assert!(migrations[20]
+            .sql
+            .as_str()
+            .contains("CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_log_media_uploads"));
+        assert!(!migrations[0]
+            .sql
+            .as_str()
+            .contains("idx_audit_log_media_uploads"));
     }
 
     #[test]
@@ -1066,7 +1158,7 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(20));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(21));
     }
 
     #[tokio::test]
@@ -1129,6 +1221,366 @@ mod tests {
         assert_eq!(after, vec![(1, Some(true)), (30_350, None)]);
     }
 
+    /// Bookkeeping-race recovery: a process dies after the DDL succeeds but
+    /// before SQLx records version 21 → a valid same-named index is left
+    /// unrecorded. `IF NOT EXISTS` makes the retry idempotent: PG skips
+    /// creation and SQLx records version 21 — the relay starts normally.
+    /// The repair guard must NOT drop a valid index in this scenario.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migration_21_retries_after_leftover_unrecorded_index() {
+        let pool = connect_test_pool().await;
+        reset_public_schema(&pool).await;
+
+        MIGRATOR
+            .run_to(20, &pool)
+            .await
+            .expect("apply migrations 1-20");
+
+        sqlx::query(
+            "CREATE INDEX idx_audit_log_media_uploads \
+             ON audit_log (community_id, actor_pubkey, object_id) \
+             WHERE action = 'media_uploaded'",
+        )
+        .execute(&pool)
+        .await
+        .expect("plant a same-named valid index (simulates successful DDL before bookkeeping)");
+
+        let planted: bool = sqlx::query_scalar(
+            "SELECT EXISTS ( \
+                 SELECT 1 FROM pg_class WHERE relname = 'idx_audit_log_media_uploads' \
+             )",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("check planted index");
+        assert!(planted, "planted index must exist before retry");
+
+        run_migrations(&pool)
+            .await
+            .expect("migration 21 must not wedge when a same-named index already exists");
+
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(21));
+
+        let index_exists: bool = sqlx::query_scalar(
+            "SELECT EXISTS ( \
+                 SELECT 1 FROM pg_indexes \
+                 WHERE indexname = 'idx_audit_log_media_uploads' \
+             )",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("check index still present");
+        assert!(
+            index_exists,
+            "index must still exist after idempotent retry"
+        );
+    }
+
+    /// Failed-build recovery: a prior `CREATE INDEX CONCURRENTLY` failed
+    /// (e.g. OOM, unique violation, process kill) and left an `indisvalid=false`
+    /// index behind. Without the startup repair guard, `IF NOT EXISTS` would
+    /// skip creation, SQLx would record version 21, and the relay would run
+    /// permanently without the required access path. The repair guard drops
+    /// the invalid index before the migrator runs, so a fresh valid build
+    /// proceeds.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migration_21_recovers_after_failed_concurrent_build() {
+        let pool = connect_test_pool().await;
+        reset_public_schema(&pool).await;
+
+        MIGRATOR
+            .run_to(20, &pool)
+            .await
+            .expect("apply migrations 1-20");
+
+        // Plant duplicate data so a UNIQUE concurrent build fails, leaving
+        // an indisvalid=false index behind.
+        let community_id = uuid::Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("invalid-idx-{}.example", community_id.simple()))
+            .execute(&pool)
+            .await
+            .expect("insert community for duplicate data");
+        let hash_a = vec![0xAA_u8; 32];
+        let hash_b = vec![0xBB_u8; 32];
+        let actor = vec![0xCC_u8; 32];
+        for (seq, hash) in [(1_i64, &hash_a), (2_i64, &hash_b)] {
+            sqlx::query(
+                "INSERT INTO audit_log (community_id, seq, hash, action, actor_pubkey, object_id, detail) \
+                 VALUES ($1, $2, $3, 'media_uploaded', $4, 'same-obj', '{}'::jsonb)",
+            )
+            .bind(community_id)
+            .bind(seq)
+            .bind(hash)
+            .bind(&actor)
+            .execute(&pool)
+            .await
+            .expect("insert duplicate audit row");
+        }
+
+        // Attempt a UNIQUE concurrent build — it will fail on the duplicates,
+        // leaving an indisvalid=false index.
+        let unique_result = sqlx::query(
+            "CREATE UNIQUE INDEX CONCURRENTLY idx_audit_log_media_uploads \
+             ON audit_log (community_id, actor_pubkey, object_id) \
+             WHERE action = 'media_uploaded'",
+        )
+        .execute(&pool)
+        .await;
+        assert!(
+            unique_result.is_err(),
+            "UNIQUE build must fail on duplicate data"
+        );
+
+        // Verify the invalid index exists.
+        let invalid: bool = sqlx::query_scalar(
+            "SELECT EXISTS ( \
+                 SELECT 1 FROM pg_index i \
+                 JOIN pg_class c ON c.oid = i.indexrelid \
+                 WHERE c.relname = 'idx_audit_log_media_uploads' \
+                   AND NOT i.indisvalid \
+             )",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("check invalid index");
+        assert!(
+            invalid,
+            "failed concurrent build must leave indisvalid=false index"
+        );
+
+        // Clean up the duplicate data so the real (non-unique) index build
+        // in migration 21 succeeds.
+        sqlx::query("DELETE FROM audit_log WHERE community_id = $1 AND seq = 2")
+            .bind(community_id)
+            .execute(&pool)
+            .await
+            .expect("remove duplicate so real migration succeeds");
+
+        // Run migrations — repair guard should drop the invalid index,
+        // then migration 21 creates a fresh valid one.
+        run_migrations(&pool)
+            .await
+            .expect("repair guard + migration 21 must recover from invalid index");
+
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(21));
+
+        // The index must exist AND be valid.
+        let (exists, valid): (bool, bool) = sqlx::query_as(
+            "SELECT \
+                 EXISTS (SELECT 1 FROM pg_class WHERE relname = 'idx_audit_log_media_uploads'), \
+                 COALESCE(( \
+                     SELECT i.indisvalid FROM pg_index i \
+                     JOIN pg_class c ON c.oid = i.indexrelid \
+                     WHERE c.relname = 'idx_audit_log_media_uploads' \
+                 ), false)",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("check repaired index");
+        assert!(exists, "index must exist after repair");
+        assert!(valid, "index must be valid (indisvalid=true) after repair");
+
+        // Postcondition: advisory lock must be released after run_migrations
+        // returns. The pool may return the same physical session, so this is
+        // a global pg_locks absence check — valid because pg_locks is
+        // cluster-wide. Scoped to current database to avoid false failures
+        // from an unrelated database using the same key.
+        let lock_key: i64 = 0x42757a7a4d696772;
+        let lock_held: bool = sqlx::query_scalar(
+            "SELECT EXISTS ( \
+                 SELECT 1 FROM pg_locks \
+                 WHERE locktype = 'advisory' \
+                   AND database = (SELECT oid FROM pg_database WHERE datname = current_database()) \
+                   AND classid = ($1 >> 32)::oid \
+                   AND objid = ($1 & x'FFFFFFFF'::bigint)::oid \
+             )",
+        )
+        .bind(lock_key)
+        .fetch_one(&pool)
+        .await
+        .expect("check pg_locks for advisory lock");
+        assert!(
+            !lock_held,
+            "advisory lock must not be held after run_migrations returns"
+        );
+    }
+
+    /// Same as `migration_21_recovers_after_failed_concurrent_build` but
+    /// under a non-public `search_path`. This exercises the probe/drop path
+    /// where the invalid index lives in a schema other than `public` — the
+    /// repair guard must resolve the actual schema from the table OID
+    /// (`to_regclass('audit_log')`) rather than hardcoding `public`.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migration_21_recovers_after_failed_concurrent_build_nonpublic_schema() {
+        let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
+            .or_else(|_| std::env::var("DATABASE_URL"))
+            .unwrap_or_else(|_| TEST_DB_URL.to_owned());
+
+        let schema_name = format!(
+            "test_schema_{}",
+            uuid::Uuid::new_v4().simple().to_string().get(..8).unwrap()
+        );
+
+        // Bootstrap pool (public schema) to create the custom schema.
+        let bootstrap_pool = PgPool::connect(&database_url)
+            .await
+            .expect("connect bootstrap pool");
+        sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DROP SCHEMA IF EXISTS {} CASCADE",
+            &schema_name
+        )))
+        .execute(&bootstrap_pool)
+        .await
+        .expect("drop old test schema");
+        sqlx::query(sqlx::AssertSqlSafe(format!(
+            "CREATE SCHEMA {}",
+            &schema_name
+        )))
+        .execute(&bootstrap_pool)
+        .await
+        .expect("create test schema");
+        bootstrap_pool.close().await;
+
+        // Build a pool whose connections default to the custom schema.
+        let opts: sqlx::postgres::PgConnectOptions =
+            database_url.parse().expect("parse database URL");
+        let opts = opts.options([("search_path", schema_name.as_str())]);
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(2)
+            .connect_with(opts)
+            .await
+            .expect("connect custom-schema pool");
+
+        // Verify search_path is set correctly.
+        let current: String = sqlx::query_scalar("SELECT current_schema()")
+            .fetch_one(&pool)
+            .await
+            .expect("check current_schema");
+        assert_eq!(
+            current, schema_name,
+            "pool connections must default to the custom schema"
+        );
+
+        MIGRATOR
+            .run_to(20, &pool)
+            .await
+            .expect("apply migrations 1-20 in custom schema");
+
+        // Plant duplicate data so a UNIQUE concurrent build fails.
+        let community_id = uuid::Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("nonpub-idx-{}.example", community_id.simple()))
+            .execute(&pool)
+            .await
+            .expect("insert community");
+        let hash_a = vec![0xAA_u8; 32];
+        let hash_b = vec![0xBB_u8; 32];
+        let actor = vec![0xCC_u8; 32];
+        for (seq, hash) in [(1_i64, &hash_a), (2_i64, &hash_b)] {
+            sqlx::query(
+                "INSERT INTO audit_log (community_id, seq, hash, action, actor_pubkey, object_id, detail) \
+                 VALUES ($1, $2, $3, 'media_uploaded', $4, 'same-obj', '{}'::jsonb)",
+            )
+            .bind(community_id)
+            .bind(seq)
+            .bind(hash)
+            .bind(&actor)
+            .execute(&pool)
+            .await
+            .expect("insert duplicate audit row");
+        }
+
+        let unique_result = sqlx::query(
+            "CREATE UNIQUE INDEX CONCURRENTLY idx_audit_log_media_uploads \
+             ON audit_log (community_id, actor_pubkey, object_id) \
+             WHERE action = 'media_uploaded'",
+        )
+        .execute(&pool)
+        .await;
+        assert!(
+            unique_result.is_err(),
+            "UNIQUE build must fail on duplicate data"
+        );
+
+        // Verify the invalid index exists in the custom schema.
+        let invalid: bool = sqlx::query_scalar(
+            "SELECT EXISTS ( \
+                 SELECT 1 FROM pg_index i \
+                 JOIN pg_class c ON c.oid = i.indexrelid \
+                 JOIN pg_namespace n ON n.oid = c.relnamespace \
+                 WHERE c.relname = 'idx_audit_log_media_uploads' \
+                   AND n.nspname = $1 \
+                   AND NOT i.indisvalid \
+             )",
+        )
+        .bind(&schema_name)
+        .fetch_one(&pool)
+        .await
+        .expect("check invalid index in custom schema");
+        assert!(
+            invalid,
+            "failed concurrent build must leave indisvalid=false in custom schema"
+        );
+
+        // Remove duplicate so the real migration build succeeds.
+        sqlx::query("DELETE FROM audit_log WHERE community_id = $1 AND seq = 2")
+            .bind(community_id)
+            .execute(&pool)
+            .await
+            .expect("remove duplicate");
+
+        run_migrations(&pool)
+            .await
+            .expect("repair guard must handle non-public schema");
+
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(21));
+
+        // The index must exist AND be valid in the custom schema.
+        let (exists, valid): (bool, bool) = sqlx::query_as(
+            "SELECT \
+                 EXISTS ( \
+                     SELECT 1 FROM pg_class c \
+                     JOIN pg_namespace n ON n.oid = c.relnamespace \
+                     WHERE c.relname = 'idx_audit_log_media_uploads' AND n.nspname = $1 \
+                 ), \
+                 COALESCE(( \
+                     SELECT i.indisvalid FROM pg_index i \
+                     JOIN pg_class c ON c.oid = i.indexrelid \
+                     JOIN pg_namespace n ON n.oid = c.relnamespace \
+                     WHERE c.relname = 'idx_audit_log_media_uploads' AND n.nspname = $1 \
+                 ), false)",
+        )
+        .bind(&schema_name)
+        .fetch_one(&pool)
+        .await
+        .expect("check repaired index in custom schema");
+        assert!(exists, "index must exist in custom schema after repair");
+        assert!(
+            valid,
+            "index must be valid (indisvalid=true) in custom schema after repair"
+        );
+
+        // Cleanup: close the pool and drop the custom schema.
+        pool.close().await;
+        let cleanup_pool = PgPool::connect(&database_url)
+            .await
+            .expect("connect cleanup pool");
+        sqlx::query(sqlx::AssertSqlSafe(format!(
+            "DROP SCHEMA IF EXISTS {} CASCADE",
+            &schema_name
+        )))
+        .execute(&cleanup_pool)
+        .await
+        .expect("cleanup test schema");
+        cleanup_pool.close().await;
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {

crates/buzz-db/src/usage.rs

@@ -333,6 +333,97 @@ pub async fn active_channel_counts(
         .collect())
 }
 
+/// Per-user storage usage derived from the audit log (logical attribution).
+///
+/// "Logical attribution" means one full-size charge per uploader per distinct
+/// blob (`object_id`).  Re-uploading the same SHA is idempotent and does not
+/// double-count.
+#[derive(Debug)]
+pub struct UserStorageCounts {
+    /// The UUID of the community.
+    pub community_id: Uuid,
+    /// 32-byte pubkey of the uploading actor.
+    pub actor_pubkey: Vec<u8>,
+    /// Sum of bytes across distinct blobs owned by this user in this community.
+    pub total_bytes: i64,
+    /// Number of distinct blobs (unique `object_id` values).
+    pub object_count: i64,
+}
+
+/// Return per-user logical storage, deduped by `(community_id, actor_pubkey,
+/// object_id)`.  Each distinct blob is charged once at its full size (the
+/// largest *valid* recorded size wins when multiple audit rows exist for the
+/// same triple, so a malformed row never shadows a good one).  Rows with
+/// NULL `actor_pubkey` or NULL `object_id` are excluded.
+///
+/// `detail->>'size'` is historical, operator-written JSONB — never trust it
+/// to be a well-formed non-negative integer. A raw `::bigint` cast throws on
+/// anything that isn't, which would abort the whole poller over one bad row.
+/// Instead: validate with a digits-only regex, then bound-check against
+/// `i64::MAX` lexicographically (strip leading zeros with `ltrim`, reject
+/// if the stripped length exceeds 19 digits or if it equals 19 and exceeds
+/// `'9223372036854775807'`). All predicates are non-throwing on arbitrary
+/// text; the `::bigint` cast is inside `THEN`, which CASE only evaluates
+/// after both guards pass. Fall back to 0 — the object is still counted
+/// once, just at zero bytes.
+///
+/// Each individual `blob_bytes` is bounded to `i64::MAX`, but the per-user
+/// `SUM` of many valid blobs can still exceed it — `SUM(bigint)` returns
+/// `numeric` (wider precision) precisely so that doesn't overflow
+/// mid-aggregation. The final cast to `BIGINT` is what would throw, so
+/// the sum is clamped into `[0, i64::MAX]` *before* that cast — an
+/// extreme (or adversarial) total degrades to a saturated gauge reading,
+/// not an aborted poller.
+pub async fn storage_byte_counts(pool: &PgPool) -> Result<Vec<UserStorageCounts>> {
+    let rows = sqlx::query_as::<_, (Uuid, Vec<u8>, i64, i64)>(
+        r#"
+        SELECT
+            community_id,
+            actor_pubkey,
+            CAST(
+                LEAST(
+                    GREATEST(COALESCE(SUM(blob_bytes), 0), 0::numeric),
+                    9223372036854775807::numeric
+                ) AS BIGINT
+            )                    AS total_bytes,
+            COUNT(*)             AS object_count
+        FROM (
+            SELECT DISTINCT ON (community_id, actor_pubkey, object_id)
+                community_id,
+                actor_pubkey,
+                CASE
+                    WHEN detail->>'size' ~ '^[0-9]+$'
+                         AND (length(ltrim(detail->>'size', '0')) < 19
+                              OR (length(ltrim(detail->>'size', '0')) = 19
+                                  AND ltrim(detail->>'size', '0') <= '9223372036854775807'))
+                    THEN (detail->>'size')::bigint
+                    ELSE 0::bigint
+                END AS blob_bytes
+            FROM audit_log
+            WHERE action = 'media_uploaded'
+              AND actor_pubkey IS NOT NULL
+              AND object_id IS NOT NULL
+            ORDER BY community_id, actor_pubkey, object_id, blob_bytes DESC
+        ) deduped
+        GROUP BY community_id, actor_pubkey
+        "#,
+    )
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(
+            |(community_id, actor_pubkey, total_bytes, object_count)| UserStorageCounts {
+                community_id,
+                actor_pubkey,
+                total_bytes,
+                object_count,
+            },
+        )
+        .collect())
+}
+
 /// Mapping from community UUID to host string, used by the poller to resolve
 /// Prometheus label values.
 #[derive(Debug)]
@@ -719,4 +810,506 @@ mod tests {
             "no stream row after last channel deleted — poller will zero-fill"
         );
     }
+
+    /// Insert a media_uploaded audit entry for `object_id` with a raw JSON
+    /// `size` value (as text, so malformed values can be exercised — e.g.
+    /// `"not-a-number"`, `"-5"`, or omission via `None`).
+    async fn insert_media_audit_raw(
+        pool: &PgPool,
+        community_id: Uuid,
+        actor: Option<&[u8]>,
+        object_id: Option<&str>,
+        seq: i64,
+        size: Option<&str>,
+    ) {
+        let hash = random_pubkey(); // unique 32 bytes
+        let mut detail = serde_json::Map::new();
+        if let Some(s) = size {
+            detail.insert("size".to_string(), serde_json::Value::String(s.to_string()));
+        }
+        sqlx::query(
+            "INSERT INTO audit_log (community_id, seq, hash, action, actor_pubkey, object_id, detail)
+             VALUES ($1, $2, $3, 'media_uploaded', $4, $5, $6)",
+        )
+        .bind(community_id)
+        .bind(seq)
+        .bind(&hash)
+        .bind(actor)
+        .bind(object_id)
+        .bind(serde_json::Value::Object(detail))
+        .execute(pool)
+        .await
+        .expect("insert media audit entry");
+    }
+
+    /// Insert a media_uploaded audit entry with a well-formed numeric size.
+    async fn insert_media_audit(
+        pool: &PgPool,
+        community_id: Uuid,
+        actor: &[u8],
+        object_id: &str,
+        seq: i64,
+        size: u64,
+    ) {
+        insert_media_audit_raw(
+            pool,
+            community_id,
+            Some(actor),
+            Some(object_id),
+            seq,
+            Some(&size.to_string()),
+        )
+        .await;
+    }
+
+    /// Two users in the same community get independent per-user storage rows.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_two_users_same_community_independent() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+
+        let user_a = random_pubkey();
+        let user_b = random_pubkey();
+
+        // User A: 2 distinct blobs totalling 300 bytes.
+        insert_media_audit(&pool, comm_uuid, &user_a, "blob-a1", 1, 100).await;
+        insert_media_audit(&pool, comm_uuid, &user_a, "blob-a2", 2, 200).await;
+        // User B: 1 blob of 500 bytes.
+        insert_media_audit(&pool, comm_uuid, &user_b, "blob-b1", 3, 500).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+
+        let a_row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user_a);
+        let b_row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user_b);
+
+        let a = a_row.expect("user A row");
+        assert_eq!(a.total_bytes, 300, "user A: 100 + 200");
+        assert_eq!(a.object_count, 2, "user A: 2 distinct blobs");
+
+        let b = b_row.expect("user B row");
+        assert_eq!(b.total_bytes, 500, "user B: 500");
+        assert_eq!(b.object_count, 1, "user B: 1 blob");
+    }
+
+    /// Storage counts are scoped per-community — same pubkey in two communities
+    /// gets separate rows.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_cross_community_isolation() {
+        let pool = get_pool().await;
+        let (comm_a, _, _) = make_community(&pool).await;
+        let (comm_b, _, _) = make_community(&pool).await;
+
+        let user = random_pubkey();
+
+        insert_media_audit(&pool, comm_a, &user, "blob-1", 1, 100).await;
+        insert_media_audit(&pool, comm_b, &user, "blob-1", 1, 900).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+
+        let in_a = rows
+            .iter()
+            .find(|r| r.community_id == comm_a && r.actor_pubkey == user)
+            .expect("row in community A");
+        let in_b = rows
+            .iter()
+            .find(|r| r.community_id == comm_b && r.actor_pubkey == user)
+            .expect("row in community B");
+
+        assert_eq!(in_a.total_bytes, 100, "community A: 100 bytes");
+        assert_eq!(in_b.total_bytes, 900, "community B: 900 bytes");
+    }
+
+    /// Multiple distinct blobs by the same user accumulate correctly.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_multiple_uploads_same_user_accumulate() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        for seq in 1..=5 {
+            insert_media_audit(&pool, comm_uuid, &user, &format!("blob-{seq}"), seq, 10).await;
+        }
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(row.total_bytes, 50, "5 × 10 bytes");
+        assert_eq!(row.object_count, 5, "5 distinct blobs");
+    }
+
+    /// Rows with missing `size` key in detail JSONB contribute 0 bytes but
+    /// still count the blob once (fail-safe, not fail-closed).
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_missing_size_contributes_zero_bytes() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        // One normal blob + one with missing size.
+        insert_media_audit(&pool, comm_uuid, &user, "blob-1", 1, 100).await;
+        insert_media_audit_raw(&pool, comm_uuid, Some(&user), Some("blob-2"), 2, None).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes, 100,
+            "only the well-formed size contributes"
+        );
+        assert_eq!(row.object_count, 2, "both blobs counted");
+    }
+
+    /// Malformed (non-numeric, negative, or otherwise unparseable) `size`
+    /// values must not abort the whole aggregate — they fall back to 0 bytes
+    /// while the blob is still counted once. Proves the poller survives
+    /// historical junk in `detail->>'size'` instead of erroring the query.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_malformed_size_fails_safe_to_zero() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-text"),
+            1,
+            Some("not-a-number"),
+        )
+        .await;
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-neg"),
+            2,
+            Some("-5"),
+        )
+        .await;
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-float"),
+            3,
+            Some("3.5"),
+        )
+        .await;
+        insert_media_audit(&pool, comm_uuid, &user, "blob-good", 4, 42).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes, 42,
+            "malformed sizes contribute 0 bytes; only the valid blob's 42 counts"
+        );
+        assert_eq!(row.object_count, 4, "all four blobs are still counted");
+    }
+
+    /// A `detail->>'size'` value exceeding `i64::MAX` or with an extreme
+    /// digit count must not abort the poller. The lexicographic bound check
+    /// (`ltrim` + length/comparison) rejects these without throwing, so the
+    /// blob falls back to 0 bytes and the poller survives.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_oversized_digits_fails_safe_to_zero() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        let oversized = "9".repeat(131_073);
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-oversized"),
+            1,
+            Some(&oversized),
+        )
+        .await;
+        insert_media_audit(&pool, comm_uuid, &user, "blob-good", 2, 42).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts must not abort on oversized digit string");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes, 42,
+            "oversized-digit value contributes 0 bytes; only the valid blob's 42 counts"
+        );
+        assert_eq!(row.object_count, 2, "both blobs are still counted");
+    }
+
+    /// Boundary battery for the lexicographic bigint bound check: i64::MAX
+    /// accepted, i64::MAX+1 → 0, 131,073 digits → 0, leading-zero forms
+    /// ('09223372036854775807' → MAX, '0000' → 0), mixed together with a
+    /// valid 42-byte blob. All must resolve without error.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_bigint_boundary_battery() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        // i64::MAX — must be accepted.
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-i64max"),
+            1,
+            Some("9223372036854775807"),
+        )
+        .await;
+        // i64::MAX + 1 — rejected, falls back to 0.
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-i64max-plus-1"),
+            2,
+            Some("9223372036854775808"),
+        )
+        .await;
+        // 131,073 digits — rejected, falls back to 0.
+        let oversized = "9".repeat(131_073);
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-oversized-digits"),
+            3,
+            Some(&oversized),
+        )
+        .await;
+        // Leading-zero i64::MAX — accepted as MAX.
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-leading-zero-max"),
+            4,
+            Some("09223372036854775807"),
+        )
+        .await;
+        // All zeros — accepted as 0.
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            Some(&user),
+            Some("blob-all-zeros"),
+            5,
+            Some("0000"),
+        )
+        .await;
+        // Valid reference blob.
+        insert_media_audit(&pool, comm_uuid, &user, "blob-valid", 6, 42).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("boundary battery must not abort");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        // i64::MAX + leading-zero-MAX + 42 = 2 * 9223372036854775807 + 42
+        // exceeds i64::MAX → saturates to i64::MAX.
+        assert_eq!(
+            row.total_bytes,
+            i64::MAX,
+            "i64::MAX + leading-zero-MAX + 42 saturates to i64::MAX"
+        );
+        assert_eq!(
+            row.object_count, 6,
+            "all six blobs counted (including rejected-size ones)"
+        );
+    }
+
+    /// Rows with a NULL `actor_pubkey` or NULL `object_id` are excluded from
+    /// per-user attribution entirely — they must not collapse into a fake
+    /// shared "unknown" user and must not be miscounted against a real user.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_null_actor_or_object_id_excluded() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        // Real attributable upload.
+        insert_media_audit(&pool, comm_uuid, &user, "blob-real", 1, 100).await;
+        // NULL actor_pubkey (e.g. system-initiated action) — excluded.
+        insert_media_audit_raw(
+            &pool,
+            comm_uuid,
+            None,
+            Some("blob-no-actor"),
+            2,
+            Some("999"),
+        )
+        .await;
+        // NULL object_id — excluded.
+        insert_media_audit_raw(&pool, comm_uuid, Some(&user), None, 3, Some("999")).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes, 100,
+            "only the real, fully-attributed upload counts"
+        );
+        assert_eq!(row.object_count, 1, "only one attributable blob");
+
+        // No row should exist with an empty/placeholder pubkey standing in
+        // for the excluded NULL-actor entry.
+        assert_eq!(
+            rows.iter().filter(|r| r.community_id == comm_uuid).count(),
+            1,
+            "excluded rows must not create a second, fake user row"
+        );
+    }
+
+    /// Re-uploading the same blob (same `object_id`) twice for one user is
+    /// charged once — idempotent re-upload must not double-count logical
+    /// storage even though the audit log records both attempts.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_same_user_same_blob_reuploaded_charged_once() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        // Same object_id uploaded twice (idempotent re-upload).
+        insert_media_audit(&pool, comm_uuid, &user, "blob-dup", 1, 250).await;
+        insert_media_audit(&pool, comm_uuid, &user, "blob-dup", 2, 250).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes, 250,
+            "re-upload of the same blob charged once"
+        );
+        assert_eq!(row.object_count, 1, "one distinct blob, not two");
+    }
+
+    /// Two different users uploading the same blob (`object_id`, e.g. same
+    /// file contents hashing to the same SHA) are each charged once — the
+    /// dedup key includes actor_pubkey, so blob sharing across users does not
+    /// collapse their independent attribution.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_two_users_same_blob_each_charged_once() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user_a = random_pubkey();
+        let user_b = random_pubkey();
+
+        insert_media_audit(&pool, comm_uuid, &user_a, "shared-blob", 1, 400).await;
+        insert_media_audit(&pool, comm_uuid, &user_b, "shared-blob", 2, 400).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts");
+        let a = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user_a)
+            .expect("user A row");
+        let b = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user_b)
+            .expect("user B row");
+
+        assert_eq!(
+            a.total_bytes, 400,
+            "user A charged once for the shared blob"
+        );
+        assert_eq!(a.object_count, 1);
+        assert_eq!(
+            b.total_bytes, 400,
+            "user B charged once for the shared blob"
+        );
+        assert_eq!(b.object_count, 1);
+    }
+
+    /// Two distinct blobs whose sizes are each individually valid (≤
+    /// `i64::MAX`) but sum past it must not abort the query — Postgres
+    /// `SUM(bigint)` returns `numeric` (wider precision than `bigint`), so
+    /// the accumulation itself won't overflow for any feasible row count,
+    /// but a naive final `CAST(... AS BIGINT)` throws `bigint out of range`
+    /// on the result. The aggregate must clamp to `i64::MAX` instead, and
+    /// the object is still counted (both blobs are individually valid, so
+    /// nothing here is malformed input to fail-safe on).
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn test_storage_aggregate_overflow_saturates_to_i64_max() {
+        let pool = get_pool().await;
+        let (comm_uuid, _, _) = make_community(&pool).await;
+        let user = random_pubkey();
+
+        insert_media_audit(&pool, comm_uuid, &user, "blob-huge-1", 1, i64::MAX as u64).await;
+        insert_media_audit(&pool, comm_uuid, &user, "blob-huge-2", 2, 1).await;
+
+        let rows = storage_byte_counts(&pool)
+            .await
+            .expect("storage_byte_counts must not abort on aggregate overflow");
+        let row = rows
+            .iter()
+            .find(|r| r.community_id == comm_uuid && r.actor_pubkey == user)
+            .expect("user row");
+
+        assert_eq!(
+            row.total_bytes,
+            i64::MAX,
+            "aggregate exceeding i64::MAX saturates instead of erroring"
+        );
+        assert_eq!(
+            row.object_count, 2,
+            "both individually-valid blobs are still counted"
+        );
+    }
 }

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

@@ -368,6 +368,11 @@ pub async fn upload_blob(
         "community" => auth.tenant.host().to_owned()
     )
     .increment(1);
+    metrics::counter!(
+        "buzz_media_upload_bytes_total",
+        "community" => auth.tenant.host().to_owned()
+    )
+    .increment(descriptor.size);
 
     // Audit via bounded channel — same pattern as event audit.
     let desc = descriptor.clone();

crates/buzz-relay/src/main.rs

@@ -1372,6 +1372,7 @@ async fn emit_db_usage_metrics(
     let active_users_30d = state.db.usage_active_user_counts("30 days").await?;
     let active_channels_1d = state.db.usage_active_channel_counts("1 day").await?;
     let active_channels_7d = state.db.usage_active_channel_counts("7 days").await?;
+    let storage_rows = state.db.usage_storage_byte_counts().await?;
 
     // --- Determine which community IDs receive per-community series (K1) ---
     //
@@ -1672,9 +1673,74 @@ async fn emit_db_usage_metrics(
         }
     }
 
+    // --- D. Storage — per-user logical bytes and object counts from audit log ---
+    //
+    // Each `UserStorageCounts.total_bytes` is already clamped to `i64::MAX`
+    // in SQL (see `storage_byte_counts`), but summing many such rows into
+    // community/fleet rollups can itself exceed `i64::MAX`. These are
+    // observability gauges, not a billing ledger, so rollups saturate at
+    // `i64::MAX` instead of panicking (overflow checks) or wrapping
+    // (release mode) on that addition.
+    {
+        let mut community_bytes: HashMap<Uuid, i64> = HashMap::new();
+        let mut community_objects: HashMap<Uuid, i64> = HashMap::new();
+        for row in &storage_rows {
+            if !active_set.contains(&row.community_id) {
+                continue;
+            }
+            let bytes_entry = community_bytes.entry(row.community_id).or_default();
+            *bytes_entry = bytes_entry.saturating_add(row.total_bytes);
+            let objects_entry = community_objects.entry(row.community_id).or_default();
+            *objects_entry = objects_entry.saturating_add(row.object_count);
+            let community_label = match host_map.get(&row.community_id) {
+                Some(h) => h.clone(),
+                None => continue,
+            };
+            let pubkey_hex = hex::encode(&row.actor_pubkey);
+            metrics::gauge!(
+                "buzz_user_storage_bytes",
+                "community" => community_label.clone(),
+                "pubkey" => pubkey_hex.clone()
+            )
+            .set(row.total_bytes as f64);
+            metrics::gauge!(
+                "buzz_user_storage_objects",
+                "community" => community_label,
+                "pubkey" => pubkey_hex
+            )
+            .set(row.object_count as f64);
+        }
+        // Fleet totals (always emitted).
+        let total_bytes = saturating_sum(storage_rows.iter().map(|r| r.total_bytes));
+        let total_objects = saturating_sum(storage_rows.iter().map(|r| r.object_count));
+        metrics::gauge!("buzz_total_storage_bytes").set(total_bytes as f64);
+        metrics::gauge!("buzz_total_storage_objects").set(total_objects as f64);
+        // Per-community rollups (gated by active_set).
+        for (&id, community) in host_map {
+            if !active_set.contains(&id) {
+                continue;
+            }
+            let bytes = community_bytes.get(&id).copied().unwrap_or(0);
+            let objects = community_objects.get(&id).copied().unwrap_or(0);
+            metrics::gauge!("buzz_community_storage_bytes", "community" => community.clone())
+                .set(bytes as f64);
+            metrics::gauge!("buzz_community_storage_objects", "community" => community.clone())
+                .set(objects as f64);
+        }
+    }
+
     Ok(())
 }
 
+/// Sum `i64` values with saturation instead of panicking (debug overflow
+/// checks) or wrapping (release mode). Used for storage rollups, which are
+/// observability gauges rather than an authoritative ledger — degrading to
+/// a saturated reading on an extreme aggregate is preferable to losing the
+/// whole usage-metrics snapshot for the tick.
+fn saturating_sum(values: impl Iterator<Item = i64>) -> i64 {
+    values.fold(0i64, i64::saturating_add)
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashSet;
@@ -1686,8 +1752,8 @@ mod tests {
 
     use super::{
         buzz_auto_migrate_enabled, dropped_in_memory_keys, idle_timeout_secs,
-        refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, EmissionScope,
-        InMemoryMetricKey,
+        refresh_legacy_active_gauge_recency, run_periodic_until_cancelled, saturating_sum,
+        EmissionScope, InMemoryMetricKey,
     };
     use metrics::GaugeFn;
     use metrics_util::{
@@ -1807,4 +1873,32 @@ mod tests {
         assert_eq!(idle_timeout_secs(None, 300), 900);
         assert_eq!(idle_timeout_secs(Some(10), 1_000), 3_000);
     }
+
+    /// Storage rollups sum many already-clamped per-user `i64` values;
+    /// individually valid inputs can still overflow `i64` in aggregate.
+    /// `saturating_sum` must clamp at `i64::MAX`/`i64::MIN` instead of
+    /// panicking (debug overflow checks) or wrapping (release mode).
+    #[test]
+    fn test_saturating_sum_clamps_instead_of_panicking_or_wrapping() {
+        assert_eq!(
+            saturating_sum([i64::MAX, 1].into_iter()),
+            i64::MAX,
+            "aggregate exceeding i64::MAX saturates high"
+        );
+        assert_eq!(
+            saturating_sum([i64::MIN, -1].into_iter()),
+            i64::MIN,
+            "aggregate exceeding i64::MIN saturates low"
+        );
+        assert_eq!(
+            saturating_sum([10, 20, 30].into_iter()),
+            60,
+            "normal sums are unaffected"
+        );
+        assert_eq!(
+            saturating_sum(std::iter::empty()),
+            0,
+            "empty input sums to zero"
+        );
+    }
 }

migrations/0021_audit_log_media_index.sql

@@ -0,0 +1,20 @@
+-- no-transaction
+-- Partial index for per-user logical-storage queries against the audit log.
+-- object_id trails the key so the per-user dedup subquery (community_id,
+-- actor_pubkey, object_id) drives off an Index Scan on this index instead of
+-- a sequential scan; Postgres still adds an Incremental Sort + Unique to
+-- pick the max-size row per (community_id, actor_pubkey, object_id), since
+-- the DISTINCT ON tiebreaker column (blob_bytes) isn't part of the index.
+-- CREATE INDEX CONCURRENTLY requires running outside a transaction (SQLx
+-- marker above).
+--
+-- Retry safety has two layers:
+-- 1. Startup repair guard (repair_invalid_media_index in migration.rs):
+--    drops any indisvalid=false leftover from a failed concurrent build
+--    before the migrator runs, so a fresh valid build can proceed.
+-- 2. IF NOT EXISTS: if a prior build succeeded but SQLx died before
+--    recording version 20 (bookkeeping race), the valid index already
+--    exists — PG skips creation and SQLx records version 20 normally.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_log_media_uploads
+    ON audit_log (community_id, actor_pubkey, object_id)
+    WHERE action = 'media_uploaded';