initial commit

src/fileserver.py

@@ -0,0 +1,174 @@
+"""
+File Server API for Meta-Stremio
+
+Serves files by CID by looking up the path from metadata.
+This is a generic file serving endpoint that can serve any file
+whose CID and path are stored in the metadata.
+
+Endpoints:
+- GET /file/{cid} - Serve file by CID
+- GET /file/{cid}/w{width} - Serve image resized to width (images only)
+"""
+from __future__ import annotations
+
+import os
+import io
+from typing import Optional, Tuple
+
+# Optional PIL for image resizing
+try:
+    from PIL import Image
+    PIL_AVAILABLE = True
+except ImportError:
+    PIL_AVAILABLE = False
+    Image = None
+
+# Configuration
+FILES_PATH = os.environ.get('FILES_PATH', '/files')
+
+# Storage reference (set by init)
+_storage = None
+
+
+def init(storage) -> None:
+    """Initialize with storage reference."""
+    global _storage
+    _storage = storage
+
+
+def lookup_path_by_cid(cid: str) -> Optional[str]:
+    """
+    Look up a file path by its CID from Redis.
+
+    Files (including poster images) are stored in Redis as file:{cid} entries.
+    The file path is stored in the 'path' field.
+
+    Returns the relative path if found, None otherwise.
+    """
+    if not _storage:
+        return None
+
+    try:
+        # Look up directly by CID - poster files are processed like any other file
+        file_path = _storage.get_file_path_by_cid(cid)
+        if file_path:
+            return file_path
+        return None
+    except Exception as e:
+        print(f"[FileServer] Error looking up CID {cid}: {e}")
+        return None
+
+
+def get_file_path(cid: str) -> Optional[str]:
+    """Get the absolute file path for a CID."""
+    rel_path = lookup_path_by_cid(cid)
+    if rel_path:
+        return os.path.join(FILES_PATH, rel_path)
+    return None
+
+
+def resize_image(image_data: bytes, width: int) -> bytes:
+    """Resize an image to the specified width, maintaining aspect ratio."""
+    if not PIL_AVAILABLE:
+        raise RuntimeError("PIL not available for image resizing")
+
+    img = Image.open(io.BytesIO(image_data))
+
+    # Calculate new height maintaining aspect ratio
+    ratio = width / img.width
+    height = int(img.height * ratio)
+
+    # Use high-quality resampling
+    resized = img.resize((width, height), Image.Resampling.LANCZOS)
+
+    # Save to bytes
+    output = io.BytesIO()
+    # Preserve format, default to JPEG
+    fmt = img.format or 'JPEG'
+    if fmt.upper() == 'PNG' and img.mode == 'RGBA':
+        resized.save(output, format='PNG', optimize=True)
+    else:
+        # Convert to RGB for JPEG
+        if resized.mode in ('RGBA', 'P'):
+            resized = resized.convert('RGB')
+        resized.save(output, format='JPEG', quality=85, optimize=True)
+
+    return output.getvalue()
+
+
+def serve_file(cid: str, width: Optional[int] = None) -> Tuple[Optional[bytes], str, int]:
+    """
+    Serve a file by CID.
+
+    Args:
+        cid: The CID of the file
+        width: Optional width for resizing (images only)
+
+    Returns:
+        Tuple of (file_data, content_type, status_code)
+    """
+    # Get file path from CID
+    file_path = get_file_path(cid)
+
+    if not file_path:
+        return None, 'text/plain', 404
+
+    if not os.path.exists(file_path):
+        print(f"[FileServer] File not found: {file_path}")
+        return None, 'text/plain', 404
+
+    # Determine content type from extension
+    ext = os.path.splitext(file_path)[1].lower()
+    content_types = {
+        '.jpg': 'image/jpeg',
+        '.jpeg': 'image/jpeg',
+        '.png': 'image/png',
+        '.webp': 'image/webp',
+        '.gif': 'image/gif',
+        '.mp4': 'video/mp4',
+        '.mkv': 'video/x-matroska',
+        '.avi': 'video/x-msvideo',
+        '.webm': 'video/webm',
+    }
+    content_type = content_types.get(ext, 'application/octet-stream')
+
+    try:
+        with open(file_path, 'rb') as f:
+            file_data = f.read()
+
+        # Resize if width is specified and it's an image
+        if width and PIL_AVAILABLE and content_type.startswith('image/'):
+            try:
+                file_data = resize_image(file_data, width)
+                # After resize, it's always JPEG (unless PNG with alpha)
+                if ext != '.png':
+                    content_type = 'image/jpeg'
+            except Exception as e:
+                print(f"[FileServer] Resize error: {e}")
+                # Fall through to serve original
+
+        return file_data, content_type, 200
+
+    except Exception as e:
+        print(f"[FileServer] Error reading {file_path}: {e}")
+        return None, 'text/plain', 500
+
+
+def get_file_url(cid: str, base_url: str, width: Optional[int] = None) -> str:
+    """
+    Generate a file URL for a CID.
+
+    Args:
+        cid: The CID of the file
+        base_url: Base URL of the server
+        width: Optional width for resizing (images only)
+
+    Returns:
+        Full URL to the file
+    """
+    if not cid:
+        return ""
+
+    if width:
+        return f"{base_url}/file/{cid}/w{width}"
+    return f"{base_url}/file/{cid}"

src/plugin_subscriber.py

@@ -0,0 +1,265 @@
+"""
+Plugin Completion Subscriber for meta-stremio.
+
+Subscribes to Redis pub/sub channel 'meta-sort:plugin:complete' to be notified
+when specific plugins finish processing files. This allows Stremio to:
+1. Log metadata updates for debugging
+2. Invalidate any local caches (if implemented)
+3. Track which files have complete metadata
+
+Plugins of interest to Stremio:
+- filename-parser: Provides season/episode info for TV vs Movie categorization
+- tmdb: Provides rich metadata (title, year, poster, description, etc.)
+"""
+from __future__ import annotations
+
+import os
+import json
+import threading
+from typing import Callable, Optional, List, Set
+
+try:
+    import redis
+    REDIS_AVAILABLE = True
+except ImportError:
+    REDIS_AVAILABLE = False
+    redis = None
+
+
+# Redis configuration
+REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')
+
+# Plugins that Stremio cares about
+WATCHED_PLUGINS: Set[str] = {'filename-parser', 'tmdb'}
+
+# Channel for plugin completion events
+PLUGIN_COMPLETE_CHANNEL = 'meta-sort:plugin:complete'
+
+
+class PluginSubscriber:
+    """
+    Subscribes to plugin completion events from meta-sort.
+
+    When filename-parser or tmdb plugins complete, the subscriber
+    receives notifications that can be used to update Stremio's view
+    of the metadata.
+    """
+
+    def __init__(
+        self,
+        url: str = None,
+        watched_plugins: Set[str] = None,
+        on_plugin_complete: Callable[[str, str, str], None] = None
+    ):
+        """
+        Initialize the plugin subscriber.
+
+        Args:
+            url: Redis connection URL
+            watched_plugins: Set of plugin IDs to watch (default: filename-parser, tmdb)
+            on_plugin_complete: Callback when a watched plugin completes
+                                Signature: (file_hash, plugin_id, file_path) -> None
+        """
+        if not REDIS_AVAILABLE:
+            raise ImportError("redis package not installed. Run: pip install redis")
+
+        self._url = url or REDIS_URL
+        self._watched_plugins = watched_plugins or WATCHED_PLUGINS
+        self._on_plugin_complete = on_plugin_complete
+        self._pubsub: Optional[redis.client.PubSub] = None
+        self._thread: Optional[threading.Thread] = None
+        self._running = False
+        self._client: Optional[redis.Redis] = None
+
+        # Track files with complete metadata (all watched plugins finished)
+        self._file_plugin_status: dict[str, Set[str]] = {}
+
+    def start(self) -> bool:
+        """Start the subscriber in a background thread."""
+        if self._running:
+            return True
+
+        try:
+            self._client = redis.from_url(self._url, decode_responses=True)
+            self._client.ping()  # Test connection
+
+            self._pubsub = self._client.pubsub()
+            self._pubsub.subscribe(PLUGIN_COMPLETE_CHANNEL)
+
+            self._running = True
+            self._thread = threading.Thread(
+                target=self._listen_loop,
+                daemon=True,
+                name='PluginSubscriber'
+            )
+            self._thread.start()
+
+            print(f"[PluginSubscriber] Started, watching plugins: {', '.join(self._watched_plugins)}")
+            return True
+
+        except Exception as e:
+            print(f"[PluginSubscriber] Failed to start: {e}")
+            self._running = False
+            return False
+
+    def stop(self) -> None:
+        """Stop the subscriber."""
+        self._running = False
+
+        if self._pubsub:
+            try:
+                self._pubsub.unsubscribe()
+                self._pubsub.close()
+            except Exception:
+                pass
+            self._pubsub = None
+
+        if self._client:
+            try:
+                self._client.close()
+            except Exception:
+                pass
+            self._client = None
+
+        print("[PluginSubscriber] Stopped")
+
+    def _listen_loop(self) -> None:
+        """Background loop that listens for plugin completion events."""
+        while self._running and self._pubsub:
+            try:
+                message = self._pubsub.get_message(timeout=1.0)
+                if message and message['type'] == 'message':
+                    self._handle_message(message['data'])
+            except redis.ConnectionError:
+                print("[PluginSubscriber] Connection lost, attempting reconnect...")
+                self._reconnect()
+            except Exception as e:
+                if self._running:
+                    print(f"[PluginSubscriber] Error in listen loop: {e}")
+
+    def _reconnect(self) -> None:
+        """Attempt to reconnect to Redis."""
+        import time
+
+        for attempt in range(5):
+            if not self._running:
+                return
+
+            try:
+                time.sleep(2 ** attempt)  # Exponential backoff
+
+                if self._pubsub:
+                    self._pubsub.close()
+                if self._client:
+                    self._client.close()
+
+                self._client = redis.from_url(self._url, decode_responses=True)
+                self._client.ping()
+
+                self._pubsub = self._client.pubsub()
+                self._pubsub.subscribe(PLUGIN_COMPLETE_CHANNEL)
+
+                print("[PluginSubscriber] Reconnected successfully")
+                return
+
+            except Exception as e:
+                print(f"[PluginSubscriber] Reconnect attempt {attempt + 1} failed: {e}")
+
+        print("[PluginSubscriber] Failed to reconnect after 5 attempts")
+        self._running = False
+
+    def _handle_message(self, data: str) -> None:
+        """Handle a plugin completion message."""
+        try:
+            payload = json.loads(data)
+            file_hash = payload.get('fileHash', '')
+            plugin_id = payload.get('pluginId', '')
+            file_path = payload.get('filePath', '')
+
+            # Only process plugins we're watching
+            if plugin_id not in self._watched_plugins:
+                return
+
+            # Track plugin completion for this file
+            if file_hash not in self._file_plugin_status:
+                self._file_plugin_status[file_hash] = set()
+            self._file_plugin_status[file_hash].add(plugin_id)
+
+            # Log the completion
+            print(f"[PluginSubscriber] Plugin '{plugin_id}' completed for {file_hash[:12]}...")
+
+            # Check if all watched plugins have completed for this file
+            completed_plugins = self._file_plugin_status[file_hash]
+            if self._watched_plugins.issubset(completed_plugins):
+                print(f"[PluginSubscriber] All watched plugins complete for {file_hash[:12]}...")
+                # Could trigger a cache refresh here if we had local caching
+
+            # Call the callback if provided
+            if self._on_plugin_complete:
+                try:
+                    self._on_plugin_complete(file_hash, plugin_id, file_path)
+                except Exception as e:
+                    print(f"[PluginSubscriber] Callback error: {e}")
+
+        except json.JSONDecodeError:
+            print(f"[PluginSubscriber] Invalid JSON message: {data[:100]}")
+        except Exception as e:
+            print(f"[PluginSubscriber] Error handling message: {e}")
+
+    def is_running(self) -> bool:
+        """Check if the subscriber is running."""
+        return self._running
+
+    def get_status(self) -> dict:
+        """Get subscriber status for debugging."""
+        return {
+            'running': self._running,
+            'watched_plugins': list(self._watched_plugins),
+            'tracked_files': len(self._file_plugin_status),
+        }
+
+
+# Global subscriber instance
+_subscriber: Optional[PluginSubscriber] = None
+
+
+def init_subscriber(
+    on_plugin_complete: Callable[[str, str, str], None] = None
+) -> Optional[PluginSubscriber]:
+    """
+    Initialize and start the global plugin subscriber.
+
+    Args:
+        on_plugin_complete: Optional callback when watched plugins complete
+
+    Returns:
+        The subscriber instance, or None if Redis is unavailable
+    """
+    global _subscriber
+
+    if not REDIS_AVAILABLE:
+        print("[PluginSubscriber] Redis not available, skipping subscriber init")
+        return None
+
+    if _subscriber and _subscriber.is_running():
+        return _subscriber
+
+    _subscriber = PluginSubscriber(on_plugin_complete=on_plugin_complete)
+    if _subscriber.start():
+        return _subscriber
+
+    _subscriber = None
+    return None
+
+
+def get_subscriber() -> Optional[PluginSubscriber]:
+    """Get the global subscriber instance."""
+    return _subscriber
+
+
+def stop_subscriber() -> None:
+    """Stop the global subscriber."""
+    global _subscriber
+    if _subscriber:
+        _subscriber.stop()
+        _subscriber = None

src/poster.py

@@ -0,0 +1,178 @@
+"""
+Poster API for Meta-Stremio
+
+Serves poster and backdrop images by CID with optional resizing.
+Builds a CID -> file path index from video metadata.
+
+Endpoints:
+- GET /poster/{cid} - Serve original image
+- GET /poster/{cid}/w{width} - Serve resized image (width specified)
+"""
+from __future__ import annotations
+
+import os
+import io
+from typing import Optional, Dict, Tuple
+from threading import Lock
+
+# Optional PIL for image resizing
+try:
+    from PIL import Image
+    PIL_AVAILABLE = True
+except ImportError:
+    PIL_AVAILABLE = False
+    Image = None
+
+# Configuration
+FILES_PATH = os.environ.get('FILES_PATH', '/files')
+
+# CID -> file path index
+_cid_index: Dict[str, str] = {}
+_index_lock = Lock()
+_index_built = False
+
+
+def build_cid_index(storage) -> None:
+    """Build CID to file path index from storage metadata."""
+    global _cid_index, _index_built
+
+    with _index_lock:
+        _cid_index.clear()
+
+        try:
+            videos = storage.get_all_videos()
+            for video in videos:
+                # Index poster
+                if video.poster and video.poster_path:
+                    _cid_index[video.poster] = video.poster_path
+                # Index backdrop
+                if video.backdrop and video.backdrop_path:
+                    _cid_index[video.backdrop] = video.backdrop_path
+
+            _index_built = True
+            print(f"[Poster] Built CID index: {len(_cid_index)} images")
+        except Exception as e:
+            print(f"[Poster] Error building CID index: {e}")
+
+
+def get_image_path(cid: str) -> Optional[str]:
+    """Get the file path for a CID."""
+    with _index_lock:
+        rel_path = _cid_index.get(cid)
+        if rel_path:
+            return os.path.join(FILES_PATH, rel_path)
+    return None
+
+
+def is_index_built() -> bool:
+    """Check if the CID index has been built."""
+    return _index_built
+
+
+def get_index_size() -> int:
+    """Get the number of images in the index."""
+    with _index_lock:
+        return len(_cid_index)
+
+
+def resize_image(image_data: bytes, width: int) -> bytes:
+    """Resize an image to the specified width, maintaining aspect ratio."""
+    if not PIL_AVAILABLE:
+        raise RuntimeError("PIL not available for image resizing")
+
+    img = Image.open(io.BytesIO(image_data))
+
+    # Calculate new height maintaining aspect ratio
+    ratio = width / img.width
+    height = int(img.height * ratio)
+
+    # Use high-quality resampling
+    resized = img.resize((width, height), Image.Resampling.LANCZOS)
+
+    # Save to bytes
+    output = io.BytesIO()
+    # Preserve format, default to JPEG
+    fmt = img.format or 'JPEG'
+    if fmt.upper() == 'PNG' and img.mode == 'RGBA':
+        resized.save(output, format='PNG', optimize=True)
+    else:
+        # Convert to RGB for JPEG
+        if resized.mode in ('RGBA', 'P'):
+            resized = resized.convert('RGB')
+        resized.save(output, format='JPEG', quality=85, optimize=True)
+
+    return output.getvalue()
+
+
+def serve_poster(cid: str, width: Optional[int] = None) -> Tuple[Optional[bytes], str, int]:
+    """
+    Serve a poster image by CID.
+
+    Args:
+        cid: The CID of the image
+        width: Optional width for resizing
+
+    Returns:
+        Tuple of (image_data, content_type, status_code)
+    """
+    # Get file path from CID
+    file_path = get_image_path(cid)
+
+    if not file_path:
+        return None, 'text/plain', 404
+
+    if not os.path.exists(file_path):
+        print(f"[Poster] File not found: {file_path}")
+        return None, 'text/plain', 404
+
+    # Determine content type from extension
+    ext = os.path.splitext(file_path)[1].lower()
+    content_types = {
+        '.jpg': 'image/jpeg',
+        '.jpeg': 'image/jpeg',
+        '.png': 'image/png',
+        '.webp': 'image/webp',
+        '.gif': 'image/gif',
+    }
+    content_type = content_types.get(ext, 'image/jpeg')
+
+    try:
+        with open(file_path, 'rb') as f:
+            image_data = f.read()
+
+        # Resize if width is specified
+        if width and PIL_AVAILABLE:
+            try:
+                image_data = resize_image(image_data, width)
+                # After resize, it's always JPEG (unless PNG with alpha)
+                if not (ext == '.png'):
+                    content_type = 'image/jpeg'
+            except Exception as e:
+                print(f"[Poster] Resize error: {e}")
+                # Fall through to serve original
+
+        return image_data, content_type, 200
+
+    except Exception as e:
+        print(f"[Poster] Error reading {file_path}: {e}")
+        return None, 'text/plain', 500
+
+
+def get_poster_url(cid: str, base_url: str, width: Optional[int] = None) -> str:
+    """
+    Generate a poster URL for a CID.
+
+    Args:
+        cid: The CID of the image
+        base_url: Base URL of the server
+        width: Optional width for resizing
+
+    Returns:
+        Full URL to the poster image
+    """
+    if not cid:
+        return ""
+
+    if width:
+        return f"{base_url}/poster/{cid}/w{width}"
+    return f"{base_url}/poster/{cid}"

