# --- ステートフル vs ステートレスサーバー ---
 
# ❌ ステートフル（スケールアウトが困難）
class StatefulServer:
    def __init__(self):
        self.sessions = {}  # セッションをサーバーメモリに保持
 
    def login(self, user_id, password):
        session_id = generate_session_id()
        self.sessions[session_id] = {
            "user_id": user_id,
            "login_time": datetime.now()
        }
        return session_id
 
    def get_user(self, session_id):
        # このサーバーにしかセッション情報がない!
        session = self.sessions.get(session_id)
        if not session:
            raise AuthenticationError("Invalid session")
        return session["user_id"]
 
# ✅ ステートレス（自由にスケールアウト可能）
class StatelessServer:
    def __init__(self, session_store):
        # セッションは外部ストア（Redis等）に保存
        self.session_store = session_store
 
    def login(self, user_id, password):
        session_id = generate_session_id()
        self.session_store.set(session_id, {
            "user_id": user_id,
            "login_time": datetime.now().isoformat()
        }, ttl=3600)  # 1時間で期限切れ
        return session_id
 
    def get_user(self, session_id):
        # どのサーバーからでもセッションを参照可能
        session = self.session_store.get(session_id)
        if not session:
            raise AuthenticationError("Invalid session")
        return session["user_id"]

import hashlib
from bisect import bisect_right
 
class ConsistentHash:
    """コンシステントハッシュの実装例"""
 
    def __init__(self, nodes=None, replicas=150):
        """
        Args:
            nodes: 初期ノードのリスト
            replicas: 各ノードの仮想ノード数（多いほど均一に分散）
        """
        self.replicas = replicas
        self.ring = {}       # ハッシュ値 → ノード名
        self.sorted_keys = []  # ソート済みハッシュ値リスト
 
        if nodes:
            for node in nodes:
                self.add_node(node)
 
    def _hash(self, key: str) -> int:
        """キーのハッシュ値を計算"""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
 
    def add_node(self, node: str):
        """ノードを追加"""
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_value = self._hash(virtual_key)
            self.ring[hash_value] = node
            self.sorted_keys.append(hash_value)
        self.sorted_keys.sort()
 
    def remove_node(self, node: str):
        """ノードを削除"""
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_value = self._hash(virtual_key)
            del self.ring[hash_value]
            self.sorted_keys.remove(hash_value)
 
    def get_node(self, key: str) -> str:
        """キーが属するノードを取得"""
        if not self.ring:
            raise ValueError("No nodes available")
 
        hash_value = self._hash(key)
        idx = bisect_right(self.sorted_keys, hash_value)
        if idx == len(self.sorted_keys):
            idx = 0  # リングを一周
        return self.ring[self.sorted_keys[idx]]
 
# 使用例
ch = ConsistentHash(["server-1", "server-2", "server-3"])
 
# データの割り当て
for key in ["user:1001", "user:1002", "user:1003", "order:5001"]:
    node = ch.get_node(key)
    print(f"{key} → {node}")
 
# ノード追加 → 影響を受けるキーは約 1/N のみ
ch.add_node("server-4")
 
# ノード削除 → 影響を受けるキーは約 1/N のみ
ch.remove_node("server-2")

# ヘルスチェックエンドポイントの実装例
from flask import Flask, jsonify
import psycopg2
import redis
 
app = Flask(__name__)
 
@app.route("/health")
def health_check():
    """簡易ヘルスチェック（サーバーが動いているか）"""
    return jsonify({"status": "ok"}), 200
 
@app.route("/health/detailed")
def detailed_health_check():
    """詳細ヘルスチェック（依存サービスの状態も確認）"""
    checks = {}
 
    # データベースの疎通確認
    try:
        conn = psycopg2.connect("postgresql://localhost/myapp")
        conn.execute("SELECT 1")
        conn.close()
        checks["database"] = {"status": "healthy"}
    except Exception as e:
        checks["database"] = {"status": "unhealthy", "error": str(e)}
 
    # Redisの疎通確認
    try:
        r = redis.Redis()
        r.ping()
        checks["cache"] = {"status": "healthy"}
    except Exception as e:
        checks["cache"] = {"status": "unhealthy", "error": str(e)}
 
    # ディスク容量の確認
    import shutil
    usage = shutil.disk_usage("/")
    free_percent = usage.free / usage.total * 100
    if free_percent > 10:
        checks["disk"] = {"status": "healthy", "free_percent": round(free_percent, 1)}
    else:
        checks["disk"] = {"status": "warning", "free_percent": round(free_percent, 1)}
 
    # 全体のステータス判定
    overall = "healthy"
    for check in checks.values():
        if check["status"] == "unhealthy":
            overall = "unhealthy"
            break
        if check["status"] == "warning":
            overall = "degraded"
 
    status_code = 200 if overall == "healthy" else 503
    return jsonify({"status": overall, "checks": checks}), status_code

import redis
import json
from datetime import timedelta
 
r = redis.Redis()
 
# --- 1. Cache-Aside（Lazy Loading） ---
# アプリケーションがキャッシュの読み書きを制御
# 最も一般的なパターン
 
def get_user_cache_aside(user_id: str) -> dict:
    """Cache-Aside: まずキャッシュを確認、なければDBから取得してキャッシュ"""
    cache_key = f"user:{user_id}"
 
    # 1. キャッシュを確認
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)  # キャッシュヒット
 
    # 2. キャッシュミス → DBから取得
    user = db.find_user(user_id)
    if user is None:
        return None
 
    # 3. キャッシュに保存（TTL付き）
    r.setex(cache_key, timedelta(hours=1), json.dumps(user.to_dict()))
 
    return user.to_dict()
 
def update_user_cache_aside(user_id: str, data: dict):
    """更新時はキャッシュを無効化"""
    db.update_user(user_id, data)
    r.delete(f"user:{user_id}")  # キャッシュを削除
 
 
# --- 2. Write-Through ---
# 書き込み時にキャッシュとDBを同時更新
 
def save_user_write_through(user_id: str, data: dict):
    """Write-Through: DBとキャッシュを同時に更新"""
    # DBに書き込み
    db.save_user(user_id, data)
 
    # キャッシュにも書き込み
    cache_key = f"user:{user_id}"
    r.setex(cache_key, timedelta(hours=1), json.dumps(data))
 
 
# --- 3. Write-Behind（Write-Back） ---
# まずキャッシュに書き込み、非同期でDBに反映
 
class WriteBehindCache:
    def __init__(self):
        self.write_queue = []
 
    def save(self, user_id: str, data: dict):
        """まずキャッシュに書き込み"""
        cache_key = f"user:{user_id}"
        r.setex(cache_key, timedelta(hours=1), json.dumps(data))
 
        # 書き込みキューに追加（非同期でDBに反映）
        self.write_queue.append(("user", user_id, data))
 
    def flush(self):
        """キューの内容をDBに一括書き込み"""
        while self.write_queue:
            entity_type, entity_id, data = self.write_queue.pop(0)
            db.save(entity_type, entity_id, data)
 
 
# --- 4. Read-Through ---
# キャッシュ自体がDBからの読み込みを管理
 
class ReadThroughCache:
    def __init__(self, loader):
        self.loader = loader  # データ取得関数
 
    def get(self, key: str) -> dict:
        cached = r.get(key)
        if cached:
            return json.loads(cached)
 
        # キャッシュが自動的にDBから取得
        data = self.loader(key)
        if data:
            r.setex(key, timedelta(hours=1), json.dumps(data))
        return data
 
# 使用例
user_cache = ReadThroughCache(loader=lambda key: db.find_user(key.split(":")[1]))
user = user_cache.get("user:1001")

# キャッシュスタンピード対策の実装例
import time
import threading
 
class StampedeProtectedCache:
    """キャッシュスタンピード対策付きキャッシュ"""
 
    def __init__(self):
        self.locks = {}  # キーごとのロック
        self.lock_manager = threading.Lock()
 
    def get_or_compute(self, key: str, compute_fn, ttl_seconds: int = 3600):
        """キャッシュ取得。ミス時はロックを取得して1リクエストのみDB問い合わせ"""
        # まずキャッシュを確認
        cached = r.get(key)
        if cached:
            return json.loads(cached)
 
        # ロックを取得（同一キーは1リクエストのみ）
        lock = self._get_lock(key)
        acquired = lock.acquire(timeout=5)
 
        if not acquired:
            # ロック取得失敗 → 少し待ってリトライ
            time.sleep(0.1)
            cached = r.get(key)
            return json.loads(cached) if cached else None
 
        try:
            # ダブルチェック（ロック取得中に別スレッドがキャッシュした可能性）
            cached = r.get(key)
            if cached:
                return json.loads(cached)
 
            # DBから取得してキャッシュ
            value = compute_fn()
            if value is not None:
                # TTLにジッターを追加（雪崩防止）
                import random
                jitter = random.randint(0, ttl_seconds // 10)
                r.setex(key, ttl_seconds + jitter, json.dumps(value))
            return value
        finally:
            lock.release()
 
    def _get_lock(self, key):
        with self.lock_manager:
            if key not in self.locks:
                self.locks[key] = threading.Lock()
            return self.locks[key]

-- インデックスの基本原則
 
-- 1. プライマリキーは自動的にインデックスが作成される
CREATE TABLE users (
    id BIGINT PRIMARY KEY,         -- 自動インデックス
    email VARCHAR(255) NOT NULL,
    name VARCHAR(100) NOT NULL,
    status VARCHAR(20) NOT NULL,
    created_at TIMESTAMP NOT NULL
);
 
-- 2. 検索条件に使うカラムにインデックスを作成
CREATE INDEX idx_users_email ON users (email);
CREATE INDEX idx_users_status ON users (status);
 
-- 3. 複合インデックス（カラムの順序が重要）
-- WHERE status = 'active' AND created_at > '2024-01-01' をサポート
CREATE INDEX idx_users_status_created ON users (status, created_at);
-- 左端のカラムから使われる（Leftmost Prefix Rule）
-- ✅ WHERE status = 'active' → 使われる
-- ✅ WHERE status = 'active' AND created_at > ... → 使われる
-- ❌ WHERE created_at > ... → 使われない（status が先頭）
 
-- 4. カバリングインデックス（必要な全カラムがインデックスに含まれる）
-- SELECT email, name FROM users WHERE status = 'active';
CREATE INDEX idx_users_covering ON users (status, email, name);
-- テーブルデータへのアクセス不要 → 非常に高速
 
-- 5. ユニークインデックス（一意制約）
CREATE UNIQUE INDEX idx_users_email_unique ON users (email);
 
-- インデックスを作りすぎない注意点:
-- - INSERT/UPDATE/DELETE が遅くなる（インデックスも更新が必要）
-- - ストレージを消費する
-- - 目安: テーブルあたり5-10個まで

# --- SQS + Pythonの例 ---
import boto3
import json
 
sqs = boto3.client("sqs", region_name="ap-northeast-1")
QUEUE_URL = "https://sqs.ap-northeast-1.amazonaws.com/123456789/my-queue"
 
# プロデューサー（メッセージ送信）
def send_email_task(to: str, subject: str, body: str):
    """メール送信タスクをキューに投入"""
    message = {
        "task": "send_email",
        "payload": {
            "to": to,
            "subject": subject,
            "body": body
        },
        "created_at": datetime.now().isoformat()
    }
    sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps(message),
        MessageGroupId="email-tasks"  # FIFOキューの場合
    )
 
