# Rate Limiting Pattern (Clean Architecture + FastAPI)

Rate limiting as an infrastructure service behind a port, with in-memory and Redis backends.

## Overview

Rate limiting is a cross-cutting concern that lives in `core/` (not in any module). It uses the same port/adapter pattern as other infrastructure services.

## Port Design

```python
from typing import Protocol

class RateLimiter(Protocol):
    async def check(self, key: str, max_requests: int, window_seconds: int) -> RateLimitResult: ...
    async def reset(self, key: str) -> None: ...

class RateLimitResult:
    __slots__ = ("allowed", "remaining", "reset_after_seconds")
    def __init__(self, allowed: bool, remaining: int, reset_after_seconds: int) -> None:
        self.allowed = allowed
        self.remaining = remaining
        self.reset_after_seconds = reset_after_seconds
```

## Key Design

Key functions keep rate-limit concerns out of route handlers:

```python
def login_rate_limit_key(ip: str, email: str) -> str:
    return f"login:{ip}:{email}"

def refresh_rate_limit_key(ip: str) -> str:
    return f"refresh:{ip}"

def resolve_rate_limit_key(user_id: int, provider: str) -> str:
    return f"resolve:{user_id}:{provider}"
```

## In-Memory Sliding Window

**Design decisions:**
- Sliding window (not fixed window) — avoids burst edge at window boundaries
- Lazy pruning — expired entries only cleaned on access of that key
- `time.monotonic()` — immune to system clock changes
- Thread-safe for single-process ASGI (but not multi-process — use Redis for that)

## Redis Sorted-Set Sliding Window

For multi-worker deployments, use Redis sorted sets (`ZADD` + `ZREMRANGEBYSCORE` + `ZCARD`) with per-key TTL:

```python
class RedisRateLimiter:
    def __init__(self, redis, key_prefix="ratelimit:"):
        self._redis = redis
        self._key_prefix = key_prefix

    async def check(self, key, max_requests, window_seconds):
        redis_key = f"{self._key_prefix}{key}"
        now = time.time()

        try:
            # Remove expired entries outside the window
            await self._redis.zremrangebyscore(redis_key, 0, now - window_seconds)

            # Count active entries
            count = await self._redis.zcard(redis_key)

            if count < max_requests:
                await self._redis.zadd(redis_key, {str(now): now})
                await self._redis.expire(redis_key, window_seconds)
                remaining = max_requests - (count + 1)
                return RateLimitResult(allowed=True, remaining=remaining, ...)

            # Rate limited — find oldest timestamp for Retry-After
            oldest = await self._redis.zrange(redis_key, 0, 0, withscores=True)
            if oldest and isinstance(oldest[0], tuple):
                _, score = oldest[0]
                reset_after = math.ceil(max(0.0, float(score) + window_seconds - now))
            else:
                reset_after = window_seconds

            return RateLimitResult(allowed=False, remaining=0, reset_after_seconds=max(1, reset_after))
        except Exception:
            # Fail-open: allow through if Redis is down
            return RateLimitResult(allowed=True, remaining=1, ...)
```

**`_RedisClient` Protocol for testing:**

```python
class _RedisClient(Protocol):
    async def zadd(self, key: str, mapping: dict[str | bytes, float]) -> int: ...
    async def zremrangebyscore(self, key: str, min: float | str, max: float | str) -> int: ...
    async def zcard(self, key: str) -> int: ...
    async def zrange(self, key: str, start: int, stop: int, withscores: bool = ...) -> list[tuple[str, float]] | list[str]: ...
    async def expire(self, key: str, seconds: int) -> bool: ...
    async def delete(self, *keys: str) -> int: ...
```

**Design decisions:**
- Sorted set members = timestamp strings, scores = same values
- ZREMRANGEBYSCORE prunes expired entries atomically before counting
- EXPIRE on the key acts as a safety net — if no new requests come, the key self-destructs after window_seconds
- Fail-open: Redis exception → allow request through rather than blocking all users
- `time.time()` (wall clock, not monotonic) since Redis scores are wall-clock timestamps

## FastAPI Integration

### Body Consumption Problem

Rate limiting login by email requires reading the request body. But `await request.json()` consumes the body stream, breaking Pydantic schema parsing in the route handler.

**Solution:** For login, rate-limit by IP only (without email). This still prevents brute-force from a single origin. For IP+email granularity, implement at middleware level with body-caching.

