feat(external): download worker for file-host links (Phase 4b)

tasks/external.py drives the external_link ledger:
- fetch_external_link(link_id): atomic claim (pending/failed→downloading, so a
  duplicate enqueue no-ops), per-host Redis serialize lock (#720 pattern;
  requeue-with-countdown if busy), fetch via external_fetch into the artist
  library tree, then route each file through importer.attach_in_place via a
  synthesized sidecar so it links to the SAME post (archive→ImageRecords,
  else→PostAttachment; on-disk original removed for captured files, art stays);
  thumbnail+ML enqueue for new images; status downloaded | failed | dead with
  attempts/last_error/completed_at/duration.
- sweep_external_links(): enqueue a bounded batch of actionable links.
- recover_external_links() + prune_external_links(): recovery + retention (#89).
- per-host enable read via getattr (forward-compatible; Settings UI adds the
  columns in 4d — defaults on, rule #26).

Wiring: celery include + route (download lane) + beat (sweep 10m, recover +
prune daily); download_service phase 3 enqueues a sweep after recording links.
Integration tests: download+attach, failure, dead-letter, non-claimable, sweep.

mega still needs the MEGAcmd binary in the runtime image (Phase 4c). Refs #830.

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

backend/app/celery_app.py

@@ -30,6 +30,7 @@ def make_celery() -> Celery:
             "backend.app.tasks.maintenance",
             "backend.app.tasks.ml",
             "backend.app.tasks.download",
+            "backend.app.tasks.external",
             "backend.app.tasks.backup",
             "backend.app.tasks.admin",
             "backend.app.tasks.library_audit",
@@ -42,6 +43,9 @@ def make_celery() -> Celery:
             "backend.app.tasks.ml.*": {"queue": "ml"},
             "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"},
             "backend.app.tasks.download.*": {"queue": "download"},
+            # External file-host fetches are downloads — same lane (they can run
+            # long, but the download worker already tolerates long backfills).
+            "backend.app.tasks.external.*": {"queue": "download"},
             "backend.app.tasks.scan.*": {"queue": "scan"},
             # `maintenance` is the QUICK lane — recovery sweeps, vacuum, cleanup
             # (concurrency-1 on the scheduler). The long one-shots (DB backups,
@@ -150,6 +154,21 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.thumbnail.backfill_thumbnails",
                 "schedule": 86400.0,
             },
+            # External file-host downloads (#830): a steady sweep catches links
+            # the post-download hook missed (worker down, etc.); recovery re-tries
+            # dead links daily; retention prunes long-dead rows.
+            "extdl-sweep": {
+                "task": "backend.app.tasks.external.sweep_external_links",
+                "schedule": 600.0,  # every 10 min
+            },
+            "extdl-recover-daily": {
+                "task": "backend.app.tasks.external.recover_external_links",
+                "schedule": 86400.0,
+            },
+            "extdl-prune-daily": {
+                "task": "backend.app.tasks.external.prune_external_links",
+                "schedule": 86400.0,
+            },
         },
         timezone="UTC",
     )

backend/app/services/download_service.py

@@ -396,6 +396,13 @@ class DownloadService:
 
             await loop.run_in_executor(None, _upsert)
 
+        # Kick the off-platform file-host downloader for any links this run
+        # recorded (mega/gdrive/…). Global + idempotent (only claims pending/
+        # retryable rows); the beat sweep is the backstop. Lazy import dodges a
+        # task-module import cycle.
+        from ..tasks.external import sweep_external_links
+        sweep_external_links.delay()
+
         ev = (await self.async_session.execute(
             select(DownloadEvent).where(DownloadEvent.id == event_id)
         )).scalar_one()

backend/app/tasks/external.py

