"""PatreonIngester tests (native ingester build step 3).

The HTTP client and the media downloader are stubbed (injected seams), so these
exercise the ingester's own logic — mode behavior, the two-tier skip, cursor
emission, the budget cutoff, and the Postgres seen-ledger — without network or a
real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so
the tier-1 skip and the idempotent mark-seen run against actual rows.
"""

import pytest
from sqlalchemy import func, select
from sqlalchemy.orm import sessionmaker

from backend.app.models import (
    Artist,
    PatreonFailedMedia,
    PatreonSeenMedia,
    Source,
)
from backend.app.services.gallery_dl import ErrorType
from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
from backend.app.services.patreon_client import (
    MediaItem,
    PatreonAPIError,
    PatreonAuthError,
    PatreonDriftError,
)
from backend.app.services.patreon_downloader import MediaOutcome, PostRecordOutcome
from backend.app.services.patreon_ingester import PatreonIngester, _ledger_key

pytestmark = pytest.mark.integration


# --- fakes ----------------------------------------------------------------


def _media(post_id, n, *, filehash=None, kind="images"):
    # Deterministic 32-hex-ish key, unique per (post_id, n).
    fh = filehash if filehash is not None else f"{post_id}{n}".encode().hex().ljust(32, "0")[:32]
    return MediaItem(
        url=f"https://cdn.patreon.com/{fh}/{n}.jpg",
        filename=f"{n}.jpg",
        kind=kind,
        filehash=fh,
        post_id=post_id,
    )


class _FakeClient:
    """Stub PatreonClient. `pages` is a list of (page_cursor, [posts]); each post
    is (post_id, [MediaItem]). `raise_on_first` lets a test trip drift."""

    def __init__(self, pages, raise_exc=None, empty_body=False, gated=None):
        self._pages = pages
        self._raise_exc = raise_exc
        # empty_body simulates a body-field schema break: every post comes back
        # with no content (the #862 canary's trip condition).
        self._empty_body = empty_body
        # #874: post_ids the account can't view → current_user_can_view=False,
        # so the walk must skip them entirely (no media, no post-record).
        self._gated = set(gated or ())
        self.consumed_posts = 0

    def iter_posts(self, campaign_id, cursor=None):
        if self._raise_exc is not None:
            raise self._raise_exc
        for page_cursor, posts in self._pages:
            for post_id, media in posts:
                self.consumed_posts += 1
                # Carry feed attributes (title/post_type/content) like the real
                # client — ingest_core reads these for the per-post diagnostic.
                content = "" if self._empty_body else f"<p>body {post_id}</p>"
                yield (
                    {
                        "id": post_id,
                        "_media": media,
                        "attributes": {
                            "title": f"Post {post_id}",
                            "post_type": "image_file",
                            "content": content,
                            "current_user_can_view": post_id not in self._gated,
                        },
                    },
                    {},
                    page_cursor,
                )

    def extract_media(self, post, included_index):
        return post["_media"]

    def post_meta(self, post):
        return {"title": post.get("id"), "date": None}

    @staticmethod
    def post_is_gated(post):
        return (post.get("attributes") or {}).get("current_user_can_view") is False

    @staticmethod
    def post_record_key(post):
        pid = str(post.get("id") or "")
        return (f"post:{pid}", pid) if pid else None


