//! Per-user, per-page tag persistence.
//!
//! Same plain-function pattern as the rest of `repo`. Tag names resolve
//! to the shared `tags` table (the same one `manga_tags` uses) via
//! [`crate::repo::tag::upsert_by_name`]; `page_tags` links rows by
//! `tag_id` (see migration 0024). The API still speaks tag *names* — the
//! name<->id resolution lives entirely here. Idempotent upserts via
//! `ON CONFLICT DO NOTHING` (the caller distinguishes 201/200 from the
//! returned `bool`), FK-violation remap to NotFound so the handler can
//! return 404 when the page was deleted between an existence check and
//! the insert.

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

use crate::domain::page_tag::{
    PageTagSummary, TaggedChapterAggregate, TaggedMangaAggregate, TaggedPageItem,
};
use crate::error::{AppError, AppResult};

/// Sort direction for the per-chapter / per-manga aggregations. We
/// inline this into the SQL via `format!()` because Postgres won't
/// accept ASC/DESC as a parameter — and the value space is closed
/// (just two variants), so there's no SQL-injection vector.
#[derive(Debug, Clone, Copy)]
pub enum Order {
    Desc,
    Asc,
}

impl Order {
    fn as_sql(self) -> &'static str {
        match self {
            Order::Desc => "DESC",
            Order::Asc => "ASC",
        }
    }
}

/// Insert a tag for `(user_id, page_id)`. The tag name is resolved to
/// (or created in) the shared `tags` table first — the same table and
/// `upsert_by_name` helper the manga-tag path uses — then linked by
/// `tag_id`. Returns `true` if a new link was inserted (handler → 201),
/// `false` if the tag was already present (handler → 200).
pub async fn upsert(
    pool: &PgPool,
    user_id: Uuid,
    page_id: Uuid,
    tag: &str,
) -> AppResult<bool> {
    let mut tx = pool.begin().await?;
    let tag_row = crate::repo::tag::upsert_by_name(&mut *tx, tag).await?;
    let result = sqlx::query(
        r#"
        INSERT INTO page_tags (user_id, page_id, tag_id)
        VALUES ($1, $2, $3)
        ON CONFLICT (user_id, page_id, tag_id) DO NOTHING
        "#,
    )
    .bind(user_id)
    .bind(page_id)
    .bind(tag_row.id)
    .execute(&mut *tx)
    .await
    .map_err(|e| match e {
        sqlx::Error::Database(ref db_err) if db_err.is_foreign_key_violation() => {
            AppError::NotFound
        }
        other => AppError::Database(other),
    })?;
    tx.commit().await?;
    Ok(result.rows_affected() > 0)
}

/// Remove a tag from `(user_id, page_id)`. The tag arrives as a name; we
/// resolve it to the shared `tags` row case-insensitively in a subquery
/// so a no-op delete (unknown name) simply matches nothing.
pub async fn remove(
    pool: &PgPool,
    user_id: Uuid,
    page_id: Uuid,
    tag: &str,
) -> AppResult<()> {
    sqlx::query(
        r#"
        DELETE FROM page_tags
         WHERE user_id = $1
           AND page_id = $2
           AND tag_id = (SELECT id FROM tags WHERE lower(name) = lower($3))
        "#,
    )
    .bind(user_id)
    .bind(page_id)
    .bind(tag)
    .execute(pool)
    .await?;
    Ok(())
}

/// `user_id`'s tags on `page_id`, oldest-first so the context-menu
/// summary line reads in the order the user added them.
pub async fn list_for_page(
    pool: &PgPool,
    user_id: Uuid,
    page_id: Uuid,
) -> AppResult<Vec<String>> {
    let rows: Vec<(String,)> = sqlx::query_as(
        r#"
        SELECT t.name AS tag
          FROM page_tags pt
          JOIN tags t ON t.id = pt.tag_id
         WHERE pt.user_id = $1 AND pt.page_id = $2
         ORDER BY pt.created_at ASC, t.name
        "#,
    )
    .bind(user_id)
    .bind(page_id)
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().map(|(t,)| t).collect())
}

