Mangalord/backend/tests/api_admin_crawler.rs

//! Integration tests for the admin crawler observability/control API.
//!
//! The default test harness wires `AppState.crawler = None` (no daemon),
//! so the *control* endpoints return 503 and the *read* endpoints that
//! work off the DB (status shell, dead-jobs list/requeue) still function.
//! This is exactly the production "daemon disabled" posture.

mod common;

use std::time::Duration;

use axum::http::StatusCode;
use axum::Router;
use http_body_util::BodyExt;
use serde_json::json;
use sqlx::PgPool;
use tower::ServiceExt;
use uuid::Uuid;

use common::{
    body_json, get, get_with_cookie, harness, harness_with_admin_origins,
    post_json_with_cookie, post_json_with_cookie_origin, register_user,
};

async fn seed_admin(pool: &PgPool, app: &Router) -> String {
    let (username, cookie) = register_user(app).await;
    let u = mangalord::repo::user::find_by_username(pool, &username)
        .await
        .unwrap()
        .unwrap();
    mangalord::repo::user::set_is_admin_unchecked(pool, u.id, true)
        .await
        .unwrap();
    cookie
}

async fn seed_dead_job(pool: &PgPool, title: &str) -> Uuid {
    let manga_id = Uuid::new_v4();
    let chapter_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
        .bind(manga_id)
        .bind(title)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
        .bind(chapter_id)
        .bind(manga_id)
        .execute(pool)
        .await
        .unwrap();
    let job_id = Uuid::new_v4();
    sqlx::query(
        "INSERT INTO crawler_jobs (id, payload, state, attempts, last_error) \
         VALUES ($1, $2, 'dead', 5, 'boom')",
    )
    .bind(job_id)
    .bind(json!({
        "kind": "sync_chapter_content",
        "source_id": "target",
        "chapter_id": chapter_id,
        "source_chapter_key": "k",
    }))
    .execute(pool)
    .await
    .unwrap();
    job_id
}

/// Seed a chapter-content job in a given state ('pending'/'running').
async fn seed_job(pool: &PgPool, title: &str, state: &str) {
    let manga_id = Uuid::new_v4();
    let chapter_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
        .bind(manga_id)
        .bind(title)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
        .bind(chapter_id)
        .bind(manga_id)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, $3)")
        .bind(Uuid::new_v4())
        .bind(json!({
            "kind": "sync_chapter_content",
            "source_id": "target",
            "chapter_id": chapter_id,
            "source_chapter_key": "k",
        }))
        .bind(state)
        .execute(pool)
        .await
        .unwrap();
}

/// Seed a manga with no cover + a live source row (queued for cover fetch).
async fn seed_missing_cover(pool: &PgPool, title: &str) {
    let manga_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title, cover_image_path) VALUES ($1, $2, NULL)")
        .bind(manga_id)
        .bind(title)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO sources (id, name, base_url) VALUES ('target','T','http://x') ON CONFLICT DO NOTHING")
        .execute(pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url) \
         VALUES ('target', $1, $2, 'http://x/m')",
    )
    .bind(format!("k-{manga_id}"))
    .bind(manga_id)
    .execute(pool)
    .await
    .unwrap();
}

#[sqlx::test(migrations = "./migrations")]
async fn active_jobs_and_covers_lists_over_http(pool: PgPool) {
    seed_job(&pool, "Naruto", "pending").await;
    seed_job(&pool, "Bleach", "running").await;
    seed_missing_cover(&pool, "One Piece").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    // Queued/active chapters.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/active-jobs", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 2);

    // Queued covers.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/covers", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    assert_eq!(body["items"][0]["manga_title"], "One Piece");

    // Both are admin-gated.
    let (_u, plain) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/active-jobs", &plain))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn get_status_requires_admin(pool: PgPool) {
    let h = harness(pool);
    // Unauthenticated → 401.
    let resp = h.app.clone().oneshot(get("/api/v1/admin/crawler")).await.unwrap();
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

    // Authenticated non-admin → 403.
    let (_u, cookie) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn get_status_reports_disabled_daemon_with_queue_counts(pool: PgPool) {
    seed_dead_job(&pool, "Naruto").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["daemon"], "disabled");
    assert_eq!(body["queue"]["dead"], 1);
    assert_eq!(body["browser"], "down");
}

