"""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 datetime import UTC, datetime, timedelta

from sqlalchemy import delete, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert

from .gallery_dl import (
    DownloadResult,
    ErrorType,
    classify_tier_gated,
    make_run_stats,
    tier_gated_message,
)
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

# How far back a tick keeps looking even once everything is already-have-it —
# the REVISIT WINDOW. Operator, 2026-09-23, holding up a Floppystack post:
# *"this post has been updated as he implements hot fixes — any chance we have a
# way to scan for or see updated posts so we can update ours to match and pull
# the new attachments and pictures etc."*
#
# A creator who edits a three-day-old post to append a hotfix build was
# structurally unreachable: that post sits twenty-odd already-seen items down
# the feed, so the count early-out above fired before the walk ever got to it.
# Not a bug in the early-out — a COUNT cannot express "recent".
#
# So the early-out now needs BOTH conditions: the run of already-seen items AND
# a post published before the horizon. Strictly a widening. Two properties this
# shape has and a plain "walk the last N days" would not:
#
#   * window 0 is exactly the old behaviour, so the feature has an off switch
#     that costs nothing to reason about;
#   * no window can make a tick stop EARLIER than it used to. A source paused
#     for months has an unseen backlog stretching well past any horizon, and
#     the walk still runs to the end of it — the horizon is a FLOOR on how far
#     to look, never a ceiling.
#
# The live value is `ImportSettings.download_revisit_days` (rule 25 — an
# operator tuning how far back their creators edit should not need a redeploy).
# This is the fallback for a caller that passes none.
DEFAULT_REVISIT_DAYS = 30

# 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


def _parse_published(raw: object) -> datetime | None:
    """An ISO-8601 post date from either native client, as aware UTC.

    Patreon's `published_at` is tz-aware with a `Z` or `+00:00` offset;
    SubscribeStar's is NAIVE (`_parse_ss_datetime` renders a parsed local
    timestamp with no zone). A naive value is read as UTC — the alternative is
    discarding it, and a post whose date we refuse to read is a post the revisit
    window can never reach.

    Anything unparseable returns None, which reads downstream as "not provably
    recent" and leaves the walk on its pre-revisit behaviour. Never raises: a
    date we cannot read must not fail a walk that is otherwise working.
    """
    if not isinstance(raw, str) or not raw.strip():
        return None
    try:
        parsed = datetime.fromisoformat(raw.strip().replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)


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

        `revisit_days` is the tick's revisit window (see DEFAULT_REVISIT_DAYS):
        inside it a tick neither early-outs nor trusts the post-record gate, so
        a post edited after we first captured it is re-read and its new
        attachments downloaded. 0 turns the window off.

        `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)
        # The revisit window (see DEFAULT_REVISIT_DAYS). `post_meta` is an
        # existing client seam — both native clients already implement it, for
        # a preview sample whose caller has since gone, so this needed no new
        # contract, only a live consumer for one. Absent seam, an unreadable
        # date or a window of 0 → `horizon` never matches and the walk behaves
        # exactly as it did before 2026-09-23.
        #
        # The window applies to TICKS only. A backfill is gated on purpose
        # (capture each post once) and `recapture` mode already exists for the
        # operator-driven "re-read every body" pass; a horizon there would be a
        # third overlapping answer to a question that has two.
        post_meta = getattr(self.client, "post_meta", None)
        horizon: datetime | None = None
        if mode == "tick" and revisit_days > 0 and post_meta is not None:
            horizon = datetime.now(UTC) - timedelta(days=revisit_days)
        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
        # Posts inside the revisit window that we had already captured, and the
        # media those revisits turned up. Reported in the run summary — the
        # operator's ask was to SEE the updated posts, not only to end up with
        # their files ("so we can update ours to match").
        revisited = 0
        revisit_downloads = 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,
                    # #874 follow-up: the native path counted gated posts but
                    # never reported them, so DownloadDetailModal's "Tier-gated"
                    # field read 0 on every native walk while gallery-dl's read
                    # true. A paywalled creator was indistinguishable from a
                    # silent one.
                    tier_gated_count=gated_skipped,
                ),
            )

        # #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
                # Inside the revisit window? Computed per post rather than
                # "stop once one post is old" because the feed is only MOSTLY
                # date-ordered — a pinned or re-pinned post can sit above older
                # ones, and one such post must not end the walk.
                in_window = False
                if horizon is not None:
                    published = _parse_published((post_meta(post) or {}).get("date"))
                    in_window = published is not None and published >= horizon
                # Set by the post-record block below when this post was already
                # captured on an earlier walk. Stays False when the platform has
                # no post-record seam, so the revisit accounting simply reports
                # nothing rather than guessing.
                post_already_recorded = False
                downloaded_before = downloaded
                # 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])
                        )
                        post_already_recorded = pkey in already
                        # A post inside the revisit window is re-read even
                        # though the gate has it: that gate's whole job is to
                        # stop us paying for a post twice, and an EDITED post is
                        # not the same post. `revisit=True` keeps the cost at
                        # zero requests — the downloader re-reads the body from
                        # the feed response already in hand and declines to
                        # write at all if that body came back empty, so a
                        # detail-fetched body is never overwritten by a blank.
                        if not post_already_recorded or in_window:
                            rec = write_post_record(
                                post, artist_slug, revisit=post_already_recorded,
                            )
                            if not post_already_recorded:
                                # FIRST captures only feed the #862 body canary.
                                # A revisit legitimately comes back empty — a
                                # post whose body only ever arrived from the
                                # detail endpoint has none in the feed, and the
                                # downloader declines to write it. Counting
                                # those into the sample would walk the canary
                                # toward firing on healthy ticks, which is the
                                # one thing a drift alarm must never do.
                                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 '?'}] "
                                + ("re-read, " if post_already_recorded else "")
                                + 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.

                    # `not in_window` is the revisit window's half of the
                    # early-out: a run of already-seen items is only permission
                    # to stop once the walk is BELOW the horizon. Both halves,
                    # never either alone — see DEFAULT_REVISIT_DAYS.
                    if (
                        mode == "tick"
                        and not in_window
                        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)

                # An already-captured post that yielded NEW media is an edited
                # post — the operator's Floppystack case, and the one thing in
                # this walk worth naming individually in the run log. The media
                # half needed no new detection: `extract_media` reads the media
                # list off the live feed response, so a hotfix build appended
                # last night is simply a ledger key we have never seen.
                new_here = downloaded - downloaded_before
                if post_already_recorded and new_here:
                    revisited += 1
                    revisit_downloads += new_here
                    log_lines.append(
                        f"  post {post.get('id')} — updated: "
                        f"{new_here} new file(s)"
                    )

                # 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,
                        # Ticks during the walk, not only at finalization: a
                        # deep backfill on a creator we've lost access to is
                        # otherwise a long run of zeros with no explanation.
                        "gated": gated_skipped,
                    })

                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 "")
            # Only when it happened: on a quiet tick this is 0 and saying so
            # every run would bury the times it is not.
            + (
                f", {revisited} post(s) updated ({revisit_downloads} new file(s))"
                if revisited 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. A
        # zero-download walk still returns success here — a re-confirming walk
        # that found nothing new genuinely completed. A tick that early-outed
        # also lands here; ticks never set backfill state so the lifecycle is a
        # no-op for them.
        #
        # success=True and return_code=0 are load-bearing, not cosmetic. They
        # are what make this a COMPLETE walk for
        # download_service._apply_backfill_lifecycle (via walk_completed) and
        # what map it to status "ok", so a walk that fetched nothing doesn't
        # accrue consecutive_failures or a backoff it hasn't earned.
        #
        # #874 follow-up: "nothing new" and "everything sat behind a tier you
        # don't hold" are different facts, and returning None for both made a
        # paywalled creator indistinguishable from a silent one. TIER_LIMITED is
        # classified LAST — every real failure has already returned above —
        # because tier-gating is the weakest signal and must never mask a
        # genuine error. It is informational, so walk_completed still counts
        # this walk as finished (see that predicate for why re-walking a
        # paywalled creator forever is the bug being avoided).
        gated_error = classify_tier_gated(gated_skipped)
        return _result(
            success=True, return_code=0,
            error_type=gated_error,
            error_message=(
                tier_gated_message(gated_skipped) if gated_error else None
            ),
        )

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