Implement Phase 6: MeTube download integration

- MeTubeClient encapsulates all MeTube HTTP/Socket.IO calls (verified against
  the real MeTube source: /add returns no job id, GET /history gives a queue
  snapshot for reconciliation, filenames arrive already relative, percent is
  a 0-100 float)
- download_jobs table + service: request/dedup active downloads, apply live
  Socket.IO events (added/updated/completed/canceled/cleared) matched by
  canonical YouTube URL, safe relative-path -> public media URL construction
- Reconciliation on startup against MeTube's live queue/done state (section 19):
  non-terminal jobs recovered where possible, else marked "unknown"; already
  completed jobs are left untouched
- POST/GET /api/videos/{id}/download(-status), recheck-local; feed/video
  detail now report real local availability instead of a stub
- Frontend: download button with live status polling (queued/downloading %/
  postprocessing/completed/failed+retry), local <video> playback with
  YouTube fallback on playback error
- health.py now delegates to MeTubeClient (single place for MeTube calls)

26 new backend tests (63 total). Verified live: Socket.IO connects
successfully to the real MeTube instance on deploy.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/api/feed.py

@@ -10,6 +10,7 @@ from app.db import get_db
 from app.models.channel import Channel
 from app.models.channel_category import channel_categories
 from app.models.video import Video
+from app.services.download_jobs import latest_jobs_map
 from app.services.video_presentation import channel_categories_map, serialize_video
 
 router = APIRouter(dependencies=[Depends(require_session)])