src/server.py

@@ -0,0 +1,955 @@
+#!/usr/bin/env python3
+"""
+Meta-Stremio Server
+
+A standalone Stremio addon that:
+1. Reads video metadata from KV storage (written by meta-sort)
+2. Transcodes video on-the-fly using FFmpeg with adaptive quality
+3. Serves HLS streams with muxed video+audio segments
+4. Provides a dashboard for monitoring and setup
+5. Registers with service discovery for inter-service navigation
+
+Environment Variables:
+- MEDIA_DIR: Directory containing video files (default: /data/media)
+- CACHE_DIR: Directory for transcoded segments (default: /data/cache)
+- PORT: HTTP server port (default: 7000)
+- STORAGE_MODE: Storage backend 'leader' or 'redis' (default: leader)
+- META_CORE_PATH: Path to meta-core shared volume (default: /meta-core)
+- FILES_PATH: Path to files shared volume (default: /files)
+- REDIS_URL: Redis connection URL (default: redis://localhost:6379)
+- REDIS_PREFIX: Redis key prefix (default: meta-sort:)
+- SCHEME: URL scheme for generated URLs (http/https/auto)
+- SEGMENT_DURATION: HLS segment length in seconds (default: 4)
+- PREFETCH_SEGMENTS: How many segments to prefetch ahead (default: 4)
+"""
+from __future__ import annotations
+
+import os
+import re
+import json
+import signal
+import sys
+import base64
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from urllib.parse import urlparse, unquote
+import threading
+
+import stremio
+import transcoder
+from storage import init_service_discovery, get_service_discovery
+import fileserver
+import plugin_subscriber
+
+# Configuration
+PORT = int(os.environ.get('PORT', '7000'))
+SCHEME = os.environ.get('SCHEME', '').lower().strip()
+META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
+BASE_URL = os.environ.get('BASE_URL', '')
+
+# Initialize storage
+storage = stremio.init_storage()
+
+# Initialize service discovery (only if META_CORE_PATH exists)
+service_discovery = None
+if os.path.exists(META_CORE_PATH):
+    try:
+        service_discovery = init_service_discovery(base_url=BASE_URL if BASE_URL else None)
+    except Exception as e:
+        print(f"[Server] Service discovery init failed: {e}")
+
+# Global Stremio handler
+stremio_handler = stremio.StremioHandler()
+
+
+def parse_config_from_path(path: str) -> tuple[str, dict]:
+    """
+    Parse Stremio config from URL path.
+
+    Stremio passes config as URL-safe base64-encoded JSON in the path:
+    /{config}/manifest.json -> config dict
+    /manifest.json -> empty dict
+
+    Returns:
+        (remaining_path, config_dict)
+    """
+    # Check if path starts with a potential config segment
+    # Config paths look like: /eyJsYW5n.../manifest.json
+    parts = path.strip('/').split('/', 1)
+
+    if len(parts) >= 1 and parts[0]:
+        potential_config = parts[0]
+
+        # Skip known non-config paths
+        if potential_config in ('manifest.json', 'catalog', 'meta', 'stream', 'stremio',
+                                 'health', 'api', 'transcode', 'direct', 'poster', 'configure'):
+            return path, {}
+
+        # Try to decode as base64 config
+        try:
+            # Add padding if needed
+            padded = potential_config + '=' * (4 - len(potential_config) % 4)
+            decoded = base64.urlsafe_b64decode(padded).decode('utf-8')
+            config = json.loads(decoded)
+
+            # Return remaining path with leading slash
+            remaining = '/' + parts[1] if len(parts) > 1 else '/'
+            return remaining, config
+        except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
+            # Not a valid config, return original path
+            pass
+
+    return path, {}
+
+
+def encode_config(config: dict) -> str:
+    """Encode config dict to URL-safe base64 string."""
+    json_str = json.dumps(config, separators=(',', ':'))
+    encoded = base64.urlsafe_b64encode(json_str.encode('utf-8')).decode('utf-8')
+    return encoded.rstrip('=')
+
+
+class Handler(BaseHTTPRequestHandler):
+    def log_message(self, fmt, *args):
+        print(f"[{self.address_string()}] {fmt % args}")
+
+    def send_json(self, data: dict, status: int = 200):
+        body = json.dumps(data).encode()
+        self.send_response(status)
+        self.send_header('Content-Type', 'application/json')
+        self.send_header('Access-Control-Allow-Origin', '*')
+        self.end_headers()
+        self.wfile.write(body)
+
+    def send_data(self, data: bytes, content_type: str):
+        self.send_response(200)
+        self.send_header('Content-Type', content_type)
+        self.send_header('Content-Length', len(data))
+        self.send_header('Access-Control-Allow-Origin', '*')
+        self.end_headers()
+        self.wfile.write(data)
+
+    def do_OPTIONS(self):
+        self.send_response(200)
+        self.send_header('Access-Control-Allow-Origin', '*')
+        self.send_header('Access-Control-Allow-Methods', 'GET, HEAD, POST, OPTIONS')
+        self.send_header('Access-Control-Allow-Headers', 'Content-Type, Range')
+        self.end_headers()
+
+    def do_HEAD(self):
+        parsed = urlparse(self.path)
+        path = unquote(parsed.path)
+
+        if path.startswith('/stremio/') or path == '/' or path.startswith('/manifest.json'):
+            self.send_response(200)
+            self.send_header('Access-Control-Allow-Origin', '*')
+            if path.endswith('.json'):
+                self.send_header('Content-Type', 'application/json')
+            else:
+                self.send_header('Content-Type', 'text/html')
+            self.end_headers()
+        elif path.startswith('/transcode/') or path.startswith('/direct/'):
+            self.send_response(200)
+            self.send_header('Access-Control-Allow-Origin', '*')
+            if path.endswith('.m3u8'):
+                self.send_header('Content-Type', 'application/vnd.apple.mpegurl')
+            elif path.endswith('.ts'):
+                self.send_header('Content-Type', 'video/mp2t')
+            elif path.endswith('.vtt'):
+                self.send_header('Content-Type', 'text/vtt')
+            else:
+                self.send_header('Content-Type', 'application/octet-stream')
+            self.end_headers()
+        elif path.startswith('/file/') or path.startswith('/poster/'):
+            # Generic file serving by CID - /file/{cid} or /poster/{cid} (backward compat)
+            m = re.match(r'^/(?:file|poster)/([a-zA-Z0-9]+)(?:/w(\d+))?$', path)
+            if m:
+                cid = m.group(1)
+                # Check if file exists by looking up CID in metadata
+                file_path = fileserver.get_file_path(cid)
+                if file_path and os.path.exists(file_path):
+                    self.send_response(200)
+                    self.send_header('Access-Control-Allow-Origin', '*')
+                    self.send_header('Content-Type', 'image/jpeg')
+                    self.send_header('Cache-Control', 'public, max-age=604800')
+                    self.end_headers()
+                else:
+                    self.send_error(404, f"File not found: {cid}")
+            else:
+                self.send_error(404, "Invalid file path")
+        else:
+            self.send_error(404)
+
+    def do_POST(self):
+        parsed = urlparse(self.path)
+        path = unquote(parsed.path)
+
+        # Library refresh endpoint
+        if path == '/api/library/refresh':
+            try:
+                # Reconnect storage to refresh
+                storage = stremio.get_storage()
+                if hasattr(storage, 'disconnect'):
+                    storage.disconnect()
+                stremio.init_storage()
+                # Reinitialize fileserver with new storage
+                fileserver.init(stremio.get_storage())
+                return self.send_json({'status': 'ok', 'message': 'Library refreshed'})
+            except Exception as e:
+                return self.send_json({'status': 'error', 'message': str(e)}, 500)
+
+        # Reset metrics
+        if path == '/transcode/reset-metrics':
+            transcoder.reset_metrics()
+            return self.send_json({'status': 'ok'})
+
+        self.send_error(404)
+
+    def do_GET(self):
+        parsed = urlparse(self.path)
+        raw_path = unquote(parsed.path)
+
+        # Parse config from path (e.g., /eyJsYW5n.../manifest.json)
+        path, config = parse_config_from_path(raw_path)
+
+        # Root - serve setup page
+        if path == '/' or path == '/index.html':
+            return self.serve_setup_page()
+
+        # Configure page
+        if path == '/configure':
+            return self.serve_configure_page()
+
+        # API: Get supported languages
+        if path == '/api/languages':
+            return self.send_json({
+                'languages': [{'code': code, 'name': name} for code, name in stremio.get_supported_languages()],
+            })
+
+        # Health check
+        if path == '/health':
+            storage = stremio.get_storage()
+            return self.send_json({
+                'status': 'ok',
+                'storage': storage.get_status() if storage else {'connected': False},
+            })
+
+        # === Dashboard API ===
+
+        # Storage status
+        if path == '/api/storage-status':
+            storage = stremio.get_storage()
+            return self.send_json(storage.get_status() if storage else {'connected': False})
+
+        # Library stats
+        if path == '/api/stats':
+            stats = stremio.get_library_stats()
+            return self.send_json(stats)
+
+        # Library list
+        if path == '/api/library':
+            storage = stremio.get_storage()
+            videos = storage.get_all_videos() if storage else []
+            return self.send_json({
+                'videos': [v.to_dict() for v in videos],
+                'count': len(videos),
+            })
+
+        # Discovered services (for inter-service navigation)
+        if path == '/api/services':
+            return self.handle_services_api()
+
+        # Plugin subscriber status
+        if path == '/api/plugin-subscriber':
+            subscriber = plugin_subscriber.get_subscriber()
+            if subscriber:
+                return self.send_json(subscriber.get_status())
+            return self.send_json({'running': False, 'reason': 'Not initialized'})
+
+        # === File API ===
+        # /file/{cid} or /poster/{cid} (backward compat) - serve file by CID
+        # /file/{cid}/w{width} - serve resized image
+        m = re.match(r'^/(?:file|poster)/([a-zA-Z0-9]+)(?:/w(\d+))?$', path)
+        if m:
+            return self.handle_file(m.group(1), int(m.group(2)) if m.group(2) else None)
+
+        # === Stremio manifest ===
+        if path == '/manifest.json' or path == '/stremio/manifest.json':
+            return self.handle_stremio_manifest(config)
+
+        # Stremio catalog
+        m = re.match(r'^(?:/stremio)?/catalog/(\w+)/([^/]+)(?:/([^.]+))?\.json$', path)
+        if m:
+            return self.handle_stremio_catalog(m.group(1), m.group(2), m.group(3), config)
+
+        # Stremio meta
+        m = re.match(r'^(?:/stremio)?/meta/(\w+)/([^/]+)\.json$', path)
+        if m:
+            return self.handle_stremio_meta(m.group(1), m.group(2), config)
+
+        # Stremio stream
+        m = re.match(r'^(?:/stremio)?/stream/(\w+)/([^/]+)\.json$', path)
+        if m:
+            return self.handle_stremio_stream(m.group(1), m.group(2), config)
+
+        # === Transcoder endpoints ===
+
+        # Metrics
+        if path == '/transcode/metrics':
+            metrics = transcoder.get_metrics()
+            metrics['total_files'] = stremio.get_storage().get_video_count()
+            return self.send_json(metrics)
+
+        # Direct file serving
+        m = re.match(r'^/direct/(.+)$', path)
+        if m:
+            return self.handle_direct_file(m.group(1))
+
+        # Master playlist variants
+        m = re.match(r'^/transcode/(.+?)/master_(\w+)_a(\d+)\.m3u8$', path)
+        if m:
+            return self.handle_master_playlist(m.group(1), m.group(2), int(m.group(3)))
+
+        m = re.match(r'^/transcode/(.+?)/master_a(\d+)\.m3u8$', path)
+        if m:
+            return self.handle_master_playlist(m.group(1), None, int(m.group(2)))
+
+        m = re.match(r'^/transcode/(.+?)/master\.m3u8$', path)
+        if m:
+            return self.handle_master_playlist(m.group(1))
+
+        m = re.match(r'^/transcode/(.+?)/master_(\w+)\.m3u8$', path)
+        if m:
+            return self.handle_master_playlist(m.group(1), m.group(2))
+
+        # Stream playlist
+        m = re.match(r'^/transcode/(.+?)/stream_a(\d+)_(\w+)\.m3u8$', path)
+        if m:
+            return self.handle_stream_playlist(m.group(1), int(m.group(2)), m.group(3))
+
+        # Segment
+        m = re.match(r'^/transcode/(.+?)/seg_a(\d+)_(\w+)_(\d+)\.ts$', path)
+        if m:
+            return self.handle_segment(m.group(1), int(m.group(2)), m.group(3), int(m.group(4)))
+
+        # Subtitle playlist
+        m = re.match(r'^/transcode/(.+?)/subtitle_(\d+)\.m3u8$', path)
+        if m:
+            return self.handle_subtitle_playlist(m.group(1), int(m.group(2)))
+
+        # Subtitle VTT
+        m = re.match(r'^/transcode/(.+?)/subtitle_(\d+)\.vtt$', path)
+        if m:
+            return self.handle_subtitle_vtt(m.group(1), int(m.group(2)))
+
+        self.send_error(404)
+
+    def serve_setup_page(self):
+        """Serve the setup/dashboard page."""
+        html_path = os.path.join(os.path.dirname(__file__), 'www', 'index.html')
+        if not os.path.exists(html_path):
+            # Try parent directory
+            html_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'www', 'index.html')
+
+        if os.path.exists(html_path):
+            with open(html_path, 'r', encoding='utf-8') as f:
+                content = f.read()
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html')
+            self.send_header('Content-Length', len(content.encode('utf-8')))
+            self.end_headers()
+            self.wfile.write(content.encode('utf-8'))
+        else:
+            # Fallback minimal page
+            content = f"""<!DOCTYPE html>
+<html>
+<head><title>Meta-Stremio</title></head>
+<body style="font-family: sans-serif; background: #1a1a2e; color: #fff; padding: 2rem;">
+<h1>Meta-Stremio</h1>
+<p>Install URL: <code>{self.get_base_url()}/manifest.json</code></p>
+<p><a href="stremio://{self.headers.get('Host', 'localhost')}/manifest.json" style="color: #4dabf7;">Install in Stremio</a></p>
+</body>
+</html>"""
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html')
+            self.end_headers()
+            self.wfile.write(content.encode('utf-8'))
+
+    def serve_configure_page(self):
+        """Serve the configuration page for language settings."""
+        html_path = os.path.join(os.path.dirname(__file__), 'www', 'configure.html')
+        if not os.path.exists(html_path):
+            html_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'www', 'configure.html')
+
+        if os.path.exists(html_path):
+            with open(html_path, 'r', encoding='utf-8') as f:
+                content = f.read()
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html')
+            self.send_header('Content-Length', len(content.encode('utf-8')))
+            self.end_headers()
+            self.wfile.write(content.encode('utf-8'))
+        else:
+            # Inline configure page (fallback)
+            base_url = self.get_base_url()
+            host = self.headers.get('Host', 'localhost')
+            languages = stremio.get_supported_languages()
+
+            lang_options = '\n'.join(
+                f'<option value="{code}">{name}</option>'
+                for code, name in languages
+            )
+
+            content = f"""<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Configure Meta-Stremio</title>
+    <style>
+        :root {{
+            --bg-primary: #0a0a0f;
+            --bg-secondary: #12121a;
+            --bg-tertiary: #1a1a2e;
+            --text-primary: #e0e0e0;
+            --text-secondary: #a0a0a0;
+            --accent-primary: #4ecdc4;
+            --border-color: #2a2a3e;
+        }}
+        * {{ box-sizing: border-box; margin: 0; padding: 0; }}
+        body {{
+            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
+            background: var(--bg-primary);
+            color: var(--text-primary);
+            min-height: 100vh;
+            padding: 2rem;
+            display: flex;
+            justify-content: center;
+            align-items: center;
+        }}
+        .container {{
+            max-width: 500px;
+            width: 100%;
+        }}
+        h1 {{
+            font-size: 1.8rem;
+            margin-bottom: 0.5rem;
+            background: linear-gradient(135deg, var(--accent-primary), #ff6b6b);
+            -webkit-background-clip: text;
+            background-clip: text;
+            -webkit-text-fill-color: transparent;
+        }}
+        .subtitle {{ color: var(--text-secondary); margin-bottom: 2rem; }}
+        .card {{
+            background: var(--bg-secondary);
+            border-radius: 12px;
+            padding: 1.5rem;
+            border: 1px solid var(--border-color);
+            margin-bottom: 1.5rem;
+        }}
+        .form-group {{ margin-bottom: 1.5rem; }}
+        .form-group:last-child {{ margin-bottom: 0; }}
+        label {{
+            display: block;
+            margin-bottom: 0.5rem;
+            font-weight: 500;
+        }}
+        .hint {{ font-size: 0.85rem; color: var(--text-secondary); margin-top: 0.25rem; }}
+        select {{
+            width: 100%;
+            padding: 0.75rem;
+            border-radius: 8px;
+            border: 1px solid var(--border-color);
+            background: var(--bg-tertiary);
+            color: var(--text-primary);
+            font-size: 1rem;
+        }}
+        .url-box {{
+            background: var(--bg-tertiary);
+            border-radius: 8px;
+            padding: 1rem;
+            font-family: monospace;
+            font-size: 0.85rem;
+            word-break: break-all;
+            margin-bottom: 1rem;
+            border: 1px solid var(--border-color);
+            color: var(--accent-primary);
+        }}
+        .btn {{
+            display: inline-flex;
+            align-items: center;
+            justify-content: center;
+            gap: 0.5rem;
+            padding: 0.75rem 1.5rem;
+            border-radius: 8px;
+            text-decoration: none;
+            font-weight: 600;
+            font-size: 0.9rem;
+            border: none;
+            cursor: pointer;
+            width: 100%;
+            margin-bottom: 0.75rem;
+        }}
+        .btn-primary {{
+            background: var(--accent-primary);
+            color: var(--bg-primary);
+        }}
+        .btn-secondary {{
+            background: var(--bg-tertiary);
+            color: var(--text-primary);
+            border: 1px solid var(--border-color);
+        }}
+        .warning {{
+            background: rgba(255, 217, 61, 0.15);
+            color: #ffd93d;
+            padding: 0.75rem;
+            border-radius: 8px;
+            font-size: 0.85rem;
+            margin-bottom: 1rem;
+        }}
+    </style>
+</head>
+<body>
+    <div class="container">
+        <h1>Configure Meta-Stremio</h1>
+        <p class="subtitle">Set your language preferences for metadata display</p>
+
+
+        <div class="card">
+            <div class="form-group">
+                <label for="displayLang">Display Language</label>
+                <select id="displayLang">
+                    {lang_options}
+                </select>
+                <p class="hint">Language preference for titles (uses metadata from meta-sort's TMDB plugin)</p>
+            </div>
+        </div>
+
+        <div class="card">
+            <p style="margin-bottom: 0.5rem; font-weight: 500;">Install URL:</p>
+            <div class="url-box" id="installUrl">{base_url}/manifest.json</div>
+
+            <a href="#" id="installBtn" class="btn btn-primary">Install in Stremio</a>
+            <button onclick="copyUrl()" class="btn btn-secondary">Copy URL</button>
+        </div>
+    </div>
+
+    <script>
+        const baseUrl = '{base_url}';
+        const host = '{host}';
+
+        function encodeConfig(config) {{
+            const json = JSON.stringify(config);
+            // URL-safe base64 encode
+            const encoded = btoa(json).replace(/\\+/g, '-').replace(/\\//g, '_').replace(/=+$/, '');
+            return encoded;
+        }}
+
+        function updateUrl() {{
+            const displayLang = document.getElementById('displayLang').value;
+
+            let url, stremioUrl;
+            if (displayLang === 'en') {{
+                // No config needed for default
+                url = baseUrl + '/manifest.json';
+                stremioUrl = 'stremio://' + host + '/manifest.json';
+            }} else {{
+                const config = {{ displayLanguage: displayLang }};
+                const encoded = encodeConfig(config);
+                url = baseUrl + '/' + encoded + '/manifest.json';
+                stremioUrl = 'stremio://' + host + '/' + encoded + '/manifest.json';
+            }}
+
+            document.getElementById('installUrl').textContent = url;
+            document.getElementById('installBtn').href = stremioUrl;
+        }}
+
+        function copyUrl() {{
+            const url = document.getElementById('installUrl').textContent;
+            navigator.clipboard.writeText(url).then(() => {{
+                const btn = event.target;
+                const original = btn.textContent;
+                btn.textContent = 'Copied!';
+                setTimeout(() => btn.textContent = original, 2000);
+            }});
+        }}
+
+        // Update URL when language changes
+        document.getElementById('displayLang').addEventListener('change', updateUrl);
+
+        // Try to detect browser language
+        const browserLang = navigator.language.split('-')[0];
+        const langSelect = document.getElementById('displayLang');
+        for (const opt of langSelect.options) {{
+            if (opt.value === browserLang) {{
+                opt.selected = true;
+                break;
+            }}
+        }}
+
+        // Initial URL update
+        updateUrl();
+    </script>
+</body>
+</html>"""
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html')
+            self.end_headers()
+            self.wfile.write(content.encode('utf-8'))
+
+    def handle_services_api(self):
+        """Return list of discovered services for dashboard navigation."""
+        services = []
+        leader_info = None
+
+        try:
+            sd = get_service_discovery()
+            if sd and sd.is_started():
+                all_services = sd.discover_all_services()
+
+                for svc in all_services:
+                    # Build dashboard URL from API URL
+                    api_url = svc.get('api', '')
+                    dashboard_path = svc.get('endpoints', {}).get('dashboard', '/')
+
+                    services.append({
+                        'name': svc.get('name', 'Unknown'),
+                        'url': api_url + dashboard_path if api_url else '',
+                        'api': api_url,
+                        'status': svc.get('status', 'unknown'),
+                        'capabilities': svc.get('capabilities', []),
+                        'version': svc.get('version', ''),
+                    })
+        except Exception as e:
+            print(f"[Server] Error discovering services: {e}")
+
+        # Get leader info from leader storage
+        try:
+            storage = stremio.get_storage()
+            if storage and hasattr(storage, '_leader_discovery'):
+                discovery = storage._leader_discovery
+                if discovery:
+                    info = discovery.get_leader_info()
+                    if info:
+                        leader_info = {
+                            'host': info.host if hasattr(info, 'host') else info.get('host', ''),
+                            'api': info.api if hasattr(info, 'api') else info.get('api', ''),
+                            'http': info.http if hasattr(info, 'http') else info.get('http', ''),
+                        }
+        except Exception as e:
+            print(f"[Server] Error getting leader info: {e}")
+
+        return self.send_json({
+            'services': services,
+            'current': 'meta-stremio',
+            'leader': leader_info,
+            'isLeader': False,  # meta-stremio is never the leader
+        })
+
+    def handle_file(self, cid: str, width: int = None):
+        """Serve a file by CID with optional resizing (for images)."""
+        file_data, content_type, status = fileserver.serve_file(cid, width)
+
+        if status == 200 and file_data:
+            self.send_response(200)
+            self.send_header('Content-Type', content_type)
+            self.send_header('Content-Length', len(file_data))
+            self.send_header('Cache-Control', 'public, max-age=604800')  # 7 days
+            self.send_header('Access-Control-Allow-Origin', '*')
+            self.end_headers()
+            self.wfile.write(file_data)
+        elif status == 404:
+            self.send_error(404, f"File not found: {cid}")
+        else:
+            self.send_error(status, "Error serving file")
+
+    def handle_stremio_manifest(self, config: dict = None):
+        host = self.get_host()
+        # Set base URL for poster URLs in responses
+        stremio.set_base_url(self.get_base_url())
+        data, content_type = stremio_handler.handle_manifest(host, config)
+        self.send_data(data, content_type)
+
+    def get_host(self) -> str:
+        host = self.headers.get('X-Forwarded-Host') or self.headers.get('Host', 'localhost')
+        if ':' in host and not host.startswith('['):
+            host = host.rsplit(':', 1)[0]
+        return host
+
+    def handle_stremio_catalog(self, catalog_type: str, catalog_id: str, extra: str = None, config: dict = None):
+        extra_dict = {}
+        if extra:
+            for param in extra.split('&'):
+                if '=' in param:
+                    key, value = param.split('=', 1)
+                    extra_dict[key] = unquote(value)
+
+        # Set base URL for poster URLs in responses
+        stremio.set_base_url(self.get_base_url())
+        data, content_type = stremio_handler.handle_catalog(catalog_type, catalog_id, extra_dict, config)
+        self.send_data(data, content_type)
+
+    def handle_stremio_meta(self, meta_type: str, meta_id: str, config: dict = None):
+        # Set base URL for poster URLs in responses
+        stremio.set_base_url(self.get_base_url())
+        data, content_type = stremio_handler.handle_meta(meta_type, meta_id, config)
+        if data:
+            self.send_data(data, content_type)
+        else:
+            self.send_error(404, "Video not found")
+
+    def handle_stremio_stream(self, stream_type: str, stream_id: str, config: dict = None):
+        base_url = self.get_base_url()
+        # Set base URL for poster URLs in responses
+        stremio.set_base_url(base_url)
+        data, content_type = stremio_handler.handle_stream(stream_type, stream_id, base_url, config)
+        if data:
+            self.send_data(data, content_type)
+        else:
+            self.send_error(404, "Video not found")
+
+    # === Transcoder handlers ===
+
+    def get_file_info(self, filepath: str):
+        # Try multiple path resolutions:
+        # 1. Relative to MEDIA_DIR (e.g., /files/watch/filename.mp4)
+        # 2. Absolute path starting with /files/ (e.g., /files/test/filename.mp4)
+        # 3. Prepend /files/ if path looks like files/... (from URL encoding)
+        candidates = [
+            os.path.join(transcoder.MEDIA_DIR, filepath),  # Relative to MEDIA_DIR
+            '/' + filepath if not filepath.startswith('/') else filepath,  # Make absolute
+        ]
+
+        full_path = None
+        for candidate in candidates:
+            if os.path.exists(candidate):
+                full_path = candidate
+                break
+
+        if not full_path:
+            self.send_error(404, f"File not found: {filepath}")
+            return None, None, None
+
+        info = transcoder.get_video_info(full_path)
+        if not info:
+            self.send_error(500, "Could not probe file")
+            return None, None, None
+
+        return full_path, transcoder.get_file_hash(filepath), info
+
+    def handle_master_playlist(self, filepath: str, resolution: str = None, audio: int = None):
+        full_path, file_hash, info = self.get_file_info(filepath)
+        if not info:
+            return
+
+        playlist = transcoder.generate_master_playlist(info, resolution, audio)
+        self.send_data(playlist.encode(), 'application/vnd.apple.mpegurl')
+
+    def handle_stream_playlist(self, filepath: str, audio: int, resolution: str):
+        full_path, file_hash, info = self.get_file_info(filepath)
+        if not info:
+            return
+
+        playlist = transcoder.generate_stream_playlist(info, audio, resolution)
+        self.send_data(playlist.encode(), 'application/vnd.apple.mpegurl')
+
+    def handle_segment(self, filepath: str, audio: int, resolution: str, segment: int):
+        full_path, file_hash, info = self.get_file_info(filepath)
+        if not info:
+            return
+
+        video_codec, audio_codec = transcoder.extract_codecs(info)
+        transcoder.segment_manager.set_codec_info(video_codec, audio_codec, filepath)
+
+        data = transcoder.get_or_transcode_segment(filepath, file_hash, audio, resolution, segment, info)
+        if data:
+            self.send_data(data, 'video/mp2t')
+        else:
+            self.send_error(500, "Transcode failed")
+
+    def handle_subtitle_playlist(self, filepath: str, sub_index: int):
+        full_path, file_hash, info = self.get_file_info(filepath)
+        if not info:
+            return
+
+        playlist = transcoder.generate_subtitle_playlist(info, sub_index)
+        self.send_data(playlist.encode(), 'application/vnd.apple.mpegurl')
+
+    def handle_subtitle_vtt(self, filepath: str, sub_index: int):
+        full_path, file_hash, info = self.get_file_info(filepath)
+        if not info:
+            return
+
+        key = f"{file_hash}:sub:{sub_index}"
+        content, error = transcoder.subtitle_manager.get_subtitle(key, full_path, file_hash, sub_index, info)
+
+        if content:
+            self.send_data(content.encode('utf-8'), 'text/vtt')
+        else:
+            error_vtt = f"WEBVTT\n\nNOTE Subtitle extraction failed: {error or 'Unknown error'}\n"
+            self.send_data(error_vtt.encode('utf-8'), 'text/vtt')
+
+    def handle_direct_file(self, filepath: str):
+        # Try multiple path resolutions (same logic as get_file_info)
+        candidates = [
+            os.path.join(transcoder.MEDIA_DIR, filepath),  # Relative to MEDIA_DIR
+            '/' + filepath if not filepath.startswith('/') else filepath,  # Make absolute
+        ]
+
+        full_path = None
+        for candidate in candidates:
+            if os.path.exists(candidate):
+                full_path = candidate
+                break
+
+        if not full_path:
+            self.send_error(404, f"File not found: {filepath}")
+            return
+
+        ext = os.path.splitext(filepath)[1].lower()
+        content_types = {
+            '.mp4': 'video/mp4',
+            '.mkv': 'video/x-matroska',
+            '.webm': 'video/webm',
+            '.avi': 'video/x-msvideo',
+            '.mov': 'video/quicktime',
+            '.m4v': 'video/x-m4v',
+            '.ts': 'video/mp2t',
+            '.m2ts': 'video/mp2t',
+        }
+        content_type = content_types.get(ext, 'application/octet-stream')
+
+        file_size = os.path.getsize(full_path)
+        range_header = self.headers.get('Range')
+
+        if range_header:
+            range_match = re.match(r'bytes=(\d*)-(\d*)', range_header)
+            if range_match:
+                start = int(range_match.group(1)) if range_match.group(1) else 0
+                end = int(range_match.group(2)) if range_match.group(2) else file_size - 1
+                end = min(end, file_size - 1)
+                length = end - start + 1
+
+                self.send_response(206)
+                self.send_header('Content-Type', content_type)
+                self.send_header('Content-Length', length)
+                self.send_header('Content-Range', f'bytes {start}-{end}/{file_size}')
+                self.send_header('Accept-Ranges', 'bytes')
+                self.send_header('Access-Control-Allow-Origin', '*')
+                self.end_headers()
+
+                with open(full_path, 'rb') as f:
+                    f.seek(start)
+                    remaining = length
+                    chunk_size = 64 * 1024
+                    while remaining > 0:
+                        chunk = f.read(min(chunk_size, remaining))
+                        if not chunk:
+                            break
+                        self.wfile.write(chunk)
+                        remaining -= len(chunk)
+                return
+
+        self.send_response(200)
+        self.send_header('Content-Type', content_type)
+        self.send_header('Content-Length', file_size)
+        self.send_header('Accept-Ranges', 'bytes')
+        self.send_header('Access-Control-Allow-Origin', '*')
+        self.end_headers()
+
+        with open(full_path, 'rb') as f:
+            chunk_size = 64 * 1024
+            while True:
+                chunk = f.read(chunk_size)
+                if not chunk:
+                    break
+                self.wfile.write(chunk)
+
+    def get_base_url(self) -> str:
+        host = self.headers.get('X-Forwarded-Host') or self.headers.get('Host', 'localhost')
+
+        # SCHEME can be 'http', 'https', 'auto', or empty
+        # 'auto' or empty means auto-detect based on host
+        if SCHEME and SCHEME not in ('auto', ''):
+            proto = SCHEME
+        else:
+            is_localhost = 'localhost' in host or '127.0.0.1' in host or '::1' in host
+            proto = 'http' if is_localhost else 'https'
+
+        clean_host = host.replace(':80', '').replace(':443', '')
+        return f"{proto}://{clean_host}"
+
+
+class ThreadedServer(HTTPServer):
+    def process_request(self, request, client_address):
+        t = threading.Thread(target=self._handle, args=(request, client_address), daemon=True)
+        t.start()
+
+    def _handle(self, request, client_address):
+        try:
+            self.finish_request(request, client_address)
+        except Exception:
+            self.handle_error(request, client_address)
+        finally:
+            self.shutdown_request(request)
+
+
+def shutdown_handler(signum, frame):
+    """Handle graceful shutdown."""
+    print("\n[Server] Shutting down...")
+
+    # Stop plugin subscriber
+    try:
+        plugin_subscriber.stop_subscriber()
+    except Exception as e:
+        print(f"[Server] Error stopping plugin subscriber: {e}")
+
+    # Stop service discovery
+    try:
+        sd = get_service_discovery()
+        if sd:
+            sd.stop()
+    except Exception as e:
+        print(f"[Server] Error stopping service discovery: {e}")
+
+    sys.exit(0)
+
+
+def main():
+    # Register shutdown handlers
+    signal.signal(signal.SIGINT, shutdown_handler)
+    signal.signal(signal.SIGTERM, shutdown_handler)
+
+    storage = stremio.get_storage()
+    storage_type = storage.get_status().get('type', 'unknown')
+    video_count = storage.get_video_count()
+
+    # Initialize fileserver with storage reference
+    fileserver.init(storage)
+
+    # Initialize plugin subscriber to receive notifications when
+    # filename-parser and tmdb plugins complete
+    subscriber = plugin_subscriber.init_subscriber()
+
+    print(f"Meta-Stremio starting on port {PORT}")
+    print(f"Storage: {storage_type} | Videos: {video_count}")
+    print(f"Media: {transcoder.MEDIA_DIR} | Cache: {transcoder.CACHE_DIR}")
+    print(f"Segment: {transcoder.SEGMENT_DURATION}s | Prefetch: {transcoder.PREFETCH_SEGMENTS} segments")
+    print(f"Manifest URL: http://localhost:{PORT}/manifest.json")
+    print(f"Dashboard: http://localhost:{PORT}/")
+    print("Adaptive quality: target 60-80% transcode ratio")
+
+    if service_discovery:
+        print(f"Service discovery: enabled (registered as meta-stremio)")
+    else:
+        print(f"Service discovery: disabled (META_CORE_PATH not found)")
+
+    if subscriber:
+        print(f"Plugin subscriber: enabled (watching: filename-parser, tmdb)")
+    else:
+        print(f"Plugin subscriber: disabled (Redis not available)")
+
+    ThreadedServer(('0.0.0.0', PORT), Handler).serve_forever()
+
+
+if __name__ == '__main__':
+    main()

