Mangalord/backend/src/repo/storage_stats.rs

//! Cross-cutting storage-usage roll-ups for the admin dashboard, plus
//! the one-shot size backfill helpers. These join `pages` → `chapters`
//! → `mangas`, so they live here rather than on any single-resource repo
//! module (mirrors `repo::admin_view`). Every figure is a single indexed
//! aggregate; the `pages_chapter_size_idx` (migration 0027) keeps the
//! per-chapter / per-manga roll-ups cheap.
//!
//! `size_bytes` / `cover_size_bytes` are nullable: NULL means "not yet
//! measured". SUM/AVG ignore NULL, so an un-backfilled library reports
//! honest partial totals; the backfill fills the NULLs.

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

use crate::domain::{TopChapter, TopManga};
use crate::error::AppResult;

/// Max rows stat-ed + updated per backfill batch. Bounds memory and
/// keeps each UPDATE statement's array parameters reasonable.
pub const BACKFILL_BATCH: i64 = 500;

/// `(covers_bytes, chapters_bytes)` — total measured bytes split by kind.
/// Unmeasured (NULL) rows contribute nothing.
pub async fn totals(pool: &PgPool) -> AppResult<(i64, i64)> {
    let (covers,): (i64,) =
        sqlx::query_as("SELECT COALESCE(SUM(cover_size_bytes), 0)::bigint FROM mangas")
            .fetch_one(pool)
            .await?;
    let (chapters,): (i64,) =
        sqlx::query_as("SELECT COALESCE(SUM(size_bytes), 0)::bigint FROM pages")
            .fetch_one(pool)
            .await?;
    Ok((covers, chapters))
}

/// `(avg_all, avg_cover, avg_chapter_page)` average image sizes. Each is
/// `None` when its population has no measured rows (no division by zero).
/// `AVG` ignores NULL, so unmeasured blobs are naturally excluded.
pub async fn averages(
    pool: &PgPool,
) -> AppResult<(Option<f64>, Option<f64>, Option<f64>)> {
    let (avg_chapter,): (Option<f64>,) =
        sqlx::query_as("SELECT AVG(size_bytes)::float8 FROM pages")
            .fetch_one(pool)
            .await?;
    let (avg_cover,): (Option<f64>,) =
        sqlx::query_as("SELECT AVG(cover_size_bytes)::float8 FROM mangas")
            .fetch_one(pool)
            .await?;
    let (avg_all,): (Option<f64>,) = sqlx::query_as(
        "SELECT AVG(s)::float8 FROM (
             SELECT size_bytes AS s FROM pages WHERE size_bytes IS NOT NULL
             UNION ALL
             SELECT cover_size_bytes FROM mangas WHERE cover_size_bytes IS NOT NULL
         ) x",
    )
    .fetch_one(pool)
    .await?;
    Ok((avg_all, avg_cover, avg_chapter))
}

/// Largest mangas by cover + summed chapter-page bytes. Only *fully
/// measured* mangas are ranked — a manga with any unmeasured page (or an
/// unmeasured cover) is excluded rather than ranked by an understated
/// partial total, so the board never disagrees with the per-manga
/// "unknown" the detail page shows. (Pre-backfill the board is empty; the
/// dashboard's unmeasured-count banner explains why.)
pub async fn top_mangas(pool: &PgPool, limit: i64) -> AppResult<Vec<TopManga>> {
    let rows: Vec<(Uuid, String, i64)> = sqlx::query_as(
        "SELECT id, title, total_bytes FROM (
             SELECT m.id, m.title,
                    (COALESCE(m.cover_size_bytes, 0)
                     + COALESCE((SELECT SUM(p.size_bytes)
                                   FROM pages p
                                   JOIN chapters c ON c.id = p.chapter_id
                                  WHERE c.manga_id = m.id), 0))::bigint AS total_bytes,
                    ((m.cover_image_path IS NULL OR m.cover_size_bytes IS NOT NULL)
                     AND NOT EXISTS (SELECT 1 FROM pages p
                                       JOIN chapters c ON c.id = p.chapter_id
                                      WHERE c.manga_id = m.id
                                        AND p.size_bytes IS NULL)) AS complete
               FROM mangas m
         ) t
         WHERE total_bytes > 0 AND complete
         ORDER BY total_bytes DESC, id
         LIMIT $1",
    )
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|(id, title, total_bytes)| TopManga { id, title, total_bytes })
        .collect())
}

/// Largest chapters by summed stored page bytes. Only *fully measured*
/// chapters are ranked: INNER JOIN drops page-less chapters, and the
/// `bool_or(... IS NULL)` guard drops any chapter with an unmeasured page
/// so the board matches the chapter list's "unknown" (em-dash) instead of
/// ranking by an understated partial.
pub async fn top_chapters(pool: &PgPool, limit: i64) -> AppResult<Vec<TopChapter>> {
    let rows: Vec<(Uuid, Uuid, i32, Option<String>, i64)> = sqlx::query_as(
        "SELECT c.id, c.manga_id, c.number, c.title,
                SUM(p.size_bytes)::bigint AS total_bytes
           FROM chapters c
           JOIN pages p ON p.chapter_id = c.id
          GROUP BY c.id
         HAVING bool_or(p.size_bytes IS NULL) IS NOT TRUE
            AND SUM(p.size_bytes) > 0
          ORDER BY total_bytes DESC, c.id
          LIMIT $1",
    )
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|(id, manga_id, number, title, total_bytes)| TopChapter {
            id,
            manga_id,
            number,
            title,
            total_bytes,
        })
        .collect())
}

