From babadb728fefbe77365469525201322cba66c8d6 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 8 Feb 2026 21:54:21 +0100 Subject: [PATCH] new flat KV structure --- docker-compose.yml | 2 +- src/storage/leader_storage.py | 147 +++++++++++++++++++++----------- src/storage/meta_consumer.py | 154 ++++++++++++++++++++++++++++++++++ src/storage/redis_storage.py | 102 ++++++++++++---------- 4 files changed, 311 insertions(+), 94 deletions(-) create mode 100644 src/storage/meta_consumer.py diff --git a/docker-compose.yml b/docker-compose.yml index 37f4cf4..32ce1a8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,7 +20,7 @@ services: - CACHE_DIR=/data/cache - STORAGE_MODE=${STORAGE_MODE:-leader} - REDIS_URL=${REDIS_URL:-redis://redis:6379} - - REDIS_PREFIX=${REDIS_PREFIX:-meta-sort:} + - REDIS_PREFIX=${REDIS_PREFIX:-} - SEGMENT_DURATION=${SEGMENT_DURATION:-4} - PREFETCH_SEGMENTS=${PREFETCH_SEGMENTS:-4} - SCHEME=${SCHEME:-auto} diff --git a/src/storage/leader_storage.py b/src/storage/leader_storage.py index 6af4f59..8078dcc 100644 --- a/src/storage/leader_storage.py +++ b/src/storage/leader_storage.py @@ -13,6 +13,7 @@ from typing import List, Optional, Callable from .provider import StorageProvider, VideoMetadata from .leader_client import LeaderClient, LeaderLockInfo +from .meta_consumer import MetaConsumer # Import webdav_client to configure it after leader discovery import webdav_client @@ -28,7 +29,7 @@ except ImportError: # 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:') +REDIS_PREFIX = os.environ.get('REDIS_PREFIX', '') class LeaderStorage(StorageProvider): @@ -59,6 +60,11 @@ class LeaderStorage(StorageProvider): self._connected = False self._lock = threading.Lock() + # Meta events consumer for real-time updates + self._meta_consumer: Optional[MetaConsumer] = None + self._video_cache: dict = {} + self._cache_dirty = True + # Callbacks self._on_ready_callbacks: List[Callable[[], None]] = [] self._on_disconnect_callbacks: List[Callable[[], None]] = [] @@ -150,6 +156,17 @@ class LeaderStorage(StorageProvider): print(f"[LeaderStorage] Connected to Redis at {url}") + # Start meta events consumer for real-time updates + try: + self._meta_consumer = MetaConsumer(self._client, self._prefix) + 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}") + + # Mark cache as dirty + self._cache_dirty = True + # Notify ready self._notify_ready() @@ -165,6 +182,14 @@ class LeaderStorage(StorageProvider): def _disconnect_redis_unlocked(self) -> None: """Disconnect from Redis (without lock).""" + # Stop meta consumer + if self._meta_consumer: + try: + self._meta_consumer.stop() + except Exception: + pass + self._meta_consumer = None + if self._client: try: self._client.close() @@ -173,6 +198,10 @@ class LeaderStorage(StorageProvider): self._client = None self._connected = False + # Clear cache + self._video_cache.clear() + self._cache_dirty = True + def _notify_ready(self) -> None: """Notify ready callbacks.""" for callback in self._on_ready_callbacks: @@ -189,6 +218,22 @@ class LeaderStorage(StorageProvider): except Exception as e: print(f"[LeaderStorage] Error in disconnect callback: {e}") + def _on_metadata_change(self, key: str, event_type: str) -> None: + """Handle metadata change events from meta:events stream.""" + # Extract hash_id from key: file:midhash256:abc123/tmdb -> midhash256:abc123 + if not key.startswith("file:"): + return + + # Remove "file:" prefix and get the hash part (before /) + parts = key[5:].split("/") + if len(parts) >= 1: + hash_id = parts[0] + + # Invalidate cache for this file + self._video_cache.pop(hash_id, None) + self._cache_dirty = True + print(f"[LeaderStorage] Metadata updated: {hash_id} ({event_type})") + # ======================================================================== # Event Registration # ======================================================================== @@ -209,9 +254,23 @@ class LeaderStorage(StorageProvider): # 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 _get_file_key_prefix(self, hash_id: str) -> str: + """Get the Redis key prefix for a file's flat keys.""" + 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.""" + 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):] + value = self._client.get(key) + if value: + data[field] = value if isinstance(value, str) else value.decode() + + return data def _resolve_path(self, path: str) -> str: """Resolve path by prepending FILES_PATH if needed.""" @@ -406,22 +465,23 @@ class LeaderStorage(StorageProvider): 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 + # Get all hash IDs from index + index_key = f"{self._prefix}file:__index__" + hash_ids = self._client.smembers(index_key) + + for hash_id in hash_ids: + hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode() + + # Get all metadata for this file + data = self._get_file_metadata(hash_id_str) # Check file type - only include video files - file_type = self._client.hget(key, 'fileType') - if file_type != 'video': + if not data or data.get('fileType') != 'video': continue - hash_id = key.replace(f"{self._prefix}file:", "") - data = self._client.hgetall(key) - video = self._parse_video(hash_id, data) + video = self._parse_video(hash_id_str, data) if video and video.file_path: videos.append(video) @@ -438,9 +498,8 @@ class LeaderStorage(StorageProvider): return None try: - key = self._get_file_key(hash_id) - data = self._client.hgetall(key) - return self._parse_video(hash_id, data) + data = self._get_file_metadata(hash_id) + return self._parse_video(hash_id, data) if data else None except Exception as e: print(f"[LeaderStorage] Error getting video {hash_id}: {e}") return None @@ -470,18 +529,19 @@ class LeaderStorage(StorageProvider): 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 + # 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/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) + # 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) except Exception as e: print(f"[LeaderStorage] Error finding video by IMDB ID {imdb_id}: {e}") @@ -492,7 +552,7 @@ class LeaderStorage(StorageProvider): """ Get the file path for a file by its CID. - Looks up file:{cid} in Redis and returns the path. + 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. @@ -502,18 +562,19 @@ class LeaderStorage(StorageProvider): return None try: - key = self._get_file_key(cid) + prefix = self._get_file_key_prefix(cid) # Try 'path' field first (relative path) - path = self._client.hget(key, 'path') + path = self._client.get(f"{prefix}path") if path: - return path + return path if isinstance(path, str) else path.decode() # Fall back to 'filePath' (full path from meta-sort) - file_path = self._client.hget(key, 'filePath') + file_path = self._client.get(f"{prefix}filePath") if file_path: + file_path_str = file_path if isinstance(file_path, str) else file_path.decode() # Convert absolute path to relative by removing /files/ prefix - if file_path.startswith('/files/'): - return file_path[7:] # Remove '/files/' - return file_path + if file_path_str.startswith('/files/'): + return file_path_str[7:] # Remove '/files/' + return file_path_str return None except Exception as e: print(f"[LeaderStorage] Error getting path for CID {cid}: {e}") @@ -525,19 +586,9 @@ class LeaderStorage(StorageProvider): return 0 try: - # Try index first + # Use index set for accurate count 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 + return self._client.scard(index_key) except Exception as e: print(f"[LeaderStorage] Error counting videos: {e}") diff --git a/src/storage/meta_consumer.py b/src/storage/meta_consumer.py new file mode 100644 index 0000000..136c55b --- /dev/null +++ b/src/storage/meta_consumer.py @@ -0,0 +1,154 @@ +""" +Meta Events Consumer for meta-stremio. + +Consumes the meta:events Redis stream published by meta-core. +Provides real-time notifications when metadata changes. +""" +from __future__ import annotations + +import threading +import logging +from typing import Callable, List, Optional + +try: + from redis import Redis + REDIS_AVAILABLE = True +except ImportError: + REDIS_AVAILABLE = False + Redis = None + +logger = logging.getLogger(__name__) + + +class MetaConsumer: + """ + Consumes meta:events stream for real-time metadata updates. + + Uses Redis consumer groups for reliable delivery. + Filters events to only notify about interesting field changes. + """ + + STREAM = "meta:events" + GROUP = "stremio-consumer" + CONSUMER = "stremio-1" + + # Fields that trigger cache invalidation + INTERESTING_FIELDS = [ + "tmdb", "tmdbId", "title", "poster", "backdrop", + "imdbId", "imdbid", "year", "type", "rating", + "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 + self._running = False + self._thread: Optional[threading.Thread] = None + self._callbacks: List[Callable[[str, str], None]] = [] + + 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.) + """ + 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._thread = threading.Thread(target=self._consume_loop, daemon=True) + self._thread.start() + logger.info("[MetaConsumer] Started consuming meta:events stream") + + def stop(self) -> None: + """Stop consuming events.""" + if not self._running: + return + + self._running = False + if self._thread: + self._thread.join(timeout=5) + self._thread = None + logger.info("[MetaConsumer] Stopped") + + def is_running(self) -> bool: + """Check if consumer is running.""" + return self._running + + def _consume_loop(self) -> None: + """Main consumer loop.""" + 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) + + except Exception as e: + if self._running: + logger.error(f"[MetaConsumer] Error consuming: {e}") + + 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") + + if key is None or event_type is None: + 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 + + # Filter: only care about interesting field changes + if not self._is_interesting_field(key): + return + + # Notify callbacks + for callback in self._callbacks: + try: + callback(key, event_type) + except Exception as e: + logger.error(f"[MetaConsumer] Callback error: {e}") + + 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 + return False diff --git a/src/storage/redis_storage.py b/src/storage/redis_storage.py index 05ca6b2..43c4128 100644 --- a/src/storage/redis_storage.py +++ b/src/storage/redis_storage.py @@ -29,8 +29,8 @@ 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 + Reads video metadata from Redis using flat key schema: + - file:{hashId}/{field} -> String value for each metadata property """ def __init__(self, url: str = None, prefix: str = None): @@ -78,7 +78,7 @@ class RedisStorage(StorageProvider): """ Get the file path for a file by its CID. - Looks up file:{cid} in Redis and returns the path. + 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. @@ -88,26 +88,41 @@ class RedisStorage(StorageProvider): return None try: - key = self._get_file_key(cid) + prefix = self._get_file_key_prefix(cid) # Try 'path' field first (relative path) - path = self._client.hget(key, 'path') + path = self._client.get(f"{prefix}path") if path: - return path + return path if isinstance(path, str) else path.decode() # Fall back to 'filePath' (full path from meta-sort) - file_path = self._client.hget(key, 'filePath') + file_path = self._client.get(f"{prefix}filePath") if file_path: + file_path_str = file_path if isinstance(file_path, str) else file_path.decode() # Convert absolute path to relative by removing /files/ prefix - if file_path.startswith('/files/'): - return file_path[7:] # Remove '/files/' - return file_path + if file_path_str.startswith('/files/'): + return file_path_str[7:] # Remove '/files/' + return file_path_str 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 _get_file_key_prefix(self, hash_id: str) -> str: + """Get the Redis key prefix for a file's flat keys.""" + 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.""" + 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):] + value = self._client.get(key) + if value: + data[field] = value if isinstance(value, str) else value.decode() + + return data def _parse_video(self, hash_id: str, data: dict) -> Optional[VideoMetadata]: """Parse Redis hash data into VideoMetadata. @@ -280,30 +295,26 @@ class RedisStorage(StorageProvider): 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 + # Get all hash IDs from index + index_key = f"{self._prefix}file:__index__" + hash_ids = self._client.smembers(index_key) - # Check key type - only process hash keys - key_type = self._client.type(key) - if key_type != 'hash': - continue + for hash_id in hash_ids: + hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode() + + # Get all metadata for this file + data = self._get_file_metadata(hash_id_str) # Check file type - only include video files - file_type = self._client.hget(key, 'fileType') - if file_type != 'video': + if not data or data.get('fileType') != '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) + video = self._parse_video(hash_id_str, data) if video and video.file_path: videos.append(video) + except Exception as e: print(f"[RedisStorage] Error getting all videos: {e}") @@ -317,9 +328,8 @@ class RedisStorage(StorageProvider): return None try: - key = self._get_file_key(hash_id) - data = self._client.hgetall(key) - return self._parse_video(hash_id, data) + data = self._get_file_metadata(hash_id) + return self._parse_video(hash_id, data) if data else None except Exception as e: print(f"[RedisStorage] Error getting video {hash_id}: {e}") return None @@ -349,15 +359,19 @@ class RedisStorage(StorageProvider): 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) + # 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) except Exception as e: print(f"[RedisStorage] Error finding video by IMDB ID {imdb_id}: {e}") @@ -370,11 +384,9 @@ class RedisStorage(StorageProvider): return 0 try: - pattern = f"{self._prefix}file:*" - count = 0 - for _ in self._client.scan_iter(pattern, count=100): - count += 1 - return count + # Use index set for accurate count + index_key = f"{self._prefix}file:__index__" + return self._client.scard(index_key) except Exception as e: print(f"[RedisStorage] Error counting videos: {e}") return 0