"""DownloadService orchestrator tests.

GalleryDLService is mocked to return a canned DownloadResult; tests
provide on-disk files for the importer to attach_in_place.
"""

from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from sqlalchemy import func, select

from backend.app.models import Artist, DownloadEvent, ImageRecord, ImportSettings, Source
from backend.app.services.credential_crypto import CredentialCrypto
from backend.app.services.credential_service import CredentialService
from backend.app.services.thumbnailer import Thumbnailer

pytestmark = pytest.mark.integration


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


def _make_jpg(path: Path, split: str = "h"):
    """Write a structured JPEG so phash isn't degenerate.

    Solid-color images all phash-collapse (DCT of a flat image is zero).
    Use `split='h'` (top/bottom halves) or `split='v'` (left/right) to
    give each fixture file a distinct phash.
    """
    from PIL import Image
    path.parent.mkdir(parents=True, exist_ok=True)
    im = Image.new("RGB", (64, 64), (255, 255, 255))
    px = im.load()
    if split == "h":
        for y in range(32):
            for x in range(64):
                px[x, y] = (0, 0, 0)
    else:  # 'v'
        for y in range(64):
            for x in range(32):
                px[x, y] = (0, 0, 0)
    im.save(path, "JPEG")


def _make_fake_dl_result(
    *, success=True, written_paths=None, quarantined_paths=None,
    files_downloaded=0, error_type=None, error_message=None,
    stdout="", stderr="", cursor=None, run_stats=None, posts_processed=0,
    relink_source_paths=None,
):
    return SimpleNamespace(
        relink_source_paths=relink_source_paths or [],
        success=success,
        url="https://patreon.com/alice",
        artist_slug="alice",
        platform="patreon",
        files_downloaded=files_downloaded,
        files_quarantined=len(quarantined_paths or []),
        quarantined_paths=quarantined_paths or [],
        written_paths=written_paths or [],
        stdout=stdout,
        stderr=stderr,
        return_code=0 if success else 1,
        error_type=error_type,
        error_message=error_message,
        duration_seconds=1.23,
        started_at="2026-05-20T14:00:00+00:00",
        completed_at="2026-05-20T14:01:00+00:00",
        # plan #704: native ingester now returns the cursor + run_stats
        # structurally (no more stderr `Cursor:` scraping).
        cursor=cursor,
        run_stats=run_stats,
        posts_processed=posts_processed,
    )


def _fake_gdl_with_result(result):
    fake = MagicMock()
    fake.download = AsyncMock(return_value=result)
    # Real values for the attrs run_download's native branch reads off the gdl
    # (the native pacing config, plan #703) — a MagicMock would break the
    # arithmetic (max(0.5, rate_limit/4)).
    fake._rate_limit = 3.0
    fake._validate_files = True
    fake._compute_run_stats = lambda *a, **k: {
        "exit_code": 0, "downloaded_count": len(result.written_paths),
        "skipped_count": 0, "per_item_failures": 0,
        "warning_count": 0, "tier_gated_count": 0,
    }
    fake._extract_errors_warnings = lambda *a, **k: ""
    fake._truncate_log = lambda x, **k: x
    return fake


def _stub_patreon_ingester(svc, result, resolved_campaign_id=None):
    """Route the phase-2 dispatch (download_backends.run_download via
    DownloadService._run_download, plan #707 A5) to a canned DownloadResult so
    these tests exercise download_service's phase-2-result→phase-3 handling
    (status mapping, backfill lifecycle, health) without the real ingester's
    network/DB. The ingester itself is covered by test_patreon_ingester.py; the
    native construction/resolution by the run_download tests below. Returns a list
    capturing (ctx, source_config, skip_value, mode) per call so a test can assert
    mode / resume_cursor threading."""
    calls = []

    async def fake(*, ctx, source_config, skip_value, mode):
        calls.append({
            "ctx": ctx, "source_config": source_config,
            "skip_value": skip_value, "mode": mode,
        })
        return result, resolved_campaign_id

    svc._run_download = fake
    return calls


