GallerySubscriber/docs/superpowers/plans/2026-04-18-patreon-campaign-resolver.md

Patreon Campaign-ID Auto-Resolver Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Auto-heal gallery-dl "Failed to extract campaign ID" failures on Patreon sources by resolving the campaign ID via Patreon's campaigns API, caching it in Source.metadata_, and retrying the download once in the same task run.

Architecture: A new patreon_resolver service performs an authenticated HTTP GET to patreon.com/api/campaigns?filter[vanity]=… using the stored Patreon cookies, returning the campaign ID (or None on any failure). The Celery download task adds three pure helpers (URL rewrite, error-pattern match, vanity extraction) and a retry hook between its existing Phase 2 (download execution) and Phase 3 (DB write) blocks. Cached campaign IDs are written to Source.metadata_['patreon_campaign_id'] on retry success; subsequent runs always use patreon.com/id:<id> as the effective URL.

Tech Stack: Python 3.13, aiohttp, pytest + pytest-asyncio, aioresponses (new dev dep), SQLAlchemy JSONB, Celery.

---

Spec Reference

Design spec: docs/superpowers/specs/2026-04-18-patreon-campaign-resolver-design.md. Consult it for motivation, non-goals, and rationale. This plan implements that spec faithfully.

---

File Structure

Create:

• backend/pytest.ini — pytest config (testpaths, asyncio_mode, pythonpath)
• backend/tests/__init__.py — empty package marker
• backend/tests/conftest.py — pytest fixtures / path wiring
• backend/tests/services/__init__.py — empty
• backend/tests/services/test_patreon_resolver.py — unit tests
• backend/tests/tasks/__init__.py — empty
• backend/tests/tasks/test_downloads_helpers.py — unit tests for the three pure helpers
• backend/tests/tasks/test_downloads_patreon_retry.py — integration test for retry hook
• backend/app/services/patreon_resolver.py — resolve_campaign_id() async function

Modify:

• backend/requirements.txt — add aioresponses under the # Testing section
• backend/app/tasks/downloads.py — add helpers, URL rewrite, retry hook inside _download_source_async (Phase 2 block at lines 211–228)

---

Task 1: Bootstrap backend pytest infrastructure

Why: The backend has pytest and pytest-asyncio listed in requirements.txt but no tests, no test directory, and no pytest config. Before writing any test, the infrastructure must exist. This task adds the skeleton and a single sanity test that proves the config works.

Files:

• Create: backend/pytest.ini

• Create: backend/tests/__init__.py

• Create: backend/tests/conftest.py

• Create: backend/tests/test_sanity.py

• Modify: backend/requirements.txt (add aioresponses)

• Step 1: Add aioresponses to requirements.txt

Edit backend/requirements.txt. Under the existing # Testing section (lines 26-28), append aioresponses so the block reads:

# Testing
pytest
pytest-asyncio
aioresponses

• Step 2: Install the new dep in the local venv (if one exists)

Run (from repo root): pip install aioresponses (only if the engineer has a local venv — otherwise the Dockerfile will pick it up on next build).

Expected: either Successfully installed aioresponses-… or, if already installed, Requirement already satisfied.

• Step 3: Create backend/pytest.ini

Create with exactly these contents:

[pytest]
testpaths = tests
asyncio_mode = auto
pythonpath = .
filterwarnings =
    ignore::DeprecationWarning:celery.*

Notes:

• pythonpath = . means pytest adds backend/ to sys.path when run from that directory, so import app.services.patreon_resolver works in tests.

• asyncio_mode = auto means async def test_* functions don't need @pytest.mark.asyncio decorators.

• Step 4: Create empty backend/tests/__init__.py

Empty file. Required so pytest discovers the directory as a package.

• Step 5: Create backend/tests/conftest.py

"""Shared pytest fixtures for backend tests."""
import asyncio
import pytest


@pytest.fixture
def anyio_backend():
    return "asyncio"

This is minimal — we don't need DB fixtures for this feature. It's here for future tests.

• Step 6: Create sanity test backend/tests/test_sanity.py

"""Sanity test to verify pytest infrastructure works."""
import pytest


def test_sync_sanity():
    assert 1 + 1 == 2


async def test_async_sanity():
    await asyncio.sleep(0)
    assert True

Add import asyncio at the top (required by the async test):

"""Sanity test to verify pytest infrastructure works."""
import asyncio
import pytest


def test_sync_sanity():
    assert 1 + 1 == 2


