diff --git a/delivery-kid/ansible/playbook.yml b/delivery-kid/ansible/playbook.yml index 42b6922..d0d131d 100644 --- a/delivery-kid/ansible/playbook.yml +++ b/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,8 @@ reverse_proxy /draft-content* localhost:3001 reverse_proxy /unpin/* localhost:3001 reverse_proxy /enrich* localhost:3001 + reverse_proxy /torrent* localhost:3001 + reverse_proxy /staging/* localhost:3001 respond "delivery-kid pinning service" 200 } } diff --git a/delivery-kid/pinning-service/Dockerfile b/delivery-kid/pinning-service/Dockerfile index e27101f..1db96a0 100644 --- a/delivery-kid/pinning-service/Dockerfile +++ b/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"] diff --git a/delivery-kid/pinning-service/app/auth.py b/delivery-kid/pinning-service/app/auth.py index 87388db..c947688 100644 --- a/delivery-kid/pinning-service/app/auth.py +++ b/delivery-kid/pinning-service/app/auth.py @@ -16,20 +16,35 @@ logger = logging.getLogger(__name__) -def create_upload_token(api_key: str, username: str, timestamp: int) -> str: - """Create an HMAC upload token for wiki-authenticated users.""" - message = f"upload:{username}:{timestamp}" +def create_upload_token(api_key: str, username: str, timestamp: int, action: str = "upload") -> str: + """Create an HMAC token for wiki-authenticated users. + + Args: + api_key: Shared secret between wiki and delivery-kid. + username: Wiki username. + timestamp: Millisecond timestamp. + action: Token action prefix — "upload" for staging, "finalize" for pinning. + """ + message = f"{action}:{username}:{timestamp}" return hmac.new(api_key.encode(), message.encode(), hashlib.sha256).hexdigest() -def verify_upload_token(token: str, username: str, timestamp: int, settings: Settings) -> bool: - """Verify an HMAC upload token.""" +def verify_upload_token(token: str, username: str, timestamp: int, settings: Settings, action: str = "upload") -> bool: + """Verify an HMAC token. + + Args: + token: The HMAC token to verify. + username: Wiki username claimed. + timestamp: Millisecond timestamp claimed. + settings: App settings. + action: Expected action prefix — "upload" or "finalize". + """ if not settings.api_key: logger.warning("HMAC verify failed: no api_key configured") return False - expected = create_upload_token(settings.api_key, username, timestamp) + expected = create_upload_token(settings.api_key, username, timestamp, action=action) if not hmac.compare_digest(token, expected): - logger.warning("HMAC verify failed: token mismatch for user=%s", username) + logger.warning("HMAC verify failed: token mismatch for user=%s action=%s", username, action) return False # Check timestamp freshness now_ms = int(time.time() * 1000) @@ -38,8 +53,8 @@ def verify_upload_token(token: str, username: str, timestamp: int, settings: Set if drift_ms > max_drift_ms: logger.warning( "HMAC verify failed: token expired. drift=%dms (max=%dms), " - "token_ts=%d, server_now=%d, user=%s", - drift_ms, max_drift_ms, timestamp, now_ms, username + "token_ts=%d, server_now=%d, user=%s, action=%s", + drift_ms, max_drift_ms, timestamp, now_ms, username, action ) return False return True @@ -103,6 +118,34 @@ def verify_signature(signature: str, timestamp: int, settings: Settings) -> Auth return AuthResult(valid=True, address=address) +def _verify_hmac_headers(request: Request, settings: Settings, action: str = "upload") -> str | None: + """Verify HMAC upload/finalize token from request headers. + + Returns identity string on success, None if headers not present. + Raises HTTPException on invalid token. + """ + upload_token = request.headers.get("X-Upload-Token") + if not upload_token: + return None + + username = request.headers.get("X-Upload-User", "") + timestamp_str = request.headers.get("X-Upload-Timestamp", "") + try: + timestamp = int(timestamp_str) + except (ValueError, TypeError): + raise HTTPException( + status_code=401, + detail={"error": "Invalid upload timestamp"} + ) + + if not verify_upload_token(upload_token, username, timestamp, settings, action=action): + raise HTTPException( + status_code=401, + detail={"error": "Invalid or expired upload token"} + ) + return f"wiki:{username}" + + async def require_wallet_auth( request: Request, settings: Settings = Depends(get_settings) @@ -153,25 +196,47 @@ async def require_auth( Returns an identity string. """ # 1. HMAC upload token (wiki-issued, for direct browser uploads) - upload_token = request.headers.get("X-Upload-Token") - if upload_token: - username = request.headers.get("X-Upload-User", "") - timestamp_str = request.headers.get("X-Upload-Timestamp", "") - try: - timestamp = int(timestamp_str) - except (ValueError, TypeError): + identity = _verify_hmac_headers(request, settings, action="upload") + if identity: + return identity + + # 2. API key (server-to-server) + api_key = request.headers.get("X-API-Key") + if api_key: + if not settings.api_key: raise HTTPException( - status_code=401, - detail={"error": "Invalid upload timestamp"} + status_code=500, + detail={"error": "API key auth not configured on server"} ) - if not verify_upload_token(upload_token, username, timestamp, settings): + if api_key != settings.api_key: raise HTTPException( status_code=401, - detail={"error": "Invalid or expired upload token"} + detail={"error": "Invalid API key"} ) - return f"wiki:{username}" + return request.headers.get("X-Uploaded-By", "api-user") - # 2. API key (server-to-server) + # 3. Wallet signature + return await require_wallet_auth(request, settings) + + +async def require_finalize_auth( + request: Request, + settings: Settings = Depends(get_settings) +) -> str: + """ + FastAPI dependency for finalization endpoints. + + Requires a finalize-prefixed HMAC token (issued only to users with + finalize-release permission), an API key, or a wallet signature. + + Returns an identity string. + """ + # 1. HMAC finalize token (wiki-issued, for finalize-release users) + identity = _verify_hmac_headers(request, settings, action="finalize") + if identity: + return identity + + # 2. API key (server-to-server — always allowed to finalize) api_key = request.headers.get("X-API-Key") if api_key: if not settings.api_key: diff --git a/delivery-kid/pinning-service/app/config.py b/delivery-kid/pinning-service/app/config.py index 5bb4bd3..c11c1bf 100644 --- a/delivery-kid/pinning-service/app/config.py +++ b/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) @@ -40,7 +47,7 @@ class Settings(BaseSettings): "https://www.cryptograss.live", "https://pickipedia.xyz", ] - cors_origin_regex: str = r"https://\w+\d*\.hunter\.cryptograss\.live" + cors_origin_regex: str = r"https://[\w.-]+\.hunter\.cryptograss\.live" class Config: env_file = ".env" @@ -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") diff --git a/delivery-kid/pinning-service/app/main.py b/delivery-kid/pinning-service/app/main.py index ba43a01..6ad580e 100644 --- a/delivery-kid/pinning-service/app/main.py +++ b/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, staging 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,9 @@ 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.include_router(staging.router) @app.get("/") diff --git a/delivery-kid/pinning-service/app/models/content.py b/delivery-kid/pinning-service/app/models/content.py index 5b9d4ec..3719600 100644 --- a/delivery-kid/pinning-service/app/models/content.py +++ b/delivery-kid/pinning-service/app/models/content.py @@ -1,5 +1,6 @@ """Pydantic models for general content drafts (video, audio, arbitrary files).""" +import secrets from datetime import datetime from typing import Optional from pydantic import BaseModel, Field @@ -23,6 +24,7 @@ class ContentFile(BaseModel): audio_codec: Optional[str] = None # Common size_bytes: int + creation_time: Optional[str] = Field(default=None, description="ISO 8601 creation time from container metadata") class ContentDraftState(BaseModel): @@ -34,6 +36,13 @@ class ContentDraftState(BaseModel): uploaded_by: str = Field(description="Wallet address that created the draft") files: list[ContentFile] metadata: dict = Field(default_factory=dict, description="User-supplied metadata") + # Preview transcoding (background, after upload) + preview_token: str = Field(default_factory=lambda: secrets.token_urlsafe(32), + description="One-time token for Coconut to fetch source video from staging") + preview_status: str = Field(default="none", description="none, pending, processing, ready, failed") + preview_job_id: Optional[str] = Field(default=None, description="Coconut job ID for preview transcode") + preview_cid: Optional[str] = Field(default=None, description="IPFS CID of AV1 HLS output") + preview_mp4_cid: Optional[str] = Field(default=None, description="IPFS CID of 480p H.264 preview MP4") class ContentDraftResponse(BaseModel): @@ -43,6 +52,10 @@ 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") + preview_status: str = Field(default="none", description="none, pending, processing, ready, failed") + preview_cid: Optional[str] = Field(default=None, description="IPFS CID of AV1 HLS output") + preview_mp4_cid: Optional[str] = Field(default=None, description="IPFS CID of 480p preview MP4") class ContentFinalizeRequest(BaseModel): @@ -51,5 +64,23 @@ 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." + ) + trim_start_seconds: Optional[float] = Field( + default=None, description="Start time in seconds for trimming the video" + ) + trim_end_seconds: Optional[float] = Field( + default=None, description="End time in seconds for trimming the video" + ) + preserve_original: bool = Field( + default=False, description="Save the original source file to permanent storage instead of deleting it" + ) diff --git a/delivery-kid/pinning-service/app/models/draft.py b/delivery-kid/pinning-service/app/models/draft.py index 3d2c432..918221e 100644 --- a/delivery-kid/pinning-service/app/models/draft.py +++ b/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): diff --git a/delivery-kid/pinning-service/app/routes/coconut.py b/delivery-kid/pinning-service/app/routes/coconut.py new file mode 100644 index 0000000..c8eec43 --- /dev/null +++ b/delivery-kid/pinning-service/app/routes/coconut.py @@ -0,0 +1,237 @@ +"""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 json +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 ..models.content import ContentDraftState +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"]) + + +def _update_draft_preview(staging_dir: Path, job: dict) -> None: + """Update a content draft's preview state after Coconut webhook.""" + draft_id = job["draftId"] + draft_json = staging_dir / "drafts" / draft_id / "draft.json" + if not draft_json.exists(): + logger.warning("[%s] Preview draft not found: %s", job["id"], draft_id) + return + try: + data = json.loads(draft_json.read_text()) + if job["status"] == "complete" and job.get("hlsCid"): + data["preview_status"] = "ready" + data["preview_cid"] = job["hlsCid"] + # previewCid = 480p MP4 for the video player on the ReleaseDraft page + if job.get("previewCid"): + data["preview_mp4_cid"] = job["previewCid"] + logger.info("[%s] Draft %s preview ready: hls=%s mp4=%s", + job["id"], draft_id[:8], job["hlsCid"], job.get("previewCid", "none")) + else: + data["preview_status"] = "failed" + logger.warning("[%s] Draft %s preview failed", job["id"], draft_id[:8]) + draft_json.write_text(json.dumps(data, indent=2, default=str)) + except Exception as e: + logger.error("[%s] Failed to update draft preview state: %s", job["id"], e) + + +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) + + # If this is a preview job, update the draft state + if job.get("isPreview") and job.get("draftId"): + _update_draft_preview(staging_dir, 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} diff --git a/delivery-kid/pinning-service/app/routes/content.py b/delivery-kid/pinning-service/app/routes/content.py index 4481cf0..7e024f8 100644 --- a/delivery-kid/pinning-service/app/routes/content.py +++ b/delivery-kid/pinning-service/app/routes/content.py @@ -2,7 +2,9 @@ import asyncio import json +import logging import shutil +import time import uuid from datetime import datetime, timedelta, timezone from pathlib import Path @@ -10,12 +12,15 @@ from fastapi import APIRouter, Depends, File, UploadFile, HTTPException from sse_starlette.sse import EventSourceResponse -from ..auth import require_auth -from ..config import get_settings, Settings +from ..auth import require_auth, require_finalize_auth +from ..config import get_settings, get_commit, Settings from ..models.content import ( ContentFile, ContentDraftState, ContentDraftResponse, ContentFinalizeRequest ) from ..services import analyze, ipfs, transcode +from ..services.coconut import submit_to_coconut, save_job, load_job + +logger = logging.getLogger(__name__) router = APIRouter(prefix="/draft-content", tags=["content"]) @@ -123,6 +128,7 @@ async def create_content_draft( video_codec=a.video_codec, audio_codec=a.audio_codec, size_bytes=a.size_bytes, + creation_time=a.creation_time, )) if not draft_files: @@ -136,6 +142,10 @@ async def create_content_draft( now = datetime.now(timezone.utc) expires_at = now + timedelta(hours=settings.draft_ttl_hours) + # Determine if this is a single-video upload that should get a preview + video_files = [f for f in draft_files if f.media_type == "video"] + should_preview = len(draft_files) == 1 and len(video_files) == 1 and settings.coconut_api_key + state = ContentDraftState( draft_id=draft_id, draft_type="content", @@ -143,13 +153,22 @@ async def create_content_draft( expires_at=expires_at, uploaded_by=wallet_address, files=draft_files, + preview_status="pending" if should_preview else "none", ) save_draft_state(draft_dir, state) + # Kick off background preview transcoding for video uploads + if should_preview: + asyncio.create_task( + _submit_preview_transcode(draft_id, state, settings) + ) + return ContentDraftResponse( draft_id=draft_id, expires_at=expires_at, files=draft_files, + commit=get_commit(), + preview_status=state.preview_status, ) except HTTPException: @@ -186,6 +205,10 @@ async def get_content_draft( expires_at=state.expires_at, files=state.files, metadata=state.metadata, + commit=get_commit(), + preview_status=state.preview_status, + preview_cid=state.preview_cid, + preview_mp4_cid=state.preview_mp4_cid, ) @@ -210,6 +233,94 @@ async def delete_content_draft( return {"message": "Draft deleted", "draft_id": draft_id} +async def _submit_preview_transcode( + draft_id: str, state: ContentDraftState, settings: Settings +) -> None: + """Background task: submit video to Coconut for AV1 HLS preview. + + Coconut fetches the source from our staging endpoint via preview_token, + transcodes to AV1 HLS, and delivers via webhook. The webhook handler + pins the HLS output to IPFS and updates draft state with the CID. + """ + staging_dir = Path(settings.staging_dir) + draft_dir = get_draft_dir(staging_dir, draft_id) + + try: + video_file = state.files[0] + + # Build the source URL: Coconut will fetch from our staging endpoint + # using the preview_token for auth (no IPFS pin of the original needed) + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) + source_url = ( + f"{base_url}/staging/drafts/{draft_id}/{video_file.original_filename}" + f"?preview_token={state.preview_token}" + ) + + # Build webhook URL — reuses existing /webhook/coconut handler + job_id = f"preview-{draft_id[:12]}-{int(time.time())}" + webhook_url = f"{base_url}/webhook/coconut?job_id={job_id}" + + logger.info("[preview:%s] Submitting to Coconut, source=%s", draft_id[:8], source_url[:80]) + + coconut_result = await submit_to_coconut( + source_url=source_url, + api_key=settings.coconut_api_key, + webhook_url=webhook_url, + include_preview=True, + ) + coconut_job_id = coconut_result.get("id") + logger.info("[preview:%s] Coconut job created: %s", draft_id[:8], coconut_job_id) + + # Save job state for the webhook handler + job_state = { + "id": job_id, + "coconutJobId": coconut_job_id, + "status": "processing", + "draftId": draft_id, + "isPreview": True, + "createdAt": datetime.now(timezone.utc).isoformat(), + "identity": state.uploaded_by, + } + save_job(staging_dir, job_id, job_state) + + # Update draft state + state.preview_status = "processing" + state.preview_job_id = job_id + save_draft_state(draft_dir, state) + + except Exception as e: + logger.error("[preview:%s] Failed to submit preview: %s", draft_id[:8], e) + try: + state.preview_status = "failed" + save_draft_state(draft_dir, state) + except Exception: + pass + + +def _should_use_coconut(request: ContentFinalizeRequest, settings: Settings) -> bool: + """Determine if we should try Coconut cloud transcoding.""" + strategy = request.transcoding_strategy + if strategy == "none": + return False + if strategy == "local": + return False + if strategy == "coconut": + return bool(settings.coconut_api_key) + # "auto" — use Coconut if available, otherwise local + return bool(settings.coconut_api_key) + + +def _should_transcode_video(request: ContentFinalizeRequest) -> bool: + """Determine if video transcoding is requested.""" + if request.transcoding_strategy == "none": + return False + # Legacy field support + if request.transcode_hls: + return True + # Auto/coconut/local all imply transcoding for video + return request.transcoding_strategy in ("auto", "coconut", "local") + + async def finalize_sse_generator( draft_id: str, request: ContentFinalizeRequest, @@ -217,11 +328,19 @@ async def finalize_sse_generator( state: ContentDraftState, settings: Settings ): - """SSE generator for content finalization — transcode if needed, then pin.""" + """SSE generator for content finalization — transcode if needed, then pin. + + Fast path: if preview transcoding already produced an HLS CID and no trim + is requested, finalization is instant — just emit the existing CID. + Slow path (trim requested or no preview): Coconut cloud transcoding first, + local ffmpeg fallback. Coconut fetches source from staging via preview_token. + """ async def send_event(event: str, data: dict): return {"event": event, "data": json.dumps(data)} + has_trim = request.trim_start_seconds is not None or request.trim_end_seconds is not None + try: upload_dir = draft_dir / "upload" output_dir = draft_dir / "output" @@ -233,15 +352,117 @@ async def send_event(event: str, data: dict): "progress": 5 }) - # Determine what we're working with + # Preserve original file if requested + if request.preserve_original: + originals_dir = Path(settings.staging_dir) / "originals" / draft_id + originals_dir.mkdir(parents=True, exist_ok=True) + for f in state.files: + src = upload_dir / f.original_filename + if src.exists(): + shutil.copy2(src, originals_dir / f.original_filename) + logger.info("[content:%s] Original files preserved to %s", draft_id[:8], originals_dir) + video_files = [f for f in state.files if f.media_type == "video"] - audio_files = [f for f in state.files if f.media_type == "audio"] - image_files = [f for f in state.files if f.media_type == "image"] + wants_transcode = len(state.files) == 1 and video_files and _should_transcode_video(request) + + # === Fast path: preview already done, no trim === + if wants_transcode and state.preview_cid and not has_trim: + logger.info("[content:%s] Using existing preview HLS: %s", draft_id[:8], state.preview_cid) + yield await send_event("progress", { + "stage": "transcode", + "message": "AV1 HLS already transcoded — using preview.", + "progress": 80 + }) + + gateway_url = f"{settings.ipfs_gateway_url}/ipfs/{state.preview_cid}" + + yield await send_event("complete", { + "cid": state.preview_cid, + "gateway_url": gateway_url, + "title": request.title, + "file_type": request.file_type, + "subsequent_to": request.subsequent_to, + }) + return + + # === Coconut cloud transcoding (with trim, or no preview available) === + if wants_transcode and _should_use_coconut(request, settings): + video_file = video_files[0] + src_path = upload_dir / video_file.original_filename + + # Build source URL — Coconut fetches from staging via preview_token + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) + source_url = ( + f"{base_url}/staging/drafts/{draft_id}/{video_file.original_filename}" + f"?preview_token={state.preview_token}" + ) + + trim_msg = "" + if has_trim: + s = request.trim_start_seconds or 0 + e = request.trim_end_seconds + trim_msg = f" (trimming {s:.1f}s–{e:.1f}s)" if e else f" (trimming from {s:.1f}s)" + yield await send_event("progress", { + "stage": "transcode", + "message": f"Submitting to Coconut for AV1 transcoding{trim_msg}...", + "progress": 30 + }) - # For single-file uploads: pin the file directly (or transcode first) - # For multi-file uploads: pin as a directory - if len(state.files) == 1 and video_files and request.transcode_hls: - # Single video → HLS transcode → pin directory + job_id = f"coconut-{int(time.time())}-{id(src_path) % 100000:05d}" + webhook_url = f"{base_url}/webhook/coconut?job_id={job_id}" + + try: + coconut_result = await submit_to_coconut( + source_url=source_url, + api_key=settings.coconut_api_key, + webhook_url=webhook_url, + qualities=request.transcoding_qualities, + trim_start=request.trim_start_seconds, + trim_end=request.trim_end_seconds, + ) + coconut_job_id = coconut_result.get("id") + logger.info("[content:%s] Coconut job created: %s", draft_id[:8], coconut_job_id) + + job_state = { + "id": job_id, + "coconutJobId": coconut_job_id, + "status": "processing", + "keepOriginal": request.preserve_original, + "title": request.title, + "fileType": request.file_type, + "subsequentTo": request.subsequent_to, + "createdAt": datetime.now(timezone.utc).isoformat(), + "identity": state.uploaded_by, + } + save_job(Path(settings.staging_dir), job_id, job_state) + + # Don't delete draft dir yet — source file still needed if Coconut + # hasn't fetched it. Draft TTL cleanup handles it. + + yield await send_event("transcoding-submitted", { + "jobId": job_id, + "coconutJobId": coconut_job_id, + "message": "Video submitted for AV1 cloud transcoding. HLS output will be pinned automatically when complete.", + "pollUrl": f"/job/{job_id}", + "title": request.title, + "fileType": request.file_type, + "subsequentTo": request.subsequent_to, + }) + return + + except Exception as e: + logger.warning( + "[content:%s] Coconut submission failed, falling back to local: %s", + draft_id[:8], e + ) + yield await send_event("progress", { + "stage": "transcode", + "message": "Cloud transcoding unavailable, using local ffmpeg...", + "progress": 15 + }) + + if wants_transcode: + # === Local ffmpeg transcoding path (sync) === video_file = video_files[0] src_path = upload_dir / video_file.original_filename @@ -252,7 +473,11 @@ async def send_event(event: str, data: dict): }) hls_dir = output_dir / "hls" - result = await transcode.transcode_video_to_hls(src_path, hls_dir) + result = await transcode.transcode_video_to_hls( + src_path, hls_dir, + trim_start=request.trim_start_seconds, + trim_end=request.trim_end_seconds, + ) if not result.success: yield await send_event("error", { @@ -293,7 +518,6 @@ async def send_event(event: str, data: dict): "created_at": datetime.now(timezone.utc).isoformat(), **request.metadata, } - # Remove None values metadata = {k: v for k, v in metadata.items() if v is not None} with open(pin_path / "metadata.json", "w") as f: @@ -305,7 +529,6 @@ async def send_event(event: str, data: dict): "progress": 70 }) - # Pin to IPFS result = await ipfs.add_directory(pin_path) if not result.success: @@ -346,7 +569,7 @@ async def send_event(event: str, data: dict): async def finalize_content_draft( draft_id: str, request: ContentFinalizeRequest, - wallet_address: str = Depends(require_auth), + wallet_address: str = Depends(require_finalize_auth), settings: Settings = Depends(get_settings) ): """ diff --git a/delivery-kid/pinning-service/app/routes/drafts.py b/delivery-kid/pinning-service/app/routes/drafts.py index 99b70bf..f21ee2e 100644 --- a/delivery-kid/pinning-service/app/routes/drafts.py +++ b/delivery-kid/pinning-service/app/routes/drafts.py @@ -10,8 +10,8 @@ from fastapi import APIRouter, Depends, File, UploadFile, HTTPException from sse_starlette.sse import EventSourceResponse -from ..auth import require_auth -from ..config import get_settings, Settings +from ..auth import require_auth, require_finalize_auth +from ..config import get_settings, get_commit, Settings from ..models.draft import DraftFile, DraftState, DraftResponse, FinalizeRequest from ..services import analyze, ipfs, transcode @@ -143,7 +143,8 @@ async def create_draft( return DraftResponse( draft_id=draft_id, expires_at=expires_at, - files=draft_files + files=draft_files, + commit=get_commit(), ) except HTTPException: @@ -186,7 +187,8 @@ async def get_draft( return DraftResponse( draft_id=state.draft_id, expires_at=state.expires_at, - files=state.files + files=state.files, + commit=get_commit(), ) @@ -509,7 +511,7 @@ async def send_event(event: str, data: dict): async def finalize_draft( draft_id: str, request: FinalizeRequest, - wallet_address: str = Depends(require_auth), + wallet_address: str = Depends(require_finalize_auth), settings: Settings = Depends(get_settings) ): """ diff --git a/delivery-kid/pinning-service/app/routes/enrich.py b/delivery-kid/pinning-service/app/routes/enrich.py index 9dad97f..a4d4ab5 100644 --- a/delivery-kid/pinning-service/app/routes/enrich.py +++ b/delivery-kid/pinning-service/app/routes/enrich.py @@ -21,6 +21,7 @@ from ..auth import require_auth from ..config import get_settings, Settings from ..services.torrent import create_torrent, DEFAULT_TRACKERS +from ..services.seeder import get_seeder logger = logging.getLogger(__name__) @@ -37,6 +38,8 @@ class TorrentResponse(BaseModel): cid: str infohash: str | None = None trackers: list[str] | None = None + webseeds: list[str] | None = None + torrent_url: str | None = None file_count: int | None = None total_size: int | None = None piece_length: int | None = None @@ -119,12 +122,17 @@ async def generate_torrent( try: torrent_name = req.name or cid + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) result = create_torrent( directory=album_dir, name=torrent_name, + # Multi-file: Caddy rewrites /webseed/{cid}/{name}/{file} → /ipfs/{cid}/{file} webseeds=[ - f"{settings.ipfs_gateway_url}/ipfs/{cid}/", - f"https://ipfs.io/ipfs/{cid}/", + f"{base_url}/webseed/{cid}/", + ], + # Single-file: BEP 19 fetches URL directly + single_file_webseeds=[ + f"{settings.ipfs_gateway_url}/ipfs/{cid}", ], ) @@ -135,11 +143,25 @@ async def generate_torrent( error=f"Torrent generation failed: {result.error}", ) + # Add to BitTorrent seeder (copies content to seeding dir) + torrent_url = None + seeder = get_seeder() + if seeder and result.torrent_bytes: + infohash_added = seeder.add_torrent(cid, result.torrent_bytes, album_dir) + if infohash_added: + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) + torrent_url = f"{base_url}/torrent/{result.infohash}.torrent" + logger.info("Seeding torrent for %s (infohash %s)", cid, infohash_added) + else: + logger.warning("Failed to add torrent to seeder for %s", cid) + return TorrentResponse( success=True, cid=cid, infohash=result.infohash, trackers=DEFAULT_TRACKERS, + webseeds=result.webseeds, + torrent_url=torrent_url, file_count=result.file_count, total_size=result.total_size, piece_length=result.piece_length, diff --git a/delivery-kid/pinning-service/app/routes/staging.py b/delivery-kid/pinning-service/app/routes/staging.py new file mode 100644 index 0000000..3aea60b --- /dev/null +++ b/delivery-kid/pinning-service/app/routes/staging.py @@ -0,0 +1,113 @@ +"""Serve staged draft files for preview (e.g., video embed on ReleaseDraft pages). + +Files are served from the staging directory at /drafts/{draft_id}/upload/{filename}. +Requires a valid upload token (any logged-in wiki user). Does NOT check draft +ownership — the unguessable UUID is sufficient access control for preview. + +Supports HTTP range requests for video seeking via FastAPI's FileResponse. + +Auth can be provided via headers (standard require_auth flow) OR via query +parameters (?token=...&user=...×tamp=...) so that