Mangalord/backend/src/api/admin/storage.rs

//! Admin storage-usage stats + the one-shot size backfill.
//!
//! Kept separate from `/admin/system` on purpose: that handler eats a
//! 250 ms CPU sample on every call. Storage figures are pure DB
//! aggregates with no such cost, so they get their own endpoint.
//!
//! Both handlers are admin-only (`RequireAdmin`, cookie-only).

use axum::extract::State;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::Serialize;
use serde_json::json;
use uuid::Uuid;

use crate::app::AppState;
use crate::auth::extractor::RequireAdmin;
use crate::domain::StorageStats;
use crate::error::AppResult;
use crate::repo;
use crate::repo::storage_stats::BACKFILL_BATCH;
use crate::storage::StorageError;

/// How many entries each leaderboard returns.
const TOP_N: i64 = 5;

pub fn routes() -> Router<AppState> {
    Router::new()
        .route("/admin/storage", get(storage))
        .route("/admin/storage/backfill", post(backfill))
}

async fn storage(
    State(state): State<AppState>,
    _admin: RequireAdmin,
) -> AppResult<Json<StorageStats>> {
    // Independent reads — run concurrently so dashboard latency is the
    // slowest query, not their sum.
    let (totals, averages, top_mangas, top_chapters, unmeasured) = tokio::try_join!(
        repo::storage_stats::totals(&state.db),
        repo::storage_stats::averages(&state.db),
        repo::storage_stats::top_mangas(&state.db, TOP_N),
        repo::storage_stats::top_chapters(&state.db, TOP_N),
        repo::storage_stats::unmeasured_counts(&state.db),
    )?;
    let (covers_bytes, chapters_bytes) = totals;
    let (avg_image_bytes, avg_cover_bytes, avg_chapter_page_bytes) = averages;
    let (unmeasured_pages, unmeasured_covers) = unmeasured;
    let total_bytes = covers_bytes + chapters_bytes;

    // Reuse the System tab's statvfs formula so the disk total and the
    // ratio can never drift apart.
    let disk_total_bytes = state
        .storage
        .local_root()
        .and_then(super::system::disk_stats_for)
        .map(|d| d.total_bytes);
    let ratio_of_disk = disk_total_bytes.and_then(|d| {
        if d > 0 {
            Some(total_bytes as f64 / d as f64)
        } else {
            None
        }
    });

    Ok(Json(StorageStats {
        total_bytes,
        covers_bytes,
        chapters_bytes,
        disk_total_bytes,
        ratio_of_disk,
        avg_image_bytes,
        avg_cover_bytes,
        avg_chapter_page_bytes,
        top_mangas,
        top_chapters,
        unmeasured_pages,
        unmeasured_covers,
    }))
}

/// Upper bound on blobs stat-ed in a single backfill request. Each stat
/// is a syscall (or, for a future S3 backend, a network round-trip), so
/// an unbounded run over a huge legacy library could exceed a proxy /
/// client timeout. Capping keeps each request bounded in wall-clock; the
/// run is idempotent and `more_remaining` tells the caller to run again.
const MAX_PER_RUN: usize = 20_000;

#[derive(Debug, Serialize)]
pub struct BackfillResponse {
    /// Pages whose size was written this run (rows actually updated).
    pub pages: usize,
    /// Covers whose size was written this run.
    pub covers: usize,
    /// Blobs the row points at but storage reports as absent
    /// (deleted/orphaned). Expected during cleanup; left unmeasured.
    pub missing: usize,
    /// Blobs that errored for another reason (bad key, transient IO).
    /// Distinct from `missing` so an operator doesn't mistake a retryable
    /// failure for a genuinely orphaned blob. Re-running may resolve these.
    pub errored: usize,
    /// `true` when the per-run cap was hit before the backlog drained, so
    /// the caller should run the backfill again to finish.
    pub more_remaining: bool,
}

