app/core/prompt_cache.py

"""In-memory LRU cache for AI prompt results.

Prevents duplicate AI calls for identical prompts within the TTL window.
Uses an ``OrderedDict``-based LRU with a per-entry timestamp for TTL checks.

Default limits
--------------
- ``max_size``: 1000 entries
- ``ttl_seconds``: 3600 (1 hour)

Thread-safe via a single ``threading.Lock``.

Usage::

    from app.core.prompt_cache import get_prompt_cache

    cache = get_prompt_cache()
    key = make_cache_key(task, model, prompt_hash)

    cached = cache.get(key)
    if cached is not None:
        return cached

    result = provider.generate_notes(...)
    cache.set(key, result)
    return result
"""
from __future__ import annotations

import hashlib
import json
import logging
import threading
import time
from collections import OrderedDict
from copy import deepcopy
from typing import Any

logger = logging.getLogger(__name__)

_REDIS_PREFIX = "pc:"  # prompt-cache namespace in Redis


class PromptLRUCache:
    """Two-tier cache: in-process LRU (L1) + optional shared Redis (L2).

    L1 keeps single-instance hot reads zero-latency. L2 (active only when
    ``REDIS_URL`` is set) shares results across every replica so a cache warmed
    by one instance is reused by all — the difference between per-instance and
    fleet-wide caching. Falls back cleanly to L1-only when Redis is absent.
    """

    def __init__(self, max_size: int = 1000, ttl_seconds: float = 3600.0) -> None:
        self.max_size = max_size
        self.ttl_seconds = ttl_seconds

        # OrderedDict value: (created_at: float, payload: dict)
        self._store: OrderedDict[str, tuple[float, dict[str, Any]]] = OrderedDict()
        self._lock = threading.Lock()

        # Counters for observability
        self._hits = 0
        self._misses = 0

    # ── L2 (Redis) helpers ─────────────────────────────────────────────────────

    def _redis(self) -> Any | None:
        try:
            from app.core.redis_client import get_redis

            return get_redis()
        except Exception:
            return None

    # ── Public API ────────────────────────────────────────────────────────────

    def get(self, key: str) -> dict[str, Any] | None:
        """Return a deep-copy of the cached value or ``None`` on miss / expiry."""
        now = time.monotonic()
        with self._lock:
            entry = self._store.get(key)
            if entry is not None:
                created_at, value = entry
                if now - created_at <= self.ttl_seconds:
                    self._store.move_to_end(key)
                    self._hits += 1
                    return deepcopy(value)
                del self._store[key]  # expired in L1

        # L2: shared Redis lookup (outside lock — network call)
        client = self._redis()
        if client is not None:
            try:
                raw = client.get(_REDIS_PREFIX + key)
                if raw:
                    value = json.loads(raw)
                    self._set_local(key, value)  # promote into L1
                    with self._lock:
                        self._hits += 1
                    return deepcopy(value)
            except Exception as exc:
                logger.debug("Prompt cache L2 get failed: %s", type(exc).__name__)

        with self._lock:
            self._misses += 1
        return None

    def _set_local(self, key: str, value: dict[str, Any]) -> None:
        now = time.monotonic()
        with self._lock:
            if key in self._store:
                self._store.move_to_end(key)
            self._store[key] = (now, value)
            while len(self._store) > self.max_size:
                self._store.popitem(last=False)

    def set(self, key: str, value: dict[str, Any]) -> None:
        """Insert or refresh an entry in L1 and (when configured) shared L2."""
        self._set_local(key, value)
        client = self._redis()
        if client is not None:
            try:
                client.setex(
                    _REDIS_PREFIX + key,
                    int(self.ttl_seconds),
                    json.dumps(value, ensure_ascii=False, separators=(",", ":")),
                )
            except Exception as exc:
                logger.debug("Prompt cache L2 set failed: %s", type(exc).__name__)

    def clear(self) -> None:
        """Empty the cache (used in tests)."""
        with self._lock:
            self._store.clear()
            self._hits = 0
            self._misses = 0

    # ── Observability ─────────────────────────────────────────────────────────

    @property
    def size(self) -> int:
        with self._lock:
            return len(self._store)

    def stats(self) -> dict[str, Any]:
        with self._lock:
            total = self._hits + self._misses
            return {
                "size": len(self._store),
                "max_size": self.max_size,
                "ttl_seconds": self.ttl_seconds,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate": round(self._hits / total, 3) if total else 0.0,
            }


# ── Singleton ─────────────────────────────────────────────────────────────────

_cache_instance: PromptLRUCache | None = None
_cache_lock = threading.Lock()


def get_prompt_cache() -> PromptLRUCache:
    """Return the process-level singleton PromptLRUCache."""
    global _cache_instance
    if _cache_instance is None:
        with _cache_lock:
            if _cache_instance is None:
                _cache_instance = PromptLRUCache(max_size=1000, ttl_seconds=3600.0)
    return _cache_instance


# ── Key helpers ───────────────────────────────────────────────────────────────

def make_prompt_cache_key(task_key: str, model_id: str, prompt: str) -> str:
    """Return a short SHA-256 hex key from task + model + prompt content."""
    raw = f"{task_key}\x00{model_id}\x00{prompt}"
    return hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()