@@ -0,0 +1,292 @@
+"""Celery worker for off-platform file-host downloads (external_link ledger).
+
+Walks the external_link table and fetches each pending/retryable link via
+external_fetch, then routes the downloaded file(s) through the existing importer
+(attach_in_place) so an archive becomes ImageRecords and any other file a
+PostAttachment — linked to the SAME post the link came from (via a synthesized
+sidecar). Long-running-task hygiene (rule #89): per-fetch wall-clock timeout,
+attempt tracking + dead-letter, a recovery sweep, and retention of dead rows.
+
+Concurrency: a per-host Redis lock serializes fetches to one-per-host (mega /
+gdrive ban-avoidance), and an atomic status claim (pending/failed →
+downloading) stops two workers grabbing the same link.
+
+Files land in the artist's library tree (so an attached art image stays in
+place, mirroring the gallery-dl download path); a captured archive/attachment is
+copied into its store and the on-disk original removed.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import redis
+from sqlalchemy import delete, select, update
+
+from ..celery_app import celery
+from ..config import get_config
+from ..models import Artist, ExternalLink, ImportSettings, Post, Source
+from ..services.external_fetch import fetch_external
+from ..services.importer import Importer
+from ..services.thumbnailer import Thumbnailer
+from ._sync_engine import sync_session_factory as _sync_session_factory
+
+log = logging.getLogger(__name__)
+
+IMAGES_ROOT = Path("/images")
+
+# After this many failed attempts a link is dead-lettered (skipped by routine
+# sweeps; an operator recovery still re-attempts). Mirrors the ingester ledger.
+DEAD_LETTER_THRESHOLD = 3
+# Per-fetch wall-clock budget — films/packs are large; the celery time_limit is
+# the backstop above this.
+_FETCH_TIMEOUT = 3000.0
+# Links enqueued per sweep — bounds the burst when a big backfill records many.
+_SWEEP_BATCH = 50
+# Dead rows older than this are pruned (retention).
+_RETENTION_DAYS = 30
+# Per-host serialize lock.
+_LOCK_PREFIX = "fc:extdl_lock:"
+_LOCK_TTL = 3600
+_SERIALIZE_COUNTDOWN = 120
+_MAX_SERIALIZE_WAITS = 30
+
+_redis_client: redis.Redis | None = None
+
+
+def _redis() -> redis.Redis:
+    global _redis_client
+    if _redis_client is None:
+        _redis_client = redis.from_url(get_config().celery_broker_url)
+    return _redis_client
+
+
+def _host_enabled(settings: ImportSettings, host: str) -> bool:
+    """Per-host enable flag — defaults True (rule #26: works out of the box).
+    The Settings UI slice adds real columns; getattr keeps this forward-
+    compatible without a migration in this slice."""
+    return bool(getattr(settings, f"extdl_{host}_enabled", True))
+
+
+def _write_link_sidecar(file: Path, post: Post, platform: str, artist: Artist) -> None:
+    """Sidecar next to a fetched file so the importer links it to `post`
+    (category+id resolve the source+post; matches gallery-dl's sidecar shape)."""
+    data = {
+        "category": platform,
+        "id": post.external_post_id,
+        "url": post.post_url,
+        "title": post.post_title,
+        "artist": artist.name,
+    }
+    Path(str(file) + ".json").write_text(json.dumps(data))
+
+
+@celery.task(
+    name="backend.app.tasks.external.fetch_external_link",
+    bind=True,
+    soft_time_limit=3300,
+    time_limit=3600,
+)
+def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict:
+    """Fetch one external_link and route its file(s) through the importer."""
+    SessionLocal = _sync_session_factory()
+
+    # Atomic claim: only an actionable (pending/failed) row transitions to
+    # downloading, so a duplicate enqueue (sweep + post-download hook) no-ops.
+    with SessionLocal() as session:
+        claimed = session.execute(
+            update(ExternalLink)
+            .where(
+                ExternalLink.id == link_id,
+                ExternalLink.status.in_(("pending", "failed")),
+            )
+            .values(status="downloading")
+            .returning(ExternalLink.id)
+        ).first()
+        session.commit()
+        if claimed is None:
+            return {"link_id": link_id, "skipped": "not claimable"}
+
+        link = session.get(ExternalLink, link_id)
+        post = session.get(Post, link.post_id) if link.post_id else None
+        if post is None:
+            link.status = "dead"
+            link.last_error = "post missing"
+            link.completed_at = datetime.now(UTC)
+            session.commit()
+            return {"link_id": link_id, "error": "post missing"}
+        artist = session.get(Artist, post.artist_id)
+        source = session.get(Source, post.source_id) if post.source_id else None
+        platform = source.platform if source is not None else "patreon"
+        settings = ImportSettings.load_sync(session)
+
+        if not _host_enabled(settings, link.host):
+            link.status = "skipped"
+            link.last_error = f"{link.host} disabled"
+            link.completed_at = datetime.now(UTC)
+            session.commit()
+            return {"link_id": link_id, "skipped": "host disabled"}
+
+        host, url, attempts = link.host, link.url, link.attempts
+
+    # Per-host serialize: one fetch per host at a time. If busy, requeue.
+    lock = None
+    try:
+        lock = _redis().lock(f"{_LOCK_PREFIX}{host}", timeout=_LOCK_TTL, blocking=False)
+        got = bool(lock.acquire(blocking=False))
+    except redis.RedisError as exc:
+        log.warning("extdl lock unavailable for %s: %s — running uncapped", host, exc)
+        lock, got = None, True
+    if not got:
+        # Release the claim back to pending so a later run can pick it up, then
+        # requeue with a countdown (bounded, like download_source).
+        with SessionLocal() as session:
+            session.execute(
+                update(ExternalLink).where(ExternalLink.id == link_id)
+                .values(status="pending")
+            )
+            session.commit()
+        if _serialize_waits < _MAX_SERIALIZE_WAITS:
+            fetch_external_link.apply_async(
+                (link_id,), {"_serialize_waits": _serialize_waits + 1},
+                countdown=_SERIALIZE_COUNTDOWN,
+            )
+        return {"link_id": link_id, "requeued": "host busy"}
+
+    started = time.monotonic()
+    post_dir = IMAGES_ROOT / artist.slug / platform / "external" / str(post.external_post_id)
+    try:
+        result = fetch_external(host, url, post_dir, timeout=_FETCH_TIMEOUT)
+        with SessionLocal() as session:
+            link = session.get(ExternalLink, link_id)
+            if not result.ok:
+                _record_failure(link, attempts, result.error or "fetch failed")
+                session.commit()
+                return {"link_id": link_id, "error": result.error}
+
+            # Re-load post/artist/source attached to THIS session before handing
+            # them to the importer (the claim block's instances are detached).
+            post = session.get(Post, link.post_id)
+            artist = session.get(Artist, post.artist_id)
+            source = session.get(Source, post.source_id) if post.source_id else None
+            image_ids = _route_files(session, result.files, post, platform, artist, source)
+            link.status = "downloaded"
+            link.last_error = None
+            link.completed_at = datetime.now(UTC)
+            link.duration_seconds = time.monotonic() - started
+            session.commit()
+
+        # Thumbnails + ML for any newly-attached images (mirrors the download
+        # path). Lazy import to dodge a task-module import cycle.
+        if image_ids:
+            from .ml import tag_and_embed
+            from .thumbnail import generate_thumbnail
+            for img_id in image_ids:
+                generate_thumbnail.delay(img_id)
+                tag_and_embed.delay(img_id)
+        return {"link_id": link_id, "files": len(result.files), "images": len(image_ids)}
+    except Exception as exc:  # never leave a link stuck in 'downloading'
+        log.exception("external fetch task failed for link %s", link_id)
+        with SessionLocal() as session:
+            link = session.get(ExternalLink, link_id)
+            if link is not None and link.status == "downloading":
+                _record_failure(link, attempts, f"{type(exc).__name__}: {exc}")
+                session.commit()
+        raise
+    finally:
+        if lock is not None:
+            try:
+                lock.release()
+            except Exception:  # noqa: BLE001 — TTL may have already freed it
+                pass
+
+
+def _record_failure(link: ExternalLink, attempts: int, error: str) -> None:
+    nxt = attempts + 1
+    link.attempts = nxt
+    link.last_error = error[:1000]
+    link.status = "dead" if nxt >= DEAD_LETTER_THRESHOLD else "failed"
+    link.completed_at = datetime.now(UTC)
+
+
+def _route_files(session, files, post, platform, artist, source) -> list[int]:
+    """Import each fetched file via the existing pipeline (in the artist tree, so
+    art stays in place); return new image ids for thumb/ML enqueue. A captured
+    archive/attachment is copied into its store, so its on-disk original is
+    removed (mirrors download_service); art images stay."""
+    settings = ImportSettings.load_sync(session)
+    importer = Importer(
+        session=session,
+        images_root=IMAGES_ROOT,
+        import_root=IMAGES_ROOT,
+        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
+        settings=settings,
+    )
+    image_ids: list[int] = []
+    for f in files:
+        _write_link_sidecar(f, post, platform, artist)
+        result = importer.attach_in_place(f, artist=artist, source=source)
+        session.commit()
+        if result.status in ("imported", "superseded"):
+            ids = list(getattr(result, "member_image_ids", []) or [])
+            if result.image_id is not None and result.image_id not in ids:
+                ids.append(result.image_id)
+            image_ids.extend(ids)
+        elif result.status in ("attached", "failed"):
+            # Copied into the attachment store (or failed) — drop the on-disk
+            # original so we don't keep two copies / an orphan.
+            f.unlink(missing_ok=True)
+        # The synthesized sidecar has done its job — don't litter the tree.
+        Path(str(f) + ".json").unlink(missing_ok=True)
+    return image_ids
+
+
+@celery.task(name="backend.app.tasks.external.sweep_external_links")
+def sweep_external_links() -> dict:
+    """Enqueue a bounded batch of actionable links (pending, or failed below the
+    dead-letter threshold). Driven by the post-download hook and a beat tick."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        ids = session.execute(
+            select(ExternalLink.id).where(
+                ExternalLink.status.in_(("pending", "failed")),
+                ExternalLink.attempts < DEAD_LETTER_THRESHOLD,
+            ).order_by(ExternalLink.id.asc()).limit(_SWEEP_BATCH)
+        ).scalars().all()
+    for link_id in ids:
+        fetch_external_link.delay(link_id)
+    return {"enqueued": len(ids)}
+
+
+@celery.task(name="backend.app.tasks.external.recover_external_links")
+def recover_external_links() -> dict:
+    """Recovery sweep (rule #89): reset dead links back to retryable so a stuck
+    host outage or a since-fixed bug gets another pass, then enqueue. Bounded."""
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        session.execute(
+            update(ExternalLink).where(ExternalLink.status == "dead")
+            .values(status="failed", attempts=0, last_error=None)
+        )
+        session.commit()
+    return sweep_external_links()
+
+
+@celery.task(name="backend.app.tasks.external.prune_external_links")
+def prune_external_links() -> dict:
+    """Retention (rule #89): delete long-dead links so the ledger doesn't grow
+    unbounded. Downloaded links are kept (they're the record of what we have)."""
+    cutoff = datetime.now(UTC) - timedelta(days=_RETENTION_DAYS)
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        res = session.execute(
+            delete(ExternalLink).where(
+                ExternalLink.status == "dead",
+                ExternalLink.created_at < cutoff,
+            )
+        )
+        session.commit()
+    return {"pruned": res.rowcount or 0}

tests/test_external_worker.py

@@ -0,0 +1,137 @@
+"""Integration tests for the external-link download worker (tasks/external)."""
+
+import pytest
+from sqlalchemy import func, select
+
+import backend.app.tasks.external as ext
+from backend.app.models import Artist, ExternalLink, Post, PostAttachment, Source
+from backend.app.services.external_fetch import FetchResult
+
+pytestmark = pytest.mark.integration
+
+
+class _FakeLock:
+    def acquire(self, blocking=False):
+        return True
+
+    def release(self):
+        pass
+
+
+class _FakeRedis:
+    def lock(self, name, timeout=None, blocking=False):
+        return _FakeLock()
+
+
+def _seed(db_sync, *, host="pixeldrain", status="pending"):
+    artist = Artist(name="Ext Artist", slug="ext-artist")
+    db_sync.add(artist)
+    db_sync.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/ext", enabled=True, config_overrides={},
+    )
+    db_sync.add(source)
+    db_sync.flush()
+    post = Post(
+        source_id=source.id, artist_id=artist.id, external_post_id="EXT1",
+        post_url="https://patreon.com/posts/EXT1",
+    )
+    db_sync.add(post)
+    db_sync.flush()
+    link = ExternalLink(
+        post_id=post.id, artist_id=artist.id, host=host,
+        url=f"https://{host}.test/file", status=status,
+    )
+    db_sync.add(link)
+    db_sync.commit()
+    return post, link
+
+
+def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypatch):
+    post, link = _seed(db_sync)
+    link_id = link.id
+
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+
+    def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False):
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        f = dest_dir / "film.bin"  # non-art → PostAttachment (no thumb/ML enqueue)
+        f.write_bytes(b"a film pack")
+        return FetchResult(files=[f], bytes=f.stat().st_size)
+
+    monkeypatch.setattr(ext, "fetch_external", fake_fetch)
+
+    out = ext.fetch_external_link(link_id)
+    assert out.get("files") == 1
+
+    db_sync.expire_all()
+    refreshed = db_sync.get(ExternalLink, link_id)
+    assert refreshed.status == "downloaded"
+    assert refreshed.completed_at is not None
+    # The file was captured as a PostAttachment linked to the SAME post.
+    atts = db_sync.execute(
+        select(PostAttachment).where(PostAttachment.post_id == post.id)
+    ).scalars().all()
+    assert len(atts) == 1
+
+
+def test_fetch_external_link_records_failure(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync)
+    link_id = link.id
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(
+        ext, "fetch_external",
+        lambda *a, **k: FetchResult(error="host 503"),
+    )
+
+    ext.fetch_external_link(link_id)
+    db_sync.expire_all()
+    refreshed = db_sync.get(ExternalLink, link_id)
+    assert refreshed.status == "failed"
+    assert refreshed.attempts == 1
+    assert "503" in refreshed.last_error
+
+
+def test_fetch_external_link_dead_letters_at_threshold(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync)
+    link.attempts = ext.DEAD_LETTER_THRESHOLD - 1
+    db_sync.commit()
+    link_id = link.id
+    monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(ext, "fetch_external", lambda *a, **k: FetchResult(error="nope"))
+
+    ext.fetch_external_link(link_id)
+    db_sync.expire_all()
+    assert db_sync.get(ExternalLink, link_id).status == "dead"
+
+
+def test_fetch_external_link_skips_non_claimable(db_sync, tmp_path, monkeypatch):
+    _, link = _seed(db_sync, status="downloaded")
+    link_id = link.id
+    called = []
+    monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
+    monkeypatch.setattr(
+        ext, "fetch_external",
+        lambda *a, **k: called.append(1) or FetchResult(),
+    )
+    out = ext.fetch_external_link(link_id)
+    assert out.get("skipped") == "not claimable"
+    assert called == []  # never fetched
+
+
+def test_sweep_enqueues_pending_and_retryable(db_sync, monkeypatch):
+    _, l1 = _seed(db_sync, host="pixeldrain")
+    _, l2 = _seed(db_sync, host="mega", status="failed")
+    # A dead one must NOT be swept.
+    _, l3 = _seed(db_sync, host="dropbox", status="dead")
+
+    enqueued = []
+    monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid))
+
+    out = ext.sweep_external_links()
+    assert out["enqueued"] == 2
+    assert set(enqueued) == {l1.id, l2.id}