feat(storage): admin storage-usage stats + per-manga/chapter sizes

Persist blob sizes (covers + chapter pages) and surface upload storage
usage to admins in two places:

- Admin System tab: total/covers/chapters totals, ratio of disk, average
  image sizes, and top-5 largest mangas/chapters leaderboards, behind a
  new GET /api/v1/admin/storage endpoint (separate from /admin/system so
  the cheap DB aggregates aren't coupled to its 250ms CPU sample).
- Manga detail page: total chapter-content size and per-chapter size,
  with an em-dash for uncrawled or not-yet-measured chapters.

Sizes are captured at write time (crawler put_stream return, uploaded
page/cover byte length) into nullable pages.size_bytes /
mangas.cover_size_bytes columns; NULL means "not yet measured", distinct
from a real 0. A one-shot, idempotent admin "Backfill sizes" action
(POST /api/v1/admin/storage/backfill) stats pre-existing blobs in
keyset-paginated, capped batches and reports more_remaining so a large
legacy library can be drained over multiple runs.

Aggregates and leaderboards treat NULL honestly (only fully-measured
entities are ranked); a dashboard banner flags unmeasured rows so partial
figures aren't read as complete. persist_pages now prunes stale page rows
on a shrinking re-crawl so totals and page_count stay consistent.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/src/crawler/content.rs

@@ -334,6 +334,9 @@ pub(crate) struct StoredPage {
     page_number: i32,
     storage_key: String,
     content_type: String,
+    /// Bytes written to storage, captured from `put_stream`'s return so
+    /// storage-usage stats are a pure DB SUM.
+    size_bytes: i64,
 }
 
 /// Bytes accumulated for content-type sniffing. `infer` only needs the
@@ -437,7 +440,7 @@ async fn download_and_store_page(
         )))),
     });
     let combined = prefix_stream.chain(rest_stream);
-    storage
+    let size_bytes = storage
         .put_stream(&key, Box::pin(combined))
         .await
         .with_context(|| format!("put_stream {key}"))?;
@@ -445,6 +448,7 @@ async fn download_and_store_page(
         page_number: img.page_number,
         storage_key: key,
         content_type: format!("image/{ext}"),
+        size_bytes: size_bytes as i64,
     })
 }
 
@@ -458,24 +462,40 @@ pub(crate) async fn persist_pages(
 ) -> anyhow::Result<()> {
     let mut tx = db.begin().await.context("open chapter sync tx")?;
     let mut page_ids: Vec<Uuid> = Vec::with_capacity(stored.len());
+    let mut page_numbers: Vec<i32> = Vec::with_capacity(stored.len());
     for page in stored {
         let (id,): (Uuid,) = sqlx::query_as(
-            "INSERT INTO pages (chapter_id, page_number, storage_key, content_type)
-             VALUES ($1, $2, $3, $4)
+            "INSERT INTO pages (chapter_id, page_number, storage_key, content_type, size_bytes)
+             VALUES ($1, $2, $3, $4, $5)
              ON CONFLICT (chapter_id, page_number) DO UPDATE
              SET storage_key = EXCLUDED.storage_key,
-                 content_type = EXCLUDED.content_type
+                 content_type = EXCLUDED.content_type,
+                 size_bytes = EXCLUDED.size_bytes
              RETURNING id",
         )
         .bind(chapter_id)
         .bind(page.page_number)
         .bind(&page.storage_key)
         .bind(&page.content_type)
+        .bind(page.size_bytes)
         .fetch_one(&mut *tx)
         .await
         .with_context(|| format!("insert page row {}", page.page_number))?;
         page_ids.push(id);
+        page_numbers.push(page.page_number);
     }
+    // Drop any rows left over from a prior, larger crawl of this chapter
+    // (re-crawl that now yields fewer pages). Without this the stale rows —
+    // and their stale size_bytes — would inflate the storage aggregates
+    // and the page_count would disagree with the row count. Cascades to
+    // collection_pages / page_tags by design (re-upload drops saved-page
+    // references).
+    sqlx::query("DELETE FROM pages WHERE chapter_id = $1 AND page_number <> ALL($2::int[])")
+        .bind(chapter_id)
+        .bind(&page_numbers)
+        .execute(&mut *tx)
+        .await
+        .context("prune stale page rows")?;
     sqlx::query("UPDATE chapters SET page_count = $1 WHERE id = $2")
         .bind(stored.len() as i32)
         .bind(chapter_id)
@@ -576,11 +596,13 @@ mod tests {
                 page_number: 1,
                 storage_key: "k/0001.jpg".into(),
                 content_type: "image/jpeg".into(),
+                size_bytes: 111,
             },
             StoredPage {
                 page_number: 2,
                 storage_key: "k/0002.jpg".into(),
                 content_type: "image/jpeg".into(),
+                size_bytes: 222,
             },
         ];
         persist_pages(&pool, chapter_id, &stored, false).await.unwrap();