/// Idempotent: only touches unmeasured rows (`size_bytes IS NULL`,
/// `cover_size_bytes IS NULL`). Keyset-paginated in batches so the whole
/// backlog is never held in memory; capped at `MAX_PER_RUN` stats so the
/// request stays bounded in wall-clock. The batched UPDATEs are
/// `IS NULL`-guarded so a concurrent write is never clobbered, and the
/// reported counts are the rows actually written. Blobs that can't be
/// stat-ed are counted and skipped — never fatal — so a single orphan
/// doesn't abort the run.
async fn backfill(
    State(state): State<AppState>,
    admin: RequireAdmin,
) -> AppResult<Json<BackfillResponse>> {
    let mut resp = BackfillResponse {
        pages: 0,
        covers: 0,
        missing: 0,
        errored: 0,
        more_remaining: false,
    };
    // Stats attempted this run, across pages + covers, against the cap.
    let mut budget = MAX_PER_RUN;

    // Pages. `pages_after` ends at the max id processed (rows are scanned in
    // id order), so it cleanly separates attempted (id <= cursor) from
    // not-yet-reached (id > cursor) rows.
    let mut pages_after = Uuid::nil();
    while budget > 0 {
        let limit = (budget as i64).min(BACKFILL_BATCH);
        let batch =
            repo::storage_stats::pages_needing_backfill_after(&state.db, pages_after, limit)
                .await?;
        if batch.is_empty() {
            break;
        }
        let mut ids = Vec::with_capacity(batch.len());
        let mut sizes = Vec::with_capacity(batch.len());
        for (id, key) in &batch {
            pages_after = *id; // advance past every row, including failures
            budget -= 1;
            if let Some(size) =
                classify(state.storage.size(key).await, &mut resp.missing, &mut resp.errored)
            {
                ids.push(*id);
                sizes.push(size);
            }
        }
        resp.pages +=
            repo::storage_stats::batch_set_page_sizes(&state.db, &ids, &sizes).await? as usize;
    }

    // Covers.
    let mut covers_after = Uuid::nil();
    while budget > 0 {
        let limit = (budget as i64).min(BACKFILL_BATCH);
        let batch =
            repo::storage_stats::covers_needing_backfill_after(&state.db, covers_after, limit)
                .await?;
        if batch.is_empty() {
            break;
        }
        let mut ids = Vec::with_capacity(batch.len());
        let mut sizes = Vec::with_capacity(batch.len());
        for (id, key) in &batch {
            covers_after = *id;
            budget -= 1;
            if let Some(size) =
                classify(state.storage.size(key).await, &mut resp.missing, &mut resp.errored)
            {
                ids.push(*id);
                sizes.push(size);
            }
        }
        resp.covers +=
            repo::storage_stats::batch_set_cover_sizes(&state.db, &ids, &sizes).await? as usize;
    }

    // `more_remaining` = the scan stopped at the cap AND unmeasured rows
    // remain *beyond the cursor we reached*. The keyset peek (LIMIT 1 past
    // the cursor) excludes rows we already attempted this run, so a
    // permanently un-stat-able orphan we passed over (id <= cursor) never
    // keeps this true — it converges instead of looping forever. It's also
    // exact at the boundary: a backlog of exactly the cap drains to "no rows
    // past the cursor" → false (no spurious follow-up prompt).
    if budget == 0 {
        let more_pages = !repo::storage_stats::pages_needing_backfill_after(
            &state.db,
            pages_after,
            1,
        )
        .await?
        .is_empty();
        let more_covers = !repo::storage_stats::covers_needing_backfill_after(
            &state.db,
            covers_after,
            1,
        )
        .await?
        .is_empty();
        resp.more_remaining = more_pages || more_covers;
    }

    // Audit is written after the action here *by necessity*, not oversight
    // (unlike reenqueue/analyze_page, which wrap their single DB mutation +
    // audit in one tx). The backfill is a budgeted scan that interleaves
    // filesystem `storage.size()` reads with per-batch DB writes across many
    // iterations; wrapping it in a transaction would hold one open across all
    // that I/O (a long-running tx). The batch size-writes are also idempotent
    // recomputations, so a lost audit row has negligible impact.
    repo::admin_audit::insert(
        &state.db,
        admin.0.id,
        "storage_backfill",
        "system",
        None,
        json!({
            "pages": resp.pages,
            "covers": resp.covers,
            "missing": resp.missing,
            "errored": resp.errored,
        }),
    )
    .await?;

    Ok(Json(resp))
}

/// Turn a `Storage::size` result into the byte count to persist, bumping
/// the right counter on failure. `NotFound` → `missing` (orphaned);
/// anything else (bad key, IO) → `errored` (possibly retryable).
fn classify(
    result: Result<u64, StorageError>,
    missing: &mut usize,
    errored: &mut usize,
) -> Option<i64> {
    match result {
        Ok(size) => Some(size as i64),
        Err(StorageError::NotFound) => {
            *missing += 1;
            None
        }
        Err(e) => {
            tracing::warn!(error = ?e, "storage backfill: cannot stat blob");
            *errored += 1;
            None
        }
    }
}