class _FakeDownloader:
    """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in
    `on_disk` report skipped_disk (tier-2); media in `quarantine` report
    quarantined; everything else downloads."""

    def __init__(self, tmp_path, on_disk=None, quarantine=None, error=None):
        self.tmp_path = tmp_path
        self.on_disk = set(on_disk or ())
        self.quarantine = set(quarantine or ())
        self.error = set(error or ())
        self.download_calls = 0
        self.post_records = 0

    def download_post(self, post, media_items, artist_slug, *, is_seen,
                      should_stop=lambda: False, recapture=False):
        outcomes = []
        for m in media_items:
            if should_stop():
                break
            seen = is_seen(m)
            # Recapture mirrors the real downloader: a seen item isn't tier-1
            # short-circuited — it falls through so an on-disk file surfaces as
            # skipped_disk (for source_filehash backfill); a seen file NOT on
            # disk is left alone (skipped_seen, never refetched).
            if seen and not recapture:
                outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
            elif _ledger_key(m) in self.on_disk:
                p = self.tmp_path / f"{m.post_id}_{m.filename}"
                outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None))
            elif seen:
                outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
            elif _ledger_key(m) in self.error:
                self.download_calls += 1
                outcomes.append(MediaOutcome(media=m, status="error", path=None, error="boom"))
            elif _ledger_key(m) in self.quarantine:
                self.download_calls += 1
                q = self.tmp_path / "_quarantine" / f"{m.post_id}_{m.filename}"
                outcomes.append(MediaOutcome(media=m, status="quarantined", path=q, error="bad image"))
            else:
                self.download_calls += 1
                p = self.tmp_path / f"{m.post_id}_{m.filename}"
                p.write_bytes(b"x")
                outcomes.append(MediaOutcome(media=m, status="downloaded", path=p, error=None))
        return outcomes

    def write_post_record(self, post, artist_slug):
        self.post_records += 1
        p = self.tmp_path / f"{post.get('id')}__post.json"
        p.write_text("{}")
        attrs = post.get("attributes") or {}
        body = attrs.get("content")
        return PostRecordOutcome(
            path=p,
            post_type=attrs.get("post_type"),
            title=attrs.get("title"),
            body_chars=len(body) if isinstance(body, str) else 0,
        )


@pytest.fixture
async def source_id(db):
    artist = Artist(name="Ingest", slug="ingest")
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform="patreon",
        url="https://patreon.com/ingest", enabled=True, config_overrides={},
    )
    db.add(source)
    await db.commit()
    return source.id


def _ingester(sync_engine, tmp_path, client, downloader):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    return PatreonIngester(
        images_root=tmp_path, cookies_path=None, session_factory=factory,
        client=client, downloader=downloader,
    )


def _count_ledger(sync_engine, source_id):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        return s.execute(
            select(func.count(PatreonSeenMedia.id)).where(
                PatreonSeenMedia.source_id == source_id
            )
        ).scalar_one()


# --- tick -----------------------------------------------------------------