/// Escape a string for use as a LIKE pattern fragment: `%`, `_`, and
/// `\` get a leading backslash so they're matched literally rather
/// than as wildcards / escapes. The matching queries below pair this
/// with `ESCAPE '\'` for explicitness — a single backslash, since the
/// SQL lives in a raw string and Postgres treats `\\` in a single-
/// quoted literal as one backslash under `standard_conforming_strings`.
///
/// The public API rejects `%`/`_`/`\` in `normalize_tag` before
/// they reach this repo, so this is defence-in-depth — a future
/// internal caller (worker, CLI) that bypasses the normalizer can't
/// turn a prefix filter into a wildcard search by accident.
fn escape_like_prefix(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for ch in s.chars() {
        if matches!(ch, '\\' | '%' | '_') {
            out.push('\\');
        }
        out.push(ch);
    }
    out
}

/// Paged list of `user_id`'s tagged pages, with breadcrumb. When
/// `tag_filter` is `Some(_)`, restrict to that exact tag (used by the
/// library Page-tags chip filter). `prefix_filter` does a `LIKE
/// 'prefix%'` against the normalized tag (used by autocomplete when
/// the user is typing a chip). The prefix is LIKE-escaped here so
/// stray `%` / `_` from a bypass-the-API caller don't widen the
/// pattern.
pub async fn list_for_user(
    pool: &PgPool,
    user_id: Uuid,
    tag_filter: Option<&str>,
    prefix_filter: Option<&str>,
    limit: i64,
    offset: i64,
) -> AppResult<(Vec<TaggedPageItem>, i64)> {
    let escaped_prefix = prefix_filter.map(escape_like_prefix);
    let rows = sqlx::query_as::<_, TaggedPageItem>(
        r#"
        SELECT
            t.name         AS tag,
            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,
            pt.created_at  AS tagged_at
          FROM page_tags pt
          JOIN tags     t  ON t.id = pt.tag_id
          JOIN pages    p  ON p.id = pt.page_id
          JOIN chapters ch ON ch.id = p.chapter_id
          JOIN mangas   m  ON m.id = ch.manga_id
         WHERE pt.user_id = $1
           AND ($2::text IS NULL OR lower(t.name) = $2)
           AND ($3::text IS NULL OR lower(t.name) LIKE $3 || '%' ESCAPE '\')
         ORDER BY pt.created_at DESC, pt.id
         LIMIT $4 OFFSET $5
        "#,
    )
    .bind(user_id)
    .bind(tag_filter)
    .bind(escaped_prefix.as_deref())
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;

    let escaped_prefix = prefix_filter.map(escape_like_prefix);
    let (total,): (i64,) = sqlx::query_as(
        r#"
        SELECT count(*)
          FROM page_tags pt
          JOIN tags t ON t.id = pt.tag_id
         WHERE pt.user_id = $1
           AND ($2::text IS NULL OR lower(t.name) = $2)
           AND ($3::text IS NULL OR lower(t.name) LIKE $3 || '%' ESCAPE '\')
        "#,
    )
    .bind(user_id)
    .bind(tag_filter)
    .bind(escaped_prefix.as_deref())
    .fetch_one(pool)
    .await?;
    Ok((rows, total))
}

/// Distinct tag list for the user, with per-tag counts. When `prefix`
/// is `Some(_)`, restrict to tags starting with that prefix — drives
/// the autocomplete dropdown in the "Add tag" sheet. The prefix is
/// LIKE-escaped here so stray `%` / `_` from a bypass-the-API caller
/// don't widen the pattern.
pub async fn distinct_tags_for_user(
    pool: &PgPool,
    user_id: Uuid,
    prefix: Option<&str>,
    limit: i64,
) -> AppResult<Vec<PageTagSummary>> {
    let escaped_prefix = prefix.map(escape_like_prefix);
    let rows = sqlx::query_as::<_, PageTagSummary>(
        r#"
        SELECT t.name AS tag, count(*) AS count
          FROM page_tags pt
          JOIN tags t ON t.id = pt.tag_id
         WHERE pt.user_id = $1
           AND ($2::text IS NULL OR lower(t.name) LIKE $2 || '%' ESCAPE '\')
         GROUP BY t.name
         ORDER BY count DESC, t.name
         LIMIT $3
        "#,
    )
    .bind(user_id)
    .bind(escaped_prefix.as_deref())
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Paged list of chapters that contain pages tagged `tag` for `user_id`,
/// ranked by `match_count`. Each row carries up to 3 sample page
/// storage keys (page-number ascending) so the row can render a
/// thumbnail strip without a follow-up fetch.
///
/// When `text` is non-blank, results are further restricted to pages whose
/// analysis `search_doc` matches the query (OCR text search), and both
/// `match_count` and the sample thumbnails reflect that filtered set —
/// i.e. `match_count` counts tagged pages whose OCR also matches `text`.
///
/// `order` is inlined via `format!()` — the enum value space is
/// closed (`ASC` / `DESC`) so this is not a SQL-injection vector.
pub async fn aggregate_chapters_for_tag(
    pool: &PgPool,
    user_id: Uuid,
    tag: &str,
    order: Order,
    limit: i64,
    offset: i64,
    text: Option<&str>,
) -> AppResult<(Vec<TaggedChapterAggregate>, i64)> {
    // OCR text filter: when present, additionally require the page's analysis
    // `search_doc` to match the query. Reuses the precomputed tsvector exactly
    // like `repo::page_analysis::page_search`. `None`/blank ⇒ tag-only.
    let text = text.map(str::trim).filter(|s| !s.is_empty());
    let (text_join, text_where) = match text {
        // `$5` in the main query, `$3` in the count query (see binds below).
        Some(_) => (
            "JOIN page_analysis pa ON pa.page_id = p.id",
            "AND pa.search_doc @@ plainto_tsquery('simple', {n})",
        ),
        None => ("", ""),
    };
    // Same filter inside the correlated sample-thumbnail subquery (its page is
    // aliased `p`), so the thumbnails match what the text search matched. The
    // subquery lives in the main query, so it reuses the `$5` text bind.
    let (sample_join, sample_where) = match text {
        Some(_) => (
            "JOIN page_analysis pa2 ON pa2.page_id = p.id",
            "AND pa2.search_doc @@ plainto_tsquery('simple', $5)",
        ),
        None => ("", ""),
    };
    let sql = format!(
        r#"
        SELECT
            ch.id           AS chapter_id,
            ch.manga_id     AS manga_id,
            m.title         AS manga_title,
            ch.number       AS chapter_number,
            ch.title        AS chapter_title,
            count(*)        AS match_count,
            COALESCE(
                (
                    SELECT array_agg(p2.storage_key ORDER BY p2.page_number ASC)
                    FROM (
                        SELECT p.storage_key, p.page_number
                          FROM pages p
                          JOIN page_tags pt2 ON pt2.page_id = p.id
                          JOIN tags      t2  ON t2.id = pt2.tag_id
                          {sample_join}
                         WHERE p.chapter_id = ch.id
                           AND pt2.user_id = $1
                           AND lower(t2.name) = $2
                           {sample_where}
                         ORDER BY p.page_number ASC
                         LIMIT 3
                    ) p2
                ),
                ARRAY[]::text[]
            ) AS sample_storage_keys
          FROM page_tags pt
          JOIN tags     t  ON t.id = pt.tag_id
          JOIN pages    p  ON p.id = pt.page_id
          JOIN chapters ch ON ch.id = p.chapter_id
          JOIN mangas   m  ON m.id = ch.manga_id
          {text_join}
         WHERE pt.user_id = $1
           AND lower(t.name) = $2
           {text_where}
         GROUP BY ch.id, ch.manga_id, m.title, ch.number, ch.title
         ORDER BY match_count {dir}, ch.id
         LIMIT $3 OFFSET $4
        "#,
        dir = order.as_sql(),
        text_join = text_join,
        text_where = text_where.replace("{n}", "$5"),
        sample_join = sample_join,
        sample_where = sample_where,
    );
    let mut q = sqlx::query_as::<_, TaggedChapterAggregate>(&sql)
        .bind(user_id)
        .bind(tag)
        .bind(limit)
        .bind(offset);
    if let Some(text) = text {
        q = q.bind(text);
    }
    let rows = q.fetch_all(pool).await?;

    let count_sql = format!(
        r#"
        SELECT count(*) FROM (
            SELECT 1
              FROM page_tags pt
              JOIN tags  t ON t.id = pt.tag_id
              JOIN pages p ON p.id = pt.page_id
              {text_join}
             WHERE pt.user_id = $1 AND lower(t.name) = $2
               {text_where}
             GROUP BY p.chapter_id
        ) c
        "#,
        text_join = text_join,
        text_where = text_where.replace("{n}", "$3"),
    );
    let mut cq = sqlx::query_as::<_, (i64,)>(&count_sql).bind(user_id).bind(tag);
    if let Some(text) = text {
        cq = cq.bind(text);
    }
    let (total,) = cq.fetch_one(pool).await?;
    Ok((rows, total))
}

/// Paged list of mangas containing pages tagged `tag` for `user_id`,
/// ranked by `match_count` summed across all their chapters.
///
/// `text` behaves as in [`aggregate_chapters_for_tag`]: non-blank restricts to
/// pages whose OCR `search_doc` matches, and both `match_count` and the sample
/// thumbnails reflect that filtered set.
pub async fn aggregate_mangas_for_tag(
    pool: &PgPool,
    user_id: Uuid,
    tag: &str,
    order: Order,
    limit: i64,
    offset: i64,
    text: Option<&str>,
) -> AppResult<(Vec<TaggedMangaAggregate>, i64)> {
    // OCR text filter — see `aggregate_chapters_for_tag` for the rationale.
    let text = text.map(str::trim).filter(|s| !s.is_empty());
    let (text_join, text_where) = match text {
        Some(_) => (
            "JOIN page_analysis pa ON pa.page_id = p.id",
            "AND pa.search_doc @@ plainto_tsquery('simple', {n})",
        ),
        None => ("", ""),
    };
    // Same filter inside the sample-thumbnail subquery (page aliased `p`),
    // reusing the main query's `$5` text bind.
    let (sample_join, sample_where) = match text {
        Some(_) => (
            "JOIN page_analysis pa2 ON pa2.page_id = p.id",
            "AND pa2.search_doc @@ plainto_tsquery('simple', $5)",
        ),
        None => ("", ""),
    };
    let sql = format!(
        r#"
        SELECT
            m.id                AS manga_id,
            m.title             AS manga_title,
            m.cover_image_path  AS manga_cover_image_path,
            count(*)            AS match_count,
            COALESCE(
                (
                    SELECT array_agg(p2.storage_key ORDER BY p2.page_number ASC)
                    FROM (
                        SELECT p.storage_key, p.page_number
                          FROM pages p
                          JOIN chapters ch2 ON ch2.id = p.chapter_id
                          JOIN page_tags pt2 ON pt2.page_id = p.id
                          JOIN tags      t2  ON t2.id = pt2.tag_id
                          {sample_join}
                         WHERE ch2.manga_id = m.id
                           AND pt2.user_id  = $1
                           AND lower(t2.name) = $2
                           {sample_where}
                         ORDER BY p.page_number ASC
                         LIMIT 3
                    ) p2
                ),
                ARRAY[]::text[]
            ) AS sample_storage_keys
          FROM page_tags pt
          JOIN tags     t  ON t.id  = pt.tag_id
          JOIN pages    p  ON p.id  = pt.page_id
          JOIN chapters ch ON ch.id = p.chapter_id
          JOIN mangas   m  ON m.id  = ch.manga_id
          {text_join}
         WHERE pt.user_id = $1
           AND lower(t.name) = $2
           {text_where}
         GROUP BY m.id, m.title, m.cover_image_path
         ORDER BY match_count {dir}, m.id
         LIMIT $3 OFFSET $4
        "#,
        dir = order.as_sql(),
        text_join = text_join,
        text_where = text_where.replace("{n}", "$5"),
        sample_join = sample_join,
        sample_where = sample_where,
    );
    let mut q = sqlx::query_as::<_, TaggedMangaAggregate>(&sql)
        .bind(user_id)
        .bind(tag)
        .bind(limit)
        .bind(offset);
    if let Some(text) = text {
        q = q.bind(text);
    }
    let rows = q.fetch_all(pool).await?;

    let count_sql = format!(
        r#"
        SELECT count(*) FROM (
            SELECT 1
              FROM page_tags pt
              JOIN tags     t  ON t.id = pt.tag_id
              JOIN pages    p  ON p.id = pt.page_id
              JOIN chapters ch ON ch.id = p.chapter_id
              {text_join}
             WHERE pt.user_id = $1 AND lower(t.name) = $2
               {text_where}
             GROUP BY ch.manga_id
        ) m
        "#,
        text_join = text_join,
        text_where = text_where.replace("{n}", "$3"),
    );
    let mut cq = sqlx::query_as::<_, (i64,)>(&count_sql).bind(user_id).bind(tag);
    if let Some(text) = text {
        cq = cq.bind(text);
    }
    let (total,) = cq.fetch_one(pool).await?;
    Ok((rows, total))
}

