feat: implement Redis caching for catalog retrieval and add Redis service integration

TimilsinaBimal · TimilsinaBimal · commit ce9b6814695a · 2025-12-29T13:23:22.000+05:45
- Introduced RedisService for managing Redis operations.
- Updated catalog retrieval to cache results in Redis for improved performance.
- Added CATALOG_CACHE_TTL setting for cache expiration.
- Refactored CatalogService and TokenStore to utilize RedisService for data storage and retrieval.
diff --git a/app/api/endpoints/catalogs.py b/app/api/endpoints/catalogs.py
@@ -1,14 +1,15 @@
+import json
+
 from fastapi import APIRouter, HTTPException, Response
 from loguru import logger
 
+from app.core.config import settings
 from app.core.security import redact_token
-from app.services.recommendation.catalog_service import CatalogService
+from app.services.recommendation.catalog_service import catalog_service
+from app.services.redis_service import redis_service
 
 router = APIRouter()
 
-# Initialize catalog service (singleton)
-_catalog_service = CatalogService()
-
 
 def _clean_meta(meta: dict) -> dict:
     """Return a sanitized Stremio meta object without internal fields.
@@ -45,8 +46,14 @@ async def get_catalog(type: str, id: str, response: Response, token: str):
     This endpoint delegates all logic to CatalogService facade.
     """
     try:
+        # catalog_key
+        catalog_key = f"watchly:catalog:{token}:{type}:{id}"
+        cached_data = await redis_service.get(catalog_key)
+        if cached_data:
+            return json.loads(cached_data)
+
         # Delegate to catalog service facade
-        recommendations, headers = await _catalog_service.get_catalog(token, type, id)
+        recommendations, headers = await catalog_service.get_catalog(token, type, id)
 
         # Set response headers
         for key, value in headers.items():
@@ -56,7 +63,9 @@ async def get_catalog(type: str, id: str, response: Response, token: str):
         cleaned = [_clean_meta(m) for m in recommendations]
         cleaned = [m for m in cleaned if m is not None]
 
-        return {"metas": cleaned}
+        data = {"metas": cleaned}
+        await redis_service.set(catalog_key, json.dumps(data), settings.CATALOG_CACHE_TTL)
+        return data
 
     except HTTPException:
         raise
diff --git a/app/core/config.py b/app/core/config.py
@@ -38,6 +38,8 @@ class Settings(BaseSettings):
     RECOMMENDATION_SOURCE_ITEMS_LIMIT: int = 10
     LIBRARY_ITEMS_LIMIT: int = 20
 
+    CATALOG_CACHE_TTL: int = 12 * 60 * 60  # 12 hours
+
     # AI
     DEFAULT_GEMINI_MODEL: str = "gemma-3-27b-it"
     GEMINI_API_KEY: str | None = None
