//! Manga persistence.
//!
//! Plain async functions over `&PgPool` rather than a repository struct —
//! each function is easy to test in isolation with `#[sqlx::test]`, and
//! handlers depend only on `sqlx::PgPool`, not on a trait object. Swap to
//! a trait + impl if a second backend ever becomes necessary.

use serde::Deserialize;
use sqlx::{PgConnection, PgExecutor, PgPool};
use uuid::Uuid;

use crate::domain::manga::{Manga, MangaCard, MangaDetail};
use crate::error::{AppError, AppResult};
use crate::repo;

/// Status values mirror the CHECK constraint in 0009. Centralized so
/// the API layer can validate uploads against the same vocabulary.
pub const STATUSES: &[&str] = &["ongoing", "completed"];
pub const DEFAULT_STATUS: &str = "ongoing";

#[derive(Debug, Clone, Copy, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ListSort {
    /// Newest first (default).
    #[default]
    Recent,
    /// A→Z by title (case-insensitive).
    Title,
}

#[derive(Debug, Clone, Default)]
pub struct ListQuery {
    pub search: Option<String>,
    pub status: Option<String>,
    pub author_ids: Vec<Uuid>,
    pub genre_ids: Vec<Uuid>,
    pub tag_ids: Vec<Uuid>,
    /// Content warnings (canonical lowercase names) the manga must carry on
    /// at least one page each — AND across the list.
    pub cw_include: Vec<String>,
    /// Content warnings the manga must NOT carry on any page.
    pub cw_exclude: Vec<String>,
    pub limit: i64,
    pub offset: i64,
    pub sort: ListSort,
}

/// Single source of truth for the `mangas` columns that hydrate a [`Manga`],
/// so the plain and join-aliased select lists stay in lockstep with the
/// struct's `FromRow`.
const MANGA_COLS: [&str; 8] = [
    "id",
    "title",
    "status",
    "alt_titles",
    "description",
    "cover_image_path",
    "created_at",
    "updated_at",
];

/// `MANGA_COLS` rendered as a select list. A non-empty `alias` qualifies each
/// column (`"m"` → `m.id, m.title, …`) for queries that join other tables;
/// an empty alias yields the bare names used by single-table queries.
fn manga_cols(alias: &str) -> String {
    if alias.is_empty() {
        MANGA_COLS.join(", ")
    } else {
        MANGA_COLS
            .iter()
            .map(|c| format!("{alias}.{c}"))
            .collect::<Vec<_>>()
            .join(", ")
    }
}

/// Shared WHERE used by both the rows and the count queries. Filters
/// are AND across facets: every supplied author_id (or genre, or tag)
/// must be attached to the manga for it to match. Empty arrays mean
/// "no filter" because `NOT EXISTS` over an empty unnest is vacuously
/// true.
const FILTER_WHERE: &str = r#"
    ($1::text IS NULL
     OR title ILIKE '%' || $1 || '%'
     OR title % $1
     OR EXISTS (
            SELECT 1 FROM manga_authors ma
              JOIN authors a ON a.id = ma.author_id
             WHERE ma.manga_id = mangas.id
               AND (a.name ILIKE '%' || $1 || '%' OR a.name % $1)
        )
    )
    AND ($2::text IS NULL OR status = $2)
    AND NOT EXISTS (
        SELECT 1 FROM unnest($3::uuid[]) AS req(id)
         WHERE NOT EXISTS (
                SELECT 1 FROM manga_authors ma
                 WHERE ma.manga_id = mangas.id AND ma.author_id = req.id
             )
    )
    AND NOT EXISTS (
        SELECT 1 FROM unnest($4::uuid[]) AS req(id)
         WHERE NOT EXISTS (
                SELECT 1 FROM manga_genres mg
                 WHERE mg.manga_id = mangas.id AND mg.genre_id = req.id
             )
    )
    AND NOT EXISTS (
        SELECT 1 FROM unnest($5::uuid[]) AS req(id)
         WHERE NOT EXISTS (
                SELECT 1 FROM manga_tags mt
                 WHERE mt.manga_id = mangas.id AND mt.tag_id = req.id
             )
    )
    AND NOT EXISTS (
        SELECT 1 FROM unnest($6::text[]) AS req(w)
         WHERE NOT EXISTS (
                SELECT 1 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 = mangas.id AND pw.warning = req.w
             )
    )
    AND NOT EXISTS (
        SELECT 1 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 = mangas.id AND pw.warning = ANY($7::text[])
    )
"#;

