feat: Refactor catalog fetching and update migration task handling

Author: TimilsinaBimal

app/api/endpoints/health.py

@@ -1,36 +1,8 @@
 from fastapi import APIRouter
-from loguru import logger
-
-from app.services.token_store import token_store
 
 router = APIRouter(tags=["health"])
 
 
 @router.get("/health", summary="Simple readiness probe")
 async def health_check() -> dict[str, str]:
     return {"status": "ok"}
-
-
-@router.get("/metrics", summary="Runtime metrics (lightweight)")
-async def metrics() -> dict:
-    """Return lightweight runtime metrics useful for diagnosing Redis connection growth."""
-    try:
-        client = await token_store._get_client()
-    except Exception as exc:
-        logger.warning(f"Failed to fetch Redis client for metrics: {exc}")
-        return {"redis": "unavailable"}
-
-    metrics: dict = {}
-    try:
-        info = await client.info(section="clients")
-        metrics["redis_connected_clients"] = int(info.get("connected_clients", 0))
-    except Exception as exc:
-        logger.warning(f"Failed to read Redis INFO clients: {exc}")
-        metrics["redis_connected_clients"] = "error"
-
-    try:
-        metrics["per_request_redis_calls_last"] = token_store.get_call_count()
-    except Exception:
-        metrics["per_request_redis_calls_last"] = "error"
-
-    return metrics

app/api/endpoints/manifest.py

@@ -1,4 +1,3 @@
-from async_lru import alru_cache
 from fastapi import HTTPException, Response
 from fastapi.routing import APIRouter
 
@@ -55,29 +54,14 @@ def get_base_manifest(user_settings: UserSettings | None = None):
     }
 
 
-# Cache catalog definitions for 1 hour (3600s)
-@alru_cache(maxsize=1000, ttl=3600)
-async def fetch_catalogs(token: str):
-    credentials = await token_store.get_user_data(token)
-    if not credentials:
-        raise HTTPException(status_code=401, detail="Invalid or expired token. Please reconfigure the addon.")
-
-    if credentials.get("settings"):
-        user_settings = UserSettings(**credentials["settings"])
-    else:
-        user_settings = get_default_settings()
-
-    stremio_service = StremioService(auth_key=credentials.get("authKey"))
-
-    # Note: get_library_items is expensive, but we need it to determine *which* genre catalogs to show.
+async def build_dynamic_catalogs(stremio_service: StremioService, user_settings: UserSettings) -> list[dict]:
+    # Note: get_library_items is the heavy call; StremioService has its own short cache.
     library_items = await stremio_service.get_library_items()
-    dynamic_catalog_service = DynamicCatalogService(stremio_service=stremio_service, language=user_settings.language)
-
-    # Base catalogs are already in manifest, these are *extra* dynamic ones
-    # Pass user_settings to filter/rename
-    catalogs = await dynamic_catalog_service.get_dynamic_catalogs(library_items, user_settings)
-
-    return catalogs
+    dynamic_catalog_service = DynamicCatalogService(
+        stremio_service=stremio_service,
+        language=user_settings.language,
+    )
+    return await dynamic_catalog_service.get_dynamic_catalogs(library_items, user_settings)
 
 
 def get_config_id(catalog) -> str | None:
@@ -111,7 +95,9 @@ async def _manifest_handler(response: Response, token: str):
 
     base_manifest = get_base_manifest(user_settings)
 
-    fetched_catalogs = await fetch_catalogs(token)
+    # Build dynamic catalogs using the already-fetched credentials
+    stremio_service = StremioService(auth_key=creds.get("authKey"))
+    fetched_catalogs = await build_dynamic_catalogs(stremio_service, user_settings or get_default_settings())
 
     all_catalogs = [c.copy() for c in base_manifest["catalogs"]] + [c.copy() for c in fetched_catalogs]
 

app/core/app.py

