Mangalord/backend/tests/analysis_worker.rs

//! Integration tests for the analysis worker daemon: it leases only
//! `analyze_page` jobs, acks done on success, skips already-done pages
//! (unless forced), and on terminal failure writes a `failed` row.

use std::sync::Arc;
use std::time::Duration;

use mangalord::analysis::daemon::{self, test_support::CountingDispatcher, AnalysisDaemonConfig};
use mangalord::domain::page_analysis::{
    AnalysisStatus, SafetyFlag, VisionAnalysis,
};
use mangalord::repo::page_analysis;
use sqlx::PgPool;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

async fn seed_page(pool: &PgPool) -> Uuid {
    let manga_id: Uuid =
        sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id")
            .fetch_one(pool)
            .await
            .unwrap();
    let chapter_id: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(pool)
    .await
    .unwrap();
    sqlx::query_scalar(
        "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
         VALUES ($1, 1, 'k/1.png', 'image/png') RETURNING id",
    )
    .bind(chapter_id)
    .fetch_one(pool)
    .await
    .unwrap()
}

async fn job_state(pool: &PgPool, page_id: Uuid) -> Option<String> {
    sqlx::query_scalar(
        "SELECT state FROM crawler_jobs \
          WHERE payload->>'kind' = 'analyze_page' AND payload->>'page_id' = $1",
    )
    .bind(page_id.to_string())
    .fetch_optional(pool)
    .await
    .unwrap()
}

/// Poll until the page's analyze_page job reaches `want`, or fail after 5s.
async fn wait_for_state(pool: &PgPool, page_id: Uuid, want: &str) {
    let deadline = Duration::from_secs(5);
    let result = tokio::time::timeout(deadline, async {
        loop {
            if job_state(pool, page_id).await.as_deref() == Some(want) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await;
    assert!(result.is_ok(), "job for {page_id} never reached state {want}");
}

fn spawn_with(
    pool: &PgPool,
    dispatcher: Arc<CountingDispatcher>,
) -> (daemon::AnalysisDaemonHandle, CancellationToken) {
    let cancel = CancellationToken::new();
    let handle = daemon::spawn(
        pool.clone(),
        cancel.clone(),
        AnalysisDaemonConfig {
            dispatcher,
            workers: 1,
            job_timeout: Duration::from_secs(5),
        },
    );
    (handle, cancel)
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_dispatches_and_acks_done(pool: PgPool) {
    let page_id = seed_page(&pool).await;
    page_analysis::enqueue_for_page(&pool, page_id, false)
        .await
        .unwrap();

    let dispatcher = CountingDispatcher::ok();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "done").await;
    handle.shutdown().await;

    assert_eq!(dispatcher.call_count(), 1);
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_skips_already_done_page_unless_forced(pool: PgPool) {
    let page_id = seed_page(&pool).await;
    // Pre-mark the page analyzed.
    page_analysis::persist_analysis(
        &pool,
        page_id,
        &VisionAnalysis {
            ocr_results: vec![],
            tagging_results: vec![],
            scene_description: String::new(),
            safety_flag: SafetyFlag::default(),
        },
        "m",
    )
    .await
    .unwrap();
    page_analysis::enqueue_for_page(&pool, page_id, false)
        .await
        .unwrap();

    let dispatcher = CountingDispatcher::ok();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "done").await;
    handle.shutdown().await;

    assert_eq!(
        dispatcher.call_count(),
        0,
        "a non-forced job for a done page must skip dispatch"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_reanalyzes_done_page_when_forced(pool: PgPool) {
    let page_id = seed_page(&pool).await;
    page_analysis::persist_analysis(
        &pool,
        page_id,
        &VisionAnalysis {
            ocr_results: vec![],
            tagging_results: vec![],
            scene_description: String::new(),
            safety_flag: SafetyFlag::default(),
        },
        "m",
    )
    .await
    .unwrap();
    page_analysis::enqueue_for_page(&pool, page_id, true)
        .await
        .unwrap();

    let dispatcher = CountingDispatcher::ok();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "done").await;
    handle.shutdown().await;

    assert_eq!(dispatcher.call_count(), 1, "force must re-dispatch");
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_marks_failed_row_on_terminal_failure(pool: PgPool) {
    let page_id = seed_page(&pool).await;
    page_analysis::enqueue_for_page(&pool, page_id, false)
        .await
        .unwrap();
    // Make the first failure terminal (no backoff wait in the test).
    sqlx::query(
        "UPDATE crawler_jobs SET max_attempts = 1 \
          WHERE payload->>'page_id' = $1",
    )
    .bind(page_id.to_string())
    .execute(&pool)
    .await
    .unwrap();

    let dispatcher = CountingDispatcher::failing();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "dead").await;
    handle.shutdown().await;

    let row = page_analysis::load(&pool, page_id).await.unwrap().unwrap();
    assert_eq!(row.status, AnalysisStatus::Failed);
    assert!(row.error.is_some());
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_isolates_dispatcher_panics(pool: PgPool) {
    let page_id = seed_page(&pool).await;
    page_analysis::enqueue_for_page(&pool, page_id, false)
        .await
        .unwrap();
    sqlx::query(
        "UPDATE crawler_jobs SET max_attempts = 1 WHERE payload->>'page_id' = $1",
    )
    .bind(page_id.to_string())
    .execute(&pool)
    .await
    .unwrap();

    let dispatcher = CountingDispatcher::panicking();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "dead").await;
    handle.shutdown().await;

    // The worker survived the panic and recorded a failed row.
    let row = page_analysis::load(&pool, page_id).await.unwrap().unwrap();
    assert_eq!(row.status, AnalysisStatus::Failed);
}

#[sqlx::test(migrations = "./migrations")]
async fn worker_ignores_non_analyze_jobs(pool: PgPool) {
    // A crawl job must be left untouched by the analysis worker.
    use mangalord::crawler::jobs::{self, JobPayload};
    let crawl = jobs::enqueue(
        &pool,
        &JobPayload::SyncManga {
            source_id: "s".into(),
            source_manga_key: "k".into(),
        },
    )
    .await
    .unwrap();
    let crawl_id = match crawl {
        jobs::EnqueueResult::Inserted(id) => id,
        jobs::EnqueueResult::Skipped => panic!("expected insert"),
    };

    let page_id = seed_page(&pool).await;
    page_analysis::enqueue_for_page(&pool, page_id, false)
        .await
        .unwrap();

    let dispatcher = CountingDispatcher::ok();
    let (handle, _c) = spawn_with(&pool, dispatcher.clone());
    wait_for_state(&pool, page_id, "done").await;
    handle.shutdown().await;

    let crawl_state: String =
        sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
            .bind(crawl_id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(crawl_state, "pending", "crawl job must be left for the crawler");
}