Mangalord/backend/tests/crawler_reconcile.rs

//! Integration tests for the reconcile diff queries:
//! `existing_source_keys` (dropped rows count as present) and
//! `sync_manga_keys_with_blocking_job` (pending/running/dead block).
//!
//! The browser-driven full walk in `reconcile::reconcile_missing` needs a
//! real `chromiumoxide::Browser`, so it is covered by the manual/E2E path;
//! the pure set-diff (`select_missing`) is unit-tested in `crawler::reconcile`.

use mangalord::crawler::jobs::{self, JobPayload};
use mangalord::repo::crawler;
use sqlx::PgPool;
use uuid::Uuid;

async fn ensure_target(pool: &PgPool) {
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ('target','T','http://x') \
         ON CONFLICT DO NOTHING",
    )
    .execute(pool)
    .await
    .unwrap();
}

/// Insert a `manga_sources` row (with its backing manga) under `target`.
async fn seed_source_manga(pool: &PgPool, key: &str, dropped: bool) {
    let manga_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
        .bind(manga_id)
        .bind(format!("M {key}"))
        .execute(pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url, dropped_at) \
         VALUES ('target', $1, $2, $3, CASE WHEN $4 THEN now() ELSE NULL END)",
    )
    .bind(key)
    .bind(manga_id)
    .bind(format!("http://x/{key}"))
    .bind(dropped)
    .execute(pool)
    .await
    .unwrap();
}

async fn enqueue_sync_manga(pool: &PgPool, key: &str) -> Uuid {
    let res = jobs::enqueue(
        pool,
        &JobPayload::SyncManga {
            source_id: "target".into(),
            source_manga_key: key.into(),
            url: format!("http://x/{key}"),
            title: format!("M {key}"),
        },
    )
    .await
    .unwrap();
    match res {
        jobs::EnqueueResult::Inserted(id) => id,
        jobs::EnqueueResult::Skipped => panic!("expected insert"),
    }
}

async fn set_job_state(pool: &PgPool, id: Uuid, state: &str) {
    sqlx::query("UPDATE crawler_jobs SET state = $2::text WHERE id = $1")
        .bind(id)
        .bind(state)
        .execute(pool)
        .await
        .unwrap();
}

#[sqlx::test(migrations = "./migrations")]
async fn existing_source_keys_includes_dropped_rows(pool: PgPool) {
    ensure_target(&pool).await;
    seed_source_manga(&pool, "live", false).await;
    seed_source_manga(&pool, "dropped", true).await;

    let keys = crawler::existing_source_keys(&pool, "target").await.unwrap();
    assert!(keys.contains("live"));
    assert!(
        keys.contains("dropped"),
        "a dropped manga_sources row must still count as present"
    );
    assert_eq!(keys.len(), 2);
}

#[sqlx::test(migrations = "./migrations")]
async fn sync_manga_blocking_job_matches_pending_running_dead_only(pool: PgPool) {
    ensure_target(&pool).await;
    // One key per state.
    enqueue_sync_manga(&pool, "pending").await; // stays pending
    let running = enqueue_sync_manga(&pool, "running").await;
    let dead = enqueue_sync_manga(&pool, "dead").await;
    let done = enqueue_sync_manga(&pool, "done").await;
    let failed = enqueue_sync_manga(&pool, "failed").await;
    set_job_state(&pool, running, "running").await;
    set_job_state(&pool, dead, "dead").await;
    set_job_state(&pool, done, "done").await;
    set_job_state(&pool, failed, "failed").await;

    let blocked = crawler::sync_manga_keys_with_blocking_job(&pool, "target")
        .await
        .unwrap();
    assert!(blocked.contains("pending"));
    assert!(blocked.contains("running"));
    assert!(blocked.contains("dead"));
    assert!(!blocked.contains("done"), "done must not block re-enqueue");
    assert!(
        !blocked.contains("failed"),
        "failed (mid-backoff) must not block re-enqueue"
    );
    assert_eq!(blocked.len(), 3);
}

#[sqlx::test(migrations = "./migrations")]
async fn sync_manga_blocking_job_is_per_source(pool: PgPool) {
    ensure_target(&pool).await;
    sqlx::query(
        "INSERT INTO sources (id, name, base_url) VALUES ('other','O','http://y') \
         ON CONFLICT DO NOTHING",
    )
    .execute(&pool)
    .await
    .unwrap();
    enqueue_sync_manga(&pool, "shared").await;
    jobs::enqueue(
        &pool,
        &JobPayload::SyncManga {
            source_id: "other".into(),
            source_manga_key: "elsewhere".into(),
            url: "http://y/elsewhere".into(),
            title: "Elsewhere".into(),
        },
    )
    .await
    .unwrap();

    let blocked = crawler::sync_manga_keys_with_blocking_job(&pool, "target")
        .await
        .unwrap();
    assert!(blocked.contains("shared"));
    assert!(!blocked.contains("elsewhere"));
    assert_eq!(blocked.len(), 1);
}