scout_rs 0.95.95

use anyhow::{anyhow, Result};
use futures::stream::{self, StreamExt};
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<MirrorStatus> {
        if storage_path.is_empty() {
            return Err(anyhow!("storage path must not be empty"));
        }

        let gcs_path = object_store::path::Path::from(storage_path);
        if object_exists(self.gcs.as_ref(), &gcs_path).await? {
            return Ok(MirrorStatus::Existing);
        }

        let Some(mut response) = self
            .storage
            .open_storage_object_from_bucket(storage_path, source_bucket)
            .await?
        else {
            return Ok(MirrorStatus::Missing);
        };
        let upload = self
            .gcs
            .put_multipart(&gcs_path)
            .await
            .map_err(|e| anyhow!("gcs multipart {storage_path}: {e}"))?;

        if !stream_to_gcs(&mut response, WriteMultipart::new(upload), storage_path).await? {
            self.gcs
                .put(&gcs_path, PutPayload::new())
                .await
                .map_err(|e| anyhow!("gcs put empty {storage_path}: {e}"))?;
        }
        Ok(MirrorStatus::Mirrored)
    }

    pub(crate) async fn mirror_objects(
        &self,
        paths: &[String],
        source_bucket: &str,
        concurrency: usize,
    ) -> Result<MirrorOutcome> {
        if concurrency == 0 {
            return Err(anyhow!("concurrency must be at least 1"));
        }

        // Collect every future so an error never drops sibling uploads before they can abort.
        let results: Vec<Result<(&String, MirrorStatus)>> = stream::iter(unique_paths(paths))
            .map(|path| async move {
                let status = self.mirror_object(path, source_bucket).await?;
                Ok((path, status))
            })
            .buffer_unordered(concurrency)
            .collect()
            .await;

        let mut outcome = MirrorOutcome::default();
        let mut failures = Vec::new();
        for result in results {
            match result {
                Ok((_, MirrorStatus::Mirrored)) => outcome.mirrored += 1,
                Ok((_, MirrorStatus::Existing)) => outcome.existing += 1,
                Ok((path, MirrorStatus::Missing)) => {
                    tracing::warn!(
                        storage_path = %path,
                        bucket = %source_bucket,
                        "source object missing in Supabase Storage; skipping"
                    );
                    outcome.missing.push(path.clone());
                }
                Err(e) => {
                    tracing::error!(error = %e, "failed to mirror object");
                    failures.push(e);
                }
            }
        }

        if let Some(first) = failures.first() {
            return Err(anyhow!(
                "{} of {} objects failed to mirror; first error: {first}",
                failures.len(),
                outcome.mirrored + outcome.existing + outcome.missing.len() + failures.len()
            ));
        }

        outcome.missing.sort();
        Ok(outcome)
    }
}

#[derive(Debug, Default)]
pub(crate) struct MirrorOutcome {
    pub mirrored: usize,
    pub existing: usize,
    pub missing: Vec<String>,
}

#[derive(Debug)]
enum MirrorStatus {
    Mirrored,
    Existing,
    Missing,
}

/// Concurrent part uploads per object.
const UPLOAD_PART_CONCURRENCY: usize = 4;

async fn object_exists(store: &dyn ObjectStore, path: &object_store::path::Path) -> Result<bool> {
    match store.head(path).await {
        Ok(_) => Ok(true),
        Err(object_store::Error::NotFound { .. }) => Ok(false),
        Err(e) => Err(anyhow!("gcs head {path}: {e}")),
    }
}

/// First occurrence of each path, so concurrent workers never race on the same object key.
fn unique_paths(paths: &[String]) -> Vec<&String> {
    let mut seen = std::collections::HashSet::new();
    paths
        .iter()
        .filter(|path| seen.insert(path.as_str()))
        .collect()
}

/// Streams into GCS with abort-on-error and returns false for an empty source body.
async fn stream_to_gcs(
    response: &mut reqwest::Response,
    mut writer: WriteMultipart,
    storage_path: &str,
) -> Result<bool> {
    let mut wrote_bytes = false;
    loop {
        let chunk = match response.chunk().await {
            Ok(Some(chunk)) => chunk,
            Ok(None) => break,
            Err(e) => {
                let _ = writer.abort().await;
                return Err(anyhow!("download stream {storage_path}: {e}"));
            }
        };
        if chunk.is_empty() {
            continue;
        }
        if let Err(e) = writer.wait_for_capacity(UPLOAD_PART_CONCURRENCY).await {
            let _ = writer.abort().await;
            return Err(anyhow!("gcs upload {storage_path}: {e}"));
        }
        writer.put(chunk);
        wrote_bytes = true;
    }

    if !wrote_bytes {
        let _ = writer.abort().await;
        return Ok(false);
    }
    writer
        .finish()
        .await
        .map_err(|e| anyhow!("gcs complete {storage_path}: {e}"))?;
    Ok(true)
}

#[cfg(test)]
mod tests {
    use super::*;
    use object_store::memory::InMemory;

    #[tokio::test]
    async fn object_exists_distinguishes_present_and_missing_objects() {
        let store = InMemory::new();
        let path = object_store::path::Path::from("artifacts/1/a.mp4");

        assert!(!object_exists(&store, &path).await.unwrap());
        store
            .put(&path, PutPayload::from_static(b"media"))
            .await
            .unwrap();
        assert!(object_exists(&store, &path).await.unwrap());
    }

    #[test]
    fn unique_paths_keeps_first_occurrence_only() {
        let paths = vec![
            "artifacts/1/a.mp4".to_string(),
            "artifacts/1/b.mp4".to_string(),
            "artifacts/1/a.mp4".to_string(),
        ];
        let unique = unique_paths(&paths);
        assert_eq!(
            unique,
            vec!["artifacts/1/a.mp4", "artifacts/1/b.mp4"],
            "duplicates would let concurrent workers race on one object key"
        );
    }
}