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::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 abs = ctx.storage.absolute_path(&upload.item_path);
        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;
            }
        };

        for (suffix, buf) in derived {
            if let Err(e) = ctx.storage.put_derived(&upload.item_path, &suffix, &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
    })
}