### Dependency Functions

```python
# core/rate_limit_deps.py

# Singleton factory — Redis when configured, in-memory otherwise
_GLOBAL_RATE_LIMITER: RateLimiter | None = None

def get_rate_limiter() -> RateLimiter:
    global _GLOBAL_RATE_LIMITER
    if _GLOBAL_RATE_LIMITER is not None:
        return _GLOBAL_RATE_LIMITER

    settings = get_settings()
    if settings.REDIS_RATE_LIMITER and settings.REDIS_HOST:
        from redis.asyncio import Redis as AsyncRedis
        redis = AsyncRedis(host=..., port=..., db=..., password=..., decode_responses=True)
        _GLOBAL_RATE_LIMITER = RedisRateLimiter(redis)
    else:
        _GLOBAL_RATE_LIMITER = InMemoryRateLimiter()
    return _GLOBAL_RATE_LIMITER

# Login: IP only, 10 requests per 5 minutes
async def rate_limit_login(request: Request) -> None:
    client_ip = request.client.host if request.client else "unknown"
    key = login_rate_limit_key(client_ip, "global")
    result = await _GLOBAL_RATE_LIMITER.check(key, max_requests=10, window_seconds=300)
    if not result.allowed:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Too many login attempts...",
            headers={"Retry-After": str(result.reset_after_seconds)},
        )

# Refresh: IP only, 20 per 5 minutes
async def rate_limit_refresh(request: Request) -> None:
    ...

# Resolve: user_id + provider, 30 per minute
# This one is called from inside the route handler (not a Depends) because
# it needs user_id from JWT and optionally provider from the request body
```

### Route Integration

```python
# For Depends-based (login/refresh):
@router.post("/login")
async def login(
    data: LoginRequest,
    use_cases: ...,
    _rate_limit: None = Depends(rate_limit_login),  # runs before body is consumed? No — Depends runs after
) -> dict[str, object]:
    ...

# For body-dependent (resolve): call inline after request params are available
@router.post("/resolve")
async def resolve_stream(data, user_id, use_cases):
    # ... validate source ...
    rl_result = await _RESOLVE_LIMITER.check(
        resolve_rate_limit_key(user_id, "global"),
        max_requests=30,
        window_seconds=60,
    )
    if not rl_result.allowed:
        return JSONResponse(
            status_code=429,
            content={"error": "Too many requests..."},
            headers={"Retry-After": str(rl_result.reset_after_seconds)},
        )
```

**Important:** `Depends` functions run AFTER the request body has been parsed by Pydantic but BEFORE the route handler executes. The body is available via `request.json()` in a Depends callable because FastAPI buffers the request body. However, `request.json()` consumes the body and subsequent Pydantic parsing will fail. So if you need body fields (email, provider) for rate limiting, either:
1. Use IP-only keys (simplest, recommended)
2. Parse body fields in middleware before route processing
3. Call the rate limiter inline inside the route handler after Pydantic has consumed the body (used above for resolve)

### 429 Response Format

```python
raise HTTPException(
    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
    detail="User-friendly error message",
    headers={"Retry-After": str(result.reset_after_seconds)},
)
```

Always include `Retry-After` header with the number of seconds. This is consumed by HTTP clients and libraries.

## Fail-Open Policy

Rate limiter failures should be **fail-open** by default: if Redis is down, allow the request rather than blocking all traffic. The risk of a DoS during a Redis outage is acceptable compared to the certainty of blocking all users.

```python
# In the rate limit dependency:
try:
    result = await limiter.check(key, max_requests, window_seconds)
    if not result.allowed:
        raise HTTPException(429, ...)
except Exception:
    # Fail open — don't block users if Redis/limiter is down
    pass
```

## Testing

```python
async def test_under_limit_all_allowed(self, limiter):
    for _ in range(3):
        result = await limiter.check("key", max_requests=5, window_seconds=60)
        assert result.allowed is True

async def test_exceeded_limit_denies(self, limiter):
    for _ in range(5):
        await limiter.check("key", max_requests=5, window_seconds=60)
    result = await limiter.check("key", max_requests=5, window_seconds=60)
    assert result.allowed is False
    assert result.remaining == 0

async def test_different_keys_independent(self, limiter):
    for _ in range(10):
        await limiter.check("key-a", max_requests=5, window_seconds=60)
    result = await limiter.check("key-b", max_requests=5, window_seconds=60)
    assert result.allowed is True
```