neiam /dms

use futures_util::future::BoxFuture;
use sea_orm::*;
use tracing::{error, info, warn};

use crate::entities::uploads;
use crate::jobs;
use crate::queue::{JobKind, JobMessage, JobOutcome};
use crate::storage::Storage;
use crate::workers::poster::POSTER_EXT;
use crate::workers::WorkerContext;

const SIZES: &[(&str, u32)] = &[("thumb", 256), ("medium", 1024)];

pub fn run(ctx: WorkerContext, msg: JobMessage) -> BoxFuture<'static, JobOutcome> {
    Box::pin(async move {
        // Idempotency: if upload_jobs row says complete, skip.
        if matches!(jobs::is_complete(ctx.db.as_ref(), msg.upload_id, JobKind::Thumbnail).await, Ok(true)) {
            info!(upload_id = %msg.upload_id, "thumbnail already complete, skipping");
            return JobOutcome::Ack;
        }
        let _ = jobs::mark_processing(ctx.db.as_ref(), msg.upload_id, JobKind::Thumbnail).await;

        let upload = match uploads::Entity::find_by_id(msg.upload_id)
            .one(ctx.db.as_ref())
            .await
        {
            Ok(Some(u)) => u,
            Ok(None) => {
                warn!(upload_id = %msg.upload_id, "upload not found, dropping");
                let _ = jobs::mark_failed(ctx.db.as_ref(), msg.upload_id, JobKind::Thumbnail, "upload not found").await;
                return JobOutcome::Reject;
            }
            Err(e) => {
                error!(error = %e, "db error loading upload");
                return JobOutcome::Retry;
            }
        };

        let source = crate::workers::source_image_relative(&upload);
        let abs = ctx.storage.absolute_path(&source);
        let bytes = match tokio::fs::read(&abs).await {
            Ok(b) => b,
            Err(e) => {
                error!(error = %e, path = %abs.display(), "read original failed");
                let _ = jobs::mark_failed(ctx.db.as_ref(), upload.id, JobKind::Thumbnail, &format!("read: {e}")).await;
                return JobOutcome::Reject;
            }
        };

        // Decode + resize on a blocking thread, then write out async.
        let res = tokio::task::spawn_blocking(move || -> Result<Vec<(String, Vec<u8>)>, String> {
            let img = image::load_from_memory(&bytes).map_err(|e: image::ImageError| e.to_string())?;
            let mut out = Vec::new();
            for (suffix, max_dim) in SIZES {
                let resized = img.thumbnail(*max_dim, *max_dim);
                let mut buf: Vec<u8> = Vec::new();
                resized
                    .write_to(&mut std::io::Cursor::new(&mut buf), image::ImageFormat::Jpeg)
                    .map_err(|e: image::ImageError| e.to_string())?;
                out.push((ToString::to_string(suffix), buf));
            }
            Ok(out)
        })
        .await;

        let derived = match res {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => {
                error!(upload_id = %upload.id, error = %e, "thumbnail generation failed");
                let _ = jobs::mark_failed(ctx.db.as_ref(), upload.id, JobKind::Thumbnail, &e).await;
                return JobOutcome::Reject;
            }
            Err(e) => {
                error!(error = %e, "thumbnail blocking task panicked");
                return JobOutcome::Retry;
            }
        };

        // Thumbnails are always JPEG bytes. For an image the artifact keeps the
        // original's extension (long-standing naming, left alone); for a video
        // that would produce a `.thumb.mp4` full of JPEG, so name it honestly.
        let derived_ext = if crate::workers::is_video(&upload) {
            POSTER_EXT.to_string()
        } else {
            std::path::Path::new(&upload.item_path)
                .extension()
                .and_then(|s| s.to_str())
                .unwrap_or("bin")
                .to_string()
        };
        for (suffix, buf) in derived {
            if let Err(e) = ctx
                .storage
                .put_derived_as(&upload.item_path, &suffix, &derived_ext, &buf)
                .await
            {
                error!(error = %e, "write thumbnail failed");
                return JobOutcome::Retry;
            }
        }

        let _ = jobs::mark_complete(ctx.db.as_ref(), upload.id, JobKind::Thumbnail).await;
        info!(upload_id = %upload.id, "thumbnails generated");
        JobOutcome::Ack
    })
}