Mangalord/backend/src/repo/admin_view.rs

//! Admin-facing read queries that join manga/chapter with the crawler
//! signals (`manga_sources`, `chapter_sources`, `crawler_jobs`) to
//! derive a sync state per row at query time.
//!
//! Priority order for `MangaSyncState`:
//!   1. `InProgress` — any pending/running `sync_manga` or
//!      `sync_chapter_list` job matches this manga.
//!   2. `Dropped` — manga has source rows AND every one of them is
//!      `dropped_at IS NOT NULL`.
//!   3. `Synced` — default (includes user-uploaded mangas with no
//!      `manga_sources` rows at all).
//!
//! Priority order for `ChapterSyncState`:
//!   1. `Downloading` — pending/running `sync_chapter_content` for this id
//!   2. `Dropped` — chapter has source rows AND all are dropped
//!   3. `Failed` — `page_count = 0` AND a `dead` `sync_chapter_content`
//!      row exists for this chapter. Constrained to `page_count = 0`
//!      because once pages are on disk the chapter IS synced — a
//!      historical dead job (likely from a re-download attempt that
//!      crashed) is noise that gets reaped after retention. Surfacing
//!      "Failed" when content is present would contradict
//!      `ChapterSyncState::Synced`'s "downloaded at some point" contract.
//!   4. `NotDownloaded` — `page_count = 0`, no in-flight, no dead job
//!   5. `Synced` — `page_count > 0`
//!
//! Reminder: `done` jobs are reaped after `CRAWLER_JOB_RETENTION_DAYS`,
//! so `chapters.page_count > 0` is the durable "this is synced" signal,
//! not the job table.

use chrono::{DateTime, Utc};
use serde::Serialize;
use sqlx::{FromRow, PgPool};
use uuid::Uuid;

use crate::domain::{ChapterSyncState, MangaSyncState};
use crate::error::AppResult;

#[derive(Debug, Serialize, FromRow)]
pub struct AdminMangaRow {
    pub id: Uuid,
    pub title: String,
    pub status: String,
    pub cover_image_path: Option<String>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub sync_state: MangaSyncState,
    pub chapter_count: i64,
    pub latest_seen_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Default)]
pub struct ListAdminMangasQuery {
    pub search: Option<String>,
    pub sync_state: Option<MangaSyncState>,
    pub limit: i64,
    pub offset: i64,
}

const MANGA_SYNC_STATE_CASE: &str = r#"
    CASE
      WHEN EXISTS (
        SELECT 1 FROM crawler_jobs cj
         WHERE cj.state IN ('pending','running')
           AND (
                (cj.payload->>'kind' = 'sync_chapter_list'
                 AND (cj.payload->>'manga_id')::uuid = m.id)
             OR (cj.payload->>'kind' = 'sync_manga'
                 AND EXISTS (
                       SELECT 1 FROM manga_sources ms
                        WHERE ms.manga_id = m.id
                          AND ms.source_id = cj.payload->>'source_id'
                          AND ms.source_manga_key = cj.payload->>'source_manga_key'
                     ))
              )
      ) THEN 'in_progress'
      WHEN EXISTS (SELECT 1 FROM manga_sources ms WHERE ms.manga_id = m.id)
       AND NOT EXISTS (
             SELECT 1 FROM manga_sources ms
              WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL
           )
      THEN 'dropped'
      ELSE 'synced'
    END
"#;

/// Library shape for the admin overview: total mangas split by derived
/// sync state, plus library-wide chapter/page totals and the newest manga.
#[derive(Debug, Clone, Serialize)]
pub struct MangaStats {
    pub total: i64,
    pub synced: i64,
    pub in_progress: i64,
    pub dropped: i64,
    pub total_chapters: i64,
    pub total_pages: i64,
    pub newest_title: Option<String>,
    pub newest_seen_at: Option<DateTime<Utc>>,
}

