FabledGrimoire/src/grimoire/rate_limit.py

"""In-process sliding-window rate limiter.

IMPORTANT — deployment note:
  Rate limit counters are stored in memory and are lost on process restart.
  When deployed behind a reverse proxy (nginx, Caddy, Traefik) you MUST set
  TRUST_PROXY_HEADERS=true so that the real client IP is used as the bucket key
  rather than the proxy's IP (which would cause all users to share one bucket).
"""

import asyncio
import time
from collections import defaultdict

_buckets: dict[str, list[float]] = defaultdict(list)
_lock = asyncio.Lock()


async def is_rate_limited(key: str, max_requests: int, window_seconds: int) -> bool:
    """Returns True if request should be blocked (limit exceeded)."""
    async with _lock:
        now = time.monotonic()
        cutoff = now - window_seconds
        _buckets[key] = [t for t in _buckets[key] if t > cutoff]
        if len(_buckets[key]) >= max_requests:
            return True
        _buckets[key].append(now)
        return False