diff --git a/app/services/recommendation/catalog_service.py b/app/services/recommendation/catalog_service.py
@@ -329,3 +329,6 @@ async def _get_recommendations(
             recommendations = []
 
         return recommendations
+
+
+catalog_service = CatalogService()
diff --git a/app/services/redis_service.py b/app/services/redis_service.py
@@ -0,0 +1,116 @@
+from typing import Any
+
+import redis.asyncio as redis
+from loguru import logger
+
+from app.core.config import settings
+
+
+class RedisService:
+    def __init__(self) -> None:
+        self._client: redis.Redis | None = None
+        if not settings.REDIS_URL:
+            logger.warning("REDIS_URL is not set. Redis operations will fail until configured.")
+
+    async def get_client(self) -> redis.Redis:
+        if self._client is None:
+            logger.info("Creating Redis client for RedisService")
+            self._client = redis.from_url(
+                settings.REDIS_URL,
+                decode_responses=True,
+                encoding="utf-8",
+                socket_connect_timeout=5,
+                socket_timeout=5,
+                max_connections=getattr(settings, "REDIS_MAX_CONNECTIONS", 100),
+                health_check_interval=30,
+                socket_keepalive=True,
+            )
+        return self._client
+
+    async def set(self, key: str, value: Any, ttl: int | None = None) -> bool:
+        """Store a value in Redis with optional TTL.
+
+        Args:
+            key: The key to store the value under
+            value: The value to store (will be converted to string)
+            ttl: Optional time-to-live in seconds. If None, key never expires.
+
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            client = await self.get_client()
+            str_value = str(value)
+            if ttl is not None:
+                result = await client.setex(key, ttl, str_value)
+            else:
+                result = await client.set(key, str_value)
+            return bool(result)
+        except (redis.RedisError, OSError) as exc:
+            logger.error(f"Failed to set key '{key}' in Redis: {exc}")
+            return False
+
+    async def get(self, key: str) -> str | None:
+        """Get a value from Redis by key.
+
+        Args:
+            key: The key to retrieve
+
+        Returns:
+            The value as a string, or None if key doesn't exist or error occurred
+        """
+        try:
+            client = await self.get_client()
+            value = await client.get(key)
+            return value
+        except (redis.RedisError, OSError) as exc:
+            logger.error(f"Failed to get key '{key}' from Redis: {exc}")
+            return None
+
+    async def delete(self, key: str) -> bool:
+        """Delete a key from Redis.
+
+        Args:
+            key: The key to delete
+
+        Returns:
+            True if key was deleted, False otherwise
+        """
+        try:
+            client = await self.get_client()
+            result = await client.delete(key)
+            return bool(result)
+        except (redis.RedisError, OSError) as exc:
+            logger.error(f"Failed to delete key '{key}' from Redis: {exc}")
+            return False
+
+    async def exists(self, key: str) -> bool:
+        """Check if a key exists in Redis.
+
+        Args:
+            key: The key to check
+
+        Returns:
+            True if key exists, False otherwise
+        """
+        try:
+            client = await self.get_client()
+            result = await client.exists(key)
+            return bool(result)
+        except (redis.RedisError, OSError) as exc:
+            logger.error(f"Failed to check existence of key '{key}' in Redis: {exc}")
+            return False
+
+    async def close(self) -> None:
+        """Close and disconnect the Redis client"""
+        if self._client is not None:
+            try:
+                await self._client.close()
+                logger.info("RedisService client closed")
+            except Exception as exc:
+                logger.warning(f"Failed to close RedisService client: {exc}")
+            finally:
+                self._client = None
+
+
+redis_service = RedisService()
diff --git a/app/services/token_store.py b/app/services/token_store.py
@@ -12,6 +12,7 @@
 
 from app.core.config import settings
 from app.core.security import redact_token
+from app.services.redis_service import redis_service
 
 
 class TokenStore:
@@ -20,12 +21,9 @@ class TokenStore:
     KEY_PREFIX = settings.REDIS_TOKEN_KEY
 
     def __init__(self) -> None:
-        self._client: redis.Redis | None = None
         # Negative cache for missing tokens to avoid repeated Redis GETs
         # when external probes request non-existent tokens.
         self._missing_tokens: TTLCache = TTLCache(maxsize=10000, ttl=86400)
-        if not settings.REDIS_URL:
-            logger.warning("REDIS_URL is not set. Token storage will fail until a Redis instance is configured.")
 
         if not settings.TOKEN_SALT or settings.TOKEN_SALT == "change-me":
             logger.warning(
@@ -34,10 +32,8 @@ def __init__(self) -> None:
 
     def _ensure_secure_salt(self) -> None:
         if not settings.TOKEN_SALT or settings.TOKEN_SALT == "change-me":
-            logger.error("Refusing to store credentials because TOKEN_SALT is unset or using the insecure default.")
-            raise RuntimeError(
-                "Server misconfiguration: TOKEN_SALT must be set to a non-default value before storing" " credentials."
-            )
+            logger.error("TOKEN_SALT is unset or using the insecure default.")
+            raise RuntimeError("TOKEN_SALT must be set to a non-default value before storing credentials.")
 
     def _get_cipher(self) -> Fernet:
         salt = b"x7FDf9kypzQ1LmR32b8hWv49sKq2Pd8T"
@@ -59,59 +55,6 @@ def decrypt_token(self, enc: str) -> str:
         cipher = self._get_cipher()
         return cipher.decrypt(enc.encode("utf-8")).decode("utf-8")
 
-    async def _get_client(self) -> redis.Redis:
-        if self._client is None:
-            # Add socket timeouts to avoid hanging on Redis operations
-            import traceback
-
-            logger.info("Creating shared Redis client")
-            # Limit the number of pooled connections to avoid unbounded growth
-            # `max_connections` is forwarded to ConnectionPool.from_url
-            self._client = redis.from_url(
-                settings.REDIS_URL,
-                decode_responses=True,
-                encoding="utf-8",
-                socket_connect_timeout=5,
-                socket_timeout=5,
-                max_connections=getattr(settings, "REDIS_MAX_CONNECTIONS", 100),
-                health_check_interval=30,
-                socket_keepalive=True,
-            )
-            if getattr(self, "_creation_count", None) is None:
-                self._creation_count = 1
-            else:
-                self._creation_count += 1
-                logger.warning(
-                    f"Redis client creation invoked again (count={self._creation_count})."
-                    f" Stack:\n{''.join(traceback.format_stack())}"
-                )
-        return self._client
-
-    async def close(self) -> None:
-        """Close and disconnect the shared Redis client (call on shutdown)."""
-        if self._client is None:
-            return
-        try:
-            logger.info("Closing shared Redis client")
-            # Close client and disconnect underlying pool
-            try:
-                await self._client.close()
-            except Exception as e:
-                logger.debug(f"Silent failure closing redis client: {e}")
-            try:
-                pool = getattr(self._client, "connection_pool", None)
-                if pool is not None:
-                    # connection_pool.disconnect may be a coroutine in some redis implementations
-                    disconnect = getattr(pool, "disconnect", None)
-                    if disconnect:
-                        res = disconnect()
-                        if hasattr(res, "__await__"):
-                            await res
-            except Exception as e:
-                logger.debug(f"Silent failure disconnecting redis pool: {e}")
-        finally:
-            self._client = None
-
     def _format_key(self, token: str) -> str:
         """Format Redis key from token."""
         return f"{self.KEY_PREFIX}{token}"
@@ -145,13 +88,12 @@ async def store_user_data(self, user_id: str, payload: dict[str, Any]) -> str:
                 # Do not store plaintext passwords
                 raise RuntimeError("PASSWORD_ENCRYPT_FAILED")
 
-        client = await self._get_client()
         json_str = json.dumps(storage_data)
 
         if settings.TOKEN_TTL_SECONDS and settings.TOKEN_TTL_SECONDS > 0:
-            await client.setex(key, settings.TOKEN_TTL_SECONDS, json_str)
+            await redis_service.set(key, json_str, settings.TOKEN_TTL_SECONDS)
         else:
-            await client.set(key, json_str)
+            await redis_service.set(key, json_str)
 
         # Invalidate async LRU cache for fresh reads on subsequent requests
         try:
@@ -193,8 +135,7 @@ async def get_user_data(self, token: str) -> dict[str, Any] | None:
 
         logger.debug(f"[REDIS] Cache miss. Fetching data from redis for {token}")
         key = self._format_key(token)
-        client = await self._get_client()
-        data_raw = await client.get(key)
+        data_raw = await redis_service.get(key)
 
         if not data_raw:
             # remember negative result briefly
@@ -232,8 +173,7 @@ async def delete_token(self, token: str = None, key: str = None) -> None:
         if token:
             key = self._format_key(token)
 
-        client = await self._get_client()
-        await client.delete(key)
+        await redis_service.delete(key)
 
         # Invalidate async LRU cache so future reads reflect deletion
         try:
@@ -261,7 +201,7 @@ async def count_users(self) -> int:
         Cached for 12 hours to avoid frequent Redis scans.
         """
         try:
-            client = await self._get_client()
+            client = await redis_service.get_client()
         except (redis.RedisError, OSError) as exc:
             logger.warning(f"Cannot count users; Redis unavailable: {exc}")
             return 0
