refactor: replace lru_cache with async_lru for improved caching and performance; remove unused streams endpoint; update manifest to include dynamic catalogs

Author: TimilsinaBimal

app/api/endpoints/manifest.py

@@ -1,3 +1,5 @@
+from async_lru import alru_cache
+from fastapi import Response
 from fastapi.routing import APIRouter
 
 from app.core.config import settings
@@ -15,10 +17,7 @@ def get_base_manifest():
         "name": settings.ADDON_NAME,
         "description": "Movie and series recommendations based on your Stremio library",
         "logo": "https://raw.githubusercontent.com/TimilsinaBimal/Watchly/refs/heads/main/static/logo.png",
-        "resources": [
-            {"name": "catalog", "types": ["movie", "series"], "idPrefixes": ["tt"]},
-            {"name": "stream", "types": ["movie", "series"], "idPrefixes": ["tt"]},
-        ],
+        "resources": [{"name": "catalog", "types": ["movie", "series"], "idPrefixes": ["tt"]}],
         "types": ["movie", "series"],
         "idPrefixes": ["tt"],
         "catalogs": [
@@ -29,32 +28,40 @@ def get_base_manifest():
     }
 
 
+# Cache catalog definitions for 1 hour (3600s)
+@alru_cache(maxsize=1000, ttl=3600)
 async def fetch_catalogs(token: str | None = None):
     if not token:
         return []
+
     credentials = await resolve_user_credentials(token)
     stremio_service = StremioService(
         username=credentials.get("username") or "",
         password=credentials.get("password") or "",
         auth_key=credentials.get("authKey"),
     )
+    # Note: get_library_items is expensive, but we need it to determine *which* genre catalogs to show.
     library_items = await stremio_service.get_library_items()
     dynamic_catalog_service = DynamicCatalogService(stremio_service=stremio_service)
+
+    # Base catalogs are already in manifest, these are *extra* dynamic ones
     catalogs = await dynamic_catalog_service.get_watched_loved_catalogs(library_items=library_items)
     catalogs += await dynamic_catalog_service.get_genre_based_catalogs(library_items=library_items)
+
     return catalogs
 
 
 @router.get("/manifest.json")
 @router.get("/{token}/manifest.json")
-async def manifest(token: str | None = None):
+async def manifest(response: Response, token: str | None = None):
     """Stremio manifest endpoint with optional credential token in the path."""
     # Cache manifest for 1 day (86400 seconds)
-    # response.headers["Cache-Control"] = "public, max-age=86400"
+    response.headers["Cache-Control"] = "public, max-age=86400"
 
     base_manifest = get_base_manifest()
     if token:
         catalogs = await fetch_catalogs(token)
         if catalogs:
+            # Append dynamic catalogs to the base ones
             base_manifest["catalogs"] += catalogs
     return base_manifest

app/api/endpoints/streams.py

@@ -3,7 +3,6 @@
 from .endpoints.catalogs import router as catalogs_router
 from .endpoints.health import router as health_router
 from .endpoints.manifest import router as manifest_router
-from .endpoints.streams import router as streams_router
 from .endpoints.tokens import router as tokens_router
 
 api_router = APIRouter()
@@ -16,6 +15,5 @@ async def root():
 
 api_router.include_router(manifest_router)
 api_router.include_router(catalogs_router)
-api_router.include_router(streams_router)
 api_router.include_router(tokens_router)
 api_router.include_router(health_router)

app/api/main.py

@@ -1,6 +1,5 @@
-from functools import lru_cache
-
 import httpx
+from async_lru import alru_cache
 from loguru import logger
 
 from app.core.config import settings
@@ -46,7 +45,9 @@ async def close(self):
             await self._addon_client.aclose()
             self._addon_client = None
 
-    @lru_cache(maxsize=1000)
+    # Increase cache size significantly as this is a hot path for recommendations
+    # 5000 items * ~1KB per item = ~5MB memory, very safe
+    @alru_cache(maxsize=5000)
     async def get_addon_meta(self, type: str, id: str) -> dict:
         """Get addon metadata for a specific type and ID."""
         url = f"{self.addon_url}/meta/{type}/{id}.json"
@@ -87,7 +88,7 @@ async def _make_request(self, endpoint: str, params: dict | None = None) -> dict
             logger.error(f"TMDB API request error for {endpoint}: {e}")
             raise
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=2000)
     async def find_by_imdb_id(self, imdb_id: str) -> tuple[int | None, str | None]:
         """Find TMDB ID and type by IMDB ID."""
         try:
@@ -128,33 +129,33 @@ async def find_by_imdb_id(self, imdb_id: str) -> tuple[int | None, str | None]:
             logger.warning(f"Unexpected error finding TMDB ID for IMDB {imdb_id}: {e}")
             return None, None
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=1000)
     async def get_movie_details(self, movie_id: int) -> dict:
         """Get details of a specific movie with credits and external IDs."""
         params = {"append_to_response": "credits,external_ids"}
         return await self._make_request(f"/movie/{movie_id}", params=params)
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=1000)
     async def get_tv_details(self, tv_id: int) -> dict:
         """Get details of a specific TV series with credits and external IDs."""
         params = {"append_to_response": "credits,external_ids"}
         return await self._make_request(f"/tv/{tv_id}", params=params)
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=1000)
     async def get_recommendations(self, tmdb_id: int, media_type: str, page: int = 1) -> dict:
         """Get recommendations based on TMDB ID and media type."""
         params = {"page": page}
         endpoint = f"/{media_type}/{tmdb_id}/recommendations"
         return await self._make_request(endpoint, params=params)
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=1000)
     async def get_similar(self, tmdb_id: int, media_type: str, page: int = 1) -> dict:
         """Get similar content based on TMDB ID and media type."""
         params = {"page": page}
         endpoint = f"/{media_type}/{tmdb_id}/similar"
         return await self._make_request(endpoint, params=params)
 
