From 772fb6f9d0dd55ec554e57e6ad7be6a7877496a3 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Wed, 20 May 2026 20:43:16 -0400
Subject: [PATCH] feat(fc3c): download_source Celery task + register in
 celery_app

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 backend/app/celery_app.py          |  1 +
 backend/app/tasks/download.py      | 76 ++++++++++++++++++++++++++++++
 tests/test_download_source_task.py | 14 ++++++
 3 files changed, 91 insertions(+)
 create mode 100644 backend/app/tasks/download.py
 create mode 100644 tests/test_download_source_task.py

diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py
index a2d3a56..d839ba6 100644
--- a/backend/app/celery_app.py
+++ b/backend/app/celery_app.py
@@ -29,6 +29,7 @@ def make_celery() -> Celery:
             "backend.app.tasks.thumbnail",
             "backend.app.tasks.maintenance",
             "backend.app.tasks.ml",
+            "backend.app.tasks.download",
         ],
     )
     app.conf.update(
diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py
new file mode 100644
index 0000000..a6416ab
--- /dev/null
+++ b/backend/app/tasks/download.py
@@ -0,0 +1,76 @@
+"""download_source Celery task — runs DownloadService for one source."""
+
+import asyncio
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import ImportSettings
+from ..services.credential_crypto import CredentialCrypto
+from ..services.credential_service import CredentialService
+from ..services.download_service import DownloadService
+from ..services.gallery_dl import GalleryDLService
+from ..services.importer import Importer
+from ..services.thumbnailer import Thumbnailer
+from .import_file import _sync_session_factory
+
+IMAGES_ROOT = Path("/images")
+_KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
+
+
+def _async_session_factory():
+    cfg = get_config()
+    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
+    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine
+
+
+@celery.task(name="backend.app.tasks.download.download_source", bind=True, acks_late=True)
+def download_source(self, source_id: int) -> int:
+    """Returns the DownloadEvent.id."""
+
+    async def _run():
+        async_factory, async_engine = _async_session_factory()
+        SyncFactory = _sync_session_factory()
+        try:
+            with SyncFactory() as sync_session:
+                settings = sync_session.execute(
+                    select(ImportSettings).where(ImportSettings.id == 1)
+                ).scalar_one()
+                rate_limit = settings.download_rate_limit_seconds
+                validate_files = settings.download_validate_files
+
+            gdl = GalleryDLService(
+                images_root=IMAGES_ROOT,
+                rate_limit=rate_limit,
+                validate_files=validate_files,
+            )
+            crypto = CredentialCrypto(_KEY_PATH)
+
+            async with async_factory() as async_session:
+                cred_service = CredentialService(async_session, crypto)
+                with SyncFactory() as sync_session:
+                    sync_settings = sync_session.execute(
+                        select(ImportSettings).where(ImportSettings.id == 1)
+                    ).scalar_one()
+                    importer = Importer(
+                        session=sync_session,
+                        images_root=IMAGES_ROOT,
+                        import_root=IMAGES_ROOT,
+                        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
+                        settings=sync_settings,
+                    )
+                    svc = DownloadService(
+                        async_session=async_session,
+                        sync_session=sync_session,
+                        gdl=gdl,
+                        importer=importer,
+                        cred_service=cred_service,
+                    )
+                    return await svc.download_source(source_id)
+        finally:
+            await async_engine.dispose()
+
+    return asyncio.run(_run())
diff --git a/tests/test_download_source_task.py b/tests/test_download_source_task.py
new file mode 100644
index 0000000..e625d7b
--- /dev/null
+++ b/tests/test_download_source_task.py
@@ -0,0 +1,14 @@
+"""Smoke test that the Celery task is registered and routed correctly."""
+
+from backend.app.celery_app import celery
+
+
+def test_download_source_is_registered():
+    name = "backend.app.tasks.download.download_source"
+    assert name in celery.tasks
+
+
+def test_download_source_routes_to_download_queue():
+    routes = celery.conf.task_routes
+    assert "backend.app.tasks.download.*" in routes
+    assert routes["backend.app.tasks.download.*"]["queue"] == "download"