feat(media): stamp upload attribution on media objects and sidecars

Media uploads now carry upload attribution so operators and out-of-band
consumers can attribute any stored object without relay internals:

- S3 object metadata on the blob and thumbnail PUTs:
  x-amz-meta-buzz-uploader-id (authenticated Blossom uploader pubkey,
  hex) and x-amz-meta-buzz-community-id (host-resolved community UUID),
  readable from a bare HEAD on the object.
- The same fields on the BlobMeta sidecar (uploader_id / community_id),
  nullable with serde defaults so older sidecars still parse.

The community always comes from the server-resolved TenantContext
(row-zero host binding), never from client input. All three upload
paths are covered: image, generic file, and streaming video (the video
path attaches metadata via bucket extra_headers so it survives
multipart uploads).

Note: blobs are shared content-addressed storage across communities, so
a re-upload of identical bytes under another tenant overwrites the
object metadata with the most recent uploader; the community-scoped
sidecar remains the authoritative per-tenant record.

Cargo.lock

@@ -945,6 +945,7 @@ dependencies = [
  "futures-core",
  "futures-util",
  "hex",
+ "http",
  "image",
  "imagesize",
  "infer",

crates/buzz-media/Cargo.toml

@@ -20,6 +20,7 @@ hex = { workspace = true }
 chrono = { workspace = true }
 axum = { workspace = true }
 s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] }
+http = "1"
 infer = "0.19"
 image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
 blurhash = "0.2"

crates/buzz-media/src/storage.rs

@@ -77,6 +77,32 @@ impl MediaStorage {
         Ok(())
     }
 
+    /// Store an object from a byte slice with `x-amz-meta-*` object metadata.
+    ///
+    /// `metadata` keys are bare names (e.g. `buzz-uploader-id`); the S3 client
+    /// adds the `x-amz-meta-` prefix. Used for media blobs that carry upload
+    /// attribution so out-of-band consumers can read it from a HEAD without
+    /// touching relay internals.
+    pub async fn put_with_metadata(
+        &self,
+        key: &str,
+        bytes: &[u8],
+        content_type: &str,
+        metadata: &[(&str, &str)],
+    ) -> Result<(), MediaError> {
+        let mut builder = self
+            .bucket
+            .put_object_builder(key, bytes)
+            .with_content_type(content_type);
+        for (k, v) in metadata {
+            builder = builder
+                .with_metadata(k, v)
+                .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        }
+        builder.execute().await?;
+        Ok(())
+    }
+
     /// Stream a file from disk into S3 without loading it into RAM.
     ///
     /// Uses rust-s3's `put_object_stream_with_content_type` which reads from
