//! Persistence for AI page-analysis results.
//!
//! [`persist_analysis`] is the single transactional writer: it replaces a
//! page's OCR text, global auto-tags, and content warnings, upserts the
//! `page_analysis` row, and computes the kind-weighted `search_doc`
//! tsvector — all atomically, so re-analysis (delete+reinsert) is
//! idempotent and never leaves a half-written page. It deliberately does
//! NOT touch the per-user `page_tags` table.
//!
//! Auto-tag names resolve to the shared `tags` vocabulary via
//! [`crate::repo::tag::upsert_by_name`], the same path manga tags and
//! personal page tags use.

use sqlx::PgPool;
use uuid::Uuid;

use crate::crawler::jobs::{self, JobPayload};
use crate::domain::page_analysis::{
    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
/// every one, satisfiable by EITHER the user's page tag or a global auto
/// tag); `text` ranks via the weighted tsvector; `cw_include` requires all
/// listed warnings; `cw_exclude` rejects any. All string values arrive
/// pre-normalized (tags lowercased, warnings validated) from the handler.
#[derive(Debug, Default, Clone)]
pub struct PageSearchQuery {
    pub user_id: uuid::Uuid,
    pub tags: Vec<String>,
    pub text: Option<String>,
    pub cw_include: Vec<String>,
    pub cw_exclude: Vec<String>,
    pub limit: i64,
    pub offset: i64,
}

/// Longest tag the shared `tags` table accepts (`upsert_by_name` enforces
/// the same bound). Auto-tags over this are dropped here rather than
/// aborting the whole page's analysis on one bad model output.
const MAX_TAG_CHARS: usize = 64;

/// Outcome of [`enqueue_for_page`] — surfaces whether the admin's intent
/// actually landed as a worker-visible change.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EnqueueForPageOutcome {
    /// A fresh `analyze_page` job row was inserted.
    Inserted,
    /// A pending `force=false` row was upgraded in-place to `force=true`,
    /// because the partial unique index `crawler_jobs_analyze_page_dedup_idx`
    /// blocks a second pending insert per page. Returned only for force
    /// requests.
    UpgradedToForce,
    /// A matching `(pending|running)` job already encodes the request;
    /// nothing changed. For force requests this is the retry-race tail:
    /// the INSERT was Skipped, the UPDATE matched zero rows (the sibling
    /// drained between the two), and the second INSERT was Skipped too
    /// (a fresh sibling landed). The next ack of either job will leave
    /// the page in the admin's intended state without further action.
    AlreadyEnqueued,
}

/// Enqueue an `analyze_page` job for `page_id`. `force` re-analyzes a page
/// that is already `done`.
///
/// **Force-vs-dedup correctness.** Migration 0031's partial unique index
/// is keyed on `page_id` (not on `(page_id, force)`), so a plain
/// `jobs::enqueue` of a `force=true` payload when a `force=false` job is
/// already pending silently goes to `Skipped` — the worker would then
/// pick up the non-force row, hit the skip-if-done net, and ack done
/// without re-analyzing. The admin's force request was lost.
///
/// This function fixes that by: (1) trying the insert; (2) on Skipped
/// with `force=true`, running an UPDATE that flips the existing pending
/// row's `force` flag to `true`; (3) reporting which path ran so the
/// caller can audit accurately.
/// Pool convenience wrapper for [`enqueue_for_page_conn`]. Use this from
/// callers that don't need to share a transaction (chapter upload, the
/// crawler). The admin force-reanalyze handler uses the `_conn` form so the
/// enqueue and its `admin_audit` row commit together.
pub async fn enqueue_for_page(
    pool: &PgPool,
    page_id: Uuid,
    force: bool,
) -> AppResult<EnqueueForPageOutcome> {
    let mut conn = pool.acquire().await?;
    enqueue_for_page_conn(&mut conn, page_id, force).await
}