#[sqlx::test(migrations = "./migrations")]
async fn control_endpoints_return_503_when_daemon_disabled(pool: PgPool) {
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    for uri in [
        "/api/v1/admin/crawler/run",
        "/api/v1/admin/crawler/reconcile",
        "/api/v1/admin/crawler/browser/restart",
        "/api/v1/admin/crawler/session/clear-expired",
    ] {
        let resp = h
            .app
            .clone()
            .oneshot(post_json_with_cookie(uri, json!({}), &cookie))
            .await
            .unwrap();
        assert_eq!(
            resp.status(),
            StatusCode::SERVICE_UNAVAILABLE,
            "{uri} should be 503 when daemon disabled"
        );
    }
}

#[sqlx::test(migrations = "./migrations")]
async fn status_stream_requires_admin(pool: PgPool) {
    let h = harness(pool);
    // Unauthenticated → 401.
    let resp = h
        .app
        .clone()
        .oneshot(get("/api/v1/admin/crawler/stream"))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    // Non-admin → 403.
    let (_u, cookie) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/stream", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn status_stream_emits_initial_event(pool: PgPool) {
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/stream", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let ct = resp
        .headers()
        .get(axum::http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or_default()
        .to_string();
    assert!(ct.starts_with("text/event-stream"), "content-type was {ct:?}");

    // Accumulate frames (the immediate snapshot may arrive split across
    // frames) until the status payload appears, with an overall timeout so
    // the never-ending stream can't hang the test.
    let mut body = resp.into_body();
    let mut acc = String::new();
    let deadline = tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let Some(frame) = body.frame().await else { break };
            if let Ok(data) = frame.expect("frame ok").into_data() {
                acc.push_str(&String::from_utf8_lossy(&data));
                if acc.contains("\"daemon\"") {
                    break;
                }
            }
        }
    })
    .await;
    assert!(deadline.is_ok(), "did not receive status within 5s; got: {acc:?}");
    assert!(acc.contains("\"daemon\""), "missing status payload: {acc}");
    assert!(acc.contains("status"), "missing SSE event name: {acc}");
}

#[sqlx::test(migrations = "./migrations")]
async fn mutating_endpoints_reject_non_admin(pool: PgPool) {
    let h = harness(pool);
    // A logged-in non-admin must be forbidden from a mutating endpoint.
    let (_u, cookie) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "all" }),
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

// ---------------------------------------------------------------------------
// CSRF Origin/Referer allowlist (T2)
// ---------------------------------------------------------------------------