async def test_async_sanity():
    await asyncio.sleep(0)
    assert True

• Step 7: Run sanity tests

Run (from backend/): pytest tests/test_sanity.py -v

Expected output:

tests/test_sanity.py::test_sync_sanity PASSED
tests/test_sanity.py::test_async_sanity PASSED

If test_async_sanity fails with "coroutine was never awaited", the asyncio_mode = auto line isn't being read — re-check pytest.ini path and contents.

• Step 8: Commit

git add backend/pytest.ini backend/requirements.txt backend/tests/__init__.py backend/tests/conftest.py backend/tests/test_sanity.py
git commit -m "test: bootstrap backend pytest infrastructure with sanity tests"

---

Task 2: Implement patreon_resolver service (TDD)

Why: This is the core new unit: an async function that turns a Patreon vanity into a campaign ID. Testable in complete isolation from Celery, DB, and gallery-dl.

Files:

• Create: backend/tests/services/__init__.py

• Create: backend/tests/services/test_patreon_resolver.py

• Create: backend/app/services/patreon_resolver.py

• Step 1: Create empty backend/tests/services/__init__.py

• Step 2: Write failing tests in backend/tests/services/test_patreon_resolver.py

"""Unit tests for the Patreon campaign-ID resolver."""
import aiohttp
import pytest
from aioresponses import aioresponses

from app.services.patreon_resolver import resolve_campaign_id


CAMPAIGNS_URL = "https://www.patreon.com/api/campaigns?filter%5Bvanity%5D=mstsu&fields%5Bcampaign%5D=name"


async def test_happy_path_returns_campaign_id():
    with aioresponses() as mock:
        mock.get(
            CAMPAIGNS_URL,
            payload={"data": [{"id": "12345", "type": "campaign", "attributes": {"name": "MSTSU"}}]},
        )
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result == "12345"


async def test_empty_data_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, payload={"data": []})
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_missing_data_key_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, payload={"errors": [{"code": 1}]})
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_missing_id_field_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, payload={"data": [{"type": "campaign"}]})
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_http_401_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, status=401, body="Unauthorized")
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_network_error_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, exception=aiohttp.ClientConnectionError("boom"))
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_timeout_returns_none():
    import asyncio
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, exception=asyncio.TimeoutError())
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_invalid_json_returns_none():
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, body="<html>not json</html>", content_type="text/html")
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result is None


async def test_missing_cookies_path_proceeds_without_jar(tmp_path):
    """When cookies_path is None, the request still goes out (cookies are optional)."""
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, payload={"data": [{"id": "99", "type": "campaign"}]})
        result = await resolve_campaign_id("mstsu", cookies_path=None)
    assert result == "99"


async def test_nonexistent_cookies_path_proceeds_without_jar(tmp_path):
    """When cookies_path points to a missing file, don't crash — just skip the jar."""
    ghost = tmp_path / "does-not-exist.txt"
    with aioresponses() as mock:
        mock.get(CAMPAIGNS_URL, payload={"data": [{"id": "99"}]})
        result = await resolve_campaign_id("mstsu", cookies_path=str(ghost))
    assert result == "99"

• Step 3: Run tests to verify they fail

Run (from backend/): pytest tests/services/test_patreon_resolver.py -v

Expected: ALL FAIL with ModuleNotFoundError: No module named 'app.services.patreon_resolver' or ImportError: cannot import name 'resolve_campaign_id'.

• Step 4: Implement backend/app/services/patreon_resolver.py

"""Patreon campaign-ID resolver.

When gallery-dl fails with "Failed to extract campaign ID" on a Patreon
source, call resolve_campaign_id(vanity, cookies_path) to look up the
campaign ID via Patreon's campaigns API. The caller can then retry with
the id-URL form (patreon.com/id:<id>) to bypass the broken vanity lookup.
"""
import asyncio
import http.cookiejar
import logging
import os
from typing import Optional

import aiohttp

logger = logging.getLogger(__name__)

_CAMPAIGNS_URL = "https://www.patreon.com/api/campaigns"
_USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
)
_REQUEST_TIMEOUT_SECONDS = 10.0


