use anyhow::{anyhow, Result};
use chrono::Utc;
use serde::Deserialize;
use std::path::PathBuf;
use crate::db_client::{DatabaseConfig, ScoutDbClient};
use crate::scan::{
FileCheckpointStore, HasRowId, HasScanTime, ScanCheckpoint, ScanCursor, ScanRequest,
};
use super::mirror::GcsMediaMirror;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MediaSource {
Artifacts,
Events,
}
impl MediaSource {
pub fn table(self) -> &'static str {
match self {
Self::Artifacts => "artifacts",
Self::Events => "events",
}
}
fn select(self) -> &'static str {
match self {
Self::Artifacts => "id,file_path,thumbnail_file_path,proxy_file_path,updated_at",
Self::Events => "id,file_path",
}
}
fn time_column(self) -> Option<&'static str> {
match self {
Self::Artifacts => Some("updated_at"),
Self::Events => None,
}
}
fn storage_bucket(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,
}
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,
}
}
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,
}
#[derive(Debug, Deserialize)]
struct ArtifactMediaRow {
id: Option<i64>,
file_path: String,
thumbnail_file_path: Option<String>,
proxy_file_path: Option<String>,
updated_at: Option<String>,
}
#[derive(Debug, Deserialize)]
struct EventMediaRow {
id: Option<i64>,
file_path: Option<String>,
}
impl HasRowId for ArtifactMediaRow {
fn row_id(&self) -> Option<i64> {
self.id
}
}
impl HasScanTime for ArtifactMediaRow {
fn scan_time(&self, time_column: &str) -> Option<&str> {
match time_column {
"updated_at" => self.updated_at.as_deref(),
_ => None,
}
}
}
impl HasRowId for EventMediaRow {
fn row_id(&self) -> Option<i64> {
self.id
}
}
impl HasScanTime for EventMediaRow {
fn scan_time(&self, _time_column: &str) -> Option<&str> {
None
}
}
async fn sync_media(
db: &mut ScoutDbClient,
mirror: &GcsMediaMirror,
options: MediaSyncOptions,
) -> Result<MediaSyncSummary> {
match options.source {
MediaSource::Artifacts => {
run_source::<ArtifactMediaRow>(db, mirror, options, artifact_paths).await
}
MediaSource::Events => run_source::<EventMediaRow>(db, mirror, options, event_paths).await,
}
}
async fn run_source<T>(
db: &mut ScoutDbClient,
mirror: &GcsMediaMirror,
options: MediaSyncOptions,
paths_fn: fn(&[T]) -> Vec<String>,
) -> Result<MediaSyncSummary>
where
T: for<'de> serde::Deserialize<'de> + HasRowId + HasScanTime,
{
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) => {
let timed = options.source.time_column().is_some();
match (&checkpoint.cursor, timed) {
(ScanCursor::Start, _)
| (ScanCursor::AfterTimeId { .. }, true)
| (ScanCursor::AfterId { .. }, false) => checkpoint.cursor,
_ => {
return Err(anyhow!(
"checkpoint '{}' cursor does not match {} scan mode; use a new job_id",
options.job_id,
table
));
}
}
}
None => ScanCursor::Start,
};
let mut pages = 0usize;
let mut rows = 0usize;
let mut media_objects = 0usize;
loop {
let mut request = ScanRequest::new(table, options.source.select())
.with_batch_size(options.batch_size)
.with_cursor(cursor.clone());
if let Some(col) = options.source.time_column() {
request = request.with_time_column(col);
}
let page = db.scan_page::<T>(request).await?;
let Some(next_cursor) = page.next.clone() else {
break;
};
media_objects += mirror
.mirror_objects(&paths_fn(&page.rows), options.source.storage_bucket())
.await?;
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,
})
}
fn push_nonempty(paths: &mut Vec<String>, value: &str) {
let value = value.trim();
if !value.is_empty() {
paths.push(value.to_string());
}
}
fn artifact_paths(rows: &[ArtifactMediaRow]) -> Vec<String> {
let mut paths = Vec::new();
for row in rows {
push_nonempty(&mut paths, &row.file_path);
if let Some(p) = &row.thumbnail_file_path {
push_nonempty(&mut paths, p);
}
if let Some(p) = &row.proxy_file_path {
push_nonempty(&mut paths, p);
}
}
paths
}
fn event_paths(rows: &[EventMediaRow]) -> Vec<String> {
let mut paths = Vec::new();
for row in rows {
if let Some(p) = &row.file_path {
push_nonempty(&mut paths, p);
}
}
paths
}