providers/cache.py

"""In-memory cache for Short Hunter datasource responses."""

from __future__ import annotations

import os
import threading
import time
from copy import deepcopy
from typing import Any, Dict, Optional


def _ttl_seconds() -> int:
    return int(os.getenv("PROVIDER_CACHE_TTL_SECONDS", "30"))


def _stale_seconds() -> int:
    return int(os.getenv("PROVIDER_STALE_CACHE_THRESHOLD_SECONDS", "300"))


class GatewayCache:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._items: Dict[str, Dict[str, Any]] = {}

    def set(self, key: str, payload: Dict[str, Any], *, provider: str, capability: str, ttl: Optional[int] = None) -> None:
        now = time.time()
        ttl_seconds = int(ttl or _ttl_seconds())
        with self._lock:
            self._items[key] = {
                "key": key,
                "provider": provider,
                "capability": capability,
                "storedAt": now,
                "expiresAt": now + ttl_seconds,
                "staleAfter": now + _stale_seconds(),
                "payload": deepcopy(payload),
            }

    def get(self, key: str, *, allow_stale: bool = True) -> Optional[Dict[str, Any]]:
        now = time.time()
        with self._lock:
            item = self._items.get(key)
            if not item:
                return None
            if not allow_stale and now > item["expiresAt"]:
                return None
            result = deepcopy(item)
            result["ageSeconds"] = max(0.0, now - item["storedAt"])
            result["isStale"] = now > item["staleAfter"]
            result["isExpired"] = now > item["expiresAt"]
            return result

    def snapshot(self) -> Dict[str, Any]:
        now = time.time()
        with self._lock:
            items = []
            for item in self._items.values():
                cloned = dict(item)
                cloned["ageSeconds"] = max(0.0, now - item["storedAt"])
                cloned["isStale"] = now > item["staleAfter"]
                cloned["isExpired"] = now > item["expiresAt"]
                items.append(cloned)
        return {
            "size": len(items),
            "ttlSeconds": _ttl_seconds(),
            "staleThresholdSeconds": _stale_seconds(),
            "items": items,
        }


gateway_cache = GatewayCache()