#[sqlx::test(migrations = "./migrations")]
async fn csrf_rejects_mutation_with_cross_origin_header(pool: PgPool) {
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie_origin(
            "/api/v1/admin/crawler/session/clear-expired",
            json!({}),
            &cookie,
            Some("https://evil.example.com"),
            None,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_allows_mutation_with_allowed_origin(pool: PgPool) {
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie_origin(
            "/api/v1/admin/crawler/session/clear-expired",
            json!({}),
            &cookie,
            Some("http://localhost:3000"),
            None,
        ))
        .await
        .unwrap();
    // Daemon disabled in the harness → 503 from the handler; the CSRF
    // gate must have let the request through before that response.
    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_rejects_cookie_auth_without_origin_or_referer(pool: PgPool) {
    // Cookie-auth + neither Origin nor Referer was previously waved through
    // ("curl bypass"). It's now refused: an extension or no-referrer-policy
    // page can produce exactly this shape and would otherwise be a CSRF
    // vector. Real curl/script callers should authenticate with a bearer
    // token instead — see `csrf_bearer_only_request_bypasses_gate`.
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie_origin(
            "/api/v1/admin/crawler/session/clear-expired",
            json!({}),
            &cookie,
            None,
            None,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_bearer_only_request_bypasses_gate(pool: PgPool) {
    // Bearer-only (no session cookie, no Origin) bypasses the CSRF gate
    // because a browser can't set `Authorization` on a cross-site POST.
    // RequireAdmin then rejects bot tokens for admin endpoints — we
    // expect 401, not the CSRF 403. The path proves the bypass branch
    // is reachable for legitimate bot callers.
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let req = axum::http::Request::builder()
        .method("POST")
        .uri("/api/v1/admin/crawler/session/clear-expired")
        .header(axum::http::header::CONTENT_TYPE, "application/json")
        .header(axum::http::header::AUTHORIZATION, "Bearer fake-bot-token")
        .body(axum::body::Body::from("{}"))
        .unwrap();
    let resp = h.app.clone().oneshot(req).await.unwrap();
    // 401 from RequireAdmin (bad/non-admin token), NOT 403 from CSRF.
    // If the CSRF gate had fired, we'd get FORBIDDEN.
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_bearer_plus_cookie_still_enforces_gate(pool: PgPool) {
    // The cookie-ride vector: an attacker page mints `Authorization:
    // Bearer junk` on a cross-site credentialed POST. The cookie is
    // attached automatically by the browser; the bearer header alone
    // used to bypass CSRF, then `CurrentUser` authenticated the victim
    // via the cookie. Cookie precedence here means: as soon as a
    // session cookie is present, the CSRF gate fires regardless of
    // Authorization — the bearer can't whitelist the cookie ride.
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    let req = axum::http::Request::builder()
        .method("POST")
        .uri("/api/v1/admin/crawler/session/clear-expired")
        .header(axum::http::header::CONTENT_TYPE, "application/json")
        .header(axum::http::header::COOKIE, cookie)
        .header(axum::http::header::AUTHORIZATION, "Bearer junk")
        .header(axum::http::header::ORIGIN, "https://evil.example.com")
        .body(axum::body::Body::from("{}"))
        .unwrap();
    let resp = h.app.clone().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_no_auth_falls_through_to_401(pool: PgPool) {
    // No cookie, no bearer — there's no authority to ride, so the CSRF
    // gate yields and lets `RequireAdmin` return the more meaningful 401.
    // Pins the carve-out that keeps anonymous curl-against-admin from
    // 403-via-CSRF (which would mislead operators about what's blocking).
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let req = axum::http::Request::builder()
        .method("POST")
        .uri("/api/v1/admin/crawler/session/clear-expired")
        .header(axum::http::header::CONTENT_TYPE, "application/json")
        .body(axum::body::Body::from("{}"))
        .unwrap();
    let resp = h.app.clone().oneshot(req).await.unwrap();
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}


#[sqlx::test(migrations = "./migrations")]
async fn csrf_falls_back_to_referer_when_origin_missing(pool: PgPool) {
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie_origin(
            "/api/v1/admin/crawler/session/clear-expired",
            json!({}),
            &cookie,
            None,
            Some("http://localhost:3000/admin/crawler"),
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_skipped_on_safe_methods(pool: PgPool) {
    let h = harness_with_admin_origins(
        pool.clone(),
        vec!["http://localhost:3000".to_string()],
    );
    let cookie = seed_admin(&pool, &h.app).await;
    // GET from a hostile origin is fine — browsers can't mutate via GET.
    let resp = h
        .app
        .clone()
        .oneshot(common::get_with_cookie_origin(
            "/api/v1/admin/crawler",
            &cookie,
            Some("https://evil.example.com"),
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
}

#[sqlx::test(migrations = "./migrations")]
async fn csrf_rejects_cookie_auth_when_allowlist_empty(pool: PgPool) {
    // Previously an empty allowlist silently skipped the CSRF check, so an
    // operator who forgot to set ADMIN_ALLOWED_ORIGINS shipped an unguarded
    // admin surface. We now fail-closed for cookie-auth admin mutations
    // when the allowlist is empty — operators must either set the env var
    // (browser deploy) or authenticate with `Authorization: Bearer …`.
    let h = harness_with_admin_origins(pool.clone(), vec![]);
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie_origin(
            "/api/v1/admin/crawler/session/clear-expired",
            json!({}),
            &cookie,
            Some("https://evil.example.com"),
            None,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

// ---------------------------------------------------------------------------
// Cache-Control: no-store on admin responses (S3)
// ---------------------------------------------------------------------------

#[sqlx::test(migrations = "./migrations")]
async fn admin_responses_have_no_store(pool: PgPool) {
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let cc = resp
        .headers()
        .get(axum::http::header::CACHE_CONTROL)
        .and_then(|v| v.to_str().ok())
        .unwrap_or_default()
        .to_string();
    assert!(
        cc.contains("no-store"),
        "admin response missing Cache-Control: no-store (got {cc:?})"
    );
}

#[sqlx::test(migrations = "./migrations")]
async fn non_admin_responses_unaffected_by_no_store(pool: PgPool) {
    let h = harness(pool);
    let resp = h
        .app
        .clone()
        .oneshot(get("/api/v1/health"))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let cc = resp
        .headers()
        .get(axum::http::header::CACHE_CONTROL)
        .and_then(|v| v.to_str().ok());
    assert!(
        cc.is_none() || !cc.unwrap_or_default().contains("no-store"),
        "non-admin response should not get no-store (got {cc:?})"
    );
}

// ---------------------------------------------------------------------------
// scope=all confirm:true guard (S1)
// ---------------------------------------------------------------------------

#[sqlx::test(migrations = "./migrations")]
async fn requeue_all_without_confirm_rejected(pool: PgPool) {
    seed_dead_job(&pool, "X").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "all" }),
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    // The dead row must NOT have been touched.
    let state: String = sqlx::query_scalar("SELECT state FROM crawler_jobs LIMIT 1")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(state, "dead");
}

#[sqlx::test(migrations = "./migrations")]
async fn requeue_all_with_confirm_flips_dead_pile(pool: PgPool) {
    seed_dead_job(&pool, "X").await;
    seed_dead_job(&pool, "Y").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "all", "confirm": true }),
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let pending: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM crawler_jobs WHERE state = 'pending'",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(pending, 2);
}

// ---------------------------------------------------------------------------
// admin_audit target_id + PHPSESSID fingerprint (S2)
// ---------------------------------------------------------------------------

#[sqlx::test(migrations = "./migrations")]
async fn requeue_chapter_audit_records_chapter_target_id(pool: PgPool) {
    let job_id = seed_dead_job(&pool, "Bleach").await;
    let chapter_id: Uuid = sqlx::query_scalar(
        "SELECT (payload->>'chapter_id')::uuid FROM crawler_jobs WHERE id = $1",
    )
    .bind(job_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "chapter", "chapter_id": chapter_id }),
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let (target_kind, target_id): (String, Option<Uuid>) = sqlx::query_as(
        "SELECT target_kind, target_id FROM admin_audit \
         WHERE action = 'crawler_dead_jobs_requeue' \
         ORDER BY at DESC LIMIT 1",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(target_kind, "chapter");
    assert_eq!(target_id, Some(chapter_id));
}

#[sqlx::test(migrations = "./migrations")]
async fn requeue_all_audit_omits_target_id_but_logs_count(pool: PgPool) {
    seed_dead_job(&pool, "X").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;
    let _ = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "all", "confirm": true }),
            &cookie,
        ))
        .await
        .unwrap();
    let (target_kind, target_id, payload): (String, Option<Uuid>, serde_json::Value) =
        sqlx::query_as(
            "SELECT target_kind, target_id, payload FROM admin_audit \
             WHERE action = 'crawler_dead_jobs_requeue' \
             ORDER BY at DESC LIMIT 1",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(target_kind, "crawler");
    assert_eq!(target_id, None);
    assert_eq!(payload["scope"], "all");
    assert_eq!(payload["requeued"], 1);
}

// ---------------------------------------------------------------------------
// Job history: unified, searchable, filterable list over all states/kinds.
// ---------------------------------------------------------------------------

/// Seed a chapter-content job in a given state and return its (manga, chapter).
async fn seed_history_chapter_job(
    pool: &PgPool,
    title: &str,
    number: i32,
    state: &str,
) -> (Uuid, Uuid) {
    let manga_id = Uuid::new_v4();
    let chapter_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
        .bind(manga_id)
        .bind(title)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, $3)")
        .bind(chapter_id)
        .bind(manga_id)
        .bind(number)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, $3)")
        .bind(Uuid::new_v4())
        .bind(json!({
            "kind": "sync_chapter_content",
            "source_id": "target",
            "chapter_id": chapter_id,
            "source_chapter_key": "k",
        }))
        .bind(state)
        .execute(pool)
        .await
        .unwrap();
    (manga_id, chapter_id)
}

/// Seed a `done` analyze_page job pointing at a real page so the history
/// row resolves manga/chapter/page-number through the page breadcrumb.
async fn seed_history_analyze_job(pool: &PgPool, title: &str, page_number: i32) {
    let manga_id = Uuid::new_v4();
    let chapter_id = Uuid::new_v4();
    let page_id = Uuid::new_v4();
    sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)")
        .bind(manga_id)
        .bind(title)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 3)")
        .bind(chapter_id)
        .bind(manga_id)
        .execute(pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO pages (id, chapter_id, page_number, storage_key, content_type) \
         VALUES ($1, $2, $3, 'k', 'image/png')",
    )
    .bind(page_id)
    .bind(chapter_id)
    .bind(page_number)
    .execute(pool)
    .await
    .unwrap();
    sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, 'done')")
        .bind(Uuid::new_v4())
        .bind(json!({ "kind": "analyze_page", "page_id": page_id, "force": false }))
        .execute(pool)
        .await
        .unwrap();
}