/// Returns the page of mangas matching `query` plus the unfiltered total
/// count for the same filter. The trigram GIN indexes (see 0005_search.sql
/// and 0009_manga_metadata.sql) keep both queries cheap as the library
/// grows.
pub async fn list(pool: &PgPool, query: &ListQuery) -> AppResult<(Vec<Manga>, i64)> {
    // `order_by` is interpolated from a hard-coded enum, never from request
    // input, so this is not a SQL injection seam.
    let order_by = match query.sort {
        ListSort::Recent => "created_at DESC, id",
        ListSort::Title => "lower(title) ASC, id",
    };

    let search = query.search.as_deref();
    let status = query.status.as_deref();

    let list_sql = format!(
        r#"
        SELECT {cols}
        FROM mangas
        WHERE {FILTER_WHERE}
        ORDER BY {order_by}
        LIMIT $8 OFFSET $9
        "#,
        cols = manga_cols(""),
    );

    let rows = sqlx::query_as::<_, Manga>(&list_sql)
        .bind(search)
        .bind(status)
        .bind(&query.author_ids)
        .bind(&query.genre_ids)
        .bind(&query.tag_ids)
        .bind(&query.cw_include)
        .bind(&query.cw_exclude)
        .bind(query.limit)
        .bind(query.offset)
        .fetch_all(pool)
        .await?;

    let count_sql = format!(
        r#"
        SELECT count(*) FROM mangas
        WHERE {FILTER_WHERE}
        "#
    );
    let (total,): (i64,) = sqlx::query_as(&count_sql)
        .bind(search)
        .bind(status)
        .bind(&query.author_ids)
        .bind(&query.genre_ids)
        .bind(&query.tag_ids)
        .bind(&query.cw_include)
        .bind(&query.cw_exclude)
        .fetch_one(pool)
        .await?;

    Ok((rows, total))
}

/// Same filter as `list`, but wraps each row with its authors + genres
/// in a single batched round-trip. Tags are intentionally not loaded
/// here — see `MangaCard` in the domain layer.
pub async fn list_cards(
    pool: &PgPool,
    query: &ListQuery,
) -> AppResult<(Vec<MangaCard>, i64)> {
    let (rows, total) = list(pool, query).await?;
    let cards = cards_from_rows(pool, rows).await?;
    Ok((cards, total))
}

/// Top-`limit` mangas ranked by tag similarity to `id`, as cards.
///
/// Similarity is the Jaccard index of the two tag sets —
/// `shared / (|base| + |other| - shared)` — so a heavily-tagged manga
/// sharing many generic tags does not crowd out genuinely-close matches.
/// (A simpler raw shared-tag count is one ORDER BY term away, but it
/// biases toward mangas with large tag sets; see the tie-break below.)
///
/// The candidate set comes from the self-join on `manga_tags`, which by
/// construction only includes mangas sharing at least one tag and never
/// the source itself. A source with no tags — or no overlap — yields an
/// empty result. Existence of `id` is the caller's concern (the empty
/// result here cannot distinguish "no tags" from "no such manga").
pub async fn list_similar(
    pool: &PgPool,
    id: Uuid,
    limit: i64,
) -> AppResult<Vec<MangaCard>> {
    let sql = format!(
        r#"
        SELECT {cols}
        FROM manga_tags base
        JOIN manga_tags other
          ON other.tag_id = base.tag_id
         AND other.manga_id <> base.manga_id
        JOIN mangas m ON m.id = other.manga_id
        WHERE base.manga_id = $1
        GROUP BY m.id
        ORDER BY
          count(*)::float
            / ( (SELECT count(*) FROM manga_tags WHERE manga_id = $1)
                + (SELECT count(*) FROM manga_tags WHERE manga_id = m.id)
                - count(*) ) DESC,
          count(*) DESC,
          m.updated_at DESC,
          lower(m.title) ASC,
          m.id
        LIMIT $2
        "#,
        // The query joins `manga_tags`, so the mangas columns need the `m.` alias.
        cols = manga_cols("m"),
    );

    let rows = sqlx::query_as::<_, Manga>(&sql)
        .bind(id)
        .bind(limit)
        .fetch_all(pool)
        .await?;

    cards_from_rows(pool, rows).await
}

/// Hydrate a batch of `Manga` rows into `MangaCard`s by attaching their
/// authors and genres in two batched round-trips. The input order is
/// preserved (callers rely on this to keep list/ranking order), so we
/// iterate `rows` rather than the id-ordered `BTreeMap`s.
async fn cards_from_rows(pool: &PgPool, rows: Vec<Manga>) -> AppResult<Vec<MangaCard>> {
    let ids: Vec<Uuid> = rows.iter().map(|m| m.id).collect();
    // Authors and genres are independent reads — load them concurrently.
    let (mut authors, mut genres) = tokio::try_join!(
        repo::author::load_for_mangas(pool, &ids),
        repo::genre::load_for_mangas(pool, &ids),
    )?;
    let cards = rows
        .into_iter()
        .map(|manga| MangaCard {
            authors: authors.remove(&manga.id).unwrap_or_default(),
            genres: genres.remove(&manga.id).unwrap_or_default(),
            manga,
        })
        .collect();
    Ok(cards)
}

pub async fn get(pool: &PgPool, id: Uuid) -> AppResult<Manga> {
    sqlx::query_as::<_, Manga>(&format!(
        "SELECT {} FROM mangas WHERE id = $1",
        manga_cols("")
    ))
    .bind(id)
    .fetch_optional(pool)
    .await?
    .ok_or(AppError::NotFound)
}