@pytest.mark.asyncio
async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_path):
    m1, m2 = _media("p1", 1), _media("p1", 2)
    client = _FakeClient([(None, [("p1", [m1, m2])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )

    assert result.success is True
    assert result.return_code == 0
    assert result.files_downloaded == 2
    assert len(result.written_paths) == 2
    # plan #704: structured run_stats carry the real counts.
    assert result.run_stats["downloaded_count"] == 2
    assert result.posts_processed == 1
    # 2 media keys + 1 synthetic post key (body/links recaptured per post).
    assert _count_ledger(sync_engine, source_id) == 3
    # The post body + links are captured for media posts too (rides the walk).
    assert len(result.post_record_paths) == 1


@pytest.mark.asyncio
async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_path):
    """plan #704 (#4): a media that fails validation is reported as quarantined —
    a real files_quarantined count + paths + run_stats — not a blank 0, and is
    NOT written or marked seen."""
    m1, m2 = _media("p1", 1), _media("p1", 2)
    client = _FakeClient([(None, [("p1", [m1, m2])])])
    downloader = _FakeDownloader(tmp_path, quarantine={_ledger_key(m2)})
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    assert result.files_downloaded == 1          # only m1
    assert result.files_quarantined == 1         # m2
    assert len(result.quarantined_paths) == 1
    assert result.run_stats["quarantined_count"] == 1
    assert result.run_stats["downloaded_count"] == 1
    assert len(result.written_paths) == 1        # quarantined NOT written
    # Quarantined media is NOT marked seen (a fixed file may be re-fetched);
    # m1 + the synthetic post key (body/links captured per post) = 2.
    assert _count_ledger(sync_engine, source_id) == 2


@pytest.mark.asyncio
async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
    m1 = _media("p1", 1)
    # Pre-seed the ledger with m1's key.
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
        s.commit()

    client = _FakeClient([(None, [("p1", [m1])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    assert result.files_downloaded == 0
    assert downloader.download_calls == 0


@pytest.mark.asyncio
async def test_tick_early_out_after_threshold(source_id, sync_engine, tmp_path):
    # Three posts, one already-seen media each; threshold 2 → stop before the 3rd.
    seen_media = [_media(f"p{i}", 1) for i in range(1, 4)]
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        for m in seen_media:
            s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id))
        s.commit()

    pages = [(None, [(m.post_id, [m]) for m in seen_media])]
    client = _FakeClient(pages)
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
    )
    assert result.success is True
    # Stopped after the 2nd contiguous seen item — never consumed the 3rd post.
    assert client.consumed_posts == 2


# --- backfill -------------------------------------------------------------


@pytest.mark.asyncio
async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, tmp_path):
    pages = [
        (None, [("p1", [_media("p1", 1)])]),
        ("CUR2", [("p2", [_media("p2", 1)])]),
    ]
    client = _FakeClient(pages)
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    # rc 0 + error_type None is what _apply_backfill_lifecycle reads as "complete".
    assert result.return_code == 0
    assert result.error_type is None
    assert result.files_downloaded == 2
    # The page-2 cursor is carried structurally for checkpointing (plan #704).
    assert result.cursor == "CUR2"
    assert result.posts_processed == 2
    assert result.run_stats["downloaded_count"] == 2


@pytest.mark.asyncio
async def test_backfill_checkpoints_cursor_mid_walk(source_id, sync_engine, tmp_path, db):
    """plan #705 #6: the cursor is persisted to the DB at each page boundary
    DURING the walk (not just at the end via phase 3), so a worker SIGKILL
    mid-chunk resumes near the crash. After the walk the source carries the last
    page's cursor — written by the ingester, not phase 3 (which the unit run
    never reaches)."""
    pages = [
        (None, [("p1", [_media("p1", 1)])]),
        ("CUR2", [("p2", [_media("p2", 1)])]),
        ("CUR3", [("p3", [_media("p3", 1)])]),
    ]
    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
    ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source_id)
    )).scalar_one()
    assert co.get("_backfill_cursor") == "CUR3"


@pytest.mark.asyncio
async def test_backfill_persists_live_posts_count(source_id, sync_engine, tmp_path, db):
    """plan #704 #5: the ingester writes a monotonic _backfill_posts absolute
    DURING the walk (seeded from prior chunks via posts_base), EXCLUDING the
    re-walked resume page so the count doesn't inflate across chunks."""
    # Resume from CUR2 (a prior chunk left off here): its page is re-walked and
    # must NOT be re-counted. posts_base=4 stands in for the prior chunks.
    pages = [
        ("CUR2", [("p1", [_media("p1", 1)])]),   # the resumed page — not counted
        ("CUR3", [("p2", [_media("p2", 1)])]),   # net-new
        ("CUR4", [("p3", [_media("p3", 1)])]),   # net-new
    ]
    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
    ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
        resume_cursor="CUR2", posts_base=4,
    )
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source_id)
    )).scalar_one()
    # 4 prior + 2 net-new (p2, p3); p1 on the resumed CUR2 page is excluded.
    assert co.get("_backfill_posts") == 6


@pytest.mark.asyncio
async def test_tick_does_not_persist_posts(source_id, sync_engine, tmp_path, db):
    """A tick has no resumable backfill state, so it must not write _backfill_posts
    (the badge is a backfill/recovery concept)."""
    ing = _ingester(
        sync_engine, tmp_path,
        _FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]),
        _FakeDownloader(tmp_path),
    )
    ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source_id)
    )).scalar_one()
    assert "_backfill_posts" not in (co or {})


@pytest.mark.asyncio
async def test_tick_does_not_checkpoint_cursor(source_id, sync_engine, tmp_path, db):
    """A tick has no resumable backfill state, so it must not write a cursor."""
    ing = _ingester(
        sync_engine, tmp_path,
        _FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]),
        _FakeDownloader(tmp_path),
    )
    ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source_id)
    )).scalar_one()
    assert "_backfill_cursor" not in (co or {})


