# コード例 1: Token Bucket 実装 (Redis + Lua Script)
import redis
import time
from typing import NamedTuple
 
class RateLimitResult(NamedTuple):
    """レート制限の判定結果"""
    allowed: bool
    remaining: int
    reset_at: float
    limit: int
 
class TokenBucketLimiter:
    """
    Token Bucket レートリミッター。
 
    WHY Lua Script を使うのか:
    1. アトミック操作: 複数の Redis コマンドを1回のラウンドトリップで実行
    2. レースコンディション防止: 複数サーバーからの同時アクセスでも正確
    3. パフォーマンス: ネットワークラウンドトリップを最小化
 
    WHY Redis Hash を使うのか:
    1. tokens と last_refill を1つのキーで管理 → アトミックに更新可能
    2. TTL で自動クリーンアップ → メモリリーク防止
    """
 
    LUA_SCRIPT = """
    local key = KEYS[1]
    local max_tokens = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])   -- tokens per second
    local now = tonumber(ARGV[3])
 
    -- 現在の状態を取得
    local data = redis.call('HMGET', key, 'tokens', 'last_refill')
    local tokens = tonumber(data[1]) or max_tokens
    local last_refill = tonumber(data[2]) or now
 
    -- トークン補充 (経過時間に比例)
    local elapsed = math.max(0, now - last_refill)
    local new_tokens = math.min(max_tokens, tokens + elapsed * refill_rate)
 
    if new_tokens >= 1 then
        -- トークン消費: 許可
        new_tokens = new_tokens - 1
        redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
        redis.call('EXPIRE', key, math.ceil(max_tokens / refill_rate) * 2)
        return {1, math.floor(new_tokens), 0}   -- {許可, 残りトークン, 待ち時間}
    else
        -- トークン不足: 拒否
        local wait_time = (1 - new_tokens) / refill_rate
        return {0, 0, math.ceil(wait_time)}      -- {拒否, 残り0, 待ち時間(秒)}
    end
    """
 
    def __init__(self, redis_client: redis.Redis,
                 max_tokens: int = 100,
                 refill_rate: float = 10.0):
        """
        Args:
            redis_client: Redis クライアント
            max_tokens: バケット容量 (バースト許容量)
            refill_rate: 補充レート (tokens/sec)
        """
        self._redis = redis_client
        self._max_tokens = max_tokens
        self._refill_rate = refill_rate
        self._script = self._redis.register_script(self.LUA_SCRIPT)
 
    def allow_request(self, key: str) -> RateLimitResult:
        """
        リクエストを許可するか判定する。
 
        Args:
            key: レートリミットキー (例: "user:123:/api/orders")
 
        Returns:
            RateLimitResult: 判定結果
        """
        now = time.time()
        result = self._script(
            keys=[f"ratelimit:tb:{key}"],
            args=[self._max_tokens, self._refill_rate, now]
        )
        allowed = bool(result[0])
        remaining = int(result[1])
        retry_after = float(result[2])
 
        return RateLimitResult(
            allowed=allowed,
            remaining=remaining,
            reset_at=now + retry_after if not allowed else 0,
            limit=self._max_tokens,
        )
 
# 使用例
limiter = TokenBucketLimiter(
    redis.Redis(host='localhost'),
    max_tokens=100,    # バースト: 最大100リクエスト
    refill_rate=10,    # 定常: 10 req/sec
)
 
result = limiter.allow_request("user:123:/api/orders")
if not result.allowed:
    return Response(
        content='{"error":"Rate limit exceeded"}',
        status_code=429,
        headers={
            "Retry-After": str(int(result.reset_at - time.time())),
            "X-RateLimit-Limit": str(result.limit),
            "X-RateLimit-Remaining": "0",
        },
    )

