"""Distributed sliding window rate limiter backed by a central Redis store.

This is the multi-replica version of ``SlidingWindowRateLimiter``. The per-client
``deque`` becomes a Redis **sorted set** (one per client, scored by timestamp) and
the per-client ``threading.Lock`` becomes an atomic **Lua script**: Redis runs the
script as one indivisible step, so two replicas can never both observe spare
capacity and over-admit. All replicas share the one Redis, so the quota is global
rather than per-process.
"""

from __future__ import annotations

import uuid
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from redis import Redis

# Runs entirely inside Redis, atomically. Mirrors the local evict/check/append:
#   1. ZREMRANGEBYSCORE  -> drop timestamps older than the window (popleft)
#   2. ZCARD             -> how many remain in the window (len)
#   3. if under limit, ZADD the new request and allow; else deny
#   4. EXPIRE keeps idle clients from leaking memory (the cleanup the local
#      version lacked)
# KEYS[1] = the client's set   ARGV = now, window_seconds, limit, unique_member
_ALLOW_SCRIPT = """
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local member = ARGV[4]

redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window)
local count = redis.call('ZCARD', key)
if count < limit then
    redis.call('ZADD', key, now, member)
    redis.call('EXPIRE', key, math.ceil(window))
    return 1
end
return 0
"""


class RedisSlidingWindowRateLimiter:
    """Exact sliding window rate limiter shared across replicas via Redis.

    Each replica constructs one of these against the same Redis. Because the
    decision is made by an atomic server-side script, the limit is enforced
    globally no matter how many replicas call concurrently.
    """

    def __init__(self, redis_client: Redis, limit: int, window_seconds: float, prefix: str = "ratelimit") -> None:
        """Initialise the limiter.

        Args:
            redis_client (Redis): A connected ``redis.Redis`` client.
            limit (int): Maximum allowed requests within the window.
            window_seconds (float): Length of the rolling window in seconds.
            prefix (str): Key prefix namespacing this limiter's keys in Redis.
        """
        self.limit = limit
        self.window_seconds = window_seconds
        self.prefix = prefix
        # register_script ships the Lua to Redis once and calls it by SHA there
        # after, so the script body is not resent on every request.
        self._allow = redis_client.register_script(_ALLOW_SCRIPT)

    def is_allowed(self, client_id: str, now: float) -> bool:
        """Decide whether a request from ``client_id`` is within its global quota.

        Args:
            client_id (str): Identifier for the calling client.
            now (float): Current time in seconds. Use a shared wall-clock source
                (e.g. ``time.time()``) so every replica agrees on the window;
                ``time.monotonic`` is only comparable within one process.

        Returns:
            bool: ``True`` if the request is allowed, ``False`` if it is throttled.
        """
        key = f"{self.prefix}:{client_id}"
        # A unique member so two requests with an identical timestamp are both
        # recorded (sorted-set members must be distinct).
        member = f"{now}:{uuid.uuid4().hex}"
        result = self._allow(keys=[key], args=[now, self.window_seconds, self.limit, member])
        return bool(result)