Phase 1: Final stabilization. Purged legacy auth (auth_data, verify_token) and resolved all IDE diagnostic errors.

Author: sanjeevafk

api/auth.py

@@ -16,53 +16,6 @@
 
 logger = structlog.get_logger(__name__)
 
-_PRO_STATE_CACHE: "OrderedDict[str, tuple[bool, float]]" = OrderedDict()
-_PRO_STATE_CACHE_LOCK = threading.Lock()
-
-
-def _pro_cache_ttl_seconds() -> int:
-    try:
-        configured_ttl = int(getattr(get_settings(), "pro_state_cache_ttl_seconds", 30) or 30)
-    except (ValueError, TypeError):
-        configured_ttl = 30
-    return min(max(configured_ttl, 1), 30)
-
-
-def _pro_cache_max_entries() -> int:
-    try:
-        configured_max = int(getattr(get_settings(), "pro_state_cache_max_entries", 1000) or 1000)
-    except (ValueError, TypeError):
-        configured_max = 1000
-    return min(max(configured_max, 1), 10000)
-
-
-def _prune_pro_cache_locked(now: float) -> None:
-    expired_keys = [user_id for user_id, (_is_pro, expires_at) in _PRO_STATE_CACHE.items() if expires_at <= now]
-    for user_id in expired_keys:
-        _PRO_STATE_CACHE.pop(user_id, None)
-
-    max_entries = _pro_cache_max_entries()
-    while len(_PRO_STATE_CACHE) > max_entries:
-        _PRO_STATE_CACHE.popitem(last=False)
-
-
-def invalidate_pro_cache(user_id: str) -> None:
-    if not user_id:
-        return
-    with _PRO_STATE_CACHE_LOCK:
-        _PRO_STATE_CACHE.pop(user_id, None)
-    async def _clear() -> None:
-        redis = await safe_redis_call(get_redis, operation="connect")
-        if redis is None:
-            return
-        await safe_redis_call(redis.delete, f"depthapi:user:is_pro:{user_id}", operation="delete")
-
-    try:
-        asyncio.create_task(_clear())
-    except Exception as exc:
-        logger.debug("auth_invalidate_pro_cache_failed", user_id_hash=anonymize_user_id(user_id), error=str(exc))
-
-
 @lru_cache(maxsize=1)
 def get_supabase() -> Client | None:
     settings = get_settings()
@@ -82,51 +35,3 @@ def get_supabase_admin() -> Client | None:
     if hasattr(secret_key, "get_secret_value"):
         secret_key = secret_key.get_secret_value()
     return create_client(settings.supabase_url, secret_key) # type: ignore
-
-
-async def check_is_pro(user_id: str, force_refresh: bool = False) -> bool:
-    """Check if a user has pro status in the database (deprecated for v1)."""
-    if not user_id:
-        return False
-
-    now = time.time()
-    if not force_refresh:
-        with _PRO_STATE_CACHE_LOCK:
-            _prune_pro_cache_locked(now)
-            cached = _PRO_STATE_CACHE.get(user_id)
-            if cached and cached[1] > now:
-                _PRO_STATE_CACHE.move_to_end(user_id)
-                return cached[0]
-        try:
-            redis = await safe_redis_call(get_redis, operation="connect")
-            raw = await safe_redis_call(redis.get, f"depthapi:user:is_pro:{user_id}", operation="get") if redis else None
-            if raw is not None:
-                value = raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else str(raw)
-                is_pro = value == "1"
-                with _PRO_STATE_CACHE_LOCK:
-                    _prune_pro_cache_locked(now)
-                    _PRO_STATE_CACHE[user_id] = (is_pro, now + _pro_cache_ttl_seconds())
-                    _PRO_STATE_CACHE.move_to_end(user_id)
-                return is_pro
-        except Exception as exc:
-            logger.debug("auth_pro_cache_redis_read_failed", user_id_hash=anonymize_user_id(user_id), error=str(exc))
-
-    supabase = get_supabase_admin()
-    if not supabase:
-        return False
-        
-    try:
-        response = await asyncio.to_thread(
-            lambda: supabase.table("users").select("is_pro").eq("id", user_id).single().execute()
-        )
-        data = getattr(response, "data", None)
-        is_pro = bool(data.get("is_pro", False)) if isinstance(data, dict) else False
-        
-        with _PRO_STATE_CACHE_LOCK:
-            _prune_pro_cache_locked(now)
-            _PRO_STATE_CACHE[user_id] = (is_pro, now + _pro_cache_ttl_seconds())
-            _PRO_STATE_CACHE.move_to_end(user_id)
-        return is_pro
-    except Exception as e:
-        logger.error("auth_check_is_pro_failed", user_id_hash=anonymize_user_id(user_id), error=str(e))
-        return False

api/routers/messages.py

