feat(catalog)!: add staged provider catalog indexing

apps/api/.env.example

@@ -78,6 +78,65 @@ MEGAKINO_BASE_URL=https://megakino1.to
 # Default: 12
 MEGAKINO_TITLES_REFRESH_HOURS=12
 
+## Scheduled Provider Catalog Index
+# What: Default refresh cadence (hours) for the persistent provider catalog index
+# Default: 24
+PROVIDER_INDEX_REFRESH_HOURS=24
+# What: Provider-specific refresh cadence override for AniWorld
+# Default: PROVIDER_INDEX_REFRESH_HOURS
+PROVIDER_INDEX_REFRESH_HOURS_ANIWORLD=24
+# What: Provider-specific refresh cadence override for s.to
+# Default: PROVIDER_INDEX_REFRESH_HOURS
+PROVIDER_INDEX_REFRESH_HOURS_STO=24
+# What: Provider-specific refresh cadence override for megakino
+# Default: PROVIDER_INDEX_REFRESH_HOURS
+PROVIDER_INDEX_REFRESH_HOURS_MEGAKINO=24
+# What: Scheduler poll interval in seconds for checking due provider refreshes
+# Default: 60
+PROVIDER_INDEX_SCHEDULER_POLL_SECONDS=60
+# What: Maximum number of provider refreshes allowed to run in parallel
+# Default: 1
+PROVIDER_INDEX_GLOBAL_CONCURRENCY=1
+# What: Per-provider crawl worker count for AniWorld title refreshes
+# Default: 4
+PROVIDER_INDEX_CONCURRENCY_ANIWORLD=4
+# What: Per-provider crawl worker count for s.to title refreshes
+# Default: 4
+PROVIDER_INDEX_CONCURRENCY_STO=4
+# What: Per-provider crawl worker count for megakino title refreshes
+# Default: 2
+PROVIDER_INDEX_CONCURRENCY_MEGAKINO=2
+# What: Hard timeout in seconds for one provider title crawl
+# Default: 45
+PROVIDER_INDEX_TITLE_TIMEOUT_SECONDS=45
+# What: Maximum number of completed title payloads allowed to wait in memory for DB persistence
+# Default: 8
+PROVIDER_INDEX_QUEUE_SIZE=8
+# What: Max number of title payloads committed per SQLite writer batch
+# Default: 32
+PROVIDER_INDEX_WRITER_BATCH_SIZE=32
+# What: Max seconds the writer may hold a partial batch before forcing a commit
+# Default: 1.0
+PROVIDER_INDEX_WRITER_FLUSH_SECONDS=1.0
+# What: Abort a refresh when failed title crawls exceed this percentage of the discovered title set
+# Default: 20
+PROVIDER_INDEX_FAILURE_THRESHOLD_PERCENT=20
+# What: Minimum seconds between repeated queue-backpressure log lines while crawlers are blocked on persistence
+# Default: 15
+PROVIDER_INDEX_BACKPRESSURE_LOG_SECONDS=15
+# What: Maximum parallel canonical metadata lookups per provider stage run
+# Default: 2
+CANONICAL_INDEX_CONCURRENCY=2
+# What: Max hot in-memory canonical search cache entries
+# Default: 512
+CANONICAL_CACHE_MEMORY_MAX_SEARCH=512
+# What: Max hot in-memory canonical show cache entries
+# Default: 256
+CANONICAL_CACHE_MEMORY_MAX_SHOW=256
+# What: TTL in seconds for hot in-memory canonical cache entries
+# Default: 3600
+CANONICAL_CACHE_TTL_SECONDS=3600
+
 # What: Domain check interval in minutes (0 disables background checks)
 # Default: 100
 MEGAKINO_DOMAIN_CHECK_INTERVAL_MIN=100
@@ -108,11 +167,21 @@ PROVIDER_ORDER=VOE,Filemoon,Streamtape,Vidmoly,Doodstream,LoadX,Luluvdo,Vidoza
 # Default: 12
 PROVIDER_REDIRECT_TIMEOUT_SECONDS=12
 