@pytest.mark.asyncio
async def test_download_source_attaches_written_files(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
    # Structured h/v split → distinct sha256 AND distinct phash;
    # also pin phash_threshold=0 so even tiny perceptual similarity
    # between the two doesn't trigger duplicate_phash.
    _make_jpg(f1, split="h")
    _make_jpg(f2, split="v")

    result = _make_fake_dl_result(
        success=True,
        written_paths=[str(f1), str(f2)],
        files_downloaded=2,
        stdout=f"{f1}\n{f2}\n",
    )
    # The ledger is marked only once the files are in: a run killed before
    # import must leave them unmarked so the next walk imports them.
    imported_when_marked = []
    result.mark_seen_after_import = lambda: imported_when_marked.append(
        db_sync.execute(
            select(func.count(ImageRecord.id)).where(ImageRecord.path.in_([str(f1), str(f2)]))
        ).scalar_one()
    )
    fake_gdl = _fake_gdl_with_result(result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    sync_settings.phash_threshold = 0  # only exact phash collisions dedup

    importer = Importer(
        session=db_sync, images_root=images_root,
        import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root),
        settings=sync_settings,
    )
    crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
    cred_service = CredentialService(db, crypto)

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, result)
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    assert ev.files_count == 2
    assert ev.metadata_["import_summary"]["attached"] == 2
    assert ev.metadata_["run_stats"]["downloaded_count"] == 2
    assert imported_when_marked == [2]


@pytest.mark.asyncio
async def test_import_step_keeps_only_the_files_the_library_needs_on_disk(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """After the import step, a downloaded image the library took stays where it
    is, while an archive whose images were extracted (the archive is in the
    attachments store) and an image the import filters rejected are removed, so
    nothing is left on disk without a record for a later walk to import again."""
    import zipfile

    from PIL import Image

    from backend.app.models import PostAttachment
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source
    images_root = tmp_path / "images"
    post_dir = images_root / "alice" / "patreon" / "post"

    kept = post_dir / "a.jpg"
    _make_jpg(kept, split="h")
    member = tmp_path / "member.jpg"
    _make_jpg(member, split="v")
    archive = post_dir / "pack.zip"
    with zipfile.ZipFile(archive, "w") as z:
        z.writestr("art.jpg", member.read_bytes())
    tiny = post_dir / "tiny.png"
    Image.new("RGB", (16, 16), (200, 30, 30)).save(tiny)  # below min_width

    result = _make_fake_dl_result(
        success=True, written_paths=[str(kept), str(archive), str(tiny)],
        files_downloaded=3,
    )
    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    sync_settings.phash_threshold = 0
    sync_settings.min_width = 32
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
    svc = DownloadService(
        async_session=db, sync_session=db_sync, gdl=_fake_gdl_with_result(result),
        importer=importer, cred_service=CredentialService(db, crypto),
    )
    _stub_patreon_ingester(svc, result)
    await svc.download_source(source.id)

    assert kept.exists()
    assert not archive.exists()
    assert not tiny.exists()
    # The archive and its image are both still in the library.
    assert db_sync.execute(
        select(func.count(PostAttachment.id)).where(
            PostAttachment.original_filename == "pack.zip"
        )
    ).scalar_one() == 1
    assert db_sync.execute(select(func.count(ImageRecord.id))).scalar_one() == 2


@pytest.mark.asyncio
async def test_download_source_skipped_when_disabled(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source
    source.enabled = False
    await db.commit()

    fake_gdl = MagicMock()
    fake_gdl.download = AsyncMock()

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path,
        import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path),
        settings=sync_settings,
    )
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "skipped"
    assert ev.error == "source disabled"
    fake_gdl.download.assert_not_called()


