align with metadata standard

.gitattributes

@@ -0,0 +1,6 @@
+# FIX CRLF always when developer has not set
+# Linux/Mac: git config --global core.autocrlf input
+# Windows: git config --global core.autocrlf true
+# Auto detect text files and perform LF normalization
+* text=auto
+* eol=lf

README.md

@@ -1,16 +1,11 @@
 # Meta-Stremio
 
-A standalone Stremio addon service for streaming personal media with on-the-fly HLS transcoding. Part of the MetaMesh decentralized media ecosystem.
+Stremio addon for the MetaMesh stack. Reads file metadata from the shared KV store, reads files via WebDAV (or direct mount), and serves HLS with on-the-fly FFmpeg transcoding plus the standard Stremio addon protocol.
+
+External port in the dev stack: **8182** (auth-gated, via Caddy + nginx-hash-lock). Internal container port is `7000`. Debug-direct port (bypasses the perimeter): **18182**. Container name: `metastremio-app` (backend) / `metastremio` (hash-lock proxy).
 
 ## Overview
 
-Meta-Stremio is a **process-read** service that:
-- Reads file metadata from the shared KV store (Redis)
-- Reads media files via WebDAV (from meta-sort) or direct volume access
-- Provides HLS streaming with adaptive real-time FFmpeg transcoding
-- Serves as a Stremio addon for seamless playback
-- Supports SMB/rclone mounts via meta-sort's WebDAV endpoint
-
 ```
 ┌─────────────────────────────────────────────────────────────────────┐
 │                        MetaMesh Ecosystem                           │
@@ -24,245 +19,223 @@ Meta-Stremio is a **process-read** service that:
 │         ▼                   ▼                   ▼                   │
 │  ┌─────────────────────────────────────────────────────┐           │
 │  │              Shared KV Store (Redis)                │           │
-│  │   Leader election via flock on shared filesystem    │           │
-│  │   DB stored in: DATA/Apps/meta-core/DB              │           │
+│  │   meta-core is leader, exposes /urls + /meta APIs   │           │
 │  └─────────────────────────────────────────────────────┘           │
 │                             │                                       │
 │                             ▼                                       │
 │  ┌─────────────────────────────────────────────────────┐           │
-│  │           Shared DATA Volume (media files)          │           │
-│  │   meta-sort writes → meta-stremio/meta-fuse reads   │           │
+│  │      File access via WebDAV (or direct mount)       │           │
 │  └─────────────────────────────────────────────────────┘           │
-│                                                                     │
 └─────────────────────────────────────────────────────────────────────┘
 ```
 
 ## Features
 
