//! 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::{
    ContentWarning, OcrKind, PageAnalysis, PageSearchItem, VisionAnalysis,
};
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;

/// Enqueue an `analyze_page` job for `page_id`. `force` re-analyzes a page
/// that is already `done`. Enqueue is idempotent at the job level only in
/// that duplicate pending jobs are harmless — processing is idempotent.
pub async fn enqueue_for_page(pool: &PgPool, page_id: Uuid, force: bool) -> AppResult<()> {
    jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await?;
    Ok(())
}

/// 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(
    pool: &PgPool,
    scope: ReenqueueScope,
    only_unanalyzed: bool,
) -> AppResult<u64> {
    // 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",
    };
    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}
           AND NOT 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'))
        "#
    );
    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(pool).await?.rows_affected())
}

/// 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(())
}

/// 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(' ');
}