def _load_cookie_jar(cookies_path: Optional[str]) -> Optional[aiohttp.CookieJar]:
    """Load a Netscape-format cookie file into an aiohttp CookieJar.

    Returns None if the path is None, the file is missing, or parsing fails.
    Never raises.
    """
    if not cookies_path or not os.path.isfile(cookies_path):
        return None
    try:
        moz = http.cookiejar.MozillaCookieJar(cookies_path)
        moz.load(ignore_discard=True, ignore_expires=True)
    except (OSError, http.cookiejar.LoadError) as e:
        logger.debug(f"Could not load cookies from {cookies_path}: {e}")
        return None

    jar = aiohttp.CookieJar()
    for cookie in moz:
        jar.update_cookies(
            {cookie.name: cookie.value},
            response_url=aiohttp.helpers.URL(
                f"https://{cookie.domain.lstrip('.')}/"
            ),
        )
    return jar


async def resolve_campaign_id(
    vanity: str,
    cookies_path: Optional[str],
) -> Optional[str]:
    """Resolve a Patreon vanity (creator slug) to a campaign ID.

    Returns the campaign ID as a string, or None on any failure (network
    error, auth error, empty response, unexpected shape). Never raises.
    """
    jar = _load_cookie_jar(cookies_path)
    headers = {
        "User-Agent": _USER_AGENT,
        "Accept": "application/vnd.api+json",
    }
    params = {
        "filter[vanity]": vanity,
        "fields[campaign]": "name",
    }
    timeout = aiohttp.ClientTimeout(total=_REQUEST_TIMEOUT_SECONDS)

    try:
        async with aiohttp.ClientSession(
            cookie_jar=jar,
            headers=headers,
            timeout=timeout,
        ) as session:
            async with session.get(_CAMPAIGNS_URL, params=params) as resp:
                if resp.status != 200:
                    logger.warning(
                        f"Patreon campaigns API returned HTTP {resp.status} for vanity={vanity}"
                    )
                    return None
                try:
                    payload = await resp.json(content_type=None)
                except (aiohttp.ContentTypeError, ValueError) as e:
                    logger.warning(
                        f"Patreon campaigns API returned non-JSON for vanity={vanity}: {e}"
                    )
                    return None
    except asyncio.TimeoutError:
        logger.warning(f"Patreon campaigns API timed out for vanity={vanity}")
        return None
    except aiohttp.ClientError as e:
        logger.warning(
            f"Patreon campaigns API request failed for vanity={vanity}: "
            f"{type(e).__name__}: {e}"
        )
        return None

    data = payload.get("data") if isinstance(payload, dict) else None
    if not isinstance(data, list) or not data:
        logger.warning(
            f"Patreon campaigns API returned empty/missing data for vanity={vanity}: "
            f"payload keys={list(payload) if isinstance(payload, dict) else type(payload).__name__}"
        )
        return None

    first = data[0] if isinstance(data[0], dict) else None
    campaign_id = first.get("id") if first else None
    if not isinstance(campaign_id, str) or not campaign_id:
        logger.warning(
            f"Patreon campaigns API response missing 'id' for vanity={vanity}: "
            f"first entry keys={list(first) if isinstance(first, dict) else type(first).__name__}"
        )
        return None

    logger.info(f"Resolved Patreon vanity={vanity} → campaign_id={campaign_id}")
    return campaign_id

• Step 5: Run tests to verify they pass

Run (from backend/): pytest tests/services/test_patreon_resolver.py -v

Expected: ALL PASS (10 tests).

If test_missing_cookies_path_proceeds_without_jar or test_nonexistent_cookies_path_proceeds_without_jar fail, the guard in _load_cookie_jar isn't short-circuiting correctly.

• Step 6: Commit

git add backend/tests/services/__init__.py backend/tests/services/test_patreon_resolver.py backend/app/services/patreon_resolver.py
git commit -m "feat(patreon): add campaign-ID resolver service"

---

Task 3: Add URL-rewrite and error-pattern helpers to downloads.py (TDD)

Why: Three pure functions encapsulate the Patreon quirk: rewriting the URL when a cached ID exists, matching the failure pattern, and extracting a vanity. Testing them in isolation (without mocking Celery, DB, or gallery-dl) keeps the logic simple and debuggable.

Files:

• Create: backend/tests/tasks/__init__.py

• Create: backend/tests/tasks/test_downloads_helpers.py

• Modify: backend/app/tasks/downloads.py (add import re if absent; append helpers after imports)

• Step 1: Create empty backend/tests/tasks/__init__.py

• Step 2: Write failing tests in backend/tests/tasks/test_downloads_helpers.py

"""Unit tests for the pure helpers added to downloads.py."""
from types import SimpleNamespace

from app.tasks.downloads import (
    _effective_patreon_url,
    _extract_vanity_from_url,
    _looks_like_campaign_id_failure,
)