@@ -87,6 +113,24 @@ impl MediaStorage {
         key: &str,
         path: &Path,
         content_type: &str,
+    ) -> Result<(), MediaError> {
+        self.put_file_with_metadata(key, path, content_type, &[])
+            .await
+    }
+
+    /// Stream a file from disk into S3 with `x-amz-meta-*` object metadata.
+    ///
+    /// Metadata is attached via bucket-level `extra_headers` (not the stream
+    /// builder's `with_metadata`) because rust-s3's streaming multipart path
+    /// only forwards builder headers on the small-file (single PUT) branch;
+    /// bucket `extra_headers` are applied to `InitiateMultipartUpload` too, so
+    /// metadata survives files larger than the 8 MiB chunk threshold.
+    pub async fn put_file_with_metadata(
+        &self,
+        key: &str,
+        path: &Path,
+        content_type: &str,
+        metadata: &[(&str, &str)],
     ) -> Result<(), MediaError> {
         const BUF: usize = 8 * 1024 * 1024; // 8 MiB read buffer
 
@@ -95,7 +139,19 @@ impl MediaStorage {
             .map_err(|e| MediaError::Io(e.to_string()))?;
         let mut reader = tokio::io::BufReader::with_capacity(BUF, file);
 
-        self.bucket
+        if metadata.is_empty() {
+            self.bucket
+                .put_object_stream_with_content_type(&mut reader, key, content_type)
+                .await?;
+            return Ok(());
+        }
+
+        let headers = build_amz_meta_headers(metadata)?;
+        let bucket = self
+            .bucket
+            .with_extra_headers(headers)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        bucket
             .put_object_stream_with_content_type(&mut reader, key, content_type)
             .await?;
         Ok(())
@@ -336,6 +392,71 @@ mod tests {
             "video/mp4"
         );
     }
+
+    #[test]
+    fn amz_meta_headers_are_prefixed_and_validated() {
+        let headers = build_amz_meta_headers(&[
+            ("buzz-uploader-id", "aabbcc"),
+            ("buzz-community-id", "0000-1111"),
+        ])
+        .unwrap();
+        assert_eq!(
+            headers.get("x-amz-meta-buzz-uploader-id").unwrap(),
+            "aabbcc"
+        );
+        assert_eq!(
+            headers.get("x-amz-meta-buzz-community-id").unwrap(),
+            "0000-1111"
+        );
+
+        // Control characters in values are rejected, not silently mangled.
+        assert!(build_amz_meta_headers(&[("buzz-uploader-id", "bu\nzz")]).is_err());
+        // Invalid header-name characters in the key are rejected.
+        assert!(build_amz_meta_headers(&[("bad key", "v")]).is_err());
+    }
+
+    /// Old sidecars (written before upload attribution) must still parse, and
+    /// new fields must round-trip.
+    #[test]
+    fn sidecar_attribution_fields_are_backward_compatible() {
+        // Pre-attribution sidecar JSON — no uploader_id/community_id keys.
+        let old = r#"{"dim":"800x600","blurhash":"","thumb_url":"","ext":"jpg","mime_type":"image/jpeg","size":123,"uploaded_at":1700000000}"#;
+        let meta: BlobMeta = serde_json::from_str(old).unwrap();
+        assert_eq!(meta.uploader_id, None);
+        assert_eq!(meta.community_id, None);
+
+        // Absent attribution is omitted from serialized output (not null).
+        let json = serde_json::to_value(&meta).unwrap();
+        assert!(json.get("uploader_id").is_none());
+        assert!(json.get("community_id").is_none());
+
+        // Populated attribution round-trips.
+        let meta = BlobMeta {
+            uploader_id: Some("aa".repeat(32)),
+            community_id: Some("6b8e1c2a-0000-0000-0000-000000000000".to_string()),
+            ..meta
+        };
+        let round: BlobMeta = serde_json::from_str(&serde_json::to_string(&meta).unwrap()).unwrap();
+        assert_eq!(round.uploader_id, meta.uploader_id);
+        assert_eq!(round.community_id, meta.community_id);
+    }
+}
+
+/// Build an `x-amz-meta-*` [`http::HeaderMap`] from bare metadata key/value
+/// pairs. Keys must be valid header-name characters; values must be valid
+/// header values (S3 object metadata is US-ASCII).
+fn build_amz_meta_headers(metadata: &[(&str, &str)]) -> Result<http::HeaderMap, MediaError> {
+    let mut headers = http::HeaderMap::new();
+    for (k, v) in metadata {
+        let name: http::HeaderName = format!("x-amz-meta-{k}")
+            .parse()
+            .map_err(|_| MediaError::StorageError(format!("invalid metadata key: {k}")))?;
+        let value: http::HeaderValue = v
+            .parse()
+            .map_err(|_| MediaError::StorageError(format!("invalid metadata value for {k}")))?;
+        headers.insert(name, value);
+    }
+    Ok(headers)
 }
 
 /// Metadata returned by HEAD — just enough for BUD-01 response headers.
@@ -364,4 +485,14 @@ pub struct BlobMeta {
     /// Video duration in seconds. `None` for non-video blobs.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub duration_secs: Option<f64>,
+    /// Authenticated uploader pubkey (hex). Upload attribution for
+    /// out-of-band consumers; `None` on sidecars written before attribution
+    /// existed.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub uploader_id: Option<String>,
+    /// Host-resolved community id (UUID string). Mirrors the community segment
+    /// of the sidecar key so attribution survives even if the object is copied
+    /// out of its keyed location; `None` on pre-attribution sidecars.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub community_id: Option<String>,
 }

crates/buzz-media/src/upload.rs

@@ -86,6 +86,17 @@ where
     // Compute uploaded_at once — single source of truth for sidecar and response.
     let uploaded_at = chrono::Utc::now().timestamp();
 
+    // Upload attribution: the authenticated uploader pubkey and the
+    // host-resolved community, stamped as S3 object metadata so out-of-band
+    // consumers can attribute a blob from a HEAD
+    // alone. The community comes from the server-resolved `TenantContext`,
+    // never from client input. Note the blob is shared CAS across communities:
+    // if the same bytes are later uploaded under another tenant, the re-put
+    // overwrites this metadata with the most recent uploader — the
+    // community-scoped sidecar remains the authoritative per-tenant record.
+    let uploader_id = auth_event.pubkey.to_hex();
+    let community_id = ctx.community().to_string();
+
     // Store blob first, then metadata.
     // On failure we intentionally do NOT delete the orphan blob — concurrent
     // uploads of the same hash could race and delete a blob that another
