object-store: provider-neutral seam + native Google Cloud Storage provider with generation CAS

Cargo.toml

@@ -18,6 +18,7 @@ members = [
     "crates/buzz-deletion",
     "crates/buzz-workflow",
     "crates/buzz-media",
+    "crates/buzz-object-store",
     "crates/buzz-cli",
     "crates/buzz-pairing-cli",
     "crates/buzz-sdk",
@@ -146,6 +147,7 @@ buzz-search = { path = "crates/buzz-search" }
 buzz-audit = { path = "crates/buzz-audit" }
 buzz-workflow = { path = "crates/buzz-workflow" }
 buzz-media = { path = "crates/buzz-media" }
+buzz-object-store = { path = "crates/buzz-object-store" }
 buzz-sdk = { path = "crates/buzz-sdk" }
 buzz-ws-client = { path = "crates/buzz-ws-client" }
 buzz-relay-mesh = { path = "crates/buzz-relay-mesh" }

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

@@ -699,7 +699,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 40);
+        assert_eq!(migrations.len(), 41);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -2400,6 +2400,75 @@ mod tests {
         assert_eq!(after, vec![(1, Some(true)), (30_179, None), (30_350, None)]);
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn function_search_path_migration_repairs_restore_context() {
+        let pool = connect_test_pool().await;
+        reset_public_schema(&pool).await;
+        run_migrations_through(&pool, 40)
+            .await
+            .expect("apply migrations through 40");
+
+        let community = uuid::Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community)
+            .bind(format!("restore-proof-{}.example", community.simple()))
+            .execute(&pool)
+            .await
+            .expect("insert restore proof community");
+
+        let mut before = pool.acquire().await.expect("pre-migration connection");
+        sqlx::query("BEGIN")
+            .execute(&mut *before)
+            .await
+            .expect("begin pre-migration transaction");
+        sqlx::query("SET LOCAL search_path TO ''")
+            .execute(&mut *before)
+            .await
+            .expect("clear pre-migration search path");
+        let before_error = sqlx::query("SELECT public.assert_community_write_allowed($1)")
+            .bind(community)
+            .execute(&mut *before)
+            .await
+            .expect_err("migration 40 must reproduce restore name-resolution failure");
+        assert_eq!(
+            before_error
+                .as_database_error()
+                .and_then(sqlx::error::DatabaseError::code)
+                .as_deref(),
+            Some("42883")
+        );
+        sqlx::query("ROLLBACK")
+            .execute(&mut *before)
+            .await
+            .expect("rollback pre-migration transaction");
+        drop(before);
+
+        run_migrations(&pool)
+            .await
+            .expect("apply function search-path migration");
+
+        let mut after = pool.acquire().await.expect("post-migration connection");
+        sqlx::query("BEGIN")
+            .execute(&mut *after)
+            .await
+            .expect("begin restore-like transaction");
+        sqlx::query("SET LOCAL search_path TO ''")
+            .execute(&mut *after)
+            .await
+            .expect("clear restore-like search path");
+        sqlx::query("INSERT INTO public.users (community_id, pubkey) VALUES ($1, $2)")
+            .bind(community)
+            .bind(vec![0x41_u8; 32])
+            .execute(&mut *after)
+            .await
+            .expect("restore-like insert must traverse the repaired write fence");
+        sqlx::query("ROLLBACK")
+            .execute(&mut *after)
+            .await
+            .expect("finish restore-like transaction");
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {
@@ -2475,6 +2544,57 @@ mod tests {
                 .await
                 .expect("insert late-table test community");
         }
+
+        let unpinned_application_functions: i64 = sqlx::query_scalar(
+            "SELECT count(*)::BIGINT \
+               FROM pg_proc AS procedure \
+               JOIN pg_namespace AS namespace \
+                 ON namespace.oid = procedure.pronamespace \
+              WHERE namespace.nspname = 'public' \
+                AND procedure.prokind = 'f' \
+                AND procedure.proowner = (SELECT oid FROM pg_roles WHERE rolname = current_user) \
+                AND NOT EXISTS (\
+                    SELECT 1 FROM pg_depend AS dependency \
+                     WHERE dependency.classid = 'pg_proc'::REGCLASS \
+                       AND dependency.objid = procedure.oid \
+                       AND dependency.deptype = 'e'\
+                ) \
+                AND NOT coalesce(procedure.proconfig, ARRAY[]::TEXT[]) \
+                    @> ARRAY['search_path=public, pg_catalog']",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("inspect application function search paths");
+        assert_eq!(
+            unpinned_application_functions, 0,
+            "every application-owned function must resolve independently of the restore session"
+        );
+
+        let mut restore_connection = pool.acquire().await.expect("restore-like connection");
+        sqlx::query("BEGIN")
+            .execute(&mut *restore_connection)
+            .await
+            .expect("begin restore-like transaction");
+        sqlx::query("SET LOCAL search_path TO ''")
+            .execute(&mut *restore_connection)
+            .await
+            .expect("clear restore-like search path");
+        sqlx::query("SELECT public.assert_community_write_allowed($1)")
+            .bind(active_a)
+            .execute(&mut *restore_connection)
+            .await
+            .expect("write fence must resolve with an empty invoker search path");
+        sqlx::query("INSERT INTO public.users (community_id, pubkey) VALUES ($1, $2)")
+            .bind(active_a)
+            .bind(vec![0x41_u8; 32])
+            .execute(&mut *restore_connection)
+            .await
+            .expect("restore-like insert must traverse the write-fence trigger");
+        sqlx::query("ROLLBACK")
+            .execute(&mut *restore_connection)
+            .await
+            .expect("finish restore-like transaction");
+
         sqlx::query(
             "CREATE TABLE late_created_scoped (\
                  community_id UUID NOT NULL, id BIGINT PRIMARY KEY, value TEXT NOT NULL\

crates/buzz-deletion/Cargo.toml

@@ -13,6 +13,7 @@ thiserror = { workspace = true }
 buzz-core = { workspace = true }
 buzz-db = { workspace = true }
 buzz-media = { workspace = true }
+buzz-object-store = { workspace = true }
 chrono = { workspace = true }
 clap = { version = "4", features = ["derive"] }
 deadpool-redis = { workspace = true }

crates/buzz-deletion/src/lib.rs

@@ -561,27 +561,8 @@ async fn connect_services() -> Result<Services> {
 }
 
 async fn connect_services_with_store(store: DeletionStore) -> Result<Services> {
-    let (s3_access_key, s3_secret_key) = s3_key_pair_from_env();
-    let media_config = buzz_media::MediaConfig {
-        s3_endpoint: required_env("BUZZ_S3_ENDPOINT")?,
-        s3_access_key,
-        s3_secret_key,
-        s3_bucket: required_env("BUZZ_S3_BUCKET")?,
-        s3_region: s3_region_from_env(),
-        s3_addressing_style: std::env::var("BUZZ_S3_ADDRESSING_STYLE")
-            .unwrap_or_else(|_| "path".to_string())
-            .parse()
-            .map_err(anyhow::Error::msg)?,
-        max_image_bytes: 1,
-        max_gif_bytes: 1,
-        max_video_bytes: 1,
-        max_file_bytes: 1,
-        public_base_url: "http://localhost/media".to_string(),
-        upload_records_enabled: false,
-        upload_ip_header: None,
-        upload_port_header: None,
-    };
-    let media = Arc::new(MediaStorage::new(&media_config)?);
+    let object_store = buzz_object_store::connect(&deletion_object_store_config()?).await?;
+    let media = Arc::new(MediaStorage::with_store(object_store));
     let redis_url = required_env("REDIS_URL")?;
     let mut redis_config = deadpool_redis::Config::from_url(&redis_url);
     redis_config.pool = Some(deadpool_redis::PoolConfig::new(env_parse(
@@ -598,6 +579,37 @@ async fn connect_services_with_store(store: DeletionStore) -> Result<Services> {
     })
 }
 
+/// Resolve the same deployment-level provider used by the relay.
+///
+/// The deletion executable is a separate composition root, so it must select
+/// the provider independently while preserving exactly the relay's environment
+/// contract. Domain deletion code continues to consume `MediaStorage` only.
+fn deletion_object_store_config() -> Result<buzz_object_store::ObjectStoreConfig> {
+    match buzz_object_store::ProviderSelection::from_env().map_err(anyhow::Error::msg)? {
+        buzz_object_store::ProviderSelection::Gcs { bucket } => {
+            Ok(buzz_object_store::ObjectStoreConfig::Gcs(
+                buzz_object_store::GcsStoreConfig::new(bucket),
+            ))
+        }
+        buzz_object_store::ProviderSelection::S3 => {
+            let (access_key, secret_key) = s3_key_pair_from_env();
+            Ok(buzz_object_store::ObjectStoreConfig::S3(
+                buzz_object_store::S3StoreConfig {
+                    endpoint: required_env("BUZZ_S3_ENDPOINT")?,
+                    access_key,
+                    secret_key,
+                    bucket: required_env("BUZZ_S3_BUCKET")?,
+                    region: s3_region_from_env(),
+                    addressing_style: std::env::var("BUZZ_S3_ADDRESSING_STYLE")
+                        .unwrap_or_else(|_| "path".to_string())
+                        .parse()
+                        .map_err(anyhow::Error::msg)?,
+                },
+            ))
+        }
+    }
+}
+
 fn s3_region_from_env() -> String {
     resolve_s3_region(
         std::env::var("BUZZ_S3_REGION").ok(),
@@ -1605,6 +1617,12 @@ fn print_json(value: &impl Serialize) -> Result<()> {
 mod tests {
     use super::*;
 
+    fn s3_media_storage(config: buzz_object_store::S3StoreConfig) -> Arc<MediaStorage> {
+        let store =
+            buzz_object_store::S3ObjectStore::new(&config).expect("construct S3 test object store");
+        Arc::new(MediaStorage::with_store(Arc::new(store)))
+    }
+
     #[test]
     fn submit_host_prefers_explicit_host() {
         assert_eq!(
@@ -1693,25 +1711,14 @@ mod tests {
             .expect("runnable deletion request");
         let services = Services {
             store,
-            media: Arc::new(
-                MediaStorage::new(&buzz_media::MediaConfig {
-                    s3_endpoint: "http://127.0.0.1:1".to_string(),
-                    s3_access_key: "unused".to_string(),
-                    s3_secret_key: "unused".to_string(),
-                    s3_bucket: "unused".to_string(),
-                    s3_region: "us-east-1".to_string(),
-                    s3_addressing_style: buzz_media::S3AddressingStyle::Path,
-                    max_image_bytes: 1,
-                    max_gif_bytes: 1,
-                    max_video_bytes: 1,
-                    max_file_bytes: 1,
-                    public_base_url: "http://localhost/media".to_string(),
-                    upload_records_enabled: false,
-                    upload_ip_header: None,
-                    upload_port_header: None,
-                })
-                .expect("construct unused media service"),
-            ),
+            media: s3_media_storage(buzz_object_store::S3StoreConfig {
+                endpoint: "http://127.0.0.1:1".to_string(),
+                access_key: "unused".to_string(),
+                secret_key: "unused".to_string(),
+                bucket: "unused".to_string(),
+                region: "us-east-1".to_string(),
+                addressing_style: buzz_object_store::S3AddressingStyle::Path,
+            }),
             redis: deadpool_redis::Config::from_url("redis://127.0.0.1:1")
                 .create_pool(Some(deadpool_redis::Runtime::Tokio1))
                 .expect("construct unused Redis pool"),
@@ -1794,27 +1801,16 @@ mod tests {
         let bucket = std::env::var("BUZZ_TEST_S3_BUCKET")
             .or_else(|_| std::env::var("BUZZ_S3_BUCKET"))
             .expect("BUZZ_TEST_S3_BUCKET or BUZZ_S3_BUCKET is required");
-        Arc::new(
-            MediaStorage::new(&buzz_media::MediaConfig {
-                s3_endpoint: endpoint,
-                s3_access_key: access_key,
-                s3_secret_key: secret_key,
-                s3_bucket: bucket,
-                s3_region: std::env::var("BUZZ_TEST_S3_REGION")
-                    .or_else(|_| std::env::var("BUZZ_S3_REGION"))
-                    .unwrap_or_else(|_| "us-east-1".to_string()),
-                s3_addressing_style: buzz_media::S3AddressingStyle::Path,
-                max_image_bytes: 1,
-                max_gif_bytes: 1,
-                max_video_bytes: 1,
-                max_file_bytes: 1,
-                public_base_url: "http://localhost/media".to_string(),
-                upload_records_enabled: false,
-                upload_ip_header: None,
-                upload_port_header: None,
-            })
-            .expect("construct deletion test media service"),
-        )
+        s3_media_storage(buzz_object_store::S3StoreConfig {
+            endpoint,
+            access_key,
+            secret_key,
+            bucket,
+            region: std::env::var("BUZZ_TEST_S3_REGION")
+                .or_else(|_| std::env::var("BUZZ_S3_REGION"))
+                .unwrap_or_else(|_| "us-east-1".to_string()),
+            addressing_style: buzz_object_store::S3AddressingStyle::Path,
+        })
     }
 
     #[tokio::test]

crates/buzz-media/Cargo.toml

@@ -9,6 +9,7 @@ description = "Media storage, validation, and thumbnail generation for Buzz"
 
 [dependencies]
 buzz-core = { workspace = true }
+buzz-object-store = { workspace = true }
 nostr = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
@@ -21,7 +22,6 @@ chrono = { workspace = true }
 ulid = "1"
 uuid = { workspace = true }
 axum = { workspace = true }
-s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] }
 infer = "0.19"
 image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
 blurhash = "0.2"
@@ -32,7 +32,6 @@ tempfile = "3"
 tokio-util = { version = "0.7", features = ["io"] }
 futures-util = "0.3"
 futures-core = "0.3"
-quick-xml = { version = "0.38", features = ["serialize"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }