feat(media): add configurable sharded writes

.env.example

@@ -95,6 +95,13 @@ BUZZ_S3_BUCKET=buzz-media
 BUZZ_S3_REGION=us-east-1
 BUZZ_S3_ADDRESSING_STYLE=path
 
+# Media payload write migration. Unset defaults to legacy, so upgrades preserve
+# existing flat-only writes and do not duplicate objects automatically.
+# Stages: legacy (flat only) -> dual (sharded + flat rollback copy) -> sharded.
+# Deploy compatibility readers everywhere and backfill/reconcile objects before
+# advancing; do not roll sharded writers back to a version without sharded reads.
+# BUZZ_MEDIA_KEY_LAYOUT=legacy
+
 # -----------------------------------------------------------------------------
 # Media Upload Admission
 # -----------------------------------------------------------------------------

crates/buzz-media/src/config.rs

@@ -33,6 +33,34 @@ impl FromStr for S3AddressingStyle {
     }
 }
 
+/// Payload object-key layout used for new media writes.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum MediaKeyLayout {
+    /// Write only the pre-migration flat key.
+    #[default]
+    Legacy,
+    /// Write the sharded key first, then a flat compatibility copy.
+    Dual,
+    /// Write only the hash-leading, community-scoped key.
+    Sharded,
+}
+
+impl FromStr for MediaKeyLayout {
+    type Err = String;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "legacy" => Ok(Self::Legacy),
+            "dual" => Ok(Self::Dual),
+            "sharded" => Ok(Self::Sharded),
+            _ => Err(format!(
+                "BUZZ_MEDIA_KEY_LAYOUT must be 'legacy', 'dual', or 'sharded', got {value:?}"
+            )),
+        }
+    }
+}
+
 fn default_max_video_bytes() -> u64 {
     524_288_000 // 500 MB
 }