# コード例 2: Sliding Window Log 実装
class SlidingWindowLogLimiter:
    """
    Sliding Window Log レートリミッター。
 
    正確な時間窓でのレート制限を提供する。
    各リクエストのタイムスタンプを Redis Sorted Set に記録し、
    ウィンドウ内のリクエスト数をカウントする。
 
    トレードオフ:
    - 長所: 最も正確なカウント、Fixed Window の境界問題なし
    - 短所: メモリ使用量が O(N)、高トラフィックではコスト大
    """
 
    LUA_SCRIPT = """
    local key = KEYS[1]
    local max_requests = tonumber(ARGV[1])
    local window_size = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local request_id = ARGV[4]
 
    -- ウィンドウ外の古いエントリを削除
    redis.call('ZREMRANGEBYSCORE', key, 0, now - window_size)
 
    -- 現在のカウント
    local count = redis.call('ZCARD', key)
 
    if count < max_requests then
        -- リクエストを記録 (member はユニークにする)
        redis.call('ZADD', key, now, request_id)
        redis.call('EXPIRE', key, window_size + 1)
        return {1, max_requests - count - 1}   -- 許可, 残り
    else
        -- 制限超過: 最も古いリクエストの時刻を取得して待ち時間を計算
        local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
        local retry_after = 0
        if #oldest > 0 then
            retry_after = math.ceil(tonumber(oldest[2]) + window_size - now)
        end
        return {0, 0, retry_after}             -- 拒否, 残り0, 待ち時間
    end
    """
 
    def __init__(self, redis_client, max_requests: int = 100,
                 window_seconds: int = 60):
        self._redis = redis_client
        self._max = max_requests
        self._window = window_seconds
        self._script = self._redis.register_script(self.LUA_SCRIPT)
 
    def allow_request(self, key: str) -> RateLimitResult:
        now = time.time()
        request_id = f"{now}:{id(self)}:{hash(key)}"
 
        result = self._script(
            keys=[f"ratelimit:sw:{key}"],
            args=[self._max, self._window, now, request_id]
        )
 
        allowed = bool(result[0])
        remaining = int(result[1])
        retry_after = float(result[2]) if len(result) > 2 else 0
 
        return RateLimitResult(
            allowed=allowed,
            remaining=remaining,
            reset_at=now + retry_after if not allowed else 0,
            limit=self._max,
        )

# コード例 3: Fixed Window Counter (最もシンプル)
class FixedWindowLimiter:
    """
    Fixed Window Counter レートリミッター。
 
    最もシンプルな実装。Redis の INCR + EXPIRE で実現。
    境界問題があるため、精度が必要な場面には不向き。
 
    WHY それでも使われるのか:
    1. 実装が最もシンプル (Lua Script 不要)
    2. メモリ効率が最良 (O(1))
    3. 処理速度が最速
    4. DDoS 防御など「大まかに制限できれば良い」ケースに最適
    """
 
    def __init__(self, redis_client, max_requests: int = 100,
                 window_seconds: int = 60):
        self._redis = redis_client
        self._max = max_requests
        self._window = window_seconds
 
    def allow_request(self, key: str) -> RateLimitResult:
        """
        Fixed Window でリクエストを判定。
 
        キー設計: ratelimit:fw:{key}:{window_id}
        window_id = int(now / window_seconds)
        → ウィンドウが変わると新しいキーが使われる
        → 古いキーは EXPIRE で自動削除
        """
        now = time.time()
        window_id = int(now) // self._window
        window_key = f"ratelimit:fw:{key}:{window_id}"
 
        pipe = self._redis.pipeline()
        pipe.incr(window_key)
        pipe.expire(window_key, self._window + 1)  # +1秒マージン
        count, _ = pipe.execute()
 
        allowed = count <= self._max
        remaining = max(0, self._max - count)
        reset_at = (window_id + 1) * self._window
 
        return RateLimitResult(
            allowed=allowed,
            remaining=remaining,
            reset_at=reset_at,
            limit=self._max,
        )