@pytest.mark.asyncio
async def test_in_flight_idempotency_returns_existing_event(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    running = DownloadEvent(source_id=source.id, status="running")
    db.add(running)
    await db.commit()
    await db.refresh(running)

    fake_gdl = MagicMock()
    fake_gdl.download = AsyncMock()
    importer_stub = MagicMock()
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer_stub, cred_service=cred_service,
    )
    event_id = await svc.download_source(source.id)
    assert event_id == running.id
    fake_gdl.download.assert_not_called()


@pytest.mark.asyncio
async def test_patreon_resolved_campaign_id_is_cached(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #697: the native ingester resolves a Patreon vanity → campaign id
    on the fly (replacing gallery-dl's reactive campaign-id retry). When phase 2
    reports a freshly-resolved id, phase 3 caches it on the source so later runs
    skip the lookup. Stub the ingester to report a resolved id; assert it lands
    in config_overrides."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path,
        import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path),
        settings=sync_settings,
    )
    cred_service = CredentialService(db, MagicMock())

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
        importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(
        svc, _make_fake_dl_result(success=True, written_paths=[]),
        resolved_campaign_id="99",
    )
    await svc.download_source(source.id)

    overrides = db_sync.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    ).scalar_one()
    assert overrides.get("patreon_campaign_id") == "99"


@pytest.mark.asyncio
async def test_run_download_native_resolves_vanity_and_runs(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """run_download (native branch, plan #707 A5): with no cached campaign id,
    resolve the vanity and pass the resolved id straight to the ingester; report
    it back so phase 3 can cache it."""
    from backend.app.services import download_backends as db_mod
    from backend.app.services.gallery_dl import SourceConfig

    _artist, source = seed_artist_and_source

    monkeypatch.setattr(
        db_mod, "resolve_campaign_id_for_source",
        AsyncMock(return_value=("4242", "4242")),
    )
    run_kwargs = {}

    class _FakeIngester:
        def __init__(self, **kw):
            pass

        def run(self, **kw):
            run_kwargs.update(kw)
            return _make_fake_dl_result(success=True, written_paths=[])

    monkeypatch.setattr(db_mod, "PatreonIngester", _FakeIngester)

    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
        ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True,
        mode="tick",
        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
        sync_session_factory=MagicMock(),
    )
    assert result.success is True
    assert resolved == "4242"
    assert run_kwargs["campaign_id"] == "4242"
    assert run_kwargs["mode"] == "tick"


@pytest.mark.asyncio
async def test_run_download_native_unresolvable_fails_loud(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """A campaign id we can't resolve is a loud NOT_FOUND failure, never a
    silent empty success."""
    from backend.app.services import download_backends as db_mod
    from backend.app.services.gallery_dl import ErrorType, SourceConfig

    _artist, source = seed_artist_and_source

    monkeypatch.setattr(
        db_mod, "resolve_campaign_id_for_source",
        AsyncMock(return_value=(None, None)),
    )
    ctx = {
        "platform": "patreon",
        "source_id": source.id, "url": "https://patreon.com/alice",
        "artist_slug": "alice", "cookies_path": None, "auth_token": None,
        "config_overrides": {},
    }
    result, resolved = await db_mod.run_download(
        ctx=ctx, source_config=SourceConfig.from_dict({}), skip_value=True,
        mode="tick", gdl=MagicMock(), sync_session_factory=MagicMock(),
    )
    assert result.success is False
    assert result.error_type == ErrorType.NOT_FOUND
    assert resolved is None


# --- FC-3d: finalize hook updates Source health columns -------------------


async def _seed_source_with_health(
    db, *, failures=0, last_error=None, suffix="fz", error_type=None,
):
    artist = Artist(name=f"alice-{suffix}", slug=f"alice-{suffix}")
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform="patreon",
        url=f"https://patreon.com/alice-{suffix}", enabled=True,
        consecutive_failures=failures, last_error=last_error,
        error_type=error_type,
    )
    db.add(source)
    await db.flush()
    event = DownloadEvent(source_id=source.id, status="running")
    db.add(event)
    await db.commit()
    return source.id, event.id


