scout_rs 0.95.95

use anyhow::{anyhow, Result};
use chrono::Utc;
use serde::Deserialize;
use std::path::PathBuf;

use crate::db_client::{DatabaseConfig, ScoutDbClient};
use crate::scan::{FileCheckpointStore, ScanCheckpoint, ScanCursor, ScanRequest, ScanRow};

use super::mirror::GcsMediaMirror;

/// Both media tables expose the same paths, so one projection serves either source.
const MEDIA_SELECT: &str = "id,file_path,thumbnail_file_path,proxy_file_path,updated_at";

/// Resuming on `(updated_at, id)` catches derivative paths written after insert.
const MEDIA_TIME_COLUMN: &str = "updated_at";

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MediaSource {
    Artifacts,
    Events,
}

impl MediaSource {
    /// Also the Storage bucket holding the table's originals; derivative paths carry their own
    /// bucket prefix.
    pub fn table(self) -> &'static str {
        match self {
            Self::Artifacts => "artifacts",
            Self::Events => "events",
        }
    }
}

pub struct MediaSyncClient {
    db: ScoutDbClient,
    mirror: GcsMediaMirror,
}

impl MediaSyncClient {
    pub fn from_env(gcs_bucket: impl Into<String>) -> Result<Self> {
        dotenv::dotenv().ok();
        let config = DatabaseConfig::from_env_service_role().ok_or_else(|| {
            anyhow!("SCOUT_DATABASE_REST_URL and SUPABASE_SERVICE_ROLE_KEY are required")
        })?;
        Self::new(config, gcs_bucket)
    }

    pub fn new(mut config: DatabaseConfig, gcs_bucket: impl Into<String>) -> Result<Self> {
        config.use_service_key = true;
        let mirror = GcsMediaMirror::new(&config, gcs_bucket)?;
        let mut db = ScoutDbClient::new(config);
        db.connect()?;
        Ok(Self { db, mirror })
    }

    pub async fn sync(&mut self, options: MediaSyncOptions) -> Result<MediaSyncSummary> {
        sync_media(&mut self.db, &self.mirror, options).await
    }
}

#[derive(Debug, Clone)]
pub struct MediaSyncOptions {
    source: MediaSource,
    job_id: String,
    checkpoint_dir: PathBuf,
    batch_size: usize,
    concurrency: usize,
}

impl MediaSyncOptions {
    pub fn new(source: MediaSource, job_id: impl Into<String>) -> Self {
        Self {
            source,
            job_id: job_id.into(),
            checkpoint_dir: "./.scout/checkpoints".into(),
            batch_size: 500,
            concurrency: 4,
        }
    }

    /// Objects mirrored at once. Each holds up to ~25 MiB of upload buffers.
    pub fn with_concurrency(mut self, concurrency: usize) -> Self {
        self.concurrency = concurrency;
        self
    }

    pub fn with_checkpoint_dir(mut self, checkpoint_dir: impl Into<PathBuf>) -> Self {
        self.checkpoint_dir = checkpoint_dir.into();
        self
    }

    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
        self.batch_size = batch_size;
        self
    }
}

#[derive(Debug, Clone)]
pub struct MediaSyncSummary {
    pub table: String,
    pub job_id: String,
    pub pages: usize,
    pub rows: usize,
    pub media_objects: usize,
    /// Objects skipped because the destination key already exists.
    pub existing_objects: usize,
    /// Paths referenced by rows whose Storage object no longer exists.
    pub missing_paths: Vec<String>,
}

#[derive(Debug, Deserialize)]
struct MediaRow {
    id: i64,
    file_path: Option<String>,
    thumbnail_file_path: Option<String>,
    proxy_file_path: Option<String>,
    updated_at: String,
}

impl ScanRow for MediaRow {
    fn row_id(&self) -> i64 {
        self.id
    }

    fn scan_time(&self) -> &str {
        &self.updated_at
    }
}

async fn sync_media(
    db: &mut ScoutDbClient,
    mirror: &GcsMediaMirror,
    options: MediaSyncOptions,
) -> Result<MediaSyncSummary> {
    let table = options.source.table();
    let store = FileCheckpointStore::new(&options.checkpoint_dir);
    let mut cursor = match store.load(&options.job_id)? {
        Some(checkpoint) if checkpoint.table != table => {
            return Err(anyhow!(
                "checkpoint '{}' belongs to table '{}', not '{}'; use a distinct job_id per source",
                options.job_id,
                checkpoint.table,
                table
            ));
        }
        Some(checkpoint) => checkpoint.cursor,
        None => ScanCursor::Start,
    };

    let mut pages = 0usize;
    let mut rows = 0usize;
    let mut media_objects = 0usize;
    let mut existing_objects = 0usize;
    let mut missing_paths = Vec::new();

    loop {
        let request = ScanRequest::new(table, MEDIA_SELECT, MEDIA_TIME_COLUMN)
            .with_batch_size(options.batch_size)
            .with_cursor(cursor.clone());
        let page = db.scan_page::<MediaRow>(request).await?;
        let Some(next_cursor) = page.next.clone() else {
            break;
        };

        let outcome = mirror
            .mirror_objects(&media_paths(&page.rows), table, options.concurrency)
            .await?;
        media_objects += outcome.mirrored;
        existing_objects += outcome.existing;
        missing_paths.extend(outcome.missing);

        store.save(&ScanCheckpoint {
            table: table.to_string(),
            job_id: options.job_id.clone(),
            cursor: next_cursor.clone(),
            updated_at: Utc::now().to_rfc3339(),
        })?;

        pages += 1;
        rows += page.rows.len();
        let exhausted = page.exhausted;
        cursor = next_cursor;
        if exhausted {
            break;
        }
    }

    Ok(MediaSyncSummary {
        table: table.to_string(),
        job_id: options.job_id,
        pages,
        rows,
        media_objects,
        existing_objects,
        missing_paths,
    })
}

/// Every non-empty storage path a page of rows refers to; the mirror dedupes them.
fn media_paths(rows: &[MediaRow]) -> Vec<String> {
    rows.iter()
        .flat_map(|row| {
            [
                &row.file_path,
                &row.thumbnail_file_path,
                &row.proxy_file_path,
            ]
        })
        .flatten()
        .map(|path| path.trim())
        .filter(|path| !path.is_empty())
        .map(str::to_string)
        .collect()
}