diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py
new file mode 100644
index 0000000..2a8778a
--- /dev/null
+++ b/backend/app/services/download_service.py
@@ -0,0 +1,273 @@
+"""FC-3c download orchestrator.
+
+Three-phase pipeline:
+  Phase 1 — DB setup (brief session): load source, mark event running,
+            fetch credentials + settings.
+  Phase 2 — Execute (no DB connection): call GalleryDLService; on
+            Patreon campaign-ID failure, resolve vanity and retry once.
+  Phase 3 — Import + persist (fresh session): attach_in_place each
+            written file; clean up duplicates; persist rich metadata.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session as SyncSession, joinedload
+
+from ..models import Artist, DownloadEvent, Source
+from .credential_service import CredentialService
+from .gallery_dl import GalleryDLService, SourceConfig
+from .importer import Importer
+from .patreon_resolver import resolve_campaign_id
+
+log = logging.getLogger(__name__)
+
+
+_PATREON_VANITY_RE = re.compile(
+    r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
+    re.IGNORECASE,
+)
+_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"
+
+
+def _extract_patreon_vanity(url: str) -> str | None:
+    m = _PATREON_VANITY_RE.match(url)
+    return m.group(1) if m else None
+
+
+def _looks_like_campaign_id_failure(stdout: str, stderr: str) -> bool:
+    return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()
+
+
+def _effective_url(platform: str, source_url: str, overrides: dict) -> str:
+    if platform == "patreon" and overrides.get("patreon_campaign_id"):
+        return f"https://www.patreon.com/id:{overrides['patreon_campaign_id']}"
+    return source_url
+
+
+class DownloadService:
+    """Async orchestrator. The Celery task runs `asyncio.run(svc.download_source(N))`.
+
+    The Importer is sync (existing FC convention); calls to attach_in_place
+    happen via run_in_executor.
+    """
+
+    def __init__(
+        self,
+        async_session: AsyncSession,
+        sync_session: SyncSession,
+        gdl: GalleryDLService,
+        importer: Importer,
+        cred_service: CredentialService,
+    ):
+        self.async_session = async_session
+        self.sync_session = sync_session
+        self.gdl = gdl
+        self.importer = importer
+        self.cred_service = cred_service
+
+    async def download_source(self, source_id: int) -> int:
+        """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is."""
+        setup = await self._phase1_setup(source_id)
+        if setup["status"] in ("skipped", "in_flight"):
+            return setup["event_id"]
+        ctx = setup
+
+        source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
+        effective_url = _effective_url(
+            ctx["platform"], ctx["url"], ctx["config_overrides"] or {}
+        )
+
+        dl_result = await self.gdl.download(
+            url=effective_url,
+            artist_slug=ctx["artist_slug"],
+            platform=ctx["platform"],
+            source_config=source_config,
+            cookies_path=ctx["cookies_path"],
+            auth_token=ctx["auth_token"],
+        )
+
+        resolved_campaign_id: str | None = None
+        if (
+            ctx["platform"] == "patreon"
+            and not dl_result.success
+            and "patreon_campaign_id" not in (ctx["config_overrides"] or {})
+            and _looks_like_campaign_id_failure(dl_result.stdout, dl_result.stderr)
+        ):
+            vanity = _extract_patreon_vanity(ctx["url"])
+            if vanity:
+                log.info(
+                    "Attempting campaign-ID resolution for %s (%s)",
+                    ctx["artist_slug"], vanity,
+                )
+                resolved_campaign_id = await resolve_campaign_id(
+                    vanity, ctx["cookies_path"]
+                )
+                if resolved_campaign_id:
+                    dl_result = await self.gdl.download(
+                        url=f"https://www.patreon.com/id:{resolved_campaign_id}",
+                        artist_slug=ctx["artist_slug"],
+                        platform=ctx["platform"],
+                        source_config=source_config,
+                        cookies_path=ctx["cookies_path"],
+                        auth_token=ctx["auth_token"],
+                    )
+
+        return await self._phase3_persist(
+            ctx["event_id"], ctx, 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)
+        )).scalar_one_or_none()
+        if source is None:
+            raise LookupError(f"source {source_id} not found")
+
+        if not source.enabled:
+            ev = DownloadEvent(
+                source_id=source_id, status="skipped",
+                error="source disabled",
+                finished_at=datetime.now(UTC),
+            )
+            self.async_session.add(ev)
+            await self.async_session.commit()
+            await self.async_session.refresh(ev)
+            return {"status": "skipped", "event_id": ev.id}
+
+        existing = (await self.async_session.execute(
+            select(DownloadEvent).where(
+                DownloadEvent.source_id == source_id,
+                DownloadEvent.status.in_(["pending", "running"]),
+            ).order_by(DownloadEvent.id.desc()).limit(1)
+        )).scalar_one_or_none()
+
+        if existing and existing.status == "running":
+            return {"status": "in_flight", "event_id": existing.id}
+        if existing and existing.status == "pending":
+            existing.status = "running"
+            await self.async_session.commit()
+            event_id = existing.id
+        else:
+            ev = DownloadEvent(source_id=source_id, status="running")
+            self.async_session.add(ev)
+            await self.async_session.commit()
+            await self.async_session.refresh(ev)
+            event_id = ev.id
+
+        artist = source.artist
+        if source.platform in ("discord", "pixiv"):
+            cookies_path = None
+            auth_token = await self.cred_service.get_token(source.platform)
+        else:
+            cookies_path_obj = await self.cred_service.get_cookies_path(source.platform)
+            cookies_path = str(cookies_path_obj) if cookies_path_obj else None
+            auth_token = None
+
+        return {
+            "status": "ok",
+            "event_id": event_id,
+            "source_id": source_id,
+            "url": source.url,
+            "platform": source.platform,
+            "artist_slug": artist.slug if artist else "_unknown",
+            "artist_id": artist.id if artist else None,
+            "config_overrides": dict(source.config_overrides or {}),
+            "cookies_path": cookies_path,
+            "auth_token": auth_token,
+        }
+
+    async def _phase3_persist(
+        self,
+        event_id: int,
+        ctx: dict,
+        dl_result,
+        resolved_campaign_id: str | None,
+    ) -> int:
+        # Cache Patreon campaign id on the source.
+        if resolved_campaign_id:
+            src = self.sync_session.get(Source, ctx["source_id"])
+            if src is not None:
+                src.config_overrides = {
+                    **(src.config_overrides or {}),
+                    "patreon_campaign_id": resolved_campaign_id,
+                }
+                self.sync_session.commit()
+
+        artist = None
+        source_row = None
+        if ctx.get("artist_id"):
+            artist = self.sync_session.get(Artist, ctx["artist_id"])
+        source_row = self.sync_session.get(Source, ctx["source_id"])
+
+        import_summary = {"attached": 0, "skipped": 0, "errors": 0}
+        bytes_downloaded = 0
+
+        loop = asyncio.get_running_loop()
+
+        for path_str in dl_result.written_paths or []:
+            if path_str in (dl_result.quarantined_paths or []):
+                continue
+            path = Path(path_str)
+            if not path.exists():
+                continue
+
+            def _attach(p=path):
+                return self.importer.attach_in_place(
+                    p, artist=artist, source=source_row,
+                )
+
+            result = await loop.run_in_executor(None, _attach)
+
+            if result.status in ("imported", "superseded"):
+                import_summary["attached"] += 1
+                try:
+                    bytes_downloaded += path.stat().st_size
+                except OSError:
+                    pass
+            elif result.status == "skipped" and result.skip_reason and result.skip_reason.value in (
+                "duplicate_hash", "duplicate_phash",
+            ):
+                import_summary["skipped"] += 1
+                try:
+                    path.unlink(missing_ok=True)
+                except OSError:
+                    pass
+            else:
+                import_summary["errors"] += 1
+
+        ev = (await self.async_session.execute(
+            select(DownloadEvent).where(DownloadEvent.id == event_id)
+        )).scalar_one()
+
+        run_stats = self.gdl._compute_run_stats(
+            dl_result.return_code, dl_result.stdout, dl_result.stderr
+        )
+        run_stats["quarantined_count"] = dl_result.files_quarantined
+        stderr_summary = self.gdl._extract_errors_warnings(dl_result.stderr)
+
+        ev.status = "ok" if (dl_result.success and import_summary["errors"] == 0) else "error"
+        ev.finished_at = datetime.now(UTC)
+        ev.files_count = import_summary["attached"]
+        ev.bytes_downloaded = bytes_downloaded
+        ev.error = dl_result.error_message if not dl_result.success else None
+        ev.metadata_ = {
+            "run_stats": run_stats,
+            "error_type": dl_result.error_type.value if dl_result.error_type else None,
+            "stdout": self.gdl._truncate_log(dl_result.stdout) or None,
+            "stderr": self.gdl._truncate_log(dl_result.stderr) or None,
+            "stderr_errors_warnings": stderr_summary or None,
+            "duration_seconds": dl_result.duration_seconds,
+            "quarantined_paths": dl_result.quarantined_paths or None,
+            "import_summary": import_summary,
+        }
+        await self.async_session.commit()
+        return event_id
diff --git a/tests/test_download_service.py b/tests/test_download_service.py
new file mode 100644
index 0000000..496a3a6
--- /dev/null
+++ b/tests/test_download_service.py
@@ -0,0 +1,254 @@
+"""DownloadService orchestrator tests.
+
+GalleryDLService is mocked to return a canned DownloadResult; tests
+provide on-disk files for the importer to attach_in_place.
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist, DownloadEvent, ImportSettings, Source
+from backend.app.services.credential_crypto import CredentialCrypto
+from backend.app.services.credential_service import CredentialService
+from backend.app.services.thumbnailer import Thumbnailer
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def seed_artist_and_source(db, db_sync):
+    artist = Artist(name="Alice", slug="alice")
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/alice", enabled=True,
+        config_overrides={},
+    )
+    db.add(source)
+    await db.commit()
+    return artist, source
+
+
+def _make_jpg(path: Path):
+    from PIL import Image
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", (32, 32), (100, 100, 100)).save(path, "JPEG")
+
+
+def _make_fake_dl_result(
+    *, success=True, written_paths=None, quarantined_paths=None,
+    files_downloaded=0, error_type=None, error_message=None,
+    stdout="", stderr="",
+):
+    return SimpleNamespace(
+        success=success,
+        url="https://patreon.com/alice",
+        artist_slug="alice",
+        platform="patreon",
+        files_downloaded=files_downloaded,
+        files_quarantined=len(quarantined_paths or []),
+        quarantined_paths=quarantined_paths or [],
+        written_paths=written_paths or [],
+        stdout=stdout,
+        stderr=stderr,
+        return_code=0 if success else 1,
+        error_type=error_type,
+        error_message=error_message,
+        duration_seconds=1.23,
+        started_at="2026-05-20T14:00:00+00:00",
+        completed_at="2026-05-20T14:01:00+00:00",
+    )
+
+
+def _fake_gdl_with_result(result):
+    fake = MagicMock()
+    fake.download = AsyncMock(return_value=result)
+    fake._compute_run_stats = lambda *a, **k: {
+        "exit_code": 0, "downloaded_count": len(result.written_paths),
+        "skipped_count": 0, "per_item_failures": 0,
+        "warning_count": 0, "tier_gated_count": 0,
+    }
+    fake._extract_errors_warnings = lambda *a, **k: ""
+    fake._truncate_log = lambda x, **k: x
+    return fake
+
+
+@pytest.mark.asyncio
+async def test_download_source_attaches_written_files(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    images_root = tmp_path / "images"
+    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
+    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
+    _make_jpg(f1)
+    _make_jpg(f2)
+
+    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+        success=True,
+        written_paths=[str(f1), str(f2)],
+        files_downloaded=2,
+        stdout=f"{f1}\n{f2}\n",
+    ))
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+
+    importer = Importer(
+        session=db_sync, images_root=images_root,
+        import_root=images_root,
+        thumbnailer=Thumbnailer(images_root=images_root),
+        settings=sync_settings,
+    )
+    crypto = CredentialCrypto(tmp_path / "key.b64")
+    cred_service = CredentialService(db, crypto)
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+
+    ev = (await db.execute(
+        select(DownloadEvent).where(DownloadEvent.id == event_id)
+    )).scalar_one()
+    assert ev.status == "ok"
+    assert ev.files_count == 2
+    assert ev.metadata_["import_summary"]["attached"] == 2
+    assert ev.metadata_["run_stats"]["downloaded_count"] == 2
+
+
+@pytest.mark.asyncio
+async def test_download_source_skipped_when_disabled(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+    source.enabled = False
+    await db.commit()
+
+    fake_gdl = MagicMock()
+    fake_gdl.download = AsyncMock()
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path,
+        import_root=tmp_path,
+        thumbnailer=Thumbnailer(images_root=tmp_path),
+        settings=sync_settings,
+    )
+    cred_service = CredentialService(db, MagicMock())
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+
+    ev = (await db.execute(
+        select(DownloadEvent).where(DownloadEvent.id == event_id)
+    )).scalar_one()
+    assert ev.status == "skipped"
+    assert ev.error == "source disabled"
+    fake_gdl.download.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_in_flight_idempotency_returns_existing_event(
+    db, db_sync, tmp_path, seed_artist_and_source,
+):
+    from backend.app.services.download_service import DownloadService
+
+    _artist, source = seed_artist_and_source
+    running = DownloadEvent(source_id=source.id, status="running")
+    db.add(running)
+    await db.commit()
+    await db.refresh(running)
+
+    fake_gdl = MagicMock()
+    fake_gdl.download = AsyncMock()
+    importer_stub = MagicMock()
+    cred_service = CredentialService(db, MagicMock())
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer_stub, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+    assert event_id == running.id
+    fake_gdl.download.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_patreon_retry_caches_campaign_id(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    failed = _make_fake_dl_result(
+        success=False, written_paths=[], stdout="",
+        stderr="[patreon][error] Failed to extract campaign ID for alice\n",
+    )
+    succeeded = _make_fake_dl_result(success=True, written_paths=[])
+
+    download_calls = []
+
+    async def fake_download(url, **k):
+        download_calls.append(url)
+        return failed if len(download_calls) == 1 else succeeded
+
+    fake_gdl = MagicMock()
+    fake_gdl.download = fake_download
+    fake_gdl._compute_run_stats = lambda *a, **k: {
+        "exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
+        "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
+    }
+    fake_gdl._extract_errors_warnings = lambda *a, **k: ""
+    fake_gdl._truncate_log = lambda x, **k: x
+
+    monkeypatch.setattr(
+        "backend.app.services.download_service.resolve_campaign_id",
+        AsyncMock(return_value="99"),
+    )
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path,
+        import_root=tmp_path,
+        thumbnailer=Thumbnailer(images_root=tmp_path),
+        settings=sync_settings,
+    )
+    cred_service = CredentialService(db, MagicMock())
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    await svc.download_source(source.id)
+
+    assert len(download_calls) == 2
+    assert "id:99" in download_calls[1]
+
+    overrides = db_sync.execute(
+        select(Source.config_overrides).where(Source.id == source.id)
+    ).scalar_one()
+    assert overrides.get("patreon_campaign_id") == "99"