The interleaved metadata pass misses mangas to list drift (a title slips a pagination slot during the slow detail walk). Add a reconcile pass: a cheap, full, list-only walk (refs only, no detail visit, no early stop) that set-diffs the walked keys against manga_sources and enqueues the strictly- missing ones as SyncManga jobs. A set-diff is immune to drift — a manga only has to appear somewhere in the list. - Build the previously-dead SyncManga worker by extending RealChapterDispatcher to run the shared pipeline::process_manga_ref (fetch → upsert → cover → chapters), refactored out of run_metadata_pass so both paths stay in lockstep. The crawl worker now leases both sync_chapter_content and sync_manga (jobs::lease_kinds); both serialize on the single exclusive browser. - SyncManga payload carries url + title so the worker can rebuild the ref and the dead-jobs/history UI can label a missing manga that has no manga row yet. - "Missing" = strict NOT EXISTS (dropped rows count as present). Re-enqueue is skipped when a pending/running/dead SyncManga job already exists, so a gone manga (detail 404 → retries → dead) is left dead and not retried. - New POST /v1/admin/crawler/reconcile (fire-and-forget, shares manual_pass_lock) + "Reconcile missing" admin button + Reconciling status phase streamed over SSE. - dead-jobs/history queries surface payload title/url/key; tables fall back to them for sync_manga rows; both searches match the payload title. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
954 lines
30 KiB
Rust
954 lines
30 KiB
Rust
//! 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_allows_mutation_without_origin_or_referer(pool: PgPool) {
|
|
// curl/server-to-server callers send neither — they can't be a CSRF
|
|
// vector since there's no third-party browser context.
|
|
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(
|
|
"/api/v1/admin/crawler/session/clear-expired",
|
|
json!({}),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
|
}
|
|
|
|
#[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_disabled_when_allowlist_empty(pool: PgPool) {
|
|
// Default harness has admin_allowed_origins = empty → operator
|
|
// opt-out documented in .env.example. A cross-origin POST passes
|
|
// through to the handler.
|
|
let h = harness(pool.clone());
|
|
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::SERVICE_UNAVAILABLE);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// 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");
|
|
}
|