# --- _effective_patreon_url -------------------------------------------------

def test_effective_url_patreon_with_cached_id_uses_id_url():
    assert (
        _effective_patreon_url("patreon", "https://www.patreon.com/mstsu", {"patreon_campaign_id": "12345"})
        == "https://www.patreon.com/id:12345"
    )


def test_effective_url_patreon_without_cached_id_uses_original():
    assert (
        _effective_patreon_url("patreon", "https://www.patreon.com/mstsu", {})
        == "https://www.patreon.com/mstsu"
    )


def test_effective_url_patreon_none_metadata_uses_original():
    assert (
        _effective_patreon_url("patreon", "https://www.patreon.com/mstsu", None)
        == "https://www.patreon.com/mstsu"
    )


def test_effective_url_non_patreon_ignores_cached_id():
    # Even if campaign_id somehow got set on a non-patreon source, don't rewrite.
    assert (
        _effective_patreon_url("subscribestar", "https://subscribestar.adult/foo", {"patreon_campaign_id": "12345"})
        == "https://subscribestar.adult/foo"
    )


# --- _looks_like_campaign_id_failure ---------------------------------------

def _result(stdout="", stderr=""):
    return SimpleNamespace(stdout=stdout, stderr=stderr)


def test_failure_pattern_in_stderr():
    assert _looks_like_campaign_id_failure(
        _result(stderr="[patreon][error] Failed to extract campaign ID")
    ) is True


def test_failure_pattern_in_stdout():
    assert _looks_like_campaign_id_failure(
        _result(stdout="something Failed to extract campaign ID something")
    ) is True


def test_failure_pattern_case_insensitive():
    assert _looks_like_campaign_id_failure(
        _result(stderr="FAILED TO EXTRACT CAMPAIGN ID")
    ) is True


def test_failure_pattern_absent():
    assert _looks_like_campaign_id_failure(
        _result(stderr="HTTP 401 Unauthorized")
    ) is False


def test_failure_pattern_empty():
    assert _looks_like_campaign_id_failure(_result()) is False


def test_failure_pattern_none_streams():
    assert _looks_like_campaign_id_failure(
        SimpleNamespace(stdout=None, stderr=None)
    ) is False


# --- _extract_vanity_from_url ----------------------------------------------

def test_vanity_plain_patreon_url():
    assert _extract_vanity_from_url("https://www.patreon.com/mstsu") == "mstsu"


def test_vanity_with_c_prefix():
    assert _extract_vanity_from_url("https://www.patreon.com/c/mstsu/posts") == "mstsu"


def test_vanity_with_c_prefix_no_trailing_path():
    assert _extract_vanity_from_url("https://www.patreon.com/c/mstsu") == "mstsu"


def test_vanity_without_www():
    assert _extract_vanity_from_url("https://patreon.com/mstsu") == "mstsu"


def test_vanity_http_scheme():
    assert _extract_vanity_from_url("http://www.patreon.com/mstsu") == "mstsu"


def test_vanity_id_url_returns_none():
    assert _extract_vanity_from_url("https://www.patreon.com/id:12345") is None


def test_vanity_bare_domain_returns_none():
    assert _extract_vanity_from_url("https://www.patreon.com/") is None


def test_vanity_unrelated_url_returns_none():
    assert _extract_vanity_from_url("https://subscribestar.adult/mstsu") is None


def test_vanity_ignores_trailing_query_and_hash():
    assert _extract_vanity_from_url("https://www.patreon.com/mstsu?x=1") == "mstsu"
    assert _extract_vanity_from_url("https://www.patreon.com/mstsu#anchor") == "mstsu"

• Step 3: Run tests to verify they fail

Run (from backend/): pytest tests/tasks/test_downloads_helpers.py -v

Expected: ALL FAIL with ImportError: cannot import name '_effective_patreon_url' from 'app.tasks.downloads'.

• Step 4: Add helpers to backend/app/tasks/downloads.py

Add import re to the top of the file if it isn't already imported (it is not — verify against the existing imports at lines 1-22). Insert import re right after import logging (line 4):

import asyncio
import logging
import re
from datetime import datetime, timedelta

Then append these three helper functions immediately after the existing _get_db_setting_sync function (after line 70, before async def _download_source_async). Exact insertion point is the blank line between line 70 (end of _get_db_setting_sync) and line 73 (start of _download_source_async):

