neiam /dms

//! Queue the `preview` job for videos and GIFs that don't have one yet.
//!
//! New uploads get a preview as part of their default job set; this is for
//! everything uploaded before that job existed. Selection is by what's on
//! disk rather than by job rows: an upload counts as done once its
//! `<stem>.preview.webp` exists, so rerunning after a partial pass only
//! picks up what's still missing.

use clap::Parser;
use dms::{
    config::{create_db_connection, Config},
    entities::uploads,
    jobs,
    queue::{connect, publish_job, JobKind, JobMessage},
    storage::{derived_relative, LocalStorage, Storage},
    workers::preview::{wants_preview, PREVIEW_EXT, PREVIEW_SUFFIX},
};
use sea_orm::{ColumnTrait, Condition, EntityTrait, QueryFilter, QueryOrder};

#[derive(Parser)]
#[command(name = "backfill-previews")]
#[command(about = "Queue animated grid previews for videos and GIFs that lack one")]
struct Cli {
    /// Queue at most this many (newest first). Unlimited by default.
    #[arg(short = 'n', long)]
    count: Option<usize>,

    /// Requeue even uploads whose preview file already exists.
    #[arg(long)]
    force: bool,

    /// Show what would be queued and exit without touching anything.
    #[arg(long)]
    dry_run: bool,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();
    let cli = Cli::parse();

    let config = Config::from_env()?;
    let db = create_db_connection(&config.database_url).await?;
    let storage = LocalStorage::new(&config.storage_root);

    let candidates = uploads::Entity::find()
        .filter(
            Condition::any()
                .add(uploads::Column::ContentType.like("video/%"))
                .add(uploads::Column::ContentType.eq("image/gif")),
        )
        .order_by_desc(uploads::Column::DateUploaded)
        .all(&db)
        .await?;

    let mut selected: Vec<uploads::Model> = candidates
        .into_iter()
        .filter(wants_preview)
        .filter(|u| {
            cli.force
                || !storage
                    .absolute_path(&derived_relative(&u.item_path, PREVIEW_SUFFIX, PREVIEW_EXT))
                    .exists()
        })
        .collect();
    if let Some(n) = cli.count {
        selected.truncate(n);
    }

    if selected.is_empty() {
        println!("Every video and GIF already has a preview.");
        return Ok(());
    }

    println!("{} upload(s) need a preview:", selected.len());
    for up in &selected {
        println!(
            "  {}  {}  {}",
            up.date_uploaded.format("%Y-%m-%d %H:%M"),
            up.id,
            up.file_name_original,
        );
    }
    if cli.dry_run {
        println!("\nDry run: nothing queued.");
        return Ok(());
    }

    let conn = connect(&config.amqp_url).await.map_err(|e| {
        eprintln!("Could not reach the broker at {}: {e}", config.amqp_url);
        e
    })?;

    let mut queued = 0usize;
    let mut failed = 0usize;
    for up in &selected {
        // Resets the row (and its attempts), so a preview that failed before
        // gets a fresh try — and the worker's "already complete" skip doesn't
        // swallow a --force.
        if let Err(e) = jobs::upsert_pending(&db, up.id, JobKind::Preview).await {
            eprintln!("  {} upsert_pending failed: {e}", up.id);
            failed += 1;
            continue;
        }
        let msg = JobMessage::new(JobKind::Preview, up.id, up.user_id);
        if let Err(e) = publish_job(&conn, &msg).await {
            eprintln!("  {} publish failed: {e}", up.id);
            failed += 1;
            continue;
        }
        queued += 1;
    }

    println!("Queued {queued} preview job(s).");
    if failed > 0 {
        eprintln!("{failed} upload(s) could not be queued.");
        std::process::exit(1);
    }
    Ok(())
}