Coconut-first transcoding, commit provenance in drafts

delivery-kid/ansible/playbook.yml

@@ -128,10 +128,13 @@
                 - IPFS_API_URL=http://ipfs:5001
                 - IPFS_GATEWAY_URL=https://{{ ipfs_gateway_domain }}
                 - STAGING_DIR=/staging
+                - COCONUT_API_KEY={{ coconut_api_key | default('') }}
               volumes:
                 - /mnt/storage-box/staging:/staging
               ports:
                 - "127.0.0.1:3001:3001"
+                - "6881:6881"      # BitTorrent TCP
+                - "6881:6881/udp"  # BitTorrent UDP
               networks:
                 - delivery-kid-net
               restart: unless-stopped
@@ -329,6 +332,14 @@
         dest: /etc/caddy/Caddyfile
         content: |
           {{ domain_name }} {
+              route /webseed/* {
+                  # BEP 19 webseed rewrite: BT clients request
+                  # /webseed/{cid}/{torrent_name}/{file_path}
+                  # Strip the torrent name and proxy to IPFS gateway as
+                  # /ipfs/{cid}/{file_path}
+                  uri path_regexp ^/webseed/([^/]+)/[^/]+/(.+)$ /ipfs/$1/$2
+                  reverse_proxy localhost:8080
+              }
               route {
                   reverse_proxy /health localhost:3001
                   reverse_proxy /version localhost:3001
@@ -344,6 +355,7 @@
                   reverse_proxy /draft-content* localhost:3001
                   reverse_proxy /unpin/* localhost:3001
                   reverse_proxy /enrich* localhost:3001
+                  reverse_proxy /torrent* localhost:3001
                   respond "delivery-kid pinning service" 200
               }
           }

delivery-kid/pinning-service/Dockerfile

@@ -26,5 +26,6 @@ RUN useradd -m -u 1000 appuser
 USER appuser
 
 EXPOSE 3001
+EXPOSE 6881
 
 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "3001"]

delivery-kid/pinning-service/app/config.py

@@ -1,5 +1,6 @@
 """Configuration settings loaded from environment variables."""
 
+import os
 from pydantic_settings import BaseSettings
 from functools import lru_cache
 
@@ -19,9 +20,15 @@ class Settings(BaseSettings):
     # Staging directory for uploads and transcoding
     staging_dir: str = "/staging"
 
+    # Seeding directory for BitTorrent (persistent, on storage box)
+    seeding_dir: str = "/staging/seeding"
+
     # Authorized wallets (comma-separated)
     authorized_wallets: str = ""
 
+    # Coconut.co cloud transcoding
+    coconut_api_key: str = ""
+
     # Auth settings
     max_timestamp_drift_seconds: int = 3600  # 1 hour — token generated at page load, user may browse before uploading
     api_key: str = ""  # Shared API key for server-to-server auth (e.g., from PickiPedia)
@@ -56,3 +63,8 @@ def authorized_wallet_list(self) -> list[str]:
 @lru_cache
 def get_settings() -> Settings:
     return Settings()
+
+
+def get_commit() -> str:
+    """Return the git commit hash baked into this build."""
+    return os.environ.get("GIT_COMMIT", "unknown")

delivery-kid/pinning-service/app/main.py

@@ -9,8 +9,9 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from .config import get_settings
-from .routes import health, albums, drafts, content, enrich
+from .routes import health, albums, drafts, content, enrich, torrent, coconut
 from .services import cleanup
+from .services.seeder import init_seeder, stop_seeder
 
 # Configure logging
 logging.basicConfig(
@@ -47,10 +48,16 @@ async def lifespan(app: FastAPI):
         cleanup.periodic_cleanup(staging_dir, interval_seconds=3600)
     )
 
+    # Start BitTorrent seeder
+    init_seeder(settings.seeding_dir)
+
     logger.info("Delivery Kid pinning service started")
     yield
 
-    # Shutdown: cancel cleanup task
+    # Shutdown: stop seeder first
+    stop_seeder()
+
+    # Cancel cleanup task
     cleanup_task.cancel()
     try:
         await cleanup_task
@@ -84,6 +91,8 @@ async def lifespan(app: FastAPI):
 app.include_router(drafts.router)
 app.include_router(content.router)
 app.include_router(enrich.router)
+app.include_router(torrent.router)
+app.include_router(coconut.router)
 
 
 @app.get("/")

delivery-kid/pinning-service/app/models/content.py

@@ -43,6 +43,7 @@ class ContentDraftResponse(BaseModel):
     expires_at: datetime
     files: list[ContentFile]
     metadata: dict = Field(default_factory=dict)
+    commit: str = Field(default="unknown", description="Git commit hash of the build that created this draft")
 
 
 class ContentFinalizeRequest(BaseModel):
@@ -51,5 +52,14 @@ class ContentFinalizeRequest(BaseModel):
     description: Optional[str] = None
     file_type: Optional[str] = Field(default=None, description="MIME type override (e.g., video/webm)")
     metadata: dict = Field(default_factory=dict, description="Arbitrary metadata for Release page")
-    transcode_hls: bool = Field(default=False, description="Transcode video to HLS before pinning")
+    transcode_hls: bool = Field(default=False, description="Transcode video to HLS before pinning (legacy, use transcoding_strategy)")
+    transcoding_strategy: str = Field(
+        default="auto",
+        description="Transcoding strategy for video: 'auto' (Coconut first, local fallback), 'coconut', 'local', 'none'"
+    )
     subsequent_to: Optional[str] = Field(default=None, description="CID this content supersedes")
+    transcoding_qualities: Optional[list[int]] = Field(
+        default=None,
+        description="Output video heights for HLS transcoding, e.g. [1080, 720, 480]. "
+                    "Default [720, 480]. Common values: 2160 (4K), 1080, 720, 480, 360."
+    )

delivery-kid/pinning-service/app/models/draft.py

@@ -31,6 +31,7 @@ class DraftResponse(BaseModel):
     draft_id: str
     expires_at: datetime
     files: list[DraftFile]
+    commit: str = Field(default="unknown", description="Git commit hash of the build that created this draft")
 
 
 class FinalizeTrack(BaseModel):

delivery-kid/pinning-service/app/routes/coconut.py

@@ -0,0 +1,206 @@
+"""Coconut.co cloud transcoding routes.
+
+POST /transcode-coconut  — submit a video for AV1 HLS transcoding
+POST /webhook/coconut     — receive completion/failure from Coconut
+GET  /job/{job_id}        — check job status
+GET  /jobs                — list recent jobs
+"""
+
+import logging
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, File, UploadFile, HTTPException, Request
+from pydantic import BaseModel
+
+from ..auth import require_auth
+from ..config import get_settings, Settings
+from ..services import ipfs
+from ..services.coconut import (
+    submit_to_coconut,
+    save_job,
+    load_job,
+    list_jobs,
+    process_completed_job,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["coconut"])
+
+
+class TranscodeRequest(BaseModel):
+    qualities: list[int] = [720, 480]
+    keep_original: bool = False
+
+
+class TranscodeResponse(BaseModel):
+    """Response uses camelCase to match arthel frontend expectations."""
+    jobId: str
+    coconutJobId: str | None = None
+    status: str
+    sourceCid: str | None = None
+    message: str
+
+
+@router.post("/transcode-coconut", response_model=TranscodeResponse)
+async def transcode_coconut(
+    video: UploadFile = File(...),
+    identity: str = Depends(require_auth),
+    settings: Settings = Depends(get_settings),
+):
+    """Submit a video for AV1 HLS transcoding via Coconut.co.
+
+    Pins the source video to IPFS first (so Coconut can fetch it),
+    then submits a transcoding job. Poll GET /job/{job_id} for status.
+    """
+    if not settings.coconut_api_key:
+        raise HTTPException(500, "Coconut API not configured")
+
+    job_id = f"coconut-{int(time.time())}-{id(video) % 100000:05d}"
+    staging_dir = Path(settings.staging_dir)
+
+    try:
+        # Save uploaded video to staging
+        video_dir = staging_dir / f"coconut-src-{job_id}"
+        video_dir.mkdir(parents=True, exist_ok=True)
+        video_path = video_dir / (video.filename or "video.mp4")
+
+        content = await video.read()
+        video_path.write_bytes(content)
+
+        # Pin source to IPFS so Coconut can fetch it via gateway
+        logger.info("[%s] Pinning source video to IPFS...", job_id)
+        pin_result = await ipfs.add_file(video_path)
+
+        if not pin_result.success:
+            raise HTTPException(500, f"Failed to pin source video: {pin_result.error}")
+
+        source_cid = pin_result.cid
+        source_url = f"{settings.ipfs_gateway_url}/ipfs/{source_cid}"
+        logger.info("[%s] Source pinned: %s", job_id, source_cid)
+
+        # Parse qualities from form data (arthel frontend sends as JSON string)
+        # but our Pydantic model handles the default
+
+        # Build webhook URL
+        base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1)
+        webhook_url = f"{base_url}/webhook/coconut?job_id={job_id}"
+
+        # Submit to Coconut
+        logger.info("[%s] Submitting to Coconut...", job_id)
+        coconut_result = await submit_to_coconut(
+            source_url=source_url,
+            api_key=settings.coconut_api_key,
+            webhook_url=webhook_url,
+            qualities=[720, 480],  # Default for now
+        )
+
+        coconut_job_id = coconut_result.get("id")
+        logger.info("[%s] Coconut job created: %s", job_id, coconut_job_id)
+
+        # Save job state (camelCase to match arthel frontend polling)
+        job_state = {
+            "id": job_id,
+            "coconutJobId": coconut_job_id,
+            "status": "processing",
+            "sourceCid": source_cid,
+            "keepOriginal": False,
+            "createdAt": datetime.now(timezone.utc).isoformat(),
+            "identity": identity,
+        }
+        save_job(staging_dir, job_id, job_state)
+
+        # Clean up temp video file (source is on IPFS now)
+        import shutil
+        shutil.rmtree(video_dir, ignore_errors=True)
+
+        return TranscodeResponse(
+            jobId=job_id,
+            coconutJobId=coconut_job_id,
+            status="processing",
+            sourceCid=source_cid,
+            message="Video submitted for transcoding. Check /job/{job_id} for status.",
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("[%s] Error: %s", job_id, e)
+        raise HTTPException(500, str(e))
+
+
+@router.post("/webhook/coconut")
+async def webhook_coconut(request: Request, settings: Settings = Depends(get_settings)):
+    """Receive completion/failure webhook from Coconut.co."""
+    job_id = request.query_params.get("job_id")
+    if not job_id:
+        raise HTTPException(400, "Missing job_id")
+
+    staging_dir = Path(settings.staging_dir)
+    job = load_job(staging_dir, job_id)
+    if not job:
+        raise HTTPException(404, "Job not found")
+
+    event = await request.json()
+    event_type = event.get("event", "unknown")
+    logger.info("[%s] Coconut webhook: %s", job_id, event_type)
+
+    try:
+        if event_type == "job.completed":
+            outputs = event.get("outputs", {})
+            hls_cid = await process_completed_job(
+                job=job,
+                outputs=outputs,
+                staging_dir=staging_dir,
+                ipfs_api_url=settings.ipfs_api_url,
+                pinata_jwt=settings.pinata_jwt,
+            )
+
+            if hls_cid:
+                job["status"] = "complete"
+                job["hlsCid"] = hls_cid
+                job["completedAt"] = datetime.now(timezone.utc).isoformat()
+                logger.info("[%s] Job complete! HLS CID: %s", job_id, hls_cid)
+            else:
+                job["status"] = "failed"
+                job["error"] = "Failed to pin HLS output to IPFS"
+
+        elif event_type == "job.failed":
+            logger.error("[%s] Coconut job failed: %s", job_id, event.get("error"))
+            job["status"] = "failed"
+            job["error"] = event.get("error", "Unknown error")
+            job["failedAt"] = datetime.now(timezone.utc).isoformat()
+
+        save_job(staging_dir, job_id, job)
+        return {"received": True}
+
+    except Exception as e:
+        logger.error("[%s] Webhook processing error: %s", job_id, e)
+        job["status"] = "failed"
+        job["error"] = str(e)
+        save_job(staging_dir, job_id, job)
+        raise HTTPException(500, str(e))
+
+
+@router.get("/job/{job_id}")
+async def get_job_status(
+    job_id: str,
+    settings: Settings = Depends(get_settings),
+):
+    """Get transcoding job status."""
+    job = load_job(Path(settings.staging_dir), job_id)
+    if not job:
+        raise HTTPException(404, "Job not found")
+    return job
+
+
+@router.get("/jobs")
+async def get_jobs(
+    identity: str = Depends(require_auth),
+    settings: Settings = Depends(get_settings),
+):
+    """List recent transcoding jobs."""
+    jobs = list_jobs(Path(settings.staging_dir))
+    return {"jobs": jobs}