cyper-rag-server

"""
utils/cache.py — Thread-safe TTL query-deduplication cache.
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from threading import Lock
from typing import Generic, TypeVar

V = TypeVar("V")


@dataclass
class TTLCache(Generic[V]):
    """
    A simple key→value store where entries expire after `ttl_seconds`.
    All operations are protected by a threading.Lock, making this safe
    for concurrent MCP tool calls.
    """
    ttl_seconds: float
    _store: dict[str, tuple[float, V]] = field(default_factory=dict)
    _lock: Lock = field(default_factory=Lock)

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

    def get(self, key: str) -> V | None:
        normalized = self._normalize(key)
        with self._lock:
            entry = self._store.get(normalized)
            if entry is None:
                return None
            ts, value = entry
            if time.monotonic() - ts > self.ttl_seconds:
                del self._store[normalized]
                return None
            return value

    def set(self, key: str, value: V) -> None:
        normalized = self._normalize(key)
        with self._lock:
            self._store[normalized] = (time.monotonic(), value)

    def invalidate(self, key: str) -> None:
        normalized = self._normalize(key)
        with self._lock:
            self._store.pop(normalized, None)

    def clear(self) -> None:
        with self._lock:
            self._store.clear()

    def __len__(self) -> int:
        with self._lock:
            return len(self._store)

    # ── Internal ──────────────────────────────────────────────────────────────

    @staticmethod
    def _normalize(key: str) -> str:
        """Lowercase + collapse whitespace for fuzzy dedup."""
        return " ".join(key.lower().split())