services/cache.py

"""
Cache Service — Unified Caching & Semantic Cache
=================================================
Unified caching layer with two backends:
  - Redis   (local dev / Docker) — uses REDIS_URL env var
  - In-memory dict (HF Spaces)  — used when REDIS_URL is not set

Features:
  - Key-value caching with TTL
  - Rate-limit counter tracking
  - Semantic Caching (avoids redundant LLM generation for semantically matching queries)
"""
from __future__ import annotations

import asyncio
import json
import time
from typing import Any, Dict, Optional, Tuple, List

import numpy as np
import structlog

logger = structlog.get_logger(__name__)


class InMemoryCache:
    """Simple thread-safe TTL in-memory cache used when Redis is unavailable."""

    def __init__(self):
        self._store: Dict[str, Tuple[str, float]] = {}
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> Optional[str]:
        async with self._lock:
            entry = self._store.get(key)
            if entry is None:
                return None
            value, expiry = entry
            if expiry > 0 and time.time() > expiry:
                del self._store[key]
                return None
            return value

    async def set(self, key: str, value: str, ttl: int = 300) -> None:
        expiry = time.time() + ttl if ttl > 0 else -1
        async with self._lock:
            self._store[key] = (value, expiry)

    async def delete(self, key: str) -> None:
        async with self._lock:
            self._store.pop(key, None)

    async def exists(self, key: str) -> bool:
        return await self.get(key) is not None

    async def incr(self, key: str, ttl: int = 60) -> int:
        async with self._lock:
            entry = self._store.get(key)
            if entry is None or (entry[1] > 0 and time.time() > entry[1]):
                count = 1
            else:
                try:
                    count = int(entry[0]) + 1
                except ValueError:
                    count = 1
            expiry = time.time() + ttl
            self._store[key] = (str(count), expiry)
            return count


class CacheService:
    """
    Auto-selects Redis or InMemoryCache.
    Provides semantic caching via query vector similarity.
    """

    def __init__(self):
        self._backend = None
        self._semantic_store: List[Dict[str, Any]] = []

    def _init_backend(self):
        if self._backend is None:
            import os
            redis_url = os.environ.get("REDIS_URL")
            if redis_url:
                logger.info("Using Redis cache", url=redis_url)
                from services.cache import RedisCache
                self._backend = RedisCache(redis_url)
            else:
                logger.info("Using in-memory cache (no Redis configured)")
                self._backend = InMemoryCache()
        return self._backend

    async def get(self, key: str) -> Optional[str]:
        return await self._init_backend().get(key)

    async def set(self, key: str, value: Any, ttl: int = 300) -> None:
        if not isinstance(value, str):
            value = json.dumps(value)
        await self._init_backend().set(key, value, ttl)

    async def get_json(self, key: str) -> Optional[Any]:
        raw = await self.get(key)
        if raw is None:
            return None
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return raw

    async def delete(self, key: str) -> None:
        await self._init_backend().delete(key)

    async def exists(self, key: str) -> bool:
        return await self._init_backend().exists(key)

    async def incr(self, key: str, ttl: int = 60) -> int:
        return await self._init_backend().incr(key, ttl)

    # ── Semantic Caching ───────────────────────────────────────
    def get_semantic(self, query_vector: List[float], similarity_threshold: float = 0.93) -> Optional[Dict[str, Any]]:
        """Look up cached response for semantically matching query vector."""
        if not self._semantic_store:
            return None

        q_vec = np.array(query_vector)
        for entry in self._semantic_store:
            cached_vec = np.array(entry["vector"])
            similarity = float(np.dot(q_vec, cached_vec) / (np.linalg.norm(q_vec) * np.linalg.norm(cached_vec)))
            if similarity >= similarity_threshold:
                logger.info("Semantic cache hit!", similarity=round(similarity, 4))
                return entry["response"]
        return None

    def set_semantic(self, query_vector: List[float], response_data: Dict[str, Any]) -> None:
        """Store query vector and response in semantic cache."""
        self._semantic_store.append({
            "vector": query_vector,
            "response": response_data,
            "timestamp": time.time(),
        })
        # Keep store size bounded
        if len(self._semantic_store) > 200:
            self._semantic_store.pop(0)


# ── Singleton instance ─────────────────────────────────────────
cache_service = CacheService()