# --- Patreon campaign-ID auto-resolution helpers -------------------------
# These support auto-healing gallery-dl's "Failed to extract campaign ID"
# failures on Patreon sources. See docs/superpowers/specs/2026-04-18-patreon-campaign-resolver-design.md.

_PATREON_VANITY_RE = re.compile(
    r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
    re.IGNORECASE,
)
_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"


def _effective_patreon_url(platform: str, source_url: str, metadata: Optional[dict]) -> str:
    """Return the URL to pass to gallery-dl for this source.

    For Patreon sources with a cached campaign_id in metadata, returns
    patreon.com/id:<id> to bypass the broken vanity-URL lookup path.
    All other cases return source_url unchanged.
    """
    if platform == "patreon" and metadata:
        campaign_id = metadata.get("patreon_campaign_id")
        if campaign_id:
            return f"https://www.patreon.com/id:{campaign_id}"
    return source_url


def _looks_like_campaign_id_failure(dl_result) -> bool:
    """True if the gallery-dl output contains the campaign-ID extraction failure marker."""
    stdout = getattr(dl_result, "stdout", None) or ""
    stderr = getattr(dl_result, "stderr", None) or ""
    return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()


def _extract_vanity_from_url(url: str) -> Optional[str]:
    """Return the vanity (creator slug) from a Patreon URL, or None if not extractable.

    Handles patreon.com/<vanity>, patreon.com/c/<vanity>[/...], with or without www.
    Returns None for id:<N> URLs, bare domain, and non-Patreon URLs.
    """
    m = _PATREON_VANITY_RE.match(url)
    return m.group(1) if m else None

• Step 5: Run tests to verify they pass

Run (from backend/): pytest tests/tasks/test_downloads_helpers.py -v

Expected: ALL PASS (18 tests).

• Step 6: Commit

git add backend/tests/tasks/__init__.py backend/tests/tasks/test_downloads_helpers.py backend/app/tasks/downloads.py
git commit -m "feat(downloads): add patreon URL-rewrite and error-pattern helpers"

---

Task 4: Wire URL rewrite and retry hook into _download_source_async (TDD)

Why: This is the one integration point. Phase 2 of _download_source_async currently calls gdl_service.download() once with source_url. After this task, Phase 2:

1. Calls download() with the effective URL (cached ID when present).
2. On failure matching the campaign-ID pattern for an uncached Patreon source, runs the resolver and retries once with the id-URL.
3. Propagates any resolved campaign ID to Phase 3 so it gets persisted with the rest of the DB writes.

Files:

• Create: backend/tests/tasks/test_downloads_patreon_retry.py

• Modify: backend/app/tasks/downloads.py (import resolver; rewrite Phase 2 block lines 211-228; add metadata write in Phase 3)

• Step 1: Write failing integration test in backend/tests/tasks/test_downloads_patreon_retry.py

These tests exercise the retry logic by mocking gdl_service.download and resolve_campaign_id at the module level. They do NOT run the full _download_source_async function (which requires DB + Celery setup); instead they test a new extracted helper _execute_download_with_retry that owns the Phase-2 logic in pure form.