+# What: Hard timeout in seconds for one video host direct-link resolution
+# attempt before AniBridge abandons that host and tries the next fallback.
+# Default: 15
+PROVIDER_DIRECT_LINK_TIMEOUT_SECONDS=15
+
 # What: Extra retry attempts when provider redirect resolution times out or
 # hits transient network errors
 # Default: 2
 PROVIDER_REDIRECT_RETRIES=2
 
+# What: How often the background scheduler flushes coalesced download
+# progress updates to SQLite while yt-dlp is running.
+# Default: 0.5
+JOB_PROGRESS_FLUSH_SECONDS=0.5
+
 # What: Base cool-down in seconds before retrying a Serienstream redirect that
 # returned a Turnstile/captcha page. Retries back off linearly from this value.
 # Default: 300

apps/api/app/api/health.py

@@ -2,9 +2,19 @@
 
 from fastapi import APIRouter
 
+from app.catalog import get_catalog_indexer
+
 router = APIRouter()
 
 
 @router.get("/health")
-async def healthcheck():
-    return {"status": "ok"}
+def healthcheck():
+    return {
+        "status": "ok",
+        "catalog": get_catalog_indexer().get_progress_snapshot(),
+    }
+
+
+@router.get("/health/catalog")
+def catalog_healthcheck():
+    return get_catalog_indexer().get_progress_snapshot()

apps/api/app/api/qbittorrent/common.py

@@ -15,6 +15,27 @@ def public_save_path() -> str:
     return QBIT_PUBLIC_SAVE_PATH or str(DOWNLOAD_DIR)
 
 