# コード例 4: Sliding Window Counter 実装
class SlidingWindowCounterLimiter:
    """
    Sliding Window Counter (ハイブリッド方式)。
 
    Fixed Window の O(1) メモリ効率と、
    Sliding Window の精度を両立する。
 
    計算式:
    estimated_count = prev_window_count * overlap_ratio + current_window_count
    overlap_ratio = 1 - (current_time - current_window_start) / window_size
    """
 
    LUA_SCRIPT = """
    local curr_key = KEYS[1]
    local prev_key = KEYS[2]
    local max_requests = tonumber(ARGV[1])
    local window_size = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
 
    -- 現在と前のウィンドウのカウントを取得
    local curr_count = tonumber(redis.call('GET', curr_key) or '0')
    local prev_count = tonumber(redis.call('GET', prev_key) or '0')
 
    -- 現在のウィンドウ内の経過率
    local window_start = math.floor(now / window_size) * window_size
    local elapsed_ratio = (now - window_start) / window_size
 
    -- スライディングウィンドウの推定カウント
    local estimated = prev_count * (1 - elapsed_ratio) + curr_count
 
    if estimated < max_requests then
        -- 許可: 現在のウィンドウのカウントをインクリメント
        redis.call('INCR', curr_key)
        redis.call('EXPIRE', curr_key, window_size * 2)
        return {1, math.floor(max_requests - estimated - 1)}
    else
        return {0, 0}
    end
    """
 
    def __init__(self, redis_client, max_requests: int = 100,
                 window_seconds: int = 60):
        self._redis = redis_client
        self._max = max_requests
        self._window = window_seconds
        self._script = self._redis.register_script(self.LUA_SCRIPT)
 
    def allow_request(self, key: str) -> RateLimitResult:
        now = time.time()
        curr_window = int(now) // self._window
        prev_window = curr_window - 1
 
        curr_key = f"ratelimit:swc:{key}:{curr_window}"
        prev_key = f"ratelimit:swc:{key}:{prev_window}"
 
        result = self._script(
            keys=[curr_key, prev_key],
            args=[self._max, self._window, now]
        )
 
        allowed = bool(result[0])
        remaining = int(result[1])
 
        return RateLimitResult(
            allowed=allowed,
            remaining=remaining,
            reset_at=(curr_window + 1) * self._window if not allowed else 0,
            limit=self._max,
        )

# コード例 5: Leaky Bucket 実装
class LeakyBucketLimiter:
    """
    Leaky Bucket レートリミッター。
 
    リクエストを均等なレートで処理する。
    バーストを平滑化し、後続サービスへの負荷を均一にする。
    """
 
    LUA_SCRIPT = """
    local key = KEYS[1]
    local burst_size = tonumber(ARGV[1])   -- バケツ容量
    local leak_rate = tonumber(ARGV[2])     -- 処理レート (req/sec)
    local now = tonumber(ARGV[3])
 
    local data = redis.call('HMGET', key, 'water_level', 'last_leak')
    local water_level = tonumber(data[1]) or 0
    local last_leak = tonumber(data[2]) or now
 
    -- 漏れた水の計算 (時間経過分を減少)
    local elapsed = math.max(0, now - last_leak)
    local leaked = elapsed * leak_rate
    water_level = math.max(0, water_level - leaked)
 
    if water_level < burst_size then
        -- バケツに空きあり: リクエストを受け入れ
        water_level = water_level + 1
        redis.call('HMSET', key, 'water_level', water_level, 'last_leak', now)
        redis.call('EXPIRE', key, math.ceil(burst_size / leak_rate) * 2)
        return {1, math.floor(burst_size - water_level)}
    else
        -- バケツ満杯: 拒否
        local wait_time = (water_level - burst_size + 1) / leak_rate
        return {0, 0, math.ceil(wait_time)}
    end
    """
 
    def __init__(self, redis_client, burst_size: int = 10,
                 leak_rate: float = 1.0):
        self._redis = redis_client
        self._burst_size = burst_size
        self._leak_rate = leak_rate
        self._script = self._redis.register_script(self.LUA_SCRIPT)
 
    def allow_request(self, key: str) -> RateLimitResult:
        now = time.time()
        result = self._script(
            keys=[f"ratelimit:lb:{key}"],
            args=[self._burst_size, self._leak_rate, now]
        )
 
        allowed = bool(result[0])
        remaining = int(result[1])
        retry_after = float(result[2]) if len(result) > 2 else 0
 
        return RateLimitResult(
            allowed=allowed,
            remaining=remaining,
            reset_at=now + retry_after if not allowed else 0,
            limit=self._burst_size,
        )

# コード例 6: FastAPI ミドルウェアとしての統合
from fastapi import FastAPI, Request, Response, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import json
import time
 
app = FastAPI()
 
class RateLimitConfig:
    """エンドポイント別のレート制限設定"""
 
    # デフォルト設定
    DEFAULT = {"max_requests": 100, "window_seconds": 60}
 
    # エンドポイント別の設定
    ENDPOINT_RULES = {
        "/api/v1/search": {"max_requests": 30, "window_seconds": 60},
        "/api/v1/login": {"max_requests": 5, "window_seconds": 300},
        "/api/v1/signup": {"max_requests": 3, "window_seconds": 3600},
        "/api/v1/health": None,  # None = 制限なし
    }
 
    # プラン別の設定
    PLAN_RULES = {
        "free": {"multiplier": 1.0},
        "starter": {"multiplier": 5.0},
        "pro": {"multiplier": 20.0},
        "enterprise": {"multiplier": 100.0},
    }
 
class RateLimitMiddleware(BaseHTTPMiddleware):
    """
    FastAPI のレートリミッティングミドルウェア。
 
    キーの構築:
    1. API Key がある場合: API Key + エンドポイント
    2. 認証済みユーザー: User ID + エンドポイント
    3. 未認証: IP アドレス + エンドポイント
 
    レスポンスヘッダー:
    - 常に X-RateLimit-* ヘッダーを付与
    - 429 の場合は Retry-After も付与
    """
 
    def __init__(self, app, limiter: TokenBucketLimiter,
                 config: RateLimitConfig = None):
        super().__init__(app)
        self.limiter = limiter
        self.config = config or RateLimitConfig()
 
    async def dispatch(self, request: Request, call_next):
        # ヘルスチェック等の除外パス
        endpoint = request.url.path
        endpoint_rule = self.config.ENDPOINT_RULES.get(endpoint)
        if endpoint_rule is None and endpoint in self.config.ENDPOINT_RULES:
            # 明示的に None が設定されている = 制限なし
            return await call_next(request)
 
        # レートリミットキーの構築
        client_id = self._extract_client_id(request)
        key = f"{client_id}:{endpoint}"
 
        # 判定
        result = self.limiter.allow_request(key)
 
        if not result.allowed:
            retry_after = max(1, int(result.reset_at - time.time()))
            return Response(
                content=json.dumps({
                    "error": {
                        "code": "RATE_LIMIT_EXCEEDED",
                        "message": f"リクエスト制限を超過しました。{retry_after}秒後に再試行してください。",
                        "retry_after": retry_after,
                    }
                }),
                status_code=429,
                headers={
                    "Content-Type": "application/json",
                    "Retry-After": str(retry_after),
                    "X-RateLimit-Limit": str(result.limit),
                    "X-RateLimit-Remaining": "0",
                    "X-RateLimit-Reset": str(int(result.reset_at)),
                },
            )
 
        # 正常レスポンスにもヘッダーを付与
        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(result.limit)
        response.headers["X-RateLimit-Remaining"] = str(result.remaining)
        return response
 
    def _extract_client_id(self, request: Request) -> str:
        """リクエストからクライアント識別子を抽出"""
        # 優先順位: API Key > User ID > IP
        api_key = request.headers.get("X-API-Key")
        if api_key:
            return f"apikey:{api_key}"
 
        auth = request.headers.get("Authorization")
        if auth and auth.startswith("Bearer "):
            # JWT から user_id を抽出 (簡略化)
            user_id = extract_user_id_from_jwt(auth[7:])
            if user_id:
                return f"user:{user_id}"
 
        # フォールバック: IP アドレス
        client_ip = request.headers.get(
            "X-Forwarded-For", request.client.host
        )
        return f"ip:{client_ip}"
 
app.add_middleware(RateLimitMiddleware, limiter=limiter)

# コード例 7: デコレータベースのエンドポイント別レート制限
import functools
from fastapi import APIRouter, Request, HTTPException
 
router = APIRouter()
 
def rate_limit(max_requests: int, window_seconds: int,
               key_func=None):
    """
    エンドポイント固有のレート制限デコレータ。
 
    使い方:
    @rate_limit(max_requests=10, window_seconds=60)
    async def search(query: str):
        ...
 
    key_func で制限キーをカスタマイズ:
    @rate_limit(max_requests=5, window_seconds=300,
                key_func=lambda req: req.client.host)
    async def login(credentials):
        ...
    """
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, request: Request = None, **kwargs):
            if request is None:
                return await func(*args, **kwargs)
 
            # キーの構築
            if key_func:
                key = key_func(request)
            else:
                user_id = getattr(request.state, 'user_id', request.client.host)
                key = f"{user_id}:{request.url.path}"
 
            # 専用リミッターで判定
            limiter = SlidingWindowCounterLimiter(
                redis_client=get_redis(),
                max_requests=max_requests,
                window_seconds=window_seconds,
            )
 
            result = limiter.allow_request(key)
            if not result.allowed:
                raise HTTPException(
                    status_code=429,
                    detail={
                        "error": "Rate limit exceeded",
                        "retry_after": int(result.reset_at - time.time()),
                    },
                    headers={
                        "Retry-After": str(int(result.reset_at - time.time())),
                    },
                )
 
            return await func(*args, request=request, **kwargs)
        return wrapper
    return decorator
 