# コンシューマー（メッセージ処理）
def process_messages():
    """キューからメッセージを取得して処理"""
    while True:
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20,  # ロングポーリング
            VisibilityTimeout=60  # 処理中は他のワーカーから見えない
        )
 
        messages = response.get("Messages", [])
        for msg in messages:
            try:
                task = json.loads(msg["Body"])
                handle_task(task)
 
                # 処理完了 → メッセージ削除
                sqs.delete_message(
                    QueueUrl=QUEUE_URL,
                    ReceiptHandle=msg["ReceiptHandle"]
                )
            except Exception as e:
                logger.error(f"Failed to process message: {e}")
                # 削除しない → VisibilityTimeout後に再処理される
 
def handle_task(task: dict):
    """タスクの種類に応じて処理"""
    if task["task"] == "send_email":
        payload = task["payload"]
        email_service.send(payload["to"], payload["subject"], payload["body"])
    elif task["task"] == "generate_report":
        report_service.generate(task["payload"]["report_id"])
    else:
        logger.warning(f"Unknown task type: {task['task']}")

# --- サービス間通信の実装例 ---
 
# 1. REST API 呼び出し（サーキットブレーカー付き）
import requests
from circuitbreaker import circuit
 
@circuit(failure_threshold=5, recovery_timeout=30)
def call_payment_service(order_id: str, amount: float) -> dict:
    """決済サービスを呼び出す（サーキットブレーカー付き）"""
    try:
        response = requests.post(
            "http://payment-service/api/v1/charge",
            json={"order_id": order_id, "amount": amount},
            timeout=5  # タイムアウト設定は必須
        )
        response.raise_for_status()
        return response.json()
    except requests.Timeout:
        raise PaymentServiceTimeout("Payment service timed out")
    except requests.ConnectionError:
        raise PaymentServiceUnavailable("Payment service is unavailable")
 
# 2. サガパターン（分散トランザクション）
class OrderSaga:
    """注文処理のサガ（各ステップに補償トランザクションを用意）"""
 
    def execute(self, order):
        try:
            # Step 1: 在庫を予約
            reservation = inventory_service.reserve(order.items)
 
            try:
                # Step 2: 決済処理
                payment = payment_service.charge(order.amount)
 
                try:
                    # Step 3: 配送手配
                    shipping = shipping_service.schedule(order)
                except Exception:
                    # Step 3 失敗 → Step 2 の補償（返金）
                    payment_service.refund(payment.transaction_id)
                    raise
            except Exception:
                # Step 2 失敗 → Step 1 の補償（在庫解放）
                inventory_service.cancel_reservation(reservation.id)
                raise
        except Exception as e:
            order.status = "failed"
            order.failure_reason = str(e)
            raise OrderFailedError(str(e)) from e
 
        order.status = "completed"
        return order

# --- 統一されたレスポンスフォーマット ---
 
# 成功レスポンス
{
    "status": "success",
    "data": {
        "id": "user-001",
        "name": "Alice",
        "email": "alice@example.com",
        "created_at": "2024-01-15T10:30:00Z"
    }
}
 
# 一覧レスポンス（ページネーション付き）
{
    "status": "success",
    "data": [
        {"id": "user-001", "name": "Alice"},
        {"id": "user-002", "name": "Bob"}
    ],
    "pagination": {
        "total": 150,
        "page": 1,
        "per_page": 20,
        "total_pages": 8,
        "next_cursor": "eyJpZCI6InVzZXItMDIwIn0="
    }
}
 
# エラーレスポンス
{
    "status": "error",
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "入力値に問題があります",
        "details": [
            {
                "field": "email",
                "message": "メールアドレスの形式が正しくありません"
            },
            {
                "field": "password",
                "message": "パスワードは8文字以上で入力してください"
            }
        ]
    }
}

# --- レート制限の実装パターン ---
 
# 1. 固定ウィンドウカウンター
import time
 
class FixedWindowRateLimiter:
    """固定ウィンドウ方式のレート制限"""
 
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
 
    def is_allowed(self, client_id: str) -> bool:
        current_window = int(time.time() / self.window_seconds)
        key = f"rate_limit:{client_id}:{current_window}"
 
        count = r.incr(key)
        if count == 1:
            r.expire(key, self.window_seconds)
 
        return count <= self.max_requests
 
