From 56eced073ad41b17e10f6ee2ca31e79078ba60af Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 24 May 2026 17:26:29 +0200 Subject: [PATCH] adaptation related to meta-core api changes --- src/storage/leader_client.py | 9 ++ src/storage/leader_storage.py | 36 ++++- src/storage/meta_consumer.py | 252 +++++++++++++++++++++++----------- 3 files changed, 208 insertions(+), 89 deletions(-) diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py index 4698dde..ed1df16 100644 --- a/src/storage/leader_client.py +++ b/src/storage/leader_client.py @@ -137,6 +137,15 @@ 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 — + # log a warning so it surfaces in the operator's face. + if data.get('redisUrl'): + print( + "[LeaderClient] WARNING: meta-core still publishes redisUrl; " + "direct Redis access was retired by the api-mediated-access " + "lockdown. Verify meta-core version." + ) self._cached_urls = URLsResponse( hostname=data.get('hostname', ''), base_url=data.get('baseUrl', ''), diff --git a/src/storage/leader_storage.py b/src/storage/leader_storage.py index 4cee553..aad661b 100644 --- a/src/storage/leader_storage.py +++ b/src/storage/leader_storage.py @@ -122,9 +122,23 @@ 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: redis_url may be empty when + # meta-core no longer advertises it. Synthesise from the + # hostname in that case so meta-stremio's read paths still + # work until they're migrated to HTTP. The synthesised URL + # assumes the docker-network `meta-core` alias or the + # canonical hostname is reachable on port 6379. + redis_url = info.redis_url if info else None + if info and not redis_url and info.hostname: + redis_url = f"redis://{info.hostname}:6379" + print( + "[LeaderStorage] redis_url absent in leader info; " + f"falling back to {redis_url}" + ) + + if info and redis_url: + print(f"[LeaderStorage] Found leader at {redis_url} (attempt {attempt})") + self._connect_to_redis(redis_url) if self._connected: # Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access) @@ -274,11 +288,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}") diff --git a/src/storage/meta_consumer.py b/src/storage/meta_consumer.py index 136c55b..aaa5eae 100644 --- a/src/storage/meta_consumer.py +++ b/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:/" 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