From 224aab37afaf2c8bd599f1ee247c60540a4c893a Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Sun, 15 Mar 2026 02:21:47 +0000 Subject: [PATCH 1/8] Add webseed rewrite to fix BEP 19 path mismatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BT clients construct webseed URLs as base_url + torrent_name + file_path, but IPFS gateway serves files without the torrent name directory level. New /webseed/{cid}/ Caddy route rewrites: /webseed/{cid}/{torrent_name}/{file} → /ipfs/{cid}/{file} Also updates enrich endpoint to use the new webseed URL. --- delivery-kid/ansible/playbook.yml | 8 ++++++++ delivery-kid/pinning-service/app/routes/enrich.py | 7 +++++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/delivery-kid/ansible/playbook.yml b/delivery-kid/ansible/playbook.yml index 42b6922..e3806d1 100644 --- a/delivery-kid/ansible/playbook.yml +++ b/delivery-kid/ansible/playbook.yml @@ -329,6 +329,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/{re.1}/{re.2} + reverse_proxy localhost:8080 + } route { reverse_proxy /health localhost:3001 reverse_proxy /version localhost:3001 diff --git a/delivery-kid/pinning-service/app/routes/enrich.py b/delivery-kid/pinning-service/app/routes/enrich.py index 9dad97f..a497485 100644 --- a/delivery-kid/pinning-service/app/routes/enrich.py +++ b/delivery-kid/pinning-service/app/routes/enrich.py @@ -119,12 +119,15 @@ async def generate_torrent( try: torrent_name = req.name or cid + # Webseed URLs use /webseed/{cid}/ path — Caddy rewrites + # /webseed/{cid}/{torrent_name}/{file} → /ipfs/{cid}/{file} + # to bridge the BEP 19 path convention with IPFS gateway paths. + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) result = create_torrent( directory=album_dir, name=torrent_name, webseeds=[ - f"{settings.ipfs_gateway_url}/ipfs/{cid}/", - f"https://ipfs.io/ipfs/{cid}/", + f"{base_url}/webseed/{cid}/", ], ) From a5a426fbad9b42c29dba558fd1a730ec37de378b Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Sun, 15 Mar 2026 02:30:53 +0000 Subject: [PATCH 2/8] Fix Caddy webseed rewrite capture group syntax Use $1/$2 (uri replacement syntax) instead of {re.1}/{re.2} (matcher syntax). --- delivery-kid/ansible/playbook.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/delivery-kid/ansible/playbook.yml b/delivery-kid/ansible/playbook.yml index e3806d1..6da3bcf 100644 --- a/delivery-kid/ansible/playbook.yml +++ b/delivery-kid/ansible/playbook.yml @@ -334,7 +334,7 @@ # /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/{re.1}/{re.2} + uri path_regexp ^/webseed/([^/]+)/[^/]+/(.+)$ /ipfs/$1/$2 reverse_proxy localhost:8080 } route { From c22d9d669aab8d3806fb765098b6501b35068ce2 Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Sun, 15 Mar 2026 02:39:55 +0000 Subject: [PATCH 3/8] Use single-file torrent format for single-file CIDs BEP 19 webseed behavior differs by torrent type: - Single-file: client fetches the URL directly - Multi-file: client appends name/path to the URL Most releases (23/26) are single video files. Using single-file format means the webseed URL can point directly to the IPFS gateway without any path rewriting needed. Multi-file torrents (albums) still use the /webseed/ rewrite route. --- .../pinning-service/app/routes/enrich.py | 8 ++- .../pinning-service/app/services/torrent.py | 60 +++++++++++++------ 2 files changed, 46 insertions(+), 22 deletions(-) diff --git a/delivery-kid/pinning-service/app/routes/enrich.py b/delivery-kid/pinning-service/app/routes/enrich.py index a497485..9972ae9 100644 --- a/delivery-kid/pinning-service/app/routes/enrich.py +++ b/delivery-kid/pinning-service/app/routes/enrich.py @@ -119,16 +119,18 @@ async def generate_torrent( try: torrent_name = req.name or cid - # Webseed URLs use /webseed/{cid}/ path — Caddy rewrites - # /webseed/{cid}/{torrent_name}/{file} → /ipfs/{cid}/{file} - # to bridge the BEP 19 path convention with IPFS gateway paths. 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"{base_url}/webseed/{cid}/", ], + # Single-file: BEP 19 fetches URL directly + single_file_webseeds=[ + f"{settings.ipfs_gateway_url}/ipfs/{cid}", + ], ) if not result.success: diff --git a/delivery-kid/pinning-service/app/services/torrent.py b/delivery-kid/pinning-service/app/services/torrent.py index 4dd0dea..f8d1217 100644 --- a/delivery-kid/pinning-service/app/services/torrent.py +++ b/delivery-kid/pinning-service/app/services/torrent.py @@ -91,6 +91,7 @@ def create_torrent( output_path: Optional[Path] = None, trackers: Optional[list[str]] = None, webseeds: Optional[list[str]] = None, + single_file_webseeds: Optional[list[str]] = None, comment: Optional[str] = None, ) -> TorrentResult: """ @@ -99,12 +100,18 @@ def create_torrent( The infohash depends only on: file contents, file paths (sorted), piece length (deterministic from total size), and name. + Uses single-file torrent format when the directory contains exactly + one file (common for video releases). This matters for BEP 19 webseed + compatibility: single-file torrents fetch the URL directly, while + multi-file torrents append name/path to the URL. + Args: directory: Path to the directory to torrent - name: Torrent name (use CID for determinism) + name: Torrent name output_path: Where to write the .torrent file (optional) trackers: List of tracker announce URLs (outside info dict, doesn't affect infohash) - webseeds: List of webseed URLs (outside info dict, doesn't affect infohash) + webseeds: List of webseed URLs for multi-file torrents (outside info dict) + single_file_webseeds: Webseed URLs for single-file torrents (BEP 19 fetches directly) comment: Optional comment (outside info dict, doesn't affect infohash) Returns: @@ -124,6 +131,7 @@ def create_torrent( if not files: return TorrentResult(success=False, error="No files in directory") + is_single_file = len(files) == 1 total_size = sum(size for _, size, _ in files) piece_length = _deterministic_piece_length(total_size) @@ -147,20 +155,30 @@ def create_torrent( if piece_buffer: pieces += hashlib.sha1(piece_buffer).digest() - # Build info dict (only deterministic fields) - file_list = [] - for rel_path, size, _ in files: - file_list.append({ + # Build info dict — single-file or multi-file format + if is_single_file: + # Single-file torrent: name is the filename, length at top level + _, size, _ = files[0] + info = { b"length": size, - b"path": [part.encode("utf-8") for part in rel_path.parts], - }) - - info = { - b"files": file_list, - b"name": name.encode("utf-8"), - b"piece length": piece_length, - b"pieces": pieces, - } + b"name": name.encode("utf-8"), + b"piece length": piece_length, + b"pieces": pieces, + } + else: + # Multi-file torrent: name is directory name, files list + file_list = [] + for rel_path, size, _ in files: + file_list.append({ + b"length": size, + b"path": [part.encode("utf-8") for part in rel_path.parts], + }) + info = { + b"files": file_list, + b"name": name.encode("utf-8"), + b"piece length": piece_length, + b"pieces": pieces, + } # Compute infohash info_bencoded = _bencode(info) @@ -178,11 +196,15 @@ def create_torrent( metainfo[b"announce-list"] = [[t.encode("utf-8")] for t in tracker_list] # Webseeds (BEP 19 url-list, outside info dict) - if webseeds: - if len(webseeds) == 1: - metainfo[b"url-list"] = webseeds[0].encode("utf-8") + # Single-file torrents: client fetches the URL directly + # Multi-file torrents: client appends name/path to the URL + ws_urls = (single_file_webseeds if is_single_file and single_file_webseeds + else webseeds) + if ws_urls: + if len(ws_urls) == 1: + metainfo[b"url-list"] = ws_urls[0].encode("utf-8") else: - metainfo[b"url-list"] = [ws.encode("utf-8") for ws in webseeds] + metainfo[b"url-list"] = [ws.encode("utf-8") for ws in ws_urls] if comment: metainfo[b"comment"] = comment.encode("utf-8") From 40a629681b3738a704190ace1b9ed9e024366c67 Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Sun, 15 Mar 2026 10:38:11 +0000 Subject: [PATCH 4/8] Return webseed URLs in torrent endpoint response The endpoint now returns the actual webseed URLs used in the torrent, so Blue Railroad can store them in Release YAML and the PHP renderer can use them directly in magnet links. This is needed because single-file and multi-file torrents use different webseed URL formats. --- delivery-kid/pinning-service/app/routes/enrich.py | 2 ++ delivery-kid/pinning-service/app/services/torrent.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/delivery-kid/pinning-service/app/routes/enrich.py b/delivery-kid/pinning-service/app/routes/enrich.py index 9972ae9..ff562f4 100644 --- a/delivery-kid/pinning-service/app/routes/enrich.py +++ b/delivery-kid/pinning-service/app/routes/enrich.py @@ -37,6 +37,7 @@ class TorrentResponse(BaseModel): cid: str infohash: str | None = None trackers: list[str] | None = None + webseeds: list[str] | None = None file_count: int | None = None total_size: int | None = None piece_length: int | None = None @@ -145,6 +146,7 @@ async def generate_torrent( cid=cid, infohash=result.infohash, trackers=DEFAULT_TRACKERS, + webseeds=result.webseeds, file_count=result.file_count, total_size=result.total_size, piece_length=result.piece_length, diff --git a/delivery-kid/pinning-service/app/services/torrent.py b/delivery-kid/pinning-service/app/services/torrent.py index f8d1217..a0cb897 100644 --- a/delivery-kid/pinning-service/app/services/torrent.py +++ b/delivery-kid/pinning-service/app/services/torrent.py @@ -82,6 +82,7 @@ class TorrentResult: piece_length: Optional[int] = None total_size: Optional[int] = None file_count: Optional[int] = None + webseeds: Optional[list[str]] = None error: Optional[str] = None @@ -227,4 +228,5 @@ def create_torrent( piece_length=piece_length, total_size=total_size, file_count=len(files), + webseeds=ws_urls or [], ) From ad1773990091dfb6ef039accb17b2d7166804b63 Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Sun, 15 Mar 2026 14:57:13 +0000 Subject: [PATCH 5/8] Add integrated libtorrent seeder to delivery-kid Instead of relying solely on webseeds (which require torrent metadata that magnet-link-only clients can't bootstrap without peers), this adds a proper libtorrent seeder that participates in the BitTorrent swarm. - New seeder service: manages libtorrent session, seeds all enriched content from /staging/seeding/{cid}/ directories - Torrent file serving: GET /torrent/{infohash}.torrent endpoint - Seeder status: GET /torrent/status for monitoring - Enrich endpoint now adds torrents to seeder and returns torrent_url - Dockerfile exposes port 6881 for BitTorrent traffic - Playbook forwards BT ports (6881 TCP+UDP) and adds /torrent* route - Added libtorrent dependency to requirements.txt --- delivery-kid/ansible/playbook.yml | 3 + delivery-kid/pinning-service/Dockerfile | 1 + delivery-kid/pinning-service/app/config.py | 3 + delivery-kid/pinning-service/app/main.py | 12 +- .../pinning-service/app/routes/enrich.py | 15 ++ .../pinning-service/app/routes/torrent.py | 35 +++ .../pinning-service/app/services/seeder.py | 247 ++++++++++++++++++ delivery-kid/pinning-service/requirements.txt | 3 + 8 files changed, 317 insertions(+), 2 deletions(-) create mode 100644 delivery-kid/pinning-service/app/routes/torrent.py create mode 100644 delivery-kid/pinning-service/app/services/seeder.py diff --git a/delivery-kid/ansible/playbook.yml b/delivery-kid/ansible/playbook.yml index 6da3bcf..cc0c4c7 100644 --- a/delivery-kid/ansible/playbook.yml +++ b/delivery-kid/ansible/playbook.yml @@ -132,6 +132,8 @@ - /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 @@ -352,6 +354,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 } } 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/config.py b/delivery-kid/pinning-service/app/config.py index 5bb4bd3..852e1c7 100644 --- a/delivery-kid/pinning-service/app/config.py +++ b/delivery-kid/pinning-service/app/config.py @@ -19,6 +19,9 @@ 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 = "" diff --git a/delivery-kid/pinning-service/app/main.py b/delivery-kid/pinning-service/app/main.py index ba43a01..2529aeb 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 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,7 @@ 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.get("/") diff --git a/delivery-kid/pinning-service/app/routes/enrich.py b/delivery-kid/pinning-service/app/routes/enrich.py index ff562f4..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__) @@ -38,6 +39,7 @@ class TorrentResponse(BaseModel): 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 @@ -141,12 +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/torrent.py b/delivery-kid/pinning-service/app/routes/torrent.py new file mode 100644 index 0000000..6f2da82 --- /dev/null +++ b/delivery-kid/pinning-service/app/routes/torrent.py @@ -0,0 +1,35 @@ +"""Serve .torrent files for download.""" + +from fastapi import APIRouter, HTTPException +from fastapi.responses import Response + +from ..services.seeder import get_seeder + +router = APIRouter(prefix="/torrent", tags=["torrent"]) + + +@router.get("/{infohash}.torrent") +async def get_torrent_file(infohash: str): + """Serve a .torrent file by infohash.""" + seeder = get_seeder() + if not seeder: + raise HTTPException(503, "Seeder not running") + + torrent_bytes = seeder.get_torrent_file(infohash) + if not torrent_bytes: + raise HTTPException(404, "Torrent not found") + + return Response( + content=torrent_bytes, + media_type="application/x-bittorrent", + headers={"Content-Disposition": f'attachment; filename="{infohash}.torrent"'}, + ) + + +@router.get("/status") +async def seeder_status(): + """Get seeder status.""" + seeder = get_seeder() + if not seeder: + return {"running": False} + return seeder.status() diff --git a/delivery-kid/pinning-service/app/services/seeder.py b/delivery-kid/pinning-service/app/services/seeder.py new file mode 100644 index 0000000..1e573fe --- /dev/null +++ b/delivery-kid/pinning-service/app/services/seeder.py @@ -0,0 +1,247 @@ +"""BitTorrent seeder — keeps a libtorrent session alive to seed generated torrents.""" + +import logging +import shutil +from pathlib import Path +from typing import Optional + +import libtorrent as lt + +logger = logging.getLogger(__name__) + + +class Seeder: + """Manages a libtorrent session for seeding torrents.""" + + def __init__(self, seeding_dir: str, listen_ports: tuple[int, int] = (6881, 6891)): + self.seeding_dir = Path(seeding_dir) + self.seeding_dir.mkdir(parents=True, exist_ok=True) + self.listen_ports = listen_ports + self.session: Optional[lt.session] = None + self._handles: dict[str, lt.torrent_handle] = {} # infohash -> handle + self._torrent_files: dict[str, bytes] = {} # infohash -> .torrent bytes + + def start(self): + """Start the libtorrent session and load existing torrents.""" + settings = { + 'listen_interfaces': f'0.0.0.0:{self.listen_ports[0]}', + 'enable_dht': True, + 'enable_lsd': True, + 'enable_upnp': False, # VPS, no UPnP + 'enable_natpmp': False, + 'alert_mask': ( + lt.alert.category_t.error_notification + | lt.alert.category_t.status_notification + ), + } + self.session = lt.session(settings) + logger.info("libtorrent session started on port %d", self.listen_ports[0]) + + # Load all existing torrents from seeding directory + self._load_existing() + + def stop(self): + """Stop the session gracefully.""" + if self.session: + self.session.pause() + logger.info("libtorrent session stopped (%d torrents)", len(self._handles)) + self.session = None + self._handles.clear() + self._torrent_files.clear() + + def _load_existing(self): + """Scan seeding directory and load all saved torrents.""" + count = 0 + for cid_dir in self.seeding_dir.iterdir(): + if not cid_dir.is_dir(): + continue + torrent_file = cid_dir / "torrent.dat" + data_dir = cid_dir / "data" + if torrent_file.exists() and data_dir.exists(): + try: + torrent_bytes = torrent_file.read_bytes() + self._add_to_session(torrent_bytes, data_dir) + count += 1 + except Exception as e: + logger.error("Failed to load torrent from %s: %s", cid_dir, e) + logger.info("Loaded %d existing torrents for seeding", count) + + def _add_to_session(self, torrent_bytes: bytes, data_dir: Path) -> Optional[str]: + """Add a torrent to the libtorrent session. Returns infohash.""" + if not self.session: + return None + + ti = lt.torrent_info(lt.bdecode(torrent_bytes)) + infohash = str(ti.info_hash()) + + if infohash in self._handles: + logger.debug("Torrent %s already loaded", infohash) + return infohash + + params = lt.add_torrent_params() + params.ti = ti + params.save_path = str(data_dir) + params.flags |= lt.torrent_flags.seed_mode # We generated the data, skip hash check + + handle = self.session.add_torrent(params) + self._handles[infohash] = handle + self._torrent_files[infohash] = torrent_bytes + logger.info("Seeding torrent %s (%s)", ti.name(), infohash) + return infohash + + def add_torrent(self, cid: str, torrent_bytes: bytes, content_dir: Path) -> Optional[str]: + """Add a new torrent for seeding. + + Copies content from content_dir to the seeding directory, + saves the .torrent file, and loads into the session. + + Handles file renaming for single-file torrents: libtorrent expects the + file at ``save_path / torrent_name``, but the source file from IPFS may + have a different name (e.g. the CID). We parse the torrent metadata to + determine expected file paths and rename accordingly. + + Args: + cid: IPFS CID (used as directory name) + torrent_bytes: The .torrent file bytes + content_dir: Path to the directory containing the files to seed + + Returns: + infohash string, or None on failure + """ + cid_dir = self.seeding_dir / cid + data_dir = cid_dir / "data" + torrent_file = cid_dir / "torrent.dat" + + try: + # If already seeding this CID, check whether the torrent changed + if torrent_file.exists() and data_dir.exists(): + old_bytes = torrent_file.read_bytes() + old_ti = lt.torrent_info(lt.bdecode(old_bytes)) + old_hash = str(old_ti.info_hash()) + if old_hash in self._handles: + self.session.remove_torrent(self._handles[old_hash]) + del self._handles[old_hash] + self._torrent_files.pop(old_hash, None) + shutil.rmtree(cid_dir, ignore_errors=True) + + # Set up seeding directory structure + cid_dir.mkdir(parents=True, exist_ok=True) + + # Copy content into data subdirectory + if content_dir.exists(): + shutil.copytree(content_dir, data_dir) + + # Parse torrent to figure out expected file layout + ti = lt.torrent_info(lt.bdecode(torrent_bytes)) + fs = ti.files() + + if fs.num_files() == 1: + # Single-file torrent: libtorrent expects the file at + # data_dir / ti.name(). The actual file from IPFS likely + # has a different name (the CID). + expected_name = ti.name() + expected_path = data_dir / expected_name + + if not expected_path.exists(): + # Find the actual file and rename it + actual_files = [f for f in data_dir.iterdir() if f.is_file()] + if len(actual_files) == 1: + actual_files[0].rename(expected_path) + logger.debug( + "Renamed %s -> %s for single-file torrent", + actual_files[0].name, expected_name, + ) + else: + logger.warning( + "Single-file torrent but found %d files in %s", + len(actual_files), data_dir, + ) + else: + # Multi-file torrent: libtorrent expects files at + # data_dir / ti.name() / . + # create_torrent built the torrent from the directory contents + # directly, so the files should be at data_dir/. + # But libtorrent expects them under data_dir//. + torrent_dir_name = ti.name() + nested_dir = data_dir / torrent_dir_name + + if not nested_dir.exists(): + # Move all files into a subdirectory named after the torrent + nested_dir.mkdir(parents=True, exist_ok=True) + for item in list(data_dir.iterdir()): + if item != nested_dir: + item.rename(nested_dir / item.name) + logger.debug( + "Moved content into %s/ for multi-file torrent", + torrent_dir_name, + ) + + # Save .torrent file + torrent_file.write_bytes(torrent_bytes) + + # Load into session + return self._add_to_session(torrent_bytes, data_dir) + + except Exception as e: + logger.error("Failed to add torrent for CID %s: %s", cid, e) + return None + + def get_torrent_file(self, infohash: str) -> Optional[bytes]: + """Get .torrent file bytes by infohash.""" + return self._torrent_files.get(infohash) + + def get_torrent_file_by_cid(self, cid: str) -> Optional[bytes]: + """Get .torrent file bytes by CID (looks on disk).""" + torrent_file = self.seeding_dir / cid / "torrent.dat" + if torrent_file.exists(): + return torrent_file.read_bytes() + return None + + def status(self) -> dict: + """Get seeder status.""" + if not self.session: + return {"running": False, "torrents": 0} + + stats = [] + for infohash, handle in self._handles.items(): + s = handle.status() + stats.append({ + "infohash": infohash, + "name": s.name, + "num_peers": s.num_peers, + "num_seeds": s.num_seeds, + "upload_rate": s.upload_rate, + "total_upload": s.total_upload, + "state": str(s.state), + }) + + return { + "running": True, + "torrents": len(self._handles), + "details": stats, + } + + +# Global seeder instance +_seeder: Optional[Seeder] = None + + +def get_seeder() -> Optional[Seeder]: + """Get the global seeder instance.""" + return _seeder + + +def init_seeder(seeding_dir: str) -> Seeder: + """Initialize and start the global seeder.""" + global _seeder + _seeder = Seeder(seeding_dir) + _seeder.start() + return _seeder + + +def stop_seeder(): + """Stop the global seeder.""" + global _seeder + if _seeder: + _seeder.stop() + _seeder = None diff --git a/delivery-kid/pinning-service/requirements.txt b/delivery-kid/pinning-service/requirements.txt index 1bad650..cae3164 100644 --- a/delivery-kid/pinning-service/requirements.txt +++ b/delivery-kid/pinning-service/requirements.txt @@ -17,3 +17,6 @@ pydantic-settings>=2.1.0 # Audio processing pydub>=0.25.1 + +# BitTorrent seeding +libtorrent>=2.0.0 From 39271bd01ba5b04605ffb12b4bcb631fd3601be3 Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Mon, 16 Mar 2026 02:34:47 +0000 Subject: [PATCH 6/8] Port Coconut.co cloud transcoding to delivery-kid Python service Restores the AV1+Opus HLS transcoding integration that was in the original Node.js pinning service but was lost in the Python rewrite. - POST /transcode-coconut: pins source to IPFS, submits to Coconut API - POST /webhook/coconut: receives completion, downloads HLS, pins to IPFS - GET /job/{id}: poll for job status (camelCase to match arthel frontend) - GET /jobs: list recent jobs - COCONUT_API_KEY added to config and docker-compose env - Local ffmpeg transcoding remains as fallback --- delivery-kid/ansible/playbook.yml | 1 + delivery-kid/pinning-service/app/config.py | 3 + delivery-kid/pinning-service/app/main.py | 3 +- .../pinning-service/app/routes/coconut.py | 206 ++++++++++++++++ .../pinning-service/app/services/coconut.py | 225 ++++++++++++++++++ 5 files changed, 437 insertions(+), 1 deletion(-) create mode 100644 delivery-kid/pinning-service/app/routes/coconut.py create mode 100644 delivery-kid/pinning-service/app/services/coconut.py diff --git a/delivery-kid/ansible/playbook.yml b/delivery-kid/ansible/playbook.yml index cc0c4c7..29d62aa 100644 --- a/delivery-kid/ansible/playbook.yml +++ b/delivery-kid/ansible/playbook.yml @@ -128,6 +128,7 @@ - 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: diff --git a/delivery-kid/pinning-service/app/config.py b/delivery-kid/pinning-service/app/config.py index 852e1c7..3336ce4 100644 --- a/delivery-kid/pinning-service/app/config.py +++ b/delivery-kid/pinning-service/app/config.py @@ -25,6 +25,9 @@ class Settings(BaseSettings): # 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) diff --git a/delivery-kid/pinning-service/app/main.py b/delivery-kid/pinning-service/app/main.py index 2529aeb..3fabbaa 100644 --- a/delivery-kid/pinning-service/app/main.py +++ b/delivery-kid/pinning-service/app/main.py @@ -9,7 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware from .config import get_settings -from .routes import health, albums, drafts, content, enrich, torrent +from .routes import health, albums, drafts, content, enrich, torrent, coconut from .services import cleanup from .services.seeder import init_seeder, stop_seeder @@ -92,6 +92,7 @@ async def lifespan(app: FastAPI): app.include_router(content.router) app.include_router(enrich.router) app.include_router(torrent.router) +app.include_router(coconut.router) @app.get("/") 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..2394566 --- /dev/null +++ b/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} diff --git a/delivery-kid/pinning-service/app/services/coconut.py b/delivery-kid/pinning-service/app/services/coconut.py new file mode 100644 index 0000000..1eb2cfe --- /dev/null +++ b/delivery-kid/pinning-service/app/services/coconut.py @@ -0,0 +1,225 @@ +"""Coconut.co cloud transcoding — AV1 HLS via external API. + +Submits videos to Coconut for AV1+Opus HLS transcoding (royalty-free codecs), +receives webhook on completion, downloads outputs, and pins to IPFS. +""" + +import json +import logging +import shutil +from pathlib import Path +from typing import Optional + +import httpx + +from . import ipfs + +logger = logging.getLogger(__name__) + +COCONUT_API_URL = "https://api.coconut.co/v2/jobs" + + +def _jobs_dir(staging_dir: Path) -> Path: + d = staging_dir / "jobs" + d.mkdir(parents=True, exist_ok=True) + return d + + +def save_job(staging_dir: Path, job_id: str, data: dict) -> None: + path = _jobs_dir(staging_dir) / f"{job_id}.json" + path.write_text(json.dumps(data, indent=2, default=str)) + + +def load_job(staging_dir: Path, job_id: str) -> Optional[dict]: + path = _jobs_dir(staging_dir) / f"{job_id}.json" + if not path.exists(): + return None + try: + return json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return None + + +def list_jobs(staging_dir: Path, limit: int = 50) -> list[dict]: + """List recent jobs, newest first.""" + jobs_path = _jobs_dir(staging_dir) + jobs = [] + for f in sorted(jobs_path.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True): + try: + jobs.append(json.loads(f.read_text())) + except (json.JSONDecodeError, OSError): + continue + if len(jobs) >= limit: + break + return jobs + + +async def submit_to_coconut( + source_url: str, + api_key: str, + webhook_url: str, + qualities: list[int] | None = None, +) -> dict: + """Submit a video to Coconut for AV1 HLS transcoding. + + Args: + source_url: Public URL of the source video (IPFS gateway URL) + api_key: Coconut API key + webhook_url: URL Coconut will POST to on completion + qualities: List of output heights, e.g. [720, 480] + + Returns: + Coconut API response dict + """ + if qualities is None: + qualities = [720, 480] + + # Build output config — AV1 video + Opus audio for each quality tier + outputs = {} + for q in qualities: + key = f"hls_av1_{q}p" + outputs[key] = { + "path": f"/output/{q}p/playlist.m3u8", + "video": { + "codec": "av1", + "height": q, + "bitrate": "4000k" if q >= 1080 else "2000k" if q >= 720 else "1000k", + }, + "audio": { + "codec": "opus", + "bitrate": "128k", + }, + "hls": { + "segment_duration": 6, + }, + } + + # Master playlist + outputs["hls_master"] = { + "path": "/output/master.m3u8", + "hls": { + "master": True, + "variants": [f"hls_av1_{q}p" for q in qualities], + }, + } + + job_config = { + "input": {"url": source_url}, + "outputs": outputs, + "webhook": webhook_url, + } + + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post( + COCONUT_API_URL, + json=job_config, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + ) + resp.raise_for_status() + return resp.json() + + +async def download_hls_outputs( + outputs: dict, + hls_dir: Path, +) -> None: + """Download HLS playlists and segments from Coconut output URLs. + + Args: + outputs: Dict of output key -> {url: ...} from Coconut webhook + hls_dir: Local directory to write files into + """ + hls_dir.mkdir(parents=True, exist_ok=True) + + async with httpx.AsyncClient(timeout=120.0) as client: + for key, output in outputs.items(): + url = output.get("url") + if not url: + continue + + if key == "hls_master": + local_path = hls_dir / "master.m3u8" + elif key.startswith("hls_av1_"): + quality = key.replace("hls_av1_", "").replace("p", "") + quality_dir = hls_dir / f"{quality}p" + quality_dir.mkdir(parents=True, exist_ok=True) + local_path = quality_dir / "playlist.m3u8" + else: + continue + + # Download playlist + logger.info("Downloading %s from %s", key, url) + resp = await client.get(url) + if not resp.is_success: + logger.warning("Failed to download %s: %s", key, resp.status_code) + continue + local_path.write_text(resp.text) + + # Download segments referenced in playlist + if local_path.name.endswith(".m3u8") and key != "hls_master": + await _download_segments(client, resp.text, url, local_path.parent) + + +async def _download_segments( + client: httpx.AsyncClient, + playlist_text: str, + playlist_url: str, + output_dir: Path, +) -> None: + """Download .ts/.m4s segments referenced in an HLS playlist.""" + from urllib.parse import urljoin + + for line in playlist_text.splitlines(): + line = line.strip() + if line.endswith(".ts") or line.endswith(".m4s"): + segment_url = urljoin(playlist_url, line) + segment_path = output_dir / line + try: + resp = await client.get(segment_url) + if resp.is_success: + segment_path.write_bytes(resp.content) + else: + logger.warning("Failed to download segment %s: %s", line, resp.status_code) + except Exception as e: + logger.warning("Error downloading segment %s: %s", line, e) + + +async def process_completed_job( + job: dict, + outputs: dict, + staging_dir: Path, + ipfs_api_url: str, + pinata_jwt: str = "", +) -> Optional[str]: + """Process a completed Coconut job: download HLS, pin to IPFS. + + Returns the HLS directory CID, or None on failure. + """ + job_id = job["id"] + hls_dir = staging_dir / f"hls-{job_id}" + + try: + # Download all HLS outputs + await download_hls_outputs(outputs, hls_dir) + + # Pin to IPFS + logger.info("[%s] Pinning HLS directory to IPFS...", job_id) + result = await ipfs.add_directory(hls_dir) + + if not result.success: + logger.error("[%s] IPFS pin failed: %s", job_id, result.error) + return None + + logger.info("[%s] HLS pinned: %s", job_id, result.cid) + return result.cid + + except Exception as e: + logger.error("[%s] Error processing completed job: %s", job_id, e) + return None + + finally: + # Clean up temp directory + shutil.rmtree(hls_dir, ignore_errors=True) From fd363b0cb3c2a273dc8bc91d2448253288159e57 Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Tue, 17 Mar 2026 13:35:06 +0000 Subject: [PATCH 7/8] Coconut-first transcoding, commit provenance in draft responses - Video finalization now tries Coconut.co first (AV1+Opus HLS), falls back to local ffmpeg if Coconut unavailable or fails - New transcoding_strategy field on ContentFinalizeRequest: auto (default), coconut, local, none - New transcoding-submitted SSE event for async Coconut path - All draft responses include commit hash (GIT_COMMIT env var) for staleness detection - config.get_commit() helper exposes build identity Depends on Dockerfile baking GIT_COMMIT at build time: docker build --build-arg GIT_COMMIT=$(git rev-parse --short HEAD) ... --- delivery-kid/pinning-service/app/config.py | 6 + .../pinning-service/app/models/content.py | 7 +- .../pinning-service/app/models/draft.py | 1 + .../pinning-service/app/routes/content.py | 140 ++++++++++++++++-- .../pinning-service/app/routes/drafts.py | 8 +- 5 files changed, 147 insertions(+), 15 deletions(-) diff --git a/delivery-kid/pinning-service/app/config.py b/delivery-kid/pinning-service/app/config.py index 3336ce4..9f49031 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 @@ -62,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/models/content.py b/delivery-kid/pinning-service/app/models/content.py index 5b9d4ec..bf0f8a6 100644 --- a/delivery-kid/pinning-service/app/models/content.py +++ b/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,9 @@ 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") 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/content.py b/delivery-kid/pinning-service/app/routes/content.py index 4481cf0..fa7efed 100644 --- a/delivery-kid/pinning-service/app/routes/content.py +++ b/delivery-kid/pinning-service/app/routes/content.py @@ -11,11 +11,12 @@ from sse_starlette.sse import EventSourceResponse from ..auth import require_auth -from ..config import get_settings, Settings +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 router = APIRouter(prefix="/draft-content", tags=["content"]) @@ -150,6 +151,7 @@ async def create_content_draft( draft_id=draft_id, expires_at=expires_at, files=draft_files, + commit=get_commit(), ) except HTTPException: @@ -186,6 +188,7 @@ async def get_content_draft( expires_at=state.expires_at, files=state.files, metadata=state.metadata, + commit=get_commit(), ) @@ -210,6 +213,30 @@ async def delete_content_draft( return {"message": "Draft deleted", "draft_id": draft_id} +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,7 +244,17 @@ 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. + + For video with transcoding enabled: + - Tries Coconut.co cloud transcoding first (AV1+Opus HLS, async via webhook) + - Falls back to local ffmpeg if Coconut is unavailable + - Coconut path: pins source to IPFS, submits job, returns job_id for polling + - Local path: synchronous transcode via SSE progress events + """ + import logging + import time + logger = logging.getLogger(__name__) async def send_event(event: str, data: dict): return {"event": event, "data": json.dumps(data)} @@ -233,15 +270,98 @@ async def send_event(event: str, data: dict): "progress": 5 }) - # Determine what we're working with 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) + + if wants_transcode and _should_use_coconut(request, settings): + # === Coconut cloud transcoding path (async) === + video_file = video_files[0] + src_path = upload_dir / video_file.original_filename + + yield await send_event("progress", { + "stage": "ipfs", + "message": "Pinning source video to IPFS...", + "progress": 10 + }) + + # Pin source to IPFS so Coconut can fetch it via gateway + pin_result = await ipfs.add_file(src_path) + if not pin_result.success: + yield await send_event("error", { + "message": f"Failed to pin source video: {pin_result.error}" + }) + return + + source_cid = pin_result.cid + source_url = f"{settings.ipfs_gateway_url}/ipfs/{source_cid}" + logger.info("[content:%s] Source pinned: %s", draft_id[:8], source_cid) + + yield await send_event("progress", { + "stage": "transcode", + "message": "Submitting to Coconut for AV1 transcoding...", + "progress": 30 + }) + + # Build webhook URL + base_url = settings.ipfs_gateway_url.replace("ipfs.", "", 1) + 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=[720, 480], + ) + coconut_job_id = coconut_result.get("id") + logger.info("[content:%s] Coconut job created: %s", draft_id[:8], coconut_job_id) + + # Save job state for webhook handler + job_state = { + "id": job_id, + "coconutJobId": coconut_job_id, + "status": "processing", + "sourceCid": source_cid, + "keepOriginal": False, + "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) + + # Clean up draft dir — source is on IPFS now + shutil.rmtree(draft_dir, ignore_errors=True) + + yield await send_event("transcoding-submitted", { + "sourceCid": source_cid, + "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}", + "gatewayUrl": f"{settings.ipfs_gateway_url}/ipfs/{source_cid}", + "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 + }) + # Fall through to local transcoding below - # 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 + if wants_transcode: + # === Local ffmpeg transcoding path (sync) === video_file = video_files[0] src_path = upload_dir / video_file.original_filename @@ -293,7 +413,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 +424,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: diff --git a/delivery-kid/pinning-service/app/routes/drafts.py b/delivery-kid/pinning-service/app/routes/drafts.py index 99b70bf..de6d6de 100644 --- a/delivery-kid/pinning-service/app/routes/drafts.py +++ b/delivery-kid/pinning-service/app/routes/drafts.py @@ -11,7 +11,7 @@ from sse_starlette.sse import EventSourceResponse from ..auth import require_auth -from ..config import get_settings, Settings +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(), ) From 094bdd524e539f2837bd86890fa4c84c4abf621c Mon Sep 17 00:00:00 2001 From: Justin Holmes Date: Tue, 17 Mar 2026 16:28:37 +0000 Subject: [PATCH 8/8] Add transcoding_qualities option, tests for coconut + config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add transcoding_qualities field to ContentFinalizeRequest — lets callers specify output heights (e.g. [1080, 720, 480]). Defaults to [720, 480] when not provided. Threaded through to Coconut API. - Document available quality values and bitrate tiers in coconut.py - Add test suite: 13 tests covering job config building (default and custom qualities, bitrate tiers, AV1+Opus codec enforcement), job persistence (save/load/list), model defaults --- .../pinning-service/app/models/content.py | 5 + .../pinning-service/app/routes/content.py | 2 +- .../pinning-service/app/services/coconut.py | 6 +- .../pinning-service/tests/__init__.py | 0 .../pinning-service/tests/test_coconut.py | 183 ++++++++++++++++++ .../pinning-service/tests/test_config.py | 20 ++ 6 files changed, 214 insertions(+), 2 deletions(-) create mode 100644 delivery-kid/pinning-service/tests/__init__.py create mode 100644 delivery-kid/pinning-service/tests/test_coconut.py create mode 100644 delivery-kid/pinning-service/tests/test_config.py diff --git a/delivery-kid/pinning-service/app/models/content.py b/delivery-kid/pinning-service/app/models/content.py index bf0f8a6..a8fa913 100644 --- a/delivery-kid/pinning-service/app/models/content.py +++ b/delivery-kid/pinning-service/app/models/content.py @@ -58,3 +58,8 @@ class ContentFinalizeRequest(BaseModel): 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." + ) diff --git a/delivery-kid/pinning-service/app/routes/content.py b/delivery-kid/pinning-service/app/routes/content.py index fa7efed..673abdf 100644 --- a/delivery-kid/pinning-service/app/routes/content.py +++ b/delivery-kid/pinning-service/app/routes/content.py @@ -312,7 +312,7 @@ async def send_event(event: str, data: dict): source_url=source_url, api_key=settings.coconut_api_key, webhook_url=webhook_url, - qualities=[720, 480], + qualities=request.transcoding_qualities, ) coconut_job_id = coconut_result.get("id") logger.info("[content:%s] Coconut job created: %s", draft_id[:8], coconut_job_id) diff --git a/delivery-kid/pinning-service/app/services/coconut.py b/delivery-kid/pinning-service/app/services/coconut.py index 1eb2cfe..05fd36a 100644 --- a/delivery-kid/pinning-service/app/services/coconut.py +++ b/delivery-kid/pinning-service/app/services/coconut.py @@ -66,7 +66,11 @@ async def submit_to_coconut( source_url: Public URL of the source video (IPFS gateway URL) api_key: Coconut API key webhook_url: URL Coconut will POST to on completion - qualities: List of output heights, e.g. [720, 480] + qualities: List of output heights for HLS variants. Each gets its + own AV1+Opus stream. Common values: 2160 (4K), 1080, 720, 480, 360. + Default [720, 480]. Higher values increase Coconut processing time + and cost. Can be passed from the UI via ContentFinalizeRequest's + transcoding_qualities field. Returns: Coconut API response dict diff --git a/delivery-kid/pinning-service/tests/__init__.py b/delivery-kid/pinning-service/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/delivery-kid/pinning-service/tests/test_coconut.py b/delivery-kid/pinning-service/tests/test_coconut.py new file mode 100644 index 0000000..74f918a --- /dev/null +++ b/delivery-kid/pinning-service/tests/test_coconut.py @@ -0,0 +1,183 @@ +"""Tests for app.services.coconut — job config building and quality tiers.""" + +import json +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from app.services.coconut import submit_to_coconut, save_job, load_job, list_jobs + + +class TestJobConfigBuilding: + """Test that submit_to_coconut builds correct Coconut API payloads.""" + + @pytest.mark.asyncio + async def test_default_qualities(self): + """Default qualities should be 720p and 480p.""" + mock_response = AsyncMock() + mock_response.json.return_value = {"id": "job-123", "status": "processing"} + mock_response.raise_for_status = lambda: None + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await submit_to_coconut( + source_url="https://example.com/video.mp4", + api_key="test-key", + webhook_url="https://example.com/webhook", + ) + + call_args = mock_client.post.call_args + job_config = call_args.kwargs.get("json") or call_args[1].get("json") + + # Should have 720p and 480p outputs plus master + assert "hls_av1_720p" in job_config["outputs"] + assert "hls_av1_480p" in job_config["outputs"] + assert "hls_master" in job_config["outputs"] + assert len(job_config["outputs"]) == 3 + + @pytest.mark.asyncio + async def test_custom_qualities(self): + """Custom qualities should produce matching output keys.""" + mock_response = AsyncMock() + mock_response.json.return_value = {"id": "job-456", "status": "processing"} + mock_response.raise_for_status = lambda: None + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await submit_to_coconut( + source_url="https://example.com/video.mp4", + api_key="test-key", + webhook_url="https://example.com/webhook", + qualities=[1080, 720, 480, 360], + ) + + call_args = mock_client.post.call_args + job_config = call_args.kwargs.get("json") or call_args[1].get("json") + + assert "hls_av1_1080p" in job_config["outputs"] + assert "hls_av1_720p" in job_config["outputs"] + assert "hls_av1_480p" in job_config["outputs"] + assert "hls_av1_360p" in job_config["outputs"] + assert "hls_master" in job_config["outputs"] + + # Master should list all variants + master = job_config["outputs"]["hls_master"] + assert set(master["hls"]["variants"]) == { + "hls_av1_1080p", "hls_av1_720p", "hls_av1_480p", "hls_av1_360p" + } + + @pytest.mark.asyncio + async def test_bitrate_tiers(self): + """Higher resolutions should get higher bitrates.""" + mock_response = AsyncMock() + mock_response.json.return_value = {"id": "job-789", "status": "processing"} + mock_response.raise_for_status = lambda: None + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await submit_to_coconut( + source_url="https://example.com/video.mp4", + api_key="test-key", + webhook_url="https://example.com/webhook", + qualities=[1080, 720, 480], + ) + + call_args = mock_client.post.call_args + job_config = call_args.kwargs.get("json") or call_args[1].get("json") + + assert job_config["outputs"]["hls_av1_1080p"]["video"]["bitrate"] == "4000k" + assert job_config["outputs"]["hls_av1_720p"]["video"]["bitrate"] == "2000k" + assert job_config["outputs"]["hls_av1_480p"]["video"]["bitrate"] == "1000k" + + @pytest.mark.asyncio + async def test_all_outputs_use_av1_opus(self): + """All quality tiers should use AV1 video and Opus audio.""" + mock_response = AsyncMock() + mock_response.json.return_value = {"id": "job-abc", "status": "processing"} + mock_response.raise_for_status = lambda: None + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await submit_to_coconut( + source_url="https://example.com/video.mp4", + api_key="test-key", + webhook_url="https://example.com/webhook", + qualities=[720, 480], + ) + + call_args = mock_client.post.call_args + job_config = call_args.kwargs.get("json") or call_args[1].get("json") + + for key, output in job_config["outputs"].items(): + if key == "hls_master": + continue + assert output["video"]["codec"] == "av1", f"{key} should use av1" + assert output["audio"]["codec"] == "opus", f"{key} should use opus" + + +class TestJobPersistence: + """Test job save/load/list operations.""" + + def test_save_and_load(self, tmp_path): + job_data = {"id": "test-job", "status": "processing", "source_cid": "bafytest"} + save_job(tmp_path, "test-job", job_data) + + loaded = load_job(tmp_path, "test-job") + assert loaded == job_data + + def test_load_nonexistent(self, tmp_path): + assert load_job(tmp_path, "nonexistent") is None + + def test_list_jobs(self, tmp_path): + for i in range(3): + save_job(tmp_path, f"job-{i}", {"id": f"job-{i}", "index": i}) + + jobs = list_jobs(tmp_path) + assert len(jobs) == 3 + + def test_list_jobs_limit(self, tmp_path): + for i in range(5): + save_job(tmp_path, f"job-{i}", {"id": f"job-{i}"}) + + jobs = list_jobs(tmp_path, limit=2) + assert len(jobs) == 2 + + +class TestContentFinalizeRequest: + """Test the transcoding_qualities field on ContentFinalizeRequest.""" + + def test_default_qualities_is_none(self): + from app.models.content import ContentFinalizeRequest + req = ContentFinalizeRequest() + assert req.transcoding_qualities is None + + def test_custom_qualities(self): + from app.models.content import ContentFinalizeRequest + req = ContentFinalizeRequest(transcoding_qualities=[1080, 720]) + assert req.transcoding_qualities == [1080, 720] + + def test_default_strategy_is_auto(self): + from app.models.content import ContentFinalizeRequest + req = ContentFinalizeRequest() + assert req.transcoding_strategy == "auto" diff --git a/delivery-kid/pinning-service/tests/test_config.py b/delivery-kid/pinning-service/tests/test_config.py new file mode 100644 index 0000000..bb6669b --- /dev/null +++ b/delivery-kid/pinning-service/tests/test_config.py @@ -0,0 +1,20 @@ +"""Tests for app.config — commit provenance and settings.""" + +import os +from unittest.mock import patch + +from app.config import get_commit + + +class TestGetCommit: + def test_returns_env_var_when_set(self): + with patch.dict(os.environ, {"GIT_COMMIT": "abc123f"}): + assert get_commit() == "abc123f" + + def test_returns_unknown_when_unset(self): + with patch.dict(os.environ, {}, clear=True): + # GIT_COMMIT might be set in the real env, so explicitly remove it + env = os.environ.copy() + env.pop("GIT_COMMIT", None) + with patch.dict(os.environ, env, clear=True): + assert get_commit() == "unknown"