@@ -76,12 +77,15 @@ def get_feed(
     channel_ids = list({v.channel_id for v in rows})
     channels = {c.id: c for c in db.query(Channel).filter(Channel.id.in_(channel_ids)).all()}
     categories_map = channel_categories_map(db, channel_ids)
+    jobs_map = latest_jobs_map(db, [v.id for v in rows])
 
     items = []
     for video in rows:
         channel = channels.get(video.channel_id)
         if channel is None:
             continue
-        items.append(serialize_video(video, channel, categories_map.get(channel.id, [])))
+        items.append(
+            serialize_video(video, channel, categories_map.get(channel.id, []), jobs_map.get(video.id))
+        )
 
     return {"items": items, "next_cursor": next_cursor}

backend/app/api/health.py

@@ -1,12 +1,11 @@
 import logging
 
-import httpx
 from fastapi import APIRouter, Depends
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
-from app.config import settings
 from app.db import get_db
+from app.services.metube_client import MeTubeClient
 
 logger = logging.getLogger(__name__)
 
@@ -22,21 +21,10 @@ def _check_database(db: Session) -> str:
         return "error"
 
 
-def _check_metube() -> str:
-    try:
-        response = httpx.get(settings.metube_api_base_url, timeout=5)
-        if response.status_code < 500:
-            return "ok"
-        return "error"
-    except Exception:
-        logger.warning("MeTube healthcheck failed", exc_info=True)
-        return "error"
-
-
 @router.get("/health")
 def health(db: Session = Depends(get_db)) -> dict:
     return {
         "status": "ok",
         "database": _check_database(db),
-        "metube": _check_metube(),
+        "metube": "ok" if MeTubeClient().health() else "error",
     }

backend/app/api/videos.py

@@ -1,24 +1,89 @@
+import logging
+
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 
 from app.core.auth_dependency import require_session
 from app.db import get_db
 from app.models.channel import Channel
+from app.models.download_job import DownloadJob
 from app.models.video import Video
+from app.services.download_jobs import MeTubeRejected, get_latest_job, request_download
+from app.services.metube_client import MeTubeClient
 from app.services.video_presentation import channel_categories_map, serialize_video
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(dependencies=[Depends(require_session)])
 
 
-@router.get("/videos/{youtube_video_id}")
-def get_video(youtube_video_id: str, db: Session = Depends(get_db)) -> dict:
+def _get_video_or_404(db: Session, youtube_video_id: str) -> Video:
     video = db.query(Video).filter(Video.youtube_video_id == youtube_video_id).one_or_none()
     if video is None:
         raise HTTPException(status_code=404, detail="Video not found")
+    return video
+
+
+def _serialize_job(job: DownloadJob) -> dict:
+    return {
+        "status": job.status,
+        "progress_percent": job.progress_percent,
+        "media_url": job.media_url if job.status == "completed" else None,
+        "error_message": job.error_message,
+        "requested_at": job.requested_at,
+        "started_at": job.started_at,
+        "completed_at": job.completed_at,
+    }
+
+
+@router.get("/videos/{youtube_video_id}")
+def get_video(youtube_video_id: str, db: Session = Depends(get_db)) -> dict:
+    video = _get_video_or_404(db, youtube_video_id)
 
     channel = db.get(Channel, video.channel_id)
     if channel is None:
         raise HTTPException(status_code=404, detail="Channel not found")
 
     categories = channel_categories_map(db, [channel.id]).get(channel.id, [])
-    return serialize_video(video, channel, categories)
+    job = get_latest_job(db, video.id)
+    return serialize_video(video, channel, categories, job)
+
+
+@router.post("/videos/{youtube_video_id}/download")
+def download_video(youtube_video_id: str, db: Session = Depends(get_db)) -> dict:
+    video = _get_video_or_404(db, youtube_video_id)
+
+    try:
+        job = request_download(db, video)
+    except MeTubeRejected as exc:
+        raise HTTPException(status_code=502, detail=f"MeTube rejected the download: {exc}")
+    except Exception:
+        logger.exception("Failed to enqueue download for %s", youtube_video_id)
+        raise HTTPException(status_code=502, detail="MeTube is unavailable")
+
+    return _serialize_job(job)
+
+
+@router.get("/videos/{youtube_video_id}/download-status")
+def download_status(youtube_video_id: str, db: Session = Depends(get_db)) -> dict:
+    video = _get_video_or_404(db, youtube_video_id)
+    job = get_latest_job(db, video.id)
+    if job is None:
+        return {"status": "not_downloaded", "progress_percent": None, "media_url": None, "error_message": None}
+    return _serialize_job(job)
+
+
+@router.post("/videos/{youtube_video_id}/recheck-local")
+def recheck_local(youtube_video_id: str, db: Session = Depends(get_db)) -> dict:
+    video = _get_video_or_404(db, youtube_video_id)
+    job = get_latest_job(db, video.id)
+    if job is None or job.status != "completed":
+        raise HTTPException(status_code=400, detail="No completed download to recheck")
+
+    if job.media_url and MeTubeClient().check_media(job.media_url):
+        return _serialize_job(job)
+
+    job.status = "unknown"
+    job.media_url = None
+    db.commit()
+    return _serialize_job(job)

backend/app/main.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -29,18 +30,48 @@ from app.api.health import router as health_router
 from app.api.sync import router as sync_router
 from app.api.videos import router as videos_router
 from app.config import settings
+from app.db import SessionLocal
+from app.services import download_jobs
+from app.services.metube_client import MeTubeClient
 from app.services.scheduler import create_scheduler
 
 logging.basicConfig(level=settings.log_level)
 logger = logging.getLogger(__name__)
 
 
+async def _on_metube_event(event_name: str, payload) -> None:
+    db = SessionLocal()
+    try:
+        await download_jobs.handle_metube_event(db, event_name, payload)
+    except Exception:
+        logger.exception("Failed handling MeTube '%s' event", event_name)
+    finally:
+        db.close()
+
+
+def _reconcile_download_jobs() -> None:
+    db = SessionLocal()
+    try:
+        download_jobs.reconcile_on_startup(db)
+    except Exception:
+        logger.exception("Failed to reconcile download jobs on startup")
+    finally:
+        db.close()
+
+
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     logger.info("Application startup")
+    _reconcile_download_jobs()
+
     scheduler = create_scheduler()
     scheduler.start()
+
+    metube_task = asyncio.create_task(MeTubeClient().run_event_listener(_on_metube_event))
+
     yield
+
+    metube_task.cancel()
     scheduler.shutdown(wait=False)
     logger.info("Application shutdown")
 

backend/app/models/__init__.py

@@ -2,7 +2,16 @@ from app.models.app_settings import AppSetting
 from app.models.category import Category
 from app.models.channel import Channel
 from app.models.channel_category import channel_categories
+from app.models.download_job import DownloadJob
 from app.models.oauth_credentials import OAuthCredentials
 from app.models.video import Video
 
-__all__ = ["AppSetting", "Category", "Channel", "channel_categories", "OAuthCredentials", "Video"]
+__all__ = [
+    "AppSetting",
+    "Category",
+    "Channel",
+    "channel_categories",
+    "DownloadJob",
+    "OAuthCredentials",
+    "Video",
+]

backend/app/models/download_job.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+
+# queued -> downloading -> postprocessing -> completed
+#                                          -> failed
+# unknown: state could not be reconciled after a restart
+ACTIVE_STATUSES = ("queued", "downloading", "postprocessing")
+TERMINAL_STATUSES = ("completed", "failed")
+
+
+class DownloadJob(Base):
+    __tablename__ = "download_jobs"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    video_id: Mapped[int] = mapped_column(ForeignKey("videos.id"), nullable=False, index=True)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="queued")
+    metube_job_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    metube_filename: Mapped[str | None] = mapped_column(String, nullable=True)
+    media_url: Mapped[str | None] = mapped_column(String, nullable=True)
+    progress_percent: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    requested_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )

backend/app/services/download_jobs.py

@@ -0,0 +1,179 @@
+import json
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from app.models.download_job import ACTIVE_STATUSES, DownloadJob
+from app.models.video import Video
+from app.services.metube_client import METUBE_STATUS_MAP, MeTubeClient
+
+logger = logging.getLogger(__name__)
+
+
+class MeTubeRejected(Exception):
+    pass
+
+
+def get_latest_job(db: Session, video_id: int) -> DownloadJob | None:
+    return (
+        db.query(DownloadJob)
+        .filter(DownloadJob.video_id == video_id)
+        .order_by(DownloadJob.requested_at.desc(), DownloadJob.id.desc())
+        .first()
+    )
+
+
+def latest_jobs_map(db: Session, video_ids: list[int]) -> dict[int, DownloadJob]:
+    if not video_ids:
+        return {}
+    rows = (
+        db.query(DownloadJob)
+        .filter(DownloadJob.video_id.in_(video_ids))
+        .order_by(DownloadJob.requested_at.desc(), DownloadJob.id.desc())
+        .all()
+    )
+    result: dict[int, DownloadJob] = {}
+    for row in rows:
+        result.setdefault(row.video_id, row)
+    return result
+
+
+def request_download(db: Session, video: Video) -> DownloadJob:
+    existing = get_latest_job(db, video.id)
+    if existing is not None and existing.status in ACTIVE_STATUSES:
+        return existing
+
+    result = MeTubeClient().enqueue_video(video.youtube_url, video.youtube_video_id)
+    if result.get("status") == "error":
+        raise MeTubeRejected(result.get("msg") or "MeTube rejected the download")
+
+    job = DownloadJob(video_id=video.id, status="queued")
+    db.add(job)
+    db.commit()
+    db.refresh(job)
+    return job
+
+
+def _find_job_by_payload(db: Session, payload: dict) -> DownloadJob | None:
+    metube_id = payload.get("id")
+    if metube_id:
+        job = (
+            db.query(DownloadJob)
+            .filter(DownloadJob.metube_job_id == metube_id)
+            .order_by(DownloadJob.id.desc())
+            .first()
+        )
+        if job is not None:
+            return job
+
+    url = payload.get("url")
+    if url:
+        video = db.query(Video).filter(Video.youtube_url == url).one_or_none()
+        if video is not None:
+            job = get_latest_job(db, video.id)
+            if job is not None:
+                return job
+    return None
+
+
+async def handle_metube_event(db: Session, event_name: str, raw_payload) -> None:
+    try:
+        payload = json.loads(raw_payload) if isinstance(raw_payload, str) else raw_payload
+    except (TypeError, ValueError):
+        logger.warning("Could not parse MeTube event payload for %s: %r", event_name, raw_payload)
+        return
+
+    if event_name in ("canceled", "cleared"):
+        if not isinstance(payload, str):
+            return
+        job = (
+            db.query(DownloadJob)
+            .filter(DownloadJob.metube_job_id == payload)
+            .order_by(DownloadJob.id.desc())
+            .first()
+        )
+        if job is not None and event_name == "canceled" and job.status in ACTIVE_STATUSES:
+            job.status = "failed"
+            job.error_message = "Отменено в MeTube"
+            db.commit()
+        return
+
+    if not isinstance(payload, dict):
+        return
+
+    job = _find_job_by_payload(db, payload)
+    if job is None:
+        return
+
+    _apply_metube_info(job, payload)
+    db.commit()
+    logger.info("download_job %s updated to status=%s via '%s' event", job.id, job.status, event_name)
+
+
+def _apply_metube_info(job: DownloadJob, info: dict) -> None:
+    if info.get("id"):
+        job.metube_job_id = info["id"]
+
+    our_status = METUBE_STATUS_MAP.get(info.get("status"), "unknown")
+
+    if job.started_at is None and our_status in ("downloading", "postprocessing", "completed"):
+        job.started_at = datetime.now(timezone.utc)
+
+    percent = info.get("percent")
+    if isinstance(percent, (int, float)):
+        job.progress_percent = int(percent)
+
+    if our_status == "failed":
+        job.status = "failed"
+        job.error_message = info.get("msg") or info.get("error") or "Ошибка загрузки"
+    elif our_status == "completed":
+        filename = info.get("filename")
+        if filename:
+            job.metube_filename = filename
+            job.media_url = MeTubeClient().build_media_url(filename)
+        job.status = "completed"
+        job.progress_percent = 100
+        job.completed_at = job.completed_at or datetime.now(timezone.utc)
+    else:
+        job.status = our_status
+
+
+def reconcile_on_startup(db: Session) -> None:
+    """Section 19: after a restart we don't trust in-flight jobs until we've
+    checked MeTube's current state. Completed jobs with a media_url are left
+    alone — they don't need MeTube's history to remain trustworthy."""
+    stale_statuses = list(ACTIVE_STATUSES) + ["unknown"]
+    jobs = db.query(DownloadJob).filter(DownloadJob.status.in_(stale_statuses)).all()
+    if not jobs:
+        return
+
+    client = MeTubeClient()
+    try:
+        history = client.fetch_history()
+    except Exception:
+        logger.warning("Could not fetch MeTube history for reconciliation", exc_info=True)
+        for job in jobs:
+            job.status = "unknown"
+        db.commit()
+        return
+
+    by_url: dict[str, dict] = {}
+    for bucket in ("queue", "pending", "done"):
+        for item in history.get(bucket, []) or []:
+            url = item.get("url")
+            if url:
+                by_url[url] = item
+
+    videos = {v.id: v for v in db.query(Video).filter(Video.id.in_([j.video_id for j in jobs])).all()}
+
+    for job in jobs:
+        video = videos.get(job.video_id)
+        info = by_url.get(video.youtube_url) if video else None
+        if info is None:
+            job.status = "unknown"
+            continue
+        _apply_metube_info(job, info)
+
+    db.commit()
+    logger.info("Reconciled %d download job(s) against MeTube history", len(jobs))

