feat(subs): kick off the first backfill walk immediately on source create

A new enabled source is armed for run-until-done backfill (#693) but would sit
idle until the next scheduler tick (~60s). create_source now enqueues the first
walk right away (pending DownloadEvent + download_source.delay), skipping only
when the platform is in a rate-limit cooldown (the scheduler picks it up when
that clears). Disabled sources still don't dispatch.

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

backend/app/api/sources.py

@@ -85,6 +85,22 @@ async def create_source():
             return _bad("empty_url", detail=str(exc))
         except DuplicateSourceError as exc:
             return _bad("duplicate", status=409, existing_id=exc.existing_id)
+
+        # Immediate kickoff: a new enabled source is armed for backfill (#693)
+        # but would otherwise sit idle until the next scheduler tick (~60s).
+        # Enqueue the first walk now, skipping only if the platform is in a
+        # rate-limit cooldown (the scheduler picks it up when that clears).
+        dispatch_id = None
+        if record.enabled:
+            cooldowns = await active_platform_cooldowns(session)
+            if record.platform not in cooldowns:
+                session.add(DownloadEvent(source_id=record.id, status="pending"))
+                await session.commit()
+                dispatch_id = record.id
+
+    if dispatch_id is not None:
+        from ..tasks.download import download_source
+        download_source.delay(dispatch_id)
     return jsonify(record.to_dict()), 201
 
 

tests/test_api_sources.py

@@ -89,6 +89,39 @@ async def test_create_list_get_delete(client, artist):
     assert (await client.get(f"/api/sources/{sid}")).status_code == 404
 
 
+@pytest.mark.asyncio
+async def test_create_kicks_off_backfill_for_enabled_source(client, artist, monkeypatch):
+    """A new enabled source dispatches its first walk immediately (no waiting
+    for the next scheduler tick), unless its platform is in cooldown."""
+    delays: list = []
+    monkeypatch.setattr(
+        "backend.app.tasks.download.download_source.delay",
+        lambda *a, **k: delays.append(a),
+    )
+    resp = await client.post("/api/sources", json={
+        "artist_id": artist.id, "platform": "patreon",
+        "url": "https://patreon.com/kickoff",
+    })
+    assert resp.status_code == 201
+    sid = (await resp.get_json())["id"]
+    assert delays == [(sid,)]
+
+
+@pytest.mark.asyncio
+async def test_create_disabled_source_does_not_kick_off(client, artist, monkeypatch):
+    delays: list = []
+    monkeypatch.setattr(
+        "backend.app.tasks.download.download_source.delay",
+        lambda *a, **k: delays.append(a),
+    )
+    resp = await client.post("/api/sources", json={
+        "artist_id": artist.id, "platform": "patreon",
+        "url": "https://patreon.com/disabled", "enabled": False,
+    })
+    assert resp.status_code == 201
+    assert delays == []
+
+
 @pytest.mark.asyncio
 async def test_create_rejects_unknown_platform(client, artist):
     resp = await client.post("/api/sources", json={