src/storage/__init__.py

@@ -0,0 +1,33 @@
+"""
+Storage abstraction layer for meta-stremio.
+
+Provides a unified interface for reading video metadata from different backends:
+- LeaderStorage: Discovers and connects to the KV leader (recommended)
+- RedisStorage: Reads from Redis with direct URL (legacy)
+
+Also provides service discovery for inter-service communication.
+"""
+
+from .provider import StorageProvider, VideoMetadata
+from .redis_storage import RedisStorage
+from .leader_storage import LeaderStorage
+from .leader_discovery import LeaderDiscovery, LeaderLockInfo
+from .service_discovery import (
+    ServiceDiscovery,
+    ServiceInfo,
+    get_service_discovery,
+    init_service_discovery
+)
+
+__all__ = [
+    'StorageProvider',
+    'VideoMetadata',
+    'RedisStorage',
+    'LeaderStorage',
+    'LeaderDiscovery',
+    'LeaderLockInfo',
+    'ServiceDiscovery',
+    'ServiceInfo',
+    'get_service_discovery',
+    'init_service_discovery',
+]

src/storage/leader_discovery.py

@@ -0,0 +1,300 @@
+"""
+Leader Discovery for meta-stremio (FOLLOWER-only)
+
+This is a Python implementation of the leader discovery pattern used in meta-fuse.
+meta-stremio never spawns Redis - it only connects to the leader's Redis instance.
+
+Architecture:
+- Reads info file at META_CORE_PATH/locks/kv-leader.info
+- The leader holds flock on kv-leader.lock, writes metadata to kv-leader.info
+- Parses leader info (Redis URL, HTTP URL)
+- Watches for leader changes via polling and file watching
+- Triggers reconnection callbacks on leader failure/change
+"""
+from __future__ import annotations
+
+import os
+import json
+import time
+import threading
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Callable
+from urllib.request import urlopen
+from urllib.error import URLError
+
+# Try to import watchdog for file watching (optional)
+try:
+    from watchdog.observers import Observer
+    from watchdog.events import FileSystemEventHandler
+    WATCHDOG_AVAILABLE = True
+except ImportError:
+    WATCHDOG_AVAILABLE = False
+    Observer = None
+    FileSystemEventHandler = object
+
+
+@dataclass
+class LeaderLockInfo:
+    """Leader lock file content format."""
+    host: str
+    api: str  # Redis connection URL (e.g., redis://10.0.1.50:6379)
+    http: str  # HTTP API URL for the leader service
+    timestamp: int
+    pid: int
+
+
+class LockFileHandler(FileSystemEventHandler):
+    """File system event handler for lock file changes."""
+
+    def __init__(self, callback: Callable[[], None], lock_filename: str):
+        self.callback = callback
+        self.lock_filename = lock_filename
+
+    def on_modified(self, event):
+        if event.src_path.endswith(self.lock_filename):
+            self.callback()
+
+    def on_created(self, event):
+        if event.src_path.endswith(self.lock_filename):
+            self.callback()
+
+    def on_deleted(self, event):
+        if event.src_path.endswith(self.lock_filename):
+            self.callback()
+
+
+class LeaderDiscovery:
+    """
+    Leader Discovery for meta-stremio.
+
+    Discovers and monitors the KV leader via a lock file in the shared filesystem.
+    """
+
+    def __init__(
+        self,
+        meta_core_path: str,
+        health_check_interval: float = 5.0,
+        max_failures: int = 3
+    ):
+        self.meta_core_path = meta_core_path
+        self.health_check_interval = health_check_interval
+        self.max_failures = max_failures
+
+        self.lock_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info')
+        self.leader_info: Optional[LeaderLockInfo] = None
+
+        self._is_started = False
+        self._stop_event = threading.Event()
+        self._health_check_thread: Optional[threading.Thread] = None
+        self._file_observer = None
+        self._consecutive_failures = 0
+
+        # Event callbacks
+        self._on_leader_found_callback: Optional[Callable[[LeaderLockInfo], None]] = None
+        self._on_leader_lost_callback: Optional[Callable[[], None]] = None
+
+    def start(self) -> None:
+        """Start watching for leader."""
+        if self._is_started:
+            print("[LeaderDiscovery] Already started")
+            return
+
+        print(f"[LeaderDiscovery] Starting, lock file: {self.lock_file_path}")
+
+        # Ensure lock directory exists
+        self._ensure_lock_dir()
+
+        # Try to read existing leader info
+        self._check_for_leader()
+
+        # Start file watcher (if watchdog is available)
+        self._start_file_watcher()
+
+        # Start health check thread
+        self._start_health_check()
+
+        self._is_started = True
+
+    def stop(self) -> None:
+        """Stop watching."""
+        if not self._is_started:
+            return
+
+        print("[LeaderDiscovery] Stopping...")
+
+        self._stop_event.set()
+
+        # Stop health check thread
+        if self._health_check_thread and self._health_check_thread.is_alive():
+            self._health_check_thread.join(timeout=2.0)
+
+        # Stop file observer
+        if self._file_observer:
+            self._file_observer.stop()
+            self._file_observer.join(timeout=2.0)
+            self._file_observer = None
+
+        self._is_started = False
+
+    def _ensure_lock_dir(self) -> None:
+        """Ensure lock directory exists."""
+        lock_dir = os.path.dirname(self.lock_file_path)
+        try:
+            os.makedirs(lock_dir, exist_ok=True)
+        except OSError as e:
+            print(f"[LeaderDiscovery] Could not create lock directory: {e}")
+
+    def _check_for_leader(self) -> None:
+        """Check for leader by reading lock file."""
+        try:
+            with open(self.lock_file_path, 'r') as f:
+                content = f.read()
+
+            data = json.loads(content)
+
+            # Validate leader info
+            if data and data.get('api') and data.get('http'):
+                info = LeaderLockInfo(
+                    host=data.get('host', 'unknown'),
+                    api=data['api'],
+                    http=data['http'],
+                    timestamp=data.get('timestamp', 0),
+                    pid=data.get('pid', 0)
+                )
+
+                is_new = (
+                    self.leader_info is None or
+                    self.leader_info.api != info.api
+                )
+
+                self.leader_info = info
+                self._consecutive_failures = 0
+
+                if is_new:
+                    print(f"[LeaderDiscovery] Leader found: {info.host} at {info.api}")
+                    if self._on_leader_found_callback:
+                        self._on_leader_found_callback(info)
+
+        except FileNotFoundError:
+            # Lock file doesn't exist - no leader yet
+            if self.leader_info:
+                print("[LeaderDiscovery] Lock file removed, leader lost")
+                self._handle_leader_lost()
+
+        except json.JSONDecodeError as e:
+            print(f"[LeaderDiscovery] Error parsing lock file: {e}")
+
+        except Exception as e:
+            print(f"[LeaderDiscovery] Error reading lock file: {e}")
+
+    def _is_leader_healthy(self) -> bool:
+        """Check if current leader is still healthy."""
+        if not self.leader_info:
+            return False
+
+        try:
+            health_url = f"{self.leader_info.http}/health"
+            with urlopen(health_url, timeout=5) as response:
+                return response.status == 200
+        except (URLError, TimeoutError, Exception):
+            return False
+
+    def _handle_leader_lost(self) -> None:
+        """Handle leader being lost."""
+        self.leader_info = None
+        if self._on_leader_lost_callback:
+            self._on_leader_lost_callback()
+
+    def _start_file_watcher(self) -> None:
+        """Start watching lock file for changes."""
+        if not WATCHDOG_AVAILABLE:
+            print("[LeaderDiscovery] watchdog not available, using polling only")
+            return
+
+        try:
+            lock_dir = os.path.dirname(self.lock_file_path)
+            handler = LockFileHandler(
+                callback=self._check_for_leader,
+                lock_filename='kv-leader.info'
+            )
+
+            self._file_observer = Observer()
+            self._file_observer.schedule(handler, lock_dir, recursive=False)
+            self._file_observer.start()
+            print(f"[LeaderDiscovery] File watcher started for {lock_dir}")
+
+        except Exception as e:
+            print(f"[LeaderDiscovery] Could not start file watcher: {e}")
+
+    def _start_health_check(self) -> None:
+        """Start health check loop."""
+        def health_check_loop():
+            while not self._stop_event.is_set():
+                try:
+                    if not self.leader_info:
+                        # No leader known, try to find one
+                        self._check_for_leader()
+                    else:
+                        # Check if current leader is healthy
+                        healthy = self._is_leader_healthy()
+
+                        if healthy:
+                            self._consecutive_failures = 0
+                        else:
+                            self._consecutive_failures += 1
+                            print(f"[LeaderDiscovery] Leader health check failed ({self._consecutive_failures}/{self.max_failures})")
+
+                            if self._consecutive_failures >= self.max_failures:
+                                print("[LeaderDiscovery] Leader appears dead, marking as lost")
+                                self._handle_leader_lost()
+
+                                # Try to find new leader
+                                self._check_for_leader()
+
+                except Exception as e:
+                    print(f"[LeaderDiscovery] Health check error: {e}")
+
+                self._stop_event.wait(self.health_check_interval)
+
+        self._health_check_thread = threading.Thread(
+            target=health_check_loop,
+            daemon=True,
+            name="leader-health-check"
+        )
+        self._health_check_thread.start()
+
+    # ========================================================================
+    # Event Registration
+    # ========================================================================
+
+    def on_leader_found(self, callback: Callable[[LeaderLockInfo], None]) -> 'LeaderDiscovery':
+        """Register callback for when leader is found/changed."""
+        self._on_leader_found_callback = callback
+
+        # If we already have leader info, call immediately
+        if self.leader_info:
+            callback(self.leader_info)
+
+        return self
+
+    def on_leader_lost(self, callback: Callable[[], None]) -> 'LeaderDiscovery':
+        """Register callback for when leader is lost."""
+        self._on_leader_lost_callback = callback
+        return self
+
+    # ========================================================================
+    # Getters
+    # ========================================================================
+
+    def get_leader_info(self) -> Optional[LeaderLockInfo]:
+        """Get current leader info."""
+        return self.leader_info
+
+    def get_redis_url(self) -> Optional[str]:
+        """Get Redis URL of current leader."""
+        return self.leader_info.api if self.leader_info else None
+
+    def has_leader(self) -> bool:
+        """Check if a leader is known."""
+        return self.leader_info is not None

