"""Shared async Redis client.

Backs the distributed lock, rate limiting, refresh-token JTI store, short
TTL caches and the idempotency layer — i.e. Redis is on the critical path
of both request handling and every scheduled job.

**Resilience is configured here, not at the call sites.** ``from_url``
builds the pool directly (``Redis.from_url`` -> ``ConnectionPool.from_url``
-> ``Connection(**connection_kwargs)``), which bypasses the retry/timeout
defaults that ``Redis.__init__`` would otherwise apply: a connection built
from a bare URL ends up with ``Retry(NoBackoff(), 0)`` — no retries — and
no socket deadlines at all. See :func:`_build_client`.
"""

import redis.asyncio as aioredis

# MUST be the asyncio Retry, not ``redis.retry.Retry``. The two classes have
# identical constructors, so the sync one is accepted without complaint — but
# its ``call_with_retry`` is a plain function that returns the un-awaited
# coroutine from its first attempt, so the try/except never observes a
# failure and NOTHING is ever retried. It fails silently and only under a
# real connection error. ``test_redis_client_resilience_2026_08_13`` pins the
# class and proves recovery against a live Redis.
from redis.asyncio.retry import Retry
from redis.backoff import ExponentialWithJitterBackoff
from redis.exceptions import ConnectionError as RedisConnectionError
from redis.exceptions import TimeoutError as RedisTimeoutError

from app.core.config import settings

_redis: aioredis.Redis | None = None


def _build_client() -> aioredis.Redis:
    """Construct the shared client with connection-resilience settings.

    2026-08-13 incident: four unrelated scheduled jobs failed inside ten
    minutes, all with ``TimeoutError: Timeout connecting to server`` raised
    out of the single-flight lock acquire in
    ``scheduled_jobs.service.execute_job``. The client was created with a
    bare ``from_url(settings.REDIS_URL, decode_responses=True)``, which
    leaves four gaps that make a transient network event terminal:

    * **no retry** — via the ``from_url`` path the connection's retry
      policy is ``Retry(NoBackoff(), 0)``, so a single reaped socket or a
      failover surfaces as an exception to the caller. This is the gap that
      actually recovers the incident: a jittered exponential retry
      disconnects, reconnects and re-runs the command. Verified against a
      real Redis with the connection killed server-side — retry alone
      survives it.
    * **no liveness probe** — a pooled connection idle since the previous
      cron tick is handed straight to the next caller. Crons are the worst
      case: the gap between ticks is minutes to hours, ample time for the
      network path to drop an idle connection with nothing the client can
      observe. ``health_check_interval`` does not recover on its own (the
      same measurement: the PING just fails in the command's place) — what
      it changes is WHERE the failure lands. On a throwaway PING, before
      the caller's command has been written to a half-dead socket, which is
      what makes the retry safe for any command rather than only
      provably-idempotent ones.
    * **no deadlines** — an unanswered TCP handshake blocks on the OS SYN
      retry ceiling (~2 min on Linux) while the caller holds a scheduled
      job concurrency slot, so one Redis blip drains the gate for
      everything else.
    * **no TCP keepalive** — nothing keeps a long-idle connection fresh in
      the middleboxes between the app and Redis.

    Querystring options on ``REDIS_URL`` still win over everything set
    here (redis-py's documented precedence), so an operator can override
    any of it without a deploy.
    """
    return aioredis.from_url(
        settings.REDIS_URL,
        decode_responses=True,
        socket_connect_timeout=settings.REDIS_CONNECT_TIMEOUT_SECS,
        socket_timeout=settings.REDIS_SOCKET_TIMEOUT_SECS,
        socket_keepalive=True,
        health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL_SECS,
        retry=Retry(
            ExponentialWithJitterBackoff(base=0.05, cap=1.0),
            settings.REDIS_MAX_RETRIES,
        ),
        # Explicit, even though these are ``Retry``'s defaults: the
        # supported-error tuple is what actually gates whether a command is
        # retried, and it must not silently change under a redis-py bump.
        retry_on_error=[RedisConnectionError, RedisTimeoutError],
        max_connections=settings.REDIS_MAX_CONNECTIONS,
    )


def get_redis() -> aioredis.Redis:
    global _redis
    if _redis is None:
        _redis = _build_client()
    return _redis


async def close_redis() -> None:
    """Close the shared Redis connection pool on application shutdown."""
    global _redis
    if _redis is not None:
        await _redis.aclose()
        _redis = None