backend/app/services/metube_client.py

@@ -0,0 +1,136 @@
+"""
+Encapsulates all HTTP/Socket.IO calls to the external MeTube service.
+
+Verified against the real MeTube source (alexta69/metube, app/main.py + app/ytdl.py):
+
+- POST /add only accepts/returns {"status": "ok"|"error", "msg": ...} — no job id.
+  The job id/url/status are only observable via Socket.IO events or GET /history.
+- MeTube itself dedups by URL (a second /add for a queued URL is a no-op "ok").
+- Socket.IO 'added'/'updated'/'completed' events carry DownloadInfo.to_public_dict(),
+  JSON-*string*-encoded (json.JSONEncoder().encode(...)), not a raw object — must
+  json.loads() the payload. Relevant keys: id, title, url, status, msg, percent
+  (float 0-100 or None), filename (already relative to DOWNLOAD_DIR), error.
+  'canceled'/'cleared' carry just an id (also JSON-string-encoded).
+- MeTube status vocabulary: pending/preparing/scheduled/downloading/postprocessing/
+  finished/error — mapped to our own vocabulary in sync with download_jobs.
+- GET /history returns {"queue": [...], "pending": [...], "done": [...]} of the
+  same to_public_dict() shape — used for reconciliation after our own restart.
+- Downloaded files are served by aiohttp's static route at /download/<relative
+  path>, matching METUBE_CONTAINER_DOWNLOAD_DIR.
+"""
+
+import logging
+from pathlib import PurePosixPath
+from typing import Awaitable, Callable
+from urllib.parse import quote
+
+import httpx
+import socketio
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+METUBE_STATUS_MAP: dict[str, str] = {
+    "pending": "queued",
+    "preparing": "queued",
+    "scheduled": "queued",
+    "downloading": "downloading",
+    "postprocessing": "postprocessing",
+    "finished": "completed",
+    "error": "failed",
+}
+
+EventHandler = Callable[[str, dict | str], Awaitable[None]]
+
+
+class MeTubeClient:
+    def __init__(self) -> None:
+        self.api_base_url = settings.metube_api_base_url.rstrip("/")
+        self.public_base_url = settings.metube_public_base_url.rstrip("/")
+        self.download_dir = settings.metube_container_download_dir
+        self.timeout = settings.metube_request_timeout_seconds
+
+    def enqueue_video(self, youtube_url: str, custom_name_prefix: str) -> dict:
+        payload = {
+            "url": youtube_url,
+            "download_type": "video",
+            "codec": "auto",
+            "format": "mp4",
+            "quality": "best",
+            "auto_start": True,
+            "custom_name_prefix": custom_name_prefix,
+        }
+        response = httpx.post(f"{self.api_base_url}/add", json=payload, timeout=self.timeout)
+        response.raise_for_status()
+        return response.json()
+
+    def health(self) -> bool:
+        try:
+            response = httpx.get(self.api_base_url, timeout=5)
+            return response.status_code < 500
+        except Exception:
+            return False
+
+    def fetch_history(self) -> dict:
+        response = httpx.get(f"{self.api_base_url}/history", timeout=self.timeout)
+        response.raise_for_status()
+        return response.json()
+
+    def build_media_url(self, filename: str) -> str | None:
+        """Safely turn a MeTube-reported filename into a public /download/... URL.
+
+        `filename` is expected relative to METUBE_CONTAINER_DOWNLOAD_DIR (that's
+        what MeTube itself stores), but we defensively strip an accidental
+        absolute prefix and reject any path that escapes the download dir.
+        """
+        if not filename:
+            return None
+
+        normalized = filename.replace("\\", "/")
+        download_dir = self.download_dir.rstrip("/")
+        if download_dir and normalized.startswith(download_dir + "/"):
+            normalized = normalized[len(download_dir) + 1 :]
+        normalized = normalized.lstrip("/")
+
+        path = PurePosixPath(normalized)
+        if ".." in path.parts or path.is_absolute():
+            logger.warning("Rejected unsafe MeTube filename: %r", filename)
+            return None
+
+        encoded = "/".join(quote(part) for part in path.parts)
+        return f"{self.public_base_url}/download/{encoded}"
+
+    def check_media(self, media_url: str) -> bool:
+        try:
+            response = httpx.head(media_url, timeout=self.timeout, follow_redirects=True)
+            return response.status_code == 200
+        except Exception:
+            return False
+
+    async def run_event_listener(self, on_event: EventHandler) -> None:
+        """Runs forever, reconnecting automatically, until cancelled."""
+        sio = socketio.AsyncClient(reconnection=True, reconnection_delay=5, reconnection_delay_max=30)
+
+        for event_name in ("added", "updated", "completed", "canceled", "cleared"):
+
+            async def _handler(data, _event_name=event_name):
+                await on_event(_event_name, data)
+
+            sio.on(event_name, _handler)
+
+        @sio.event
+        async def connect():
+            logger.info("Connected to MeTube Socket.IO at %s", self.api_base_url)
+
+        @sio.event
+        async def disconnect():
+            logger.warning("Disconnected from MeTube Socket.IO")
+
+        while True:
+            try:
+                await sio.connect(self.api_base_url, wait_timeout=10)
+                await sio.wait()
+            except Exception:
+                logger.warning("MeTube Socket.IO connection failed, retrying", exc_info=True)
+                await sio.sleep(10)

