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:

```ini
[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`**

```python
"""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`**

```python
"""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):

```python
"""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**

```bash
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`**

```python
"""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`**

```python
"""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**

```bash
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`**

```python
"""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):

```python
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`):

```python
# --- 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**

```bash
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.

```python
"""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`):

```python
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`):

```python
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:

```python
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:

```python
    # --- 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:

```python
    # --- 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:

```python
            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**

```bash
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.