refactor(ingest): extract platform-agnostic Ingester core — roadmap #9 (plan #706)

Factor the native-ingest orchestration out of PatreonIngester into a reusable
ingest_core.Ingester base, parametrized by client/downloader/ledger-models/
constraints/key/platform/error_base. PatreonIngester becomes a thin adapter:
it resolves the Patreon client/downloader, wires the seen/dead-letter models +
UNIQUE-constraint names + _ledger_key into super().__init__, and overrides
_failure_result with the Patreon exception taxonomy. Behavior-preserving — no
table rename, no migration; the public surface (PatreonIngester, _ledger_key,
DEAD_LETTER_THRESHOLD, verify_patreon_credential) is unchanged.

This is the strategic seam: SubscribeStar/etc. now migrate by writing a
~40-line adapter, not by re-implementing the tick/backfill/recovery walk,
tiered skip, checkpoint, and dead-letter logic.

run() moved to ingest_core, so the budget test's monotonic patch repoints to
ingest_core.time.

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

backend/app/services/ingest_core.py

@@ -0,0 +1,459 @@
+"""Platform-agnostic native-ingest core (plan #706, build on #697/#703/#704/#705).
+
+The orchestration that drives a native subscription walk — page a feed →
+extract media → tiered skip (seen-ledger / on-disk / dead-letter) → download →
+mark-seen / record-failures / checkpoint-cursor → return a gallery-dl-shaped
+`DownloadResult`, across tick/backfill/recovery modes — is identical for every
+platform. Only four things are platform-specific, and they're INJECTED at
+construction by a thin adapter (e.g. `PatreonIngester`):
+
+  - `client`    — `.iter_posts(feed_id, cursor)` yielding `(post, included,
+                  page_cursor)` + `.extract_media(post, included) -> [media]`.
+  - `downloader`— `.download_post(post, media, artist_slug, is_seen) ->
+                  [MediaOutcome]` (status in downloaded/skipped_seen/skipped_disk
+                  /quarantined/error; `.path`/`.error`/`.post_id`).
+  - ledger      — `seen_model` + `failed_model` SQLAlchemy models (+ their
+                  on-conflict UNIQUE constraint names) and a `ledger_key(media)`.
+  - failure map — the adapter overrides `_failure_result` (platform exception
+                  → DownloadResult.error_type) and supplies `error_base` (the
+                  exception type the walk catches) + `platform` (result label).
+
+Everything DB touches a SHORT-LIVED sync session from the injected sessionmaker —
+never held across a network fetch ([[db-connection-held-across-subprocess]]).
+Plain-HTTP homelab: no secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Callable
+
+from sqlalchemy import delete, func, select, text
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+from .gallery_dl import DownloadResult, ErrorType
+
+log = logging.getLogger(__name__)
+
+# Stop a tick after this many CONTIGUOUS already-have-it media (seen-ledger or
+# on-disk) — the cheap native equivalent of gallery-dl's `exit:20`, now free of
+# per-file HEADs. Headroom against paywalled/undownloadable items interleaving.
+_TICK_SEEN_THRESHOLD = 20
+
+# plan #705 #7: after this many failed download/validate attempts a media is
+# "dead-lettered" and skipped on routine tick/backfill walks (recovery still
+# re-attempts it). Stops a permanently-broken media re-erroring forever.
+DEAD_LETTER_THRESHOLD = 3
+# last_error is Text but bound it so a giant traceback doesn't bloat the row.
+_ERROR_MAX = 1000
+
+
+class Ingester:
+    """Generic native-ingest orchestration. Subclass with a platform adapter
+    (see the module docstring) — or construct directly with the keyword seams."""
+
+    def __init__(
+        self,
+        *,
+        client,
+        downloader,
+        session_factory: Callable[[], object],
+        seen_model,
+        failed_model,
+        seen_constraint: str,
+        failed_constraint: str,
+        ledger_key: Callable[[object], str],
+        platform: str,
+        error_base: type[Exception],
+    ):
+        self.client = client
+        self.downloader = downloader
+        self.session_factory = session_factory
+        self._seen_model = seen_model
+        self._failed_model = failed_model
+        self._seen_constraint = seen_constraint
+        self._failed_constraint = failed_constraint
+        self._ledger_key = ledger_key
+        self._platform = platform
+        self._error_base = error_base
+
+    # -- public ------------------------------------------------------------
+
+    def run(
+        self,
+        *,
+        source_id: int,
+        campaign_id: str,
+        artist_slug: str,
+        url: str,
+        mode: str,
+        resume_cursor: str | None = None,
+        time_budget_seconds: float = 870.0,
+        seen_threshold: int = _TICK_SEEN_THRESHOLD,
+    ) -> DownloadResult:
+        """Walk + download for one source, returning a gallery-dl-shaped result.
+
+        `mode` is "tick" | "backfill" | "recovery". Recovery bypasses the tier-1
+        seen-ledger AND the dead-letter ledger (tier-2 disk still skips kept
+        files). The walk stops on:
+          - budget exhaustion (time_budget_seconds)         → TIMEOUT / PARTIAL
+          - tick early-out (seen_threshold contiguous seen)  → success
+          - reaching the bottom of the feed                  → success (rc 0)
+        A client-level failure (drift / auth / network) fails the whole run loud.
+        """
+        bypass_seen = mode == "recovery"
+        # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a
+        # tick has no resumable backfill state.
+        checkpoint = mode in ("backfill", "recovery")
+        ledger_key = self._ledger_key
+        start = time.monotonic()
+        log_lines: list[str] = []
+        written: list[str] = []
+        quarantined_paths: list[str] = []
+        downloaded = 0
+        errors = 0
+        quarantined = 0
+        dead_lettered = 0
+        skipped_count = 0
+        posts_processed = 0
+        consecutive_seen = 0
+        emitted_cursor: str | None = None
+        reached_bottom = False
+        budget_hit = False
+        early_out = False
+
+        def _result(
+            *, success: bool, return_code: int,
+            error_type: ErrorType | None, error_message: str | None,
+        ) -> DownloadResult:
+            # plan #704: return STRUCTURED data — phase 3 reads run_stats/cursor
+            # directly instead of regex-scraping a reconstructed stdout. stdout
+            # stays a human-readable summary (no fake `Cursor:` lines).
+            return DownloadResult(
+                success=success,
+                url=url,
+                artist_slug=artist_slug,
+                platform=self._platform,
+                files_downloaded=downloaded,
+                files_quarantined=quarantined,
+                quarantined_paths=list(quarantined_paths),
+                written_paths=written,
+                stdout="\n".join(log_lines),
+                stderr="",
+                return_code=return_code,
+                error_type=error_type,
+                error_message=error_message,
+                duration_seconds=time.monotonic() - start,
+                cursor=emitted_cursor,
+                posts_processed=posts_processed,
+                run_stats={
+                    "exit_code": return_code,
+                    "downloaded_count": downloaded,
+                    "skipped_count": skipped_count,
+                    "per_item_failures": errors,
+                    "warning_count": 0,
+                    "tier_gated_count": 0,
+                    "quarantined_count": quarantined,
+                    "dead_lettered_count": dead_lettered,
+                },
+            )
+
+        try:
+            for post, included, page_cursor in self.client.iter_posts(
+                campaign_id, cursor=resume_cursor
+            ):
+                # Checkpoint the cursor that FETCHED this page the moment we
+                # START it — so a chunk cut mid-page resumes the page, not the one
+                # after it. Carried as DownloadResult.cursor (plan #704).
+                if page_cursor and page_cursor != emitted_cursor:
+                    emitted_cursor = page_cursor
+                    # plan #705 #6: persist the cursor at each page boundary so a
+                    # worker SIGKILL mid-chunk resumes near the crash, not the
+                    # chunk start. (phase 3 still writes the final cursor — same
+                    # value; this is the crash-safety net.)
+                    if checkpoint:
+                        self._checkpoint_cursor(source_id, emitted_cursor)
+
+                # Time-box check at the post boundary (coarse, like a gallery-dl
+                # chunk). Backfill/recovery resume from emitted_cursor next chunk.
+                if time.monotonic() - start >= time_budget_seconds:
+                    budget_hit = True
+                    break
+
+                posts_processed += 1
+                media = self.client.extract_media(post, included)
+                if not media:
+                    continue
+
+                keys = [ledger_key(m) for m in media]
+                # Recovery bypasses BOTH the seen-ledger AND the dead-letter
+                # ledger (the operator's "try everything again"); routine walks
+                # skip seen + dead media (tier-1 + tier-1.5, plan #705 #7).
+                dead = set() if bypass_seen else self._dead_keys(source_id, keys)
+                seen = (
+                    set()
+                    if bypass_seen
+                    else self._seen_keys(source_id, keys)
+                )
+                skip = seen | dead
+
+                def _is_skip(m, _skip=skip) -> bool:
+                    return ledger_key(m) in _skip
+
+                outcomes = self.downloader.download_post(
+                    post, media, artist_slug, is_seen=_is_skip
+                )
+
+                to_mark: list[tuple[str, str]] = []
+                to_clear: list[str] = []  # recovered → drop any dead-letter row
+                to_fail: list[tuple[str, str, str]] = []  # (key, post_id, error)
+                for media_item, outcome in zip(media, outcomes, strict=False):
+                    key = ledger_key(media_item)
+                    if key in dead:
+                        dead_lettered += 1  # skipped because previously dead
+                    if outcome.status == "downloaded":
+                        downloaded += 1
+                        if outcome.path is not None:
+                            written.append(str(outcome.path))
+                        to_mark.append((key, media_item.post_id))
+                        to_clear.append(key)
+                        consecutive_seen = 0
+                    elif outcome.status == "skipped_disk":
+                        # Already on disk (a prior run). Reconcile the ledger so a
+                        # later tick skips it at tier-1 without a disk stat, but
+                        # do NOT re-feed it to phase 3 — attach_in_place would see
+                        # the duplicate sha256 and unlink the on-disk copy.
+                        to_mark.append((key, media_item.post_id))
+                        to_clear.append(key)
+                        skipped_count += 1
+                        consecutive_seen += 1
+                    elif outcome.status == "skipped_seen":
+                        skipped_count += 1
+                        consecutive_seen += 1
+                    elif outcome.status == "quarantined":
+                        # New content that failed validation (corrupt) — counted
+                        # distinctly so the run surfaces a real quarantined total.
+                        # Not marked seen (a later walk may re-fetch a fixed file);
+                        # it IS new content, so it breaks the run-of-seen. Counts
+                        # toward the dead-letter ledger (plan #705 #7).
+                        quarantined += 1
+                        if outcome.path is not None:
+                            quarantined_paths.append(str(outcome.path))
+                        to_fail.append((key, media_item.post_id, outcome.error or "quarantined"))
+                        consecutive_seen = 0
+                    elif outcome.status == "error":
+                        errors += 1
+                        to_fail.append((key, media_item.post_id, outcome.error or "error"))
+                        # An error neither advances nor resets the run-of-seen.
+
+                    if mode == "tick" and consecutive_seen >= seen_threshold:
+                        early_out = True
+                        break
+
+                # Persist ledger changes AFTER the network fetch, on short
+                # sessions: mark downloaded/on-disk seen, clear any dead-letter
+                # for recovered media, and record failures (plan #705 #7).
+                if to_mark:
+                    self._mark_seen(source_id, to_mark)
+                if to_clear:
+                    self._clear_failures(source_id, to_clear)
+                if to_fail:
+                    self._record_failures(source_id, to_fail)
+
+                if early_out:
+                    break
+            else:
+                reached_bottom = True
+        except self._error_base as exc:
+            # The platform's client-error base — _failure_result (adapter)
+            # maps it to a typed error.
+            return self._failure_result(exc, _result)
+
+        if errors:
+            log_lines.append(f"{errors} media item(s) failed")
+        if quarantined:
+            log_lines.append(f"{quarantined} media item(s) quarantined (invalid)")
+        if dead_lettered:
+            log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)")
+        log_lines.append(
+            f"{self._platform} ingest ({mode}): {downloaded} downloaded, "
+            f"{skipped_count} skipped, {quarantined} quarantined, "
+            f"{dead_lettered} dead-lettered, {errors} error(s), "
+            f"{posts_processed} post(s)"
+            + (", reached end" if reached_bottom else "")
+            + (", time-boxed" if budget_hit else "")
+        )
+
+        if budget_hit:
+            # A chunk that hit its time-box but made forward progress is a
+            # NORMAL chunk boundary, not a failure (PARTIAL → status "ok"); the
+            # next chunk resumes from the emitted cursor. No progress → TIMEOUT,
+            # which feeds download_service's backfill stall-guard. rc<0 mirrors
+            # subprocess TimeoutExpired so completion detection stays false.
+            made_progress = downloaded > 0 or emitted_cursor != resume_cursor
+            if made_progress:
+                return _result(
+                    success=False, return_code=-1,
+                    error_type=ErrorType.PARTIAL,
+                    error_message=(
+                        f"Backfill chunk: {downloaded} file(s) — continuing"
+                    ),
+                )
+            return _result(
+                success=False, return_code=-1,
+                error_type=ErrorType.TIMEOUT,
+                error_message="Chunk timed out with no progress",
+            )
+
+        # Normal success: reached the bottom, or a tick that early-outed. rc 0 +
+        # error_type None is REQUIRED for a backfill/recovery walk that reached
+        # the bottom to be marked COMPLETE by
+        # download_service._apply_backfill_lifecycle — so we return None even
+        # when downloaded == 0 (a re-confirming walk that found nothing new still
+        # completed). success=True maps to status "ok" regardless. A tick that
+        # early-outed also returns here; ticks never set backfill state so the
+        # lifecycle is a no-op for them.
+        return _result(
+            success=True, return_code=0,
+            error_type=None, error_message=None,
+        )
+
+    # -- failure mapping (adapter overrides) -------------------------------
+
+    def _failure_result(self, exc: Exception, _result) -> DownloadResult:
+        """Map a platform client-error to a typed failed DownloadResult. The base
+        gives a safe default; adapters override with their exception taxonomy."""
+        log.warning("%s ingest failed: %s", self._platform, exc)
+        return _result(
+            success=False, return_code=1,
+            error_type=ErrorType.UNKNOWN_ERROR, error_message=str(exc),
+        )
+
+    # -- seen-ledger (short-lived sessions) --------------------------------
+
+    def _seen_keys(self, source_id: int, keys: list[str]) -> set[str]:
+        """Which of `keys` are already in the seen-ledger for this source.
+
+        One short SELECT on its own session — opened and closed without any
+        network in between (the GETs happen after, in download_post).
+        """
+        if not keys:
+            return set()
+        with self.session_factory() as session:
+            rows = session.execute(
+                select(self._seen_model.filehash).where(
+                    self._seen_model.source_id == source_id,
+                    self._seen_model.filehash.in_(keys),
+                )
+            ).scalars().all()
+        return set(rows)
+
+    def _checkpoint_cursor(self, source_id: int, cursor: str) -> None:
+        """Persist the in-progress backfill cursor mid-walk (plan #705 #6).
+
+        ATOMIC, single-key UPDATE: cast the JSON column to jsonb, set just
+        `_backfill_cursor`, cast back — so it never clobbers operator config or
+        the other backfill keys (no read-modify-write race). The in-flight guard
+        means only this source's one download runs at a time; a concurrent
+        operator stop is benign (a stray cursor with no `_backfill_state` is
+        ignored by tick mode and cleared on the next start).
+        """
+        with self.session_factory() as session:
+            session.execute(
+                text(
+                    "UPDATE source SET config_overrides = jsonb_set("
+                    "  coalesce(config_overrides::jsonb, '{}'::jsonb),"
+                    "  '{_backfill_cursor}', to_jsonb(cast(:cur AS text))"
+                    ")::json WHERE id = :sid"
+                ),
+                {"cur": cursor, "sid": source_id},
+            )
+            session.commit()
+
+    def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None:
+        """Idempotent upsert of (filehash, post_id) seen-ledger rows for a page.
+
+        ON CONFLICT DO NOTHING against the (source_id, filehash) UNIQUE so a
+        re-sighting — or a concurrent walk — is a harmless no-op
+        ([[scalar_one_or_none-duplicates]]: never check-then-insert without the
+        DB constraint backing it). De-dup the batch locally first so a single
+        page can't present the same key twice to one INSERT.
+        """
+        seen_local: set[str] = set()
+        values = []
+        for key, post_id in items:
+            if key in seen_local:
+                continue
+            seen_local.add(key)
+            values.append(
+                {"source_id": source_id, "filehash": key, "post_id": post_id}
+            )
+        if not values:
+            return
+        with self.session_factory() as session:
+            stmt = pg_insert(self._seen_model).values(values)
+            stmt = stmt.on_conflict_do_nothing(constraint=self._seen_constraint)
+            session.execute(stmt)
+            session.commit()
+
+    # -- dead-letter ledger (plan #705 #7) ---------------------------------
+
+    def _dead_keys(self, source_id: int, keys: list[str]) -> set[str]:
+        """Which of `keys` have failed >= DEAD_LETTER_THRESHOLD times (dead).
+        One short SELECT; recovery never calls this (it re-attempts dead media)."""
+        if not keys:
+            return set()
+        with self.session_factory() as session:
+            rows = session.execute(
+                select(self._failed_model.filehash).where(
+                    self._failed_model.source_id == source_id,
+                    self._failed_model.filehash.in_(keys),
+                    self._failed_model.attempts >= DEAD_LETTER_THRESHOLD,
+                )
+            ).scalars().all()
+        return set(rows)
+
+    def _record_failures(
+        self, source_id: int, items: list[tuple[str, str, str]]
+    ) -> None:
+        """Upsert-increment the dead-letter ledger for failed media. On conflict
+        bump `attempts` and refresh last_error/last_failed_at (UNIQUE backs the
+        upsert — no check-then-insert). De-dup the batch (one row/key, last error
+        wins)."""
+        by_key: dict[str, str] = {}
+        for key, _post_id, err in items:
+            by_key[key] = (err or "")[:_ERROR_MAX]
+        if not by_key:
+            return
+        values = [
+            {"source_id": source_id, "filehash": k, "attempts": 1, "last_error": e}
+            for k, e in by_key.items()
+        ]
+        with self.session_factory() as session:
+            stmt = pg_insert(self._failed_model).values(values)
+            stmt = stmt.on_conflict_do_update(
+                constraint=self._failed_constraint,
+                set_={
+                    "attempts": self._failed_model.attempts + 1,
+                    "last_error": stmt.excluded.last_error,
+                    "last_failed_at": func.now(),
+                },
+            )
+            session.execute(stmt)
+            session.commit()
+
+    def _clear_failures(self, source_id: int, keys: list[str]) -> None:
+        """Drop dead-letter rows for media that just downloaded cleanly — they
+        recovered. A no-op DELETE for keys that were never failing."""
+        unique = list(dict.fromkeys(keys))
+        if not unique:
+            return
+        with self.session_factory() as session:
+            session.execute(
+                delete(self._failed_model).where(
+                    self._failed_model.source_id == source_id,
+                    self._failed_model.filehash.in_(unique),
+                )
+            )
+            session.commit()

backend/app/services/patreon_ingester.py

@@ -1,13 +1,14 @@
-"""Native Patreon ingester — phase-2 orchestrator (build step 3).
+"""Native Patreon ingester — the Patreon ADAPTER over the platform-agnostic core.
 
-Ties build steps 1 (`patreon_client`) and 2 (`patreon_downloader` +
+The orchestration that drives a native subscription walk (page a feed → extract
-`patreon_seen_media`) together into a single sync walk that REPLACES the
+media → tiered skip → download → mark-seen / record-failures / checkpoint-cursor
-gallery-dl subprocess for Patreon. `download_service.download_source` calls
+→ return a gallery-dl-shaped `DownloadResult`, across tick/backfill/recovery)
-`PatreonIngester.run(...)` from phase 2 (in a thread, via run_in_executor, since
+now lives in `ingest_core.Ingester` — it's identical for every platform. This
-everything here is sync `requests`/`subprocess`) and gets back a
+module is the thin Patreon adapter: it wires the Patreon `client`/`downloader`/
-`DownloadResult` — the exact shape gallery-dl returns — so phase 1 (DB setup)
+ledger models/constraints/key into the core and supplies the Patreon-specific
-and phase 3 (import → pHash dedup → thumbnails → ML) are untouched and cannot
+failure mapping. `download_service.download_source` calls `PatreonIngester.run`
-tell the difference.
+exactly as before; the public surface (this class, `_ledger_key`,
+`DEAD_LETTER_THRESHOLD`, `verify_patreon_credential`) is unchanged.
 
 Three modes (selected by `download_service` from `config_overrides` state):
   - tick      — newest→oldest, skip seen (tier-1 ledger + tier-2 disk), early-out
@@ -15,24 +16,14 @@ Three modes (selected by `download_service` from `config_overrides` state):
                 equivalent of gallery-dl's `exit:20`, now free of per-file HEADs).
   - backfill  — full-history walk in a time-boxed chunk, resuming from the
                 pagination cursor checkpoint; reaches the bottom → "complete".
-  - recovery  — like backfill but BYPASSES the tier-1 seen-ledger, so
+  - recovery  — like backfill but BYPASSES the tier-1 seen-ledger AND the
-                deliberately-dropped-and-deleted near-dups get re-fetched and
+                dead-letter ledger, so deliberately-dropped-and-deleted near-dups
-                re-evaluated under the current pHash threshold (tier-2 disk skip
+                get re-fetched and re-evaluated under the current pHash threshold
-                still spares files we kept). Triggered by the same #693 backfill
+                (tier-2 disk skip still spares files we kept).
-                state machine plus the `_backfill_bypass_seen` flag, so the whole
-                cursor/chunk/complete/stall lifecycle is reused verbatim.
-
-Cursor contract: the ingester emits gallery-dl-style ``Cursor: <token>`` lines
-into the returned `stdout`, one per page it walks. That is exactly what
-`download_service`'s existing backfill lifecycle reads via
-`parse_last_cursor(stdout, stderr)` — so checkpointing, the TIMEOUT→PARTIAL
-reclassification, and completion detection all keep working unchanged.
-
-The seen-ledger lives in Postgres (`patreon_seen_media`). The ingester opens a
-SHORT-LIVED sync session per page batch (via an injected sessionmaker) — never
-held across a network fetch — so the multi-minute walk can't strand a checked-out
-connection for the server to reap ([[db-connection-held-across-subprocess]]).
 
+The seen/dead-letter ledgers live in Postgres (`patreon_seen_media` /
+`patreon_failed_media`); the core opens SHORT-LIVED sync sessions per page batch
+— never held across a network fetch ([[db-connection-held-across-subprocess]]).
 FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
 """
 
@@ -40,15 +31,12 @@ from __future__ import annotations
 
 import asyncio
 import logging
-import time
 from collections.abc import Callable
 from pathlib import Path
 
-from sqlalchemy import delete, func, select, text
-from sqlalchemy.dialects.postgresql import insert as pg_insert
-
 from ..models import PatreonFailedMedia, PatreonSeenMedia
 from .gallery_dl import DownloadResult, ErrorType
+from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester
 from .patreon_client import (
     MediaItem,
     PatreonAPIError,
@@ -59,27 +47,19 @@ from .patreon_client import (
 from .patreon_downloader import PatreonDownloader
 from .patreon_resolver import resolve_campaign_id_for_source
 
-log = logging.getLogger(__name__)
+__all__ = [
+    "DEAD_LETTER_THRESHOLD",
+    "PatreonIngester",
+    "_ledger_key",
+    "verify_patreon_credential",
+]
 
-# gallery-dl's `exit:20` default ported over: stop a tick after this many
+log = logging.getLogger(__name__)
-# CONTIGUOUS already-have-it media (seen-ledger or on-disk). Native walks have
-# zero per-file HEADs, so the only cost of a higher number is a few extra cheap
-# ledger lookups — 20 is operator-set headroom against paywalled/undownloadable
-# items interleaving with archived ones.
-_TICK_SEEN_THRESHOLD = 20
 
 # Ledger keys are stored in patreon_seen_media.filehash VARCHAR(128); bound any
 # synthesized key so a pathologically long file_name can't overflow the column.
 _LEDGER_KEY_MAX = 128
 
-# plan #705 #7: after this many failed download/validate attempts a media is
-# "dead-lettered" and skipped on routine tick/backfill walks (recovery still
-# re-attempts it). Stops a permanently-broken media (404'd CDN, deleted post)
-# re-erroring forever and re-burning chunks.
-DEAD_LETTER_THRESHOLD = 3
-# last_error is Text but bound it so a giant traceback doesn't bloat the row.
-_ERROR_MAX = 1000
-
 
 def _ledger_key(media: MediaItem) -> str:
     """Stable per-media identity for the cross-run seen-ledger.
@@ -96,12 +76,12 @@ def _ledger_key(media: MediaItem) -> str:
     return f"{media.post_id}:{media.filename}"[:_LEDGER_KEY_MAX]
 
 
-class PatreonIngester:
+class PatreonIngester(Ingester):
     """Walk a Patreon campaign's posts, download unseen media, return a
-    `DownloadResult`.
+    `DownloadResult`. A thin adapter over `ingest_core.Ingester`.
 
     Construct with the per-source `cookies_path` and a sync sessionmaker for the
-    seen-ledger. `client` / `downloader` are injectable seams so unit tests run
+    ledgers. `client` / `downloader` are injectable seams so unit tests run
     without network, subprocess, or a real CDN.
     """
 
@@ -119,266 +99,36 @@ class PatreonIngester:
     ):
         self.images_root = Path(images_root)
         self.cookies_path = str(cookies_path) if cookies_path else None
-        self.session_factory = session_factory
         # Pacing (plan #703): request_sleep paces the API page fetches,
         # rate_limit paces the media downloads. Injected client/downloader (in
         # tests) already carry their own pacing, so these only apply to the
         # default-constructed ones.
-        self.client = (
+        resolved_client = (
             client
             if client is not None
             else PatreonClient(cookies_path, request_sleep=request_sleep)
         )
-        self.downloader = (
+        resolved_downloader = (
             downloader
             if downloader is not None
             else PatreonDownloader(
                 self.images_root, cookies_path, validate=validate, rate_limit=rate_limit
             )
         )
-
+        super().__init__(
-    # -- public ------------------------------------------------------------
+            client=resolved_client,
-
+            downloader=resolved_downloader,
-    def run(
+            session_factory=session_factory,
-        self,
+            seen_model=PatreonSeenMedia,
-        *,
+            failed_model=PatreonFailedMedia,
-        source_id: int,
+            seen_constraint="uq_patreon_seen_media_source_id",
-        campaign_id: str,
+            failed_constraint="uq_patreon_failed_media_source_id",
-        artist_slug: str,
+            ledger_key=_ledger_key,
-        url: str,
+            platform="patreon",
-        mode: str,
+            error_base=PatreonAPIError,
-        resume_cursor: str | None = None,
-        time_budget_seconds: float = 870.0,
-        seen_threshold: int = _TICK_SEEN_THRESHOLD,
-    ) -> DownloadResult:
-        """Walk + download for one source, returning a gallery-dl-shaped result.
-
-        `mode` is "tick" | "backfill" | "recovery". Recovery bypasses the tier-1
-        seen-ledger (tier-2 disk still skips kept files). The walk stops on:
-          - budget exhaustion (time_budget_seconds)         → TIMEOUT / PARTIAL
-          - tick early-out (seen_threshold contiguous seen)  → success
-          - reaching the bottom of the feed                  → success (rc 0)
-        A client-level failure (drift / auth / network) fails the whole run loud.
-        """
-        bypass_seen = mode == "recovery"
-        # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a
-        # tick has no resumable backfill state.
-        checkpoint = mode in ("backfill", "recovery")
-        start = time.monotonic()
-        log_lines: list[str] = []
-        written: list[str] = []
-        quarantined_paths: list[str] = []
-        downloaded = 0
-        errors = 0
-        quarantined = 0
-        dead_lettered = 0
-        skipped_count = 0
-        posts_processed = 0
-        consecutive_seen = 0
-        emitted_cursor: str | None = None
-        reached_bottom = False
-        budget_hit = False
-        early_out = False
-
-        def _result(
-            *, success: bool, return_code: int,
-            error_type: ErrorType | None, error_message: str | None,
-        ) -> DownloadResult:
-            # plan #704: return STRUCTURED data — phase 3 reads run_stats/cursor
-            # directly instead of regex-scraping a reconstructed stdout. stdout
-            # stays a human-readable summary (no fake `Cursor:` lines).
-            return DownloadResult(
-                success=success,
-                url=url,
-                artist_slug=artist_slug,
-                platform="patreon",
-                files_downloaded=downloaded,
-                files_quarantined=quarantined,
-                quarantined_paths=list(quarantined_paths),
-                written_paths=written,
-                stdout="\n".join(log_lines),
-                stderr="",
-                return_code=return_code,
-                error_type=error_type,
-                error_message=error_message,
-                duration_seconds=time.monotonic() - start,
-                cursor=emitted_cursor,
-                posts_processed=posts_processed,
-                run_stats={
-                    "exit_code": return_code,
-                    "downloaded_count": downloaded,
-                    "skipped_count": skipped_count,
-                    "per_item_failures": errors,
-                    "warning_count": 0,
-                    "tier_gated_count": 0,
-                    "quarantined_count": quarantined,
-                    "dead_lettered_count": dead_lettered,
-                },
-            )
-
-        try:
-            for post, included, page_cursor in self.client.iter_posts(
-                campaign_id, cursor=resume_cursor
-            ):
-                # Checkpoint the cursor that FETCHED this page the moment we
-                # START it — so a chunk cut mid-page resumes the page, not the one
-                # after it. Carried as DownloadResult.cursor (plan #704); no fake
-                # `Cursor:` stdout line to regex back out.
-                if page_cursor and page_cursor != emitted_cursor:
-                    emitted_cursor = page_cursor
-                    # plan #705 #6: persist the cursor at each page boundary so a
-                    # worker SIGKILL mid-chunk resumes near the crash, not the
-                    # chunk start. (phase 3 still writes the final cursor — same
-                    # value; this is the crash-safety net.)
-                    if checkpoint:
-                        self._checkpoint_cursor(source_id, emitted_cursor)
-
-                # Time-box check at the post boundary (coarse, like a gallery-dl
-                # chunk). Backfill/recovery resume from emitted_cursor next chunk.
-                if time.monotonic() - start >= time_budget_seconds:
-                    budget_hit = True
-                    break
-
-                posts_processed += 1
-                media = self.client.extract_media(post, included)
-                if not media:
-                    continue
-
-                keys = [_ledger_key(m) for m in media]
-                # Recovery bypasses BOTH the seen-ledger AND the dead-letter
-                # ledger (the operator's "try everything again"); routine walks
-                # skip seen + dead media (tier-1 + tier-1.5, plan #705 #7).
-                dead = set() if bypass_seen else self._dead_keys(source_id, keys)
-                seen = (
-                    set()
-                    if bypass_seen
-                    else self._seen_keys(source_id, keys)
-                )
-                skip = seen | dead
-
-                def _is_skip(m: MediaItem, _skip=skip) -> bool:
-                    return _ledger_key(m) in _skip
-
-                outcomes = self.downloader.download_post(
-                    post, media, artist_slug, is_seen=_is_skip
-                )
-
-                to_mark: list[tuple[str, str]] = []
-                to_clear: list[str] = []  # recovered → drop any dead-letter row
-                to_fail: list[tuple[str, str, str]] = []  # (key, post_id, error)
-                for media_item, outcome in zip(media, outcomes, strict=False):
-                    key = _ledger_key(media_item)
-                    if key in dead:
-                        dead_lettered += 1  # skipped because previously dead
-                    if outcome.status == "downloaded":
-                        downloaded += 1
-                        if outcome.path is not None:
-                            written.append(str(outcome.path))
-                        to_mark.append((key, media_item.post_id))
-                        to_clear.append(key)
-                        consecutive_seen = 0
-                    elif outcome.status == "skipped_disk":
-                        # Already on disk (a prior run). Reconcile the ledger so a
-                        # later tick skips it at tier-1 without a disk stat, but
-                        # do NOT re-feed it to phase 3 — attach_in_place would see
-                        # the duplicate sha256 and unlink the on-disk copy.
-                        to_mark.append((key, media_item.post_id))
-                        to_clear.append(key)
-                        skipped_count += 1
-                        consecutive_seen += 1
-                    elif outcome.status == "skipped_seen":
-                        skipped_count += 1
-                        consecutive_seen += 1
-                    elif outcome.status == "quarantined":
-                        # New content that failed validation (corrupt) — counted
-                        # distinctly so the run surfaces a real quarantined total.
-                        # Not marked seen (a later walk may re-fetch a fixed file);
-                        # it IS new content, so it breaks the run-of-seen. Counts
-                        # toward the dead-letter ledger (plan #705 #7).
-                        quarantined += 1
-                        if outcome.path is not None:
-                            quarantined_paths.append(str(outcome.path))
-                        to_fail.append((key, media_item.post_id, outcome.error or "quarantined"))
-                        consecutive_seen = 0
-                    elif outcome.status == "error":
-                        errors += 1
-                        to_fail.append((key, media_item.post_id, outcome.error or "error"))
-                        # An error neither advances nor resets the run-of-seen.
-
-                    if mode == "tick" and consecutive_seen >= seen_threshold:
-                        early_out = True
-                        break
-
-                # Persist ledger changes AFTER the network fetch, on short
-                # sessions: mark downloaded/on-disk seen, clear any dead-letter
-                # for recovered media, and record failures (plan #705 #7).
-                if to_mark:
-                    self._mark_seen(source_id, to_mark)
-                if to_clear:
-                    self._clear_failures(source_id, to_clear)
-                if to_fail:
-                    self._record_failures(source_id, to_fail)
-
-                if early_out:
-                    break
-            else:
-                reached_bottom = True
-        except PatreonAPIError as exc:
-            # Base of PatreonAuthError + PatreonDriftError — catches every
-            # client-level failure; _failure_result maps it to a typed error.
-            return self._failure_result(exc, _result)
-
-        if errors:
-            log_lines.append(f"{errors} media item(s) failed")
-        if quarantined:
-            log_lines.append(f"{quarantined} media item(s) quarantined (invalid)")
-        if dead_lettered:
-            log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)")
-        log_lines.append(
-            f"Patreon ingest ({mode}): {downloaded} downloaded, "
-            f"{skipped_count} skipped, {quarantined} quarantined, "
-            f"{dead_lettered} dead-lettered, {errors} error(s), "
-            f"{posts_processed} post(s)"
-            + (", reached end" if reached_bottom else "")
-            + (", time-boxed" if budget_hit else "")
         )
 
-        if budget_hit:
+    # -- failure mapping (Patreon exception taxonomy) ----------------------
-            # A chunk that hit its time-box but made forward progress is a
-            # NORMAL chunk boundary, not a failure (PARTIAL → status "ok"); the
-            # next chunk resumes from the emitted cursor. No progress → TIMEOUT,
-            # which feeds download_service's backfill stall-guard. rc<0 mirrors
-            # subprocess TimeoutExpired so completion detection stays false.
-            made_progress = downloaded > 0 or emitted_cursor != resume_cursor
-            if made_progress:
-                return _result(
-                    success=False, return_code=-1,
-                    error_type=ErrorType.PARTIAL,
-                    error_message=(
-                        f"Patreon backfill chunk: {downloaded} file(s) — continuing"
-                    ),
-                )
-            return _result(
-                success=False, return_code=-1,
-                error_type=ErrorType.TIMEOUT,
-                error_message="Patreon chunk timed out with no progress",
-            )
-
-        # Normal success: reached the bottom, or a tick that early-outed. rc 0 +
-        # error_type None is REQUIRED for a backfill/recovery walk that reached
-        # the bottom to be marked COMPLETE by
-        # download_service._apply_backfill_lifecycle — so we return None even
-        # when downloaded == 0 (a re-confirming walk that found nothing new still
-        # completed). NO_NEW_CONTENT would read as "not finished" there and trip
-        # the stall guard. success=True maps to status "ok" regardless. A tick
-        # that early-outed also returns here; ticks never set backfill state so
-        # the lifecycle is a no-op for them.
-        return _result(
-            success=True, return_code=0,
-            error_type=None, error_message=None,
-        )
-
-    # -- failure mapping ---------------------------------------------------
 
     def _failure_result(self, exc: Exception, _result) -> DownloadResult:
         """Map a client-level exception to a loud, typed failed DownloadResult.
@@ -417,136 +167,6 @@ class PatreonIngester:
             error_type=error_type, error_message=message,
         )
 
-    # -- seen-ledger (short-lived sessions) --------------------------------
-
-    def _seen_keys(self, source_id: int, keys: list[str]) -> set[str]:
-        """Which of `keys` are already in the ledger for this source.
-
-        One short SELECT on its own session — opened and closed without any
-        network in between (the GETs happen after, in download_post).
-        """
-        if not keys:
-            return set()
-        with self.session_factory() as session:
-            rows = session.execute(
-                select(PatreonSeenMedia.filehash).where(
-                    PatreonSeenMedia.source_id == source_id,
-                    PatreonSeenMedia.filehash.in_(keys),
-                )
-            ).scalars().all()
-        return set(rows)
-
-    def _checkpoint_cursor(self, source_id: int, cursor: str) -> None:
-        """Persist the in-progress backfill cursor mid-walk (plan #705 #6).
-
-        ATOMIC, single-key UPDATE: cast the JSON column to jsonb, set just
-        `_backfill_cursor`, cast back — so it never clobbers operator config or
-        the other backfill keys (no read-modify-write race). The in-flight guard
-        means only this source's one download runs at a time; a concurrent
-        operator stop is benign (a stray cursor with no `_backfill_state` is
-        ignored by tick mode and cleared on the next start).
-        """
-        with self.session_factory() as session:
-            session.execute(
-                text(
-                    "UPDATE source SET config_overrides = jsonb_set("
-                    "  coalesce(config_overrides::jsonb, '{}'::jsonb),"
-                    "  '{_backfill_cursor}', to_jsonb(cast(:cur AS text))"
-                    ")::json WHERE id = :sid"
-                ),
-                {"cur": cursor, "sid": source_id},
-            )
-            session.commit()
-
-    def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None:
-        """Idempotent upsert of (filehash, post_id) ledger rows for a page.
-
-        ON CONFLICT DO NOTHING against the (source_id, filehash) UNIQUE so a
-        re-sighting — or a concurrent walk — is a harmless no-op
-        ([[scalar_one_or_none-duplicates]]: never check-then-insert without the
-        DB constraint backing it). De-dup the batch locally first so a single
-        page can't present the same key twice to one INSERT.
-        """
-        seen_local: set[str] = set()
-        values = []
-        for key, post_id in items:
-            if key in seen_local:
-                continue
-            seen_local.add(key)
-            values.append(
-                {"source_id": source_id, "filehash": key, "post_id": post_id}
-            )
-        if not values:
-            return
-        with self.session_factory() as session:
-            stmt = pg_insert(PatreonSeenMedia).values(values)
-            stmt = stmt.on_conflict_do_nothing(
-                constraint="uq_patreon_seen_media_source_id"
-            )
-            session.execute(stmt)
-            session.commit()
-
-    # -- dead-letter ledger (plan #705 #7) ---------------------------------
-
-    def _dead_keys(self, source_id: int, keys: list[str]) -> set[str]:
-        """Which of `keys` have failed >= DEAD_LETTER_THRESHOLD times (dead).
-        One short SELECT; recovery never calls this (it re-attempts dead media)."""
-        if not keys:
-            return set()
-        with self.session_factory() as session:
-            rows = session.execute(
-                select(PatreonFailedMedia.filehash).where(
-                    PatreonFailedMedia.source_id == source_id,
-                    PatreonFailedMedia.filehash.in_(keys),
-                    PatreonFailedMedia.attempts >= DEAD_LETTER_THRESHOLD,
-                )
-            ).scalars().all()
-        return set(rows)
-
-    def _record_failures(
-        self, source_id: int, items: list[tuple[str, str, str]]
-    ) -> None:
-        """Upsert-increment the dead-letter ledger for failed media. On conflict
-        bump `attempts` and refresh last_error/last_failed_at (UNIQUE backs the
-        upsert — no check-then-insert, [[scalar_one_or_none-duplicates]]). De-dup
-        the batch locally (one row per key, last error wins)."""
-        by_key: dict[str, str] = {}
-        for key, _post_id, err in items:
-            by_key[key] = (err or "")[:_ERROR_MAX]
-        if not by_key:
-            return
-        values = [
-            {"source_id": source_id, "filehash": k, "attempts": 1, "last_error": e}
-            for k, e in by_key.items()
-        ]
-        with self.session_factory() as session:
-            stmt = pg_insert(PatreonFailedMedia).values(values)
-            stmt = stmt.on_conflict_do_update(
-                constraint="uq_patreon_failed_media_source_id",
-                set_={
-                    "attempts": PatreonFailedMedia.attempts + 1,
-                    "last_error": stmt.excluded.last_error,
-                    "last_failed_at": func.now(),
-                },
-            )
-            session.execute(stmt)
-            session.commit()
-
-    def _clear_failures(self, source_id: int, keys: list[str]) -> None:
-        """Drop dead-letter rows for media that just downloaded cleanly — they
-        recovered. A no-op DELETE for keys that were never failing."""
-        unique = list(dict.fromkeys(keys))
-        if not unique:
-            return
-        with self.session_factory() as session:
-            session.execute(
-                delete(PatreonFailedMedia).where(
-                    PatreonFailedMedia.source_id == source_id,
-                    PatreonFailedMedia.filehash.in_(unique),
-                )
-            )
-            session.commit()
-
 
 async def verify_patreon_credential(
     url: str,

tests/test_patreon_ingester.py

@@ -300,7 +300,9 @@ async def test_backfill_budget_cut_returns_partial_with_progress(
     source_id, sync_engine, tmp_path, monkeypatch,
 ):
     # Deterministic clock: start=0, post1 check=10 (ok), post2 check=200 (>budget).
-    import backend.app.services.patreon_ingester as mod
+    # run() lives in ingest_core now, so patch the clock there (the same global
+    # time module object, but we reference it through the module that uses it).
+    import backend.app.services.ingest_core as core
     ticks = iter([0.0, 10.0, 200.0, 250.0])
     last = [0.0]
 
@@ -311,7 +313,7 @@ async def test_backfill_budget_cut_returns_partial_with_progress(
             pass
         return last[0]
 
-    monkeypatch.setattr(mod.time, "monotonic", fake_monotonic)
+    monkeypatch.setattr(core.time, "monotonic", fake_monotonic)
 
     pages = [
         ("CUR1", [("p1", [_media("p1", 1)])]),