/// Enqueue (or force-upgrade) an `analyze_page` job on a caller-supplied
/// connection, so the admin handler can run it inside the same transaction as
/// its audit insert. The body is a sequence of single statements, each
/// reborrowing `&mut *conn`.
pub async fn enqueue_for_page_conn(
    conn: &mut sqlx::PgConnection,
    page_id: Uuid,
    force: bool,
) -> AppResult<EnqueueForPageOutcome> {
    use crate::crawler::jobs::EnqueueResult;
    match jobs::enqueue(&mut *conn, &JobPayload::AnalyzePage { page_id, force }).await? {
        EnqueueResult::Inserted(_) => Ok(EnqueueForPageOutcome::Inserted),
        EnqueueResult::Skipped if !force => Ok(EnqueueForPageOutcome::AlreadyEnqueued),
        EnqueueResult::Skipped => {
            // Force-specific upgrade. The WHERE clause matches the same
            // partial index predicate (`pending|running` + `analyze_page`
            // + this page_id) and adds `force=false` so we never overwrite
            // a row that's already what the admin wants. RETURNING tells
            // us whether the upgrade actually moved anything AND its
            // pre-update state so the running-state race fix below can
            // release the in-flight lease.
            let upgraded: Vec<(Uuid, String)> = sqlx::query_as(
                "UPDATE crawler_jobs \
                    SET payload = jsonb_set(payload, '{force}', 'true'::jsonb), \
                        updated_at = now() \
                  WHERE state IN ('pending', 'running') \
                    AND payload->>'kind' = 'analyze_page' \
                    AND payload->>'page_id' = $1 \
                    AND (payload->>'force')::boolean = false \
                  RETURNING id, state",
            )
            .bind(page_id.to_string())
            .fetch_all(&mut *conn)
            .await?;
            if upgraded.is_empty() {
                // Race: between the skipped INSERT and the UPDATE the
                // sibling row could have been completed (job state moves
                // out of pending/running). At that point a re-INSERT
                // would succeed. Retry once to close the race without
                // unbounded loops.
                return Ok(
                    match jobs::enqueue(&mut *conn, &JobPayload::AnalyzePage { page_id, force })
                        .await?
                    {
                        EnqueueResult::Inserted(_) => EnqueueForPageOutcome::Inserted,
                        EnqueueResult::Skipped => EnqueueForPageOutcome::AlreadyEnqueued,
                    },
                );
            }
            // Race fix for the running-state branch (0.87.20 followup):
            // a worker that has already leased the row holds the
            // pre-upgrade `force=false` in its in-memory `lease.payload`
            // (see `analysis::daemon::WorkerContext::process_lease`).
            // Even though we've flipped the row's payload to
            // `force=true`, the worker's skip-if-done check uses its
            // own pre-upgrade value and will ack done without
            // re-analyzing. `release_unowned` returns the row to
            // `pending` AND bumps `lease_generation` so the original
            // worker's later `ack_done(id, old_generation)` finds no
            // matching row and is a no-op — the successor's lease (a
            // fresh generation) is preserved. The state-only guard
            // that previously protected this (and was the 0.87.20
            // narrowing) wasn't enough on its own: a successor that
            // re-leased the id would re-enter `state='running'`, and
            // the original's ack would clobber it.
            for (id, state) in &upgraded {
                if state == "running" {
                    let _ = jobs::release_unowned(&mut *conn, *id).await;
                }
            }
            Ok(EnqueueForPageOutcome::UpgradedToForce)
        }
    }
}

/// How wide a bulk re-enqueue reaches.
#[derive(Debug, Clone, Copy)]
pub enum ReenqueueScope {
    /// Every page in the library.
    All,
    /// Every page across all chapters of one manga.
    Manga(uuid::Uuid),
    /// Every page of one chapter.
    Chapter(uuid::Uuid),
}