@pytest.mark.asyncio
async def test_backfill_budget_cut_returns_partial_with_progress(
    source_id, sync_engine, tmp_path, monkeypatch,
):
    # Deterministic clock: start=0; post1 gate=10 (ok); post1's per-media
    # should_stop=20 (still ok, so post1's item downloads); post1 live-progress
    # read=200; post2 gate=250 (>budget → cut). The per-media should_stop check
    # (added 2026-06-07 to bound media-dense posts) reads the clock once more per
    # post, so the sequence carries the extra tick. run() lives in ingest_core
    # now, so patch the clock there.
    import backend.app.services.ingest_core as core
    ticks = iter([0.0, 10.0, 20.0, 200.0, 250.0])
    last = [0.0]

    def fake_monotonic():
        try:
            last[0] = next(ticks)
        except StopIteration:
            pass
        return last[0]

    monkeypatch.setattr(core.time, "monotonic", fake_monotonic)

    pages = [
        ("CUR1", [("p1", [_media("p1", 1)])]),
        ("CUR2", [("p2", [_media("p2", 1)])]),
    ]
    client = _FakeClient(pages)
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
        resume_cursor=None, time_budget_seconds=100.0,
    )
    assert result.error_type == ErrorType.PARTIAL
    assert result.return_code == -1
    assert result.files_downloaded == 1  # post1 downloaded before the cut
    assert downloader.download_calls == 1  # post2's body never ran
    # The checkpoint is the page we were CUT ON (CUR2 — entered, cursor emitted,
    # then the budget check broke before processing it); the next chunk resumes
    # there. Structured (plan #704), matching the old parse_last_cursor(last).
    assert result.cursor == "CUR2"


@pytest.mark.asyncio
async def test_write_live_progress_updates_running_event(
    source_id, sync_engine, tmp_path, db,
):
    """plan #709: live counts land in the RUNNING event's metadata.live; a
    finalized event is left alone (status guard)."""
    from backend.app.models import DownloadEvent

    ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path))
    running = DownloadEvent(source_id=source_id, status="running")
    done = DownloadEvent(source_id=source_id, status="ok")
    db.add_all([running, done])
    await db.commit()

    counts = {"downloaded": 3, "skipped": 1, "errors": 0, "quarantined": 0, "posts": 4}
    ing._write_live_progress(running.id, counts)
    ing._write_live_progress(done.id, {"downloaded": 9})  # guarded: status != running

    live = (await db.execute(
        select(DownloadEvent.metadata_).where(DownloadEvent.id == running.id)
    )).scalar_one()
    assert live["live"] == counts
    done_meta = (await db.execute(
        select(DownloadEvent.metadata_).where(DownloadEvent.id == done.id)
    )).scalar_one()
    assert "live" not in (done_meta or {})


@pytest.mark.asyncio
async def test_still_running_reads_backfill_state(source_id, sync_engine, tmp_path, db):
    """plan #708 B4: _still_running reflects the source's `_backfill_state` — the
    flag an operator Stop pops to cancel a live chunk."""
    ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path))
    assert ing._still_running(source_id) is False  # absent → not running
    src = (await db.execute(select(Source).where(Source.id == source_id))).scalar_one()
    src.config_overrides = {"_backfill_state": "running"}
    await db.commit()
    assert ing._still_running(source_id) is True


@pytest.mark.asyncio
async def test_backfill_stops_when_operator_cancels_mid_walk(
    source_id, sync_engine, tmp_path,
):
    """plan #708 B4: with the running state latched, a later page boundary that
    finds it gone (Stop) bails the chunk with PARTIAL instead of finishing."""
    pages = [
        ("CUR2", [("p1", [_media("p1", 1)])]),
        ("CUR3", [("p2", [_media("p2", 1)])]),
        ("CUR4", [("p3", [_media("p3", 1)])]),
    ]
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), downloader)

    # Latch on the page-2 boundary (running), then "Stop" before page 3.
    calls = {"n": 0}

    def fake_still_running(_source_id):
        calls["n"] += 1
        return calls["n"] == 1

    ing._still_running = fake_still_running

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert result.error_type == ErrorType.PARTIAL
    assert "Stopped by operator" in result.error_message
    assert downloader.download_calls == 1  # only p1 ran; cancelled before p2