# 2. スライディングウィンドウログ
class SlidingWindowLogRateLimiter:
    """スライディングウィンドウログ方式"""
 
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
 
    def is_allowed(self, client_id: str) -> bool:
        now = time.time()
        window_start = now - self.window_seconds
        key = f"rate_limit:{client_id}"
 
        # 古いエントリを削除
        r.zremrangebyscore(key, 0, window_start)
 
        # 現在のウィンドウ内のリクエスト数をカウント
        count = r.zcard(key)
 
        if count < self.max_requests:
            r.zadd(key, {str(now): now})
            r.expire(key, self.window_seconds)
            return True
 
        return False
 
# 3. トークンバケット
class TokenBucketRateLimiter:
    """トークンバケット方式"""
 
    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: バケットの最大トークン数
            refill_rate: 1秒あたりに補充されるトークン数
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
 
    def is_allowed(self, client_id: str, tokens_needed: int = 1) -> bool:
        key = f"token_bucket:{client_id}"
        now = time.time()
 
        # 現在のトークン数と最終更新時刻を取得
        data = r.hgetall(key)
        if not data:
            # 初回: バケットを満タンにして1トークン消費
            r.hset(key, mapping={
                "tokens": str(self.capacity - tokens_needed),
                "last_refill": str(now)
            })
            r.expire(key, 3600)
            return True
 
        current_tokens = float(data[b"tokens"])
        last_refill = float(data[b"last_refill"])
 
        # 時間経過分のトークンを補充
        elapsed = now - last_refill
        new_tokens = min(
            self.capacity,
            current_tokens + elapsed * self.refill_rate
        )
 
        if new_tokens >= tokens_needed:
            r.hset(key, mapping={
                "tokens": str(new_tokens - tokens_needed),
                "last_refill": str(now)
            })
            return True
 
        return False
 
# 使い方
limiter = TokenBucketRateLimiter(capacity=100, refill_rate=10)  # 100リクエスト/バースト、10リクエスト/秒
if limiter.is_allowed("user:123"):
    process_request()
else:
    return {"error": "Rate limit exceeded"}, 429

# --- リトライ with 指数バックオフ ---
import time
import random
 
def retry_with_exponential_backoff(
    func,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True
):
    """指数バックオフ付きリトライ"""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries:
                raise  # 最後のリトライでも失敗 → 例外を上げる
 
            delay = min(base_delay * (2 ** attempt), max_delay)
            if jitter:
                delay = delay * (0.5 + random.random())  # 0.5〜1.5倍のジッター
 
            logger.warning(
                f"Attempt {attempt + 1}/{max_retries} failed: {e}. "
                f"Retrying in {delay:.1f}s..."
            )
            time.sleep(delay)
 
# 使用例
result = retry_with_exponential_backoff(
    lambda: external_api.fetch_data("user-001"),
    max_retries=3,
    base_delay=1.0
)
 
 
# --- サーキットブレーカーの実装 ---
import time
import threading
from enum import Enum
 
class CircuitState(Enum):
    CLOSED = "closed"        # 正常: リクエストを通す
    OPEN = "open"            # 遮断: リクエストを即座に拒否
    HALF_OPEN = "half_open"  # 試行: 1リクエストだけ通す
 
class CircuitBreaker:
    """サーキットブレーカーの実装"""
 
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        success_threshold: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.lock = threading.Lock()
 
    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if self._should_try_reset():
                    self.state = CircuitState.HALF_OPEN
                else:
                    raise CircuitBreakerOpenError(
                        "Circuit breaker is OPEN. Request rejected."
                    )
 
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
 
    def _on_success(self):
        with self.lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    self.success_count = 0
            else:
                self.failure_count = 0
 
    def _on_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
            elif self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
 
    def _should_try_reset(self):
        if self.last_failure_time is None:
            return True
        return time.time() - self.last_failure_time >= self.recovery_timeout