src/storage/leader_storage.py

@@ -0,0 +1,496 @@
+"""
+Leader-aware Storage Provider for meta-stremio.
+
+Uses leader discovery to automatically connect to the KV leader.
+Handles reconnection when the leader changes or becomes unavailable.
+"""
+from __future__ import annotations
+
+import os
+import threading
+from typing import List, Optional, Callable
+
+from .provider import StorageProvider, VideoMetadata
+from .leader_discovery import LeaderDiscovery, LeaderLockInfo
+
+try:
+    import redis
+    REDIS_AVAILABLE = True
+except ImportError:
+    REDIS_AVAILABLE = False
+    redis = None
+
+
+# Configuration from environment
+META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
+FILES_PATH = os.environ.get('FILES_PATH', '/files')
+REDIS_PREFIX = os.environ.get('REDIS_PREFIX', 'meta-sort:')
+
+
+class LeaderStorage(StorageProvider):
+    """
+    Storage provider that discovers and connects to the KV leader.
+
+    Uses leader discovery to automatically find the Redis instance
+    managed by meta-sort. Handles reconnection on leader failure.
+    """
+
+    def __init__(
+        self,
+        meta_core_path: str = None,
+        files_path: str = None,
+        prefix: str = None,
+        redis_url: str = None  # Override - skip leader discovery
+    ):
+        if not REDIS_AVAILABLE:
+            raise ImportError("redis package not installed. Run: pip install redis")
+
+        self._meta_core_path = meta_core_path or META_CORE_PATH
+        self._files_path = files_path or FILES_PATH
+        self._prefix = prefix or REDIS_PREFIX
+        self._override_url = redis_url  # Direct Redis URL (skip leader discovery)
+
+        self._leader_discovery: Optional[LeaderDiscovery] = None
+        self._client: Optional[redis.Redis] = None
+        self._connected = False
+        self._lock = threading.Lock()
+
+        # Callbacks
+        self._on_ready_callbacks: List[Callable[[], None]] = []
+        self._on_disconnect_callbacks: List[Callable[[], None]] = []
+
+    def connect(self) -> None:
+        """Connect to the storage backend via leader discovery."""
+        # If direct URL provided, skip leader discovery
+        if self._override_url:
+            print(f"[LeaderStorage] Using direct Redis URL: {self._override_url}")
+            self._connect_to_redis(self._override_url)
+            return
+
+        # Use leader discovery
+        self._leader_discovery = LeaderDiscovery(
+            meta_core_path=self._meta_core_path
+        )
+
+        # Set up callbacks
+        self._leader_discovery.on_leader_found(self._on_leader_found)
+        self._leader_discovery.on_leader_lost(self._on_leader_lost)
+
+        # Start discovery
+        self._leader_discovery.start()
+
+    def disconnect(self) -> None:
+        """Disconnect from the storage backend."""
+        # Stop leader discovery
+        if self._leader_discovery:
+            self._leader_discovery.stop()
+            self._leader_discovery = None
+
+        # Disconnect Redis
+        self._disconnect_redis()
+
+    def is_connected(self) -> bool:
+        """Check if connected to the storage backend."""
+        with self._lock:
+            if not self._connected or not self._client:
+                return False
+            try:
+                self._client.ping()
+                return True
+            except Exception:
+                self._connected = False
+                return False
+
+    def _on_leader_found(self, info: LeaderLockInfo) -> None:
+        """Handle leader found event."""
+        print(f"[LeaderStorage] Connecting to leader at {info.api}...")
+        self._connect_to_redis(info.api)
+
+    def _on_leader_lost(self) -> None:
+        """Handle leader lost event."""
+        print("[LeaderStorage] Leader lost, disconnecting...")
+        self._disconnect_redis()
+        self._notify_disconnect()
+
+    def _connect_to_redis(self, url: str) -> None:
+        """Connect to Redis."""
+        with self._lock:
+            try:
+                # Disconnect existing client
+                self._disconnect_redis_unlocked()
+
+                # Create new client
+                self._client = redis.from_url(url, decode_responses=True)
+                self._client.ping()  # Test connection
+                self._connected = True
+
+                print(f"[LeaderStorage] Connected to Redis at {url}")
+
+                # Notify ready
+                self._notify_ready()
+
+            except Exception as e:
+                print(f"[LeaderStorage] Failed to connect to Redis: {e}")
+                self._client = None
+                self._connected = False
+
+    def _disconnect_redis(self) -> None:
+        """Disconnect from Redis."""
+        with self._lock:
+            self._disconnect_redis_unlocked()
+
+    def _disconnect_redis_unlocked(self) -> None:
+        """Disconnect from Redis (without lock)."""
+        if self._client:
+            try:
+                self._client.close()
+            except Exception:
+                pass
+            self._client = None
+        self._connected = False
+
+    def _notify_ready(self) -> None:
+        """Notify ready callbacks."""
+        for callback in self._on_ready_callbacks:
+            try:
+                callback()
+            except Exception as e:
+                print(f"[LeaderStorage] Error in ready callback: {e}")
+
+    def _notify_disconnect(self) -> None:
+        """Notify disconnect callbacks."""
+        for callback in self._on_disconnect_callbacks:
+            try:
+                callback()
+            except Exception as e:
+                print(f"[LeaderStorage] Error in disconnect callback: {e}")
+
+    # ========================================================================
+    # Event Registration
+    # ========================================================================
+
+    def on_ready(self, callback: Callable[[], None]) -> 'LeaderStorage':
+        """Register callback for when storage is ready."""
+        self._on_ready_callbacks.append(callback)
+        if self._connected:
+            callback()
+        return self
+
+    def on_disconnect(self, callback: Callable[[], None]) -> 'LeaderStorage':
+        """Register callback for disconnection."""
+        self._on_disconnect_callbacks.append(callback)
+        return self
+
+    # ========================================================================
+    # Key Schema Helpers
+    # ========================================================================
+
+    def _get_file_key(self, hash_id: str) -> str:
+        """Get the Redis key for a file hash."""
+        return f"{self._prefix}file:{hash_id}"
+
+    def _resolve_path(self, path: str) -> str:
+        """Resolve path by prepending FILES_PATH if needed."""
+        if not path:
+            return path
+        if path.startswith(self._files_path):
+            return path
+        if path.startswith('/'):
+            return f"{self._files_path}{path}"
+        return f"{self._files_path}/{path}"
+
+    def _parse_nested_array(self, data: dict, prefix: str) -> list:
+        """Parse nested array format (prefix/0, prefix/1, etc.) into a list."""
+        items = []
+        index = 0
+        while True:
+            key = f'{prefix}/{index}'
+            if key in data:
+                items.append(data[key])
+                index += 1
+            else:
+                break
+        return items
+
+    def _parse_video(self, hash_id: str, data: dict) -> Optional[VideoMetadata]:
+        """Parse Redis hash data into VideoMetadata.
+
+        Handles both flat keys and nested keys (e.g., 'plot/eng' for description).
+        meta-sort stores data using a flattened nested key format.
+        """
+        if not data:
+            return None
+
+        import json as jsonlib
+
+        # Parse JSON fields
+        audio_tracks = []
+        subtitles = []
+        genres = []
+        studios = []
+
+        if 'audioTracks' in data:
+            try:
+                audio_tracks = jsonlib.loads(data['audioTracks'])
+            except (jsonlib.JSONDecodeError, TypeError):
+                pass
+
+        if 'subtitles' in data:
+            try:
+                subtitles = jsonlib.loads(data['subtitles'])
+            except (jsonlib.JSONDecodeError, TypeError):
+                pass
+
+        # Parse genres - can be JSON array or nested keys (genres/0, genres/1, etc.)
+        if 'genres' in data:
+            try:
+                genres = jsonlib.loads(data['genres'])
+            except (jsonlib.JSONDecodeError, TypeError):
+                if isinstance(data['genres'], str):
+                    genres = [g.strip() for g in data['genres'].split(',') if g.strip()]
+        else:
+            # Try nested key format (genres/0, genres/1, etc.)
+            genres = self._parse_nested_array(data, 'genres')
+
+        # Parse studios - nested keys (studio/0, studio/1, etc.)
+        studios = self._parse_nested_array(data, 'studio')
+
+        # Parse numeric fields
+        def parse_int(val):
+            try:
+                return int(val) if val else None
+            except (ValueError, TypeError):
+                return None
+
+        def parse_float(val):
+            try:
+                return float(val) if val else None
+            except (ValueError, TypeError):
+                return None
+
+        # Resolve file path with FILES_PATH
+        file_path = data.get('filePath', data.get('sourcePath', ''))
+        resolved_path = self._resolve_path(file_path)
+
+        # Normalize video type: 'tvshow' -> 'series' for Stremio compatibility
+        raw_type = data.get('type', data.get('videoType', 'movie'))
+        video_type = 'series' if raw_type == 'tvshow' else raw_type
+
+        # Get description from nested plot/eng or flat description/plot
+        description = data.get('description')
+        if not description:
+            description = data.get('plot')
+        if not description:
+            # Try nested format: plot/eng, plot/en, etc.
+            for lang in ['eng', 'en', 'english', 'und']:
+                description = data.get(f'plot/{lang}')
+                if description:
+                    break
+
+        # Parse rating (stored as string by TMDBProcessor)
+        rating = parse_float(data.get('rating'))
+
+        # Parse release date
+        release_date = data.get('releasedate', data.get('releaseDate'))
+
+        # Parse episode title - try episodeTitle or extract from titles/eng
+        episode_title = data.get('episodeTitle', data.get('episodeName'))
+        if not episode_title:
+            episode_title = data.get('titles/eng', data.get('titles/en'))
+
+        # Parse tagline
+        tagline = data.get('tagline')
+
+        return VideoMetadata(
+            hash_id=hash_id,
+            file_path=resolved_path,
+            title=data.get('title', data.get('originalTitle', '')),
+            video_type=video_type,
+            year=parse_int(data.get('year', data.get('movieYear'))),
+            season=parse_int(data.get('season')),
+            episode=parse_int(data.get('episode')),
+            duration=parse_float(data.get('duration', data.get('fileinfo/duration'))),
+            width=parse_int(data.get('width', data.get('fileinfo/streamdetails/video/0/width'))),
+            height=parse_int(data.get('height', data.get('fileinfo/streamdetails/video/0/height'))),
+            video_codec=data.get('videoCodec', data.get('fileinfo/streamdetails/video/0/codec')),
+            audio_codec=data.get('audioCodec', data.get('fileinfo/streamdetails/audio/0/codec')),
+            container=data.get('container'),
+            file_size=parse_int(data.get('fileSize', data.get('sizeByte'))),
+            audio_tracks=audio_tracks,
+            subtitles=subtitles,
+            imdb_id=data.get('imdbId', data.get('imdbid')),
+            tmdb_id=data.get('tmdbId', data.get('tmdbid')),
+            poster=data.get('poster') or data.get('posterUrl'),  # Fallback to posterUrl for backward compat
+            backdrop=data.get('backdrop') or data.get('backdropUrl'),  # Fallback to backdropUrl
+            poster_path=data.get('posterPath'),
+            backdrop_path=data.get('backdropPath'),
+            description=description,
+            genres=genres,
+            episode_title=episode_title,
+            rating=rating,
+            release_date=release_date,
+            tagline=tagline,
+            studios=studios,
+        )
+
+    # ========================================================================
+    # StorageProvider Interface
+    # ========================================================================
+
+    def get_all_videos(self) -> List[VideoMetadata]:
+        """Get all videos from storage."""
+        if not self.is_connected():
+            return []
+
+        videos = []
+        pattern = f"{self._prefix}file:*"
+
+        try:
+            for key in self._client.scan_iter(pattern, count=100):
+                # Skip index key
+                if '__index__' in key:
+                    continue
+
+                # Check file type - only include video files
+                file_type = self._client.hget(key, 'fileType')
+                if file_type != 'video':
+                    continue
+
+                hash_id = key.replace(f"{self._prefix}file:", "")
+                data = self._client.hgetall(key)
+                video = self._parse_video(hash_id, data)
+                if video and video.file_path:
+                    videos.append(video)
+
+        except Exception as e:
+            print(f"[LeaderStorage] Error getting all videos: {e}")
+
+        # Sort by title
+        videos.sort(key=lambda v: v.title.lower())
+        return videos
+
+    def get_video_by_hash(self, hash_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its hash ID."""
+        if not self.is_connected():
+            return None
+
+        try:
+            key = self._get_file_key(hash_id)
+            data = self._client.hgetall(key)
+            return self._parse_video(hash_id, data)
+        except Exception as e:
+            print(f"[LeaderStorage] Error getting video {hash_id}: {e}")
+            return None
+
+    def get_videos_by_type(self, video_type: str) -> List[VideoMetadata]:
+        """Get all videos of a specific type."""
+        videos = self.get_all_videos()
+        return [v for v in videos if v.video_type == video_type]
+
+    def search_videos(self, query: str) -> List[VideoMetadata]:
+        """Search videos by title."""
+        if not query:
+            return self.get_all_videos()
+
+        query_lower = query.lower()
+        videos = self.get_all_videos()
+        return [v for v in videos if query_lower in v.title.lower()]
+
+    def get_video_by_imdb_id(self, imdb_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its IMDB ID (e.g., 'tt1727587')."""
+        if not self.is_connected() or not imdb_id:
+            return None
+
+        # Normalize IMDB ID format
+        imdb_id = imdb_id.lower()
+        if not imdb_id.startswith('tt'):
+            imdb_id = f"tt{imdb_id}"
+
+        try:
+            # Scan all videos looking for matching IMDB ID
+            pattern = f"{self._prefix}file:*"
+            for key in self._client.scan_iter(pattern, count=100):
+                if '__index__' in key:
+                    continue
+
+                # Check imdbid/imdbId field
+                stored_imdb = self._client.hget(key, 'imdbid') or self._client.hget(key, 'imdbId')
+                if stored_imdb and stored_imdb.lower() == imdb_id:
+                    hash_id = key.replace(f"{self._prefix}file:", "")
+                    data = self._client.hgetall(key)
+                    return self._parse_video(hash_id, data)
+
+        except Exception as e:
+            print(f"[LeaderStorage] Error finding video by IMDB ID {imdb_id}: {e}")
+
+        return None
+
+    def get_video_count(self) -> int:
+        """Get total number of videos."""
+        if not self.is_connected():
+            return 0
+
+        try:
+            # Try index first
+            index_key = f"{self._prefix}file:__index__"
+            count = self._client.scard(index_key)
+            if count > 0:
+                return count
+
+            # Fallback to scanning
+            pattern = f"{self._prefix}file:*"
+            count = 0
+            for key in self._client.scan_iter(pattern, count=100):
+                if '__index__' not in key:
+                    count += 1
+            return count
+
+        except Exception as e:
+            print(f"[LeaderStorage] Error counting videos: {e}")
+            return 0
+
+    def get_status(self) -> dict:
+        """Get storage status for dashboard."""
+        leader_info = None
+        if self._leader_discovery:
+            info = self._leader_discovery.get_leader_info()
+            if info:
+                leader_info = {
+                    'host': info.host,
+                    'api': info.api,
+                    'http': info.http,
+                }
+
+        status = {
+            'type': 'leader',
+            'connected': self.is_connected(),
+            'meta_core_path': self._meta_core_path,
+            'files_path': self._files_path,
+            'prefix': self._prefix,
+            'video_count': 0,
+            'leader': leader_info,
+        }
+
+        if status['connected']:
+            try:
+                status['video_count'] = self.get_video_count()
+                info = self._client.info('memory')
+                status['memory_used'] = info.get('used_memory_human', 'N/A')
+            except Exception:
+                pass
+
+        return status
+
+    # ========================================================================
+    # Pub/Sub (Future enhancement)
+    # ========================================================================
+
+    def on_video_added(self, callback: Callable[[VideoMetadata], None]) -> None:
+        """Subscribe to video added events."""
+        # TODO: Implement pub/sub for real-time updates
+        pass
+
+    def on_video_removed(self, callback: Callable[[str], None]) -> None:
+        """Subscribe to video removed events."""
+        # TODO: Implement pub/sub for real-time updates
+        pass

src/storage/provider.py

@@ -0,0 +1,148 @@
+"""
+Abstract StorageProvider interface for video metadata.
+
+meta-stremio is a read-only consumer - it does NOT write metadata.
+All writes are handled by meta-sort.
+"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import List, Optional, Callable
+
+
+@dataclass
+class VideoMetadata:
+    """Video metadata stored in KV by meta-sort, read by meta-stremio."""
+    hash_id: str
+    file_path: str
+    title: str
+    video_type: str = 'movie'  # 'movie' | 'series' | 'anime'
+    year: Optional[int] = None
+    season: Optional[int] = None
+    episode: Optional[int] = None
+    duration: Optional[float] = None
+    width: Optional[int] = None
+    height: Optional[int] = None
+    video_codec: Optional[str] = None
+    audio_codec: Optional[str] = None
+    container: Optional[str] = None
+    file_size: Optional[int] = None
+    audio_tracks: List[dict] = field(default_factory=list)
+    subtitles: List[dict] = field(default_factory=list)
+    imdb_id: Optional[str] = None
+    tmdb_id: Optional[str] = None
+    poster: Optional[str] = None
+    backdrop: Optional[str] = None
+    poster_path: Optional[str] = None
+    backdrop_path: Optional[str] = None
+    description: Optional[str] = None
+    genres: List[str] = field(default_factory=list)
+    # New fields for improved listings
+    episode_title: Optional[str] = None  # Title specific to this episode
+    rating: Optional[float] = None  # TMDB/IMDB rating (0-10)
+    release_date: Optional[str] = None  # Full release date (YYYY-MM-DD)
+    tagline: Optional[str] = None  # Movie/series tagline
+    studios: List[str] = field(default_factory=list)  # Production studios
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary for JSON serialization."""
+        return {
+            'hash_id': self.hash_id,
+            'file_path': self.file_path,
+            'title': self.title,
+            'video_type': self.video_type,
+            'year': self.year,
+            'season': self.season,
+            'episode': self.episode,
+            'duration': self.duration,
+            'width': self.width,
+            'height': self.height,
+            'video_codec': self.video_codec,
+            'audio_codec': self.audio_codec,
+            'container': self.container,
+            'file_size': self.file_size,
+            'audio_tracks': self.audio_tracks,
+            'subtitles': self.subtitles,
+            'imdb_id': self.imdb_id,
+            'tmdb_id': self.tmdb_id,
+            'poster': self.poster,
+            'backdrop': self.backdrop,
+            'poster_path': self.poster_path,
+            'backdrop_path': self.backdrop_path,
+            'description': self.description,
+            'genres': self.genres,
+            'episode_title': self.episode_title,
+            'rating': self.rating,
+            'release_date': self.release_date,
+            'tagline': self.tagline,
+            'studios': self.studios,
+        }
+
+
+class StorageProvider(ABC):
+    """
+    Abstract interface for video metadata storage.
+
+    meta-stremio only reads - all writes are handled by meta-sort.
+    """
+
+    @abstractmethod
+    def connect(self) -> None:
+        """Connect to the storage backend."""
+        pass
+
+    @abstractmethod
+    def disconnect(self) -> None:
+        """Disconnect from the storage backend."""
+        pass
+
+    @abstractmethod
+    def is_connected(self) -> bool:
+        """Check if connected to the storage backend."""
+        pass
+
+    @abstractmethod
+    def get_all_videos(self) -> List[VideoMetadata]:
+        """Get all videos from storage."""
+        pass
+
+    @abstractmethod
+    def get_video_by_hash(self, hash_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its hash ID."""
+        pass
+
+    @abstractmethod
+    def get_videos_by_type(self, video_type: str) -> List[VideoMetadata]:
+        """Get all videos of a specific type (movie, series, anime)."""
+        pass
+
+    @abstractmethod
+    def search_videos(self, query: str) -> List[VideoMetadata]:
+        """Search videos by title."""
+        pass
+
+    @abstractmethod
+    def get_video_by_imdb_id(self, imdb_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its IMDB ID (e.g., 'tt1727587')."""
+        pass
+
+    @abstractmethod
+    def get_video_count(self) -> int:
+        """Get total number of videos in storage."""
+        pass
+
+    def on_video_added(self, callback: Callable[[VideoMetadata], None]) -> None:
+        """Subscribe to video added events (optional)."""
+        pass
+
+    def on_video_removed(self, callback: Callable[[str], None]) -> None:
+        """Subscribe to video removed events (optional)."""
+        pass
+
+    def get_status(self) -> dict:
+        """Get storage status for dashboard."""
+        return {
+            'connected': self.is_connected(),
+            'video_count': self.get_video_count() if self.is_connected() else 0,
+        }

src/storage/redis_storage.py

@@ -0,0 +1,379 @@
+"""
+Redis storage provider for meta-stremio.
+
+Reads video metadata from Redis (written by meta-sort).
+Uses the same key schema as meta-sort for compatibility.
+"""
+from __future__ import annotations
+
+import os
+import json
+from typing import List, Optional
+
+try:
+    import redis
+    REDIS_AVAILABLE = True
+except ImportError:
+    REDIS_AVAILABLE = False
+    redis = None
+
+from .provider import StorageProvider, VideoMetadata
+
+
+# Redis configuration
+REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')
+REDIS_PREFIX = os.environ.get('REDIS_PREFIX', 'meta-sort:')
+
+
+class RedisStorage(StorageProvider):
+    """
+    Redis storage provider.
+
+    Reads video metadata from Redis using the meta-sort key schema:
+    - meta-sort:file:{hashId} -> Hash containing all metadata fields
+    """
+
+    def __init__(self, url: str = None, prefix: str = None):
+        if not REDIS_AVAILABLE:
+            raise ImportError("redis package not installed. Run: pip install redis")
+
+        self._url = url or REDIS_URL
+        self._prefix = prefix or REDIS_PREFIX
+        self._client: Optional[redis.Redis] = None
+        self._connected = False
+
+    def connect(self) -> None:
+        """Connect to Redis."""
+        try:
+            self._client = redis.from_url(self._url, decode_responses=True)
+            # Test connection
+            self._client.ping()
+            self._connected = True
+            print(f"[RedisStorage] Connected to {self._url}")
+        except Exception as e:
+            self._connected = False
+            print(f"[RedisStorage] Failed to connect: {e}")
+            raise
+
+    def disconnect(self) -> None:
+        """Disconnect from Redis."""
+        if self._client:
+            self._client.close()
+            self._client = None
+        self._connected = False
+        print("[RedisStorage] Disconnected")
+
+    def is_connected(self) -> bool:
+        """Check if connected to Redis."""
+        if not self._connected or not self._client:
+            return False
+        try:
+            self._client.ping()
+            return True
+        except Exception:
+            self._connected = False
+            return False
+
+    def get_file_path_by_cid(self, cid: str) -> Optional[str]:
+        """
+        Get the file path for a file by its CID.
+
+        Looks up file:{cid} in Redis and returns the path.
+        Tries 'path' field first, then falls back to 'filePath'.
+        This works for any file including poster images.
+
+        Returns the relative path if found, None otherwise.
+        """
+        if not self.is_connected():
+            return None
+
+        try:
+            key = self._get_file_key(cid)
+            # Try 'path' field first (relative path)
+            path = self._client.hget(key, 'path')
+            if path:
+                return path
+            # Fall back to 'filePath' (full path from meta-sort)
+            file_path = self._client.hget(key, 'filePath')
+            if file_path:
+                # Convert absolute path to relative by removing /files/ prefix
+                if file_path.startswith('/files/'):
+                    return file_path[7:]  # Remove '/files/'
+                return file_path
+            return None
+        except Exception as e:
+            print(f"[RedisStorage] Error getting path for CID {cid}: {e}")
+            return None
+
+    def _get_file_key(self, hash_id: str) -> str:
+        """Get the Redis key for a file hash."""
+        return f"{self._prefix}file:{hash_id}"
+
+    def _parse_video(self, hash_id: str, data: dict) -> Optional[VideoMetadata]:
+        """Parse Redis hash data into VideoMetadata.
+
+        Handles both flat keys and nested keys (e.g., 'plot/eng' for description).
+        meta-sort stores data using a flattened nested key format.
+        """
+        if not data:
+            return None
+
+        # Parse JSON fields
+        audio_tracks = []
+        subtitles = []
+        genres = []
+        studios = []
+
+        if 'audioTracks' in data:
+            try:
+                audio_tracks = json.loads(data['audioTracks'])
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+        if 'subtitles' in data:
+            try:
+                subtitles = json.loads(data['subtitles'])
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+        # Parse genres - can be JSON array or nested keys (genres/0, genres/1, etc.)
+        if 'genres' in data:
+            try:
+                genres = json.loads(data['genres'])
+            except (json.JSONDecodeError, TypeError):
+                if isinstance(data['genres'], str):
+                    genres = [g.strip() for g in data['genres'].split(',') if g.strip()]
+        else:
+            # Try nested key format (genres/0, genres/1, etc.)
+            genres = self._parse_nested_array(data, 'genres')
+
+        # Parse studios - nested keys (studio/0, studio/1, etc.)
+        studios = self._parse_nested_array(data, 'studio')
+
+        # Parse numeric fields
+        def parse_int(val):
+            try:
+                return int(val) if val else None
+            except (ValueError, TypeError):
+                return None
+
+        def parse_float(val):
+            try:
+                return float(val) if val else None
+            except (ValueError, TypeError):
+                return None
+
+        # Get description from nested plot/eng or flat description/plot
+        description = data.get('description')
+        if not description:
+            description = data.get('plot')
+        if not description:
+            # Try nested format: plot/eng, plot/en, etc.
+            for lang in ['eng', 'en', 'english', 'und']:
+                description = data.get(f'plot/{lang}')
+                if description:
+                    break
+
+        # Parse rating (stored as string by TMDBProcessor)
+        rating = parse_float(data.get('rating'))
+
+        # Parse release date
+        release_date = data.get('releasedate', data.get('releaseDate'))
+
+        # Parse episode title (for series episodes)
+        episode_title = data.get('episodeTitle', data.get('episodeName'))
+
+        # Parse tagline
+        tagline = data.get('tagline')
+
+        # Get video/audio codec and resolution from various field formats
+        # meta-sort stores these in fileinfo/streamdetails/video/0/... format
+        video_codec = data.get('videoCodec')
+        if not video_codec:
+            video_codec = data.get('fileinfo/streamdetails/video/0/codec')
+
+        audio_codec = data.get('audioCodec')
+        if not audio_codec:
+            audio_codec = data.get('fileinfo/streamdetails/audio/0/codec')
+
+        width = parse_int(data.get('width'))
+        if not width:
+            width = parse_int(data.get('fileinfo/streamdetails/video/0/width'))
+
+        height = parse_int(data.get('height'))
+        if not height:
+            height = parse_int(data.get('fileinfo/streamdetails/video/0/height'))
+
+        duration = parse_float(data.get('duration'))
+        if not duration:
+            duration = parse_float(data.get('fileinfo/duration'))
+
+        return VideoMetadata(
+            hash_id=hash_id,
+            file_path=data.get('filePath', ''),
+            title=data.get('title', data.get('originalTitle', '')),
+            video_type=data.get('type', data.get('videoType', 'movie')),
+            year=parse_int(data.get('year', data.get('movieYear'))),
+            season=parse_int(data.get('season')),
+            episode=parse_int(data.get('episode')),
+            duration=duration,
+            width=width,
+            height=height,
+            video_codec=video_codec,
+            audio_codec=audio_codec,
+            container=data.get('container'),
+            file_size=parse_int(data.get('fileSize', data.get('sizeByte'))),
+            audio_tracks=audio_tracks,
+            subtitles=subtitles,
+            imdb_id=data.get('imdbId', data.get('imdbid')),
+            tmdb_id=data.get('tmdbId', data.get('tmdbid')),
+            poster=data.get('poster') or data.get('posterUrl'),  # Fallback to posterUrl for backward compat
+            backdrop=data.get('backdrop') or data.get('backdropUrl'),  # Fallback to backdropUrl
+            poster_path=data.get('posterPath'),
+            backdrop_path=data.get('backdropPath'),
+            description=description,
+            genres=genres,
+            episode_title=episode_title,
+            rating=rating,
+            release_date=release_date,
+            tagline=tagline,
+            studios=studios,
+        )
+
+    def _parse_nested_array(self, data: dict, prefix: str) -> list:
+        """Parse nested array format (prefix/0, prefix/1, etc.) into a list."""
+        items = []
+        index = 0
+        while True:
+            key = f'{prefix}/{index}'
+            if key in data:
+                items.append(data[key])
+                index += 1
+            else:
+                break
+        return items
+
+    def get_all_videos(self) -> List[VideoMetadata]:
+        """Get all videos from Redis."""
+        if not self.is_connected():
+            return []
+
+        videos = []
+        pattern = f"{self._prefix}file:*"
+
+        try:
+            for key in self._client.scan_iter(pattern, count=100):
+                # Skip special keys (like file:__index__ which is a set, not a hash)
+                if '__index__' in key or not key.startswith(f"{self._prefix}file:"):
+                    continue
+
+                # Check key type - only process hash keys
+                key_type = self._client.type(key)
+                if key_type != 'hash':
+                    continue
+
+                # Check file type - only include video files
+                file_type = self._client.hget(key, 'fileType')
+                if file_type != 'video':
+                    continue
+
+                # Extract hash_id from key
+                hash_id = key.replace(f"{self._prefix}file:", "")
+                data = self._client.hgetall(key)
+                video = self._parse_video(hash_id, data)
+                if video and video.file_path:
+                    videos.append(video)
+        except Exception as e:
+            print(f"[RedisStorage] Error getting all videos: {e}")
+
+        # Sort by title
+        videos.sort(key=lambda v: v.title.lower())
+        return videos
+
+    def get_video_by_hash(self, hash_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its hash ID."""
+        if not self.is_connected():
+            return None
+
+        try:
+            key = self._get_file_key(hash_id)
+            data = self._client.hgetall(key)
+            return self._parse_video(hash_id, data)
+        except Exception as e:
+            print(f"[RedisStorage] Error getting video {hash_id}: {e}")
+            return None
+
+    def get_videos_by_type(self, video_type: str) -> List[VideoMetadata]:
+        """Get all videos of a specific type."""
+        videos = self.get_all_videos()
+        return [v for v in videos if v.video_type == video_type]
+
+    def search_videos(self, query: str) -> List[VideoMetadata]:
+        """Search videos by title."""
+        if not query:
+            return self.get_all_videos()
+
+        query_lower = query.lower()
+        videos = self.get_all_videos()
+        return [v for v in videos if query_lower in v.title.lower()]
+
+    def get_video_by_imdb_id(self, imdb_id: str) -> Optional[VideoMetadata]:
+        """Get a video by its IMDB ID (e.g., 'tt1727587')."""
+        if not self.is_connected() or not imdb_id:
+            return None
+
+        # Normalize IMDB ID format
+        imdb_id = imdb_id.lower()
+        if not imdb_id.startswith('tt'):
+            imdb_id = f"tt{imdb_id}"
+
+        try:
+            # Scan all videos looking for matching IMDB ID
+            pattern = f"{self._prefix}file:*"
+            for key in self._client.scan_iter(pattern, count=100):
+                # Check imdbid/imdbId field
+                stored_imdb = self._client.hget(key, 'imdbid') or self._client.hget(key, 'imdbId')
+                if stored_imdb and stored_imdb.lower() == imdb_id:
+                    hash_id = key.replace(f"{self._prefix}file:", "")
+                    data = self._client.hgetall(key)
+                    return self._parse_video(hash_id, data)
+
+        except Exception as e:
+            print(f"[RedisStorage] Error finding video by IMDB ID {imdb_id}: {e}")
+
+        return None
+
+    def get_video_count(self) -> int:
+        """Get total number of videos."""
+        if not self.is_connected():
+            return 0
+
+        try:
+            pattern = f"{self._prefix}file:*"
+            count = 0
+            for _ in self._client.scan_iter(pattern, count=100):
+                count += 1
+            return count
+        except Exception as e:
+            print(f"[RedisStorage] Error counting videos: {e}")
+            return 0
+
+    def get_status(self) -> dict:
+        """Get storage status for dashboard."""
+        status = {
+            'type': 'redis',
+            'connected': self.is_connected(),
+            'url': self._url,
+            'prefix': self._prefix,
+            'video_count': 0,
+        }
+
+        if status['connected']:
+            try:
+                status['video_count'] = self.get_video_count()
+                info = self._client.info('memory')
+                status['memory_used'] = info.get('used_memory_human', 'N/A')
+            except Exception:
+                pass
+
+        return status

src/storage/service_discovery.py

@@ -0,0 +1,367 @@
+"""
+Service Discovery for meta-stremio.
+
+Python implementation of the service discovery pattern used in meta-sort.
+Each service registers itself in /meta-core/services/{service-name}.json
+and can discover other registered services.
+
+Features:
+- Service registration with heartbeat
+- Discovery of all running services
+- Stale service detection (60 second threshold)
+- Automatic heartbeat loop
+"""
+from __future__ import annotations
+
+import os
+import json
+import socket
+import threading
+import time
+from dataclasses import dataclass, field, asdict
+from datetime import datetime
+from pathlib import Path
+from typing import List, Optional, Dict, Callable
+from urllib.request import urlopen
+from urllib.error import URLError
+
+
+# Configuration
+META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
+SERVICE_NAME = os.environ.get('SERVICE_NAME', 'meta-stremio')
+SERVICE_VERSION = os.environ.get('SERVICE_VERSION', '1.0.0')
+BASE_URL = os.environ.get('BASE_URL', '')
+
+
+@dataclass
+class ServiceInfo:
+    """Service registration info stored in JSON file."""
+    name: str
+    version: str
+    api: str  # HTTP API URL (e.g., http://10.0.1.50:7000)
+    status: str = 'starting'  # starting | running | stopping | stopped
+    pid: int = 0
+    hostname: str = ''
+    startedAt: str = ''
+    lastHeartbeat: str = ''
+    capabilities: List[str] = field(default_factory=list)
+    endpoints: Dict[str, str] = field(default_factory=dict)
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary for JSON serialization."""
+        return asdict(self)
+
+
+class ServiceDiscovery:
+    """
+    Service Discovery using shared filesystem.
+
+    Each service registers itself in /meta-core/services/{service-name}.json
+    and can discover other registered services.
+    """
+
+    def __init__(
+        self,
+        meta_core_path: str = None,
+        service_name: str = None,
+        version: str = None,
+        api_url: str = None,
+        base_url: str = None,
+        capabilities: List[str] = None,
+        endpoints: Dict[str, str] = None,
+        heartbeat_interval: float = 30.0,
+        stale_threshold: float = 60.0
+    ):
+        self.meta_core_path = meta_core_path or META_CORE_PATH
+        self.service_name = service_name or SERVICE_NAME
+        self.version = version or SERVICE_VERSION
+        # Use base_url if provided, then api_url, then env BASE_URL, then auto-detect
+        self.api_url = base_url or api_url or BASE_URL or self._get_default_api_url()
+        self.capabilities = capabilities or ['streaming', 'transcoding', 'hls']
+        self.endpoints = endpoints or {
+            'health': '/health',
+            'manifest': '/manifest.json',
+            'dashboard': '/',
+            'api': '/api',
+        }
+        self.heartbeat_interval = heartbeat_interval
+        self.stale_threshold = stale_threshold
+
+        self.services_dir = os.path.join(self.meta_core_path, 'services')
+        self.service_file_path = os.path.join(self.services_dir, f'{self.service_name}.json')
+
+        self._heartbeat_thread: Optional[threading.Thread] = None
+        self._stop_event = threading.Event()
+        self._is_started = False
+
+    def _get_default_api_url(self) -> str:
+        """Get default API URL using local IP and PORT."""
+        port = os.environ.get('PORT', '7000')
+        ip = self._get_local_ip()
+        return f'http://{ip}:{port}'
+
+    def _get_local_ip(self) -> str:
+        """Get the local machine's IP address."""
+        try:
+            # Create a socket to determine the outbound IP
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            s.connect(('8.8.8.8', 80))
+            ip = s.getsockname()[0]
+            s.close()
+            return ip
+        except Exception:
+            return socket.gethostname() or 'localhost'
+
+    def _ensure_services_dir(self) -> None:
+        """Ensure services directory exists."""
+        Path(self.services_dir).mkdir(parents=True, exist_ok=True)
+
+    def _build_service_info(self, status: str = 'starting') -> ServiceInfo:
+        """Build service info for this service."""
+        now = datetime.utcnow().isoformat() + 'Z'
+        return ServiceInfo(
+            name=self.service_name,
+            version=self.version,
+            api=self.api_url,
+            status=status,
+            pid=os.getpid(),
+            hostname=socket.gethostname(),
+            startedAt=now,
+            lastHeartbeat=now,
+            capabilities=self.capabilities,
+            endpoints=self.endpoints
+        )
+
+    def register(self) -> None:
+        """Register this service."""
+        self._ensure_services_dir()
+
+        info = self._build_service_info('starting')
+
+        with open(self.service_file_path, 'w') as f:
+            json.dump(info.to_dict(), f, indent=2)
+
+        print(f'[ServiceDiscovery] Registered {self.service_name}')
+
+    def update_status(self, status: str) -> None:
+        """Update service status."""
+        try:
+            with open(self.service_file_path, 'r') as f:
+                info = json.load(f)
+
+            info['status'] = status
+            info['lastHeartbeat'] = datetime.utcnow().isoformat() + 'Z'
+
+            with open(self.service_file_path, 'w') as f:
+                json.dump(info, f, indent=2)
+
+        except Exception as e:
+            print(f'[ServiceDiscovery] Failed to update status: {e}')
+
+    def heartbeat(self) -> None:
+        """Send heartbeat (update lastHeartbeat timestamp)."""
+        try:
+            with open(self.service_file_path, 'r') as f:
+                info = json.load(f)
+
+            info['lastHeartbeat'] = datetime.utcnow().isoformat() + 'Z'
+
+            with open(self.service_file_path, 'w') as f:
+                json.dump(info, f, indent=2)
+
+        except FileNotFoundError:
+            # If file was deleted, re-register
+            self.register()
+            self.update_status('running')
+        except Exception as e:
+            print(f'[ServiceDiscovery] Heartbeat failed: {e}')
+
+    def _heartbeat_loop(self) -> None:
+        """Background heartbeat loop."""
+        while not self._stop_event.is_set():
+            try:
+                self.heartbeat()
+            except Exception as e:
+                print(f'[ServiceDiscovery] Heartbeat error: {e}')
+
+            self._stop_event.wait(self.heartbeat_interval)
+
+    def start_heartbeat(self) -> None:
+        """Start heartbeat loop in background thread."""
+        if self._heartbeat_thread and self._heartbeat_thread.is_alive():
+            return
+
+        self._stop_event.clear()
+        self._heartbeat_thread = threading.Thread(
+            target=self._heartbeat_loop,
+            daemon=True,
+            name='service-heartbeat'
+        )
+        self._heartbeat_thread.start()
+
+    def stop_heartbeat(self) -> None:
+        """Stop heartbeat loop."""
+        self._stop_event.set()
+        if self._heartbeat_thread and self._heartbeat_thread.is_alive():
+            self._heartbeat_thread.join(timeout=2.0)
+
+    def unregister(self) -> None:
+        """Unregister this service (on shutdown)."""
+        self.stop_heartbeat()
+
+        try:
+            self.update_status('stopped')
+            print(f'[ServiceDiscovery] Unregistered {self.service_name}')
+        except Exception as e:
+            print(f'[ServiceDiscovery] Failed to unregister: {e}')
+
+    # ========================================================================
+    # Discovery Methods
+    # ========================================================================
+
+    def _is_service_stale(self, info: dict) -> bool:
+        """Check if a service is stale based on lastHeartbeat."""
+        try:
+            last_heartbeat_str = info.get('lastHeartbeat', '')
+            if not last_heartbeat_str:
+                return True
+
+            # Parse ISO timestamp
+            last_heartbeat_str = last_heartbeat_str.rstrip('Z')
+            last_heartbeat = datetime.fromisoformat(last_heartbeat_str)
+            age = (datetime.utcnow() - last_heartbeat).total_seconds()
+
+            return age > self.stale_threshold
+        except Exception:
+            return True
+
+    def discover_service(self, name: str) -> Optional[dict]:
+        """Discover a service by name."""
+        service_path = os.path.join(self.services_dir, f'{name}.json')
+
+        try:
+            with open(service_path, 'r') as f:
+                info = json.load(f)
+
+            # Check if service is alive
+            if self._is_service_stale(info):
+                print(f'[ServiceDiscovery] Service {name} appears stale')
+                return None
+
+            return info
+
+        except FileNotFoundError:
+            return None
+        except Exception as e:
+            print(f'[ServiceDiscovery] Error discovering {name}: {e}')
+            return None
+
+    def discover_all_services(self) -> List[dict]:
+        """Discover all registered services."""
+        services = []
+
+        try:
+            if not os.path.exists(self.services_dir):
+                return services
+
+            for filename in os.listdir(self.services_dir):
+                if not filename.endswith('.json'):
+                    continue
+
+                name = filename.replace('.json', '')
+                service = self.discover_service(name)
+
+                if service and service.get('status') == 'running':
+                    services.append(service)
+
+        except Exception as e:
+            print(f'[ServiceDiscovery] Error discovering services: {e}')
+
+        return services
+
+    def is_service_healthy(self, name: str) -> bool:
+        """Check if a service is healthy via HTTP ping."""
+        service = self.discover_service(name)
+        if not service:
+            return False
+
+        health_endpoint = service.get('endpoints', {}).get('health', '/health')
+        url = f"{service['api']}{health_endpoint}"
+
+        try:
+            with urlopen(url, timeout=5) as response:
+                return response.status == 200
+        except Exception:
+            return False
+
+    # ========================================================================
+    # Lifecycle
+    # ========================================================================
+
+    def start(self) -> None:
+        """Full startup sequence."""
+        if self._is_started:
+            return
+
+        # Try to register, but don't fail if filesystem is read-only
+        # This allows discovery-only mode when we can't write
+        can_register = os.access(self.services_dir, os.W_OK) if os.path.exists(self.services_dir) else False
+
+        if can_register:
+            try:
+                self.register()
+                self.update_status('running')
+                self.start_heartbeat()
+                print(f'[ServiceDiscovery] Started {self.service_name} at {self.api_url}')
+            except Exception as e:
+                print(f'[ServiceDiscovery] Registration failed (read-only?): {e}')
+                print(f'[ServiceDiscovery] Running in discovery-only mode')
+        else:
+            print(f'[ServiceDiscovery] Services directory not writable, running in discovery-only mode')
+
+        self._is_started = True
+
+    def stop(self) -> None:
+        """Full shutdown sequence."""
+        if not self._is_started:
+            return
+
+        self.unregister()
+        self._is_started = False
+
+    # ========================================================================
+    # Getters
+    # ========================================================================
+
+    def get_service_file_path(self) -> str:
+        return self.service_file_path
+
+    def get_services_dir(self) -> str:
+        return self.services_dir
+
+    def is_started(self) -> bool:
+        return self._is_started
+
+
+# Singleton instance
+_service_discovery: Optional[ServiceDiscovery] = None
+
+
+def get_service_discovery() -> ServiceDiscovery:
+    """Get or create the service discovery singleton."""
+    global _service_discovery
+    if _service_discovery is None:
+        _service_discovery = ServiceDiscovery()
+    return _service_discovery
+
+
+def init_service_discovery(
+    api_url: str = None,
+    base_url: str = None,
+    **kwargs
+) -> ServiceDiscovery:
+    """Initialize and start the service discovery singleton."""
+    global _service_discovery
+    _service_discovery = ServiceDiscovery(api_url=api_url, base_url=base_url, **kwargs)
+    _service_discovery.start()
+    return _service_discovery

src/transcoder.py

@@ -0,0 +1,863 @@
+#!/usr/bin/env python3
+"""
+HLS Live Transcoder Engine
+
+Transcodes video segments on-demand using FFmpeg with adaptive quality.
+
+Design: Single-threaded transcoding with multi-ahead prefetch.
+- One FFmpeg process gets 100% CPU for fastest segment time
+- After serving segment N, prefetch N+1..N+4 in background
+- No parallel transcoding (would split CPU and increase latency)
+
+Adapted from SegmentPlayer's transcoder for standalone Stremio addon use.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+import hashlib
+import threading
+import time
+import json
+import re
+from typing import Optional
+
+# Configuration
+MEDIA_DIR = os.environ.get('MEDIA_DIR', '/data/media')
+CACHE_DIR = os.environ.get('CACHE_DIR', '/data/cache')
+SEGMENT_DURATION = int(os.environ.get('SEGMENT_DURATION', '4'))
+PREFETCH_SEGMENTS = int(os.environ.get('PREFETCH_SEGMENTS', '4'))
+
+os.makedirs(CACHE_DIR, exist_ok=True)
+
+# Resolution presets: (width, height, crf)
+RESOLUTIONS = {
+    'original': (None, None, 23),
+    '1080p': (1920, 1080, 23),
+    '720p': (1280, 720, 24),
+    '480p': (854, 480, 25),
+    '360p': (640, 360, 26),
+}
+
+# x264 presets ordered from fastest to slowest
+X264_PRESETS = ['ultrafast', 'superfast', 'veryfast', 'faster', 'fast', 'medium', 'slow', 'slower', 'veryslow']
+
+# Subtitle codecs that cannot be converted to WebVTT (image-based)
+UNSUPPORTED_SUBTITLE_CODECS = {'hdmv_pgs_subtitle', 'dvd_subtitle', 'dvb_subtitle', 'xsub'}
+
+
+class AdaptiveQuality:
+    """
+    Coordinated adaptive quality control using both preset and CRF.
+
+    Strategy - PRIORITIZE LOW CRF (quality over compression efficiency):
+    - CRF directly affects output quality (lower = better visual quality)
+    - Preset affects encoding speed (faster = less CPU, slightly worse compression)
+    - For streaming, visual quality matters more than file size
+    - So: keep CRF low, use faster presets when needed
+
+    Target: 60-80% of segment duration (transcode ratio)
+    - Below 60%: We have headroom, can increase quality
+    - 60-80%: Sweet spot, maintain current settings
+    - Above 80%: Too slow, decrease quality (preset first!)
+    - Above 100%: EMERGENCY, drop to fastest settings
+
+    Coordination rules:
+    - DECREASE quality: Drop preset first (faster encoding), CRF only as last resort
+    - INCREASE quality: Decrease CRF first (better quality), then preset (capped at 'medium')
+    - Max preset is 'medium' - slower presets waste CPU for minimal streaming benefit
+    """
+
+    MAX_PRESET_INDEX = 5  # 'medium' - slower presets not worth it for streaming
+
+    def __init__(self, initial_preset: str = 'fast'):
+        self._lock = threading.Lock()
+
+        # Preset state (capped at medium)
+        initial_index = X264_PRESETS.index(initial_preset) if initial_preset in X264_PRESETS else 4
+        self._preset_index = min(initial_index, self.MAX_PRESET_INDEX)
+
+        # CRF state
+        self._crf_offset = 0  # 0 to 7
+
+        # Shared state
+        self._recent_ratios: list[float] = []
+        self._window_size = 5
+        self._consecutive_good_signals = 0
+        self._last_change_time = 0.0
+
+        # Thresholds
+        self._target_min = 60.0
+        self._target_max = 80.0
+
+        # Stats
+        self._preset_ups = 0
+        self._preset_downs = 0
+        self._crf_ups = 0
+        self._crf_downs = 0
+
+    @property
+    def preset(self) -> str:
+        with self._lock:
+            return X264_PRESETS[self._preset_index]
+
+    def get_crf(self, base_crf: int) -> int:
+        with self._lock:
+            return min(base_crf + self._crf_offset, 30)
+
+    def record_transcode(self, elapsed: float) -> dict | None:
+        """Record transcode time and adjust quality settings."""
+        last_ratio = (elapsed / SEGMENT_DURATION) * 100
+
+        with self._lock:
+            self._recent_ratios.append(last_ratio)
+            if len(self._recent_ratios) > self._window_size:
+                self._recent_ratios.pop(0)
+
+            avg_ratio = sum(self._recent_ratios) / len(self._recent_ratios)
+            current_time = time.time()
+
+            old_preset = X264_PRESETS[self._preset_index]
+            old_crf = self._crf_offset
+
+            # EMERGENCY: ratio > 100%
+            if last_ratio > 100:
+                if self._preset_index > 0 or self._crf_offset < 7:
+                    self._preset_index = 0  # ultrafast
+                    self._crf_offset = 7
+                    self._consecutive_good_signals = 0
+                    self._last_change_time = current_time
+                    self._recent_ratios.clear()
+                    self._preset_downs += 1
+                    self._crf_ups += 1
+                    print(f"[AdaptiveQuality] EMERGENCY {last_ratio:.1f}% → ultrafast, CRF +7")
+                    return {'preset': 'ultrafast', 'crf_offset': 7, 'emergency': True}
+
+            # DECREASE QUALITY: last ratio > 80%
+            if last_ratio > self._target_max:
+                self._consecutive_good_signals = 0
+
+                if self._preset_index > 0:
+                    self._preset_index -= 1
+                    self._preset_downs += 1
+                    self._last_change_time = current_time
+                    new_preset = X264_PRESETS[self._preset_index]
+                    print(f"[AdaptiveQuality] Last {last_ratio:.1f}% > {self._target_max}% → {old_preset} → {new_preset}")
+                    return {'preset': new_preset}
+                elif self._crf_offset < 7:
+                    self._crf_offset = min(self._crf_offset + 2, 7)
+                    self._crf_ups += 1
+                    self._last_change_time = current_time
+                    print(f"[AdaptiveQuality] Last {last_ratio:.1f}% > {self._target_max}% (ultrafast) → CRF +{old_crf} → +{self._crf_offset}")
+                    return {'crf_offset': self._crf_offset}
+
+            # INCREASE QUALITY: avg ratio < 60%
+            elif avg_ratio < self._target_min:
+                self._consecutive_good_signals += 1
+
+                if self._consecutive_good_signals >= 3 and current_time - self._last_change_time >= 5.0:
+                    if self._crf_offset > 0:
+                        self._crf_offset -= 1
+                        self._crf_downs += 1
+                        self._consecutive_good_signals = 0
+                        self._last_change_time = current_time
+                        print(f"[AdaptiveQuality] Avg {avg_ratio:.1f}% < {self._target_min}% → CRF +{old_crf} → +{self._crf_offset}")
+                        return {'crf_offset': self._crf_offset}
+                    elif self._preset_index < self.MAX_PRESET_INDEX:
+                        self._preset_index += 1
+                        self._preset_ups += 1
+                        self._consecutive_good_signals = 0
+                        self._last_change_time = current_time
+                        new_preset = X264_PRESETS[self._preset_index]
+                        print(f"[AdaptiveQuality] Avg {avg_ratio:.1f}% < {self._target_min}% → {old_preset} → {new_preset}")
+                        return {'preset': new_preset}
+            else:
+                self._consecutive_good_signals = 0
+
+            return None
+
+    def get_stats(self) -> dict:
+        with self._lock:
+            avg_ratio = sum(self._recent_ratios) / len(self._recent_ratios) if self._recent_ratios else 0
+            last_ratio = self._recent_ratios[-1] if self._recent_ratios else 0
+            return {
+                'preset': X264_PRESETS[self._preset_index],
+                'crf_offset': self._crf_offset,
+                'target_range': f"{self._target_min:.0f}-{self._target_max:.0f}%",
+                'avg_ratio': round(avg_ratio, 1),
+                'last_ratio': round(last_ratio, 1),
+                'preset_adjustments': {'up': self._preset_ups, 'down': self._preset_downs},
+                'crf_adjustments': {'up': self._crf_ups, 'down': self._crf_downs},
+            }
+
+
+class SegmentManager:
+    """
+    Manages segment transcoding with single-threaded execution.
+
+    Design rationale:
+    - Single FFmpeg gets 100% CPU → fastest per-segment time
+    - Multi-ahead prefetch uses idle time while user watches current segment
+    - Multiple requests for same segment wait on shared Event
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._in_progress: dict[str, threading.Event] = {}
+        self._prefetch_thread: threading.Thread | None = None
+        self._prefetch_queue: list[tuple[str, callable]] = []
+
+        # Metrics
+        self.total_segments = 0
+        self.cache_hits = 0
+        self.total_transcode_time = 0.0
+        self.last_segment_time = 0.0
+        self.min_segment_time = float('inf')
+        self.max_segment_time = 0.0
+        self._transcode_times: list[float] = []
+
+        # Current file codec info
+        self.current_video_codec: str | None = None
+        self.current_audio_codec: str | None = None
+        self.current_file: str | None = None
+
+    def get_segment(self, key: str, transcode_fn) -> str | None:
+        """Get a segment, transcoding if necessary."""
+        with self._lock:
+            self.total_segments += 1
+
+            if key in self._in_progress:
+                event = self._in_progress[key]
+                wait_for_other = True
+            else:
+                event = threading.Event()
+                self._in_progress[key] = event
+                wait_for_other = False
+
+        if wait_for_other:
+            completed = event.wait(timeout=120)
+            if completed:
+                return transcode_fn()
+            else:
+                return None
+
+        try:
+            result = transcode_fn()
+            return result
+        finally:
+            with self._lock:
+                self._in_progress.pop(key, None)
+            event.set()
+
+    def schedule_prefetch(self, key: str, transcode_fn):
+        """Schedule prefetch for a segment."""
+        with self._lock:
+            if key in self._in_progress:
+                return
+            if any(k == key for k, _ in self._prefetch_queue):
+                return
+
+            self._prefetch_queue.append((key, transcode_fn))
+
+            if not self._prefetch_thread or not self._prefetch_thread.is_alive():
+                self._prefetch_thread = threading.Thread(
+                    target=self._process_prefetch_queue,
+                    daemon=True
+                )
+                self._prefetch_thread.start()
+
+    def _process_prefetch_queue(self):
+        """Process prefetch queue sequentially."""
+        while True:
+            with self._lock:
+                if not self._prefetch_queue:
+                    return
+
+                key, transcode_fn = self._prefetch_queue.pop(0)
+                if key in self._in_progress:
+                    continue
+
+                event = threading.Event()
+                self._in_progress[key] = event
+
+            try:
+                transcode_fn()
+            finally:
+                with self._lock:
+                    self._in_progress.pop(key, None)
+                event.set()
+
+    def is_in_progress(self, key: str) -> bool:
+        with self._lock:
+            return key in self._in_progress
+
+    def record_cache_hit(self):
+        with self._lock:
+            self.total_segments += 1
+            self.cache_hits += 1
+
+    def record_transcode_time(self, elapsed: float):
+        with self._lock:
+            self.total_transcode_time += elapsed
+            self.last_segment_time = elapsed
+            if elapsed < self.min_segment_time:
+                self.min_segment_time = elapsed
+            if elapsed > self.max_segment_time:
+                self.max_segment_time = elapsed
+            self._transcode_times.append(elapsed)
+            if len(self._transcode_times) > 100:
+                self._transcode_times.pop(0)
+
+    def set_codec_info(self, video_codec: str | None, audio_codec: str | None, filepath: str | None = None):
+        with self._lock:
+            self.current_video_codec = video_codec
+            self.current_audio_codec = audio_codec
+            if filepath:
+                self.current_file = filepath
+
+    def get_metrics(self) -> dict:
+        with self._lock:
+            avg_time = sum(self._transcode_times) / len(self._transcode_times) if self._transcode_times else 0
+            last_ratio = (self.last_segment_time / SEGMENT_DURATION * 100) if self.last_segment_time > 0 else 0
+            avg_ratio = (avg_time / SEGMENT_DURATION * 100) if avg_time > 0 else 0
+            return {
+                'total_segments': self.total_segments,
+                'cache_hits': self.cache_hits,
+                'cache_hit_rate': round((self.cache_hits / self.total_segments * 100), 1) if self.total_segments > 0 else 0,
+                'prefetch_queue_size': len(self._prefetch_queue),
+                'total_transcode_time': round(self.total_transcode_time, 2),
+                'last_segment_time': round(self.last_segment_time, 2),
+                'avg_segment_time': round(avg_time, 2),
+                'segment_duration': SEGMENT_DURATION,
+                'transcode_ratio_last': round(last_ratio, 1),
+                'transcode_ratio_avg': round(avg_ratio, 1),
+                'video_codec': self.current_video_codec,
+                'audio_codec': self.current_audio_codec,
+                'current_file': self.current_file,
+            }
+
+    def reset_metrics(self):
+        with self._lock:
+            self.total_segments = 0
+            self.cache_hits = 0
+            self.total_transcode_time = 0.0
+            self.last_segment_time = 0.0
+            self.min_segment_time = float('inf')
+            self.max_segment_time = 0.0
+            self._transcode_times.clear()
+
+
+class SubtitleManager:
+    """Manages subtitle extraction with background processing."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._extractions: dict[str, tuple[threading.Event, list]] = {}
+
+    def get_subtitle(self, key: str, filepath: str, file_hash: str, sub_index: int,
+                     info: dict = None, timeout: float = 300) -> tuple[str | None, str | None]:
+        """Get subtitle content, extracting if necessary."""
+        cache_dir = os.path.join(CACHE_DIR, file_hash)
+        vtt_file = os.path.join(cache_dir, f"subtitle_{sub_index}.vtt")
+        error_file = os.path.join(cache_dir, f"subtitle_{sub_index}.error")
+
+        # Fast path: already cached
+        if os.path.exists(vtt_file):
+            with open(vtt_file, 'r', encoding='utf-8') as f:
+                return f.read(), None
+        if os.path.exists(error_file):
+            with open(error_file, 'r', encoding='utf-8') as f:
+                return None, f.read()
+
+        with self._lock:
+            if os.path.exists(vtt_file):
+                with open(vtt_file, 'r', encoding='utf-8') as f:
+                    return f.read(), None
+            if os.path.exists(error_file):
+                with open(error_file, 'r', encoding='utf-8') as f:
+                    return None, f.read()
+
+            if key in self._extractions:
+                event, result_holder = self._extractions[key]
+            else:
+                event = threading.Event()
+                result_holder = [None]
+                self._extractions[key] = (event, result_holder)
+
+                thread = threading.Thread(
+                    target=self._extract_background,
+                    args=(key, filepath, file_hash, sub_index, info, event, result_holder),
+                    daemon=True
+                )
+                thread.start()
+
+        completed = event.wait(timeout=timeout)
+
+        if completed:
+            if result_holder[0] is not None:
+                return result_holder[0]
+            if os.path.exists(vtt_file):
+                with open(vtt_file, 'r', encoding='utf-8') as f:
+                    return f.read(), None
+            if os.path.exists(error_file):
+                with open(error_file, 'r', encoding='utf-8') as f:
+                    return None, f.read()
+            return None, "Extraction completed but result not found"
+
+        return None, f"Extraction timed out after {timeout}s"
+
+    def _extract_background(self, key: str, filepath: str, file_hash: str, sub_index: int,
+                           info: dict, event: threading.Event, result_holder: list):
+        try:
+            result = self._do_extract(filepath, file_hash, sub_index, info)
+            result_holder[0] = result
+        except Exception as e:
+            result_holder[0] = (None, f"Extraction thread error: {e}")
+        finally:
+            event.set()
+            def cleanup():
+                time.sleep(30)
+                with self._lock:
+                    self._extractions.pop(key, None)
+            threading.Thread(target=cleanup, daemon=True).start()
+
+    def _do_extract(self, filepath: str, file_hash: str, sub_index: int,
+                    info: dict = None) -> tuple[str | None, str | None]:
+        cache_dir = os.path.join(CACHE_DIR, file_hash)
+        os.makedirs(cache_dir, exist_ok=True)
+
+        vtt_file = os.path.join(cache_dir, f"subtitle_{sub_index}.vtt")
+        temp_file = vtt_file + '.tmp'
+        error_file = os.path.join(cache_dir, f"subtitle_{sub_index}.error")
+
+        if os.path.exists(vtt_file):
+            with open(vtt_file, 'r', encoding='utf-8') as f:
+                return f.read(), None
+        if os.path.exists(error_file):
+            with open(error_file, 'r', encoding='utf-8') as f:
+                return None, f.read()
+
+        # Get subtitle stream info
+        codec = None
+        track_index = None
+        if info:
+            subtitle_streams = [s for s in info.get('streams', []) if s.get('codec_type') == 'subtitle']
+            if sub_index < len(subtitle_streams):
+                stream = subtitle_streams[sub_index]
+                codec = stream.get('codec_name', '')
+                track_index = stream.get('index')
+                if codec in UNSUPPORTED_SUBTITLE_CODECS:
+                    error_msg = f"Subtitle format '{codec}' is image-based and cannot be converted to WebVTT"
+                    with open(error_file, 'w', encoding='utf-8') as f:
+                        f.write(error_msg)
+                    return None, error_msg
+
+        try:
+            print(f"[Subtitle {sub_index}] Extracting from {os.path.basename(filepath)}...")
+            is_mkv = filepath.lower().endswith(('.mkv', '.mka', '.mks'))
+
+            # Determine intermediate format based on codec
+            codec_ext_map = {
+                'subrip': 'srt', 'srt': 'srt',
+                'ass': 'ass', 'ssa': 'ssa',
+                'webvtt': 'vtt', 'mov_text': 'srt'
+            }
+            intermediate_ext = codec_ext_map.get(codec, 'srt') if codec else 'srt'
+            intermediate_file = os.path.join(cache_dir, f"subtitle_{sub_index}.{intermediate_ext}")
+
+            extracted = False
+
+            # Method 1: Try mkvextract for MKV files (fastest)
+            if is_mkv and track_index is not None:
+                try:
+                    result = subprocess.run(
+                        ['mkvextract', 'tracks', filepath, f'{track_index}:{intermediate_file}'],
+                        capture_output=True,
+                        timeout=120
+                    )
+                    if result.returncode == 0 and os.path.exists(intermediate_file):
+                        extracted = True
+                        print(f"[Subtitle {sub_index}] Extracted via mkvextract")
+                except FileNotFoundError:
+                    pass  # mkvextract not installed, fall back to ffmpeg
+                except Exception as e:
+                    print(f"[Subtitle {sub_index}] mkvextract failed: {e}, trying ffmpeg")
+
+            # Method 2: FFmpeg copy codec (fast demux, no re-encoding)
+            if not extracted:
+                result = subprocess.run(
+                    ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error',
+                     '-probesize', '1M', '-analyzeduration', '1M',
+                     '-i', filepath,
+                     '-map', f'0:s:{sub_index}',
+                     '-c:s', 'copy', intermediate_file],
+                    capture_output=True,
+                    timeout=300
+                )
+                if result.returncode == 0 and os.path.exists(intermediate_file):
+                    extracted = True
+                    print(f"[Subtitle {sub_index}] Extracted via ffmpeg copy")
+
+            # Convert to WebVTT if needed
+            if extracted and os.path.exists(intermediate_file):
+                if intermediate_ext == 'vtt':
+                    # Already VTT, just rename
+                    os.rename(intermediate_file, temp_file)
+                else:
+                    # Convert to WebVTT
+                    result = subprocess.run(
+                        ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error',
+                         '-i', intermediate_file,
+                         '-c:s', 'webvtt', '-f', 'webvtt', temp_file],
+                        capture_output=True,
+                        timeout=60
+                    )
+                    # Clean up intermediate file
+                    if os.path.exists(intermediate_file):
+                        os.remove(intermediate_file)
+
+                    if result.returncode != 0:
+                        stderr = result.stderr.decode('utf-8', errors='replace') if result.stderr else 'Unknown error'
+                        error_msg = f"VTT conversion failed: {stderr[:200]}"
+                        with open(error_file, 'w', encoding='utf-8') as f:
+                            f.write(error_msg)
+                        return None, error_msg
+
+            # Validate and finalize
+            if os.path.exists(temp_file):
+                file_size = os.path.getsize(temp_file)
+                if file_size > 10:
+                    os.rename(temp_file, vtt_file)
+                    print(f"[Subtitle {sub_index}] Extraction complete ({file_size} bytes)")
+                    with open(vtt_file, 'r', encoding='utf-8') as f:
+                        return f.read(), None
+                else:
+                    error_msg = "Extraction produced empty VTT file"
+                    os.remove(temp_file)
+            else:
+                error_msg = "Extraction failed - no output file"
+
+            with open(error_file, 'w', encoding='utf-8') as f:
+                f.write(error_msg)
+            return None, error_msg
+
+        except subprocess.TimeoutExpired:
+            error_msg = "Extraction timed out"
+            with open(error_file, 'w', encoding='utf-8') as f:
+                f.write(error_msg)
+            return None, error_msg
+        except Exception as e:
+            error_msg = f"Unexpected error: {str(e)}"
+            with open(error_file, 'w', encoding='utf-8') as f:
+                f.write(error_msg)
+            return None, error_msg
+
+
+# Global managers
+adaptive_quality = AdaptiveQuality(initial_preset='fast')
+segment_manager = SegmentManager()
+subtitle_manager = SubtitleManager()
+
+
+def get_file_hash(filepath: str) -> str:
+    """Generate cache key from filepath."""
+    return hashlib.md5(filepath.encode()).hexdigest()[:16]
+
+
+def get_video_info(filepath: str) -> dict | None:
+    """Get video metadata using ffprobe."""
+    full_path = os.path.join(MEDIA_DIR, filepath) if not filepath.startswith('/') else filepath
+    try:
+        result = subprocess.run(
+            ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', '-show_streams', full_path],
+            capture_output=True, text=True, timeout=30
+        )
+        return json.loads(result.stdout) if result.returncode == 0 else None
+    except Exception:
+        return None
+
+
+def extract_codecs(info: dict) -> tuple[str | None, str | None]:
+    """Extract video and audio codec names from ffprobe info."""
+    streams = info.get('streams', [])
+    video_codec = None
+    audio_codec = None
+
+    for stream in streams:
+        codec_type = stream.get('codec_type')
+        if codec_type == 'video' and video_codec is None:
+            video_codec = stream.get('codec_name')
+        elif codec_type == 'audio' and audio_codec is None:
+            audio_codec = stream.get('codec_name')
+
+    return video_codec, audio_codec
+
+
+def get_segment_path(file_hash: str, audio: int, resolution: str, segment: int) -> str:
+    """Get cache path for a muxed video+audio segment."""
+    return os.path.join(CACHE_DIR, file_hash, f"seg_a{audio}_{resolution}_{segment:05d}.ts")
+
+
+def transcode_segment(filepath: str, file_hash: str, audio: int, resolution: str, segment: int) -> str | None:
+    """
+    Transcode a single segment using FFmpeg.
+
+    Optimized for speed with adaptive quality:
+    - adaptive preset: auto-adjusts based on transcode ratio (target 60-80%)
+    - threads 0: use all CPU cores
+    - tune zerolatency: reduce encoding latency
+    - Muxed video+audio: single segment contains both streams
+    """
+    cache_dir = os.path.join(CACHE_DIR, file_hash)
+    os.makedirs(cache_dir, exist_ok=True)
+
+    output = get_segment_path(file_hash, audio, resolution, segment)
+
+    # Already cached?
+    if os.path.exists(output):
+        return output
+
+    full_path = os.path.join(MEDIA_DIR, filepath) if not filepath.startswith('/') else filepath
+    start_offset = segment * SEGMENT_DURATION
+    res_preset = RESOLUTIONS.get(resolution, RESOLUTIONS['original'])
+    width, height, base_crf = res_preset
+
+    # Get current adaptive preset and CRF
+    current_preset = adaptive_quality.preset
+    current_crf = adaptive_quality.get_crf(base_crf)
+
+    cpu_count = os.cpu_count() or 8
+
+    cmd = [
+        'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error',
+        '-threads', str(cpu_count),
+        '-ss', str(start_offset),
+        '-i', full_path,
+        '-t', str(SEGMENT_DURATION),
+        '-map', '0:v:0',
+        '-map', f'0:a:{audio}?',  # ? = optional, don't fail if audio track doesn't exist
+        # Video: H.264 with adaptive quality
+        '-c:v', 'libx264',
+        '-preset', current_preset,
+        '-tune', 'zerolatency',
+        '-crf', str(current_crf),
+        '-pix_fmt', 'yuv420p',
+        '-x264-params', f'threads={cpu_count}:lookahead_threads={min(cpu_count, 8)}',
+        '-force_key_frames', 'expr:gte(t,0)',
+        # Audio: AAC (muxed with video)
+        '-c:a', 'aac', '-b:a', '128k', '-ac', '2',
+        # Output format
+        '-f', 'mpegts',
+        '-mpegts_copyts', '1',
+        '-output_ts_offset', str(start_offset),
+    ]
+
+    if width and height:
+        cmd.extend(['-vf', f'scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2'])
+
+    cmd.append(output)
+
+    start_time = time.time()
+    try:
+        result = subprocess.run(cmd, capture_output=True, timeout=120)
+        elapsed = time.time() - start_time
+        if result.returncode == 0 and os.path.exists(output):
+            segment_manager.record_transcode_time(elapsed)
+            adaptive_quality.record_transcode(elapsed)
+            return output
+        print(f"FFmpeg error: {result.stderr.decode()[-500:]}")
+    except Exception as e:
+        print(f"Transcode error: {e}")
+
+    return None
+
+
+def get_or_transcode_segment(filepath: str, file_hash: str, audio: int, resolution: str,
+                              segment: int, info: dict) -> bytes | None:
+    """Get segment data, transcoding if necessary."""
+    output = get_segment_path(file_hash, audio, resolution, segment)
+    key = f"{file_hash}:{audio}:{resolution}:{segment}"
+
+    # Check if transcode in progress
+    if segment_manager.is_in_progress(key):
+        result = segment_manager.get_segment(
+            key,
+            lambda: transcode_segment(filepath, file_hash, audio, resolution, segment)
+        )
+        if result and os.path.exists(result):
+            trigger_prefetch(filepath, file_hash, audio, resolution, segment, info)
+            with open(result, 'rb') as f:
+                return f.read()
+        return None
+
+    # Fast path: cached
+    if os.path.exists(output):
+        segment_manager.record_cache_hit()
+        trigger_prefetch(filepath, file_hash, audio, resolution, segment, info)
+        with open(output, 'rb') as f:
+            return f.read()
+
+    # Need to transcode
+    result = segment_manager.get_segment(
+        key,
+        lambda: transcode_segment(filepath, file_hash, audio, resolution, segment)
+    )
+
+    if result and os.path.exists(result):
+        trigger_prefetch(filepath, file_hash, audio, resolution, segment, info)
+        with open(result, 'rb') as f:
+            return f.read()
+
+    return None
+
+
+def trigger_prefetch(filepath: str, file_hash: str, audio: int, resolution: str,
+                     current: int, info: dict):
+    """Prefetch upcoming segments."""
+    duration = float(info.get('format', {}).get('duration', 0))
+    total = int(duration / SEGMENT_DURATION) + 1
+
+    for offset in range(1, PREFETCH_SEGMENTS + 1):
+        next_seg = current + offset
+        if next_seg >= total:
+            break
+
+        output = get_segment_path(file_hash, audio, resolution, next_seg)
+        if os.path.exists(output):
+            continue
+
+        key = f"{file_hash}:{audio}:{resolution}:{next_seg}"
+        seg = next_seg
+        segment_manager.schedule_prefetch(
+            key,
+            lambda s=seg: transcode_segment(filepath, file_hash, audio, resolution, s)
+        )
+
+
+def generate_master_playlist(info: dict, resolution_filter: str = None, audio_filter: int = None) -> str:
+    """
+    Generate HLS master playlist.
+
+    Args:
+        info: ffprobe output
+        resolution_filter: Specific resolution (e.g., 'original', '720p') or None for ABR
+        audio_filter: Specific audio track index or None for all tracks
+
+    When audio_filter is specified, generates a simple playlist with just that
+    audio track muxed in - no AUDIO group needed since there's only one option.
+    """
+    streams = info.get('streams', [])
+    video = next((s for s in streams if s.get('codec_type') == 'video'), None)
+    audio_streams = [s for s in streams if s.get('codec_type') == 'audio']
+    subtitle_streams = [s for s in streams if s.get('codec_type') == 'subtitle']
+
+    lines = ["#EXTM3U", "#EXT-X-VERSION:4", ""]
+
+    # Subtitle tracks
+    sub_counter = 0
+    for i, s in enumerate(subtitle_streams):
+        codec = s.get('codec_name', '')
+        if codec in UNSUPPORTED_SUBTITLE_CODECS:
+            continue
+        lang = s.get('tags', {}).get('language', 'und')
+        title = s.get('tags', {}).get('title', '')
+        name = f"{title}" if title else (lang.upper() if lang != 'und' else f"Subtitle {sub_counter+1}")
+        lines.append(f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",LANGUAGE="{lang}",DEFAULT=NO,AUTOSELECT=YES,URI="subtitle_{i}.m3u8"')
+        sub_counter += 1
+
+    if sub_counter > 0:
+        lines.append("")
+
+    # Video variants
+    if video:
+        src_w = video.get('width', 1920)
+        src_h = video.get('height', 1080)
+        subs_ref = ',SUBTITLES="subs"' if sub_counter > 0 else ''
+
+        # Build quality ladder
+        if resolution_filter:
+            res_info = RESOLUTIONS.get(resolution_filter, RESOLUTIONS['original'])
+            bw_map = {None: 5000000, 1080: 4000000, 720: 2500000, 480: 1200000, 360: 800000}
+            quality_order = [(resolution_filter, res_info[1], bw_map.get(res_info[1], 5000000))]
+        else:
+            quality_order = [
+                ('original', None, 5000000),
+                ('1080p', 1080, 4000000),
+                ('720p', 720, 2500000),
+                ('480p', 480, 1200000),
+                ('360p', 360, 800000),
+            ]
+
+        # Determine audio track to use
+        audio_index = audio_filter if audio_filter is not None else 0
+
+        # Simple playlist - just stream variants, no AUDIO group
+        # Audio is muxed into each stream variant
+        for res_key, target_h, bw in quality_order:
+            if target_h and target_h > src_h:
+                continue
+            if target_h and target_h == src_h:
+                continue
+
+            w, h, _ = RESOLUTIONS.get(res_key, RESOLUTIONS['original'])
+            width = w or src_w
+            height = h or src_h
+            res_label = f"{height}p (Original)" if res_key == 'original' else f"{height}p"
+
+            lines.append(f'#EXT-X-STREAM-INF:BANDWIDTH={bw},RESOLUTION={width}x{height}{subs_ref},NAME="{res_label}"')
+            lines.append(f"stream_a{audio_index}_{res_key}.m3u8")
+
+    return "\n".join(lines) + "\n"
+
+
+def generate_stream_playlist(info: dict, audio: int, resolution: str) -> str:
+    """Generate HLS stream playlist with muxed video+audio segments."""
+    duration = float(info.get('format', {}).get('duration', 0))
+    num_segments = int(duration / SEGMENT_DURATION) + 1
+
+    lines = [
+        "#EXTM3U",
+        "#EXT-X-VERSION:3",
+        f"#EXT-X-TARGETDURATION:{SEGMENT_DURATION}",
+        "#EXT-X-MEDIA-SEQUENCE:0",
+        "#EXT-X-PLAYLIST-TYPE:VOD",
+        ""
+    ]
+
+    for i in range(num_segments):
+        seg_dur = min(SEGMENT_DURATION, duration - (i * SEGMENT_DURATION))
+        if seg_dur > 0.1:
+            lines.append(f"#EXTINF:{seg_dur:.3f},")
+            lines.append(f"seg_a{audio}_{resolution}_{i:05d}.ts")
+
+    lines.append("#EXT-X-ENDLIST")
+    return "\n".join(lines)
+
+
+def generate_subtitle_playlist(info: dict, sub_index: int) -> str:
+    """Generate subtitle playlist."""
+    duration = float(info.get('format', {}).get('duration', 0))
+    return "\n".join([
+        "#EXTM3U",
+        "#EXT-X-VERSION:3",
+        f"#EXT-X-TARGETDURATION:{int(duration) + 1}",
+        "#EXT-X-MEDIA-SEQUENCE:0",
+        "#EXT-X-PLAYLIST-TYPE:VOD",
+        "",
+        f"#EXTINF:{duration:.3f},",
+        f"subtitle_{sub_index}.vtt",
+        "#EXT-X-ENDLIST"
+    ])
+
+
+def get_metrics() -> dict:
+    """Get transcoding metrics."""
+    return {
+        **segment_manager.get_metrics(),
+        'adaptive_quality': adaptive_quality.get_stats(),
+    }
+
+
+def reset_metrics():
+    """Reset transcoding metrics."""
+    segment_manager.reset_metrics()