import time
import asyncio
from typing import Optional
from ..config import settings

# Redis is optional at import-time so the API runs without it.
try:
    import redis.asyncio as aioredis
    _HAS_REDIS = True
except Exception:  # pragma: no cover - redis not installed
    aioredis = None
    _HAS_REDIS = False

_redis_client: Optional[object] = None

# In-memory TTL fallback cache: {key: (value, expires_at)}
_memory_cache: dict = {}


class CircuitBreaker:
    """Fail fast to the in-memory cache when Redis is slow or down.

    Without this, `validate_batch` pays Redis' socket timeout (~2s) on every one
    of the ~8 cache calls it makes per row — measured at 10.0s per row on a
    host where the Upstash link is slow, while the engine itself needs ~70ms.
    After `failure_threshold` consecutive slow/failed ops the circuit opens and
    calls skip Redis entirely until `cooldown_seconds` elapse, then probe once.
    """

    def __init__(self, failure_threshold: int = 3,
                 cooldown_seconds: float = 30.0, slow_threshold: float = 1.0):
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.slow_threshold = slow_threshold
        self._failures = 0
        self._open_until = 0.0

    def allow(self) -> bool:
        """True when Redis may be contacted (closed, or cooldown elapsed)."""
        return time.time() >= self._open_until

    def record(self, success: bool, duration: float) -> None:
        """Record one op: slow ops count as failures even when they succeed."""
        if success and duration < self.slow_threshold:
            self._failures = 0
            return
        self._failures += 1
        if self._failures >= self.failure_threshold:
            self._open_until = time.time() + self.cooldown_seconds
            self._failures = 0

    def reset(self) -> None:
        self._failures = 0
        self._open_until = 0.0


# Trips when Redis is consistently slow/failing; see CircuitBreaker.
_circuit = CircuitBreaker()


async def get_redis():
    """Lazily create a shared Redis client (returns None if unavailable).

    Redis connections must be configured explicitly. Missing or unreachable
    Redis leaves the API on its bounded in-memory cache instead of silently
    provisioning a temporary external database.
    """
    global _redis_client
    if not _HAS_REDIS:
        return None
    if _redis_client is None:
        url = (settings.redis_url or "").strip()
        if not url:
            return None
        try:
            _redis_client = aioredis.from_url(
                url,
                encoding="utf-8",
                decode_responses=True,
                # The TLS handshake to Upstash measures ~750-850 ms on some
                # networks; a 2 s cap killed the very first PING at exactly
                # the timeout (observed live), which made every cache op pay
                # the full budget and trip the breaker. 5 s connects reliably
                # while the breaker still fails fast on real outages.
                socket_connect_timeout=5,
                socket_timeout=5,
            )
            await _redis_client.ping()
        except Exception:
            _redis_client = None
    return _redis_client


def _memory_get(key: str) -> Optional[str]:
    item = _memory_cache.get(key)
    if item is None:
        return None
    value, expires = item
    if time.time() > expires:
        _memory_cache.pop(key, None)
        return None
    return value


def _memory_set(key: str, value: str, ttl: int):
    _memory_cache[key] = (value, time.time() + ttl)
    # keep cache bounded
    if len(_memory_cache) > 20000:
        now = time.time()
        for k in [k for k, (_, exp) in _memory_cache.items() if exp < now]:
            _memory_cache.pop(k, None)


async def get_cache(key: str) -> Optional[str]:
    if _circuit.allow():
        t0 = time.perf_counter()
        try:
            client = await get_redis()
            if client is not None:
                value = await client.get(key)
                _circuit.record(success=True, duration=time.perf_counter() - t0)
                return value
            # Redis is configured but get_redis() could not establish a client
            # (connect/ping failed). Without this the 2s timeout is paid on
            # EVERY call and the breaker never trips.
            if (settings.redis_url or "").strip():
                _circuit.record(success=False,
                                duration=time.perf_counter() - t0)
        except Exception:
            _circuit.record(success=False, duration=time.perf_counter() - t0)
    return _memory_get(key)


async def set_cache(key: str, value: str, ttl: int = 86400):
    if _circuit.allow():
        t0 = time.perf_counter()
        try:
            client = await get_redis()
            if client is not None:
                await client.setex(key, ttl, value)
                _circuit.record(success=True, duration=time.perf_counter() - t0)
                return
            if (settings.redis_url or "").strip():
                _circuit.record(success=False,
                                duration=time.perf_counter() - t0)
        except Exception:
            _circuit.record(success=False, duration=time.perf_counter() - t0)
    _memory_set(key, value, ttl)


async def redis_available(timeout: float = 4.0) -> bool:
    """
    Quick non-blocking check: can we talk to Redis?

    Used to gate Celery .delay() calls so a missing Redis broker fails
    fast instead of entering Celery's infinite reconnect/retry loop.
    """
    if not _HAS_REDIS:
        return False
    try:
        client = await asyncio.wait_for(
            get_redis(), timeout=timeout
        )
        if client is None:
            return False
        await asyncio.wait_for(client.ping(), timeout=timeout)
        return True
    except asyncio.TimeoutError:
        print("[REDIS] Connection timeout - Redis may not be running")
        return False
    except Exception as e:
        print(f"[REDIS] Connection failed: {str(e)}")
        return False