+def coerce_torrent_state(*, stored_state: str | None, job_status: str | None) -> str:
+    """Map persisted task state and scheduler status to qBittorrent states."""
+    if job_status == "completed":
+        return "uploading"
+    if job_status == "failed":
+        return "error"
+    if job_status == "cancelled":
+        return "pausedDL"
+
+    normalized = (stored_state or "").strip().lower()
+    if normalized == "queued":
+        return "queuedDL"
+    if normalized == "paused":
+        return "pausedDL"
+    if normalized == "completed":
+        return "uploading"
+    if normalized == "error" or normalized == "failed":
+        return "error"
+    return "downloading"
+
+
 # Categories map compatible with qBittorrent format
 CATEGORIES: Dict[str, dict] = {
     "prowlarr": {

apps/api/app/api/qbittorrent/sync.py

@@ -8,7 +8,7 @@
 from app.db import get_session, get_job
 
 from . import router
-from .common import CATEGORIES, public_save_path
+from .common import CATEGORIES, coerce_torrent_state, public_save_path
 from app.config import DOWNLOAD_DIR, QBIT_PUBLIC_SAVE_PATH
 
 
@@ -27,14 +27,10 @@ def sync_maindata(session: Session = Depends(get_session)):
     for r in rows:
         job = get_job(session, r.job_id) if r.job_id else None
         progress = (job.progress or 0.0) / 100.0 if job else 0.0
-        state = "downloading"
-        if job:
-            if job.status == "completed":
-                state = "uploading"
-            elif job.status == "failed":
-                state = "error"
-            elif job.status == "cancelled":
-                state = "pausedDL"
+        state = coerce_torrent_state(
+            stored_state=r.state,
+            job_status=job.status if job else None,
+        )
 
         size_val = int(job.total_bytes or 0) if job else 0
         save_path_val = (

apps/api/app/api/qbittorrent/torrents.py

@@ -20,10 +20,10 @@
     delete_client_task,
     get_job,
 )
-from app.core.scheduler import schedule_download, cancel_job
+from app.core.scheduler import cancel_job, schedule_download, start_scheduled_job
 
 from . import router
-from .common import public_save_path
+from .common import coerce_torrent_state, public_save_path
 
 
 @router.post("/torrents/add")
@@ -101,8 +101,8 @@ def torrents_add(
         req["provider"] = provider
     if mode:
         req["mode"] = mode
-    job_id = schedule_download(req)
-    logger.debug(f"Scheduled job_id: {job_id}")
+    job_id = schedule_download(req, autostart=False)
+    logger.debug(f"Created scheduled job_id: {job_id}")
 
     if not savepath:
         savepath = str(DOWNLOAD_DIR)
@@ -120,13 +120,105 @@ def torrents_add(
         save_path=published_savepath,
         category=category,
         job_id=job_id,
-        state="queued" if paused else "downloading",
+        state="queued",
+        provider=provider,
+        mode=mode or None,
     )
     logger.success(
         "Torrent task upserted for hash={}, state={}, site={}".format(
-            btih, "queued" if paused else "downloading", site
+            btih, "queued", site
         )
     )
+    if not paused:
+        try:
+            start_scheduled_job(job_id, req)
+        except Exception as exc:
+            logger.error("Failed to start scheduled job {}: {}", job_id, exc)
+            upsert_client_task(
+                session,
+                hash=btih,
+                name=name,
+                slug=slug,
+                season=season,
+                episode=episode,
+                language=language,
+                site=site,
+                save_path=published_savepath,
+                category=category,
+                job_id=job_id,
+                state="failed",
+                provider=provider,
+                mode=mode or None,
+            )
+            return PlainTextResponse("Failed to start download.", status_code=500)
+        upsert_client_task(
+            session,
+            hash=btih,
+            name=name,
+            slug=slug,
+            season=season,
+            episode=episode,
+            language=language,
+            site=site,
+            save_path=published_savepath,
+            category=category,
+            job_id=job_id,
+            state="downloading",
+            provider=provider,
+            mode=mode or None,
+        )
+        logger.debug(f"Started background worker for job_id: {job_id}")
+    return PlainTextResponse("Ok.")
+
+
+@router.post("/torrents/resume")
+def torrents_resume(
+    session: Session = Depends(get_session),
+    hashes: str = Form(...),
+):
+    """Start existing queued torrent jobs."""
+    requested_hashes = [value.strip().lower() for value in hashes.split("|")]
+    if requested_hashes == ["all"]:
+        from app.db import ClientTask
+        from sqlmodel import select
+
+        tasks = session.exec(select(ClientTask)).all()
+    else:
+        tasks = [
+            task
+            for torrent_hash in requested_hashes
+            if (task := get_client_task(session, torrent_hash)) is not None
+        ]
+
+    for task in tasks:
+        job = get_job(session, task.job_id) if task.job_id else None
+        if job is None or job.status != "queued":
+            continue
+        req = {
+            "slug": task.slug,
+            "season": task.season,
+            "episode": task.episode,
+            "language": task.language,
+            "site": task.site or "aniworld.to",
+            "title_hint": task.name,
+        }
+        if task.provider:
+            req["provider"] = task.provider
+        if task.mode:
+            req["mode"] = task.mode
+        try:
+            start_scheduled_job(job.id, req)
+        except Exception as exc:
+            logger.error("Failed to resume scheduled job {}: {}", job.id, exc)
+            task.state = "failed"
+            session.add(task)
+            session.commit()
+            return PlainTextResponse("Failed to resume download.", status_code=500)
+        task.state = "downloading"
+        session.add(task)
+        session.commit()
+        logger.debug("Resumed background worker for job_id: {}", job.id)
+
     return PlainTextResponse("Ok.")
 
 
@@ -149,7 +241,9 @@ def torrents_info(
         if category and (r.category or "") != category:
             continue
         job = get_job(session, r.job_id) if r.job_id else None
-        state = r.state
+        state = coerce_torrent_state(
+            stored_state=r.state, job_status=job.status if job else None
+        )
         progress = 0.0
         dlspeed = 0
         eta = 0
@@ -162,19 +256,12 @@ def torrents_info(
                 f"Job {job.id}: status={job.status}, progress={progress}, speed={dlspeed}, eta={eta}"
             )
             if job.status == "completed":
-                state = "uploading"
                 dlspeed = 0
                 if job.result_path and os.path.exists(job.result_path):
                     try:
                         size = int(os.path.getsize(job.result_path))
                     except Exception:
                         pass
-            elif job.status == "failed":
-                state = "error"
-            elif job.status == "cancelled":
-                state = "pausedDL"
-            else:
-                state = "downloading"
 
         content_path = None
         save_path_val = r.save_path or (QBIT_PUBLIC_SAVE_PATH or str(DOWNLOAD_DIR))