@pytest.mark.asyncio
async def test_finalize_ok_resets_failures_and_clears_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=3, last_error="prev", suffix="ok",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="ok",
        error_message=None,
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 0
    assert row.last_error is None
    assert row.last_checked_at is not None


@pytest.mark.asyncio
async def test_finalize_error_increments_failures_and_sets_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=2, last_error=None, suffix="err",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="error",
        error_message="boom",
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 3
    assert row.last_error == "boom"
    assert row.last_checked_at is not None


@pytest.mark.asyncio
async def test_rate_limited_cooldown_honors_retry_after(db, monkeypatch):
    """plan #708 B1: a RATE_LIMITED result with a Retry-After stamps the platform
    cooldown at that duration, clamped to [60, 3600]; no hint → the flat default."""
    from unittest.mock import AsyncMock, MagicMock

    from backend.app.services import download_service as dl_mod
    from backend.app.services.download_service import DownloadService

    source_id, _event_id = await _seed_source_with_health(db, suffix="rl")
    cooldown = AsyncMock()
    monkeypatch.setattr(dl_mod, "set_platform_cooldown", cooldown)
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )

    async def _health(retry):
        cooldown.reset_mock()
        await svc._update_source_health(
            source_id=source_id, status="error", error_message="rate limited",
            error_type="rate_limited", retry_after_seconds=retry,
        )

    await _health(120.0)
    assert cooldown.await_args.kwargs["seconds"] == 120  # honored as-is
    await _health(5.0)
    assert cooldown.await_args.kwargs["seconds"] == 60  # floored
    await _health(99999.0)
    assert cooldown.await_args.kwargs["seconds"] == 3600  # capped
    await _health(None)
    assert "seconds" not in cooldown.await_args.kwargs  # default cooldown


@pytest.mark.asyncio
async def test_finalize_skipped_preserves_failures_clears_error(db):
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=4, last_error="stale", suffix="skip",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._finalize_event_and_source(
        event_id=event_id, source_id=source_id, status="skipped",
        error_message=None,
    )

    row = (await db.execute(
        select(
            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
        ).where(Source.id == source_id)
    )).one()
    assert row.consecutive_failures == 4
    assert row.last_error is None
    assert row.last_checked_at is not None


# --- Plan #693: backfill state machine (time-boxed chunks, run-until-done) ---