# --- recovery -------------------------------------------------------------


@pytest.mark.asyncio
async def test_recovery_bypasses_seen_ledger(source_id, sync_engine, tmp_path):
    m1 = _media("p1", 1)
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
        s.commit()

    client = _FakeClient([(None, [("p1", [m1])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="recovery",
    )
    # Ledger says seen, but recovery bypasses tier-1 → re-downloaded for a fresh
    # pHash evaluation under the current threshold.
    assert result.files_downloaded == 1
    assert downloader.download_calls == 1


@pytest.mark.asyncio
async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path):
    m1 = _media("p1", 1)
    client = _FakeClient([(None, [("p1", [m1])])])
    # File still on disk (a kept image) → tier-2 spares it even under recovery.
    downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="recovery",
    )
    assert result.files_downloaded == 0
    assert downloader.download_calls == 0
    # Disk-skip reconciles the media key + the synthetic post key (recovery
    # recaptures the body/links per post) = 2.
    assert _count_ledger(sync_engine, source_id) == 2


@pytest.mark.asyncio
async def test_backfill_recaptures_body_for_already_downloaded_post(
    source_id, sync_engine, tmp_path,
):
    """Phase 5 guarantee: a post whose media is ALREADY on disk still gets its
    body + external links recaptured on a re-walk. Re-downloading existing media
    can never fill links the system never had — so the recapture rides the walk
    itself, and a normal backfill is the backfill."""
    m1 = _media("p1", 1)
    client = _FakeClient([(None, [("p1", [m1])])])
    downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert result.files_downloaded == 0          # media already on disk
    assert len(result.post_record_paths) == 1    # body/links recaptured anyway
    assert downloader.post_records == 1


def _seed_seen(sync_engine, source_id, key, post_id=None):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        s.add(PatreonSeenMedia(source_id=source_id, filehash=key, post_id=post_id))
        s.commit()


@pytest.mark.asyncio
async def test_backfill_skips_already_captured_post_but_recapture_forces_it(
    source_id, sync_engine, tmp_path,
):
    """#830: a plain backfill GATES post-record capture on the seen-ledger, so a
    post whose `post:<id>` key is already seen is NOT recaptured (the operator's
    'Industrial Lust body missing' bug). RECAPTURE mode bypasses that gate and
    re-captures the body/links even though the post key is seen — without
    re-downloading the on-disk media."""
    m1 = _media("p1", 1)
    # Pre-seed BOTH the media key and the synthetic post key as already seen.
    _seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1")
    _seed_seen(sync_engine, source_id, "post:p1", post_id="p1")

    # 1) Plain backfill: post key is seen → gate skips body recapture.
    client = _FakeClient([(None, [("p1", [m1])])])
    downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
    ing = _ingester(sync_engine, tmp_path, client, downloader)
    res_backfill = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert res_backfill.post_record_paths == []     # gated → not recaptured
    assert downloader.post_records == 0

    # 2) Recapture: gate bypassed → body recaptured; media NOT re-downloaded;
    # the on-disk file is surfaced for source_filehash relink.
    client2 = _FakeClient([(None, [("p1", [m1])])])
    downloader2 = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
    ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
    res_recap = ing2.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="recapture",
    )
    assert len(res_recap.post_record_paths) == 1    # gate bypassed → recaptured
    assert downloader2.post_records == 1
    assert res_recap.files_downloaded == 0          # no media re-download
    assert downloader2.download_calls == 0
    # The on-disk media is surfaced as a (path, CDN url, post_id) relink triple
    # (post_id drives the #1288 image→post back-link in phase 3).
    assert len(res_recap.relink_source_paths) == 1
    rel_path, rel_url, rel_post_id = res_recap.relink_source_paths[0]
    assert rel_url == m1.url
    assert rel_post_id == m1.post_id
    # Diagnostic summary surfaces the post-record + relink counts (operator reads
    # this off the event stdout to see what a recapture actually did).
    assert "1 post-record(s), 1 relinked" in res_recap.stdout
    # #842: per-post handling is written into the SAME run stdout (reuses the
    # existing "Raw stdout" panel) — post_type + body length, so an empty body is
    # self-explanatory by its post_type.
    assert f"post p1 [image_file] body: {len('<p>body p1</p>')} chars" in res_recap.stdout