/// Bulk-enqueue `analyze_page` jobs for existing pages within `scope` — the
/// admin backfill / re-analyze path.
///
/// When `only_unanalyzed` is true, pages that already have a `done`
/// analysis row are skipped and the enqueued jobs carry `force = false`.
/// When false, ALL in-scope pages are enqueued with `force = true` so the
/// worker re-analyzes even already-done pages (otherwise the worker's
/// skip-if-done net would no-op them). Pages with a pending/running
/// `analyze_page` job are always skipped so repeated calls don't pile up
/// duplicates. Returns the number of jobs enqueued.
pub async fn enqueue_pages<'e, E>(
    executor: E,
    scope: ReenqueueScope,
    only_unanalyzed: bool,
) -> AppResult<u64>
where
    E: sqlx::PgExecutor<'e>,
{
    // Scope predicate; the bound uuid (when present) is always $2.
    let scope_clause = match scope {
        ReenqueueScope::All => "",
        ReenqueueScope::Manga(_) => {
            "AND p.chapter_id IN (SELECT id FROM chapters WHERE manga_id = $2)"
        }
        ReenqueueScope::Chapter(_) => "AND p.chapter_id = $2",
    };
    // The `NOT EXISTS(... pending|running ...)` pre-check used to live in
    // this query, but it races with concurrent admin clicks (two requests
    // both see "no in-flight job" and both INSERT). The partial unique
    // index `crawler_jobs_analyze_page_dedup_idx` (migration 0031) covers
    // (payload->>'page_id') WHERE state IN ('pending', 'running') AND
    // kind = 'analyze_page', so `ON CONFLICT DO NOTHING` is now the
    // atomic primitive — duplicates can't land and concurrent enqueue
    // calls are both observably correct.
    let sql = format!(
        r#"
        INSERT INTO crawler_jobs (payload)
        SELECT jsonb_build_object('kind', 'analyze_page', 'page_id', p.id, 'force', NOT $1)
          FROM pages p
         WHERE ($1 = false OR NOT EXISTS (
                   SELECT 1 FROM page_analysis pa
                    WHERE pa.page_id = p.id AND pa.status = 'done'))
           {scope_clause}
        ON CONFLICT DO NOTHING
        "#
    );
    let query = sqlx::query(&sql).bind(only_unanalyzed);
    let query = match scope {
        ReenqueueScope::All => query,
        ReenqueueScope::Manga(id) | ReenqueueScope::Chapter(id) => query.bind(id),
    };
    Ok(query.execute(executor).await?.rows_affected())
}

/// Per-manga analysis coverage for the admin overview. Only mangas that
/// have at least one page appear (nothing to analyze otherwise). `search`
/// filters by title (case-insensitive substring). Ordered by title.
/// Returns the page plus the total matching-manga count for pagination.
pub async fn manga_coverage(
    pool: &PgPool,
    search: Option<&str>,
    limit: i64,
    offset: i64,
) -> AppResult<(Vec<MangaCoverage>, i64)> {
    let rows = sqlx::query_as::<_, MangaCoverage>(
        r#"
        SELECT m.id AS manga_id, m.title,
               count(p.id)                                  AS total_pages,
               count(*) FILTER (WHERE pa.status = 'done')   AS analyzed_pages
          FROM mangas m
          JOIN chapters c ON c.manga_id = m.id
          JOIN pages p    ON p.chapter_id = c.id
          LEFT JOIN page_analysis pa ON pa.page_id = p.id
         WHERE ($1::text IS NULL OR m.title ILIKE '%' || $1 || '%')
         GROUP BY m.id, m.title
         ORDER BY lower(m.title), m.id
         LIMIT $2 OFFSET $3
        "#,
    )
    .bind(search)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;

    let (total,): (i64,) = sqlx::query_as(
        r#"
        SELECT count(*) FROM (
            SELECT m.id
              FROM mangas m
              JOIN chapters c ON c.manga_id = m.id
              JOIN pages p    ON p.chapter_id = c.id
             WHERE ($1::text IS NULL OR m.title ILIKE '%' || $1 || '%')
             GROUP BY m.id
        ) x
        "#,
    )
    .bind(search)
    .fetch_one(pool)
    .await?;
    Ok((rows, total))
}

/// Library-wide analysis coverage `(analyzed_pages, total_pages)` for the
/// admin overview. `total_pages` is every page; `analyzed_pages` are those
/// with a `done` analysis row.
pub async fn library_coverage(pool: &PgPool) -> AppResult<(i64, i64)> {
    let row: (i64, i64) = sqlx::query_as(
        r#"
        SELECT
            count(*) FILTER (WHERE pa.status = 'done')::bigint AS analyzed,
            count(p.id)::bigint                                AS total
          FROM pages p
          LEFT JOIN page_analysis pa ON pa.page_id = p.id
        "#,
    )
    .fetch_one(pool)
    .await?;
    Ok(row)
}