def _backfill_svc(db, db_sync, tmp_path, result):
    """DownloadService wired with a real importer (so empty written_paths just
    attaches nothing) whose Patreon phase-2 branch is stubbed to return `result`.

    The seeded source is Patreon, so phase 2 routes to the native ingester
    (plan #697); the stub returns the canned result + captures the per-call
    (ctx, source_config, mode). Returns (svc, ingester_calls)."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    images_root = tmp_path / "images"
    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
    fake_gdl = _fake_gdl_with_result(result)
    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    calls = _stub_patreon_ingester(svc, result)
    return svc, calls


@pytest.mark.asyncio
async def test_backfill_chunk_progress_advances_cursor(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A running backfill chunk that didn't finish but advanced (new cursor)
    stays 'running', checkpoints the cursor, bumps the chunk counter, and
    spends one safety-cap chunk."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[], files_downloaded=0,
        cursor="03:PAGE2:xyz",
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "running"
    assert co.get("_backfill_cursor") == "03:PAGE2:xyz"
    assert co.get("_backfill_chunks") == 1
    assert remaining == 4


@pytest.mark.asyncio
async def test_phase_3_leaves_imports_past_its_deadline_for_the_next_chunk(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """A walk that reached its end handed phase 3 more imports than fit before
    the deadline. Phase 3 imports what fits, leaves the rest on disk and
    unmarked, keeps the backfill running, and resumes it on the page holding
    the first file it left."""
    import backend.app.services.download_service as ds

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:PAGE1"}
    source.backfill_runs_remaining = 5
    await db.commit()

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
    _make_jpg(f1, split="h")
    _make_jpg(f2, split="v")
    result = _make_fake_dl_result(
        success=True, written_paths=[str(f1), str(f2)], cursor="03:PAGE3",
    )
    result.written_cursors = ["03:PAGE1", "03:PAGE2"]
    left_unmarked = []
    result.mark_seen_after_import = lambda skip_paths=(): left_unmarked.append(
        set(skip_paths)
    )

    # The clock is past the deadline once the first import is done.
    past = ds.PHASE3_DEADLINE_SECONDS + 1
    ticks = iter([0.0, 0.0])
    monkeypatch.setattr(ds, "time", SimpleNamespace(monotonic=lambda: next(ticks, past)))

    svc, _ = _backfill_svc(db, db_sync, tmp_path, result)
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    assert ev.files_count == 1
    assert "1 file(s) left for the next chunk" in ev.metadata_["stdout"]
    assert left_unmarked == [{str(f2)}]
    assert f2.exists()
    assert db_sync.execute(
        select(func.count(ImageRecord.id)).where(ImageRecord.path == str(f2))
    ).scalar_one() == 0

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "running"
    assert co.get("_backfill_cursor") == "03:PAGE2"
    assert remaining == 4


@pytest.mark.asyncio
async def test_backfill_state_running_selects_backfill_mode_and_resumes(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """state=='running' drives backfill mode (chunk budget) and threads the
    stored cursor to the native ingester as the resume point."""
    from backend.app.services.gallery_dl import BACKFILL_CHUNK_SECONDS

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:RESUME:here"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[],
        cursor="03:RESUME2:next",
    ))
    await svc.download_source(source.id)

    assert len(calls) == 1
    assert calls[0]["mode"] == "backfill"
    assert calls[0]["source_config"].resume_cursor == "03:RESUME:here"
    assert calls[0]["source_config"].timeout == BACKFILL_CHUNK_SECONDS

    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_cursor") == "03:RESUME2:next"


@pytest.mark.asyncio
async def test_backfill_completes_when_every_post_was_tier_gated(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A fully-paywalled backfill chunk still COMPLETES.

    The completion test used to require `error_type is None`, so classifying a
    gated walk as TIER_LIMITED would have dropped it into the not-finished
    branch: zero files downloaded means `advanced` is False, two strikes marks
    it 'stalled', and the creator we can see least becomes the one we re-walk
    most. `walk_completed` admits informational classes for exactly this.
    """
    from backend.app.services.gallery_dl import ErrorType

    _artist, source = seed_artist_and_source
    source.config_overrides = {
        "_backfill_state": "running", "_backfill_cursor": "03:NEAR:bottom",
    }
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
        error_type=ErrorType.TIER_LIMITED,
        error_message="Subscription tier does not grant access to 9 posts",
    ))
    await svc.download_source(source.id)

    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_cursor" not in co


