feat(admin): observability — job history, live "now analyzing", durations & metrics
Crawler + analysis admin dashboards gain a Live / History / Metrics segmented view. History: searchable, filterable, paginated job log per subsystem (crawler_jobs across all states/kinds; page_analysis terminal outcomes), with inline dead-job requeue and a Duration column. Live: enrich analysis SSE events with manga title + chapter number and add a sticky "Now analyzing" banner that jumps to and highlights the page. Metrics: new durable crawl_metrics table (migration 0028) timing every crawl op (manga list walk, manga detail, cover, whole chapter; per-page derived from chapter) plus page_analysis.duration_ms for analysis. New endpoints serve per-type average durations + success rates and a recent-ops log; a cron reaper (CRAWL_METRICS_RETENTION_DAYS) bounds growth. Tested: repo + API integration tests, vitest for the API client and fmtDuration, and Playwright for the History/Metrics tabs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -601,3 +601,255 @@ async fn coverage_requires_admin(pool: PgPool) {
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::FORBIDDEN);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Analysis history: terminal-outcome log, searchable + filterable.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Seed a manga → chapter → one page, then write a terminal `page_analysis`
|
||||
/// row for that page. Returns the page id.
|
||||
async fn seed_analyzed_page(
|
||||
pool: &PgPool,
|
||||
title: &str,
|
||||
status: &str,
|
||||
is_nsfw: bool,
|
||||
error: Option<&str>,
|
||||
) -> uuid::Uuid {
|
||||
let manga_id: uuid::Uuid =
|
||||
sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
|
||||
.bind(title)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let chapter_id: uuid::Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
|
||||
)
|
||||
.bind(manga_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let page_id: uuid::Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
|
||||
VALUES ($1, 1, 'k', 'image/png') RETURNING id",
|
||||
)
|
||||
.bind(chapter_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO page_analysis (page_id, status, is_nsfw, model, error, analyzed_at) \
|
||||
VALUES ($1, $2, $3, 'vis', $4, now())",
|
||||
)
|
||||
.bind(page_id)
|
||||
.bind(status)
|
||||
.bind(is_nsfw)
|
||||
.bind(error)
|
||||
.execute(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
page_id
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "./migrations")]
|
||||
async fn analysis_history_lists_filters_and_paginates(pool: PgPool) {
|
||||
seed_analyzed_page(&pool, "Alpha", "done", false, None).await;
|
||||
seed_analyzed_page(&pool, "Beta", "failed", false, Some("boom")).await;
|
||||
seed_analyzed_page(&pool, "Gamma", "done", true, None).await;
|
||||
// A still-pending row must NOT appear in history (terminal outcomes only).
|
||||
let pending_page = seed_analyzed_page(&pool, "Delta", "pending", false, None).await;
|
||||
let _ = pending_page;
|
||||
|
||||
let h = common::harness(pool.clone());
|
||||
let cookie = seed_admin(&pool, &h.app).await;
|
||||
|
||||
// Unfiltered: the three terminal rows, pending excluded.
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["page"]["total"], 3);
|
||||
|
||||
// Filter by status=failed.
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history?status=failed",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["page"]["total"], 1);
|
||||
assert_eq!(body["items"][0]["manga_title"], "Beta");
|
||||
assert_eq!(body["items"][0]["error"], "boom");
|
||||
|
||||
// NSFW-only.
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history?nsfw=true",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["page"]["total"], 1);
|
||||
assert_eq!(body["items"][0]["manga_title"], "Gamma");
|
||||
|
||||
// Search by manga title.
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history?search=Alpha",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["page"]["total"], 1);
|
||||
assert_eq!(body["items"][0]["manga_title"], "Alpha");
|
||||
assert_eq!(body["items"][0]["status"], "done");
|
||||
|
||||
// Pagination: limit clamps the slice; total reflects the full set.
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history?limit=1",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["items"].as_array().unwrap().len(), 1);
|
||||
assert_eq!(body["page"]["total"], 3);
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "./migrations")]
|
||||
async fn analysis_history_requires_admin(pool: PgPool) {
|
||||
let h = common::harness(pool.clone());
|
||||
let (_u, cookie) = common::register_user(&h.app).await;
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::FORBIDDEN);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Analysis metrics: durations + averages + by-model + success.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Seed a manga→chapter→page + a terminal page_analysis row carrying a
|
||||
/// duration and model. Returns the page id.
|
||||
async fn seed_timed_page(
|
||||
pool: &PgPool,
|
||||
title: &str,
|
||||
status: &str,
|
||||
model: &str,
|
||||
duration_ms: i64,
|
||||
) -> uuid::Uuid {
|
||||
let manga_id: uuid::Uuid =
|
||||
sqlx::query_scalar("INSERT INTO mangas (title) VALUES ($1) RETURNING id")
|
||||
.bind(title)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let chapter_id: uuid::Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
|
||||
)
|
||||
.bind(manga_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
let page_id: uuid::Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
|
||||
VALUES ($1, 1, 'k', 'image/png') RETURNING id",
|
||||
)
|
||||
.bind(chapter_id)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO page_analysis (page_id, status, model, duration_ms, analyzed_at) \
|
||||
VALUES ($1, $2, $3, $4, now())",
|
||||
)
|
||||
.bind(page_id)
|
||||
.bind(status)
|
||||
.bind(model)
|
||||
.bind(duration_ms)
|
||||
.execute(pool)
|
||||
.await
|
||||
.unwrap();
|
||||
page_id
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "./migrations")]
|
||||
async fn analysis_metrics_over_http(pool: PgPool) {
|
||||
seed_timed_page(&pool, "Alpha", "done", "qwen", 2000).await;
|
||||
seed_timed_page(&pool, "Beta", "done", "qwen", 4000).await;
|
||||
seed_timed_page(&pool, "Gamma", "done", "llava", 3000).await;
|
||||
seed_timed_page(&pool, "Delta", "failed", "qwen", 1000).await;
|
||||
|
||||
let h = common::harness(pool.clone());
|
||||
let cookie = seed_admin(&pool, &h.app).await;
|
||||
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/metrics",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["n"], 4);
|
||||
assert_eq!(body["ok"], 3);
|
||||
assert_eq!(body["failed"], 1);
|
||||
// Headline mean is done-only (2000 + 4000 + 3000)/3 — the failed row's
|
||||
// 1000ms is excluded so it reconciles with the by-model breakdown.
|
||||
assert_eq!(body["avg_ms"], 3000.0);
|
||||
|
||||
// by_model: qwen has the done rows 2000 + 4000 → avg 3000, n 2
|
||||
// (the failed qwen row is excluded — done only).
|
||||
let by_model = body["by_model"].as_array().unwrap();
|
||||
let qwen = by_model.iter().find(|m| m["model"] == "qwen").unwrap();
|
||||
assert_eq!(qwen["n"], 2);
|
||||
assert_eq!(qwen["avg_ms"], 3000.0);
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "./migrations")]
|
||||
async fn analysis_history_carries_duration(pool: PgPool) {
|
||||
seed_timed_page(&pool, "Alpha", "done", "qwen", 2500).await;
|
||||
let h = common::harness(pool.clone());
|
||||
let cookie = seed_admin(&pool, &h.app).await;
|
||||
|
||||
let resp = h
|
||||
.app
|
||||
.clone()
|
||||
.oneshot(common::get_with_cookie(
|
||||
"/api/v1/admin/analysis/history",
|
||||
&cookie,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let body = common::body_json(resp).await;
|
||||
assert_eq!(body["items"][0]["duration_ms"], 2500);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user