/// Per-chapter analysis coverage for one manga, ordered by chapter number.
pub async fn chapter_coverage(
    pool: &PgPool,
    manga_id: uuid::Uuid,
) -> AppResult<Vec<ChapterCoverage>> {
    let rows = sqlx::query_as::<_, ChapterCoverage>(
        r#"
        SELECT c.id AS chapter_id, c.number, c.title,
               count(p.id)                                AS total_pages,
               count(*) FILTER (WHERE pa.status = 'done') AS analyzed_pages
          FROM chapters c
          JOIN pages p ON p.chapter_id = c.id
          LEFT JOIN page_analysis pa ON pa.page_id = p.id
         WHERE c.manga_id = $1
         GROUP BY c.id, c.number, c.title
         ORDER BY c.number, c.id
        "#,
    )
    .bind(manga_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Per-page status grid for a chapter. `status` resolves to the analysis
/// row's state (`done`/`failed`), else `queued` when a pending/running
/// analyze_page job exists, else `none`.
pub async fn chapter_page_status(
    pool: &PgPool,
    chapter_id: uuid::Uuid,
) -> AppResult<Vec<PageStatusItem>> {
    let rows = sqlx::query_as::<_, PageStatusItem>(
        r#"
        SELECT p.id AS page_id, p.page_number,
               COALESCE(
                   pa.status,
                   CASE WHEN EXISTS (
                       SELECT 1 FROM crawler_jobs j
                        WHERE j.payload->>'kind' = 'analyze_page'
                          AND j.payload->>'page_id' = p.id::text
                          AND j.state IN ('pending', 'running')
                   ) THEN 'queued' ELSE 'none' END
               ) AS status
          FROM pages p
          LEFT JOIN page_analysis pa ON pa.page_id = p.id
         WHERE p.chapter_id = $1
         ORDER BY p.page_number
        "#,
    )
    .bind(chapter_id)
    .fetch_all(pool)
    .await?;
    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".
pub async fn page_detail(
    pool: &PgPool,
    page_id: uuid::Uuid,
) -> AppResult<Option<PageAnalysisDetail>> {
    let Some((page_number, chapter_id, manga_id)): Option<(i32, uuid::Uuid, uuid::Uuid)> =
        sqlx::query_as(
            "SELECT p.page_number, p.chapter_id, c.manga_id \
               FROM pages p JOIN chapters c ON c.id = p.chapter_id WHERE p.id = $1",
        )
        .bind(page_id)
        .fetch_optional(pool)
        .await?
    else {
        return Ok(None);
    };

    let analysis = load(pool, page_id).await?;

    let ocr = sqlx::query_as::<_, OcrLine>(
        "SELECT kind, text FROM page_ocr_text WHERE page_id = $1 ORDER BY ord, id",
    )
    .bind(page_id)
    .fetch_all(pool)
    .await?;

    let tags: Vec<String> = sqlx::query_scalar(
        "SELECT t.name FROM page_auto_tags pat JOIN tags t ON t.id = pat.tag_id \
          WHERE pat.page_id = $1 ORDER BY t.name",
    )
    .bind(page_id)
    .fetch_all(pool)
    .await?;

    let content_warnings = sqlx::query_scalar::<_, ContentWarning>(
        "SELECT warning FROM page_content_warnings WHERE page_id = $1 ORDER BY warning",
    )
    .bind(page_id)
    .fetch_all(pool)
    .await?;

    Ok(Some(PageAnalysisDetail {
        page_id,
        page_number,
        chapter_id,
        manga_id,
        status: analysis
            .as_ref()
            .map(|a| format!("{:?}", a.status).to_lowercase())
            .unwrap_or_else(|| "none".to_string()),
        is_nsfw: analysis.as_ref().map(|a| a.is_nsfw).unwrap_or(false),
        scene_description: analysis.as_ref().and_then(|a| a.scene_description.clone()),
        model: analysis.as_ref().and_then(|a| a.model.clone()),
        error: analysis.as_ref().and_then(|a| a.error.clone()),
        analyzed_at: analysis.as_ref().and_then(|a| a.analyzed_at),
        ocr,
        tags,
        content_warnings,
    }))
}

/// Deduplicated union of content warnings across all of a manga's pages,
/// ordered alphabetically. Powers the manga-detail content-warning banner.
pub async fn warnings_for_manga(
    pool: &PgPool,
    manga_id: uuid::Uuid,
) -> AppResult<Vec<ContentWarning>> {
    let rows = sqlx::query_scalar::<_, ContentWarning>(
        r#"
        SELECT DISTINCT pw.warning
          FROM page_content_warnings pw
          JOIN pages p    ON p.id = pw.page_id
          JOIN chapters c ON c.id = p.chapter_id
         WHERE c.manga_id = $1
         ORDER BY pw.warning
        "#,
    )
    .bind(manga_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Load a page's analysis row, if it has one.
pub async fn load(pool: &PgPool, page_id: Uuid) -> AppResult<Option<PageAnalysis>> {
    let row = sqlx::query_as::<_, PageAnalysis>(
        r#"
        SELECT page_id, status, scene_description, is_nsfw, model, error, analyzed_at
          FROM page_analysis
         WHERE page_id = $1
        "#,
    )
    .bind(page_id)
    .fetch_optional(pool)
    .await?;
    Ok(row)
}

/// Record that analysis failed terminally for a page (retries exhausted /
/// dead-lettered). Leaves a `failed` row so the page's state is
/// observable; it simply contributes nothing to search.
pub async fn mark_failed(pool: &PgPool, page_id: Uuid, error: &str) -> AppResult<()> {
    sqlx::query(
        r#"
        INSERT INTO page_analysis (page_id, status, error)
        VALUES ($1, 'failed', $2)
        ON CONFLICT (page_id) DO UPDATE
           SET status = 'failed', error = EXCLUDED.error, updated_at = now()
        "#,
    )
    .bind(page_id)
    .bind(error)
    .execute(pool)
    .await?;
    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 })
}

/// Bucketed analysis throughput/success/duration series over an optional
/// window, for the Analysis-tab trend charts. `ok`/`failed` follow the
/// `done`/`failed` statuses; mean duration is over completed pages only.
/// Empty intervals are absent (client fills the axis). The
/// `page_analysis_analyzed_at_idx` (migration 0030) backs the scan.
pub async fn analysis_series(
    pool: &PgPool,
    bucket: crate::repo::crawl_metrics::Bucket,
    since: Option<DateTime<Utc>>,
) -> AppResult<Vec<crate::domain::crawl_metrics::MetricsBucket>> {
    let rows = sqlx::query_as::<_, crate::domain::crawl_metrics::MetricsBucket>(
        r#"
        SELECT
            date_trunc($1, analyzed_at)                          AS t,
            COUNT(*)                                             AS n,
            COUNT(*) FILTER (WHERE status = 'done')              AS ok,
            COUNT(*) FILTER (WHERE status = 'failed')            AS failed,
            AVG(duration_ms) FILTER (WHERE status = 'done')::float8 AS avg_ms
          FROM page_analysis
         WHERE status <> 'pending'
           AND analyzed_at IS NOT NULL
           AND ($2::timestamptz IS NULL OR analyzed_at >= $2)
         GROUP BY t
         ORDER BY t
        "#,
    )
    .bind(bucket.as_str())
    .bind(since)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Persist a completed analysis for a page, replacing any previous result.
///
/// The model's free-form `kind` / `content_type` strings are mapped onto
/// the closed vocabularies here ([`OcrKind::from_model_str`] /
/// [`ContentWarning::from_model_str`]); unrecognized warnings are dropped,
/// unknown OCR kinds fall back to `narration`. Empty / over-long tags and
/// empty OCR text are skipped so one bad item never aborts the page.
pub async fn persist_analysis(
    pool: &PgPool,
    page_id: Uuid,
    analysis: &VisionAnalysis,
    model: &str,
) -> AppResult<()> {
    // Kind-weighted text buckets for the tsvector: A=speech/title,
    // B=narration/thought/caption, C=scene, D=sfx.
    let mut bucket_a = String::new();
    let mut bucket_b = String::new();
    let bucket_c = analysis.scene_description.trim().to_string();
    let mut bucket_d = String::new();

    let mut ocr_rows: Vec<(OcrKind, String)> = Vec::new();
    for r in &analysis.ocr_results {
        let text = r.text.trim();
        if text.is_empty() {
            continue;
        }
        let kind = OcrKind::from_model_str(&r.kind);
        match kind.weight() {
            'A' => push_token(&mut bucket_a, text),
            'B' => push_token(&mut bucket_b, text),
            'D' => push_token(&mut bucket_d, text),
            _ => {}
        }
        ocr_rows.push((kind, text.to_string()));
    }

    // Dedup tags case-insensitively, preserving first-seen order; drop
    // empties and over-long names rather than failing the transaction.
    let mut seen_tags = std::collections::HashSet::new();
    let mut tags: Vec<String> = Vec::new();
    for raw in &analysis.tagging_results {
        let t = raw.trim();
        if t.is_empty() || t.chars().count() > MAX_TAG_CHARS {
            continue;
        }
        if seen_tags.insert(t.to_lowercase()) {
            tags.push(t.to_string());
        }
    }

    // Map + dedup warnings.
    let mut seen_warn = std::collections::HashSet::new();
    let mut warnings: Vec<ContentWarning> = Vec::new();
    for raw in &analysis.safety_flag.content_type {
        if let Some(w) = ContentWarning::from_model_str(raw) {
            if seen_warn.insert(w) {
                warnings.push(w);
            }
        }
    }

    let mut tx = pool.begin().await?;

    sqlx::query("DELETE FROM page_ocr_text WHERE page_id = $1")
        .bind(page_id)
        .execute(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM page_auto_tags WHERE page_id = $1")
        .bind(page_id)
        .execute(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM page_content_warnings WHERE page_id = $1")
        .bind(page_id)
        .execute(&mut *tx)
        .await?;

    for (ord, (kind, text)) in ocr_rows.iter().enumerate() {
        sqlx::query(
            "INSERT INTO page_ocr_text (page_id, kind, text, ord) VALUES ($1, $2, $3, $4)",
        )
        .bind(page_id)
        .bind(kind)
        .bind(text)
        .bind(ord as i32)
        .execute(&mut *tx)
        .await?;
    }

    for tag in &tags {
        let tag_row = crate::repo::tag::upsert_by_name(&mut *tx, tag).await?;
        sqlx::query(
            "INSERT INTO page_auto_tags (page_id, tag_id) VALUES ($1, $2) \
             ON CONFLICT (page_id, tag_id) DO NOTHING",
        )
        .bind(page_id)
        .bind(tag_row.id)
        .execute(&mut *tx)
        .await?;
    }

    for w in &warnings {
        sqlx::query(
            "INSERT INTO page_content_warnings (page_id, warning) VALUES ($1, $2) \
             ON CONFLICT (page_id, warning) DO NOTHING",
        )
        .bind(page_id)
        .bind(w)
        .execute(&mut *tx)
        .await?;
    }

    let scene = if bucket_c.is_empty() {
        None
    } else {
        Some(bucket_c.as_str())
    };
    sqlx::query(
        r#"
        INSERT INTO page_analysis
            (page_id, status, scene_description, is_nsfw, model, analyzed_at, search_doc)
        VALUES
            ($1, 'done', $2, $3, $4, now(),
             setweight(to_tsvector('simple', $5), 'A') ||
             setweight(to_tsvector('simple', $6), 'B') ||
             setweight(to_tsvector('simple', $7), 'C') ||
             setweight(to_tsvector('simple', $8), 'D'))
        ON CONFLICT (page_id) DO UPDATE SET
            status            = 'done',
            scene_description = EXCLUDED.scene_description,
            is_nsfw           = EXCLUDED.is_nsfw,
            model             = EXCLUDED.model,
            analyzed_at       = now(),
            error             = NULL,
            search_doc        = EXCLUDED.search_doc,
            updated_at        = now()
        "#,
    )
    .bind(page_id)
    .bind(scene)
    .bind(analysis.safety_flag.is_nsfw)
    .bind(model)
    .bind(&bucket_a)
    .bind(&bucket_b)
    .bind(&bucket_c)
    .bind(&bucket_d)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(())
}

/// Content search over pages: multi-tag AND across (user page tags ∪
/// global auto tags), optional weighted text search over the OCR/scene
/// document, and content-warning include/exclude. Returns one row per
/// matching page plus the total for pagination.
///
/// Empty `tags` makes the tag clause vacuously true (text- or
/// warning-only search), mirroring the manga search's `unnest` idiom. The
/// caller is responsible for requiring at least one positive filter so
/// this never degenerates into "every page".
pub async fn page_search(
    pool: &PgPool,
    q: &PageSearchQuery,
) -> AppResult<(Vec<PageSearchItem>, i64)> {
    // `text` participates in three places ($3): the rank, the match
    // predicate, and the order key. Empty string disables text filtering.
    let text = q.text.as_deref().unwrap_or("").trim();

    const WHERE: &str = r#"
        NOT EXISTS (
            SELECT 1 FROM unnest($2::text[]) AS req(name)
             WHERE NOT EXISTS (
                SELECT 1 FROM page_tags ut
                  JOIN tags t ON t.id = ut.tag_id
                 WHERE ut.page_id = p.id AND ut.user_id = $1 AND lower(t.name) = req.name
                UNION ALL
                SELECT 1 FROM page_auto_tags at
                  JOIN tags t ON t.id = at.tag_id
                 WHERE at.page_id = p.id AND lower(t.name) = req.name
             )
        )
        AND ($3 = '' OR pa.search_doc @@ plainto_tsquery('simple', $3))
        AND NOT EXISTS (
            SELECT 1 FROM unnest($4::text[]) AS req(w)
             WHERE NOT EXISTS (
                SELECT 1 FROM page_content_warnings pw
                 WHERE pw.page_id = p.id AND pw.warning = req.w
             )
        )
        AND NOT EXISTS (
            SELECT 1 FROM page_content_warnings pw
             WHERE pw.page_id = p.id AND pw.warning = ANY($5::text[])
        )
    "#;

    let rows_sql = format!(
        r#"
        SELECT
            p.id          AS page_id,
            p.chapter_id  AS chapter_id,
            ch.manga_id   AS manga_id,
            p.page_number AS page_number,
            ch.number     AS chapter_number,
            ch.title      AS chapter_title,
            m.title       AS manga_title,
            p.storage_key AS storage_key,
            COALESCE(pa.is_nsfw, false) AS is_nsfw,
            COALESCE(
                (SELECT array_agg(pw.warning ORDER BY pw.warning)
                   FROM page_content_warnings pw WHERE pw.page_id = p.id),
                ARRAY[]::text[]
            ) AS content_warnings,
            COALESCE(ts_rank(pa.search_doc, plainto_tsquery('simple', $3)), 0)::real AS rank
          FROM pages p
          JOIN chapters ch ON ch.id = p.chapter_id
          JOIN mangas   m  ON m.id  = ch.manga_id
          LEFT JOIN page_analysis pa ON pa.page_id = p.id
         WHERE {WHERE}
         ORDER BY (CASE WHEN $3 = '' THEN 0 ELSE 1 END) DESC, rank DESC, p.id
         LIMIT $6 OFFSET $7
        "#
    );
    let rows = sqlx::query_as::<_, PageSearchItem>(&rows_sql)
        .bind(q.user_id)
        .bind(&q.tags)
        .bind(text)
        .bind(&q.cw_include)
        .bind(&q.cw_exclude)
        .bind(q.limit)
        .bind(q.offset)
        .fetch_all(pool)
        .await?;

    let count_sql = format!(
        "SELECT count(*) FROM pages p \
           JOIN chapters ch ON ch.id = p.chapter_id \
           JOIN mangas m ON m.id = ch.manga_id \
           LEFT JOIN page_analysis pa ON pa.page_id = p.id \
          WHERE {WHERE}"
    );
    let (total,): (i64,) = sqlx::query_as(&count_sql)
        .bind(q.user_id)
        .bind(&q.tags)
        .bind(text)
        .bind(&q.cw_include)
        .bind(&q.cw_exclude)
        .fetch_one(pool)
        .await?;

    Ok((rows, total))
}

/// Append a token to a tsvector text bucket with a trailing space.
fn push_token(bucket: &mut String, text: &str) {
    bucket.push_str(text);
    bucket.push(' ');
}