feat(subscriptions): dry-run backfill preview — B4 preview (plan #708)

Owning the walk lets an operator gauge "is this source worth a backfill?" before
arming one. ingest_core.Ingester.preview walks the first few feed pages and
counts media NOT already in the seen/dead ledgers, downloading nothing
(read-only). download_backends.preview_source resolves the campaign id + runs it
(native-only, mirrors verify_source_credential / run_download); POST
/api/sources/{id}/preview returns {total_new, posts_scanned, has_more, sample[]}
(409 on auth/drift/unresolvable, 400 for gallery-dl platforms). PatreonClient
gains post_meta(post) for the sample's title/date.

UI: a Patreon-only Preview button (mdi-eye-outline) on SourceRow + SourceCard
opens PreviewDialog — self-fetches with loading / error / empty / result states
and a "Start backfill" shortcut. Store action previewSource.

Tests: preview counts new media without downloading + samples only posts with
new items; page_limit caps the walk + flags has_more.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/api/sources.py

@@ -191,6 +191,50 @@ async def set_backfill(source_id: int):
     return jsonify(record.to_dict())
 
 
+@sources_bp.route("/<int:source_id>/preview", methods=["POST"])
+async def preview_source_endpoint(source_id: int):
+    """Plan #708 B4: dry-run — count what a backfill WOULD download for a native
+    platform (Patreon today), without downloading. Walks the first few feed pages
+    and counts media not already in the seen/dead ledgers. Returns
+    {total_new, posts_scanned, pages_scanned, has_more, sample[]} or 409 + reason
+    (unresolvable campaign id / auth / drift). 400 for gallery-dl platforms (no
+    cheap dry-run — their verify is a slow --simulate)."""
+    from pathlib import Path
+
+    from ..services.credential_service import CredentialService
+    from ..services.download_backends import preview_source, uses_native_ingester
+    from ..tasks._sync_engine import sync_session_factory
+    from .credentials import _get_crypto
+
+    async with get_session() as session:
+        rec = await SourceService(session).get(source_id)
+        if rec is None:
+            return _bad("not_found", status=404)
+        if not uses_native_ingester(rec.platform):
+            return _bad(
+                "unsupported",
+                detail="Preview is only available for native-ingester platforms.",
+                status=400,
+            )
+        cred = CredentialService(session, _get_crypto())
+        cookies_path = await cred.get_cookies_path(rec.platform)
+
+    # The walk + ledger reads are sync (run off the request loop); the process
+    # sync engine is the same one the download task uses.
+    result = await preview_source(
+        platform=rec.platform,
+        url=rec.url,
+        source_id=source_id,
+        config_overrides=rec.config_overrides or {},
+        cookies_path=str(cookies_path) if cookies_path else None,
+        images_root=Path("/images"),
+        sync_session_factory=sync_session_factory(),
+    )
+    if "error" in result:
+        return _bad("preview_failed", detail=result["error"], status=409)
+    return jsonify(result)
+
+
 @sources_bp.route("/<int:source_id>/check", methods=["POST"])
 async def check_source(source_id: int):
     """FC-3c: enqueue a download for this source.

backend/app/services/download_backends.py

@@ -144,6 +144,54 @@ async def _run_native_ingester(
     return dl_result, resolved_campaign_id
 
 
+async def preview_source(
+    *,
+    platform: str,
+    url: str,
+    source_id: int,
+    config_overrides: dict | None,
+    cookies_path: str | None,
+    images_root: Path,
+    sync_session_factory,
+    page_limit: int = 3,
+) -> dict:
+    """Dry-run preview for a native platform (plan #708 B4): resolve the campaign
+    id, then walk a few pages counting media not already seen/dead — no download.
+
+    Returns the preview dict (total_new / posts_scanned / pages_scanned /
+    has_more / sample), or `{"error": msg}` on a resolve / auth / drift failure.
+    Native-only — the caller gates on `uses_native_ingester`.
+    """
+    import asyncio
+
+    from .patreon_client import PatreonAPIError
+
+    campaign_id, _ = await resolve_campaign_id_for_source(
+        url, cookies_path, config_overrides or {}
+    )
+    if not campaign_id:
+        return {
+            "error": (
+                "Couldn't resolve the campaign id from the source URL "
+                "(cookies expired, or the creator moved/renamed?)."
+            )
+        }
+    ingester = PatreonIngester(
+        images_root=images_root,
+        cookies_path=cookies_path,
+        session_factory=sync_session_factory,
+    )
+    loop = asyncio.get_running_loop()
+    try:
+        result = await loop.run_in_executor(
+            None,
+            lambda: ingester.preview(source_id, campaign_id, page_limit=page_limit),
+        )
+    except PatreonAPIError as exc:
+        return {"error": f"Couldn't preview: {exc}"}
+    return result
+
+
 async def verify_source_credential(
     *,
     platform: str,

backend/app/services/ingest_core.py

@@ -361,6 +361,67 @@ class Ingester:
             error_type=None, error_message=None,
         )
 
+    # -- preview (dry-run) -------------------------------------------------
+
+    def preview(
+        self,
+        source_id: int,
+        campaign_id: str,
+        *,
+        page_limit: int = 3,
+        sample_size: int = 10,
+    ) -> dict:
+        """Dry-run (plan #708 B4): walk up to `page_limit` pages and count media
+        NOT already in the seen/dead ledgers, WITHOUT downloading anything.
+
+        Read-only — only the seen/dead SELECTs touch the DB (short sessions). Lets
+        an operator gauge "is this source worth a backfill?" cheaply. Returns:
+            {total_new, posts_scanned, pages_scanned, has_more,
+             sample: [{title, date, new}, ...]}  # sample = posts with new media
+        A client-level failure (auth/drift) propagates to the caller.
+        """
+        total_new = 0
+        posts_scanned = 0
+        pages_scanned = 0
+        has_more = False
+        sample: list[dict] = []
+        unset = object()
+        last_page: object = unset
+        for post, included, page_cursor in self.client.iter_posts(
+            campaign_id, cursor=None
+        ):
+            if page_cursor != last_page:
+                last_page = page_cursor
+                pages_scanned += 1
+                if pages_scanned > page_limit:
+                    has_more = True
+                    pages_scanned = page_limit
+                    break
+            posts_scanned += 1
+            media = self.client.extract_media(post, included)
+            if not media:
+                continue
+            keys = [self._ledger_key(m) for m in media]
+            skip = self._seen_keys(source_id, keys) | self._dead_keys(source_id, keys)
+            new_count = sum(1 for m in media if self._ledger_key(m) not in skip)
+            total_new += new_count
+            if new_count > 0 and len(sample) < sample_size:
+                meta = self.client.post_meta(post)
+                sample.append(
+                    {
+                        "title": meta.get("title") or "(untitled)",
+                        "date": meta.get("date"),
+                        "new": new_count,
+                    }
+                )
+        return {
+            "total_new": total_new,
+            "posts_scanned": posts_scanned,
+            "pages_scanned": pages_scanned,
+            "has_more": has_more,
+            "sample": sample,
+        }
+
     # -- failure mapping (adapter overrides) -------------------------------
 
     def _failure_result(self, exc: Exception, _result) -> DownloadResult:

backend/app/services/patreon_client.py

@@ -520,6 +520,18 @@ class PatreonClient:
 
         return _dedup_by_filehash(items)
 
+    @staticmethod
+    def post_meta(post: dict) -> dict:
+        """Title + published date for a post — for the preview sample (plan #708
+        B4). Part of the client contract `ingest_core.Ingester.preview` calls."""
+        attrs = post.get("attributes") or {}
+        title = attrs.get("title")
+        published = attrs.get("published_at")
+        return {
+            "title": title if isinstance(title, str) else None,
+            "date": published if isinstance(published, str) else None,
+        }
+
     # -- iteration ---------------------------------------------------------
 
     def iter_posts(