feat(relay): record who uploads each media blob

crates/buzz-db/src/runtime/migration.rs

@@ -709,7 +709,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 66);
+        assert_eq!(migrations.len(), 67);
         assert_eq!(migrations[64].version, 65);
         let retire_ttl = strip_sql_comments(migrations[64].sql.as_str());
         for statement in [
@@ -746,6 +746,11 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("CREATE TABLE community_nip_fi_issuers"));
+        assert_eq!(migrations[66].version, 67);
+        assert!(migrations[66]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE media_uploaders"));
         assert_eq!(migrations[55].version, 56);
         assert!(migrations[55]
             .sql
@@ -2317,6 +2322,15 @@ mod postgres_tests {
             .as_str()
             .contains("attach_community_write_fence('community_nip_fi_issuers')"));
         expected_fences.insert("community_nip_fi_issuers".to_owned());
+        // 0067 fences the media uploaders table.
+        assert!(MIGRATOR
+            .iter()
+            .find(|m| m.version == 67)
+            .expect("media uploaders migration")
+            .sql
+            .as_str()
+            .contains("attach_community_write_fence('media_uploaders')"));
+        expected_fences.insert("media_uploaders".to_owned());
         assert_eq!(
             expected_fences, schema.fence_attachments,
             "write-fence attachment targets differ after recovery policy"

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

@@ -90,6 +90,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
     "events",
     "git_repo_names",
     "join_policy_acceptances",
+    "media_uploaders",
     "moderation_actions",
     "moderation_reports",
     "parameterized_event_watermarks",
@@ -156,6 +157,7 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[
     "audit_log",
     "community_bans",
     "community_nip_fi_issuers",
+    "media_uploaders",
     "pubkey_allowlist",
     "relay_members",
     "users",
@@ -4423,6 +4425,7 @@ DROP TRIGGER IF EXISTS community_write_fence_event_mentions ON event_mentions;
 DROP TRIGGER IF EXISTS community_write_fence_events ON events;
 DROP TRIGGER IF EXISTS community_write_fence_git_repo_names ON git_repo_names;
 DROP TRIGGER IF EXISTS community_write_fence_join_policy_acceptances ON join_policy_acceptances;
+DROP TRIGGER IF EXISTS community_write_fence_media_uploaders ON media_uploaders;
 DROP TRIGGER IF EXISTS community_write_fence_moderation_actions ON moderation_actions;
 DROP TRIGGER IF EXISTS community_write_fence_moderation_reports ON moderation_reports;
 DROP TRIGGER IF EXISTS community_write_fence_parameterized_event_watermarks ON parameterized_event_watermarks;

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

@@ -0,0 +1,110 @@
+//! Who has uploaded each blob, as proof of possession for media reads.
+
+use buzz_datastore_tracing::datastore_span;
+
+use crate::deletion::ServingWriteLease;
+use crate::{error::Result, observability, Db};
+
+impl Db {
+    /// Record that `pubkey` uploaded the blob `sha256`; repeat uploads are
+    /// no-ops. The write presents the upload's serving lease, so it may finish
+    /// while the community quiesces but never after the lease is gone.
+    #[datastore_span(name = "record_media_uploader", system = "postgresql")]
+    pub async fn record_media_uploader(
+        &self,
+        lease: &ServingWriteLease,
+        sha256: &str,
+        pubkey: &[u8],
+    ) -> Result<()> {
+        let connection =
+            observability::acquire_writer(&self.pool, observability::WriterOperation::EventWrite)
+                .await?;
+        let mut tx = sqlx::Transaction::begin(connection, None).await?;
+        self.deletion_store()
+            .guard_transaction_with_serving_lease(&mut tx, lease)
+            .await?;
+        sqlx::query(
+            "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \
+             ON CONFLICT DO NOTHING",
+        )
+        .bind(lease.community_id.as_uuid())
+        .bind(sha256)
+        .bind(pubkey)
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod postgres_tests {
+    use std::time::Duration;
+
+    use super::*;
+    use buzz_core::CommunityId;
+    use sqlx::PgPool;
+    use uuid::Uuid;
+
+    const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
+
+    async fn db_with_lease() -> (Db, CommunityId, ServingWriteLease) {
+        let db = Db::from_pool(
+            PgPool::connect(&crate::test_support::database_url())
+                .await
+                .expect("connect to test DB"),
+        );
+        let host = format!("uploader-{}.example", Uuid::new_v4().simple());
+        let community_id = db
+            .ensure_configured_community(&host)
+            .await
+            .expect("seed community")
+            .id;
+        let lease = db
+            .deletion_store()
+            .acquire_serving_write_lease(
+                community_id,
+                "media_upload",
+                "media-uploader-test",
+                Duration::from_secs(60),
+            )
+            .await
+            .expect("acquire lease");
+        (db, community_id, lease)
+    }
+
+    async fn uploader_rows(db: &Db, community_id: CommunityId) -> i64 {
+        sqlx::query_scalar("SELECT count(*) FROM media_uploaders WHERE community_id = $1")
+            .bind(community_id.as_uuid())
+            .fetch_one(&db.pool)
+            .await
+            .expect("count uploaders")
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn repeat_upload_records_one_row() {
+        let (db, community_id, lease) = db_with_lease().await;
+        for _ in 0..2 {
+            db.record_media_uploader(&lease, SHA, &[7; 32])
+                .await
+                .expect("record uploader");
+        }
+        assert_eq!(uploader_rows(&db, community_id).await, 1);
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn released_lease_refuses_the_write() {
+        let (db, community_id, lease) = db_with_lease().await;
+        db.deletion_store()
+            .release_serving_write_lease(&lease)
+            .await
+            .expect("release lease");
+        assert!(db
+            .record_media_uploader(&lease, SHA, &[7; 32])
+            .await
+            .is_err());
+        assert_eq!(uploader_rows(&db, community_id).await, 0);
+    }
+}

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

@@ -32,6 +32,8 @@ mod event_follow_up_postgres_tests;
 pub mod feed;
 /// Git repository name registry (NIP-34 kind:30617).
 pub mod git_repo;
+/// Who has uploaded each blob, as proof of possession for media reads.
+pub mod media_uploader;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
 /// Deployment-global operator-listener mention registrations and delivery queues.

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

@@ -576,6 +576,16 @@ async fn upload_blob_result(
         auth.tenant.host(),
     );
 
+    state
+        .db
+        .record_media_uploader(
+            serving_write.lease(),
+            &descriptor.sha256,
+            auth.auth_event.pubkey.as_bytes(),
+        )
+        .await
+        .map_err(|error| serving_lease_lost(error.into()))?;
+
     // Normalize MIME to a known set to bound label cardinality.
     let mime_label = match descriptor.mime_type.as_str() {
         "image/jpeg" | "image/png" | "image/gif" | "image/webp" | "video/mp4" => {
@@ -2135,11 +2145,13 @@ mod tests {
         const TEST_ISSUER: &str = "https://issuer.example";
         const TEST_AUDIENCE: &str = "https://relay.example";
         const TEST_KID: &str = "test-key-1";
-        const TEST_EC_PKCS8_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\
-            MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n\
-            WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n\
-            zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n\
-            -----END PRIVATE KEY-----\n";
+        const TEST_EC_PKCS8_PEM: &str = concat!(
+            "-----BEGIN PRIVATE KEY-----\n", // sadscan:disable kingfisher.privkey.2 -- static test-only key
+            "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n",
+            "WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n",
+            "zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n",
+            "-----END PRIVATE KEY-----\n",
+        );
 
         /// Build an AppState with NIP-FI Enforce and a real injected P-256 verifier.
         async fn media_enforce_test_state() -> Option<Arc<AppState>> {

migrations/0067_media_uploaders.sql

@@ -0,0 +1,14 @@
+-- Who has uploaded each blob in a community. Uploading the exact bytes proves
+-- possession, so channel-scoped media reads can always admit an uploader even
+-- when the same bytes are also attached somewhere they cannot read.
+--
+-- Additive. Nothing reads this table yet.
+CREATE TABLE media_uploaders (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    sha256       TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
+    pubkey       BYTEA NOT NULL,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, sha256, pubkey)
+);
+
+SELECT attach_community_write_fence('media_uploaders');

schema/schema.sql

@@ -107,6 +107,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i tables/public/moderation_reports.sql
 \i tables/public/community_bans.sql
 \i tables/public/community_nip_fi_issuers.sql
+\i tables/public/media_uploaders.sql
 \i tables/public/_operator_global_tables.sql
 \i tables/public/git_repo_names.sql
 \i tables/public/parameterized_event_watermarks.sql
@@ -159,6 +160,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 \i triggers/event_mentions.sql
 \i triggers/git_repo_names.sql
 \i triggers/join_policy_acceptances.sql
+\i triggers/media_uploaders.sql
 \i triggers/moderation_actions.sql
 \i triggers/moderation_reports.sql
 \i triggers/parameterized_event_watermarks.sql

schema/tables/public/media_uploaders.sql

@@ -0,0 +1,11 @@
+-- ── Media uploaders ───────────────────────────────────────────────────────────
+-- Who has uploaded each blob in a community. Uploading the exact bytes proves
+-- possession, so channel-scoped media reads can always admit an uploader.
+
+CREATE TABLE media_uploaders (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    sha256       TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
+    pubkey       BYTEA NOT NULL,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, sha256, pubkey)
+);

schema/triggers/media_uploaders.sql

@@ -0,0 +1 @@
+SELECT attach_community_write_fence('media_uploaders');