@pytest.mark.asyncio
async def test_gated_post_skipped_entirely_no_media_no_record(
    source_id, sync_engine, tmp_path,
):
    """#874: a tier-gated post (current_user_can_view=False) is skipped ENTIRELY
    — no media download AND no post-record stub — while an accessible post in the
    same walk is fully ingested. Patreon serves only blurred locked-preview media
    for gated posts; downloading it produced unusable images."""
    gm = _media("gated", 1)
    am = _media("open", 1)
    client = _FakeClient([(None, [("gated", [gm]), ("open", [am])])], gated={"gated"})
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )

    # Only the accessible post's media downloaded; the gated post contributed
    # nothing — no file, no post-record.
    assert result.files_downloaded == 1
    # (substring check on the basename only — the tmp dir name contains "gated")
    assert not any(p.endswith("gated_1.jpg") for p in result.written_paths)
    assert downloader.download_calls == 1
    assert len(result.post_record_paths) == 1     # only the open post
    assert downloader.post_records == 1
    # The gated post left NO trace in the seen-ledger (no media key, no post key):
    # only the open post's media key + synthetic post key are recorded.
    assert _count_ledger(sync_engine, source_id) == 2


@pytest.mark.asyncio
async def test_recapture_does_not_refetch_seen_media_missing_from_disk(
    source_id, sync_engine, tmp_path,
):
    """Recapture must NOT re-download a seen item that's absent from disk (that's
    recovery's job) — it returns skipped_seen and contributes no relink pair."""
    m1 = _media("p1", 1)
    _seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1")
    client = _FakeClient([(None, [("p1", [m1])])])
    # NOT in on_disk → seen + missing.
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)
    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="recapture",
    )
    assert result.files_downloaded == 0
    assert downloader.download_calls == 0
    assert result.relink_source_paths == []


# --- dead-letter ledger (plan #705 #7) ------------------------------------


def _count_failed(sync_engine, source_id):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        return s.execute(
            select(func.count(PatreonFailedMedia.id)).where(
                PatreonFailedMedia.source_id == source_id
            )
        ).scalar_one()


def _seed_failed(sync_engine, source_id, key, attempts):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        s.add(PatreonFailedMedia(source_id=source_id, filehash=key, attempts=attempts))
        s.commit()


@pytest.mark.asyncio
async def test_repeated_failures_dead_letter_then_skip(source_id, sync_engine, tmp_path):
    """A media that errors DEAD_LETTER_THRESHOLD times is recorded; the next walk
    skips it (no download attempt) and counts it dead-lettered."""
    from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD

    m1 = _media("p1", 1)

    def _run():
        ing = _ingester(
            sync_engine, tmp_path,
            _FakeClient([(None, [("p1", [m1])])]),
            _FakeDownloader(tmp_path, error={_ledger_key(m1)}),
        )
        result = ing.run(
            source_id=source_id, campaign_id="c1", artist_slug="ingest",
            url="https://patreon.com/ingest", mode="tick",
        )
        return result, ing.downloader

    # Fail it up to the threshold — each run attempts the download and errors.
    for _ in range(DEAD_LETTER_THRESHOLD):
        result, dl = _run()
        assert dl.download_calls == 1
    assert _count_failed(sync_engine, source_id) == 1

    # Now attempts == threshold → dead: the next walk skips it without trying.
    result, dl = _run()
    assert dl.download_calls == 0
    assert result.run_stats["dead_lettered_count"] == 1


