Mangalord/backend/src/repo/manga.rs

//! 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 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";

/// Which column the listing is ordered by. The direction lives in a separate
/// [`SortOrder`] so any field can be sorted either way. Wire values are parsed
/// (with validation and the legacy `recent` alias) in `api::mangas`, not via
/// serde, so this stays a plain domain enum.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SortField {
    /// Creation time.
    Created,
    /// Last-modified time (default — "last updated first" pairs with `Desc`).
    #[default]
    Updated,
    /// Title (case-insensitive).
    Title,
    /// Alphabetically-first attached author (case-insensitive); authorless
    /// mangas sort last regardless of direction.
    Author,
}

impl SortField {
    /// The direction applied when the client omits `order`. Dates read
    /// newest-first (`Desc`); text reads A→Z (`Asc`). This mirrors the
    /// frontend's `defaultOrderFor` (see `frontend/src/lib/mangaSort.ts`) so a
    /// bare `?sort=<field>` link means the same thing to the UI and the API.
    pub fn default_order(self) -> SortOrder {
        match self {
            SortField::Created | SortField::Updated => SortOrder::Desc,
            SortField::Title | SortField::Author => SortOrder::Asc,
        }
    }
}

/// Sort direction.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SortOrder {
    Asc,
    #[default]
    Desc,
}

#[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: SortField,
    pub order: SortOrder,
}

/// 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 date/title sort columns are index-backed
/// (`mangas_created_at_idx`, `mangas_updated_at_idx`, `mangas_title_lower_idx`)
/// and the trigram GIN indexes (0005_search.sql, 0009_manga_metadata.sql) keep
/// the search filter cheap as the library grows.
pub async fn list(pool: &PgPool, query: &ListQuery) -> AppResult<(Vec<Manga>, Option<i64>)> {
    // Both `col` and `dir` are interpolated from hard-coded enums, never from
    // request input, so this is not a SQL injection seam. The trailing `id` is
    // a stable tie-break that keeps pagination deterministic across rows with
    // equal sort keys.
    let col = match query.sort {
        SortField::Created => "created_at",
        SortField::Updated => "updated_at",
        SortField::Title => "lower(title)",
        // Sorts on the alphabetically-first attached author. As an ORDER BY
        // key this correlated subquery is evaluated per filter-matching row
        // before LIMIT applies, so it scales worse than the indexed date/title
        // sorts — revisit with a LATERAL join or precomputed sort-name column
        // if the library grows large.
        SortField::Author => {
            "(SELECT min(lower(a.name)) \
               FROM manga_authors ma JOIN authors a ON a.id = ma.author_id \
              WHERE ma.manga_id = mangas.id)"
        }
    };
    let dir = match query.order {
        SortOrder::Asc => "ASC",
        SortOrder::Desc => "DESC",
    };
    // Only the author key can be NULL (authorless mangas); keep those last in
    // both directions. The date/title columns are NOT NULL, so they need no
    // NULLS clause — and omitting it lets a reverse index scan satisfy `DESC`.
    let nulls = match query.sort {
        SortField::Author => " NULLS LAST",
        _ => "",
    };
    let order_by = format!("{col} {dir}{nulls}, 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?;

    // The count reuses FILTER_WHERE — up to five correlated NOT EXISTS/unnest
    // subqueries (plus a page_content_warnings join under CW filters) over the
    // whole filtered set. Recomputing it on every page made pagination scale
    // with catalog size. It doesn't change as the caller walks pages, so
    // compute it once on the first page (offset 0) and return None thereafter;
    // the pagination envelope serialises that as `total: null`.
    let total = if query.offset == 0 {
        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?;
        Some(total)
    } else {
        None
    };

    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>, Option<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
}

/// Content-based "Recommended for you": rank mangas by weighted tag overlap
/// with the user's taste. Signals: explicit like = +1.0, bookmark = +0.5,
/// dislike = -1.0 (a reaction overrides a bookmark on the same manga). Per
/// tag we sum those weights into an affinity, then score each candidate by
/// the sum of its tags' affinities, normalized by the candidate's tag count
/// (same anti-tag-stuffing rationale as `list_similar`). Candidates the user
/// already reacted to, bookmarked, or read are excluded; net-negative
/// candidates (dominated by disliked-tag affinity) are dropped, so a dislike
/// down-ranks rather than the manga being hidden from normal browse. No
/// signals → empty. Reuses `cards_from_rows` for author/genre hydration.
pub async fn list_recommendations(
    pool: &PgPool,
    user_id: Uuid,
    limit: i64,
) -> AppResult<Vec<MangaCard>> {
    let sql = format!(
        r#"
        WITH signals AS (
            SELECT s.manga_id,
                   CASE WHEN r.reaction = 'dislike' THEN -1.0
                        WHEN r.reaction = 'like'    THEN  1.0
                        ELSE 0.5 END AS weight
            FROM (
                SELECT manga_id FROM manga_reactions WHERE user_id = $1
                UNION
                SELECT manga_id FROM bookmarks       WHERE user_id = $1
            ) s
            LEFT JOIN manga_reactions r
              ON r.user_id = $1 AND r.manga_id = s.manga_id
        ),
        tag_affinity AS (
            SELECT mt.tag_id, SUM(sig.weight) AS affinity
            FROM signals sig
            JOIN manga_tags mt ON mt.manga_id = sig.manga_id
            GROUP BY mt.tag_id
        )
        SELECT {cols}
        FROM manga_tags cand
        JOIN tag_affinity ta ON ta.tag_id = cand.tag_id
        JOIN mangas m ON m.id = cand.manga_id
        WHERE cand.manga_id NOT IN (SELECT manga_id FROM signals)
          AND cand.manga_id NOT IN (
                SELECT manga_id FROM read_progress WHERE user_id = $1
          )
        GROUP BY m.id
        HAVING SUM(ta.affinity) > 0
        ORDER BY
          SUM(ta.affinity)
            / (SELECT count(*) FROM manga_tags WHERE manga_id = m.id) DESC,
          SUM(ta.affinity) DESC,
          m.updated_at DESC,
          lower(m.title) ASC,
          m.id
        LIMIT $2
        "#,
        cols = manga_cols("m"),
    );

    let rows = sqlx::query_as::<_, Manga>(&sql)
        .bind(user_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))
}