@@ -52,19 +52,16 @@ async def _send_message_handler(
     _core.log_sampled_success = log_sampled_success
     _core.cache_set_value = cache_set_value
     _core.logger = logger
-    # Bridge: wrap ApiKeyRecord into the shape messages_core expects
-    auth_data = {"api_key": api_key, "is_pro": api_key.is_pro}
-    return await _core._send_message_handler(request=request, auth_data=auth_data)
+    return await _core._send_message_handler(request=request, api_key=api_key)
 
 
 @router.post("/messages")
 async def send_message(
     request: Request,
     api_key: ApiKeyRecord = Depends(verify_api_key),
 ) -> StreamingResponse:
-    auth_data = {"api_key": api_key, "is_pro": api_key.is_pro}
     return await _core._message_workflow.process_message(
         request=request,
-        auth_data=auth_data,
+        api_key=api_key,
         handler=_send_message_handler,
     )

api/routers/messages_core.py

@@ -1644,10 +1644,10 @@ async def get_next_chunk() -> str:
 @router.post("/messages")
 async def send_message(
     request: Request,
-    auth_data: dict = Depends(verify_token),
+    api_key: ApiKeyRecord = Depends(verify_api_key),
 ) -> StreamingResponse:
     return await _message_workflow.process_message(
         request=request,
-        auth_data=auth_data,
+        api_key=api_key,
         handler=_send_message_handler,
     )

api/routers/query.py

@@ -407,7 +407,7 @@ async def query_topic_stream(
         user_id_raw=api_key.id,
         user_id_hash=user_id_hash,
         topic_hash=topic_hash,
-        auth_data={"api_key": api_key},
+        api_key=api_key,
         cache_get=cache_get,
         cache_set=cache_set,
         cache_key_value=cache_key(topic, level, mode),

api/services/message_workflow.py

@@ -9,7 +9,7 @@
 from fastapi import Request
 from fastapi.responses import StreamingResponse
 
-from logging_config import logger
+from services.api_key_auth import ApiKeyRecord
 
 T = TypeVar("T")
 
@@ -35,10 +35,10 @@ async def process_message(
         self,
         *,
         request: Request,
-        auth_data: dict,
-        handler: Callable[[Request, dict], Awaitable[StreamingResponse]],
+        api_key: ApiKeyRecord,
+        handler: Callable[[Request, ApiKeyRecord], Awaitable[StreamingResponse]],
     ) -> StreamingResponse:
         async def _execute() -> StreamingResponse:
-            return await handler(request, auth_data)
+            return await handler(request, api_key)
 
         return await self.run_stage("process_message", _execute)

api/services/query_streaming.py

@@ -12,6 +12,7 @@
 import time
 from typing import Any
 from collections.abc import AsyncIterable, AsyncIterator, Iterable
+from services.api_key_auth import ApiKeyRecord
 
 from fastapi.responses import StreamingResponse
 
@@ -107,7 +108,7 @@ def build_query_stream_response(
     user_id_raw: str | None,
     user_id_hash: str | None,
     topic_hash: str,
-    auth_data,
+    api_key: ApiKeyRecord,
     cache_get,
     cache_set,
     cache_key_value: str,
@@ -207,8 +208,7 @@ def emit(event: str, payload: dict[str, Any] | str) -> str:
                     for index in range(0, len(content), chunk_size):
                         yield emit("chunk", {"chunk": content[index : index + chunk_size]})
                     yield emit("done", "[DONE]")
-                    if auth_data:
-                        await persist_history(auth_data["user"], topic, [level], mode)
+                    await persist_history(api_key.id, topic, [level], mode)
                     return
 
             stream = generate_stream_explanation(
@@ -319,17 +319,15 @@ async def get_next_chunk():
                 yield emit("done", "[DONE]")
                 if full_content.strip():
                     await cache_set(cache_key_value, {"text": full_content})
-                if auth_data:
-                    await persist_history(auth_data["user"], topic, [level], mode)
+                await persist_history(api_key.id, topic, [level], mode)
                 return
             if timed_out:
                 cutoff_message = "\n\n[Response truncated to stay within serverless limits. Retry to continue.]"
                 yield emit("chunk", {"chunk": cutoff_message})
 
             if full_content.strip():
                 await cache_set(cache_key_value, {"text": full_content})
-            if auth_data:
-                await persist_history(auth_data["user"], topic, [level], mode)
+            await persist_history(api_key.id, topic, [level], mode)
 
             yield emit("done", "[DONE]")
         except Exception as exc:
@@ -380,8 +378,7 @@ async def get_next_chunk():
                     yield emit("done", "[DONE]")
                     if full_content.strip():
                         await cache_set(cache_key_value, {"text": full_content})
-                    if auth_data:
-                        await persist_history(auth_data["user"], topic, [level], mode)
+                    await persist_history(api_key.id, topic, [level], mode)
                     return
                 except Exception as fallback_exc:
                     logger.error(
@@ -401,8 +398,7 @@ async def get_next_chunk():
                 yield emit("done", "[DONE]")
                 if full_content.strip():
                     await cache_set(cache_key_value, {"text": full_content})
-                if auth_data:
-                    await persist_history(auth_data["user"], topic, [level], mode)
+                await persist_history(api_key.id, topic, [level], mode)
                 return
             yield emit("error", {"error": "An error occurred while streaming. Please try again."})
             yield emit("done", "[DONE]")