@pytest.mark.asyncio
async def test_ok_run_keeps_the_tier_limited_chip_but_not_the_failure_state(
    db,
):
    """An informational class survives a successful run; a failure class doesn't.

    `status == "ok"` clears `error_type` (alembic 0032) because it is the
    failure-class chip. tier_limited rides an OK run, so that clear is what made
    FailingSourcesCard's `tier_limited` palette entry unreachable — the chip was
    wiped by the same success that produced it. It must persist while the
    failure bookkeeping stays clean: no accrued failures, no last_error.
    """
    from backend.app.services.download_service import DownloadService

    source_id, event_id = await _seed_source_with_health(
        db, failures=3, last_error="prev", suffix="gated",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._update_source_health(
        source_id=source_id, status="ok", error_message=None,
        error_type="tier_limited",
    )

    row = (await db.execute(
        select(
            Source.error_type, Source.consecutive_failures, Source.last_error,
        ).where(Source.id == source_id)
    )).one()
    assert row.error_type == "tier_limited"
    assert row.consecutive_failures == 0
    assert row.last_error is None


@pytest.mark.asyncio
async def test_ok_run_still_clears_a_real_failure_class(db):
    """The exemption is narrow: a genuine failure class is still cleared on OK,
    so a recovered source stops showing a stale chip."""
    from backend.app.services.download_service import DownloadService

    source_id, _event_id = await _seed_source_with_health(
        db, failures=2, last_error="boom", suffix="recovered",
        error_type="auth_error",
    )
    svc = DownloadService(
        async_session=db, sync_session=None,
        gdl=None, importer=None, cred_service=None,
    )
    await svc._update_source_health(
        source_id=source_id, status="ok", error_message=None,
        error_type=None,
    )

    row = (await db.execute(
        select(Source.error_type, Source.consecutive_failures)
        .where(Source.id == source_id)
    )).one()
    assert row.error_type is None
    assert row.consecutive_failures == 0


@pytest.mark.asyncio
async def test_backfill_clean_exit_marks_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A clean rc=0 chunk = gallery-dl reached the bottom → state 'complete',
    cursor cleared, returns to tick mode."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:NEAR:bottom"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_cursor" not in co
    assert remaining == 0


@pytest.mark.asyncio
async def test_backfill_cap_exhaustion_stalls(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A progressing chunk that spends the last safety-cap chunk without
    reaching the bottom pauses as 'stalled' (doesn't loop forever)."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 1
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, written_paths=[],
        cursor="03:MORE:left",
    ))
    await svc.download_source(source.id)

    remaining, co = (await db.execute(
        select(Source.backfill_runs_remaining, Source.config_overrides)
        .where(Source.id == source.id)
    )).one()
    assert co.get("_backfill_state") == "stalled"
    assert remaining == 0


@pytest.mark.asyncio
async def test_backfill_stuck_guard_stalls_after_two_no_advance(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """Two consecutive chunks that fail to advance the cursor → 'stalled',
    cursor cleared, so a wedged walk can't re-strand forever."""
    from unittest.mock import MagicMock

    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "stuck"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )
    ctx = {
        "source_id": source.id, "platform": "patreon",
        "config_overrides": {"_backfill_state": "running", "_backfill_cursor": "stuck"},
        "backfill_runs_remaining": 5,
    }
    stuck = _make_fake_dl_result(success=False, cursor="stuck")

    await svc._apply_backfill_lifecycle(ctx, stuck)
    await db.commit()
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "running"
    assert co.get("_backfill_cursor_stalls") == 1

    ctx["config_overrides"] = dict(co)
    await svc._apply_backfill_lifecycle(ctx, stuck)
    await db.commit()
    co2 = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co2.get("_backfill_state") == "stalled"
    assert "_backfill_cursor" not in co2


@pytest.mark.asyncio
async def test_backfill_lifecycle_leaves_posts_to_ingester(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #704 (#5), revised: _backfill_posts is OWNED by the ingester now (it
    writes a monotonic absolute live mid-walk), so the post-chunk lifecycle must
    NOT touch it — it carries the ingester's committed value forward unchanged."""
    from unittest.mock import MagicMock

    from backend.app.services.download_service import DownloadService

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_posts": 12}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
    )
    ctx = {
        "source_id": source.id, "platform": "patreon",
        "config_overrides": {"_backfill_state": "running", "_backfill_posts": 12},
        "backfill_runs_remaining": 5,
    }
    await svc._apply_backfill_lifecycle(
        ctx, _make_fake_dl_result(success=False, cursor="C1", posts_processed=10),
    )
    await db.commit()
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_posts") == 12  # untouched — the ingester owns it