/// Aggregate the manga library for the overview dashboard.
///
/// Unlike the paginated Mangas tab (`list_mangas_with_sync_state`, which
/// evaluates `MANGA_SYNC_STATE_CASE` for only the page slice), this scans the
/// WHOLE library. The per-row correlated form is therefore O(mangas x jobs):
/// each manga seqscans `crawler_jobs`, and the disabled-analysis backlog
/// inflates every scan. Instead we collect the set of "in-flight" manga ids in
/// ONE pass over the in-flight sync jobs (index-backed: 0020 + 0029), then
/// classify each manga with a hash semi-join — O(mangas + jobs).
///
/// The `in_progress` rule here MUST stay in lockstep with the first arm of
/// `MANGA_SYNC_STATE_CASE`: pending/running `sync_chapter_list` (by manga_id)
/// OR `sync_manga` (resolved through `manga_sources`). The `dropped`/`synced`
/// arms are identical to the shared case.
pub async fn manga_stats(pool: &PgPool) -> AppResult<MangaStats> {
    let counts_sql = r#"
        WITH in_progress_mangas AS (
            -- sync_chapter_list jobs carry the target manga_id directly
            SELECT (cj.payload->>'manga_id')::uuid AS manga_id
              FROM crawler_jobs cj
             WHERE cj.state IN ('pending', 'running')
               AND cj.payload->>'kind' = 'sync_chapter_list'
               AND cj.payload->>'manga_id' IS NOT NULL
            UNION
            -- sync_manga jobs resolve to a manga via manga_sources
            SELECT ms.manga_id
              FROM crawler_jobs cj
              JOIN manga_sources ms
                ON ms.source_id = cj.payload->>'source_id'
               AND ms.source_manga_key = cj.payload->>'source_manga_key'
             WHERE cj.state IN ('pending', 'running')
               AND cj.payload->>'kind' = 'sync_manga'
        )
        SELECT
            COUNT(*)::bigint                                    AS total,
            COUNT(*) FILTER (WHERE s = 'synced')::bigint       AS synced,
            COUNT(*) FILTER (WHERE s = 'in_progress')::bigint  AS in_progress,
            COUNT(*) FILTER (WHERE s = 'dropped')::bigint      AS dropped
          FROM (
            SELECT
              CASE
                WHEN m.id IN (SELECT manga_id FROM in_progress_mangas)
                  THEN 'in_progress'
                WHEN EXISTS (SELECT 1 FROM manga_sources ms WHERE ms.manga_id = m.id)
                 AND NOT EXISTS (
                       SELECT 1 FROM manga_sources ms
                        WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL
                     )
                  THEN 'dropped'
                ELSE 'synced'
              END AS s
            FROM mangas m
          ) q
        "#;
    // Backstop: even with the single-pass rewrite + indexes this is cheap, but
    // a `statement_timeout` guarantees a pathological plan can never pin a
    // backend for minutes and let pollers stack again. SET LOCAL is scoped to
    // the transaction.
    let mut tx = pool.begin().await?;
    sqlx::query("SET LOCAL statement_timeout = '5s'")
        .execute(&mut *tx)
        .await?;
    let (total, synced, in_progress, dropped): (i64, i64, i64, i64) =
        sqlx::query_as(counts_sql).fetch_one(&mut *tx).await?;
    tx.commit().await?;

    let (total_chapters,): (i64,) =
        sqlx::query_as("SELECT COUNT(*)::bigint FROM chapters")
            .fetch_one(pool)
            .await?;
    let (total_pages,): (i64,) = sqlx::query_as("SELECT COUNT(*)::bigint FROM pages")
        .fetch_one(pool)
        .await?;

    // Newest by creation, with its own latest non-dropped crawl sighting
    // (NULL for user uploads with no source rows).
    let newest: Option<(String, Option<DateTime<Utc>>)> = sqlx::query_as(
        r#"
        SELECT m.title,
               (SELECT MAX(ms.last_seen_at) FROM manga_sources ms
                 WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL)
          FROM mangas m
         ORDER BY m.created_at DESC
         LIMIT 1
        "#,
    )
    .fetch_optional(pool)
    .await?;
    let (newest_title, newest_seen_at) = match newest {
        Some((title, seen)) => (Some(title), seen),
        None => (None, None),
    };

    Ok(MangaStats {
        total,
        synced,
        in_progress,
        dropped,
        total_chapters,
        total_pages,
        newest_title,
        newest_seen_at,
    })
}