"""Integration tests for the Patreon campaign-ID retry hook in downloads.py.

These tests exercise _execute_download_with_retry — the pure async helper
that owns the download → resolver → retry logic — without standing up a
full Celery + DB environment.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from app.tasks import downloads as downloads_module
from app.tasks.downloads import _execute_download_with_retry


def _make_gdl_service(results):
    """Build a fake GalleryDLService whose .download() returns results in order."""
    service = SimpleNamespace()
    service.download = AsyncMock(side_effect=list(results))
    return service


def _success(url="https://www.patreon.com/id:12345"):
    return SimpleNamespace(
        success=True, stdout="", stderr="",
        files_downloaded=3, url=url, error_type=None, error_message=None,
    )


def _campaign_id_failure(url="https://www.patreon.com/mstsu"):
    return SimpleNamespace(
        success=False,
        stdout="",
        stderr="[patreon][error] Failed to extract campaign ID",
        files_downloaded=0, url=url, error_type=None,
        error_message="Failed to extract campaign ID",
    )


def _other_failure(url="https://www.patreon.com/mstsu"):
    return SimpleNamespace(
        success=False, stdout="", stderr="HTTP 401 Unauthorized",
        files_downloaded=0, url=url, error_type=None,
        error_message="Authentication failed",
    )


async def test_happy_path_heal(monkeypatch):
    """Patreon source fails with campaign-ID error, resolver returns ID, retry succeeds."""
    gdl = _make_gdl_service([_campaign_id_failure(), _success()])
    monkeypatch.setattr(
        downloads_module,
        "resolve_campaign_id",
        AsyncMock(return_value="12345"),
    )

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/mstsu",
        source_metadata={},
        subscription_name="MSTSU",
        source_config=SimpleNamespace(),
        cookies_path="/data/cookies/patreon_cookies.txt",
        auth_token=None,
    )

    assert result.dl_result.success is True
    assert result.resolved_campaign_id == "12345"
    assert gdl.download.call_count == 2
    first_call_url = gdl.download.call_args_list[0].kwargs["url"]
    second_call_url = gdl.download.call_args_list[1].kwargs["url"]
    assert first_call_url == "https://www.patreon.com/mstsu"
    assert second_call_url == "https://www.patreon.com/id:12345"


async def test_resolver_returns_none(monkeypatch):
    """Campaign-ID failure + resolver returns None → no retry, no metadata, original failure surfaced."""
    first_failure = _campaign_id_failure()
    gdl = _make_gdl_service([first_failure])
    monkeypatch.setattr(
        downloads_module,
        "resolve_campaign_id",
        AsyncMock(return_value=None),
    )

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/mstsu",
        source_metadata={},
        subscription_name="MSTSU",
        source_config=SimpleNamespace(),
        cookies_path="/data/cookies/patreon_cookies.txt",
        auth_token=None,
    )

    assert result.dl_result is first_failure
    assert result.resolved_campaign_id is None
    assert gdl.download.call_count == 1


async def test_already_cached_campaign_id(monkeypatch):
    """Source has cached campaign_id → first download uses id-URL, no resolver call."""
    gdl = _make_gdl_service([_success()])
    resolver_mock = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr(downloads_module, "resolve_campaign_id", resolver_mock)

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/mstsu",
        source_metadata={"patreon_campaign_id": "99999"},
        subscription_name="MSTSU",
        source_config=SimpleNamespace(),
        cookies_path="/data/cookies/patreon_cookies.txt",
        auth_token=None,
    )

    assert result.dl_result.success is True
    assert result.resolved_campaign_id is None
    assert gdl.download.call_count == 1
    assert gdl.download.call_args.kwargs["url"] == "https://www.patreon.com/id:99999"
    resolver_mock.assert_not_called()


async def test_non_patreon_failure_skips_retry(monkeypatch):
    """Non-Patreon source with similar-looking output → retry hook doesn't fire."""
    failure = SimpleNamespace(
        success=False,
        stdout="",
        stderr="some error that contains failed to extract campaign id text",
        files_downloaded=0, url="https://subscribestar.adult/foo",
        error_type=None, error_message="unknown",
    )
    gdl = _make_gdl_service([failure])
    resolver_mock = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr(downloads_module, "resolve_campaign_id", resolver_mock)

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="subscribestar",
        source_url="https://subscribestar.adult/foo",
        source_metadata={},
        subscription_name="Foo",
        source_config=SimpleNamespace(),
        cookies_path=None,
        auth_token=None,
    )

    assert result.dl_result is failure
    assert result.resolved_campaign_id is None
    resolver_mock.assert_not_called()
    assert gdl.download.call_count == 1


async def test_patreon_different_failure_skips_retry(monkeypatch):
    """Patreon source with non-campaign-ID error → no resolver, no retry."""
    failure = _other_failure()
    gdl = _make_gdl_service([failure])
    resolver_mock = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr(downloads_module, "resolve_campaign_id", resolver_mock)

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/mstsu",
        source_metadata={},
        subscription_name="MSTSU",
        source_config=SimpleNamespace(),
        cookies_path="/data/cookies/patreon_cookies.txt",
        auth_token=None,
    )

    assert result.dl_result is failure
    assert result.resolved_campaign_id is None
    resolver_mock.assert_not_called()
    assert gdl.download.call_count == 1


async def test_unextractable_vanity_skips_retry(monkeypatch):
    """Patreon source whose URL can't yield a vanity → no resolver call."""
    failure = _campaign_id_failure(url="https://www.patreon.com/")
    gdl = _make_gdl_service([failure])
    resolver_mock = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr(downloads_module, "resolve_campaign_id", resolver_mock)

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/",
        source_metadata={},
        subscription_name="?",
        source_config=SimpleNamespace(),
        cookies_path=None,
        auth_token=None,
    )

    assert result.dl_result is failure
    resolver_mock.assert_not_called()
    assert gdl.download.call_count == 1