@pytest.mark.asyncio
async def test_backfill_timeout_chunk_reclassified_to_ok(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A backfill chunk that hit its time-box (TIMEOUT) but advanced is
    reclassified to PARTIAL → event status 'ok' (progress, not error), and
    consecutive_failures is not bumped."""
    from backend.app.services.gallery_dl import ErrorType

    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running"}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=False, files_downloaded=3,
        error_type=ErrorType.TIMEOUT,
        error_message="Download timed out",
        cursor="03:NEXT:page",
    ))
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    failures = (await db.execute(
        select(Source.consecutive_failures).where(Source.id == source.id)
    )).scalar_one()
    assert failures == 0


@pytest.mark.asyncio
async def test_tick_mode_when_not_running_leaves_state_untouched(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """No _backfill_state → not backfilling; the lifecycle is a no-op and the
    counter/state aren't mutated."""
    _artist, source = seed_artist_and_source
    assert (source.config_overrides or {}).get("_backfill_state") is None

    svc, _ = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert (co or {}).get("_backfill_state") is None


@pytest.mark.asyncio
async def test_recovery_mode_selected_and_flag_cleared_on_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """plan #697: `_backfill_bypass_seen` alongside a running backfill selects
    recovery mode (ingester bypasses the seen-ledger). A clean rc=0 walk
    completes the shared lifecycle AND clears the bypass flag so the next tick
    honors the ledger again."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_bypass_seen": True}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    assert calls[0]["mode"] == "recovery"
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_bypass_seen" not in co


@pytest.mark.asyncio
async def test_recapture_mode_selected_and_flag_cleared_on_complete(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """#830: `_backfill_recapture` alongside a running backfill selects recapture
    mode. A clean rc=0 walk completes the shared lifecycle AND clears the
    recapture flag so the next tick is a plain tick again."""
    _artist, source = seed_artist_and_source
    source.config_overrides = {"_backfill_state": "running", "_backfill_recapture": True}
    source.backfill_runs_remaining = 5
    await db.commit()

    svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result(
        success=True, written_paths=[], files_downloaded=0,
    ))
    await svc.download_source(source.id)

    assert calls[0]["mode"] == "recapture"
    co = (await db.execute(
        select(Source.config_overrides).where(Source.id == source.id)
    )).scalar_one()
    assert co.get("_backfill_state") == "complete"
    assert "_backfill_recapture" not in co


@pytest.mark.asyncio
async def test_partial_error_type_maps_to_ok_status(
    db, db_sync, tmp_path, seed_artist_and_source,
):
    """A non-zero gallery-dl exit accompanied by PARTIAL error_type
    (real files were downloaded before the run was cut short) is treated
    as status=ok by the orchestrator — next tick continues."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.gallery_dl import ErrorType
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    _make_jpg(f1, split="h")

    fake_result = _make_fake_dl_result(
        success=False, written_paths=[str(f1)], files_downloaded=1,
        error_type=ErrorType.PARTIAL,
        error_message="Downloaded 1 file; run did not complete in budget",
        stdout=f"{f1}\n",
    )
    fake_gdl = _fake_gdl_with_result(fake_result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, fake_result)
    event_id = await svc.download_source(source.id)

    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"
    assert ev.error is None
    # And the source's failure counter isn't bumped — PARTIAL isn't a failure.
    src_after = (await db.execute(
        select(Source.consecutive_failures).where(Source.id == source.id)
    )).scalar_one()
    assert src_after == 0


@pytest.mark.asyncio
async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """Operator-flagged 2026-06-01: downloaded images stayed at
    thumbnail_path=NULL until periodic backfill swept them up, surfacing
    as broken-thumbnail tiles in the gallery. Importer.attach_in_place
    deliberately skips inline generation; the calling code MUST enqueue
    the thumbnail + ML tasks per attached image (matching the pattern in
    tasks/import_file.py)."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    images_root = tmp_path / "images"
    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
    _make_jpg(f1, split="h")
    _make_jpg(f2, split="v")

    result = _make_fake_dl_result(
        success=True, written_paths=[str(f1), str(f2)],
        files_downloaded=2, stdout=f"{f1}\n{f2}\n",
    )
    fake_gdl = _fake_gdl_with_result(result)

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    sync_settings.phash_threshold = 0
    importer = Importer(
        session=db_sync, images_root=images_root, import_root=images_root,
        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
    )
    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))

    # Capture the IDs that the orchestrator hands off to each Celery task.
    # The .delay() shim runs inside DownloadService._phase3_persist (lazy
    # imports under ..tasks.thumbnail / ..tasks.ml), so monkeypatch the
    # symbols on those modules.
    thumb_calls: list[int] = []
    ml_calls: list[int] = []
    from backend.app.tasks import ml as ml_mod
    from backend.app.tasks import thumbnail as thumb_mod
    monkeypatch.setattr(
        thumb_mod.generate_thumbnail, "delay",
        lambda image_id: thumb_calls.append(image_id),
    )
    monkeypatch.setattr(
        ml_mod.embed_image, "delay",
        lambda image_id: ml_calls.append(image_id),
    )

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )
    _stub_patreon_ingester(svc, result)
    await svc.download_source(source.id)

    # Two files attached → two thumbnail enqueues + two ML enqueues, IDs
    # match the actually-imported records (not ad-hoc — drawn from the
    # importer's returned image_id so future supersede paths stay covered).
    assert len(thumb_calls) == 2
    assert len(ml_calls) == 2
    assert sorted(thumb_calls) == sorted(ml_calls)


@pytest.mark.asyncio
async def test_releases_db_connections_before_subprocess(
    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
):
    """Phase 1's DB connections must be released BEFORE the (multi-minute)
    phase-2 fetch, so they don't idle-die and strand phase 3 with
    ConnectionDoesNotExistError (Anduo #40014). Spy on the async session close
    and assert it happened before the phase-2 work (here the native Patreon
    ingester) runs; phase 3 must still finalize the event."""
    from backend.app.services.download_service import DownloadService
    from backend.app.services.importer import Importer

    _artist, source = seed_artist_and_source

    closed = {"async": False}
    orig_close = db.close

    async def spy_close():
        closed["async"] = True
        await orig_close()

    monkeypatch.setattr(db, "close", spy_close)

    seen = {}
    fake_gdl = _fake_gdl_with_result(
        _make_fake_dl_result(success=True, written_paths=[], stdout="")
    )

    sync_settings = db_sync.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    importer = Importer(
        session=db_sync, images_root=tmp_path, import_root=tmp_path,
        thumbnailer=Thumbnailer(images_root=tmp_path), settings=sync_settings,
    )
    cred_service = CredentialService(
        db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
    )

    svc = DownloadService(
        async_session=db, sync_session=db_sync,
        gdl=fake_gdl, importer=importer, cred_service=cred_service,
    )

    async def fake_ingest(*, ctx, source_config, skip_value, mode):
        seen["closed_before_phase2"] = closed["async"]
        return _make_fake_dl_result(success=True, written_paths=[], stdout=""), None

    svc._run_download = fake_ingest
    event_id = await svc.download_source(source.id)

    assert seen["closed_before_phase2"] is True
    ev = (await db.execute(
        select(DownloadEvent).where(DownloadEvent.id == event_id)
    )).scalar_one()
    assert ev.status == "ok"


@pytest.mark.asyncio
async def test_async_task_engine_uses_nullpool():
    """The per-task async engine must use NullPool so phase 3 re-acquires a
    fresh connection instead of a pooled one the server reaped during the
    long subprocess (Anduo #40014)."""
    from sqlalchemy.pool import NullPool

    from backend.app.tasks._async_session import async_session_factory

    _factory, engine = async_session_factory()
    try:
        assert isinstance(engine.sync_engine.pool, NullPool)
    finally:
        await engine.dispose()