"""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,
                  should_stop) -> [MediaOutcome]` (status in downloaded/
                  skipped_seen/skipped_disk/quarantined/error;
                  `.path`/`.error`/`.post_id`). `should_stop()` is polled
                  between media so the time-box is honoured mid-post.
  - 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 json
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, make_run_stats
from .native_ingest_common import NativeAuthError, NativeDriftError

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

# plan #709: throttle the live-progress write to the running DownloadEvent to one
# every ~5s — a steady cadence for the Downloads view regardless of how big/slow a
# page is (page boundaries can be minutes apart on image-dense backfills, so a
# page-tied update would lurch). Trivial churn (~one single-row UPDATE / 5s).
_LIVE_PROGRESS_INTERVAL = 5.0

# Post-body schema-drift canary (#862). Patreon's body lives in
# content/content_json_string with NO post_type gate, so a field rename (as
# content→content_json_string already was) zeroes EVERY body at once — across
# every artist, every walk. If a native walk records at least this many posts
# and extracts a body from NONE of them, treat it as that break (fail the run
# API_DRIFT) rather than silently archiving empties. A *fraction* threshold would
# false-positive on gallery/art creators who legitimately post images with no
# caption, so the gate is "zero across a minimum sample": a real creator nearly
# always has SOME text across this many posts, a broken parser has none. Set high
# enough that a small tick (a few new posts) can't trip it — only a backfill /
# recapture (the operator's schema-test flow) reaches the sample.
_CANARY_MIN_SAMPLE = 30


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],
        drift_label: str | None = None,
        body_canary: bool = True,
    ):
        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
        # Human label for the API_DRIFT message ("<label> changed — ingester needs
        # update"). Defaults to the platform name; adapters pass a richer phrase
        # (e.g. "Patreon API", "SubscribeStar markup").
        self._drift_label = drift_label or platform
        # #862 canary opt-out: platforms whose posts legitimately have empty
        # bodies across large samples (pixiv — caption-less artists are common)
        # would false-positive the zero-bodies-means-drift alarm; their clients
        # catch drift structurally (response-shape checks) instead. The
        # "bodies X/N" summary line still surfaces the ratio either way.
        self._body_canary = body_canary

    # -- 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,
        posts_base: int = 0,
        event_id: int | None = None,
    ) -> DownloadResult:
        """Walk + download for one source, returning a gallery-dl-shaped result.

        `mode` is "tick" | "backfill" | "recovery" | "recapture". Recovery
        bypasses the tier-1 seen-ledger AND the dead-letter ledger (tier-2 disk
        still skips kept files). Recapture (#830) is the cheap "re-grab post
        text" walk: it bypasses the post-record gate (so EVERY post's body +
        external links are re-captured / detail-fetched) but KEEPS the media
        seen-ledger — on-disk media is NOT re-downloaded, only surfaced so its
        ImageRecord's source_filehash can be backfilled for inline-image
        localization. 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"
        recapture = mode == "recapture"
        # Both recovery and recapture re-capture EVERY post's body + links — they
        # bypass the post-record seen-gate (recovery via bypass_seen, recapture
        # explicitly). A plain backfill stays gated (capture once per post).
        recapture_records = bypass_seen or recapture
        # Only deep walks checkpoint their cursor mid-flight (plan #705 #6); a
        # tick has no resumable backfill state.
        checkpoint = mode in ("backfill", "recovery", "recapture")
        ledger_key = self._ledger_key
        # POST-FIRST CONTRACT (milestone #67): these two optional seams make a
        # platform "post-first" on the native core ingester — the post-record is
        # the single authoritative writer of the post body/links/metadata, and the
        # per-media sidecar carries image identity only (download_service flips
        # importer.post_first via uses_native_ingester, so the import side follows
        # automatically). A platform migrating off gallery-dl onto the native core
        # adopts post-first by implementing BOTH:
        #   client.post_record_key(post)   -> (ledger_key, post_id) | None  (gate)
        #   downloader.write_post_record(post, artist_slug) -> PostRecordOutcome
        # Absent on stub clients/downloaders (unit tests) and on not-yet-migrated
        # platforms → media-less posts are skipped as before and the body still
        # comes from the per-media sidecar (gallery-dl path). See [[post-first-ingest-contract]].
        post_record_key = getattr(self.client, "post_record_key", None)
        write_post_record = getattr(self.downloader, "write_post_record", None)
        # #874: optional client seam — skip tier-gated posts (the account can't
        # view them, so Patreon serves only blurred locked-preview media) ENTIRELY:
        # no media download, no post-record stub. Absent on stub/not-yet-migrated
        # clients → nothing is ever treated as gated.
        post_is_gated = getattr(self.client, "post_is_gated", None)
        start = time.monotonic()
        last_live = start  # plan #709: last live-progress write timestamp
        log_lines: list[str] = []
        written: list[str] = []
        post_records: list[str] = []
        quarantined_paths: list[str] = []
        # #830 recapture: (on-disk path, CDN source_url, post_id) triples for
        # already-present media, so phase 3 can (a) backfill the ImageRecord's
        # source_filehash and (b) link the on-disk image to its Post (#1288) —
        # WITHOUT re-downloading or unlinking the file. Empty outside recapture.
        relink: list[tuple[str, str, str]] = []
        downloaded = 0
        errors = 0
        quarantined = 0
        dead_lettered = 0
        skipped_count = 0
        posts_processed = 0
        # Post-body schema-drift canary counters (#862, native ingester only —
        # gallery-dl walks never enter the post-record block below so these stay 0
        # and the canary can't fire there). posts_recorded = post-records attempted
        # this walk; posts_with_body = how many yielded a non-empty body.
        posts_recorded = 0
        posts_with_body = 0
        # Tier-gated posts skipped entirely this walk (#874) — surfaced in the
        # run summary for diagnostics ("a lot of these now").
        gated_skipped = 0
        # Net-new posts THIS chunk for the live progress badge (plan #704 #5);
        # excludes the re-walked resume page so _backfill_posts stays a monotonic
        # absolute across chunks instead of an inflating sum. posts_processed
        # stays the gross per-chunk count used for the run summary.
        chunk_new_posts = 0
        consecutive_seen = 0
        emitted_cursor: str | None = None
        reached_bottom = False
        budget_hit = False
        early_out = False
        stopped = False  # plan #708 B4: operator hit Stop mid-walk
        cancel_armed = False  # latched once we observe a live "running" state

        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,
                post_record_paths=list(post_records),
                relink_source_paths=list(relink),
                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=make_run_stats(
                    exit_code=return_code,
                    downloaded_count=downloaded,
                    skipped_count=skipped_count,
                    per_item_failures=errors,
                    quarantined_count=quarantined,
                    dead_lettered_count=dead_lettered,
                ),
            )

        # #899 L1: emit run milestones through the real logger (not only the
        # in-memory log_lines → DownloadResult.stdout, which is persisted to the
        # DownloadEvent ONLY at phase 3). A worker SIGKILL/OOM/hard-time-limit
        # mid-walk would otherwise leave NO trace; these land in the container log
        # in real time regardless of whether the event gets finalized.
        log.info(
            "%s ingest START (%s): source=%s campaign=%s resume_cursor=%s",
            self._platform, mode, source_id, campaign_id, resume_cursor,
        )
        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
                    # #899 L1: a per-page breadcrumb in the container log (pages can
                    # be minutes apart on image-dense backfills) — survives a worker
                    # kill so the operator sees how far a since-died walk got.
                    log.info(
                        "%s ingest progress (%s, source=%s): posts=%d downloaded=%d "
                        "skipped=%d errors=%d quarantined=%d gated=%d cursor=%s",
                        self._platform, mode, source_id, posts_processed, downloaded,
                        skipped_count, errors, quarantined, gated_skipped, emitted_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.) plan #704 #5: persist
                    # the live posts count alongside it so the badge climbs DURING
                    # the chunk, not only when it ends.
                    if checkpoint:
                        # plan #708 B4: an operator Stop pops `_backfill_state` —
                        # bail at the page boundary (progress already checkpointed)
                        # before more network work, so the live chunk halts
                        # promptly instead of running to its time-box. LATCH on the
                        # first observed "running" state, so a run invoked WITHOUT a
                        # running state (a unit test, or a stale call) never
                        # spuriously self-cancels. A short SELECT, never held.
                        if self._still_running(source_id):
                            cancel_armed = True
                        elif cancel_armed:
                            stopped = True
                            break
                        self._checkpoint_cursor(source_id, emitted_cursor)
                        self._checkpoint_posts(source_id, posts_base + chunk_new_posts)

                # 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
                # The resume page (its cursor == resume_cursor) was already
                # counted by the chunk that checkpointed it — don't re-count it
                # into the persisted badge (plan #704 #5). First chunk has
                # resume_cursor None, so everything counts.
                if not (resume_cursor and page_cursor == resume_cursor):
                    chunk_new_posts += 1
                # Tier-gated post (#874): the account can't fully view it, so
                # Patreon serves only blurred locked-preview media. Skip it
                # ENTIRELY — no media download AND no post-record stub (operator
                # decision: gated content leaves no trace; a later walk re-ingests
                # it for real once access is gained). Skipped BEFORE the
                # post-record block so gated posts never inflate the #862 body
                # canary's sample. post_is_gated gates only on an explicit
                # current_user_can_view=False (missing/None → viewable).
                if post_is_gated and post_is_gated(post):
                    gated_skipped += 1
                    log_lines.append(
                        f"  post {post.get('id')} — gated (skipped, no access)"
                    )
                    continue
                # Capture the post body + external links ONCE per post (gated by
                # the synthetic post key in the seen-ledger), for EVERY post —
                # whether or not it has downloadable media. This is what makes a
                # backfill/recovery re-walk RECAPTURE bodies + links for posts
                # whose media is already on disk: re-downloading existing media
                # never fills links the system never had, so the body recapture
                # has to ride the walk itself. Detail-fetch (for an empty feed
                # body) happens at most once per post — the gate then spares it on
                # later walks. bypass_seen (recovery) re-captures unconditionally.
                if post_record_key and write_post_record:
                    rk = post_record_key(post)
                    if rk is not None:
                        pkey, ppid = rk
                        already = (
                            set() if recapture_records
                            else self._seen_keys(source_id, [pkey])
                        )
                        if pkey not in already:
                            rec = write_post_record(post, artist_slug)
                            posts_recorded += 1
                            if rec.body_chars:
                                posts_with_body += 1
                            if rec.path is not None:
                                post_records.append(str(rec.path))
                                self._mark_seen(source_id, [(pkey, ppid)])
                            # Per-post handling line in the run stdout (the existing
                            # "Raw stdout" panel) — the downloader already read the
                            # post; we only format its outcome here. post_type beside
                            # a 0-char body is the "why is this one empty" answer.
                            log_lines.append(
                                f"  post {ppid} [{rec.post_type or '?'}] "
                                f"body: {rec.body_chars} chars"
                                + ("" if rec.body_chars else " — EMPTY")
                                + (f" — {rec.title}" if rec.title else "")
                            )

                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

                # Honour the time-box DURING a media-dense post too, not only at
                # the per-post boundary below — else one heavy post can blow the
                # chunk budget out to the Celery soft limit (Pocketacer, 2026-06-07).
                outcomes = self.downloader.download_post(
                    post, media, artist_slug, is_seen=_is_skip,
                    should_stop=lambda: time.monotonic() - start >= time_budget_seconds,
                    recapture=recapture,
                )

                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
                        # #830/#1288 recapture: surface (on-disk path, CDN url,
                        # post_id) so phase 3 can backfill source_filehash AND link
                        # the on-disk image to its Post — a SEPARATE non-deleting
                        # channel, never the import list (which would unlink the
                        # file, per above).
                        if recapture and outcome.path is not None:
                            relink.append(
                                (str(outcome.path), media_item.url, media_item.post_id)
                            )
                    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)

                # plan #709: time-throttled live progress to the running event so
                # the Downloads view ticks ~every 5s, independent of page size.
                now = time.monotonic()
                if event_id is not None and (now - last_live) >= _LIVE_PROGRESS_INTERVAL:
                    last_live = now
                    self._write_live_progress(event_id, {
                        "downloaded": downloaded,
                        "skipped": skipped_count,
                        "errors": errors,
                        "quarantined": quarantined,
                        "posts": posts_processed,
                    })

                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)

        # plan #708 B4: a Stop already popped the backfill state (incl. cursor +
        # posts), so don't re-write them — return PARTIAL (reads as "ok/progress",
        # the lifecycle no-ops since state is gone) instead of a false "complete".
        if stopped:
            log.info(
                "%s ingest STOPPED by operator (%s, source=%s): %d file(s) this chunk",
                self._platform, mode, source_id, downloaded,
            )
            return _result(
                success=False, return_code=-1,
                error_type=ErrorType.PARTIAL,
                error_message=f"Stopped by operator: {downloaded} file(s) this chunk",
            )

        # Final authoritative posts count for the badge — captures the last page
        # after the last boundary write and the time-box break (plan #704 #5).
        if checkpoint:
            self._checkpoint_posts(source_id, posts_base + chunk_new_posts)

        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)")
        summary = (
            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), {len(post_records)} post-record(s), "
            f"{len(relink)} relinked"
            # Body-capture health (#862): even below the canary's red-alarm
            # threshold, surfacing the ratio makes a partial extraction regression
            # visible in the Raw stdout (e.g. "bodies 3/180" reads as off).
            + (f", bodies {posts_with_body}/{posts_recorded}" if posts_recorded else "")
            + (f", {gated_skipped} gated-skipped" if gated_skipped else "")
            + (", reached end" if reached_bottom else "")
            + (", time-boxed" if budget_hit else "")
        )
        log_lines.append(summary)
        # #899 L1: also to the container log (survives event-finalization failure).
        log.info("%s (source=%s)", summary, source_id)

        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",
            )

        # Post-body schema-drift canary (#862): a native walk recorded a
        # meaningful sample of posts but extracted a body from NONE of them. Since
        # the body field has no post_type gate, that's the signature of Patreon
        # renaming/restructuring the body field (as content→content_json_string
        # already was) — fail RED (API_DRIFT: "fix is the field-set/parser, not
        # creds") so the breakage screams instead of silently archiving empties.
        # Only reached on an otherwise-clean walk (timeout/stop/error returned
        # above), so it never masks a more specific failure.
        if (
            self._body_canary
            and posts_recorded >= _CANARY_MIN_SAMPLE
            and posts_with_body == 0
        ):
            msg = (
                f"Post-body canary: extracted a body from 0 of {posts_recorded} "
                "posts — Patreon's body field shape likely changed; the ingester "
                "needs a field-set/parser update."
            )
            log_lines.append(msg)
            log.error("%s (artist=%s)", msg, artist_slug)
            return _result(
                success=False, return_code=-1,
                error_type=ErrorType.API_DRIFT, error_message=msg,
            )

        # 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,
        )

    # -- 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
        # #874: same gated-post gate as run() — the preview must not count
        # blurred locked-preview media as "new", or it would overstate a gated
        # source's backlog (preview/apply parity, rule 93).
        post_is_gated = getattr(self.client, "post_is_gated", None)
        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
            if post_is_gated and post_is_gated(post):
                continue
            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:
        """Map a platform client-error to a loud, typed failed DownloadResult —
        NEVER a silent zero-download "success". The mapping is shared across
        platforms via the NativeAuthError/NativeDriftError taxonomy (the platform
        client raises subclasses), so a new platform gets it for free:
          - NativeAuthError  → AUTH_ERROR   (rotate the credential)
          - NativeDriftError → API_DRIFT    (the ingester/scraper needs updating)
          - HTTP 429 / 404   → RATE_LIMITED / NOT_FOUND
          - other HTTP status→ HTTP_ERROR;  transport failure → NETWORK_ERROR
        Auth/Drift are matched first (they also carry a status_code in some paths).
        """
        message = str(exc)
        if isinstance(exc, NativeAuthError):
            error_type = ErrorType.AUTH_ERROR
        elif isinstance(exc, NativeDriftError):
            error_type = ErrorType.API_DRIFT
            message = f"{self._drift_label} changed — ingester needs update: {message}"
        else:
            status = getattr(exc, "status_code", None)
            if status == 429:
                error_type = ErrorType.RATE_LIMITED
            elif status == 404:
                error_type = ErrorType.NOT_FOUND
            elif status is not None:
                error_type = ErrorType.HTTP_ERROR
            else:
                error_type = ErrorType.NETWORK_ERROR
        log.warning("%s ingest failed (%s): %s", self._platform, error_type.value, message)
        result = _result(
            success=False, return_code=1,
            error_type=error_type, error_message=message,
        )
        # plan #708 B1: carry the server's Retry-After up to the cooldown.
        if error_type == ErrorType.RATE_LIMITED:
            result.retry_after_seconds = getattr(exc, "retry_after", None)
        return result

    # -- 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 _write_live_progress(self, event_id: int, counts: dict) -> None:
        """Throttled mid-walk write of live counts to the RUNNING download_event
        (plan #709) so the Downloads view shows progress before the chunk
        finishes. A short session (never held across the walk); the `status =
        'running'` guard avoids clobbering an event phase 3 already finalized.
        `metadata` is JSONB — jsonb_set sets just the `live` key, leaving the rest
        for phase 3 to overwrite with the final run_stats."""
        with self.session_factory() as session:
            session.execute(
                text(
                    "UPDATE download_event SET metadata = jsonb_set("
                    "  coalesce(metadata, '{}'::jsonb), '{live}',"
                    "  cast(:live AS jsonb)) "
                    "WHERE id = :eid AND status = 'running'"
                ),
                {"live": json.dumps(counts), "eid": event_id},
            )
            session.commit()

    def _still_running(self, source_id: int) -> bool:
        """True while the source is armed for a deep walk (plan #708 B4).

        An operator Stop (`source_service.stop_backfill`) pops `_backfill_state`,
        so a False here means "cancel this chunk now". One short SELECT on its own
        session — never held across the walk
        ([[db-connection-held-across-subprocess]])."""
        with self.session_factory() as session:
            state = session.execute(
                text(
                    "SELECT config_overrides::jsonb ->> '_backfill_state' "
                    "FROM source WHERE id = :sid"
                ),
                {"sid": source_id},
            ).scalar_one_or_none()
        return state == "running"

    def _checkpoint_posts(self, source_id: int, posts: int) -> None:
        """Persist the live backfill posts-processed count mid-walk (plan #704 #5).

        Same atomic single-key jsonb_set dance as _checkpoint_cursor, on the
        `_backfill_posts` key (cast to a JSON number) — so the progress badge
        climbs DURING a chunk without clobbering operator config or the cursor.
        The ingester OWNS this key now; download_service no longer accumulates it
        post-chunk (which lagged a whole chunk and over-counted the resume page).
        """
        with self.session_factory() as session:
            session.execute(
                text(
                    "UPDATE source SET config_overrides = jsonb_set("
                    "  coalesce(config_overrides::jsonb, '{}'::jsonb),"
                    "  '{_backfill_posts}', to_jsonb(cast(:posts AS int))"
                    ")::json WHERE id = :sid"
                ),
                {"posts": posts, "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()