@@ -599,6 +621,15 @@ mod tests {
                 .await
                 .unwrap();
         assert_eq!(rows, 2);
+        // Sizes are persisted from the StoredPage so SUM(size_bytes) is
+        // the chapter's storage usage.
+        let total: i64 =
+            sqlx::query_scalar("SELECT COALESCE(SUM(size_bytes),0)::bigint FROM pages WHERE chapter_id = $1")
+                .bind(chapter_id)
+                .fetch_one(&pool)
+                .await
+                .unwrap();
+        assert_eq!(total, 333);
 
         // Idempotent re-run (force refetch path): same rows, page_count stable.
         persist_pages(&pool, chapter_id, &stored, false).await.unwrap();
@@ -611,6 +642,61 @@ mod tests {
         assert_eq!(rows2, 2, "re-run is idempotent via ON CONFLICT");
     }
 
+    #[sqlx::test(migrations = "./migrations")]
+    async fn persist_pages_prunes_stale_rows_on_smaller_recrawl(pool: PgPool) {
+        // A re-crawl that yields fewer pages must drop the leftover
+        // high-numbered rows so page_count, the row count, and the storage
+        // aggregates stay consistent.
+        let manga_id = Uuid::new_v4();
+        let chapter_id = Uuid::new_v4();
+        sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')")
+            .bind(manga_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)")
+            .bind(chapter_id)
+            .bind(manga_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+
+        let three = vec![
+            StoredPage { page_number: 1, storage_key: "k/1.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 10 },
+            StoredPage { page_number: 2, storage_key: "k/2.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 20 },
+            StoredPage { page_number: 3, storage_key: "k/3.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 30 },
+        ];
+        persist_pages(&pool, chapter_id, &three, false).await.unwrap();
+
+        // Re-crawl now yields only 2 pages.
+        let two = vec![
+            StoredPage { page_number: 1, storage_key: "k/1.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 11 },
+            StoredPage { page_number: 2, storage_key: "k/2.jpg".into(), content_type: "image/jpeg".into(), size_bytes: 22 },
+        ];
+        persist_pages(&pool, chapter_id, &two, false).await.unwrap();
+
+        let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1")
+            .bind(chapter_id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(rows, 2, "stale page 3 should be pruned");
+        let page_count: i32 = sqlx::query_scalar("SELECT page_count FROM chapters WHERE id = $1")
+            .bind(chapter_id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(page_count, 2);
+        let total: i64 = sqlx::query_scalar(
+            "SELECT COALESCE(SUM(size_bytes),0)::bigint FROM pages WHERE chapter_id = $1",
+        )
+        .bind(chapter_id)
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert_eq!(total, 33, "size reflects only the 2 re-crawled pages");
+    }
+
     #[sqlx::test(migrations = "./migrations")]
     async fn persist_pages_enqueues_analysis_only_when_flag_set(pool: PgPool) {
         let manga_id = Uuid::new_v4();
@@ -630,6 +716,7 @@ mod tests {
             page_number: 1,
             storage_key: "k/0001.jpg".into(),
             content_type: "image/jpeg".into(),
+            size_bytes: 100,
         }];
 
         // Flag off: no analyze_page jobs.

backend/src/crawler/pipeline.rs

@@ -768,7 +768,7 @@ pub(crate) async fn download_and_store_cover(
         .put(&key, &bytes)
         .await
         .with_context(|| format!("store cover at {key}"))?;
-    repo::manga::set_cover_image_path(db, manga_id, &key)
+    repo::manga::set_cover_image_path(db, manga_id, &key, bytes.len() as i64)
         .await
         .with_context(|| format!("update cover_image_path for {manga_id}"))?;
     tracing::info!(