#[sqlx::test(migrations = "./migrations")]
async fn job_history_lists_filters_and_paginates(pool: PgPool) {
    seed_history_chapter_job(&pool, "Naruto", 1, "done").await;
    seed_history_chapter_job(&pool, "Bleach", 2, "running").await;
    seed_dead_job(&pool, "Vinland").await; // dead sync_chapter_content
    seed_history_analyze_job(&pool, "One Piece", 7).await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    // Unfiltered: all four jobs.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/history", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 4);

    // Filter by state.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?state=done",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    // Naruto (sync_chapter done) + One Piece (analyze_page done).
    assert_eq!(body["page"]["total"], 2);

    // Filter by kind.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?kind=analyze_page",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    assert_eq!(body["items"][0]["manga_title"], "One Piece");
    assert_eq!(body["items"][0]["page_number"], 7);
    assert_eq!(body["items"][0]["chapter_number"], 3);

    // Search by manga title.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?search=Bleach",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    assert_eq!(body["items"][0]["manga_title"], "Bleach");
    assert_eq!(body["items"][0]["state"], "running");

    // Pagination: limit clamps the page slice but total reflects the full set.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?limit=1",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    assert_eq!(body["items"].as_array().unwrap().len(), 1);
    assert_eq!(body["page"]["total"], 4);

    // Admin-gated.
    let (_u, plain) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/history", &plain))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn job_history_surfaces_sync_manga_payload_title(pool: PgPool) {
    // A reconcile-enqueued sync_manga job that died has no manga row, so the
    // history row must fall back to the payload title/url and be searchable.
    sqlx::query("INSERT INTO crawler_jobs (id, payload, state, last_error) VALUES ($1, $2, 'dead', 'broken-page body signature')")
        .bind(Uuid::new_v4())
        .bind(json!({
            "kind": "sync_manga",
            "source_id": "target",
            "source_manga_key": "gone-1",
            "url": "http://x/manga/gone-1",
            "title": "Ghost Manga",
        }))
        .execute(&pool)
        .await
        .unwrap();
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?search=ghost",
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    let row = &body["items"][0];
    assert_eq!(row["kind"], "sync_manga");
    assert_eq!(row["manga_title"], serde_json::Value::Null);
    assert_eq!(row["payload_title"], "Ghost Manga");
    assert_eq!(row["source_url"], "http://x/manga/gone-1");
    assert_eq!(row["source_key"], "gone-1");
}