backend/app/services/video_presentation.py

@@ -4,6 +4,7 @@ from sqlalchemy.orm import Session
 from app.models.category import Category
 from app.models.channel import Channel
 from app.models.channel_category import channel_categories
+from app.models.download_job import DownloadJob
 from app.models.video import Video
 
 
@@ -21,7 +22,22 @@ def channel_categories_map(db: Session, channel_ids: list[int]) -> dict[int, lis
     return result
 
 
-def serialize_video(video: Video, channel: Channel, categories: list[dict]) -> dict:
+def _serialize_local(job: DownloadJob | None) -> dict:
+    if job is None:
+        return {"available": False, "status": "not_downloaded", "progress_percent": None, "media_url": None}
+    if job.status == "completed":
+        return {"available": True, "status": "completed", "progress_percent": 100, "media_url": job.media_url}
+    return {
+        "available": False,
+        "status": job.status,
+        "progress_percent": job.progress_percent,
+        "media_url": None,
+    }
+
+
+def serialize_video(
+    video: Video, channel: Channel, categories: list[dict], download_job: DownloadJob | None = None
+) -> dict:
     return {
         "youtube_video_id": video.youtube_video_id,
         "title": video.title,
@@ -37,10 +53,5 @@ def serialize_video(video: Video, channel: Channel, categories: list[dict]) -> d
         "duration_seconds": video.duration_seconds,
         "youtube_url": video.youtube_url,
         "categories": categories,
-        "local": {
-            "available": False,
-            "status": "not_downloaded",
-            "progress_percent": None,
-            "media_url": None,
-        },
+        "local": _serialize_local(download_job),
     }