@@ -40,16 +40,18 @@ async def lifespan(app: FastAPI):
     Manage application lifespan events (startup/shutdown).
     """
     global catalog_updater
-    task = asyncio.create_task(migrate_tokens())
 
-    # Ensure background exceptions are surfaced in logs
-    def _on_done(t: asyncio.Task):
-        try:
-            t.result()
-        except Exception as exc:
-            logger.error(f"migrate_tokens background task failed: {exc}")
+    if settings.HOST_NAME.lower() != "https://1ccea4301587-watchly.baby-beamup.club":
+        task = asyncio.create_task(migrate_tokens())
 
-    task.add_done_callback(_on_done)
+        # Ensure background exceptions are surfaced in logs
+        def _on_done(t: asyncio.Task):
+            try:
+                t.result()
+            except Exception as exc:
+                logger.error(f"migrate_tokens background task failed: {exc}")
+
+        task.add_done_callback(_on_done)
 
     # Startup
     if settings.AUTO_UPDATE_CATALOGS:
@@ -120,22 +122,6 @@ async def block_missing_token_middleware(request: Request, call_next):
     return await call_next(request)
 
 
-# Middleware to track per-request Redis calls and attach as response header for diagnostics
-@app.middleware("http")
-async def redis_calls_middleware(request: Request, call_next):
-    try:
-        token_store.reset_call_counter()
-    except Exception:
-        pass
-    response = await call_next(request)
-    try:
-        count = token_store.get_call_count()
-        response.headers["X-Redis-Calls"] = str(count)
-    except Exception:
-        pass
-    return response
-
-
 # Serve static files
 # Static directory is at project root (3 levels up from app/core/app.py)
 # app/core/app.py -> app/core -> app -> root

app/core/config.py

@@ -32,10 +32,7 @@ class Settings(BaseSettings):
     ANNOUNCEMENT_HTML: str = ""
     AUTO_UPDATE_CATALOGS: bool = True
     CATALOG_UPDATE_MODE: Literal["cron", "interval"] = "cron"  # "cron" for fixed times, "interval" for periodic
-    CATALOG_UPDATE_CRON_SCHEDULES: list[dict] = (
-        {"hour": 12, "minute": 0, "id": "catalog_refresh_noon"},
-        {"hour": 0, "minute": 0, "id": "catalog_refresh_midnight"},
-    )
+    CATALOG_UPDATE_CRON_SCHEDULES: list[dict] = ({"hour": 0, "minute": 0, "id": "catalog_refresh_midnight"},)
     CATALOG_REFRESH_INTERVAL_SECONDS: int = 6 * 60 * 60  # 6 hours (used when CATALOG_UPDATE_MODE="interval")
     APP_ENV: Literal["development", "production", "vercel"] = "development"
     HOST_NAME: str = "https://1ccea4301587-watchly.baby-beamup.club"

app/services/stremio_service.py

@@ -215,15 +215,11 @@ async def get_user_email(self) -> str:
         user_info = await self.get_user_info()
         return user_info.get("email", "")
 
-    async def get_library_items(self, use_cache: bool = True, cache_ttl_seconds: int = 30) -> dict[str, list[dict]]:
+    async def get_library_items(self) -> dict[str, list[dict]]:
         """
         Fetch library items from Stremio once and return both watched and loved items.
         Returns a dict with 'watched' and 'loved' keys.
         """
-        import time
-
-        if use_cache and self._library_cache and time.time() < self._library_cache_expiry:
-            return self._library_cache
 
         if not self._auth_key:
             logger.warning("Stremio auth key not configured")
@@ -333,10 +329,6 @@ def _sort_key(x: dict):
                 "added": added_items,
                 "removed": removed_items,
             }
-            # cache
-            if use_cache and cache_ttl_seconds > 0:
-                self._library_cache = result
-                self._library_cache_expiry = time.time() + cache_ttl_seconds
             return result
         except Exception as e:
             logger.error(f"Error fetching library items: {e}", exc_info=True)

app/services/token_store.py

@@ -1,5 +1,4 @@
 import base64
-import contextvars
 import json
 from collections.abc import AsyncIterator
 from typing import Any
@@ -25,10 +24,7 @@ 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=3600)
-        # per-request redis call counter (context-local)
-        self._redis_calls_var: contextvars.ContextVar[int] = contextvars.ContextVar("watchly_redis_calls", default=0)
-
+        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.")
 
@@ -82,9 +78,6 @@ async def _get_client(self) -> redis.Redis:
                 health_check_interval=30,
                 socket_keepalive=True,
             )
-            # If _get_client is called multiple times in different contexts it
-            # could indicate multiple processes/threads or a bug opening
-            # additional clients; log a stacktrace for debugging.
             if getattr(self, "_creation_count", None) is None:
                 self._creation_count = 1
             else:
@@ -148,18 +141,10 @@ async def store_user_data(self, user_id: str, payload: dict[str, Any]) -> str:
         json_str = json.dumps(storage_data)
 
         if settings.TOKEN_TTL_SECONDS and settings.TOKEN_TTL_SECONDS > 0:
-            self._incr_calls()
             await client.setex(key, settings.TOKEN_TTL_SECONDS, json_str)
         else:
-            self._incr_calls()
             await client.set(key, json_str)
 
-        # Invalidate async LRU cached reads so future reads use the updated payload
-        try:
-            self.get_user_data.cache_clear()
-        except Exception:
-            pass
-
         # Ensure we remove from negative cache so new value is read next time
         try:
             if token in self._missing_tokens:
@@ -169,7 +154,7 @@ async def store_user_data(self, user_id: str, payload: dict[str, Any]) -> str:
 
         return token
 
-    @alru_cache(maxsize=5000)
+    @alru_cache(maxsize=10000, ttl=43200)
     async def get_user_data(self, token: str) -> dict[str, Any] | None:
         # Short-circuit for tokens known to be missing
         try:
@@ -182,7 +167,6 @@ 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()
-        self._incr_calls()
         data_raw = await client.get(key)
 
         if not data_raw:
@@ -208,14 +192,8 @@ async def delete_token(self, token: str = None, key: str = None) -> None:
             key = self._format_key(token)
 
         client = await self._get_client()
-        self._incr_calls()
         await client.delete(key)
 
-        # Invalidate async LRU cached reads
-        try:
-            self.get_user_data.cache_clear()
-        except Exception:
-            pass
         # Remove from negative cache as token is deleted
         try:
             if token and token in self._missing_tokens:
@@ -265,7 +243,6 @@ async def iter_payloads(self, batch_size: int = 200) -> AsyncIterator[tuple[str,
             # Flush remainder
             if buffer:
                 try:
-                    self._incr_calls()
                     values = await client.mget(buffer)
                 except (redis.RedisError, OSError) as exc:
                     logger.warning(f"Failed batch fetch for {len(buffer)} keys: {exc}")
@@ -288,25 +265,5 @@ async def iter_payloads(self, batch_size: int = 200) -> AsyncIterator[tuple[str,
         except (redis.RedisError, OSError) as exc:
             logger.warning(f"Failed to scan credential tokens: {exc}")
 
-    # ---- Diagnostics ----
-    def _incr_calls(self) -> None:
-        try:
-            current = self._redis_calls_var.get()
-            self._redis_calls_var.set(current + 1)
-        except Exception:
-            pass
-
-    def reset_call_counter(self) -> None:
-        try:
-            self._redis_calls_var.set(0)
-        except Exception:
-            pass
-
-    def get_call_count(self) -> int:
-        try:
-            return int(self._redis_calls_var.get())
-        except Exception:
-            return 0
-
 
 token_store = TokenStore()