feat(admin): observability — job history, live now-analyzing, durations & metrics (0.84.0) (#6)

backend/src/repo/crawl_metrics.rs

@@ -0,0 +1,158 @@
+//! DB access for the `crawl_metrics` timing log (migration 0028).
+//!
+//! `record` is the best-effort write called from each crawl operation;
+//! `summary` and `list_ops` are the admin Metrics-tab reads. Mirrors the
+//! plain-fn + `query_as` style used across `repo`.
+
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::domain::crawl_metrics::{OpRow, OpSummary};
+
+/// Operation kinds, kept as consts so call sites and the CHECK constraint
+/// don't drift from a free-typed literal.
+pub const OP_MANGA_LIST: &str = "manga_list";
+pub const OP_MANGA_DETAIL: &str = "manga_detail";
+pub const OP_MANGA_COVER: &str = "manga_cover";
+pub const OP_CHAPTER: &str = "chapter";
+
+/// Insert one completed-operation timing row. Best-effort: callers log and
+/// continue on error so a metrics-write failure never fails the actual crawl
+/// work. `outcome` is `"ok"` or `"failed"`.
+#[allow(clippy::too_many_arguments)]
+pub async fn record(
+    pool: &PgPool,
+    op: &str,
+    manga_id: Option<Uuid>,
+    chapter_id: Option<Uuid>,
+    outcome: &str,
+    duration_ms: i64,
+    items: Option<i32>,
+    error: Option<&str>,
+) -> sqlx::Result<()> {
+    sqlx::query(
+        "INSERT INTO crawl_metrics \
+            (op, manga_id, chapter_id, outcome, duration_ms, items, error) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7)",
+    )
+    .bind(op)
+    .bind(manga_id)
+    .bind(chapter_id)
+    .bind(outcome)
+    .bind(duration_ms)
+    .bind(items)
+    .bind(error)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Per-op average roll-up over an optional time window (`since = None` → all
+/// time). One row per `op` that has any metric in the window, with mean
+/// duration, counts, success split, and mean `items`.
+pub async fn summary(
+    pool: &PgPool,
+    since: Option<DateTime<Utc>>,
+) -> sqlx::Result<Vec<OpSummary>> {
+    sqlx::query_as::<_, OpSummary>(
+        r#"
+        SELECT
+            op,
+            AVG(duration_ms)::float8                              AS avg_ms,
+            COUNT(*)                                              AS n,
+            COUNT(*) FILTER (WHERE outcome = 'ok')                AS ok,
+            COUNT(*) FILTER (WHERE outcome = 'failed')            AS failed,
+            AVG(items)::float8                                    AS avg_items
+          FROM crawl_metrics
+         WHERE ($1::timestamptz IS NULL OR finished_at >= $1)
+         GROUP BY op
+         ORDER BY op
+        "#,
+    )
+    .bind(since)
+    .fetch_all(pool)
+    .await
+}
+
+/// Filters for [`list_ops`]. `None`/empty widens the scope.
+#[derive(Debug, Default, Clone)]
+pub struct OpFilter<'a> {
+    pub op: Option<&'a str>,
+    pub outcome: Option<&'a str>,
+    pub since: Option<DateTime<Utc>>,
+}
+
+/// Paginated, newest-first recent-operations log resolved to manga/chapter
+/// labels. Returns the page slice plus the filtered total.
+pub async fn list_ops(
+    pool: &PgPool,
+    filter: OpFilter<'_>,
+    limit: i64,
+    offset: i64,
+) -> sqlx::Result<(Vec<OpRow>, i64)> {
+    let items = sqlx::query_as::<_, OpRow>(
+        r#"
+        SELECT
+            cm.id,
+            cm.op,
+            cm.manga_id,
+            m.title           AS manga_title,
+            cm.chapter_id,
+            c.number          AS chapter_number,
+            cm.outcome,
+            cm.duration_ms,
+            cm.items,
+            cm.error,
+            cm.finished_at
+          FROM crawl_metrics cm
+          LEFT JOIN mangas m   ON m.id = cm.manga_id
+          LEFT JOIN chapters c ON c.id = cm.chapter_id
+         WHERE ($1::text IS NULL OR cm.op = $1)
+           AND ($2::text IS NULL OR cm.outcome = $2)
+           AND ($3::timestamptz IS NULL OR cm.finished_at >= $3)
+         ORDER BY cm.finished_at DESC
+         LIMIT $4 OFFSET $5
+        "#,
+    )
+    .bind(filter.op)
+    .bind(filter.outcome)
+    .bind(filter.since)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await?;
+
+    let (total,): (i64,) = sqlx::query_as(
+        r#"
+        SELECT COUNT(*)
+          FROM crawl_metrics cm
+         WHERE ($1::text IS NULL OR cm.op = $1)
+           AND ($2::text IS NULL OR cm.outcome = $2)
+           AND ($3::timestamptz IS NULL OR cm.finished_at >= $3)
+        "#,
+    )
+    .bind(filter.op)
+    .bind(filter.outcome)
+    .bind(filter.since)
+    .fetch_one(pool)
+    .await?;
+
+    Ok((items, total))
+}
+
+/// Delete metric rows older than `retention_days`. `0` disables the reaper
+/// (returns 0 without touching the table). Mirrors `jobs::reap_done`.
+pub async fn reap(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
+    if retention_days == 0 {
+        return Ok(0);
+    }
+    let result = sqlx::query(
+        "DELETE FROM crawl_metrics \
+          WHERE finished_at < now() - ($1::bigint || ' days')::interval",
+    )
+    .bind(retention_days as i64)
+    .execute(pool)
+    .await?;
+    Ok(result.rows_affected())
+}

backend/src/repo/crawler.rs

@@ -774,6 +774,139 @@ pub async fn list_active_jobs(
     Ok((items, total))
 }
 
+// ---------------------------------------------------------------------------
+// Job history: unified, searchable, filterable view over the queue table.
+// ---------------------------------------------------------------------------
+
+/// A `crawler_jobs` row resolved to human context for the admin history
+/// table. Works across all job kinds: the target columns are best-effort
+/// `Option`s resolved through whichever payload reference the kind carries
+/// (`chapter_id`, `manga_id`, or `page_id` via the page breadcrumb), so a
+/// `sync_manga` job (which has neither yet) simply leaves them `None` and
+/// falls back to `source_key`.
+#[derive(Debug, Clone, Serialize, FromRow)]
+pub struct JobHistoryRow {
+    pub id: Uuid,
+    /// `pending` | `running` | `done` | `dead` (the live queue states).
+    pub state: String,
+    /// `sync_manga` | `sync_chapter_list` | `sync_chapter_content` | `analyze_page`.
+    pub kind: Option<String>,
+    pub manga_id: Option<Uuid>,
+    pub manga_title: Option<String>,
+    pub chapter_id: Option<Uuid>,
+    pub chapter_number: Option<i32>,
+    /// Set only for `analyze_page` (resolved through the page breadcrumb).
+    pub page_number: Option<i32>,
+    /// Source-side key, the only target a `sync_manga` job carries.
+    pub source_key: Option<String>,
+    pub attempts: i32,
+    pub max_attempts: i32,
+    pub last_error: Option<String>,
+    pub updated_at: DateTime<Utc>,
+    /// How long the job's work took, when a timing was recorded: the latest
+    /// `chapter` crawl-metric for a chapter job, else the page's analysis
+    /// duration for an `analyze_page` job. `None` for kinds we don't time or
+    /// jobs not yet completed.
+    pub duration_ms: Option<i64>,
+}
+
+/// Filters for [`list_job_history`]. All optional; `None` widens the scope.
+#[derive(Debug, Default, Clone)]
+pub struct JobHistoryFilter<'a> {
+    /// Exact queue state (`done`, `dead`, `running`, `pending`).
+    pub state: Option<&'a str>,
+    /// Exact payload `kind`.
+    pub kind: Option<&'a str>,
+    /// Case-insensitive manga-title substring.
+    pub search: Option<&'a str>,
+}
+
+/// Paginated, newest-first view of the job queue across every state and
+/// kind — the searchable/filterable history surface. Joins each job to its
+/// manga/chapter/page context (best-effort) so the table can label rows.
+/// Returns the page slice plus the filtered total for pagination.
+///
+/// History depth is bounded by the done-job reaper (`reap_done`): completed
+/// jobs older than the retention window are gone. Terminal `dead` jobs
+/// persist until requeued.
+pub async fn list_job_history(
+    pool: &PgPool,
+    filter: JobHistoryFilter<'_>,
+    limit: i64,
+    offset: i64,
+) -> sqlx::Result<(Vec<JobHistoryRow>, i64)> {
+    let search_pat = filter
+        .search
+        .map(|s| format!("%{}%", s.trim()))
+        .filter(|p| p.len() > 2);
+
+    // The same FROM/JOIN/WHERE drives both the page slice and the count, so
+    // they stay in lockstep. `pg` resolves analyze_page → page → chapter;
+    // `COALESCE` lets one set of joins serve every kind.
+    let items: Vec<JobHistoryRow> = sqlx::query_as(
+        r#"
+        SELECT
+            cj.id,
+            cj.state,
+            cj.payload->>'kind'                                       AS kind,
+            COALESCE(c.manga_id, (cj.payload->>'manga_id')::uuid)     AS manga_id,
+            m.title                                                   AS manga_title,
+            COALESCE((cj.payload->>'chapter_id')::uuid, pg.chapter_id) AS chapter_id,
+            c.number                                                  AS chapter_number,
+            pg.page_number                                            AS page_number,
+            cj.payload->>'source_manga_key'                           AS source_key,
+            cj.attempts,
+            cj.max_attempts,
+            cj.last_error,
+            cj.updated_at,
+            COALESCE(cm.duration_ms, pa.duration_ms)                  AS duration_ms
+          FROM crawler_jobs cj
+          LEFT JOIN pages pg    ON pg.id = (cj.payload->>'page_id')::uuid
+          LEFT JOIN chapters c  ON c.id = COALESCE((cj.payload->>'chapter_id')::uuid, pg.chapter_id)
+          LEFT JOIN mangas m    ON m.id = COALESCE(c.manga_id, (cj.payload->>'manga_id')::uuid)
+          LEFT JOIN page_analysis pa ON pa.page_id = (cj.payload->>'page_id')::uuid
+          LEFT JOIN LATERAL (
+              SELECT duration_ms FROM crawl_metrics
+               WHERE op = 'chapter'
+                 AND chapter_id = (cj.payload->>'chapter_id')::uuid
+               ORDER BY finished_at DESC LIMIT 1
+          ) cm ON true
+         WHERE ($1::text IS NULL OR cj.state = $1)
+           AND ($2::text IS NULL OR cj.payload->>'kind' = $2)
+           AND ($3::text IS NULL OR m.title ILIKE $3)
+         ORDER BY cj.updated_at DESC
+         LIMIT $4 OFFSET $5
+        "#,
+    )
+    .bind(filter.state)
+    .bind(filter.kind)
+    .bind(&search_pat)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await?;
+
+    let total: i64 = sqlx::query_scalar(
+        r#"
+        SELECT COUNT(*)
+          FROM crawler_jobs cj
+          LEFT JOIN pages pg    ON pg.id = (cj.payload->>'page_id')::uuid
+          LEFT JOIN chapters c  ON c.id = COALESCE((cj.payload->>'chapter_id')::uuid, pg.chapter_id)
+          LEFT JOIN mangas m    ON m.id = COALESCE(c.manga_id, (cj.payload->>'manga_id')::uuid)
+         WHERE ($1::text IS NULL OR cj.state = $1)
+           AND ($2::text IS NULL OR cj.payload->>'kind' = $2)
+           AND ($3::text IS NULL OR m.title ILIKE $3)
+        "#,
+    )
+    .bind(filter.state)
+    .bind(filter.kind)
+    .bind(&search_pat)
+    .fetch_one(pool)
+    .await?;
+
+    Ok((items, total))
+}
+
 /// A manga whose cover is still missing (queued for cover fetch).
 #[derive(Debug, Clone, Serialize, FromRow)]
 pub struct MissingCoverRow {

backend/src/repo/mod.rs

@@ -6,6 +6,7 @@ pub mod author;
 pub mod bookmark;
 pub mod chapter;
 pub mod collection;
+pub mod crawl_metrics;
 pub mod crawler;
 pub mod genre;
 pub mod manga;

backend/src/repo/page.rs

@@ -60,6 +60,27 @@ pub async fn locate(
     Ok(row)
 }
 
+/// Like [`locate`] but also resolves the manga title and chapter number, so
+/// live analysis events can carry human labels for the "now analyzing"
+/// banner without the dashboard having to look them up. Returns
+/// `(manga_id, manga_title, chapter_id, chapter_number, page_number)`.
+pub async fn locate_labeled(
+    pool: &PgPool,
+    page_id: Uuid,
+) -> AppResult<Option<(Uuid, String, Uuid, i32, i32)>> {
+    let row: Option<(Uuid, String, Uuid, i32, i32)> = sqlx::query_as(
+        "SELECT c.manga_id, m.title, p.chapter_id, c.number, p.page_number \
+           FROM pages p \
+           JOIN chapters c ON c.id = p.chapter_id \
+           JOIN mangas m   ON m.id = c.manga_id \
+          WHERE p.id = $1",
+    )
+    .bind(page_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row)
+}
+
 pub async fn list_for_chapter(pool: &PgPool, chapter_id: Uuid) -> AppResult<Vec<Page>> {
     let rows = sqlx::query_as::<_, Page>(
         r#"

backend/src/repo/page_analysis.rs

@@ -16,9 +16,11 @@ use uuid::Uuid;
 
 use crate::crawler::jobs::{self, JobPayload};
 use crate::domain::page_analysis::{
-    ChapterCoverage, ContentWarning, MangaCoverage, OcrKind, OcrLine, PageAnalysis,
-    PageAnalysisDetail, PageSearchItem, PageStatusItem, VisionAnalysis,
+    AnalysisHistoryRow, AnalysisMetrics, ChapterCoverage, ContentWarning, MangaCoverage,
+    ModelDuration, OcrKind, OcrLine, PageAnalysis, PageAnalysisDetail, PageSearchItem,
+    PageStatusItem, VisionAnalysis,
 };
+use chrono::{DateTime, Utc};
 use crate::error::AppResult;
 
 /// Filter set for [`page_search`]. `tags` are AND-ed (a page must carry
@@ -212,6 +214,90 @@ pub async fn chapter_page_status(
     Ok(rows)
 }
 
+/// Filters for [`list_history`]. `None` widens the scope.
+#[derive(Debug, Default, Clone)]
+pub struct AnalysisHistoryFilter<'a> {
+    /// Exact analysis status (`done` or `failed`).
+    pub status: Option<&'a str>,
+    /// Only NSFW-flagged pages.
+    pub nsfw_only: bool,
+    /// Case-insensitive manga-title substring.
+    pub search: Option<&'a str>,
+}
+
+/// Paginated, newest-first analysis history — every completed/failed page
+/// analysis resolved to its manga/chapter/page context. Reads the
+/// persistent `page_analysis` table, so unlike the crawler job history it
+/// is not bounded by job reaping. Returns the page slice plus the filtered
+/// total. `pending` rows are excluded — history is terminal outcomes only.
+pub async fn list_history(
+    pool: &PgPool,
+    filter: AnalysisHistoryFilter<'_>,
+    limit: i64,
+    offset: i64,
+) -> AppResult<(Vec<AnalysisHistoryRow>, i64)> {
+    let search_pat = filter
+        .search
+        .map(|s| format!("%{}%", s.trim()))
+        .filter(|p| p.len() > 2);
+
+    let items = sqlx::query_as::<_, AnalysisHistoryRow>(
+        r#"
+        SELECT
+            pa.page_id,
+            p.page_number,
+            p.chapter_id,
+            c.number     AS chapter_number,
+            c.manga_id,
+            m.title      AS manga_title,
+            pa.status,
+            pa.is_nsfw,
+            pa.model,
+            pa.error,
+            pa.analyzed_at,
+            pa.duration_ms
+          FROM page_analysis pa
+          JOIN pages p    ON p.id = pa.page_id
+          JOIN chapters c ON c.id = p.chapter_id
+          JOIN mangas m   ON m.id = c.manga_id
+         WHERE pa.status <> 'pending'
+           AND ($1::text IS NULL OR pa.status = $1)
+           AND ($2::bool IS FALSE OR pa.is_nsfw)
+           AND ($3::text IS NULL OR m.title ILIKE $3)
+         ORDER BY pa.analyzed_at DESC NULLS LAST, pa.page_id
+         LIMIT $4 OFFSET $5
+        "#,
+    )
+    .bind(filter.status)
+    .bind(filter.nsfw_only)
+    .bind(&search_pat)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await?;
+
+    let (total,): (i64,) = sqlx::query_as(
+        r#"
+        SELECT count(*)
+          FROM page_analysis pa
+          JOIN pages p    ON p.id = pa.page_id
+          JOIN chapters c ON c.id = p.chapter_id
+          JOIN mangas m   ON m.id = c.manga_id
+         WHERE pa.status <> 'pending'
+           AND ($1::text IS NULL OR pa.status = $1)
+           AND ($2::bool IS FALSE OR pa.is_nsfw)
+           AND ($3::text IS NULL OR m.title ILIKE $3)
+        "#,
+    )
+    .bind(filter.status)
+    .bind(filter.nsfw_only)
+    .bind(&search_pat)
+    .fetch_one(pool)
+    .await?;
+
+    Ok((items, total))
+}
+
 /// Full analysis detail for one page. `None` when the page itself doesn't
 /// exist; an existing-but-unanalyzed page returns `status = "none"` with
 /// empty OCR/tags/warnings so the UI can show "not analyzed yet".
@@ -331,6 +417,66 @@ pub async fn mark_failed(pool: &PgPool, page_id: Uuid, error: &str) -> AppResult
     Ok(())
 }
 
+/// Record the wall-clock the worker spent on a page. Best-effort: updates the
+/// existing `page_analysis` row (written by `persist_analysis`/`mark_failed`);
+/// a transient failure with no row yet simply updates nothing.
+pub async fn record_duration(
+    pool: &PgPool,
+    page_id: Uuid,
+    duration_ms: i64,
+) -> AppResult<()> {
+    sqlx::query("UPDATE page_analysis SET duration_ms = $2 WHERE page_id = $1")
+        .bind(page_id)
+        .bind(duration_ms)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+/// Aggregate analysis timing + outcome roll-up over an optional time window
+/// (`since = None` → all time). Mean duration ignores NULL `duration_ms`
+/// (pre-tracking rows); `by_model` breaks the mean out per model.
+pub async fn analysis_metrics(
+    pool: &PgPool,
+    since: Option<DateTime<Utc>>,
+) -> AppResult<AnalysisMetrics> {
+    let (n, ok, failed, avg_ms): (i64, i64, i64, Option<f64>) = sqlx::query_as(
+        r#"
+        SELECT
+            COUNT(*)                                            AS n,
+            COUNT(*) FILTER (WHERE status = 'done')             AS ok,
+            COUNT(*) FILTER (WHERE status = 'failed')           AS failed,
+            -- Mean over successful pages only, so it reconciles with the
+            -- by-model breakdown below (which is done-only). Failed /
+            -- timed-out durations would otherwise skew the headline up.
+            AVG(duration_ms) FILTER (WHERE status = 'done')::float8 AS avg_ms
+          FROM page_analysis
+         WHERE status <> 'pending'
+           AND ($1::timestamptz IS NULL OR analyzed_at >= $1)
+        "#,
+    )
+    .bind(since)
+    .fetch_one(pool)
+    .await?;
+
+    let by_model = sqlx::query_as::<_, ModelDuration>(
+        r#"
+        SELECT model, AVG(duration_ms)::float8 AS avg_ms, COUNT(*) AS n
+          FROM page_analysis
+         WHERE status = 'done'
+           AND duration_ms IS NOT NULL
+           AND ($1::timestamptz IS NULL OR analyzed_at >= $1)
+         GROUP BY model
+         ORDER BY n DESC
+        "#,
+    )
+    .bind(since)
+    .fetch_all(pool)
+    .await?;
+
+    Ok(AnalysisMetrics { n, ok, failed, avg_ms, by_model })
+}
+
 /// Persist a completed analysis for a page, replacing any previous result.
 ///
 /// The model's free-form `kind` / `content_type` strings are mapped onto