@pytest.mark.asyncio
async def test_recovery_reattempts_dead_lettered_and_clears_on_success(
    source_id, sync_engine, tmp_path,
):
    """Recovery bypasses the dead-letter ledger (re-attempts dead media); a clean
    download clears the row (it recovered)."""
    from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD

    m1 = _media("p1", 1)
    _seed_failed(sync_engine, source_id, _ledger_key(m1), DEAD_LETTER_THRESHOLD)

    downloader = _FakeDownloader(tmp_path)  # succeeds
    ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader)
    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="recovery",
    )
    assert downloader.download_calls == 1   # re-attempted despite being dead
    assert result.files_downloaded == 1
    assert _count_failed(sync_engine, source_id) == 0  # cleared on success


@pytest.mark.asyncio
async def test_clean_download_clears_below_threshold_failure(
    source_id, sync_engine, tmp_path,
):
    """A media with a sub-threshold failure that now downloads cleanly (tick) has
    its dead-letter row cleared."""
    m1 = _media("p1", 1)
    _seed_failed(sync_engine, source_id, _ledger_key(m1), 1)  # not yet dead

    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader)
    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    assert result.files_downloaded == 1
    assert _count_failed(sync_engine, source_id) == 0


# --- ledger + drift -------------------------------------------------------