// ---------------------------------------------------------------------------
// Operation metrics: durations, averages, recent-ops log.
// ---------------------------------------------------------------------------

async fn record_metric(
    pool: &PgPool,
    op: &str,
    manga_id: Option<Uuid>,
    chapter_id: Option<Uuid>,
    outcome: &str,
    duration_ms: i64,
    items: Option<i32>,
) {
    mangalord::repo::crawl_metrics::record(
        pool, op, manga_id, chapter_id, outcome, duration_ms, items, None,
    )
    .await
    .unwrap();
}

#[sqlx::test(migrations = "./migrations")]
async fn crawler_metrics_summary_and_ops_over_http(pool: PgPool) {
    // Two chapter ops (one failed) + a cover op.
    record_metric(&pool, "chapter", None, None, "ok", 6000, Some(20)).await;
    record_metric(&pool, "chapter", None, None, "failed", 8000, Some(10)).await;
    record_metric(&pool, "manga_cover", None, None, "ok", 500, None).await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    // Summary: per-type averages + success split.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/metrics", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    let summary = body["summary"].as_array().unwrap();
    let chapter = summary.iter().find(|s| s["op"] == "chapter").unwrap();
    assert_eq!(chapter["n"], 2);
    assert_eq!(chapter["ok"], 1);
    assert_eq!(chapter["failed"], 1);
    assert_eq!(chapter["avg_ms"], 7000.0);

    // Recent-ops log + outcome filter.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/metrics/ops?outcome=failed",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    assert_eq!(body["items"][0]["op"], "chapter");
    assert_eq!(body["items"][0]["outcome"], "failed");

    // Admin-gated.
    let (_u, plain) = register_user(&h.app).await;
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/metrics", &plain))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
}

