feat(cleanup): purge misgrabbed gated-post blurred previews (#874 follow-up)

A one-shot Maintenance action to remove the blurred locked-preview images
the ingester downloaded from tier-gated Patreon posts before #874.

current_user_can_view was never persisted, so the cleanup re-walks each
enabled Patreon source (read-only) to re-derive which posts are gated now
and the blurred filehashes Patreon serves for them, then matches by
CONTENT HASH against stored source_filehash. Because the hash is
content-addressed, a real file downloaded when access existed has a
different hash and can never match — regained-then-lost-access content is
provably spared (operator's hard requirement). NULL source_filehash =>
unverifiable, kept + reported.

On apply: delete matched ImageRecords + files (provenance cascades),
clear seen/dead-letter ledger rows for those hashes so the real media
re-ingests if access returns, and delete gated posts left bare. Shares
one match predicate between preview and apply (rule 93).

- cleanup_service: collect_gated_previews + purge_gated_previews
- tasks.admin: purge_gated_previews_task (async re-walk bridge, timeboxed)
- api.admin: POST /maintenance/purge-gated-previews
- GatedPurgeCard.vue in Settings > Maintenance (preview -> confirm -> apply)
- tests: collect predicate, hash-match delete/spare/unverifiable, ledger
  clear, bare-post removal, no-op

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

backend/app/tasks/admin.py

@@ -143,6 +143,106 @@ def dedup_videos_task(self, dry_run: bool = False) -> dict:
         )
 
 
+# Wall-clock budget for the gated-preview re-walk: stop walking new sources past
+# this and report partial (operator re-runs to finish). Sits under the soft limit
+# so the task returns its summary cleanly instead of being SIGKILLed mid-walk.
+_GATED_PURGE_BUDGET_SECONDS = 1500
+
+
+@celery.task(
+    name="backend.app.tasks.admin.purge_gated_previews_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=15, retry_backoff_max=180, max_retries=1,
+    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
+)
+def purge_gated_previews_task(self, dry_run: bool = True) -> dict:
+    """Cleanup (#874 follow-up): purge blurred locked-preview images grabbed from
+    tier-gated Patreon posts before the ingester fix. Re-walks every enabled
+    Patreon source (read-only, no downloads) to re-derive which posts are gated NOW
+    and the blurred filehashes Patreon serves for them, then matches by content
+    hash and (unless dry_run) deletes only those exact files — real content
+    downloaded when access existed has a different hash and is provably spared.
+
+    dry_run=True returns the projection (gated posts / matched images / reclaimable
+    bytes / unverifiable kept) WITHOUT deleting; dry_run=false applies it. The
+    summary lands in task_run.metadata (FC-3i) for the Maintenance card. Time-boxed
+    across sources: a `partial` result means re-run to finish (idempotent — a
+    re-walk re-derives the same matches and already-deleted files stay deleted)."""
+    import asyncio
+    import time as _time
+
+    from sqlalchemy import select as _select
+
+    from ..models import Source
+    from ..services.credential_crypto import CredentialCrypto
+    from ..services.credential_service import CredentialService
+    from ..services.patreon_client import PatreonClient
+    from ..services.patreon_resolver import resolve_campaign_id_for_source
+    from ._async_session import async_session_factory
+
+    key_path = IMAGES_ROOT / "secrets" / "credential_key.b64"
+
+    async def _run() -> dict:
+        async_factory, async_engine = async_session_factory()
+        SessionLocal = _sync_session_factory()
+        try:
+            async with async_factory() as async_session:
+                cred = CredentialService(async_session, CredentialCrypto(key_path))
+                cookies_obj = await cred.get_cookies_path("patreon")
+            cookies_path = str(cookies_obj) if cookies_obj else None
+
+            with SessionLocal() as session:
+                sources = session.execute(
+                    _select(Source.id, Source.url, Source.config_overrides)
+                    .where(Source.platform == "patreon", Source.enabled.is_(True))
+                ).all()
+
+            gated_map: dict[int, dict] = {}
+            scanned = 0
+            partial = False
+            loop = asyncio.get_running_loop()
+            start = _time.monotonic()
+            for sid, url, overrides in sources:
+                if _time.monotonic() - start >= _GATED_PURGE_BUDGET_SECONDS:
+                    partial = True
+                    break
+                campaign_id, _resolved = await resolve_campaign_id_for_source(
+                    url, cookies_path, overrides or {}
+                )
+                if not campaign_id:
+                    log.warning(
+                        "gated-purge: couldn't resolve campaign for source %s (%s) "
+                        "— skipping", sid, url,
+                    )
+                    continue
+                client = PatreonClient(cookies_path)
+                try:
+                    gated = await loop.run_in_executor(
+                        None, cleanup_service.collect_gated_previews, client, campaign_id
+                    )
+                except Exception as exc:  # one bad feed must not strand the sweep
+                    log.warning("gated-purge walk failed for source %s: %s", sid, exc)
+                    continue
+                scanned += 1
+                if gated:
+                    gated_map[int(sid)] = gated
+
+            with SessionLocal() as session:
+                summary = cleanup_service.purge_gated_previews(
+                    session, gated_map=gated_map, images_root=IMAGES_ROOT,
+                    dry_run=dry_run,
+                )
+            summary["sources_scanned"] = scanned
+            summary["sources_total"] = len(sources)
+            summary["partial"] = partial
+            return summary
+        finally:
+            await async_engine.dispose()
+
+    return asyncio.run(_run())
+
+
 @celery.task(
     name="backend.app.tasks.admin.bulk_delete_images_task",
     bind=True,