# 使用例
@router.get("/api/v1/search")
@rate_limit(max_requests=30, window_seconds=60)
async def search_products(query: str, request: Request):
    """検索 API: 1分あたり30リクエストに制限"""
    return await perform_search(query)
 
@router.post("/api/v1/auth/login")
@rate_limit(
    max_requests=5,
    window_seconds=300,
    key_func=lambda req: f"login:{req.client.host}",  # IP 別
)
async def login(credentials: dict, request: Request):
    """ログイン API: IP あたり5分で5回に制限 (ブルートフォース防止)"""
    return await authenticate(credentials)
 
@router.post("/api/v1/emails/send")
@rate_limit(max_requests=10, window_seconds=3600)
async def send_email(email_data: dict, request: Request):
    """メール送信 API: 1時間あたり10通に制限"""
    return await send_email_service(email_data)

# コード例 8: Redis 障害時のフォールバック戦略
import time
from threading import Lock
from collections import defaultdict
 
class ResilientRateLimiter:
    """
    Redis 障害に耐性のあるレートリミッター。
 
    フォールバック戦略:
    1. Primary: Redis Cluster (高精度)
    2. Secondary: ローカルメモリ (近似値)
    3. Tertiary: フェイルオープン (制限なし)
 
    WHY フェイルオープンを選ぶか:
    - レートリミッターの障害でサービス停止は本末転倒
    - 短時間のレート制限なしは許容可能
    - 代わりにアラートを発報し、迅速に復旧を目指す
    """
 
    def __init__(self, redis_limiter: TokenBucketLimiter,
                 fail_open: bool = True):
        self.redis_limiter = redis_limiter
        self.fail_open = fail_open
        self._local_counters = defaultdict(lambda: {"count": 0, "reset": 0})
        self._lock = Lock()
        self._redis_healthy = True
        self._last_health_check = 0
 
    def allow_request(self, key: str) -> RateLimitResult:
        # Redis の健全性確認 (10秒ごと)
        if not self._redis_healthy:
            if time.time() - self._last_health_check > 10:
                self._check_redis_health()
 
        # Primary: Redis
        if self._redis_healthy:
            try:
                result = self.redis_limiter.allow_request(key)
                return result
            except Exception as e:
                self._redis_healthy = False
                self._last_health_check = time.time()
                metrics.increment("rate_limiter.redis_error")
                logger.error(f"Redis rate limiter failed: {e}")
 
        # Secondary: ローカルメモリ (近似)
        try:
            return self._local_rate_limit(key)
        except Exception:
            pass
 
        # Tertiary: フェイルオープン/クローズ
        if self.fail_open:
            metrics.increment("rate_limiter.fail_open")
            return RateLimitResult(
                allowed=True, remaining=-1, reset_at=0, limit=0
            )
        else:
            metrics.increment("rate_limiter.fail_closed")
            return RateLimitResult(
                allowed=False, remaining=0, reset_at=time.time() + 60, limit=0
            )
 
    def _local_rate_limit(self, key: str) -> RateLimitResult:
        """ローカルメモリによる近似的なレート制限"""
        now = time.time()
        with self._lock:
            counter = self._local_counters[key]
 
            # ウィンドウリセット
            if now > counter["reset"]:
                counter["count"] = 0
                counter["reset"] = now + 60  # 1分ウィンドウ
 
            counter["count"] += 1
 
            # ★ ローカルなのでサーバー台数分の誤差がある
            # 例: 4台のサーバーで100/min 制限
            #     → 各サーバーで25/min に設定
            local_limit = 25  # max_requests / num_servers
            allowed = counter["count"] <= local_limit
 
            return RateLimitResult(
                allowed=allowed,
                remaining=max(0, local_limit - counter["count"]),
                reset_at=counter["reset"],
                limit=local_limit,
            )
 
    def _check_redis_health(self):
        """Redis の健全性チェック"""
        try:
            self.redis_limiter._redis.ping()
            self._redis_healthy = True
            logger.info("Redis rate limiter recovered")
        except Exception:
            self._redis_healthy = False
            self._last_health_check = time.time()

# コード例 9: 動的ルール管理サービス
import json
 
class RateLimitRuleService:
    """
    レート制限ルールを動的に管理するサービス。
 
    ルールの優先順位:
    1. ユーザー個別ルール (user:123 → 500/min)
    2. プラン別ルール (plan:pro → 1000/min)
    3. エンドポイント別ルール (/api/search → 30/min)
    4. デフォルトルール (100/min)
 
    WHY 動的ルールが必要か:
    - 特定ユーザーの一時的な制限緩和 (大量インポート等)
    - プラン変更の即時反映
    - インシデント対応での緊急制限
    - A/B テストでの制限値の実験
    """
 
    def __init__(self, redis_client, db):
        self.redis = redis_client
        self.db = db
        self._cache_ttl = 300  # 5分キャッシュ
 
    async def get_limit(self, client_id: str,
                         endpoint: str) -> dict:
        """適用されるレート制限ルールを取得"""
 
        # 1. キャッシュ確認
        cache_key = f"rate_rule:{client_id}:{endpoint}"
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
 
        # 2. ユーザー個別ルール
        user_rule = await self.db.fetch_one(
            "SELECT * FROM rate_limit_rules "
            "WHERE client_id = :cid AND endpoint = :ep AND is_active = TRUE",
            {"cid": client_id, "ep": endpoint},
        )
        if user_rule:
            rule = self._format_rule(user_rule)
            await self.redis.set(cache_key, json.dumps(rule), ex=self._cache_ttl)
            return rule
 
        # 3. プラン別ルール
        plan = await self._get_user_plan(client_id)
        plan_rule = await self.db.fetch_one(
            "SELECT * FROM rate_limit_rules "
            "WHERE client_id = :plan AND endpoint = :ep AND is_active = TRUE",
            {"plan": f"plan:{plan}", "ep": endpoint},
        )
        if plan_rule:
            rule = self._format_rule(plan_rule)
            await self.redis.set(cache_key, json.dumps(rule), ex=self._cache_ttl)
            return rule
 
        # 4. デフォルトルール
        default = {"max_requests": 100, "window_seconds": 60,
                    "algorithm": "token_bucket"}
        await self.redis.set(cache_key, json.dumps(default), ex=self._cache_ttl)
        return default
 
    async def set_temporary_override(self, client_id: str,
                                       endpoint: str,
                                       new_limit: dict,
                                       duration_hours: int = 24):
        """一時的なルールオーバーライド (インシデント対応等)"""
        await self.db.execute(
            "INSERT INTO rate_limit_overrides "
            "(client_id, endpoint, max_requests, window_seconds, expires_at) "
            "VALUES (:cid, :ep, :max, :win, NOW() + INTERVAL ':dur hours')",
            {
                "cid": client_id, "ep": endpoint,
                "max": new_limit["max_requests"],
                "win": new_limit["window_seconds"],
                "dur": duration_hours,
            },
        )
 
        # キャッシュを即時無効化
        cache_key = f"rate_rule:{client_id}:{endpoint}"
        await self.redis.delete(cache_key)
 
    def _format_rule(self, rule) -> dict:
        return {
            "max_requests": rule["max_requests"],
            "window_seconds": rule["window_seconds"],
            "algorithm": rule.get("algorithm", "token_bucket"),
        }

# NG: 各サーバーのメモリでカウント管理
class BadInMemoryLimiter:
    def __init__(self):
        self.counters = {}  # ローカルメモリ
 
    def allow(self, key):
        count = self.counters.get(key, 0)
        if count >= 100:
            return False
        self.counters[key] = count + 1
        return True
 
# 問題:
#   Server A: user-123 = 50 requests
#   Server B: user-123 = 50 requests
#   → 合計100リクエストだが各サーバーは50と認識
#   → 制限が全く効かない (N台のサーバーで N倍の制限)
 
# さらに:
#   - サーバー再起動でカウンターがリセットされる
#   - メモリリークのリスク (キーの自動削除がない)
 
 
# OK: Redis で一元管理
class GoodRedisLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
 
    def allow(self, key):
        # Redis で一元管理 → 全サーバーで共有
        result = self.redis.eval(LUA_SCRIPT, 1, key, ...)
        return bool(result[0])
 
# Redis クラスターの場合:
#   Server A --count--> [Redis Cluster] <--count-- Server B
#   → user-123 = 100 (正確な合計値)

