refactor(downloads): unify phase-2 dispatch in download_backends — A5 (plan #707)

Mirror verify_source_credential: download_backends.run_download is now the single
download entry, so that module is the ONE registry of how each platform both
downloads AND verifies — the seam that makes adding a platform a bounded job
(write its adapter construction next to its verify). The native-ingester
construction + campaign-id resolution moves out of download_service into
download_backends._run_native_ingester.

download_service.download_source drops its `if uses_native_ingester ... else
gdl.download` branch and calls one `self._run_download(...)` (a thin delegate to
run_download passing the service's gdl + sync sessionmaker). mode (tick/backfill/
recovery) is still chosen there from the backfill state machine and threaded
through. Removed the now-unused PatreonIngester / resolve_campaign_id_for_source
imports from download_service.

Tests: the phase-2 stub seam moves from svc._run_patreon_ingester to
svc._run_download (helper + the db-release test); the two native-construction
tests repoint to download_backends.run_download (patching
download_backends.resolve_campaign_id_for_source / PatreonIngester).

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

backend/app/services/download_backends.py

@@ -20,8 +20,13 @@ function regardless of platform:
 
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
 
+from .gallery_dl import DownloadResult, ErrorType
+from .patreon_ingester import PatreonIngester
+from .patreon_resolver import resolve_campaign_id_for_source
+
 # Platforms whose download + verify go through the native ingester rather than
 # gallery-dl. gallery-dl still serves every other platform (subscribestar,
 # hentaifoundry, discord, pixiv, deviantart) unchanged.
@@ -34,6 +39,111 @@ def uses_native_ingester(platform: str) -> bool:
     return platform in NATIVE_INGESTER_PLATFORMS
 
 
+async def run_download(
+    *,
+    ctx: dict,
+    source_config,
+    skip_value: bool | str,
+    mode: str | None,
+    gdl,
+    sync_session_factory,
+) -> tuple[DownloadResult, str | None]:
+    """Uniform download across backends — the download counterpart to
+    `verify_source_credential`, so this module is the ONE place that knows how
+    each platform both downloads AND verifies (the seam that makes adding a
+    platform a bounded job).
+
+    Returns `(DownloadResult, resolved_campaign_id)`; `resolved_campaign_id` is
+    non-None only when a native vanity lookup ran this call (so phase 3 caches
+    it). Native platforms route through their ingester in `mode`
+    (tick/backfill/recovery); gallery-dl platforms run the subprocess. The caller
+    (download_service) prepares `source_config`/`skip_value`/`mode` from the
+    backfill state machine and owns phase 3.
+    """
+    platform = ctx["platform"]
+    if uses_native_ingester(platform):
+        return await _run_native_ingester(
+            ctx, source_config, mode, gdl, sync_session_factory
+        )
+    result = await gdl.download(
+        url=ctx["url"],
+        artist_slug=ctx["artist_slug"],
+        platform=platform,
+        source_config=source_config,
+        cookies_path=ctx["cookies_path"],
+        auth_token=ctx["auth_token"],
+        skip_value=skip_value,
+    )
+    return result, None
+
+
+async def _run_native_ingester(
+    ctx: dict, source_config, mode: str | None, gdl, sync_session_factory,
+) -> tuple[DownloadResult, str | None]:
+    """Patreon (today the only native platform): resolve the campaign id, then run
+    the native ingester in a worker thread (it is sync requests/subprocess).
+
+    `resolved_campaign_id` is non-None only when we had to look it up from the
+    vanity URL this run, so phase 3 caches it the way the old gallery-dl retry
+    did. A campaign id we cannot resolve is a loud NOT_FOUND — never a silent
+    empty success.
+    """
+    overrides = ctx["config_overrides"] or {}
+    campaign_id, resolved_campaign_id = await resolve_campaign_id_for_source(
+        ctx["url"], ctx["cookies_path"], overrides
+    )
+    if not campaign_id:
+        return (
+            DownloadResult(
+                success=False,
+                url=ctx["url"],
+                artist_slug=ctx["artist_slug"],
+                platform="patreon",
+                error_type=ErrorType.NOT_FOUND,
+                error_message=(
+                    "Could not resolve Patreon campaign id from the source URL "
+                    "(vanity lookup failed — cookies expired or creator moved?)"
+                ),
+            ),
+            None,
+        )
+
+    # Honor the operator's existing rate-limit knobs on the native path (plan
+    # #703): the global download_rate_limit_seconds (gallery-dl's `rate_limit`,
+    # here on the gdl service) paces media downloads; page fetches use the
+    # per-source sleep_request override, else `max(0.5, rate_limit/4)` — the same
+    # API-pacing default gallery-dl applied as its `sleep-request`.
+    rate_limit = gdl._rate_limit
+    request_sleep = (
+        source_config.sleep_request
+        if source_config.sleep_request is not None
+        else max(0.5, rate_limit / 4)
+    )
+    ingester = PatreonIngester(
+        images_root=gdl.images_root,
+        cookies_path=ctx["cookies_path"],
+        session_factory=sync_session_factory,
+        validate=gdl._validate_files,
+        rate_limit=rate_limit,
+        request_sleep=request_sleep,
+    )
+    loop = asyncio.get_running_loop()
+    dl_result = await loop.run_in_executor(
+        None,
+        lambda: ingester.run(
+            source_id=ctx["source_id"],
+            campaign_id=campaign_id,
+            artist_slug=ctx["artist_slug"],
+            url=ctx["url"],
+            mode=mode,
+            resume_cursor=source_config.resume_cursor,
+            time_budget_seconds=source_config.timeout,
+            posts_base=int(overrides.get("_backfill_posts", 0)),
+        ),
+    )
+    return dl_result, resolved_campaign_id
+
+
 async def verify_source_credential(
     *,
     platform: str,

backend/app/services/download_service.py

@@ -24,7 +24,7 @@ from sqlalchemy.orm import joinedload
 
 from ..models import Artist, DownloadEvent, Source
 from .credential_service import CredentialService
-from .download_backends import uses_native_ingester
+from .download_backends import run_download, uses_native_ingester
 from .gallery_dl import (
     BACKFILL_CHUNK_SECONDS,
     BACKFILL_SKIP_VALUE,
@@ -37,8 +37,6 @@ from .gallery_dl import (
     truncate_log,
 )
 from .importer import Importer
-from .patreon_ingester import PatreonIngester
-from .patreon_resolver import resolve_campaign_id_for_source
 from .platforms import auth_type_for
 from .scheduler_service import set_platform_cooldown
 
@@ -122,32 +120,23 @@ class DownloadService:
         else:
             skip_value = TICK_SKIP_VALUE
 
-        resolved_campaign_id: str | None = None
+        # Phase 2 dispatch is uniform across backends (download_backends.
+        # run_download — the download counterpart to verify_source_credential):
+        # native platforms run their ingester in `mode` (zero per-file HEADs,
+        # native cursor/resume, loud drift detection); gallery-dl platforms run
+        # the subprocess. Either returns a DownloadResult-shaped object so phase 3
+        # is untouched. `mode` is None for gallery-dl (run_download ignores it).
+        mode: str | None = None
         if uses_native_ingester(ctx["platform"]):
-            # Native ingester (plan #697) fully replaces gallery-dl for Patreon
-            # in phase 2 — zero per-file HEADs, native cursor/resume, loud drift
-            # detection. Returns a DownloadResult-shaped object so phase 3 is
-            # untouched. The gallery-dl Patreon path + its campaign-id retry are
-            # removed at the step-5 cutover.
             if in_backfill and bypass_seen:
                 mode = "recovery"
             elif in_backfill:
                 mode = "backfill"
             else:
                 mode = "tick"
-            dl_result, resolved_campaign_id = await self._run_patreon_ingester(
-                ctx, source_config, mode,
-            )
-        else:
-            dl_result = await self.gdl.download(
-                url=ctx["url"],
-                artist_slug=ctx["artist_slug"],
-                platform=ctx["platform"],
-                source_config=source_config,
-                cookies_path=ctx["cookies_path"],
-                auth_token=ctx["auth_token"],
-                skip_value=skip_value,
-            )
+        dl_result, resolved_campaign_id = await self._run_download(
+            ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
+        )
 
         # A backfill chunk that hit its time-box but made forward progress is
         # NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as
@@ -171,82 +160,18 @@ class DownloadService:
             ctx["event_id"], ctx, dl_result, resolved_campaign_id,
         )
 
-    async def _run_patreon_ingester(
-        self, ctx: dict, source_config, mode: str,
+    async def _run_download(
+        self, *, ctx: dict, source_config, skip_value, mode: str | None,
     ) -> tuple[DownloadResult, str | None]:
-        """Phase-2 Patreon branch: resolve the campaign id, then run the native
-        ingester in a worker thread (it is sync requests/subprocess).
-
-        Returns (DownloadResult, resolved_campaign_id). `resolved_campaign_id` is
-        non-None only when we had to look it up from the vanity URL this run, so
-        phase 3 caches it on the source the same way the old gallery-dl retry did.
-        A campaign id we cannot resolve is a loud failure (NOT_FOUND) — never a
-        silent empty success.
-        """
-        overrides = ctx["config_overrides"] or {}
-        # Shared resolution path (override / id: URL / vanity lookup) — the same
-        # helper the credential-verify probe uses. resolved_campaign_id is
-        # non-None only when a vanity lookup ran, so phase 3 caches it.
-        campaign_id, resolved_campaign_id = await resolve_campaign_id_for_source(
-            ctx["url"], ctx["cookies_path"], overrides
+        """Phase-2 dispatch → `download_backends.run_download`, passing this
+        service's gdl + sync sessionmaker. Kept as a thin instance method so tests
+        can stub the whole phase-2 dispatch on the service, and so the per-backend
+        construction lives in download_backends (the backend registry)."""
+        return await run_download(
+            ctx=ctx, source_config=source_config, skip_value=skip_value, mode=mode,
+            gdl=self.gdl, sync_session_factory=self.sync_session_factory,
         )
 
-        if not campaign_id:
-            return (
-                DownloadResult(
-                    success=False,
-                    url=ctx["url"],
-                    artist_slug=ctx["artist_slug"],
-                    platform="patreon",
-                    error_type=ErrorType.NOT_FOUND,
-                    error_message=(
-                        "Could not resolve Patreon campaign id from the source "
-                        "URL (vanity lookup failed — cookies expired or creator "
-                        "moved?)"
-                    ),
-                ),
-                None,
-            )
-
-        # Honor the operator's existing rate-limit knobs on the native path
-        # (plan #703): the global download_rate_limit_seconds (gallery-dl's
-        # `rate_limit`, here on self.gdl) paces media downloads. Page fetches use
-        # the per-source sleep_request override, else `max(0.5, rate_limit/4)` —
-        # the SAME API-pacing default gallery-dl applied as its `sleep-request`
-        # (gallery_dl._get_default_config), so the rate-limited /api/posts
-        # endpoint stays politely paced without over-throttling.
-        rate_limit = self.gdl._rate_limit
-        request_sleep = (
-            source_config.sleep_request
-            if source_config.sleep_request is not None
-            else max(0.5, rate_limit / 4)
-        )
-        ingester = PatreonIngester(
-            images_root=self.gdl.images_root,
-            cookies_path=ctx["cookies_path"],
-            session_factory=self.sync_session_factory,
-            validate=self.gdl._validate_files,
-            rate_limit=rate_limit,
-            request_sleep=request_sleep,
-        )
-        loop = asyncio.get_running_loop()
-        dl_result = await loop.run_in_executor(
-            None,
-            lambda: ingester.run(
-                source_id=ctx["source_id"],
-                campaign_id=campaign_id,
-                artist_slug=ctx["artist_slug"],
-                url=ctx["url"],
-                mode=mode,
-                resume_cursor=source_config.resume_cursor,
-                time_budget_seconds=source_config.timeout,
-                # plan #704 #5: seed the live posts badge from prior chunks so the
-                # ingester writes a monotonic absolute mid-walk (it owns the key).
-                posts_base=int(overrides.get("_backfill_posts", 0)),
-            ),
-        )
-        return dl_result, resolved_campaign_id
-
     async def _phase1_setup(self, source_id: int) -> dict[str, Any]:
         source = (await self.async_session.execute(
             select(Source).options(joinedload(Source.artist)).where(Source.id == source_id)

tests/test_download_service.py

@@ -89,7 +89,7 @@ def _make_fake_dl_result(
 def _fake_gdl_with_result(result):
     fake = MagicMock()
     fake.download = AsyncMock(return_value=result)
-    # Real values for the attrs _run_patreon_ingester reads off the gdl service
+    # 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
@@ -105,19 +105,24 @@ def _fake_gdl_with_result(result):
 
 
 def _stub_patreon_ingester(svc, result, resolved_campaign_id=None):
-    """Route the Patreon phase-2 branch (plan #697 native ingester) 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. Returns a list capturing (ctx, source_config, mode)
-    per call so a test can assert mode / resume_cursor threading."""
+    """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, mode):
-        calls.append({"ctx": ctx, "source_config": source_config, "mode": mode})
+    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_patreon_ingester = fake
+    svc._run_download = fake
     return calls
 
 
@@ -285,20 +290,19 @@ async def test_patreon_resolved_campaign_id_is_cached(
 
 
 @pytest.mark.asyncio
-async def test_run_patreon_ingester_resolves_vanity_and_runs(
+async def test_run_download_native_resolves_vanity_and_runs(
     db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
 ):
-    """_run_patreon_ingester: 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_service as dl_mod
-    from backend.app.services.download_service import DownloadService
+    """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(
-        dl_mod, "resolve_campaign_id_for_source",
+        db_mod, "resolve_campaign_id_for_source",
         AsyncMock(return_value=("4242", "4242")),
     )
     run_kwargs = {}
@@ -311,21 +315,19 @@ async def test_run_patreon_ingester_resolves_vanity_and_runs(
             run_kwargs.update(kw)
             return _make_fake_dl_result(success=True, written_paths=[])
 
-    monkeypatch.setattr(dl_mod, "PatreonIngester", _FakeIngester)
+    monkeypatch.setattr(db_mod, "PatreonIngester", _FakeIngester)
 
-    svc = DownloadService(
-        async_session=db, sync_session=db_sync,
-        gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)),
-        importer=MagicMock(), cred_service=MagicMock(),
-        sync_session_factory=MagicMock(),
-    )
     ctx = {
+        "platform": "patreon",
         "source_id": source.id, "url": "https://patreon.com/alice",
         "artist_slug": "alice", "cookies_path": None,
         "config_overrides": {},
     }
-    result, resolved = await svc._run_patreon_ingester(
-        ctx, SourceConfig.from_dict({}), "tick",
+    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"
@@ -334,33 +336,29 @@ async def test_run_patreon_ingester_resolves_vanity_and_runs(
 
 
 @pytest.mark.asyncio
-async def test_run_patreon_ingester_unresolvable_fails_loud(
+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_service as dl_mod
-    from backend.app.services.download_service import DownloadService
+    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(
-        dl_mod, "resolve_campaign_id_for_source",
+        db_mod, "resolve_campaign_id_for_source",
         AsyncMock(return_value=(None, None)),
     )
-    svc = DownloadService(
-        async_session=db, sync_session=db_sync,
-        gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(),
-        sync_session_factory=MagicMock(),
-    )
     ctx = {
+        "platform": "patreon",
         "source_id": source.id, "url": "https://patreon.com/alice",
         "artist_slug": "alice", "cookies_path": None,
         "config_overrides": {},
     }
-    result, resolved = await svc._run_patreon_ingester(
-        ctx, SourceConfig.from_dict({}), "tick",
+    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
@@ -926,11 +924,11 @@ async def test_releases_db_connections_before_subprocess(
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
     )
 
-    async def fake_ingest(ctx, source_config, mode):
+    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_patreon_ingester = fake_ingest
+    svc._run_download = fake_ingest
     event_id = await svc.download_source(source.id)
 
     assert seen["closed_before_phase2"] is True