PolyWeather/web/scan_terminal_cache.py

from __future__ import annotations

import json
import hashlib
import os
import threading
import time
from datetime import datetime
from typing import Any, Dict, Optional

_SCAN_TERMINAL_CACHE_LOCK = threading.Lock()
_SCAN_TERMINAL_CACHE: Dict[str, Dict[str, Any]] = {}
_SCAN_TERMINAL_REFRESHING: set[str] = set()
_SCAN_TERMINAL_REDIS_CLIENT_LOCK = threading.Lock()
_SCAN_TERMINAL_REDIS_CLIENT: Any = None
_SCAN_TERMINAL_REDIS_UNAVAILABLE = False


def scan_terminal_cache_key(filters: Dict[str, Any]) -> str:
    return json.dumps(filters, ensure_ascii=True, sort_keys=True)


def _truthy_env(name: str, *, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default
    return str(value).strip().lower() not in ("", "0", "false", "no", "off")


def _redis_cache_enabled() -> bool:
    return _truthy_env(
        "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED",
        default=bool(os.getenv("POLYWEATHER_REDIS_URL")),
    )


def _redis_cache_ttl_sec() -> int:
    try:
        value = int(os.getenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_TTL_SEC", "21600"))
    except Exception:
        value = 21600
    return max(600, min(value, 86400))


def _redis_cache_prefix() -> str:
    prefix = os.getenv(
        "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX",
        "polyweather:scan_terminal:v2:",
    )
    if prefix.endswith(":v1:"):
        return f"{prefix[:-4]}:v2:"
    return prefix


def _redis_entry_key(cache_key: str) -> str:
    digest = hashlib.sha256(cache_key.encode("utf-8")).hexdigest()
    return f"{_redis_cache_prefix()}{digest}"


def _get_redis_client() -> Any:
    global _SCAN_TERMINAL_REDIS_CLIENT, _SCAN_TERMINAL_REDIS_UNAVAILABLE

    if not _redis_cache_enabled() or _SCAN_TERMINAL_REDIS_UNAVAILABLE:
        return None

    with _SCAN_TERMINAL_REDIS_CLIENT_LOCK:
        if _SCAN_TERMINAL_REDIS_CLIENT is not None:
            return _SCAN_TERMINAL_REDIS_CLIENT
        try:
            import redis  # type: ignore

            url = os.getenv("POLYWEATHER_REDIS_URL") or "redis://127.0.0.1:6379/0"
            client = redis.Redis.from_url(
                url,
                socket_timeout=float(os.getenv("POLYWEATHER_REDIS_SOCKET_TIMEOUT_SECONDS", "2")),
                socket_connect_timeout=float(
                    os.getenv("POLYWEATHER_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", "1")
                ),
                health_check_interval=30,
            )
            client.ping()
            _SCAN_TERMINAL_REDIS_CLIENT = client
            return client
        except Exception:
            _SCAN_TERMINAL_REDIS_UNAVAILABLE = True
            return None


def _read_redis_cache_entry(cache_key: str) -> Optional[Dict[str, Any]]:
    client = _get_redis_client()
    if client is None:
        return None
    try:
        raw = client.get(_redis_entry_key(cache_key))
        if not raw:
            return None
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8")
        entry = json.loads(str(raw))
        return dict(entry) if isinstance(entry, dict) else None
    except Exception:
        return None


def _write_redis_cache_entry(cache_key: str, entry: Dict[str, Any]) -> None:
    client = _get_redis_client()
    if client is None:
        return
    try:
        client.setex(
            _redis_entry_key(cache_key),
            _redis_cache_ttl_sec(),
            json.dumps(entry, ensure_ascii=False, separators=(",", ":")),
        )
    except Exception:
        return


def get_cached_scan_terminal_payload(
    filters: Dict[str, Any],
    *,
    ttl_sec: int,
) -> Optional[Dict[str, Any]]:
    now = time.time()
    cached = get_scan_terminal_cache_entry(filters)
    if not cached:
        return None
    cached_at = float(cached.get("t") or 0.0)
    if now - cached_at >= float(ttl_sec):
        return None
    payload = cached.get("payload")
    if not isinstance(payload, dict):
        return None
    return dict(payload)


def get_scan_terminal_cache_entry(filters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    cache_key = scan_terminal_cache_key(filters)
    with _SCAN_TERMINAL_CACHE_LOCK:
        cached = _SCAN_TERMINAL_CACHE.get(cache_key)
        if isinstance(cached, dict):
            return dict(cached)

    redis_entry = _read_redis_cache_entry(cache_key)
    if not redis_entry:
        return None

    with _SCAN_TERMINAL_CACHE_LOCK:
        _SCAN_TERMINAL_CACHE[cache_key] = dict(redis_entry)
    return dict(redis_entry)


def set_cached_scan_terminal_payload(
    filters: Dict[str, Any],
    payload: Dict[str, Any],
) -> None:
    cache_key = scan_terminal_cache_key(filters)
    existing = get_scan_terminal_cache_entry(filters) or {}
    now = time.time()
    existing_success_payload = existing.get("success_payload")
    previous_success_payload = existing.get("previous_success_payload")
    if isinstance(existing_success_payload, dict):
        existing_snapshot_id = existing_success_payload.get("snapshot_id")
        next_snapshot_id = payload.get("snapshot_id")
        if existing_snapshot_id and existing_snapshot_id != next_snapshot_id:
            previous_success_payload = dict(existing_success_payload)
    entry = {
        "t": now,
        "payload": dict(payload),
        "success_t": now,
        "success_payload": dict(payload),
        "previous_success_payload": previous_success_payload,
        "last_error": existing.get("last_error"),
        "last_failed_at": existing.get("last_failed_at"),
    }
    with _SCAN_TERMINAL_CACHE_LOCK:
        _SCAN_TERMINAL_CACHE[cache_key] = dict(entry)
    _write_redis_cache_entry(cache_key, entry)


def set_scan_terminal_failure_state(
    filters: Dict[str, Any],
    *,
    error_message: str,
) -> None:
    cache_key = scan_terminal_cache_key(filters)
    existing = get_scan_terminal_cache_entry(filters) or {}
    existing["last_error"] = error_message
    existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z"
    with _SCAN_TERMINAL_CACHE_LOCK:
        _SCAN_TERMINAL_CACHE[cache_key] = dict(existing)
    _write_redis_cache_entry(cache_key, existing)


def mark_scan_terminal_refreshing(filters: Dict[str, Any]) -> bool:
    cache_key = scan_terminal_cache_key(filters)
    with _SCAN_TERMINAL_CACHE_LOCK:
        if cache_key in _SCAN_TERMINAL_REFRESHING:
            return False
        _SCAN_TERMINAL_REFRESHING.add(cache_key)
    return True


def clear_scan_terminal_refreshing(filters: Dict[str, Any]) -> None:
    cache_key = scan_terminal_cache_key(filters)
    with _SCAN_TERMINAL_CACHE_LOCK:
        _SCAN_TERMINAL_REFRESHING.discard(cache_key)