/// Paginated admin manga list with derived sync state and total count.
/// Filters by `search` (substring on title, case-insensitive) and
/// `sync_state` (post-derivation). The CTE keeps the case expression
/// in one place — the same projection feeds both the page rows and the
/// totals count under the same filter.
pub async fn list_mangas_with_sync_state(
    pool: &PgPool,
    q: &ListAdminMangasQuery,
) -> AppResult<(Vec<AdminMangaRow>, i64)> {
    let search_pat = q
        .search
        .as_ref()
        .map(|s| format!("%{}%", crate::repo::escape_like(s.trim())))
        .filter(|p| p.len() > 2);
    // sqlx::Type → text: bind the snake_case representation manually so
    // the SQL can compare it as text without an explicit cast.
    let sync_filter = q.sync_state.map(|s| match s {
        MangaSyncState::InProgress => "in_progress",
        MangaSyncState::Dropped => "dropped",
        MangaSyncState::Synced => "synced",
    });

    let sql = format!(
        r#"
        WITH classified AS (
            SELECT
                m.id, m.title, m.status, m.cover_image_path,
                m.created_at, m.updated_at,
                {case} AS sync_state,
                (SELECT COUNT(*) FROM chapters c WHERE c.manga_id = m.id) AS chapter_count,
                (SELECT MAX(last_seen_at) FROM manga_sources ms
                  WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL) AS latest_seen_at
              FROM mangas m
             WHERE ($1::text IS NULL OR m.title ILIKE $1 ESCAPE '\')
        )
        SELECT * FROM classified
         WHERE ($2::text IS NULL OR sync_state = $2)
         -- `id` tiebreaker stabilises pagination across mangas that share an
         -- `updated_at` (multi-row crawl-tick bulk updates land within the
         -- same statement_timestamp() and tie on Postgres' default ts
         -- resolution). Without it, page boundaries can skip or repeat rows.
         ORDER BY updated_at DESC, id DESC
         LIMIT $3 OFFSET $4
        "#,
        case = MANGA_SYNC_STATE_CASE
    );
    let items: Vec<AdminMangaRow> = sqlx::query_as(&sql)
        .bind(&search_pat)
        .bind(sync_filter)
        .bind(q.limit)
        .bind(q.offset)
        .fetch_all(pool)
        .await?;

    let total_sql = format!(
        r#"
        WITH classified AS (
            SELECT {case} AS sync_state
              FROM mangas m
             WHERE ($1::text IS NULL OR m.title ILIKE $1 ESCAPE '\')
        )
        SELECT COUNT(*) FROM classified
         WHERE ($2::text IS NULL OR sync_state = $2)
        "#,
        case = MANGA_SYNC_STATE_CASE
    );
    let total: i64 = sqlx::query_scalar(&total_sql)
        .bind(&search_pat)
        .bind(sync_filter)
        .fetch_one(pool)
        .await?;

    Ok((items, total))
}

#[derive(Debug, Serialize, FromRow)]
pub struct AdminChapterRow {
    pub id: Uuid,
    pub manga_id: Uuid,
    pub number: i32,
    pub title: Option<String>,
    pub page_count: i32,
    pub created_at: DateTime<Utc>,
    pub sync_state: ChapterSyncState,
    pub latest_seen_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Default)]
pub struct ListAdminChaptersQuery {
    pub manga_id: Uuid,
    pub limit: i64,
    pub offset: i64,
}

/// Paginated chapter list with derived sync state. Pagination is non-
/// optional — long-runners can have thousands of chapters and the
/// per-row scalar subqueries make the unbounded variant a real
/// stall risk even behind an admin guard. Returns the page slice plus
/// the unfiltered total so the UI can render "showing N of M".
pub async fn list_chapters_with_sync_state(
    pool: &PgPool,
    q: &ListAdminChaptersQuery,
) -> AppResult<(Vec<AdminChapterRow>, i64)> {
    let items: Vec<AdminChapterRow> = sqlx::query_as(
        r#"
        SELECT
            c.id, c.manga_id, c.number, c.title, c.page_count, c.created_at,
            CASE
              WHEN EXISTS (
                SELECT 1 FROM crawler_jobs cj
                 WHERE cj.state IN ('pending','running')
                   AND cj.payload->>'kind' = 'sync_chapter_content'
                   AND (cj.payload->>'chapter_id')::uuid = c.id
              ) THEN 'downloading'
              WHEN EXISTS (SELECT 1 FROM chapter_sources cs WHERE cs.chapter_id = c.id)
               AND NOT EXISTS (
                     SELECT 1 FROM chapter_sources cs
                      WHERE cs.chapter_id = c.id AND cs.dropped_at IS NULL
                   )
              THEN 'dropped'
              WHEN c.page_count = 0
               AND EXISTS (
                SELECT 1 FROM crawler_jobs cj
                 WHERE cj.state = 'dead'
                   AND cj.payload->>'kind' = 'sync_chapter_content'
                   AND (cj.payload->>'chapter_id')::uuid = c.id
              ) THEN 'failed'
              WHEN c.page_count = 0 THEN 'not_downloaded'
              ELSE 'synced'
            END AS sync_state,
            (SELECT MAX(last_seen_at) FROM chapter_sources cs
              WHERE cs.chapter_id = c.id AND cs.dropped_at IS NULL) AS latest_seen_at
          FROM chapters c
         WHERE c.manga_id = $1
         -- `id` tiebreaker: migration 0013 dropped the (manga_id, number)
         -- UNIQUE constraint to allow variants ("Ch.14: PH" / "Ch.14:
         -- Official") and non-numeric entries (all parse to 0). Without a
         -- tiebreaker, pagination over those tied rows can skip or repeat
         -- depending on Postgres' chosen sort order.
         ORDER BY c.number ASC, c.id ASC
         LIMIT $2 OFFSET $3
        "#,
    )
    .bind(q.manga_id)
    .bind(q.limit)
    .bind(q.offset)
    .fetch_all(pool)
    .await?;

    let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM chapters WHERE manga_id = $1")
        .bind(q.manga_id)
        .fetch_one(pool)
        .await?;

    Ok((items, total))
}