scout_rs 0.95.77

use anyhow::{anyhow, Result};
use object_store::gcp::GoogleCloudStorageBuilder;
use object_store::{ObjectStore, PutPayload, WriteMultipart};
use std::sync::Arc;

use crate::db_client::DatabaseConfig;
use crate::storage::{StorageClient, StorageConfig};

/// Streams Supabase Storage objects into GCS at the same object key as the DB path.
pub struct GcsMediaMirror {
    storage: StorageClient,
    gcs: Arc<dyn ObjectStore>,
}

impl GcsMediaMirror {
    pub fn new(db: &DatabaseConfig, gcs_bucket: impl Into<String>) -> Result<Self> {
        let bucket = gcs_bucket.into();
        if bucket.is_empty() {
            return Err(anyhow!("gcs bucket must not be empty"));
        }
        let service_key = db
            .service_role_key
            .clone()
            .ok_or_else(|| anyhow!("SUPABASE_SERVICE_ROLE_KEY required to mirror media"))?;
        let storage = StorageClient::new(StorageConfig {
            supabase_url: db.rest_url.replace("/rest/v1", ""),
            supabase_anon_key: service_key,
            scout_api_key: db.scout_api_key.clone(),
            bucket_name: "artifacts".to_string(),
            allowed_extensions: vec![],
        })?;
        let gcs = GoogleCloudStorageBuilder::from_env()
            .with_bucket_name(&bucket)
            .build()
            .map_err(|e| anyhow!("gcs client: {e}"))?;
        Ok(Self {
            storage,
            gcs: Arc::new(gcs),
        })
    }

    async fn mirror_object(&self, storage_path: &str, source_bucket: &str) -> Result<()> {
        if storage_path.is_empty() {
            return Err(anyhow!("storage path must not be empty"));
        }

        let mut response = self
            .storage
            .open_storage_object_from_bucket(storage_path, source_bucket)
            .await?;
        let gcs_path = object_store::path::Path::from(storage_path);
        let upload = self
            .gcs
            .put_multipart(&gcs_path)
            .await
            .map_err(|e| anyhow!("gcs multipart {storage_path}: {e}"))?;
        let mut writer = WriteMultipart::new(upload);

        let mut wrote_bytes = false;
        loop {
            match response.chunk().await {
                Ok(Some(chunk)) if !chunk.is_empty() => {
                    writer
                        .wait_for_capacity(4)
                        .await
                        .map_err(|e| anyhow!("gcs upload {storage_path}: {e}"))?;
                    writer.put(chunk);
                    wrote_bytes = true;
                }
                Ok(Some(_)) => continue,
                Ok(None) => break,
                Err(e) => {
                    let _ = writer.abort().await;
                    return Err(anyhow!("download stream {storage_path}: {e}"));
                }
            }
        }

        if wrote_bytes {
            writer
                .finish()
                .await
                .map_err(|e| anyhow!("gcs complete {storage_path}: {e}"))?;
        } else {
            let _ = writer.abort().await;
            self.gcs
                .put(&gcs_path, PutPayload::new())
                .await
                .map_err(|e| anyhow!("gcs put empty {storage_path}: {e}"))?;
        }
        Ok(())
    }

    pub(crate) async fn mirror_objects(
        &self,
        paths: &[String],
        source_bucket: &str,
    ) -> Result<usize> {
        let mut mirrored = 0usize;
        let mut seen = std::collections::HashSet::new();
        for path in paths {
            if !seen.insert(path.as_str()) {
                continue;
            }
            self.mirror_object(path, source_bucket).await?;
            mirrored += 1;
        }
        Ok(mirrored)
    }
}