fix: invalidate stale page analysis when a re-crawl overwrites a page

Deterministic per-page storage keys mean a re-crawl overwrites the image but the
pages row survives the upsert, so analysis keyed to page_id (page_analysis/
search_doc, OCR, auto-tags, content-warnings) stayed stale forever and
force=false re-enqueue never re-analyzed it — poisoning search + the derived
manga_content_warnings (audit DB-1). persist_pages now detects conflict-updated
pages via xmax and clears their four derived tables in the same tx (the
content-warnings DELETE fires the mcw_* triggers), returning them to unanalyzed
so they re-derive. Tested.

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

backend/src/crawler/content.rs

@@ -595,15 +595,22 @@ pub(crate) async fn persist_pages(
     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());
+    // Pages whose row already existed and was overwritten by this re-crawl.
+    // Their image changed but pages.id (and any analysis keyed to it) survived,
+    // so the derived analysis is now stale and must be invalidated below.
+    let mut updated_page_ids: Vec<Uuid> = Vec::new();
     for page in stored {
-        let (id,): (Uuid,) = sqlx::query_as(
+        // `xmax <> 0` on the RETURNING row distinguishes a conflict UPDATE
+        // (existing row overwritten) from a fresh INSERT — the standard upsert
+        // idiom. Cast through text/bigint since there is no direct xid<>int op.
+        let (id, was_update): (Uuid, bool) = sqlx::query_as(
             "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,
                  size_bytes = EXCLUDED.size_bytes
-             RETURNING id",
+             RETURNING id, (xmax::text::bigint <> 0)",
         )
         .bind(chapter_id)
         .bind(page.page_number)
@@ -615,6 +622,32 @@ pub(crate) async fn persist_pages(
         .with_context(|| format!("insert page row {}", page.page_number))?;
         page_ids.push(id);
         page_numbers.push(page.page_number);
+        if was_update {
+            updated_page_ids.push(id);
+        }
+    }
+    // Invalidate stale analysis for re-crawled (overwritten) pages. Storage keys
+    // are deterministic per page number, so the row survives the upsert and its
+    // page_analysis / OCR / auto-tags / content-warnings would otherwise reflect
+    // the OLD image — poisoning search results and the derived
+    // manga_content_warnings. Clearing them (the DELETE on page_content_warnings
+    // fires the mcw_* triggers so manga_content_warnings recomputes) drops the
+    // pages back to "unanalyzed" so the enqueue below — and the admin
+    // only-unanalyzed backfill — re-derive them. Newly inserted pages have no
+    // prior analysis, so they're untouched.
+    if !updated_page_ids.is_empty() {
+        for table in [
+            "page_ocr_text",
+            "page_auto_tags",
+            "page_content_warnings",
+            "page_analysis",
+        ] {
+            sqlx::query(&format!("DELETE FROM {table} WHERE page_id = ANY($1::uuid[])"))
+                .bind(&updated_page_ids)
+                .execute(&mut *tx)
+                .await
+                .with_context(|| format!("invalidate {table} for re-crawled pages"))?;
+        }
     }
     // 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 —
@@ -1200,4 +1233,91 @@ mod tests {
         assert_eq!(fetch_n, 1);
         assert!(format!("{err:#}").contains("nav timeout"));
     }
+
+    #[sqlx::test(migrations = "./migrations")]
+    async fn persist_pages_invalidates_stale_analysis_on_recrawl_update(pool: PgPool) {
+        // Page storage keys are deterministic per page number, so a re-crawl
+        // overwrites the image but keeps pages.id. Analysis keyed to page_id
+        // would otherwise stay stale (old OCR/search_doc/warnings). persist_pages
+        // must clear the derived analysis for updated pages so it re-derives.
+        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 v1 = vec![StoredPage {
+            page_number: 1,
+            storage_key: "k/0001.jpg".into(),
+            content_type: "image/jpeg".into(),
+            size_bytes: 10,
+        }];
+        persist_pages(&pool, chapter_id, &v1, false).await.unwrap();
+        let page_id: Uuid = sqlx::query_scalar(
+            "SELECT id FROM pages WHERE chapter_id = $1 AND page_number = 1",
+        )
+        .bind(chapter_id)
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+        // Simulate a completed prior analysis pass for that page.
+        sqlx::query(
+            "INSERT INTO page_analysis (page_id, status, is_nsfw, analyzed_at) \
+             VALUES ($1, 'done', true, now())",
+        )
+        .bind(page_id)
+        .execute(&pool)
+        .await
+        .unwrap();
+        sqlx::query("INSERT INTO page_ocr_text (page_id, kind, text) VALUES ($1, 'speech', 'stale')")
+            .bind(page_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO page_content_warnings (page_id, warning) VALUES ($1, 'gore')")
+            .bind(page_id)
+            .execute(&pool)
+            .await
+            .unwrap();
+
+        // Re-crawl page 1 with a new image (same deterministic key, new size).
+        let v2 = vec![StoredPage {
+            page_number: 1,
+            storage_key: "k/0001.jpg".into(),
+            content_type: "image/jpeg".into(),
+            size_bytes: 999,
+        }];
+        persist_pages(&pool, chapter_id, &v2, false).await.unwrap();
+
+        // The page row survived (deterministic key).
+        let same_id: Uuid = sqlx::query_scalar(
+            "SELECT id FROM pages WHERE chapter_id = $1 AND page_number = 1",
+        )
+        .bind(chapter_id)
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert_eq!(same_id, page_id, "re-crawl keeps the page id");
+
+        // ...but its stale analysis is cleared so it gets re-derived.
+        for table in ["page_analysis", "page_ocr_text", "page_content_warnings"] {
+            let n: i64 = sqlx::query_scalar(&format!(
+                "SELECT COUNT(*) FROM {table} WHERE page_id = $1"
+            ))
+            .bind(page_id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+            assert_eq!(n, 0, "{table} must be invalidated when the re-crawl updates the page");
+        }
+    }
 }