pub async fn get_detail(pool: &PgPool, id: Uuid) -> AppResult<MangaDetail> {
    let manga = get(pool, id).await?;
    let authors = repo::author::list_for_manga(pool, id).await?;
    let genres = repo::genre::list_for_manga(pool, id).await?;
    let tags = repo::tag::list_for_manga(pool, id).await?;
    let content_warnings = repo::page_analysis::warnings_for_manga(pool, id).await?;
    let chapter_storage_bytes = repo::storage_stats::manga_chapter_bytes(pool, id).await?;
    Ok(MangaDetail {
        manga,
        authors,
        genres,
        tags,
        content_warnings,
        chapter_storage_bytes,
    })
}

/// Insert just the manga row. Relations (authors, genres) are written
/// by the caller via `repo::author::set_for_manga` etc. in the same
/// transaction. `status` is taken as a validated string — the handler
/// is responsible for defaulting/validating it.
///
/// `uploaded_by` records who created the manga and feeds the per-user
/// upload history. `None` means "historical / no associated user" —
/// historic rows from before the uploader columns were added carry
/// NULL.
pub async fn create<'e, E: PgExecutor<'e>>(
    executor: E,
    title: &str,
    status: &str,
    description: Option<&str>,
    alt_titles: &[String],
    uploaded_by: Option<Uuid>,
) -> AppResult<Manga> {
    let row = sqlx::query_as::<_, Manga>(&format!(
        r#"
        INSERT INTO mangas (title, status, description, alt_titles, uploaded_by)
        VALUES ($1, $2, $3, $4, $5)
        RETURNING {cols}
        "#,
        cols = manga_cols(""),
    ))
    .bind(title)
    .bind(status)
    .bind(description)
    .bind(alt_titles)
    .bind(uploaded_by)
    .fetch_one(executor)
    .await?;
    Ok(row)
}

/// Patch the inline columns. Each `Option` argument leaves the field
/// untouched when `None`. `description` uses two binds — a `provided`
/// boolean and the new value — so callers can distinguish "leave
/// alone" from "set to NULL" (both look the same in a plain
/// `Option<&str>` because serde collapses missing and explicit-null).
pub async fn update_basics(
    conn: &mut PgConnection,
    id: Uuid,
    title: Option<&str>,
    status: Option<&str>,
    description_provided: bool,
    description: Option<&str>,
    alt_titles: Option<&[String]>,
) -> AppResult<Manga> {
    let row = sqlx::query_as::<_, Manga>(&format!(
        r#"
        UPDATE mangas
           SET title       = COALESCE($2, title),
               status      = COALESCE($3, status),
               description = CASE WHEN $4::boolean THEN $5 ELSE description END,
               alt_titles  = COALESCE($6, alt_titles),
               updated_at  = now()
         WHERE id = $1
        RETURNING {cols}
        "#,
        cols = manga_cols(""),
    ))
    .bind(id)
    .bind(title)
    .bind(status)
    .bind(description_provided)
    .bind(description)
    .bind(alt_titles)
    .fetch_optional(&mut *conn)
    .await?
    .ok_or(AppError::NotFound)?;
    Ok(row)
}

pub async fn set_cover_image_path<'e, E: PgExecutor<'e>>(
    executor: E,
    id: Uuid,
    key: &str,
    size_bytes: i64,
) -> AppResult<()> {
    sqlx::query(
        "UPDATE mangas SET cover_image_path = $1, cover_size_bytes = $2, updated_at = now() \
         WHERE id = $3",
    )
    .bind(key)
    .bind(size_bytes)
    .bind(id)
    .execute(executor)
    .await?;
    Ok(())
}

pub async fn clear_cover_image_path<'e, E: PgExecutor<'e>>(
    executor: E,
    id: Uuid,
) -> AppResult<()> {
    sqlx::query(
        "UPDATE mangas SET cover_image_path = NULL, cover_size_bytes = NULL, updated_at = now() \
         WHERE id = $1",
    )
    .bind(id)
    .execute(executor)
    .await?;
    Ok(())
}

pub async fn exists(pool: &PgPool, id: Uuid) -> AppResult<bool> {
    let (exists,): (bool,) =
        sqlx::query_as("SELECT EXISTS(SELECT 1 FROM mangas WHERE id = $1)")
            .bind(id)
            .fetch_one(pool)
            .await?;
    Ok(exists)
}

/// Returns the uploader's user id for a manga. `None` either when the
/// manga doesn't exist or when the row predates the `uploaded_by`
/// column (historical NULL — see migration 0011). Callers must
/// distinguish "manga missing" via [`exists`] before relying on this
/// to make an authz decision.
pub async fn uploaded_by(pool: &PgPool, id: Uuid) -> AppResult<Option<Uuid>> {
    let row: Option<(Option<Uuid>,)> =
        sqlx::query_as("SELECT uploaded_by FROM mangas WHERE id = $1")
            .bind(id)
            .fetch_optional(pool)
            .await?;
    Ok(row.and_then(|(u,)| u))
}