-    @lru_cache(maxsize=1000)
+    @alru_cache(maxsize=1000)
     async def get_discover(self, media_type: str, params: dict[str, str]) -> dict:
         """Get discover content based on params."""
         media_type = "movie" if media_type == "movie" else "tv"

app/services/tmdb_service.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 import redis.asyncio as redis
+from cachetools import TTLCache
 from cryptography.fernet import Fernet, InvalidToken
 from loguru import logger
 
@@ -20,6 +21,9 @@ class TokenStore:
     def __init__(self) -> None:
         self._client: redis.Redis | None = None
         self._cipher: Fernet | None = None
+        # Cache decrypted payloads for 1 day (86400s) to reduce Redis hits
+        # Max size 5000 allows many active users without eviction
+        self._payload_cache: TTLCache = TTLCache(maxsize=5000, ttl=86400)
 
         if not settings.REDIS_URL:
             logger.warning("REDIS_URL is not set. Token storage will fail until a Redis instance is configured.")
@@ -93,15 +97,22 @@ async def store_payload(self, payload: dict[str, Any]) -> tuple[str, bool]:
         if settings.TOKEN_TTL_SECONDS and settings.TOKEN_TTL_SECONDS > 0:
             await client.setex(key, settings.TOKEN_TTL_SECONDS, encrypted_value)
             logger.info(
-                "Stored encrypted credential payload with TTL %s seconds",
-                settings.TOKEN_TTL_SECONDS,
+                f"Stored encrypted credential payload with TTL {settings.TOKEN_TTL_SECONDS} seconds",
             )
         else:
             await client.set(key, encrypted_value)
             logger.info("Stored encrypted credential payload without expiration")
+
+        # Cache the new payload immediately to avoid next-read hit
+        self._payload_cache[token] = normalized
+
         return token, not bool(existing)
 
     async def get_payload(self, token: str) -> dict[str, Any] | None:
+        # Check local LRU cache first
+        if token in self._payload_cache:
+            return self._payload_cache[token]
+
         hashed = self._hash_token(token)
         key = self._format_key(hashed)
         client = await self._get_client()
@@ -113,7 +124,11 @@ async def get_payload(self, token: str) -> dict[str, Any] | None:
         try:
             # Decrypt -> JSON Decode
             decrypted_json = self._get_cipher().decrypt(encrypted_raw.encode()).decode("utf-8")
-            return json.loads(decrypted_json)
+            payload = json.loads(decrypted_json)
+
+            # Cache for subsequent reads
+            self._payload_cache[token] = payload
+            return payload
         except (InvalidToken, json.JSONDecodeError, UnicodeDecodeError):
             logger.warning("Failed to decrypt or decode cached payload for token. Key might have changed.")
             return None
@@ -124,12 +139,16 @@ async def delete_token(self, token: str) -> None:
         client = await self._get_client()
         await client.delete(key)
 
+        # Invalidate local cache
+        if token in self._payload_cache:
+            del self._payload_cache[token]
+
     async def iter_payloads(self) -> AsyncIterator[tuple[str, dict[str, Any]]]:
         """Iterate over all stored payloads, yielding key and payload."""
         try:
             client = await self._get_client()
         except (redis.RedisError, OSError) as exc:
-            logger.warning("Skipping credential iteration; Redis unavailable: %s", exc)
+            logger.warning(f"Skipping credential iteration; Redis unavailable: {exc}")
             return
 
         pattern = f"{self.KEY_PREFIX}*"
@@ -140,7 +159,7 @@ async def iter_payloads(self) -> AsyncIterator[tuple[str, dict[str, Any]]]:
                 try:
                     encrypted_raw = await client.get(key)
                 except (redis.RedisError, OSError) as exc:
-                    logger.warning("Failed to fetch payload for %s: %s", key, exc)
+                    logger.warning(f"Failed to fetch payload for {key}: {exc}")
                     continue
 
                 if encrypted_raw is None:
@@ -150,12 +169,12 @@ async def iter_payloads(self) -> AsyncIterator[tuple[str, dict[str, Any]]]:
                     decrypted_json = cipher.decrypt(encrypted_raw.encode()).decode("utf-8")
                     payload = json.loads(decrypted_json)
                 except (InvalidToken, json.JSONDecodeError, UnicodeDecodeError):
-                    logger.warning("Failed to decrypt payload for key %s. Skipping.", key)
+                    logger.warning(f"Failed to decrypt payload for key {key}. Skipping.")
                     continue
 
                 yield key, payload
         except (redis.RedisError, OSError) as exc:
-            logger.warning("Failed to scan credential tokens: %s", exc)
+            logger.warning(f"Failed to scan credential tokens: {exc}")
 
 
 token_store = TokenStore()

app/services/token_store.py

@@ -5,6 +5,7 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+    "async-lru>=2.0.5",
     "cachetools>=6.2.2",
     "cryptography>=46.0.3",
     "fastapi>=0.104.1",

pyproject.toml

@@ -7,3 +7,4 @@ loguru>=0.7.2
 cachetools>=6.2.2
 redis>=5.0.1
 cryptography>=41.0.0
+async-lru>=2.0.4