/// Total bytes of a single manga's chapter pages (cover excluded), summed
/// over **all** its chapters. `None` when any of the manga's pages is
/// unmeasured (so the UI shows "unknown" rather than a silent undercount);
/// `Some(0)` when the manga has no pages. This is computed in the DB over
/// the full page set — it can't be derived correctly from a paginated
/// chapter list — so `MangaDetail` carries it even though it adds one
/// indexed aggregate to the (un-polled) detail read.
pub async fn manga_chapter_bytes(pool: &PgPool, manga_id: Uuid) -> AppResult<Option<i64>> {
    let (total,): (Option<i64>,) = sqlx::query_as(
        "SELECT CASE
                  WHEN count(*) = 0 THEN 0
                  WHEN bool_or(p.size_bytes IS NULL) THEN NULL
                  ELSE sum(p.size_bytes)
                END::bigint
           FROM pages p
           JOIN chapters c ON c.id = p.chapter_id
          WHERE c.manga_id = $1",
    )
    .bind(manga_id)
    .fetch_one(pool)
    .await?;
    Ok(total)
}

/// `(unmeasured_pages, unmeasured_covers)` — counts of rows still awaiting
/// a backfill. Drives the dashboard "N items not yet measured" banner and
/// signals whether another backfill run is needed.
pub async fn unmeasured_counts(pool: &PgPool) -> AppResult<(i64, i64)> {
    let (pages,): (i64,) =
        sqlx::query_as("SELECT count(*)::bigint FROM pages WHERE size_bytes IS NULL")
            .fetch_one(pool)
            .await?;
    let (covers,): (i64,) = sqlx::query_as(
        "SELECT count(*)::bigint FROM mangas \
         WHERE cover_size_bytes IS NULL AND cover_image_path IS NOT NULL",
    )
    .fetch_one(pool)
    .await?;
    Ok((pages, covers))
}

// --- backfill ---------------------------------------------------------
//
// Keyset-paginated so the whole un-backfilled set is never materialized
// at once and the loop terminates even when some blobs can't be stat-ed
// (those rows stay NULL but `id > after` advances past them). The UPDATEs
// are batched via UNNEST and guarded on `IS NULL` so a concurrent write
// of a real size between the scan and the update is never clobbered.

/// Up to `limit` pages with an unmeasured size, ordered by id after
/// `after`. Pass `Uuid::nil()` to start.
pub async fn pages_needing_backfill_after(
    pool: &PgPool,
    after: Uuid,
    limit: i64,
) -> AppResult<Vec<(Uuid, String)>> {
    let rows: Vec<(Uuid, String)> = sqlx::query_as(
        "SELECT id, storage_key FROM pages \
         WHERE size_bytes IS NULL AND id > $1 ORDER BY id LIMIT $2",
    )
    .bind(after)
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Mangas with a cover blob but no measured cover size, ordered by id
/// after `after`.
pub async fn covers_needing_backfill_after(
    pool: &PgPool,
    after: Uuid,
    limit: i64,
) -> AppResult<Vec<(Uuid, String)>> {
    let rows: Vec<(Uuid, String)> = sqlx::query_as(
        "SELECT id, cover_image_path FROM mangas \
         WHERE cover_size_bytes IS NULL AND cover_image_path IS NOT NULL \
           AND id > $1 ORDER BY id LIMIT $2",
    )
    .bind(after)
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Apply a batch of `(id, size)` updates in one statement and return the
/// number of rows actually written. The `IS NULL` guard makes this a
/// no-op for any row a concurrent crawl/upload has already measured,
/// closing the lost-update window; returning the affected count (rather
/// than `ids.len()`) keeps the backfill's reported total honest when that
/// happens.
pub async fn batch_set_page_sizes(
    pool: &PgPool,
    ids: &[Uuid],
    sizes: &[i64],
) -> AppResult<u64> {
    if ids.is_empty() {
        return Ok(0);
    }
    let result = sqlx::query(
        "UPDATE pages SET size_bytes = v.s \
           FROM unnest($1::uuid[], $2::bigint[]) AS v(id, s) \
          WHERE pages.id = v.id AND pages.size_bytes IS NULL",
    )
    .bind(ids)
    .bind(sizes)
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}

pub async fn batch_set_cover_sizes(
    pool: &PgPool,
    ids: &[Uuid],
    sizes: &[i64],
) -> AppResult<u64> {
    if ids.is_empty() {
        return Ok(0);
    }
    let result = sqlx::query(
        "UPDATE mangas SET cover_size_bytes = v.s \
           FROM unnest($1::uuid[], $2::bigint[]) AS v(id, s) \
          WHERE mangas.id = v.id AND mangas.cover_size_bytes IS NULL",
    )
    .bind(ids)
    .bind(sizes)
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}