@@ -67,6 +95,9 @@ pub struct MediaConfig {
     /// S3 URL addressing style. Defaults to path style for MinIO compatibility.
     #[serde(default)]
     pub s3_addressing_style: S3AddressingStyle,
+    /// Object-key layout for new media payload writes. Defaults to legacy.
+    #[serde(default)]
+    pub key_layout: MediaKeyLayout,
     /// Maximum upload size for images (bytes). Default: 50 MB.
     pub max_image_bytes: u64,
     /// Maximum upload size for animated GIFs (bytes). Default: 10 MB.
@@ -159,7 +190,7 @@ impl MediaConfig {
 
 #[cfg(test)]
 mod tests {
-    use super::{MediaConfig, S3AddressingStyle};
+    use super::{MediaConfig, MediaKeyLayout, S3AddressingStyle};
     use std::str::FromStr;
 
     fn valid_config() -> MediaConfig {
@@ -170,6 +201,7 @@ mod tests {
             s3_bucket: "buzz-media".to_string(),
             s3_region: "us-east-1".to_string(),
             s3_addressing_style: S3AddressingStyle::Path,
+            key_layout: MediaKeyLayout::Legacy,
             max_image_bytes: 1,
             max_gif_bytes: 1,
             max_video_bytes: 1,
@@ -210,6 +242,15 @@ mod tests {
         }
     }
 
+    #[test]
+    fn media_key_layout_parses_and_defaults_to_legacy() {
+        assert_eq!(MediaKeyLayout::default(), MediaKeyLayout::Legacy);
+        assert_eq!("legacy".parse(), Ok(MediaKeyLayout::Legacy));
+        assert_eq!("dual".parse(), Ok(MediaKeyLayout::Dual));
+        assert_eq!("sharded".parse(), Ok(MediaKeyLayout::Sharded));
+        assert!("new".parse::<MediaKeyLayout>().is_err());
+    }
+
     #[test]
     fn upload_record_knobs_default_off_and_validate() {
         assert!(valid_config().validate().is_ok());

crates/buzz-media/src/lib.rs

@@ -18,7 +18,7 @@ pub use bucket_index::{
     classify_key, fold_bucket_listing, BucketAggregate, BucketSnapshot, CommunityStorage, KeyClass,
     Page, SweepError,
 };
-pub use config::{MediaConfig, S3AddressingStyle};
+pub use config::{MediaConfig, MediaKeyLayout, S3AddressingStyle};
 pub use error::MediaError;
 pub use keys::{
     legacy_blob_key, legacy_thumb_key, read_candidates, sharded_blob_key, sharded_thumb_key,

crates/buzz-media/src/storage.rs

@@ -5,7 +5,7 @@ use std::pin::Pin;
 
 use buzz_core::tenant::{CommunityId, TenantContext};
 
-use crate::config::{MediaConfig, S3AddressingStyle};
+use crate::config::{MediaConfig, MediaKeyLayout, S3AddressingStyle};
 use crate::error::MediaError;
 use bytes::Bytes;
 use s3::creds::Credentials;
@@ -18,6 +18,7 @@ pub type ByteStream = Pin<Box<dyn futures_core::Stream<Item = Result<Bytes, Medi
 /// S3-compatible object storage client.
 pub struct MediaStorage {
     bucket: Box<Bucket>,
+    key_layout: MediaKeyLayout,
 }
 
 impl MediaStorage {
@@ -66,7 +67,10 @@ impl MediaStorage {
             S3AddressingStyle::Path => bucket.with_path_style(),
             S3AddressingStyle::Virtual => bucket,
         };
-        Ok(Self { bucket })
+        Ok(Self {
+            bucket,
+            key_layout: config.key_layout,
+        })
     }
 
     /// Store an object from a byte slice.
@@ -104,6 +108,88 @@ impl MediaStorage {
         Ok(())
     }
 
+    /// Store a media payload according to the configured migration layout.
+    ///
+    /// Dual mode writes the sharded primary first and the legacy compatibility
+    /// copy second. Callers publish sidecars only after this returns success.
+    pub async fn put_payload(
+        &self,
+        ctx: &TenantContext,
+        sha256: &str,
+        ext: &str,
+        bytes: &[u8],
+        content_type: &str,
+    ) -> Result<String, MediaError> {
+        let legacy = crate::keys::legacy_blob_key(sha256, ext)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        let sharded = crate::keys::sharded_blob_key(ctx.community(), sha256, ext)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        match self.key_layout {
+            MediaKeyLayout::Legacy => self.put(&legacy, bytes, content_type).await?,
+            MediaKeyLayout::Dual => {
+                self.put(&sharded, bytes, content_type).await?;
+                self.put(&legacy, bytes, content_type).await?;
+            }
+            MediaKeyLayout::Sharded => self.put(&sharded, bytes, content_type).await?,
+        }
+        Ok(match self.key_layout {
+            MediaKeyLayout::Legacy => legacy,
+            MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded,
+        })
+    }
+
+    /// Stream a media payload from disk according to the configured layout.
+    pub async fn put_payload_file(
+        &self,
+        ctx: &TenantContext,
+        sha256: &str,
+        ext: &str,
+        path: &Path,
+        content_type: &str,
+    ) -> Result<String, MediaError> {
+        let legacy = crate::keys::legacy_blob_key(sha256, ext)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        let sharded = crate::keys::sharded_blob_key(ctx.community(), sha256, ext)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        match self.key_layout {
+            MediaKeyLayout::Legacy => self.put_file(&legacy, path, content_type).await?,
+            MediaKeyLayout::Dual => {
+                self.put_file(&sharded, path, content_type).await?;
+                self.put_file(&legacy, path, content_type).await?;
+            }
+            MediaKeyLayout::Sharded => self.put_file(&sharded, path, content_type).await?,
+        }
+        Ok(match self.key_layout {
+            MediaKeyLayout::Legacy => legacy,
+            MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded,
+        })
+    }
+
+    /// Store a thumbnail according to the configured migration layout.
+    pub async fn put_thumbnail(
+        &self,
+        ctx: &TenantContext,
+        sha256: &str,
+        bytes: &[u8],
+    ) -> Result<String, MediaError> {
+        let legacy = crate::keys::legacy_thumb_key(sha256)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        let sharded = crate::keys::sharded_thumb_key(ctx.community(), sha256)
+            .map_err(|e| MediaError::StorageError(e.to_string()))?;
+        match self.key_layout {
+            MediaKeyLayout::Legacy => self.put(&legacy, bytes, "image/jpeg").await?,
+            MediaKeyLayout::Dual => {
+                self.put(&sharded, bytes, "image/jpeg").await?;
+                self.put(&legacy, bytes, "image/jpeg").await?;
+            }
+            MediaKeyLayout::Sharded => self.put(&sharded, bytes, "image/jpeg").await?,
+        }
+        Ok(match self.key_layout {
+            MediaKeyLayout::Legacy => legacy,
+            MediaKeyLayout::Dual | MediaKeyLayout::Sharded => sharded,
+        })
+    }
+
     /// Retrieve an object's bytes.
     pub async fn get(&self, key: &str) -> Result<Vec<u8>, MediaError> {
         match self.bucket.get_object(key).await {
@@ -360,6 +446,7 @@ mod tests {
             s3_bucket: "buzz-media".to_string(),
             s3_region: "us-west-2".to_string(),
             s3_addressing_style: S3AddressingStyle::Path,
+            key_layout: crate::config::MediaKeyLayout::Legacy,
             max_image_bytes: 50 * 1024 * 1024,
             max_gif_bytes: 10 * 1024 * 1024,
             max_video_bytes: 524_288_000,

crates/buzz-media/src/upload.rs

@@ -88,14 +88,19 @@ where
     .await
     .map_err(|_| MediaError::Internal)??;
 
-    let key = format!("{sha256}.{ext}");
+    let legacy_key = crate::keys::legacy_blob_key(&sha256, &ext)
+        .map_err(|e| MediaError::StorageError(e.to_string()))?;
     let meta_key = MediaStorage::ctx_sidecar_key(ctx, &sha256);
 
     // Idempotent: short-circuit only if BOTH sidecar and blob exist. If the
     // sidecar exists but the blob is missing, fall through to re-upload.
     let sidecar_exists = storage.head(&meta_key).await?;
-    let blob_exists = storage.head(&key).await?;
-    if sidecar_exists && blob_exists {
+    let existing_blob_key = match storage.resolve_read_key(ctx, &legacy_key).await {
+        Ok(key) => Some(key),
+        Err(MediaError::NotFound) => None,
+        Err(error) => return Err(error),
+    };
+    if sidecar_exists && existing_blob_key.is_some() {
         let meta = storage.get_sidecar(ctx, &sha256).await?;
         // A re-upload of known bytes is still a distinct upload *event*: no
         // blob PUT happens, so without this record the uploader would be
@@ -110,6 +115,7 @@ where
                 UploadEventFacts {
                     sha256: &sha256,
                     ext: &ext,
+                    blob_key: existing_blob_key.as_deref().expect("checked above"),
                     mime: &mime,
                     size: body.len() as u64,
                     uploaded_at: chrono::Utc::now().timestamp(),
@@ -138,7 +144,9 @@ 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?;
+    let blob_key = storage
+        .put_payload(ctx, &sha256, &ext, &body, &mime)
+        .await?;
 
     let meta = match prepare_metadata(MetadataInput {
         sha256: sha256.clone(),
@@ -168,6 +176,7 @@ where
             UploadEventFacts {
                 sha256: &sha256,
                 ext: &ext,
+                blob_key: &blob_key,
                 mime: &mime,
                 size: body.len() as u64,
                 uploaded_at,
@@ -226,7 +235,7 @@ pub async fn process_upload(
             let ext = mime_to_ext(&mime).to_string();
             Ok((mime, ext))
         },
-        |input| async move { prepare_image_metadata(storage, config, input).await },
+        |input| async move { prepare_image_metadata(storage, config, ctx, input).await },
     )
     .await
 }
@@ -423,13 +432,18 @@ pub async fn process_video_upload(
             .map_err(|_| MediaError::Internal)??;
 
     let ext = "mp4";
-    let key = format!("{sha256_hex}.{ext}");
+    let legacy_key = crate::keys::legacy_blob_key(&sha256_hex, ext)
+        .map_err(|e| MediaError::StorageError(e.to_string()))?;
     let meta_key = MediaStorage::ctx_sidecar_key(ctx, &sha256_hex);
 
     // --- 5. Idempotency check ---
     let sidecar_exists = storage.head(&meta_key).await?;
-    let blob_exists = storage.head(&key).await?;
-    if sidecar_exists && blob_exists {
+    let existing_blob_key = match storage.resolve_read_key(ctx, &legacy_key).await {
+        Ok(key) => Some(key),
+        Err(MediaError::NotFound) => None,
+        Err(error) => return Err(error),
+    };
+    if sidecar_exists && existing_blob_key.is_some() {
         let meta = storage.get_sidecar(ctx, &sha256_hex).await?;
         // Re-upload of known bytes: still a distinct upload event — see the
         // buffered path's short-circuit for the rationale.
@@ -442,6 +456,7 @@ pub async fn process_video_upload(
                 UploadEventFacts {
                     sha256: &sha256_hex,
                     ext,
+                    blob_key: existing_blob_key.as_deref().expect("checked above"),
                     mime: &mime,
                     size: file_size,
                     uploaded_at: chrono::Utc::now().timestamp(),
@@ -463,7 +478,9 @@ pub async fn process_video_upload(
     let uploaded_at = chrono::Utc::now().timestamp();
 
     // --- 6. Stream blob from temp file to S3 ---
-    storage.put_file(&key, &tmp_path, &mime).await?;
+    let blob_key = storage
+        .put_payload_file(ctx, &sha256_hex, ext, &tmp_path, &mime)
+        .await?;
     drop(tmp); // Free temp file disk space immediately after S3 upload.
 
     // --- 7. Build metadata (no thumbnail for video — desktop handles that) ---
@@ -488,6 +505,7 @@ pub async fn process_video_upload(
             UploadEventFacts {
                 sha256: &sha256_hex,
                 ext,
+                blob_key: &blob_key,
                 mime: &mime,
                 size: file_size,
                 uploaded_at,
@@ -513,6 +531,7 @@ pub async fn process_video_upload(
 async fn prepare_image_metadata(
     storage: &MediaStorage,
     config: &MediaConfig,
+    ctx: &TenantContext,
     input: MetadataInput,
 ) -> Result<BlobMeta, MediaError> {
     let body_ref = input.body.clone();
@@ -529,8 +548,7 @@ async fn prepare_image_metadata(
     meta.uploaded_at = input.uploaded_at;
 
     if let Some(ref tb) = thumb_bytes {
-        let thumb_key = format!("{}.thumb.jpg", input.sha256);
-        storage.put(&thumb_key, tb, "image/jpeg").await?;
+        storage.put_thumbnail(ctx, &input.sha256, tb).await?;
     }
 
     Ok(meta)
@@ -571,6 +589,7 @@ mod tests {
             s3_bucket: String::new(),
             s3_region: "us-east-1".to_string(),
             s3_addressing_style: crate::config::S3AddressingStyle::Path,
+            key_layout: crate::config::MediaKeyLayout::Legacy,
             max_image_bytes: 50 * 1024 * 1024,
             max_gif_bytes: 10 * 1024 * 1024,
             max_video_bytes: 524_288_000,

crates/buzz-media/src/upload_record.rs

@@ -125,6 +125,8 @@ pub struct UploadEventFacts<'a> {
     pub sha256: &'a str,
     /// Canonical extension.
     pub ext: &'a str,
+    /// Exact object key containing the payload bytes.
+    pub blob_key: &'a str,
     /// Sniffed MIME type.
     pub mime: &'a str,
     /// Uploaded byte size.
@@ -158,10 +160,7 @@ pub async fn record_upload_event(
         event_id: event_id.clone(),
         sha256: facts.sha256.to_string(),
         ext: facts.ext.to_string(),
-        blob_key: Some(
-            crate::keys::legacy_blob_key(facts.sha256, facts.ext)
-                .map_err(|e| crate::error::MediaError::StorageError(e.to_string()))?,
-        ),
+        blob_key: Some(facts.blob_key.to_string()),
         mime_type: facts.mime.to_string(),
         size: facts.size,
         uploaded_at: facts.uploaded_at,

crates/buzz-media/src/validation.rs

@@ -950,6 +950,7 @@ mod tests {
             s3_bucket: String::new(),
             s3_region: "us-east-1".to_string(),
             s3_addressing_style: crate::config::S3AddressingStyle::Path,
+            key_layout: crate::config::MediaKeyLayout::Legacy,
             max_image_bytes: 50 * 1024 * 1024,
             max_gif_bytes: 10 * 1024 * 1024,
             max_video_bytes: 524_288_000,

crates/buzz-media/tests/static_creds_minio.rs

@@ -35,6 +35,7 @@ fn minio_config() -> MediaConfig {
             .unwrap_or_else(|_| "path".to_string())
             .parse()
             .expect("BUZZ_S3_ADDRESSING_STYLE must be path or virtual"),
+        key_layout: buzz_media::MediaKeyLayout::Legacy,
         max_image_bytes: 50 * 1024 * 1024,
         max_gif_bytes: 10 * 1024 * 1024,
         max_video_bytes: 524_288_000,