async def test_retry_itself_fails(monkeypatch):
    """Resolver succeeds but retry fails → cache id, surface retry's failure."""
    first_failure = _campaign_id_failure()
    retry_failure = _other_failure()
    gdl = _make_gdl_service([first_failure, retry_failure])
    monkeypatch.setattr(
        downloads_module, "resolve_campaign_id", AsyncMock(return_value="12345")
    )

    result = await _execute_download_with_retry(
        gdl_service=gdl,
        platform="patreon",
        source_url="https://www.patreon.com/mstsu",
        source_metadata={},
        subscription_name="MSTSU",
        source_config=SimpleNamespace(),
        cookies_path="/data/cookies/patreon_cookies.txt",
        auth_token=None,
    )

    assert result.dl_result is retry_failure  # retry's result wins
    assert result.resolved_campaign_id == "12345"  # ID still cached
    assert gdl.download.call_count == 2

• Step 2: Run tests to verify they fail

Run (from backend/): pytest tests/tasks/test_downloads_patreon_retry.py -v

Expected: ALL FAIL with ImportError: cannot import name '_execute_download_with_retry' from 'app.tasks.downloads'.

• Step 3: Add _execute_download_with_retry to downloads.py

Add this import near the other from app.services imports (after line 20 from app.services.credential_manager import CredentialManager):

from app.services.patreon_resolver import resolve_campaign_id

Add a small result container and the retry helper immediately after the three helpers added in Task 3 (so the order inside the file is: Patreon helpers → new dataclass → _execute_download_with_retry → _download_source_async):

from dataclasses import dataclass


@dataclass
class _DownloadAttempt:
    """Result of executing a download, possibly with a retry after campaign-ID resolution."""
    dl_result: DownloadResult
    resolved_campaign_id: Optional[str]


async def _execute_download_with_retry(
    *,
    gdl_service,
    platform: str,
    source_url: str,
    source_metadata: dict,
    subscription_name: str,
    source_config,
    cookies_path: Optional[str],
    auth_token: Optional[str],
) -> _DownloadAttempt:
    """Run the download once, auto-heal Patreon campaign-ID failures with one retry.

    Returns a _DownloadAttempt wrapping:
      - the final DownloadResult (retry's result if we retried, else the first)
      - resolved_campaign_id: the newly-resolved ID (if we resolved one this call), else None

    The caller is responsible for persisting resolved_campaign_id into source.metadata_.
    """
    effective_url = _effective_patreon_url(platform, source_url, source_metadata)
    dl_result = await gdl_service.download(
        url=effective_url,
        subscription_name=subscription_name,
        platform=platform,
        source_config=source_config,
        cookies_path=cookies_path,
        auth_token=auth_token,
    )

    # Only retry if: patreon + campaign-ID failure + no cached id + extractable vanity
    if (
        platform == "patreon"
        and not dl_result.success
        and "patreon_campaign_id" not in (source_metadata or {})
        and _looks_like_campaign_id_failure(dl_result)
    ):
        vanity = _extract_vanity_from_url(source_url)
        if vanity:
            logger.info(
                f"Attempting campaign-ID resolution for {subscription_name}/patreon ({vanity})"
            )
            campaign_id = await resolve_campaign_id(vanity, cookies_path)
            if campaign_id:
                retry_result = await gdl_service.download(
                    url=f"https://www.patreon.com/id:{campaign_id}",
                    subscription_name=subscription_name,
                    platform=platform,
                    source_config=source_config,
                    cookies_path=cookies_path,
                    auth_token=auth_token,
                )
                return _DownloadAttempt(
                    dl_result=retry_result, resolved_campaign_id=campaign_id
                )

    return _DownloadAttempt(dl_result=dl_result, resolved_campaign_id=None)

Also move from dataclasses import dataclass to the top-level imports block (alphabetically near other stdlib imports) — keep the file clean. The insertion at the top should look like:

import asyncio
import logging
import re
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional

And remove the inline from dataclasses import dataclass above the @dataclass decorator.

• Step 4: Replace Phase 2 call site in _download_source_async

In backend/app/tasks/downloads.py, the current Phase 2 block is approximately:

    # --- Phase 2: Execute download (no DB connection held) ---
    source_config = SourceConfig.from_dict(source_metadata)
    dl_result = None
    dl_error = None

    try:
        gdl_service = GalleryDLService(rate_limit=db_rate_limit)
        dl_result = await gdl_service.download(
            url=source_url,
            subscription_name=subscription_name,
            platform=source_platform,
            source_config=source_config,
            cookies_path=cookies_path,
            auth_token=auth_token,
        )
    except Exception as e:
        logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}")
        dl_error = e