@pytest.mark.asyncio
async def test_mark_seen_is_idempotent(source_id, sync_engine, tmp_path):
    m1 = _media("p1", 1)
    client = _FakeClient([(None, [("p1", [m1])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    # Same media downloaded twice across two runs — the second marks seen again.
    ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
    ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
    assert _count_ledger(sync_engine, source_id) == 1


def _run_with_client_error(source_id, sync_engine, tmp_path, exc):
    client = _FakeClient([], raise_exc=exc)
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
    return ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )


@pytest.mark.asyncio
async def test_drift_maps_to_api_drift_and_fails_loud(source_id, sync_engine, tmp_path):
    result = _run_with_client_error(
        source_id, sync_engine, tmp_path,
        PatreonDriftError("missing top-level 'data' key"),
    )
    assert result.success is False
    assert result.error_type == ErrorType.API_DRIFT
    assert "Patreon API changed" in (result.error_message or "")


@pytest.mark.asyncio
async def test_auth_error_maps_to_auth_error(source_id, sync_engine, tmp_path):
    result = _run_with_client_error(
        source_id, sync_engine, tmp_path,
        PatreonAuthError("HTTP 403 — auth rejected", status_code=403),
    )
    assert result.success is False
    assert result.error_type == ErrorType.AUTH_ERROR


@pytest.mark.asyncio
async def test_rate_limit_status_maps_to_rate_limited(source_id, sync_engine, tmp_path):
    result = _run_with_client_error(
        source_id, sync_engine, tmp_path,
        PatreonAPIError("HTTP 429", status_code=429),
    )
    assert result.error_type == ErrorType.RATE_LIMITED


@pytest.mark.asyncio
async def test_not_found_status_maps_to_not_found(source_id, sync_engine, tmp_path):
    result = _run_with_client_error(
        source_id, sync_engine, tmp_path,
        PatreonAPIError("HTTP 404", status_code=404),
    )
    assert result.error_type == ErrorType.NOT_FOUND


@pytest.mark.asyncio
async def test_transport_error_maps_to_network_error(source_id, sync_engine, tmp_path):
    result = _run_with_client_error(
        source_id, sync_engine, tmp_path,
        PatreonAPIError("connection reset"),  # no status_code → transport
    )
    assert result.error_type == ErrorType.NETWORK_ERROR


def test_ledger_key_video_has_no_filehash():
    video = MediaItem(
        url="https://stream.mux.com/abc.m3u8", filename="clip.mp4",
        kind="postfile", filehash=None, post_id="p9",
    )
    assert _ledger_key(video) == "p9:clip.mp4"


# --- verify_patreon_credential (the verify counterpart, plan #697) ---------


@pytest.mark.asyncio
async def test_verify_patreon_credential_unresolvable_is_inconclusive():
    from backend.app.services.patreon_ingester import verify_patreon_credential

    # No campaign id resolvable from a non-Patreon URL → can't test → None.
    ok, msg = await verify_patreon_credential("not-a-patreon-url", None, {})
    assert ok is None
    assert "campaign id" in msg.lower()


@pytest.mark.asyncio
async def test_verify_patreon_credential_delegates_to_client(monkeypatch):
    import backend.app.services.patreon_ingester as mod

    async def _fake_resolve(url, cookies_path, overrides):
        return "123", None
    monkeypatch.setattr(mod, "resolve_campaign_id_for_source", _fake_resolve)
    monkeypatch.setattr(
        mod.PatreonClient, "verify_auth",
        lambda self, campaign_id: (True, f"ok:{campaign_id}"),
    )
    ok, msg = await mod.verify_patreon_credential("https://patreon.com/x", None, {})
    assert ok is True
    assert msg == "ok:123"


# --- media-less (text-only) post capture ----------------------------------


@pytest.mark.asyncio
async def test_tick_captures_media_less_post_once(source_id, sync_engine, tmp_path):
    """A post with NO downloadable media still gets a post-only record, gated by
    the seen-ledger (synthetic post key) so a second walk doesn't re-record it."""
    client = _FakeClient([(None, [("ptext", [])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    assert result.success is True
    assert len(result.post_record_paths) == 1
    assert downloader.post_records == 1
    # The synthetic `post:ptext` key was marked seen (gates re-capture).
    assert _count_ledger(sync_engine, source_id) == 1

    # Second walk: already recorded → gated, no re-write, no new ledger row.
    client2 = _FakeClient([(None, [("ptext", [])])])
    downloader2 = _FakeDownloader(tmp_path)
    ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
    result2 = ing2.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="tick",
    )
    assert result2.post_record_paths == []
    assert downloader2.post_records == 0
    assert _count_ledger(sync_engine, source_id) == 1


@pytest.mark.asyncio
async def test_recovery_bypasses_gate_and_re_records(source_id, sync_engine, tmp_path):
    """Recovery bypasses the seen-ledger, so a media-less post is re-recorded
    (matches recovery semantics for media)."""
    client = _FakeClient([(None, [("ptext", [])])])
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
    ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
            url="https://patreon.com/ingest", mode="tick")

    client2 = _FakeClient([(None, [("ptext", [])])])
    dl2 = _FakeDownloader(tmp_path)
    ing2 = _ingester(sync_engine, tmp_path, client2, dl2)
    result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
                      url="https://patreon.com/ingest", mode="recovery")
    assert len(result.post_record_paths) == 1
    assert dl2.post_records == 1


# --- post-body schema-drift canary (#862) --------------------------------


def _media_less_posts(n):
    # A single page of n media-less posts; each still gets a post-record, so the
    # canary counts all n. Unique ids keep the seen-ledger gate from collapsing them.
    return [(None, [(f"c{i}", []) for i in range(n)])]


@pytest.mark.asyncio
async def test_body_canary_fails_run_when_all_bodies_empty(source_id, sync_engine, tmp_path):
    """#862: a native walk that records >= _CANARY_MIN_SAMPLE posts but extracts a
    body from NONE of them is the signature of a Patreon body-field rename (as
    content→content_json_string was) — fail the run RED (API_DRIFT) instead of
    silently archiving empties."""
    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE), empty_body=True)
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert result.success is False
    assert result.error_type == ErrorType.API_DRIFT
    assert "canary" in (result.error_message or "").lower()


@pytest.mark.asyncio
async def test_body_canary_silent_when_some_bodies_present(source_id, sync_engine, tmp_path):
    """No false alarm: a large walk that DOES capture bodies completes normally
    (normal operation — e.g. a gallery creator who writes captions)."""
    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE))  # bodies present
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert result.success is True
    assert result.error_type is None


@pytest.mark.asyncio
async def test_body_canary_silent_below_min_sample(source_id, sync_engine, tmp_path):
    """Tick safety: a small walk (a few new captionless posts) is below the sample
    floor and must NOT trip the canary, even with every body empty."""
    client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE - 1), empty_body=True)
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))

    result = ing.run(
        source_id=source_id, campaign_id="c1", artist_slug="ingest",
        url="https://patreon.com/ingest", mode="backfill",
    )
    assert result.success is True
    assert result.error_type is None