-### High Performance
-- **Adaptive Quality** - Dynamic preset + CRF adjustment targeting 60-80% transcode ratio
-- **Intelligent Prefetch** - 4 segments ahead for smooth playback
-- **Single-threaded Encode** - 100% CPU per segment for fastest encoding
-- **Muxed Segments** - Video+audio in one FFmpeg call
-- **Segment Caching** - Persistent cache for repeated access
-
-### Stremio Integration
-- **Catalog Endpoint** - Browse media library by type (movies, series)
-- **Meta Endpoint** - Detailed video information with codec details
-- **Stream Endpoint** - Multiple stream options per video
-- **Multi-Audio** - Separate HLS streams per audio track
-- **Subtitles** - Extract and serve embedded subtitles as VTT
-- **Language Configuration** - Configure display language (uses metadata stored by meta-sort's TMDB plugin)
-
-### Storage Abstraction
-- **Leader Storage** - Auto-discovers meta-sort's Redis via leader election (recommended)
-- **Redis Storage** - Direct Redis connection via URL
-
-## Project Structure
-
-```
-meta-stremio/
-├── README.md
-├── CLAUDE.md                   # Development guide
-├── Dockerfile
-├── docker-compose.yml
-├── requirements.txt            # redis + requests
-├── src/
-│   ├── server.py               # HTTP server + routing
-│   ├── stremio.py              # Stremio handlers + storage integration
-│   ├── transcoder.py           # HLS transcoding engine
-│   ├── fileserver.py           # File serving (direct or WebDAV)
-│   ├── webdav_client.py        # WebDAV client for meta-sort file access
-│   └── storage/
-│       ├── __init__.py
-│       ├── provider.py         # Abstract StorageProvider
-│       ├── redis_storage.py    # Direct Redis connection
-│       ├── leader_storage.py   # Leader-aware Redis connection
-│       └── leader_discovery.py # Leader election discovery
-└── www/
-    └── index.html              # Dashboard
-```
+- **HLS transcoding** — adaptive preset+CRF targeting a 60–80% transcode ratio, prefetch of N segments ahead, persistent segment cache.
+- **Stremio protocol** — manifest, catalog, meta, stream; one HLS stream per audio track; subtitles extracted to VTT.
+- **Direct file serving** — range-request capable, for clients that can play the source directly.
+- **Storage abstractions** — `LeaderStorage` (auto-discovers Redis via meta-core), `RedisStorage` (direct URL), `direct` mode (reads from meta-core's HTTP API + SSE meta-stream).
+- **Path-token gate for addon URLs** — when `HASH_API_SEED` is set, addon paths must be prefixed with `/s/<token>`. Dashboard/browser paths are protected separately by the upstream hash-lock + OIDC sidecar (Stremio clients can't carry cookies, so they get the path-token instead).
+- **Language-configurable manifest** — `displayLanguage` in the per-install config (base64-in-path) selects which `titles` translation meta-sort's TMDB plugin populated.
 
 ## Installation
 
-### Docker (Recommended)
+### Docker (in the MetaMesh dev stack)
 
 ```bash
-# Build
-docker build -t meta-stremio .
+cd dev
+./scripts/start.sh
+```
 
-# Run with leader discovery (recommended - auto-discovers meta-sort's Redis)
+This brings up `metastremio-app` on port 18182 (debug-direct) and the auth-gated entry on `https://metastremio-dev.localhost:8182`.
+
+### Docker (standalone)
+
+```bash
+docker build -t meta-stremio packages/meta-stremio
+
+# Leader mode (auto-discover Redis via meta-core)
 docker run -d \
-  -p 7000:7000 \
+  -p 8182:7000 \
   -v /path/to/media:/files:ro \
   -v /path/to/meta-core:/meta-core \
   meta-stremio
 
-# Run with direct Redis connection
+# Direct Redis mode
 docker run -d \
-  -p 7000:7000 \
+  -p 8182:7000 \
   -v /path/to/media:/files:ro \
   -e STORAGE_MODE=redis \
   -e REDIS_URL=redis://your-redis:6379 \
   meta-stremio
 ```
 
-### Docker Compose
+The container always listens on port `7000` internally — map it to whatever you want externally.
+
+### Standalone (development)
 
 ```bash
-# Run as part of the MetaMesh stack (uses leader discovery)
-cd dev
-./start.sh
-```
-
-### Standalone (Development)
-
-```bash
-# Prerequisites: Python 3.9+, FFmpeg
-
-# Install dependencies
-pip install -r requirements.txt
-
-# Run
+# Requires Python 3.9+ and FFmpeg.
+pip install -r requirements.txt   # redis, watchdog, Pillow, requests
 cd src
 python server.py
 ```
 
 ## Configuration
 
-### Environment Variables
+### Environment variables
 
 | Variable | Default | Description |
-|----------|---------|-------------|
-| `PORT` | `7000` | Server listen port |
-| `MEDIA_DIR` | `/data/media` | Media files directory |
-| `CACHE_DIR` | `/data/cache` | Transcoded segment cache |
-| `STORAGE_MODE` | `leader` | `leader` or `redis` |
-| `META_CORE_PATH` | `/meta-core` | Path to meta-core shared volume (for leader discovery) |
-| `FILES_PATH` | `/files` | Path to files volume |
-| `REDIS_URL` | `redis://localhost:6379` | Redis connection URL (for `redis` mode) |
-| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix |
-| `SEGMENT_DURATION` | `4` | HLS segment length (seconds) |
-| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead |
-| `SCHEME` | `auto` | URL scheme (`http`, `https`, `auto`) |
-| `META_CORE_WEBDAV_URL` | - | WebDAV URL for file access (e.g., `http://meta-core/webdav`) |
+|---|---|---|
+| `PORT` | `7000` | Internal HTTP listen port (the dev stack maps 18182 / 8182 to this). |
+| `BASE_URL` | — | External URL emitted in manifests / poster URLs (e.g. `https://metastremio-dev.localhost:8182`). |
+| `MEDIA_DIR` | `/files/watch` | Media directory (consumed by `transcoder.py`). |
+| `CACHE_DIR` | `/data/cache` | Transcoded segment cache. |
+| `STORAGE_MODE` | `leader` | `leader`, `redis`, or `direct`. |
+| `META_CORE_PATH` | `/meta-core` | Shared meta-core volume; used by `LeaderClient` for `kv-leader.info`. |
+| `META_CORE_URL` | — | meta-core HTTP API (e.g. `http://metacore-app:9000`); required for `STORAGE_MODE=direct` since there is no leader file to read. |
+| `FILES_PATH` | `/files` | Files volume. |
+| `REDIS_URL` | `redis://localhost:6379` | Redis URL for `STORAGE_MODE=redis`. |
+| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix. |
+| `SCHEME` | `auto` | URL scheme for generated URLs: `http`, `https`, or `auto`. |
+| `SEGMENT_DURATION` | `4` | HLS segment length (s). |
+| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead. |
+| `META_CORE_WEBDAV_URL` | — | When set, files are read over HTTP from meta-core's WebDAV instead of the filesystem (lets meta-stremio reach SMB / rclone mounts that only exist inside meta-core). |
+| `HASH_API_SEED` | — | When set, derives a 16-char path-token; all addon routes must be prefixed with `/s/<token>`. Dashboard routes bypass. |
 
-**WebDAV File Access**: When `META_CORE_WEBDAV_URL` is set, meta-stremio reads files via HTTP from meta-core's WebDAV server instead of direct filesystem access. This enables access to SMB/rclone mounts that are dynamically mounted inside meta-core.
+## API endpoints
 
-## API Endpoints
+All routes below are matched in `src/server.py`. When `HASH_API_SEED` is set, **every addon path is prefixed with `/s/<token>`**; the dashboard paths in the first table are *not* prefixed (they're protected by hash-lock/OIDC upstream).
+
+### Dashboard / browser
 
-### Dashboard
 | Endpoint | Method | Description |
-|----------|--------|-------------|
-| `/` | GET | Setup dashboard |
-| `/health` | GET | Health check with storage status |
-
-### Dashboard API
-| Endpoint | Method | Description |
-|----------|--------|-------------|
+|---|---|---|
+| `/` , `/index.html` | GET | Setup dashboard |
+| `/configure` | GET | Per-install language configuration page |
+| `/health` | GET | Liveness + storage status |
 | `/api/stats` | GET | Library statistics |
-| `/api/library` | GET | Full library list |
+| `/api/library` | GET | Full library list (videos + count) |
+| `/api/services` | GET | Discovered MetaMesh services (proxied from `meta-core /services`) |
+| `/api/languages` | GET | Languages selectable from the `/configure` page |
+
+### Stremio addon protocol
+
+`/manifest.json` and `/stremio/manifest.json` are aliases; same for catalog/meta/stream. Stremio per-install configuration is encoded as URL-safe base64 JSON between the `/s/<token>` prefix and the addon path (e.g. `/s/<token>/<configB64>/manifest.json`).
 
-### Stremio Protocol
 | Endpoint | Method | Description |
-|----------|--------|-------------|
-| `/manifest.json` | GET | Addon manifest |
-| `/catalog/:type/:id.json` | GET | Video catalog |
+|---|---|---|
+| `/manifest.json` | GET, HEAD | Addon manifest |
+| `/stremio/manifest.json` | GET, HEAD | Alias of `/manifest.json` |
+| `/catalog/:type/:id.json` | GET | Catalog (with optional `…/:extra.json`) |
+| `/stremio/catalog/:type/:id.json` | GET | Alias |
 | `/meta/:type/:id.json` | GET | Video metadata |
+| `/stremio/meta/:type/:id.json` | GET | Alias |
 | `/stream/:type/:id.json` | GET | Stream URLs |
+| `/stremio/stream/:type/:id.json` | GET | Alias |
+
+### File serving
 
-### Transcoding
 | Endpoint | Method | Description |
-|----------|--------|-------------|
-| `/transcode/:path/master.m3u8` | GET | ABR master playlist |
-| `/transcode/:path/master_original_a0.m3u8` | GET | Original quality playlist |
-| `/transcode/:path/stream_a0_720p.m3u8` | GET | Quality stream playlist |
-| `/transcode/:path/seg_a0_720p_0.ts` | GET | Video segment |
-| `/transcode/:path/subtitle_0.vtt` | GET | Subtitle track |
-| `/direct/:path` | GET | Direct file serving |
-| `/transcode/metrics` | GET | Transcoding metrics |
+|---|---|---|
+| `/file/{cid}` | GET, HEAD | Serve file by CID |
+| `/file/{cid}/w{width}` | GET, HEAD | Serve resized image by CID (uses Pillow) |
+| `/poster/{cid}` | GET, HEAD | Legacy alias of `/file/{cid}` |
+| `/poster/{cid}/w{width}` | GET, HEAD | Legacy alias of `/file/{cid}/w{width}` |
+| `/direct/{path}` | GET, HEAD | Direct file serving with HTTP `Range` support |
 
-## KV Store Schema
+### Transcoder
 
-Meta-stremio reads from meta-sort's Redis using this key format:
+| Endpoint | Method | Description |
+|---|---|---|
+| `/transcode/{path}/master.m3u8` | GET, HEAD | ABR master playlist |
+| `/transcode/{path}/master_{resolution}.m3u8` | GET, HEAD | Master playlist for a single resolution |
+| `/transcode/{path}/master_a{audio}.m3u8` | GET, HEAD | Master playlist for a single audio track |
+| `/transcode/{path}/master_{resolution}_a{audio}.m3u8` | GET, HEAD | Combined variant |
+| `/transcode/{path}/stream_a{audio}_{resolution}.m3u8` | GET, HEAD | Per-variant stream playlist |
+| `/transcode/{path}/seg_a{audio}_{resolution}_{n}.ts` | GET, HEAD | Video segment |
+| `/transcode/{path}/subtitle_{idx}.m3u8` | GET, HEAD | Subtitle playlist |
+| `/transcode/{path}/subtitle_{idx}.vtt` | GET, HEAD | Subtitle VTT |
+| `/transcode/metrics` | GET | Transcoder metrics (includes `total_files`) |
+| `/transcode/reset-metrics` | POST | Reset transcoder metrics |
+
+## Project structure
 
 ```
-meta-sort:file:{hashId}              → Hash containing all metadata
-  - filePath                         → /media/movies/Movie.mkv
-  - title                            → Movie Title
-  - type                             → movie|series|anime
-  - year                             → 2024
-  - season                           → 1 (for series)
-  - episode                          → 1 (for series)
-  - duration                         → 7200 (seconds)
-  - width                            → 1920
-  - height                           → 1080
-  - videoCodec                       → h264
-  - audioCodec                       → aac
-  - audioTracks                      → JSON string
-  - subtitles                        → JSON string
-  - imdbId                           → tt1234567
-  - poster                           → URL
+meta-stremio/
+├── README.md
+├── Dockerfile
+├── docker-compose.yml
+├── docker/                            # container entry scripts
+├── requirements.txt                   # redis, watchdog, Pillow, requests
+├── src/
+│   ├── server.py                      # HTTP server + routing (entry point)
+│   ├── stremio.py                     # Stremio handlers + library/language helpers
+│   ├── transcoder.py                  # HLS transcoding engine + metrics + segment/subtitle managers
+│   ├── fileserver.py                  # serve_file(cid, width) — file/poster endpoints
+│   ├── poster.py                      # poster URL/CID helpers
+│   ├── webdav_client.py               # WebDAV client (file_exists, stream_file, stream_range, get_file_size)
+│   └── storage/
+│       ├── __init__.py
+│       ├── provider.py                # StorageProvider + VideoMetadata abstract base
+│       ├── redis_storage.py           # Direct Redis backend
+│       ├── leader_storage.py          # Leader-aware Redis backend
+│       ├── leader_client.py           # Reads /meta-core/locks/kv-leader.info + calls meta-core /urls
+│       ├── meta_consumer.py           # SSE consumer from meta-core /meta stream
+│       ├── meta_core_api_client.py    # HTTP client for meta-core REST API
+│       └── service_registration.py    # Registers meta-stremio with meta-core /services
+└── www/
+    ├── index.html                     # Dashboard
+    └── configure.html                 # Language configuration page
 ```
 
-## Stream Types
+## KV store schema
 
-Each video provides multiple stream options:
+meta-stremio reads from the per-file Redis hash `/file/{cid}` populated by meta-sort and its plugins. Field semantics (`videoType`, `originalTitle`, `titles`, `season`, `episode`, `movieYear`, `cid_*`, etc.) are documented in the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md) — that is the single source of truth, and matches the `@metazla/meta-interface` types.
 
-1. **Direct File** - Original file, no transcoding
-   - Best quality, may not play on all devices
+## Stream types
 
-2. **HLS Original** - Transcoded at source resolution
-   - H.264/AAC in HLS container
-   - One stream per audio track
+For each video, the addon advertises multiple streams:
 
-3. **HLS ABR** - Adaptive bitrate with quality ladder
-   - Multiple resolutions (1080p, 720p, 480p, 360p)
-   - Automatic quality switching
+1. **Direct file** — original bytes, served via `/direct/...` with `Range` support. Best quality, may not play on every device.
+2. **HLS original** — transcoded at source resolution, H.264/AAC, one stream per audio track.
+3. **HLS ABR** — adaptive ladder. Quality presets are baked into `transcoder.py`.
 
-## Transcoding Details
-
-### Adaptive Quality System
-
-The transcoder automatically adjusts encoding settings to maintain a 60-80% transcode ratio (segment encode time / segment duration):
-
-- **Too Fast (<60%)**: Increase quality (lower CRF, slower preset)
-- **Too Slow (>80%)**: Decrease quality (higher CRF, faster preset)
-
-### Quality Ladder
-
-| Resolution | Video Bitrate | Audio Bitrate |
-|------------|---------------|---------------|
+| Resolution | Video bitrate | Audio bitrate |
+|---|---|---|
 | 1080p | 5000 kbps | 192 kbps |
-| 720p | 3000 kbps | 128 kbps |
-| 480p | 1500 kbps | 128 kbps |
-| 360p | 800 kbps | 96 kbps |
+| 720p  | 3000 kbps | 128 kbps |
+| 480p  | 1500 kbps | 128 kbps |
+| 360p  |  800 kbps |  96 kbps |
 
 ## Adding to Stremio
 
-1. Open the dashboard at `http://localhost:7000`
-2. Click "Install in Stremio" or copy the manifest URL
-3. In Stremio: Settings → Addons → Enter URL → Install
+1. Open the dashboard at `https://metastremio-dev.localhost:8182/` (or debug-direct `http://localhost:18182/`).
+2. Pick your display language on `/configure` if you want non-English titles.
+3. Use the **Install in Stremio** button (uses the `stremio://` protocol handler) or copy the manifest URL into Stremio: Settings → Addons → Add Addon.
 
-## Development
+## Debugging
 
 ```bash
-# Run in development mode (requires meta-sort running with Redis)
-cd src
-python server.py
+# Container logs
+docker compose -f dev/docker-compose.yml logs -f metastremio-app
 
-# Or with direct Redis URL
-STORAGE_MODE=redis REDIS_URL=redis://localhost:6379 python server.py
+# Live health
+curl -k https://metastremio-dev.localhost:8182/health
+curl    http://localhost:18182/health   # debug-direct (no auth, no TLS)
 
-# View logs
-tail -f /data/cache/*.log
+# Transcoder metrics
+curl    http://localhost:18182/transcode/metrics
+curl -X POST http://localhost:18182/transcode/reset-metrics
 ```
 
-## Related Projects
+Container lifecycle: do **not** `docker restart metastremio-app` to apply config — use the dev-stack reload script (`./scripts/reload-stremio.sh`) so supervisord restarts the right process inside the container.
 
-- **[meta-sort](../meta-sort)** - File indexer (writes metadata to Redis)
-- **[meta-fuse](../meta-fuse)** - Virtual filesystem (reads metadata)
-- **[meta-orbit](../meta-orbit)** - P2P metadata sharing
+## Related projects
+
+- **[meta-sort](../meta-sort)** — file indexer; writes the metadata meta-stremio reads.
+- **[meta-fuse](../meta-fuse)** — virtual filesystem; reads the same KV store.
+- **[meta-share](../meta-share)** — decentralised metadata sharing.
 
 ## License
 

src/server.py

@@ -28,6 +28,7 @@ import json
 import signal
 import sys
 import base64
+import hashlib
 from http.server import HTTPServer, BaseHTTPRequestHandler
 from urllib.parse import urlparse, unquote
 import threading
@@ -44,6 +45,23 @@ SCHEME = os.environ.get('SCHEME', 'auto').lower().strip()
 META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
 BASE_URL = os.environ.get('BASE_URL', '')
 
+# Stremio addon path-token protection. When HASH_API_SEED is set, every addon
+# path (manifest, catalog, meta, stream, transcode, direct, file, poster) must
+# be prefixed with /s/<token>. Dashboard paths bypass this — Caddy + the
+# nginx-hash-lock sidecar gate those with OIDC, since browsers can carry
+# cookies but Stremio clients cannot.
+HASH_API_SEED = os.environ.get('HASH_API_SEED', '').strip()
+API_KEY = hashlib.sha256(HASH_API_SEED.encode()).hexdigest()[:16] if HASH_API_SEED else ''
+API_PREFIX = f'/s/{API_KEY}' if API_KEY else ''
+
+# Paths the addon serves for browsers, not Stremio clients. These are gated
+# upstream by hash-lock/OIDC instead of the path-token.
+DASHBOARD_PATHS = frozenset({
+    '/', '/index.html', '/configure', '/health',
+    '/api/stats', '/api/library', '/api/services', '/api/languages',
+    '/transcode/metrics',
+})
+
 # Initialize storage
 storage = stremio.init_storage()
 
@@ -133,10 +151,32 @@ class Handler(BaseHTTPRequestHandler):
         self.send_header('Access-Control-Allow-Headers', 'Content-Type, Range')
         self.end_headers()
 
+    def strip_api_prefix(self, path: str) -> str | None:
+        """Remove the /s/<token> prefix from addon paths.
+
+        Returns the de-prefixed path, or None when the token is missing/wrong
+        (caller should send 401). Dashboard paths are returned untouched —
+        they are protected by hash-lock/OIDC upstream, not by the token.
+        """
+        if not API_KEY:
+            return path
+        if path in DASHBOARD_PATHS:
+            return path
+        if path.startswith(API_PREFIX + '/'):
+            return path[len(API_PREFIX):]
+        if path == API_PREFIX:
+            return '/'
+        return None
+
     def do_HEAD(self):
         parsed = urlparse(self.path)
         path = unquote(parsed.path)
 
+        path = self.strip_api_prefix(path)
+        if path is None:
+            self.send_error(401, "Unauthorized")
+            return
+
         if path.startswith('/stremio/') or path == '/' or path.startswith('/manifest.json'):
             self.send_response(200)
             self.send_header('Access-Control-Allow-Origin', '*')
@@ -181,6 +221,11 @@ class Handler(BaseHTTPRequestHandler):
         parsed = urlparse(self.path)
         path = unquote(parsed.path)
 
+        path = self.strip_api_prefix(path)
+        if path is None:
+            self.send_error(401, "Unauthorized")
+            return
+
         # Reset metrics
         if path == '/transcode/reset-metrics':
             transcoder.reset_metrics()
@@ -192,8 +237,16 @@ class Handler(BaseHTTPRequestHandler):
         parsed = urlparse(self.path)
         raw_path = unquote(parsed.path)
 
+        # Strip path-token (when HASH_API_SEED is set) before any further
+        # parsing; the existing base64-config-in-path scheme then runs on the
+        # tail, so URLs look like /s/<token>/<configB64>/manifest.json.
+        stripped = self.strip_api_prefix(raw_path)
+        if stripped is None:
+            self.send_error(401, "Unauthorized")
+            return
+
         # Parse config from path (e.g., /eyJsYW5n.../manifest.json)
-        path, config = parse_config_from_path(raw_path)
+        path, config = parse_config_from_path(stripped)
 
         # Root - serve setup page
         if path == '/' or path == '/index.html':
@@ -325,11 +378,14 @@ class Handler(BaseHTTPRequestHandler):
         if os.path.exists(html_path):
             with open(html_path, 'r', encoding='utf-8') as f:
                 content = f.read()
+            prefix_script = f'<script>window.API_PREFIX = "{API_PREFIX}";</script>'
+            content = content.replace('</head>', f'{prefix_script}\n</head>', 1)
+            encoded = content.encode('utf-8')
             self.send_response(200)
             self.send_header('Content-Type', 'text/html')
-            self.send_header('Content-Length', len(content.encode('utf-8')))
+            self.send_header('Content-Length', len(encoded))
             self.end_headers()
-            self.wfile.write(content.encode('utf-8'))
+            self.wfile.write(encoded)
         else:
             # Fallback minimal page
             content = f"""<!DOCTYPE html>
@@ -338,7 +394,7 @@ class Handler(BaseHTTPRequestHandler):
 <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>
+<p><a href="stremio://{self.headers.get('Host', 'localhost')}{API_PREFIX}/manifest.json" style="color: #4dabf7;">Install in Stremio</a></p>
 </body>
 </html>"""
             self.send_response(200)
@@ -364,6 +420,7 @@ class Handler(BaseHTTPRequestHandler):
             # Inline configure page (fallback)
             base_url = self.get_base_url()
             host = self.headers.get('Host', 'localhost')
+            api_prefix = API_PREFIX
             languages = stremio.get_supported_languages()
 
             lang_options = '\n'.join(
@@ -508,6 +565,7 @@ class Handler(BaseHTTPRequestHandler):
     <script>
         const baseUrl = '{base_url}';
         const host = '{host}';
+        const apiPrefix = '{api_prefix}';
 
         function encodeConfig(config) {{
             const json = JSON.stringify(config);
@@ -523,12 +581,12 @@ class Handler(BaseHTTPRequestHandler):
             if (displayLang === 'en') {{
                 // No config needed for default
                 url = baseUrl + '/manifest.json';
-                stremioUrl = 'stremio://' + host + '/manifest.json';
+                stremioUrl = 'stremio://' + host + apiPrefix + '/manifest.json';
             }} else {{
                 const config = {{ displayLanguage: displayLang }};
                 const encoded = encodeConfig(config);
                 url = baseUrl + '/' + encoded + '/manifest.json';
-                stremioUrl = 'stremio://' + host + '/' + encoded + '/manifest.json';
+                stremioUrl = 'stremio://' + host + apiPrefix + '/' + encoded + '/manifest.json';
             }}
 
             document.getElementById('installUrl').textContent = url;
@@ -827,7 +885,7 @@ class Handler(BaseHTTPRequestHandler):
             proto = 'http' if is_localhost else 'https'
 
         clean_host = host.replace(':80', '').replace(':443', '')
-        return f"{proto}://{clean_host}"
+        return f"{proto}://{clean_host}{API_PREFIX}"
 
 
 class ThreadedServer(HTTPServer):
@@ -875,8 +933,12 @@ def main():
     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"Manifest URL: http://localhost:{PORT}{API_PREFIX}/manifest.json")
     print(f"Dashboard: http://localhost:{PORT}/")
+    if API_KEY:
+        print(f"Path-token protection: enabled (prefix: {API_PREFIX})")
+    else:
+        print("Path-token protection: disabled (set HASH_API_SEED to enable)")
     print("Adaptive quality: target 60-80% transcode ratio")
 
     if service_discovery:

src/storage/leader_client.py

@@ -137,6 +137,21 @@ class LeaderClient:
 
                 data = json.loads(response.read().decode())
 
+                # api-mediated-access PR D: redisUrl is no longer published by
+                # meta-core. If it shows up, an old build snuck back in —
+                # fail loudly so the rollback can't go unnoticed. Set
+                # ALLOW_LEGACY_REDIS_URL=1 to downgrade to a warning during
+                # a deliberate temporary rollback.
+                if data.get('redisUrl'):
+                    msg = (
+                        "[LeaderClient] meta-core still publishes redisUrl; "
+                        "direct Redis access was retired by the api-mediated-access "
+                        "lockdown. Verify meta-core version."
+                    )
+                    if os.environ.get("ALLOW_LEGACY_REDIS_URL") == "1":
+                        print("WARNING: " + msg)
+                    else:
+                        raise RuntimeError(msg)
                 self._cached_urls = URLsResponse(
                     hostname=data.get('hostname', ''),
                     base_url=data.get('baseUrl', ''),

src/storage/leader_storage.py

@@ -20,6 +20,7 @@ from typing import List, Optional, Callable
 from .provider import StorageProvider, VideoMetadata
 from .leader_client import LeaderClient, LeaderLockInfo
 from .meta_consumer import MetaConsumer
+from .meta_core_api_client import MetaCoreApiClient
 
 # Import webdav_client to configure it after leader discovery
 import webdav_client
@@ -70,6 +71,7 @@ class LeaderStorage(StorageProvider):
 
         self._leader_client: Optional[LeaderClient] = None
         self._client: Optional[redis.Redis] = None
+        self._api_client: Optional[MetaCoreApiClient] = None
         self._connected = False
         self._lock = threading.Lock()
 
@@ -122,9 +124,14 @@ class LeaderStorage(StorageProvider):
                 # Try to get leader info (short timeout per attempt)
                 info = self._leader_client.get_leader_info()
 
-                if info and info.redis_url:
-                    print(f"[LeaderStorage] Found leader at {info.redis_url} (attempt {attempt})")
-                    self._connect_to_redis(info.redis_url)
+                # api-mediated-access PR D: when redis_url is absent, run in
+                # HTTP-only mode against meta-core's /meta/* surface. Redis
+                # access has been retired; the API client covers every read
+                # this storage provider needs. Only fall back to Redis when
+                # both redis_url and api_url are present (transition state).
+                if info and info.api_url:
+                    print(f"[LeaderStorage] Found leader at {info.api_url} (attempt {attempt})")
+                    self._connect_via_api(info)
 
                     if self._connected:
                         # Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access)
@@ -140,7 +147,7 @@ class LeaderStorage(StorageProvider):
                         self._leader_client.start_watching()
                         return
                     else:
-                        print(f"[LeaderStorage] Failed to connect to Redis, retrying...")
+                        print(f"[LeaderStorage] Failed to connect to meta-core, retrying...")
                 else:
                     print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})")
 
@@ -174,9 +181,9 @@ class LeaderStorage(StorageProvider):
             try:
                 info = self._leader_client.get_leader_info()
 
-                if info and info.redis_url:
-                    print(f"[LeaderStorage] Background: Found leader at {info.redis_url}")
-                    self._connect_to_redis(info.redis_url)
+                if info and info.api_url:
+                    print(f"[LeaderStorage] Background: Found leader at {info.api_url}")
+                    self._connect_via_api(info)
 
                     if self._connected:
                         # Configure WebDAV client
@@ -214,9 +221,30 @@ class LeaderStorage(StorageProvider):
         self._disconnect_redis()
 
     def is_connected(self) -> bool:
-        """Check if connected to the storage backend."""
+        """Check if connected to the storage backend (Redis or HTTP API).
+
+        HTTP-only mode auto-recovers from transient meta-core outages: if
+        the health probe fails we drop the connected flag but keep the
+        api_client so the *next* call probes again. The background
+        reconnect thread is only needed for Redis-mode where the
+        connection itself is sticky.
+        """
         with self._lock:
-            if not self._connected or not self._client:
+            # HTTP-only mode: probe meta-core's /health on every call. This
+            # is cheap (a single HTTP GET) and naturally recovers when
+            # meta-core comes back without needing a separate reconnect
+            # loop. The api_client object stays valid through transient
+            # failures — it's stateless aside from the configured URL.
+            if self._api_client is not None and self._client is None:
+                if self._api_client.health():
+                    self._connected = True
+                    return True
+                self._connected = False
+                return False
+            # Legacy Redis mode.
+            if not self._connected:
+                return False
+            if not self._client:
                 return False
             try:
                 self._client.ping()
@@ -241,9 +269,9 @@ class LeaderStorage(StorageProvider):
                 try:
                     info = self._leader_client.get_leader_info()
 
-                    if info and info.redis_url:
-                        print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...")
-                        self._connect_to_redis(info.redis_url)
+                    if info and info.api_url:
+                        print(f"[LeaderStorage] Reconnecting to leader at {info.api_url}...")
+                        self._connect_via_api(info)
 
                         if self._connected:
                             # Reconfigure WebDAV client with new leader's internal WebDAV URL
@@ -260,8 +288,46 @@ class LeaderStorage(StorageProvider):
 
             print(f"[LeaderStorage] Failed to reconnect after {max_attempts} attempts")
 
+    def _connect_via_api(self, info: LeaderLockInfo) -> None:
+        """Connect to meta-core via its HTTP API. The api_client covers every
+        read this storage provider does — Redis access is no longer required
+        (api-mediated-access PR D). Also starts the SSE meta-events consumer
+        so cache invalidation still flows in real time.
+        """
+        with self._lock:
+            try:
+                # Drop any prior state (Redis or otherwise)
+                self._disconnect_redis_unlocked()
+
+                # Set up the HTTP read client + probe meta-core.
+                api = MetaCoreApiClient(api_url=info.api_url)
+                if not api.health():
+                    raise RuntimeError(f"meta-core health probe failed at {info.api_url}")
+                self._api_client = api
+                self._connected = True
+
+                print(f"[LeaderStorage] Connected via meta-core API at {info.api_url}")
+
+                # Start SSE consumer for cache invalidation.
+                try:
+                    self._meta_consumer = MetaConsumer(api_url=info.api_url)
+                    self._meta_consumer.on_change(self._on_metadata_change)
+                    self._meta_consumer.start()
+                except Exception as e:
+                    print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
+
+                self._cache_dirty = True
+                self._notify_ready()
+
+            except Exception as e:
+                print(f"[LeaderStorage] Failed to connect via API: {e}")
+                self._api_client = None
+                self._connected = False
+
     def _connect_to_redis(self, url: str) -> None:
-        """Connect to Redis."""
+        """Connect to Redis. Legacy path — kept for direct REDIS_URL overrides
+        and ad-hoc tests. Production discovery goes through _connect_via_api.
+        """
         with self._lock:
             try:
                 # Disconnect existing client
@@ -274,11 +340,19 @@ class LeaderStorage(StorageProvider):
 
                 print(f"[LeaderStorage] Connected to Redis at {url}")
 
-                # Start meta events consumer for real-time updates
+                # Start meta events consumer (HTTP SSE against meta-core).
+                # Prefer the api URL we discovered via the LeaderClient; fall
+                # back to META_CORE_URL env so STORAGE_MODE=direct deployments
+                # (no leader discovery) still get live updates.
                 try:
-                    self._meta_consumer = MetaConsumer(self._client, self._prefix)
-                    self._meta_consumer.on_change(self._on_metadata_change)
-                    self._meta_consumer.start()
+                    info = self._leader_client.get_cached_leader_info() if self._leader_client else None
+                    api_url = (info.api_url if info else None) or os.environ.get("META_CORE_URL")
+                    if api_url:
+                        self._meta_consumer = MetaConsumer(api_url=api_url)
+                        self._meta_consumer.on_change(self._on_metadata_change)
+                        self._meta_consumer.start()
+                    else:
+                        print("[LeaderStorage] Warning: api_url unavailable; meta consumer not started")
                 except Exception as e:
                     print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
 
@@ -299,7 +373,7 @@ class LeaderStorage(StorageProvider):
             self._disconnect_redis_unlocked()
 
     def _disconnect_redis_unlocked(self) -> None:
-        """Disconnect from Redis (without lock)."""
+        """Disconnect from Redis and/or API (without lock)."""
         # Stop meta consumer
         if self._meta_consumer:
             try:
@@ -314,6 +388,10 @@ class LeaderStorage(StorageProvider):
             except Exception:
                 pass
             self._client = None
+
+        # The API client has no socket; just drop the reference.
+        self._api_client = None
+
         self._connected = False
 
         # Clear cache
@@ -377,10 +455,14 @@ class LeaderStorage(StorageProvider):
         return f"{self._prefix}file:{hash_id}/"
 
     def _get_file_metadata(self, hash_id: str) -> dict:
-        """Get all metadata for a file using flat key scan."""
+        """Get all metadata for a file. Uses meta-core HTTP API when in
+        HTTP-only mode; falls back to Redis SCAN+GET otherwise."""
+        if self._api_client is not None:
+            flat = self._api_client.get_metadata_flat(hash_id)
+            return dict(flat) if flat else {}
+
         prefix = self._get_file_key_prefix(hash_id)
         data = {}
-
         for key in self._client.scan_iter(f"{prefix}*", count=100):
             key_str = key if isinstance(key, str) else key.decode()
             field = key_str[len(prefix):]
@@ -494,10 +576,9 @@ class LeaderStorage(StorageProvider):
         # Parse release date
         release_date = data.get('releasedate', data.get('releaseDate'))
 
-        # Parse episode title - try episodeTitle or extract from titles/eng
+        # Parse episode title. `titles/*` is the work's name set, never an
+        # episode title (METADATA_KEYS.md §3).
         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')
@@ -577,6 +658,15 @@ class LeaderStorage(StorageProvider):
     # StorageProvider Interface
     # ========================================================================
 
+    def _list_hash_ids(self) -> List[str]:
+        """Enumerate every hashId. HTTP API in api-mediated mode, SMEMBERS
+        on file:__index__ in legacy Redis mode."""
+        if self._api_client is not None:
+            return self._api_client.list_hash_ids()
+        index_key = f"{self._prefix}file:__index__"
+        ids = self._client.smembers(index_key)
+        return [h if isinstance(h, str) else h.decode() for h in ids]
+
     def get_all_videos(self) -> List[VideoMetadata]:
         """Get all videos from storage."""
         if not self.is_connected():
@@ -586,8 +676,7 @@ class LeaderStorage(StorageProvider):
 
         try:
             # Get all hash IDs from index
-            index_key = f"{self._prefix}file:__index__"
-            hash_ids = self._client.smembers(index_key)
+            hash_ids = self._list_hash_ids()
 
             for hash_id in hash_ids:
                 hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
@@ -648,31 +737,39 @@ class LeaderStorage(StorageProvider):
 
         try:
             # Get from index and check each file's imdbid
-            index_key = f"{self._prefix}file:__index__"
-            for hash_id in self._client.smembers(index_key):
-                hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
-                prefix = self._get_file_key_prefix(hash_id_str)
-
-                # Check imdbid field directly
-                stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
-                if stored_imdb:
-                    stored_imdb_str = stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
-                    if stored_imdb_str.lower() == imdb_id:
-                        data = self._get_file_metadata(hash_id_str)
-                        return self._parse_video(hash_id_str, data)
+            for hash_id_str in self._list_hash_ids():
+                stored_imdb = self._get_imdb_field(hash_id_str)
+                if stored_imdb and stored_imdb.lower() == imdb_id:
+                    data = self._get_file_metadata(hash_id_str)
+                    return self._parse_video(hash_id_str, data)
 
         except Exception as e:
             print(f"[LeaderStorage] Error finding video by IMDB ID {imdb_id}: {e}")
 
         return None
 
+    def _get_imdb_field(self, hash_id: str) -> Optional[str]:
+        """Read imdbid (or imdbId) for a file. HTTP API when in api-mediated
+        mode, direct Redis GET otherwise."""
+        if self._api_client is not None:
+            v = self._api_client.get_property(hash_id, "imdbid")
+            if v:
+                return v
+            return self._api_client.get_property(hash_id, "imdbId")
+        prefix = self._get_file_key_prefix(hash_id)
+        stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
+        if not stored_imdb:
+            return None
+        return stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
+
     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.
+        In api-mediated mode this hits meta-core's GET /api/file/{cid}/info
+        which already does the reverse-index lookup + canonical-path
+        resolution. In Redis mode we fall back to reading file:{cid}/path
+        and file:{cid}/filePath directly.
 
         Returns the relative path if found, None otherwise.
         """
@@ -680,6 +777,11 @@ class LeaderStorage(StorageProvider):
             return None
 
         try:
+            if self._api_client is not None:
+                # /api/file/{cid}/info already returns a path relative to
+                # FILES_PATH, normalised. No further trimming required.
+                return self._api_client.resolve_file_path(cid)
+
             prefix = self._get_file_key_prefix(cid)
             # Try 'path' field first (relative path)
             path = self._client.get(f"{prefix}path")
@@ -704,6 +806,8 @@ class LeaderStorage(StorageProvider):
             return 0
 
         try:
+            if self._api_client is not None:
+                return self._api_client.count_hash_ids()
             # Use index set for accurate count
             index_key = f"{self._prefix}file:__index__"
             return self._client.scard(index_key)
@@ -740,8 +844,11 @@ class LeaderStorage(StorageProvider):
         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')
+                # Memory stats only available in Redis-direct mode; the HTTP
+                # API doesn't surface them. Skip cleanly when unavailable.
+                if self._client is not None:
+                    mem = self._client.info('memory')
+                    status['memory_used'] = mem.get('used_memory_human', 'N/A')
             except Exception:
                 pass
 

src/storage/meta_consumer.py

@@ -1,153 +1,241 @@
 """
 Meta Events Consumer for meta-stremio.
 
-Consumes the meta:events Redis stream published by meta-core.
-Provides real-time notifications when metadata changes.
+Consumes the meta:events stream published by meta-core. Originally used a
+Redis Streams XREADGROUP loop; migrated to meta-core's HTTP SSE endpoint at
+/api/events/meta as part of the api-mediated-access lockdown (see
+meta-core docs/api-mediated-access.md, PR C).
+
+The callback signature is unchanged so leader_storage.py needs no changes:
+each event is delivered as (key, event_type) where key has the form
+"file:<hashId>/<field>" and event_type is "set" / "del" / "expire".
 """
 from __future__ import annotations
 
-import threading
+import json
 import logging
+import os
+import threading
+import time
+from pathlib import Path
 from typing import Callable, List, Optional
+from urllib.parse import urlparse
 
-try:
-    from redis import Redis
-    REDIS_AVAILABLE = True
-except ImportError:
-    REDIS_AVAILABLE = False
-    Redis = None
+import urllib.request
+import urllib.error
 
 logger = logging.getLogger(__name__)
 
 
 class MetaConsumer:
     """
-    Consumes meta:events stream for real-time metadata updates.
+    Streams meta:events from meta-core over SSE and dispatches a filtered
+    subset of changes to registered callbacks.
 
-    Uses Redis consumer groups for reliable delivery.
-    Filters events to only notify about interesting field changes.
+    Cursor durability: the last-seen event id is persisted to a file (default
+    /meta-core/cursors/meta-stremio-meta.cursor) so the consumer resumes
+    cleanly after a restart. On a gap event the cursor is reset to the
+    server-supplied resumeFrom value.
     """
 
-    STREAM = "meta:events"
-    GROUP = "stremio-consumer"
-    CONSUMER = "stremio-1"
+    STREAM_PATH = "/api/events/meta"
 
-    # Fields that trigger cache invalidation
+    # Fields whose changes invalidate cached Stremio responses.
     INTERESTING_FIELDS = [
         "tmdb", "tmdbId", "title", "poster", "backdrop",
         "imdbId", "imdbid", "year", "type", "rating",
-        "description", "plot", "genres", "fileType"
+        "description", "plot", "genres", "fileType",
     ]
 
-    def __init__(self, redis_client: Redis, prefix: str = ""):
-        if not REDIS_AVAILABLE:
-            raise ImportError("redis package not installed")
-
-        self._client = redis_client
-        self._prefix = prefix
+    def __init__(
+        self,
+        api_url: str,
+        cursor_path: str = "/meta-core/cursors/meta-stremio-meta.cursor",
+        request_timeout: int = 35,
+    ) -> None:
+        self._api_url = api_url.rstrip("/")
+        self._cursor_path = cursor_path
+        self._request_timeout = request_timeout
         self._running = False
         self._thread: Optional[threading.Thread] = None
         self._callbacks: List[Callable[[str, str], None]] = []
+        self._cursor: str = self._load_cursor()
+        self._stop_event = threading.Event()
 
-    def on_change(self, callback: Callable[[str, str], None]) -> 'MetaConsumer':
-        """
-        Register callback for metadata changes.
-
-        callback(key, event_type): Called when a file's metadata changes.
-            - key: The Redis key that changed (e.g., "file:abc123/tmdb")
-            - event_type: The operation type ("set", "del", etc.)
-        """
+    def on_change(self, callback: Callable[[str, str], None]) -> "MetaConsumer":
+        """Register a callback. See module docstring for the contract."""
         self._callbacks.append(callback)
         return self
 
     def start(self) -> None:
-        """Start consuming events."""
         if self._running:
             return
-
-        # Create consumer group if not exists
-        try:
-            self._client.xgroup_create(
-                self.STREAM, self.GROUP, id="$", mkstream=True
-            )
-            logger.info(f"[MetaConsumer] Created consumer group '{self.GROUP}'")
-        except Exception as e:
-            if "BUSYGROUP" not in str(e):
-                logger.warning(f"[MetaConsumer] Group creation error: {e}")
-
         self._running = True
+        self._stop_event.clear()
         self._thread = threading.Thread(target=self._consume_loop, daemon=True)
         self._thread.start()
-        logger.info("[MetaConsumer] Started consuming meta:events stream")
+        logger.info(
+            "[MetaConsumer] Started SSE consumer for %s (cursor=%s)",
+            f"{self._api_url}{self.STREAM_PATH}",
+            self._cursor or "$",
+        )
 
     def stop(self) -> None:
-        """Stop consuming events."""
         if not self._running:
             return
-
         self._running = False
+        self._stop_event.set()
         if self._thread:
             self._thread.join(timeout=5)
             self._thread = None
+        # Best-effort flush of any unsaved cursor advance.
+        self._save_cursor()
         logger.info("[MetaConsumer] Stopped")
 
     def is_running(self) -> bool:
-        """Check if consumer is running."""
         return self._running
 
+    # ---------- internals -----------------------------------------------
+
+    def _load_cursor(self) -> str:
+        try:
+            with open(self._cursor_path, "r", encoding="utf-8") as f:
+                value = f.read().strip()
+                if value:
+                    logger.info("[MetaConsumer] Resuming from cursor %s", value)
+                return value
+        except FileNotFoundError:
+            return ""
+        except Exception as e:
+            logger.warning(
+                "[MetaConsumer] Could not read cursor %s: %s", self._cursor_path, e
+            )
+            return ""
+
+    def _save_cursor(self) -> None:
+        if not self._cursor:
+            return
+        try:
+            Path(self._cursor_path).parent.mkdir(parents=True, exist_ok=True)
+            with open(self._cursor_path, "w", encoding="utf-8") as f:
+                f.write(self._cursor)
+        except Exception as e:
+            logger.warning(
+                "[MetaConsumer] Could not write cursor %s: %s", self._cursor_path, e
+            )
+
     def _consume_loop(self) -> None:
-        """Main consumer loop."""
+        backoff = 0.5
+        max_backoff = 30.0
         while self._running:
             try:
-                # Read from stream with 2s block timeout
-                entries = self._client.xreadgroup(
-                    self.GROUP,
-                    self.CONSUMER,
-                    {self.STREAM: ">"},
-                    count=100,
-                    block=2000
-                )
-
-                if not entries:
-                    continue
-
-                for stream_name, messages in entries:
-                    for msg_id, fields in messages:
-                        self._process_event(msg_id, fields)
-                        # Acknowledge
-                        self._client.xack(self.STREAM, self.GROUP, msg_id)
-
+                self._stream_once()
+                backoff = 0.5  # successful read → reset backoff
             except Exception as e:
-                if self._running:
-                    logger.error(f"[MetaConsumer] Error consuming: {e}")
+                if not self._running:
+                    return
+                logger.warning(
+                    "[MetaConsumer] SSE error, reconnecting in %.1fs: %s",
+                    backoff, e,
+                )
+                # interruptible sleep
+                self._stop_event.wait(timeout=backoff)
+                backoff = min(backoff * 2, max_backoff)
 
-    def _process_event(self, msg_id: str, fields: dict) -> None:
-        """Process a single event from the stream."""
-        # Handle both bytes and string keys
-        key = fields.get(b"key") or fields.get("key")
-        event_type = fields.get(b"type") or fields.get("type")
+    def _stream_once(self) -> None:
+        url = f"{self._api_url}{self.STREAM_PATH}"
+        req = urllib.request.Request(url, method="GET")
+        req.add_header("Accept", "text/event-stream")
+        req.add_header("Cache-Control", "no-cache")
+        if self._cursor:
+            req.add_header("Last-Event-ID", self._cursor)
 
-        if key is None or event_type is None:
+        # urllib.request doesn't natively grok SSE; we open the socket and
+        # read line-by-line until the connection drops or the thread is
+        # asked to stop.
+        with urllib.request.urlopen(req, timeout=self._request_timeout) as resp:
+            buffer = ""
+            since_save = time.time()
+            for chunk in resp:
+                if not self._running:
+                    return
+                buffer += chunk.decode("utf-8", errors="replace")
+                # Process complete events (separated by blank lines).
+                while "\n\n" in buffer:
+                    event_chunk, buffer = buffer.split("\n\n", 1)
+                    self._handle_chunk(event_chunk)
+                # Flush cursor periodically.
+                if time.time() - since_save > 2.0:
+                    self._save_cursor()
+                    since_save = time.time()
+
+    def _handle_chunk(self, chunk: str) -> None:
+        event_id = ""
+        event_type = ""
+        data_lines: List[str] = []
+        for line in chunk.split("\n"):
+            if not line or line.startswith(":"):
+                continue  # comment / heartbeat
+            colon = line.find(":")
+            if colon == -1:
+                continue
+            field = line[:colon]
+            value = line[colon + 1:]
+            if value.startswith(" "):
+                value = value[1:]
+            if field == "id":
+                event_id = value
+            elif field == "event":
+                event_type = value
+            elif field == "data":
+                data_lines.append(value)
+
+        if not data_lines:
             return
 
-        # Decode bytes if needed
-        key = key.decode() if isinstance(key, bytes) else key
-        event_type = event_type.decode() if isinstance(event_type, bytes) else event_type
+        try:
+            data = json.loads("\n".join(data_lines))
+        except json.JSONDecodeError as e:
+            logger.warning(
+                "[MetaConsumer] Skipping unparseable event %s: %s",
+                event_id or "(no id)", e,
+            )
+            return
 
-        # Filter: only care about interesting field changes
+        # `gap` is a synthetic event signalling that the server trimmed the
+        # cursor out of retention. Reset to the resumeFrom value and
+        # continue. Stremio doesn't bootstrap from the stream so loss is
+        # acceptable — it just means the cache may be stale for a moment.
+        if event_type == "gap":
+            resume = data.get("resumeFrom")
+            if isinstance(resume, str) and resume:
+                self._cursor = resume
+            logger.warning(
+                "[MetaConsumer] Stream cursor trimmed; resuming from %s",
+                resume,
+            )
+            return
+
+        key = data.get("key")
+        if not key or not event_type:
+            return
         if not self._is_interesting_field(key):
+            # Still advance the cursor so we don't replay this event after
+            # a restart.
+            if event_id:
+                self._cursor = event_id
             return
 
-        # Notify callbacks
         for callback in self._callbacks:
             try:
                 callback(key, event_type)
             except Exception as e:
-                logger.error(f"[MetaConsumer] Callback error: {e}")
+                logger.error("[MetaConsumer] Callback error: %s", e)
+
+        if event_id:
+            self._cursor = event_id
 
     def _is_interesting_field(self, key: str) -> bool:
-        """Check if the key represents an interesting field."""
-        # Key format: file:{hashId}/{field}
         for field in self.INTERESTING_FIELDS:
             if key.endswith(f"/{field}"):
                 return True

src/storage/meta_core_api_client.py

@@ -0,0 +1,120 @@
+"""
+MetaCoreApiClient — HTTP read client against meta-core.
+
+Mirrors the subset of meta-stremio's storage-read surface that previously
+spoke ioredis directly. Backs the LeaderStorage when meta-core's leader info
+omits redis_url (post api-mediated-access PR D), so the addon keeps working
+without direct Redis access.
+
+Endpoints used:
+  - GET /meta                  → enumerate hashIds
+  - GET /meta/{hash}           → flat metadata map for one file
+  - GET /meta/{hash}/{key}     → single property (text/plain)
+  - GET /api/file/{cid}/info   → resolve CID to file path
+  - GET /health                → connectivity probe
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Optional, List, Dict
+from urllib.parse import quote
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError, URLError
+
+logger = logging.getLogger(__name__)
+
+
+class MetaCoreApiClient:
+    """HTTP client for meta-core's /meta/* read surface."""
+
+    def __init__(self, api_url: str, timeout: int = 30):
+        self._api_url = api_url.rstrip('/')
+        self._timeout = timeout
+
+    # ----- enumeration ---------------------------------------------------
+
+    def list_hash_ids(self) -> List[str]:
+        """GET /meta — returns every hashId in file:__index__."""
+        body = self._get_json('/meta')
+        if not body:
+            return []
+        ids = body.get('hashIds')
+        return ids if isinstance(ids, list) else []
+
+    def count_hash_ids(self) -> int:
+        body = self._get_json('/meta')
+        if not body:
+            return 0
+        return int(body.get('count', 0))
+
+    # ----- per-file -----------------------------------------------------
+
+    def get_metadata_flat(self, hash_id: str) -> Optional[Dict[str, str]]:
+        """GET /meta/{hash}. Returns the flat property map or None on 404."""
+        body = self._get_json(f'/meta/{quote(hash_id, safe="")}', allow_404=True)
+        if not body:
+            return None
+        meta = body.get('metadata')
+        return meta if isinstance(meta, dict) else None
+
+    def get_property(self, hash_id: str, key: str) -> Optional[str]:
+        """GET /meta/{hash}/{key}. meta-core returns text/plain (or 404)."""
+        path = f'/meta/{quote(hash_id, safe="")}/{quote(key, safe="/")}'
+        url = f"{self._api_url}{path}"
+        try:
+            req = Request(url, headers={'Accept': 'text/plain'})
+            with urlopen(req, timeout=self._timeout) as resp:
+                text = resp.read().decode('utf-8', errors='replace')
+                return text if text != '' else None
+        except HTTPError as e:
+            if e.code == 404:
+                return None
+            logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
+            return None
+        except URLError as e:
+            logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
+            return None
+
+    # ----- CID resolution -----------------------------------------------
+
+    def resolve_file_path(self, cid: str) -> Optional[str]:
+        """GET /api/file/{cid}/info → relative path (or None on miss)."""
+        path = f'/api/file/{quote(cid, safe="")}/info'
+        body = self._get_json(path)
+        if not body or not body.get('exists'):
+            return None
+        return body.get('filePath') or None
+
+    # ----- health -------------------------------------------------------
+
+    def health(self) -> bool:
+        try:
+            req = Request(f"{self._api_url}/health", headers={'Accept': 'application/json'})
+            with urlopen(req, timeout=min(self._timeout, 5)) as resp:
+                return resp.status == 200
+        except Exception:
+            return False
+
+    # ----- internals ----------------------------------------------------
+
+    def _get_json(self, path: str, allow_404: bool = False) -> Optional[dict]:
+        url = f"{self._api_url}{path}"
+        try:
+            req = Request(url, headers={'Accept': 'application/json'})
+            with urlopen(req, timeout=self._timeout) as resp:
+                raw = resp.read().decode('utf-8', errors='replace')
+                if not raw:
+                    return None
+                return json.loads(raw)
+        except HTTPError as e:
+            if allow_404 and e.code == 404:
+                return None
+            logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
+            return None
+        except URLError as e:
+            logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
+            return None
+        except json.JSONDecodeError as e:
+            logger.warning("[MetaCoreApiClient] GET %s — invalid JSON: %s", path, e)
+            return None

www/index.html

@@ -477,11 +477,12 @@
 
     <script>
         const baseUrl = window.location.origin;
-        const manifestUrl = `${baseUrl}/manifest.json`;
+        const apiPrefix = window.API_PREFIX || '';
+        const manifestUrl = `${baseUrl}${apiPrefix}/manifest.json`;
 
         // Set URLs
         document.getElementById('manifestUrl').textContent = manifestUrl;
-        document.getElementById('installLink').href = `stremio://${window.location.host}/manifest.json`;
+        document.getElementById('installLink').href = `stremio://${window.location.host}${apiPrefix}/manifest.json`;
 
         function copyUrl() {
             navigator.clipboard.writeText(manifestUrl).then(() => {