# NG: 全ユーザー・全エンドポイントに同一制限
class BadSingleRuleLimiter:
    GLOBAL_LIMIT = 100  # req/min for everyone
 
    def allow(self, key):
        count = self.redis.incr(f"rate:{key}")
        return count <= self.GLOBAL_LIMIT
 
# 問題:
# 1. 無料ユーザーと有料ユーザーが同じ制限 → 有料の価値がない
# 2. /api/health (軽量) と /api/search (重量) が同じ制限
# 3. DDoS 防御と通常制限が区別できない
 
 
# OK: 多層・多次元のルール
class GoodMultiLayerLimiter:
    RULES = {
        # プラン別
        "plan:free":       {"default": 60, "search": 10},
        "plan:pro":        {"default": 1000, "search": 100},
        "plan:enterprise": {"default": 10000, "search": 1000},
 
        # エンドポイント別
        "endpoint:/api/search":  {"limit": 30, "window": 60},
        "endpoint:/api/health":  None,  # 制限なし
        "endpoint:/api/login":   {"limit": 5, "window": 300},
 
        # グローバル (DDoS 防御)
        "global:ip": {"limit": 1000, "window": 60},
    }
 
    def allow(self, client_id, endpoint, plan):
        # 1. グローバル IP 制限
        if not self._check_ip_limit(client_id):
            return False
        # 2. エンドポイント固有制限
        if not self._check_endpoint_limit(client_id, endpoint):
            return False
        # 3. プラン別制限
        if not self._check_plan_limit(client_id, plan, endpoint):
            return False
        return True

# NG: 429 を返すだけで、いつリトライすべきか教えない
@app.get("/api/data")
async def bad_endpoint(request: Request):
    if not rate_limiter.allow(get_client_id(request)):
        return Response(
            content="Too Many Requests",
            status_code=429,
        )
        # クライアントはいつリトライすればよいか不明
        # → 即座にリトライしてさらに 429 を受ける
        # → Thundering Herd 問題
 
 
# OK: 標準ヘッダーで制限状態を通知
@app.get("/api/data")
async def good_endpoint(request: Request):
    result = rate_limiter.allow_request(get_client_id(request))
 
    if not result.allowed:
        retry_after = max(1, int(result.reset_at - time.time()))
        return Response(
            content=json.dumps({
                "error": {
                    "code": "RATE_LIMIT_EXCEEDED",
                    "message": f"{retry_after}秒後に再試行してください",
                    "retry_after": retry_after,
                }
            }),
            status_code=429,
            headers={
                "Retry-After": str(retry_after),
                "X-RateLimit-Limit": str(result.limit),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(int(result.reset_at)),
            },
        )
 
    response = await process_request(request)
    response.headers["X-RateLimit-Remaining"] = str(result.remaining)
    return response

# NG: クライアント側でリトライ制御なし
class BadClient:
    async def call_api(self, url):
        for _ in range(100):  # 100回リトライ
            response = await http.get(url)
            if response.status == 200:
                return response
            # 429 でも即座にリトライ → サーバーに負荷
 
# OK: Exponential Backoff + Jitter + Retry-After 遵守
class GoodClient:
    async def call_api(self, url, max_retries=5):
        for attempt in range(max_retries):
            response = await http.get(url)
 
            if response.status == 200:
                return response
 
            if response.status == 429:
                # Retry-After ヘッダーを尊重
                retry_after = int(response.headers.get("Retry-After", 60))
 
                # Exponential Backoff + Jitter
                base_delay = min(retry_after, 2 ** attempt)
                jitter = random.uniform(0, base_delay * 0.5)
                wait_time = base_delay + jitter
 
                logger.warning(
                    f"Rate limited. Waiting {wait_time:.1f}s "
                    f"(attempt {attempt + 1}/{max_retries})"
                )
                await asyncio.sleep(wait_time)
 
            elif response.status >= 500:
                await asyncio.sleep(2 ** attempt)
            else:
                raise APIError(response.status, await response.text())
 
        raise MaxRetriesExceeded(f"Failed after {max_retries} attempts")

# 要件:
# 1. IP 単位: 1000 req/min (DDoS 防御)
# 2. ユーザー単位: 100 req/min (通常利用)
# 3. 未認証の場合は IP 制限のみ適用
# 4. 両方の制限を通過した場合のみリクエストを許可
# 5. レスポンスヘッダーにはより厳しい方の残り回数を表示
 
