Rate Limiting at Scale.

class FixedWindowLimiter:
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters: dict[str, tuple[int, int]] = {}  # key -> (count, window_start)
    
    def is_allowed(self, key: str) -> bool:
        now = int(time.time())
        window_start = now - (now % self.window_seconds)
        
        count, stored_window = self.counters.get(key, (0, window_start))
        
        # New window - reset counter
        if stored_window != window_start:
            count = 0
            stored_window = window_start
        
        if count < self.limit:
            self.counters[key] = (count + 1, stored_window)
            return True
        return False

class SlidingWindowLogLimiter:
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window_seconds = window_seconds
        self.logs: dict[str, list[float]] = {}
    
    def is_allowed(self, key: str) -> bool:
        now = time.time()
        window_start = now - self.window_seconds
        
        # Get existing timestamps, filter out old ones
        timestamps = self.logs.get(key, [])
        timestamps = [t for t in timestamps if t > window_start]
        
        if len(timestamps) < self.limit:
            timestamps.append(now)
            self.logs[key] = timestamps
            return True
        return False

class SlidingWindowCounterLimiter:
    def __init__(self, limit: int, window_seconds: int):
        self.limit = limit
        self.window_seconds = window_seconds
        self.counters: dict[str, dict] = {}
    
    def is_allowed(self, key: str) -> bool:
        now = time.time()
        current_window = int(now // self.window_seconds)
        window_position = now % self.window_seconds
        
        data = self.counters.get(key, {
            'prev_count': 0,
            'prev_window': current_window - 1,
            'curr_count': 0,
            'curr_window': current_window
        })
        
        # Handle window transitions
        if data['curr_window'] < current_window:
            data['prev_count'] = data['curr_count'] if data['curr_window'] == current_window - 1 else 0
            data['prev_window'] = data['curr_window']
            data['curr_count'] = 0
            data['curr_window'] = current_window
        
        # Calculate weighted count
        prev_weight = (self.window_seconds - window_position) / self.window_seconds
        weighted_count = (data['prev_count'] * prev_weight) + data['curr_count']
        
        if weighted_count < self.limit:
            data['curr_count'] += 1
            self.counters[key] = data
            return True
        return False

class TokenBucketLimiter:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # tokens per second
        self.capacity = capacity
        self.buckets: dict[str, tuple[float, float]] = {}  # key -> (tokens, last_update)
    
    def is_allowed(self, key: str, tokens_requested: int = 1) -> bool:
        now = time.time()
        
        current_tokens, last_update = self.buckets.get(key, (self.capacity, now))
        
        # Add tokens based on time elapsed
        elapsed = now - last_update
        current_tokens = min(self.capacity, current_tokens + (elapsed * self.rate))
        
        if current_tokens >= tokens_requested:
            current_tokens -= tokens_requested
            self.buckets[key] = (current_tokens, now)
            return True
        else:
            self.buckets[key] = (current_tokens, now)
            return False

import redis
import time

class RedisRateLimiter:
    def __init__(self, redis_client: redis.Redis, limit: int, window_seconds: int):
        self.redis = redis_client
        self.limit = limit
        self.window_seconds = window_seconds
    
    def is_allowed(self, key: str) -> tuple[bool, dict]:
        """
        Returns (allowed, info) where info contains remaining quota and reset time.
        Uses the sliding window log algorithm (one sorted-set entry per request).
        """
        now = time.time()
        window_key = f"ratelimit:{key}"
        
        # Use Redis pipeline for atomic operations
        pipe = self.redis.pipeline()
        
        # Remove old entries (outside window)
        window_start = now - self.window_seconds
        pipe.zremrangebyscore(window_key, 0, window_start)
        
        # Count current entries
        pipe.zcard(window_key)
        
        # Add current request (optimistically)
        pipe.zadd(window_key, {f"{now}:{id(now)}": now})
        
        # Set expiry on the key
        pipe.expire(window_key, self.window_seconds)
        
        results = pipe.execute()
        current_count = results[1]
        
        if current_count < self.limit:
            remaining = self.limit - current_count - 1
            return True, {
                'allowed': True,
                'remaining': max(0, remaining),
                'reset': int(now + self.window_seconds),
                'limit': self.limit
            }
        else:
            # Over limit - remove the optimistically added entry
            self.redis.zrem(window_key, f"{now}:{id(now)}")
            return False, {
                'allowed': False,
                'remaining': 0,
                'reset': int(now + self.window_seconds),
                'limit': self.limit
            }

-- Token bucket rate limiter as a Lua script (atomic execution)
-- KEYS[1] = rate limit key
-- ARGV[1] = rate (tokens per second)
-- ARGV[2] = capacity (max tokens)
-- ARGV[3] = current timestamp
-- ARGV[4] = tokens requested

local key = KEYS[1]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

-- Get current state
local data = redis.call('HMGET', key, 'tokens', 'last_update')
local tokens = tonumber(data[1]) or capacity
local last_update = tonumber(data[2]) or now

-- Calculate token refill
local elapsed = now - last_update
local new_tokens = math.min(capacity, tokens + (elapsed * rate))

-- Check if request can be fulfilled
local allowed = 0
if new_tokens >= requested then
    new_tokens = new_tokens - requested
    allowed = 1
end

-- Update state
redis.call('HMSET', key, 'tokens', new_tokens, 'last_update', now)
redis.call('EXPIRE', key, math.ceil(capacity / rate) * 2)

return {allowed, new_tokens, capacity}

class RedisTokenBucket:
    LUA_SCRIPT = """..."""  # The Lua script above
    
    def __init__(self, redis_client: redis.Redis, rate: float, capacity: int):
        self.redis = redis_client
        self.rate = rate
        self.capacity = capacity
        self.script = self.redis.register_script(self.LUA_SCRIPT)
    
    def is_allowed(self, key: str, tokens: int = 1) -> tuple[bool, float]:
        now = time.time()
        result = self.script(
            keys=[f"tokenbucket:{key}"],
            args=[self.rate, self.capacity, now, tokens]
        )
        return bool(result[0]), result[1]  # (allowed, remaining_tokens)

import threading
import time

class LocalSyncRateLimiter:
    def __init__(
        self,
        redis_client: redis.Redis,
        local_limit: int,
        global_limit: int,
        window_seconds: int,
        sync_interval: float = 1.0
    ):
        self.redis = redis_client
        self.local_limit = local_limit  # Per-server allowance
        self.global_limit = global_limit
        self.window_seconds = window_seconds
        self.sync_interval = sync_interval
        
        self.local_counts: dict[str, int] = {}
        self.lock = threading.Lock()
        
        # Start sync thread
        self.sync_thread = threading.Thread(target=self._sync_loop, daemon=True)
        self.sync_thread.start()
    
    def is_allowed(self, key: str) -> bool:
        with self.lock:
            count = self.local_counts.get(key, 0)
            if count < self.local_limit:
                self.local_counts[key] = count + 1
                return True
            return False
    
    def _sync_loop(self):
        while True:
            time.sleep(self.sync_interval)
            self._sync_to_redis()
    
    def _sync_to_redis(self):
        with self.lock:
            counts_to_sync = self.local_counts.copy()
            self.local_counts.clear()
        
        pipe = self.redis.pipeline()
        for key, count in counts_to_sync.items():
            redis_key = f"ratelimit:{key}"
            pipe.incrby(redis_key, count)
            pipe.expire(redis_key, self.window_seconds)
        pipe.execute()

class ResilientRateLimiter:
    def __init__(
        self,
        redis_client: redis.Redis,
        global_limit: int,
        local_fallback_limit: int,
        window_seconds: int,
        failure_mode: str = 'over_allow'  # or 'under_allow'
    ):
        self.redis = redis_client
        self.global_limit = global_limit
        self.local_fallback_limit = local_fallback_limit
        self.window_seconds = window_seconds
        self.failure_mode = failure_mode
        
        self.local_limiter = SlidingWindowCounterLimiter(
            local_fallback_limit, window_seconds
        )
        self.redis_available = True
        self.last_redis_check = 0
        self.redis_check_interval = 5  # seconds
    
    def is_allowed(self, key: str) -> tuple[bool, str]:
        """Returns (allowed, mode) where mode is 'redis', 'local', or 'failed'."""
        
        # Try Redis first
        if self._redis_is_available():
            try:
                return self._check_redis(key), 'redis'
            except redis.RedisError:
                self.redis_available = False
        
        # Redis unavailable - use fallback strategy
        if self.failure_mode == 'over_allow':
            # Use local limiter as soft limit
            allowed = self.local_limiter.is_allowed(key)
            return allowed, 'local'
        else:
            # Fail closed - but maybe allow some traffic
            if self.local_fallback_limit > 0:
                # Very conservative local limit
                allowed = self.local_limiter.is_allowed(key)
                return allowed, 'local_strict'
            return False, 'failed'
    
    def _redis_is_available(self) -> bool:
        now = time.time()
        if not self.redis_available:
            # Periodically retry Redis
            if now - self.last_redis_check > self.redis_check_interval:
                self.last_redis_check = now
                try:
                    self.redis.ping()
                    self.redis_available = True
                except redis.RedisError:
                    pass
        return self.redis_available
    
    def _check_redis(self, key: str) -> bool:
        # Implement your Redis rate limit check here
        ...

from prometheus_client import Counter, Histogram, Gauge

# Metrics
rate_limit_decisions = Counter(
    'rate_limit_decisions_total',
    'Rate limit decisions',
    ['result', 'mode']  # result=allowed/denied, mode=redis/local/failed
)

rate_limit_latency = Histogram(
    'rate_limit_check_seconds',
    'Time to check rate limit',
    buckets=[.001, .005, .01, .025, .05, .1, .25, .5, 1]
)

redis_available = Gauge(
    'rate_limiter_redis_available',
    'Whether Redis is available for rate limiting'
)

class InstrumentedRateLimiter:
    def __init__(self, limiter: ResilientRateLimiter):
        self.limiter = limiter
    
    def is_allowed(self, key: str) -> bool:
        with rate_limit_latency.time():
            allowed, mode = self.limiter.is_allowed(key)
        
        result = 'allowed' if allowed else 'denied'
        rate_limit_decisions.labels(result=result, mode=mode).inc()
        redis_available.set(1 if self.limiter.redis_available else 0)
        
        return allowed

@dataclass
class RateLimitRule:
    id: str
    name: str
    key_pattern: str  # e.g., "user:{user_id}", "api_key:{api_key}"
    limit: int
    window_seconds: int
    algorithm: str  # 'sliding_window' or 'token_bucket'
    burst_capacity: Optional[int]  # For token bucket
    scope: str  # 'per_user', 'per_api_key', 'global'
    tier: Optional[str]  # 'free', 'pro', 'enterprise'
    
# Example rules
rules = [
    RateLimitRule(
        id="user_basic",
        name="Basic user limit",
        key_pattern="user:{user_id}",
        limit=100,
        window_seconds=60,
        algorithm="sliding_window",
        scope="per_user",
        tier="free"
    ),
    RateLimitRule(
        id="user_pro",
        name="Pro user limit",
        key_pattern="user:{user_id}",
        limit=1000,
        window_seconds=60,
        algorithm="sliding_window",
        scope="per_user",
        tier="pro"
    ),
    RateLimitRule(
        id="api_key",
        name="API key limit",
        key_pattern="apikey:{api_key}",
        limit=10000,
        window_seconds=3600,
        algorithm="token_bucket",
        burst_capacity=500,
        scope="per_api_key"
    ),
    RateLimitRule(
        id="global_writes",
        name="Global write limit",
        key_pattern="global:writes",
        limit=5000,
        window_seconds=1,
        algorithm="sliding_window",
        scope="global"
    )
]

class MultiLevelRateLimiter:
    def __init__(self, redis_client: redis.Redis, rules: list[RateLimitRule]):
        self.redis = redis_client
        self.rules = rules
        self.limiters = self._create_limiters()
    
    def _create_limiters(self) -> dict[str, RateLimiter]:
        limiters = {}
        for rule in self.rules:
            if rule.algorithm == 'sliding_window':
                limiters[rule.id] = RedisSlidingWindowLimiter(
                    self.redis, rule.limit, rule.window_seconds
                )
            elif rule.algorithm == 'token_bucket':
                limiters[rule.id] = RedisTokenBucket(
                    self.redis, 
                    rate=rule.limit / rule.window_seconds,
                    capacity=rule.burst_capacity or rule.limit
                )
        return limiters
    
    def check_request(self, context: RequestContext) -> RateLimitResult:
        """
        Check all applicable rules for this request.
        Returns the most restrictive result.
        """
        applicable_rules = self._get_applicable_rules(context)
        
        results = []
        for rule in applicable_rules:
            key = self._build_key(rule, context)
            limiter = self.limiters[rule.id]
            
            allowed, info = limiter.is_allowed(key)
            results.append(RateLimitResult(
                rule_id=rule.id,
                allowed=allowed,
                limit=rule.limit,
                remaining=info.get('remaining', 0),
                reset=info.get('reset', 0)
            ))
        
        # If any rule denies, the request is denied
        denied_results = [r for r in results if not r.allowed]
        if denied_results:
            # Return the result with the longest wait time
            return max(denied_results, key=lambda r: r.reset)
        
        # All rules allow - return the one with lowest remaining quota
        return min(results, key=lambda r: r.remaining)
    
    def _get_applicable_rules(self, context: RequestContext) -> list[RateLimitRule]:
        applicable = []
        for rule in self.rules:
            # Check tier match
            if rule.tier and rule.tier != context.user_tier:
                continue
            # Check scope
            if rule.scope == 'per_user' and not context.user_id:
                continue
            if rule.scope == 'per_api_key' and not context.api_key:
                continue
            applicable.append(rule)
        return applicable
    
    def _build_key(self, rule: RateLimitRule, context: RequestContext) -> str:
        key = rule.key_pattern
        key = key.replace('{user_id}', context.user_id or 'anonymous')
        key = key.replace('{api_key}', context.api_key or 'none')
        key = key.replace('{ip}', context.ip_address)
        return key

def add_rate_limit_headers(response: Response, result: RateLimitResult):
    response.headers['X-RateLimit-Limit'] = str(result.limit)
    response.headers['X-RateLimit-Remaining'] = str(result.remaining)
    response.headers['X-RateLimit-Reset'] = str(result.reset)
    
    if not result.allowed:
        response.headers['Retry-After'] = str(result.reset - int(time.time()))

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import redis
from typing import Optional

app = FastAPI()

class RateLimitMiddleware:
    def __init__(self, app: FastAPI, limiter: MultiLevelRateLimiter):
        self.app = app
        self.limiter = limiter
    
    async def __call__(self, request: Request, call_next):
        # Extract context from request
        context = RequestContext(
            user_id=request.headers.get('X-User-ID'),
            api_key=request.headers.get('X-API-Key'),
            ip_address=request.client.host,
            user_tier=request.headers.get('X-User-Tier', 'free'),
            endpoint=request.url.path,
            method=request.method
        )
        
        # Check rate limits
        result = self.limiter.check_request(context)
        
        if not result.allowed:
            response = JSONResponse(
                status_code=429,
                content={
                    'error': 'rate_limit_exceeded',
                    'message': f'Rate limit exceeded. Retry after {result.reset - int(time.time())} seconds.',
                    'limit': result.limit,
                    'reset': result.reset
                }
            )
            add_rate_limit_headers(response, result)
            return response
        
        # Proceed with request
        response = await call_next(request)
        add_rate_limit_headers(response, result)
        return response


# Setup
redis_client = redis.RedisCluster(
    host='redis-cluster.example.com',
    port=6379,
    decode_responses=True
)

limiter = MultiLevelRateLimiter(redis_client, rules)
app.middleware('http')(RateLimitMiddleware(app, limiter))

TIER_LIMITS = {
    'free': {
        'requests_per_minute': 60,
        'requests_per_day': 1000,
        'burst': 10
    },
    'pro': {
        'requests_per_minute': 600,
        'requests_per_day': 50000,
        'burst': 100
    },
    'enterprise': {
        'requests_per_minute': 6000,
        'requests_per_day': None,  # Unlimited
        'burst': 1000
    }
}

def get_limits_for_user(user: User) -> dict:
    return TIER_LIMITS.get(user.tier, TIER_LIMITS['free'])

ENDPOINT_COSTS = {
    'GET /api/users': 1,
    'POST /api/users': 5,
    'GET /api/reports': 10,  # Expensive query
    'POST /api/ai/generate': 100,  # Very expensive
}

class CostAwareRateLimiter:
    def __init__(self, limiter: TokenBucketLimiter):
        self.limiter = limiter
    
    def check_request(self, user_id: str, endpoint: str) -> bool:
        cost = ENDPOINT_COSTS.get(endpoint, 1)
        return self.limiter.is_allowed(user_id, tokens_requested=cost)

class AdaptiveRateLimiter:
    def __init__(self, base_limiter: RateLimiter, load_monitor: LoadMonitor):
        self.limiter = base_limiter
        self.load_monitor = load_monitor
    
    def get_effective_limit(self, base_limit: int) -> int:
        load = self.load_monitor.get_current_load()  # 0.0 to 1.0
        
        if load < 0.5:
            # Low load - allow 20% more
            return int(base_limit * 1.2)
        elif load < 0.8:
            # Normal load - use base limit
            return base_limit
        elif load < 0.95:
            # High load - reduce by 20%
            return int(base_limit * 0.8)
        else:
            # Critical load - reduce by 50%
            return int(base_limit * 0.5)

EXEMPT_PATTERNS = [
    '/health',
    '/metrics',
    '/api/webhooks/stripe',  # Stripe webhook callbacks
]

EXEMPT_API_KEYS = {
    'internal-service-key-123',
    'monitoring-key-456',
}

def should_exempt(request: Request) -> bool:
    # Check path exemptions
    for pattern in EXEMPT_PATTERNS:
        if request.url.path.startswith(pattern):
            return True
    
    # Check API key exemptions
    api_key = request.headers.get('X-API-Key')
    if api_key in EXEMPT_API_KEYS:
        return True
    
    return False

# Kong rate limiting configuration
plugins:
  - name: rate-limiting
    config:
      minute: 100
      hour: 1000
      policy: redis
      redis_host: redis.example.com
      redis_port: 6379
      redis_database: 0
      fault_tolerant: true  # Continue if Redis fails
      hide_client_headers: false

try:
    return redis_limiter.is_allowed(key)
except redis.ClusterDownError:
    return local_fallback_limiter.is_allowed(key)

class RateLimitRule:
    ...
    failure_mode: str  # 'open' or 'closed'

# Defense against key enumeration
class SecureRateLimiter:
    def __init__(self, max_unique_keys: int = 1_000_000):
        self.max_unique_keys = max_unique_keys
    
    def is_allowed(self, key: str) -> bool:
        # Use server time, not client time
        now = time.time()
        
        # Limit total keys tracked (with LRU eviction)
        if self.key_count() > self.max_unique_keys:
            self.evict_oldest_keys()
        
        # For anonymous users, use IP + user agent hash
        # This makes it harder to generate unique keys
        ...

# Defense against distributed attacks
class IPSubnetLimiter:
    def is_allowed(self, ip: str) -> bool:
        # Limit per /24 subnet, not just per IP
        subnet = self.get_subnet(ip, prefix_length=24)
        return self.limiter.is_allowed(f"subnet:{subnet}")

import redis
import time
from dataclasses import dataclass
from typing import Optional, Tuple
from enum import Enum

class RateLimitAlgorithm(Enum):
    SLIDING_WINDOW = "sliding_window"
    TOKEN_BUCKET = "token_bucket"

@dataclass
class RateLimitResult:
    allowed: bool
    limit: int
    remaining: int
    reset: int
    retry_after: Optional[int] = None

class ProductionRateLimiter:
    """
    Production-ready rate limiter with:
    - Multiple algorithm support
    - Graceful degradation
    - Comprehensive metrics
    - Multi-key support
    """
    
    SLIDING_WINDOW_SCRIPT = """
    local key = KEYS[1]
    local now = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local limit = tonumber(ARGV[3])
    
    local window_start = now - window
    redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
    local count = redis.call('ZCARD', key)
    
    if count < limit then
        redis.call('ZADD', key, now, now .. ':' .. math.random())
        redis.call('EXPIRE', key, window)
        return {1, limit - count - 1, math.ceil(now + window)}
    else
        local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
        local reset = oldest[2] and (oldest[2] + window) or (now + window)
        return {0, 0, math.ceil(reset)}
    end
    """
    
    TOKEN_BUCKET_SCRIPT = """
    local key = KEYS[1]
    local rate = tonumber(ARGV[1])
    local capacity = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local requested = tonumber(ARGV[4])
    
    local data = redis.call('HMGET', key, 'tokens', 'last_update')
    local tokens = tonumber(data[1]) or capacity
    local last_update = tonumber(data[2]) or now
    
    local elapsed = now - last_update
    tokens = math.min(capacity, tokens + (elapsed * rate))
    
    local allowed = 0
    if tokens >= requested then
        tokens = tokens - requested
        allowed = 1
    end
    
    redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
    redis.call('EXPIRE', key, math.ceil(capacity / rate) * 2)
    
    local reset = now + ((capacity - tokens) / rate)
    return {allowed, math.floor(tokens), math.ceil(reset)}
    """
    
    def __init__(
        self,
        redis_client: redis.Redis,
        default_limit: int = 100,
        default_window: int = 60,
        algorithm: RateLimitAlgorithm = RateLimitAlgorithm.SLIDING_WINDOW,
        failure_mode: str = 'open',
        local_fallback_limit: int = 10
    ):
        self.redis = redis_client
        self.default_limit = default_limit
        self.default_window = default_window
        self.algorithm = algorithm
        self.failure_mode = failure_mode
        self.local_fallback_limit = local_fallback_limit
        
        # Register Lua scripts
        self.sliding_window_script = self.redis.register_script(self.SLIDING_WINDOW_SCRIPT)
        self.token_bucket_script = self.redis.register_script(self.TOKEN_BUCKET_SCRIPT)
        
        # Local fallback
        self.local_counts: dict[str, list[float]] = {}
        self.redis_healthy = True
    
    def is_allowed(
        self,
        key: str,
        limit: Optional[int] = None,
        window: Optional[int] = None,
        cost: int = 1
    ) -> RateLimitResult:
        """
        Check if request is allowed under rate limit.
        """
        limit = limit or self.default_limit
        window = window or self.default_window
        
        try:
            if self.algorithm == RateLimitAlgorithm.SLIDING_WINDOW:
                result = self._sliding_window_check(key, limit, window)
            else:
                result = self._token_bucket_check(key, limit, window, cost)
            
            self.redis_healthy = True
            return result
            
        except redis.RedisError as e:
            self.redis_healthy = False
            return self._fallback_check(key, limit, window)
    
    def _sliding_window_check(
        self,
        key: str,
        limit: int,
        window: int
    ) -> RateLimitResult:
        now = time.time()
        result = self.sliding_window_script(
            keys=[f"ratelimit:sw:{key}"],
            args=[now, window, limit]
        )
        
        return RateLimitResult(
            allowed=bool(result[0]),
            limit=limit,
            remaining=result[1],
            reset=result[2],
            retry_after=None if result[0] else max(0, result[2] - int(now))
        )
    
    def _token_bucket_check(
        self,
        key: str,
        limit: int,
        window: int,
        cost: int
    ) -> RateLimitResult:
        now = time.time()
        rate = limit / window
        capacity = limit
        
        result = self.token_bucket_script(
            keys=[f"ratelimit:tb:{key}"],
            args=[rate, capacity, now, cost]
        )
        
        return RateLimitResult(
            allowed=bool(result[0]),
            limit=limit,
            remaining=result[1],
            reset=result[2],
            retry_after=None if result[0] else max(0, result[2] - int(now))
        )
    
    def _fallback_check(self, key: str, limit: int, window: int) -> RateLimitResult:
        """
        Local fallback when Redis is unavailable.
        """
        if self.failure_mode == 'closed':
            return RateLimitResult(
                allowed=False,
                limit=limit,
                remaining=0,
                reset=int(time.time()) + window,
                retry_after=window
            )
        
        # Fail open with local limiting
        now = time.time()
        window_start = now - window
        
        if key not in self.local_counts:
            self.local_counts[key] = []
        
        # Remove old entries
        self.local_counts[key] = [t for t in self.local_counts[key] if t > window_start]
        
        count = len(self.local_counts[key])
        local_limit = min(self.local_fallback_limit, limit)
        
        if count < local_limit:
            self.local_counts[key].append(now)
            return RateLimitResult(
                allowed=True,
                limit=local_limit,
                remaining=local_limit - count - 1,
                reset=int(now + window)
            )
        else:
            return RateLimitResult(
                allowed=False,
                limit=local_limit,
                remaining=0,
                reset=int(now + window),
                retry_after=window
            )


# FastAPI integration
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse

def rate_limit_middleware(limiter: ProductionRateLimiter):
    async def middleware(request: Request, call_next):
        # Extract key from request
        key = request.headers.get('X-API-Key') or request.client.host
        
        result = limiter.is_allowed(key)
        
        if not result.allowed:
            return JSONResponse(
                status_code=429,
                content={
                    'error': 'rate_limit_exceeded',
                    'retry_after': result.retry_after
                },
                headers={
                    'X-RateLimit-Limit': str(result.limit),
                    'X-RateLimit-Remaining': str(result.remaining),
                    'X-RateLimit-Reset': str(result.reset),
                    'Retry-After': str(result.retry_after)
                }
            )
        
        response = await call_next(request)
        response.headers['X-RateLimit-Limit'] = str(result.limit)
        response.headers['X-RateLimit-Remaining'] = str(result.remaining)
        response.headers['X-RateLimit-Reset'] = str(result.reset)
        
        return response
    
    return middleware

import pytest
import time
import fakeredis
from unittest.mock import patch, MagicMock

class TestSlidingWindowLimiter:
    @pytest.fixture
    def limiter(self):
        redis_client = fakeredis.FakeRedis()
        return ProductionRateLimiter(
            redis_client=redis_client,
            default_limit=10,
            default_window=60,
            algorithm=RateLimitAlgorithm.SLIDING_WINDOW
        )
    
    def test_allows_under_limit(self, limiter):
        for i in range(10):
            result = limiter.is_allowed('user1')
            assert result.allowed, f"Request {i+1} should be allowed"
            assert result.remaining == 10 - i - 1
    
    def test_blocks_over_limit(self, limiter):
        for _ in range(10):
            limiter.is_allowed('user1')
        
        result = limiter.is_allowed('user1')
        assert not result.allowed
        assert result.remaining == 0
        assert result.retry_after > 0
    
    def test_separate_keys(self, limiter):
        for _ in range(10):
            limiter.is_allowed('user1')
        
        # Different key should still be allowed
        result = limiter.is_allowed('user2')
        assert result.allowed
    
    def test_window_reset(self, limiter):
        limiter.default_window = 1  # 1 second window
        
        for _ in range(10):
            limiter.is_allowed('user1')
        
        result = limiter.is_allowed('user1')
        assert not result.allowed
        
        time.sleep(1.1)
        
        result = limiter.is_allowed('user1')
        assert result.allowed


class TestFallbackBehavior:
    def test_fail_open(self):
        redis_client = MagicMock()
        redis_client.register_script.return_value = MagicMock(
            side_effect=redis.RedisError("Connection refused")
        )
        
        limiter = ProductionRateLimiter(
            redis_client=redis_client,
            failure_mode='open',
            local_fallback_limit=5
        )
        
        # Should use local fallback
        for i in range(5):
            result = limiter.is_allowed('user1')
            assert result.allowed
        
        # Local limit exceeded
        result = limiter.is_allowed('user1')
        assert not result.allowed
    
    def test_fail_closed(self):
        redis_client = MagicMock()
        redis_client.register_script.return_value = MagicMock(
            side_effect=redis.RedisError("Connection refused")
        )
        
        limiter = ProductionRateLimiter(
            redis_client=redis_client,
            failure_mode='closed'
        )
        
        result = limiter.is_allowed('user1')
        assert not result.allowed