@@ -93,7 +104,17 @@ where
     // content-addressed and bounded by the upload size limit, so the storage
     // cost is negligible. A V2 background GC job can sweep blobs with no
     // matching sidecar after a grace period.
-    storage.put(&key, &body, &mime).await?;
+    storage
+        .put_with_metadata(
+            &key,
+            &body,
+            &mime,
+            &[
+                ("buzz-uploader-id", uploader_id.as_str()),
+                ("buzz-community-id", community_id.as_str()),
+            ],
+        )
+        .await?;
 
     let meta_result = store_metadata(MetadataInput {
         sha256: sha256.clone(),
@@ -101,6 +122,8 @@ where
         mime: mime.clone(),
         body: body.clone(),
         uploaded_at,
+        uploader_id,
+        community_id,
     })
     .await;
 
@@ -131,6 +154,10 @@ struct MetadataInput {
     mime: String,
     body: Bytes,
     uploaded_at: i64,
+    /// Authenticated uploader pubkey (hex), mirrored into the sidecar.
+    uploader_id: String,
+    /// Host-resolved community id, mirrored into the sidecar.
+    community_id: String,
 }
 
 /// Process an upload end-to-end: validate, store, thumbnail, return descriptor.
@@ -195,6 +222,8 @@ pub async fn process_file_upload(
                 mime_type: input.mime,
                 uploaded_at: input.uploaded_at,
                 duration_secs: None,
+                uploader_id: Some(input.uploader_id),
+                community_id: Some(input.community_id),
             };
             storage.put_sidecar(ctx, &input.sha256, &meta).await?;
             Ok(meta)
@@ -370,8 +399,23 @@ pub async fn process_video_upload(
 
     let uploaded_at = chrono::Utc::now().timestamp();
 
+    // Upload attribution — see process_buffered_upload for the rationale and
+    // the shared-CAS re-put caveat.
+    let uploader_id = auth_event.pubkey.to_hex();
+    let community_id = ctx.community().to_string();
+
     // --- 6. Stream blob from temp file to S3 ---
-    storage.put_file(&key, &tmp_path, &mime).await?;
+    storage
+        .put_file_with_metadata(
+            &key,
+            &tmp_path,
+            &mime,
+            &[
+                ("buzz-uploader-id", uploader_id.as_str()),
+                ("buzz-community-id", community_id.as_str()),
+            ],
+        )
+        .await?;
     drop(tmp); // Free temp file disk space immediately after S3 upload.
 
     // --- 7. Write sidecar (no thumbnail for video — desktop handles that) ---
@@ -384,6 +428,8 @@ pub async fn process_video_upload(
         size: file_size,
         uploaded_at,
         duration_secs: Some(video_meta.duration_secs),
+        uploader_id: Some(uploader_id),
+        community_id: Some(community_id),
     };
     storage.put_sidecar(ctx, &sha256_hex, &meta).await?;
 
@@ -418,10 +464,30 @@ async fn generate_and_store_metadata(
     .map_err(|_| MediaError::Internal)??;
 
     meta.uploaded_at = input.uploaded_at;
+    meta.uploader_id = Some(input.uploader_id);
+    meta.community_id = Some(input.community_id);
 
     if let Some(ref tb) = thumb_bytes {
+        // The thumbnail is a derived object with its own S3 key, so it carries
+        // the same attribution metadata as the source blob.
         let thumb_key = format!("{}.thumb.jpg", input.sha256);
-        storage.put(&thumb_key, tb, "image/jpeg").await?;
+        storage
+            .put_with_metadata(
+                &thumb_key,
+                tb,
+                "image/jpeg",
+                &[
+                    (
+                        "buzz-uploader-id",
+                        meta.uploader_id.as_deref().unwrap_or_default(),
+                    ),
+                    (
+                        "buzz-community-id",
+                        meta.community_id.as_deref().unwrap_or_default(),
+                    ),
+                ],
+            )
+            .await?;
     }
 
     storage.put_sidecar(ctx, &input.sha256, &meta).await?;
@@ -484,6 +550,8 @@ mod tests {
             size: 5_000_000,
             uploaded_at: 1700000000,
             duration_secs: Some(29.5),
+            uploader_id: None,
+            community_id: None,
         };
 
         let desc = build_descriptor(
@@ -542,6 +610,8 @@ mod tests {
             size: 100_000,
             uploaded_at: 1700000000,
             duration_secs: None,
+            uploader_id: None,
+            community_id: None,
         };
 
         let desc = build_descriptor(