# スケルトンコード:
class TwoLayerRateLimiter:
    def __init__(self, redis_client):
        self.ip_limiter = FixedWindowLimiter(
            redis_client, max_requests=1000, window_seconds=60
        )
        self.user_limiter = TokenBucketLimiter(
            redis_client, max_tokens=100, refill_rate=1.67
        )
 
    def check(self, ip: str, user_id: str = None) -> RateLimitResult:
        # TODO: IP 制限チェック
        # TODO: user_id があればユーザー制限チェック
        # TODO: より厳しい方の結果を返す
        pass

# 要件:
# 1. Free プラン: 60 req/min, バースト不可
# 2. Starter プラン: 300 req/min, バースト 50
# 3. Pro プラン: 1000 req/min, バースト 200
# 4. Enterprise プラン: カスタム制限
# 5. プラン変更は即時反映
# 6. API Key から自動でプランを判定
 
# スケルトンコード:
class PlanBasedRateLimiter:
    PLAN_CONFIGS = {
        "free":       {"algorithm": "fixed_window", "max": 60, "window": 60},
        "starter":    {"algorithm": "token_bucket", "max": 300, "refill": 5.0},
        "pro":        {"algorithm": "token_bucket", "max": 1000, "refill": 16.7},
        "enterprise": None,  # DB から取得
    }
 
    async def check(self, api_key: str) -> RateLimitResult:
        # TODO: API Key からプランを取得
        # TODO: プランに対応するアルゴリズムを選択
        # TODO: レート制限を適用
        pass

# 要件:
# 1. 正常時: Redis Cluster でレート制限
# 2. Redis 障害時: ローカルメモリでフォールバック
# 3. フォールバック中のメトリクス収集
# 4. Redis 復旧の自動検知と切り戻し
# 5. 障害時のアラート発報
# 6. 障害耐性のテストシナリオ
 
# テストシナリオ:
class RateLimiterResiliencyTest:
    """
    テスト 1: Redis 正常動作
      → 100 req/min の制限が正確に動作すること
 
    テスト 2: Redis 障害発生
      → フォールバックに自動切替
      → ローカルメモリで近似的な制限が動作
      → alert が発報されること
 
    テスト 3: Redis 復旧
      → 10秒以内に Redis に切り戻しされること
      → ローカルカウンターがクリアされること
 
    テスト 4: ネットワーク分断 (Split Brain)
      → 各サーバーが独立してレート制限を維持
      → 全体としての精度は低下するが、サービスは継続
    """
 
    async def test_redis_failure_and_recovery(self):
        # TODO: Redis を停止
        # TODO: リクエストが通ることを確認 (フェイルオープン)
        # TODO: ローカルフォールバックの動作確認
        # TODO: Redis を復旧
        # TODO: Redis に切り戻されることを確認
        pass

# コード例 10: サーバー負荷に応じた動的レート制限
class AdaptiveRateLimiter:
    """
    サーバーの負荷状況に応じてレート制限を動的に調整する。
 
    WHY 適応型が必要か:
    - 固定制限だと、サーバーに余裕がある時にリクエストを不要に拒否
    - 負荷が高い時には固定制限では不十分な場合がある
    - トラフィックパターンは時間帯によって大きく変動する
 
    戦略:
    - CPU 使用率 > 80% → 制限を厳しくする (50% に削減)
    - CPU 使用率 > 90% → 最低限のみ許可 (20% に削減)
    - CPU 使用率 < 50% → 制限を緩和 (150% に拡大)
    """
 
    def __init__(self, base_limiter, health_checker):
        self.base_limiter = base_limiter
        self.health_checker = health_checker
        self._adjustment_factor = 1.0
 
    async def update_factor(self):
        """定期的に (10秒ごと) 調整係数を更新"""
        health = await self.health_checker.get_metrics()
 
        cpu_usage = health["cpu_percent"]
        if cpu_usage > 90:
            self._adjustment_factor = 0.2
        elif cpu_usage > 80:
            self._adjustment_factor = 0.5
        elif cpu_usage > 60:
            self._adjustment_factor = 0.8
        elif cpu_usage < 30:
            self._adjustment_factor = 1.5
        else:
            self._adjustment_factor = 1.0
 
        metrics.gauge("rate_limit.adjustment_factor", self._adjustment_factor)
 
    def get_effective_limit(self, base_limit: int) -> int:
        """調整後の制限値を返す"""
        return max(1, int(base_limit * self._adjustment_factor))