"""A small in-process rate limiter for the few public routes that write.

Sources of truth: this module; `routers/subscribers.py` is the first caller. One Python
process serves the site (react.md), so a dictionary of sliding windows is the whole
mechanism: no store, no dependency, reset on restart. Two windows are checked on every
call, one per client key and one for everybody, so a client that rotates addresses is
still held under the global budget.

The client key is the address the proxies in front of us report: `CF-Connecting-IP` when
the request came through Cloudflare (the edge sets it and a visitor cannot forge it through
the edge), else the last `X-Forwarded-For` hop, which Caddy writes for a direct visitor.
Only Caddy reaches the container. A visitor who reaches the origin directly and forges
`CF-Connecting-IP` can rotate keys, which is why the global window exists: it bounds what
any number of keys can do together. The key is used for counting only and is never stored
or logged with a submission.
"""

from __future__ import annotations

import threading
import time
from collections import deque

from fastapi import Request


class Throttle:
    def __init__(
        self, per_key: int, per_key_seconds: float, total: int, total_seconds: float
    ) -> None:
        self.per_key = per_key
        self.per_key_seconds = per_key_seconds
        self.total = total
        self.total_seconds = total_seconds
        self._keys: dict[str, deque[float]] = {}
        self._all: deque[float] = deque()
        self._lock = threading.Lock()

    def allow(self, key: str, now: float | None = None) -> bool:
        """Record one attempt and say whether it is within both budgets."""
        now = time.monotonic() if now is None else now
        with self._lock:
            self._trim(self._all, now - self.total_seconds)
            window = self._keys.setdefault(key, deque())
            self._trim(window, now - self.per_key_seconds)
            if len(self._all) >= self.total or len(window) >= self.per_key:
                return False
            window.append(now)
            self._all.append(now)
            # Forget idle keys so the dictionary cannot grow without bound.
            if len(self._keys) > 5000:
                for k in [k for k, w in self._keys.items() if not w]:
                    del self._keys[k]
            return True

    @staticmethod
    def _trim(window: deque[float], cutoff: float) -> None:
        while window and window[0] <= cutoff:
            window.popleft()


def client_key(request: Request) -> str:
    """The caller's address as the proxy in front of us reports it."""
    edge = request.headers.get("cf-connecting-ip")
    if edge:
        return edge.strip()
    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        return forwarded.split(",")[-1].strip()
    return request.client.host if request.client else "unknown"