app/core/cache.py

"""Database-backed response cache with stale fallback support."""

from __future__ import annotations

import hashlib
import os
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine

from app.core.config import settings
from app.utils.serialization import dumps_json, loads_json, to_jsonable


UTC = timezone.utc
TABLE_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")


@dataclass(frozen=True)
class CacheEntry:
    key: str
    namespace: str
    payload: Any
    source: str
    created_at: datetime
    expires_at: datetime
    row_count: int | None

    @property
    def expired(self) -> bool:
        return self.expires_at <= datetime.now(UTC)


class ResponseCache:
    def __init__(self, database_url: str, enabled: bool = True) -> None:
        self.enabled = enabled
        self.database_url = self._normalize_database_url(database_url)
        self.table_name = self._validate_table_name(settings.cache_table_name)
        self.engine: Engine | None = None
        if self.enabled:
            self._init_engine()

    def _normalize_database_url(self, database_url: str) -> str:
        if database_url.startswith("mysql://"):
            return database_url.replace("mysql://", "mysql+pymysql://", 1)
        return database_url

    def _validate_table_name(self, table_name: str) -> str:
        value = table_name.strip()
        if not TABLE_NAME_PATTERN.fullmatch(value):
            raise ValueError(f"Invalid CACHE_TABLE_NAME: {table_name!r}")
        return value

    def _init_engine(self) -> None:
        if self.database_url.startswith("sqlite:///"):
            path = self.database_url.removeprefix("sqlite:///")
            dirname = os.path.dirname(path)
            if dirname:
                os.makedirs(dirname, exist_ok=True)
        connect_args: dict[str, Any] = {}
        if settings.database_ssl and self.database_url.startswith("mysql+pymysql://"):
            connect_args["ssl"] = {"ssl": True}
        self.engine = create_engine(
            self.database_url,
            pool_pre_ping=True,
            future=True,
            connect_args=connect_args,
        )
        with self.engine.begin() as conn:
            conn.execute(
                text(
                    f"""
                    CREATE TABLE IF NOT EXISTS {self.table_name} (
                        cache_key VARCHAR(191) PRIMARY KEY,
                        namespace VARCHAR(96) NOT NULL,
                        request_hash VARCHAR(64) NOT NULL,
                        payload_json LONGTEXT NOT NULL,
                        source_name VARCHAR(160) NOT NULL,
                        row_count INTEGER NULL,
                        created_at DATETIME NOT NULL,
                        expires_at DATETIME NOT NULL
                    )
                    """
                )
            )

    def make_key(self, namespace: str, params: dict[str, Any]) -> str:
        canonical = dumps_json(
            {
                "version": settings.cache_schema_version,
                "namespace": namespace,
                "params": to_jsonable(params),
            }
        )
        digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        return f"{namespace}:{digest}"

    def get(self, key: str, allow_expired: bool = False) -> CacheEntry | None:
        if not self.enabled or self.engine is None:
            return None
        with self.engine.begin() as conn:
            row = conn.execute(
                text(
                    f"""
                    SELECT cache_key, namespace, payload_json, source_name, row_count, created_at, expires_at
                    FROM {self.table_name}
                    WHERE cache_key = :key
                    """
                ),
                {"key": key},
            ).mappings().first()
        if not row:
            return None

        entry = CacheEntry(
            key=str(row["cache_key"]),
            namespace=str(row["namespace"]),
            payload=loads_json(str(row["payload_json"])),
            source=str(row["source_name"]),
            row_count=row["row_count"],
            created_at=self._as_utc(row["created_at"]),
            expires_at=self._as_utc(row["expires_at"]),
        )
        if entry.expired and not allow_expired:
            return None
        return entry

    def set(
        self,
        key: str,
        namespace: str,
        payload: Any,
        source: str,
        ttl_seconds: int,
        row_count: int | None = None,
    ) -> CacheEntry | None:
        if not self.enabled or self.engine is None:
            return None
        now = datetime.now(UTC)
        expires_at = now + timedelta(seconds=ttl_seconds)
        request_hash = key.rsplit(":", 1)[-1]
        with self.engine.begin() as conn:
            conn.execute(
                text(
                    f"""
                    DELETE FROM {self.table_name}
                    WHERE cache_key = :key
                    """
                ),
                {"key": key},
            )
            conn.execute(
                text(
                    f"""
                    INSERT INTO {self.table_name} (
                        cache_key, namespace, request_hash, payload_json, source_name,
                        row_count, created_at, expires_at
                    )
                    VALUES (
                        :key, :namespace, :request_hash, :payload_json, :source_name,
                        :row_count, :created_at, :expires_at
                    )
                    """
                ),
                {
                    "key": key,
                    "namespace": namespace,
                    "request_hash": request_hash,
                    "payload_json": dumps_json(payload),
                    "source_name": source,
                    "row_count": row_count,
                    "created_at": now.replace(tzinfo=None),
                    "expires_at": expires_at.replace(tzinfo=None),
                },
            )
        return CacheEntry(key, namespace, payload, source, now, expires_at, row_count)

    def purge_expired(self) -> int:
        if not self.enabled or self.engine is None:
            return 0
        now = datetime.now(UTC).replace(tzinfo=None)
        with self.engine.begin() as conn:
            result = conn.execute(
                text(f"DELETE FROM {self.table_name} WHERE expires_at < :now"),
                {"now": now},
            )
            return int(result.rowcount or 0)

    def purge_all(self) -> int:
        if not self.enabled or self.engine is None:
            return 0
        with self.engine.begin() as conn:
            result = conn.execute(text(f"DELETE FROM {self.table_name}"))
            return int(result.rowcount or 0)

    def stats(self) -> dict[str, Any]:
        if not self.enabled or self.engine is None:
            return {"enabled": False}
        with self.engine.begin() as conn:
            total = conn.execute(text(f"SELECT COUNT(*) FROM {self.table_name}")).scalar_one()
            expired = conn.execute(
                text(f"SELECT COUNT(*) FROM {self.table_name} WHERE expires_at < :now"),
                {"now": datetime.now(UTC).replace(tzinfo=None)},
            ).scalar_one()
        return {
            "enabled": True,
            "backend": self.database_url.split(":", 1)[0],
            "total_entries": int(total),
            "expired_entries": int(expired),
        }

    def _as_utc(self, value: Any) -> datetime:
        if isinstance(value, datetime):
            if value.tzinfo is None:
                return value.replace(tzinfo=UTC)
            return value.astimezone(UTC)
        return datetime.fromisoformat(str(value)).replace(tzinfo=UTC)


cache = ResponseCache(settings.database_url, enabled=settings.cache_enabled)