#[sqlx::test(migrations = "./migrations")]
async fn job_history_includes_chapter_duration(pool: PgPool) {
    let (_m, chapter_id) = seed_history_chapter_job(&pool, "Berserk", 12, "done").await;
    // A chapter metric for that chapter → history row should surface its duration.
    record_metric(&pool, "chapter", None, Some(chapter_id), "ok", 6100, Some(20)).await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie(
            "/api/v1/admin/crawler/history?kind=sync_chapter_content",
            &cookie,
        ))
        .await
        .unwrap();
    let body = body_json(resp).await;
    assert_eq!(body["items"][0]["duration_ms"], 6100);
}

#[sqlx::test(migrations = "./migrations")]
async fn dead_jobs_list_and_requeue_over_http(pool: PgPool) {
    let job_id = seed_dead_job(&pool, "Bleach").await;
    let h = harness(pool.clone());
    let cookie = seed_admin(&pool, &h.app).await;

    // List.
    let resp = h
        .app
        .clone()
        .oneshot(get_with_cookie("/api/v1/admin/crawler/dead-jobs", &cookie))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["page"]["total"], 1);
    assert_eq!(body["items"][0]["manga_title"], "Bleach");

    // Requeue the single job.
    let resp = h
        .app
        .clone()
        .oneshot(post_json_with_cookie(
            "/api/v1/admin/crawler/dead-jobs/requeue",
            json!({ "scope": "job", "job_id": job_id }),
            &cookie,
        ))
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_json(resp).await;
    assert_eq!(body["requeued"], 1);

    let state: String = sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1")
        .bind(job_id)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(state, "pending");
}