Replace it with:

    # --- Phase 2: Execute download (no DB connection held) ---
    source_config = SourceConfig.from_dict(source_metadata)
    dl_result = None
    dl_error = None
    resolved_campaign_id: Optional[str] = None

    try:
        gdl_service = GalleryDLService(rate_limit=db_rate_limit)
        attempt = await _execute_download_with_retry(
            gdl_service=gdl_service,
            platform=source_platform,
            source_url=source_url,
            source_metadata=source_metadata,
            subscription_name=subscription_name,
            source_config=source_config,
            cookies_path=cookies_path,
            auth_token=auth_token,
        )
        dl_result = attempt.dl_result
        resolved_campaign_id = attempt.resolved_campaign_id
    except Exception as e:
        logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}")
        dl_error = e

• Step 5: Persist resolved_campaign_id into Source.metadata_ in Phase 3

In _download_source_async Phase 3 (approximately lines 230-333), locate the block where source has just been re-fetched via select(Source).where(Source.id == source_id) (around line 239). Add the metadata write immediately after the if not download or not source: guard, and before the if dl_error: branch:

            if not download or not source:
                logger.error(f"Could not re-fetch download {actual_download_id} or source {source_id} for result update")
                return {"source_id": source_id, "status": "error", "error": "Lost DB records"}

            # Persist resolved campaign ID (if the retry hook resolved one this run)
            if resolved_campaign_id:
                source.metadata_ = {
                    **(source.metadata_ or {}),
                    "patreon_campaign_id": resolved_campaign_id,
                }

            if dl_error:

The replacement-assignment (not in-place mutation) is important — SQLAlchemy's JSONB type doesn't track in-place mutations to dicts by default, so source.metadata_["patreon_campaign_id"] = … would not be persisted on commit.

• Step 6: Run retry tests to verify they pass

Run (from backend/): pytest tests/tasks/test_downloads_patreon_retry.py -v

Expected: ALL PASS (7 tests).

• Step 7: Run the entire test suite to catch regressions

Run (from backend/): pytest -v

Expected: ALL PASS. If any sanity, helper, or resolver test fails, a regression was introduced — inspect the failure and fix before committing.

• Step 8: Commit

git add backend/tests/tasks/test_downloads_patreon_retry.py backend/app/tasks/downloads.py
git commit -m "feat(downloads): auto-heal patreon campaign-ID failures via resolver retry"

---

Self-Review

Spec coverage (spec section → task):

• "Components / patreon_resolver.py" → Task 2
• "Components / downloads.py retry hook + URL rewrite" → Tasks 3 (helpers) + 4 (integration)
• "Components / gallery_dl.py no change" → covered by omission; caller supplies effective URL
• "Components / Data model no change" → covered by writing to existing JSONB column in Task 4 Step 5
• "Data flow / first-time failure" → Task 4 test_happy_path_heal
• "Data flow / subsequent runs" → Task 4 test_already_cached_campaign_id
• "Data flow / resolver failure" → Task 4 test_resolver_returns_none
• "Data flow / resolver succeeds but retry fails" → Task 4 test_retry_itself_fails
• "Error handling table" → Task 2 tests (resolver-side) + Task 4 tests (hook-side) cover each row
• "Testing" → Tasks 1-4 implement the full testing plan

Placeholder scan: None. Every step has exact code, exact commands, and exact expected output.

Type consistency:

• resolve_campaign_id(vanity: str, cookies_path: Optional[str]) -> Optional[str] — used identically in Task 2 (def) and Task 4 (call site + mock return values).
• _effective_patreon_url(platform, source_url, metadata) — signature matches between Task 3 helper tests and Task 4 _execute_download_with_retry call.
• _looks_like_campaign_id_failure(dl_result) — uses duck-typed .stdout / .stderr attributes in both Task 3 tests and the real helper.
• _extract_vanity_from_url(url: str) -> Optional[str] — consistent.
• _DownloadAttempt(dl_result, resolved_campaign_id) — defined in Task 4 Step 3, consumed in Task 4 Steps 4-5, asserted in Task 4 Step 1.
• The Optional import is used by the new retry helper (Optional[str] = None for resolved_campaign_id) and is already present on line 6 of downloads.py — no import change required.