"""CDN の DNS 解決フローの解説"""
 
# 1. ユーザーが cdn.example.com にアクセス
# 2. DNS 解決の流れ:
 
#   cdn.example.com
#     → CNAME: d123.cloudfront.net
#       → Anycast IP: 13.224.x.x (最寄り POP の IP)
 
# GeoDNS を使う場合 (Cloudflare):
#   cdn.example.com
#     → Anycast IP: 104.16.x.x
#     → BGP ルーティングで最寄り POP に到達
 
# 3. DNS TTL の設計
dns_config = {
    "cdn.example.com": {
        "type": "CNAME",
        "value": "d123456.cloudfront.net",
        "ttl": 300,  # 5分: CDN 切り替えに対応できる短さ
    },
    # A/AAAA レコード (Cloudflare Anycast)
    "api.example.com": {
        "type": "A",
        "value": "104.16.132.229",  # Anycast IP
        "ttl": 300,
        "proxied": True,  # Cloudflare プロキシ有効
    }
}

# Nginx での Cache-Control 設定例
 
# ハッシュ付き静的アセット（CSS/JS）: 1年キャッシュ + immutable
location ~* \.(?:css|js)$ {
    # ファイル名にハッシュ: main.a1b2c3.js
    add_header Cache-Control "public, max-age=31536000, immutable";
    add_header X-Content-Type-Options "nosniff";
    # Brotli / Gzip 圧縮 (事前圧縮ファイルがある場合)
    gzip_static on;
    brotli_static on;
}
 
# 画像: 30日キャッシュ + WebP/AVIF 自動変換
location ~* \.(?:jpg|jpeg|png|gif|webp|avif|svg)$ {
    add_header Cache-Control "public, max-age=2592000";
    add_header Vary "Accept";  # Accept ヘッダーでコンテンツネゴシエーション
}
 
# フォント: 1年キャッシュ + CORS
location ~* \.(?:woff2?|ttf|otf|eot)$ {
    add_header Cache-Control "public, max-age=31536000, immutable";
    add_header Access-Control-Allow-Origin "*";
}
 
# HTML: CDN 60秒、ブラウザはキャッシュなし
location ~* \.html$ {
    add_header Cache-Control "public, s-maxage=60, max-age=0, must-revalidate";
    # オリジン障害時は5分間古いキャッシュを返す
    add_header Cache-Control "stale-if-error=300" always;
}
 
# API レスポンス: CDN 10秒 + stale-while-revalidate
location /api/ {
    add_header Cache-Control "public, s-maxage=10, stale-while-revalidate=60, stale-if-error=300";
    add_header Vary "Accept-Encoding, Authorization";
}
 
# 個人情報: キャッシュ禁止
location /api/user/profile {
    add_header Cache-Control "private, no-store";
    add_header Pragma "no-cache";  # HTTP/1.0 互換
}

"""ETag を使った条件付きリクエストの実装"""
import hashlib
from fastapi import FastAPI, Request, Response
 
app = FastAPI()
 
def generate_etag(content: bytes) -> str:
    """コンテンツから ETag を生成"""
    return f'"{hashlib.sha256(content).hexdigest()[:16]}"'
 
@app.get("/api/products/{product_id}")
async def get_product(product_id: str, request: Request):
    product = await fetch_product(product_id)
    content = json.dumps(product).encode()
    etag = generate_etag(content)
 
    # クライアントの If-None-Match をチェック
    if_none_match = request.headers.get("If-None-Match")
    if if_none_match == etag:
        # コンテンツ未変更 → 304 を返す (帯域節約)
        return Response(status_code=304, headers={
            "ETag": etag,
            "Cache-Control": "public, s-maxage=30, stale-while-revalidate=60",
        })
 
    return Response(
        content=content,
        media_type="application/json",
        headers={
            "ETag": etag,
            "Cache-Control": "public, s-maxage=30, stale-while-revalidate=60",
            "Vary": "Accept-Encoding",
        }
    )
 
# リクエスト・レスポンスフロー:
#
# 1回目: GET /api/products/123
#   → 200 OK + ETag: "a1b2c3d4e5f6g7h8"
#
# 2回目: GET /api/products/123
#        If-None-Match: "a1b2c3d4e5f6g7h8"
#   → 304 Not Modified (ボディなし、帯域を節約)
#
# CDN の動作:
#   CDN キャッシュ期限切れ → オリジンに条件付きリクエスト
#   → 304 → CDN キャッシュを更新して TTL リセット
#   → オリジンのデータ転送量を大幅削減

# CloudFront ディストリビューション作成 (boto3)
import boto3
 
cf = boto3.client('cloudfront')
 
distribution_config = {
    'CallerReference': 'my-app-2026',
    'Origins': {
        'Quantity': 2,
        'Items': [
            {
                # S3 オリジン (静的アセット)
                'Id': 'S3-static-assets',
                'DomainName': 'my-static-assets.s3.amazonaws.com',
                'S3OriginConfig': {
                    'OriginAccessIdentity':
                        'origin-access-identity/cloudfront/XXXXXXX'
                },
                'OriginShield': {
                    'Enabled': True,
                    'OriginShieldRegion': 'ap-northeast-1',  # 東京
                },
            },
            {
                # ALB オリジン (動的 API)
                'Id': 'ALB-api',
                'DomainName': 'api-internal.example.com',
                'CustomOriginConfig': {
                    'HTTPPort': 80,
                    'HTTPSPort': 443,
                    'OriginProtocolPolicy': 'https-only',
                    'OriginSslProtocols': {
                        'Quantity': 1, 'Items': ['TLSv1.2']
                    },
                    'OriginKeepaliveTimeout': 60,  # Keep-Alive 秒数
                    'OriginReadTimeout': 30,
                },
            },
        ]
    },
    'DefaultCacheBehavior': {
        'TargetOriginId': 'S3-static-assets',
        'ViewerProtocolPolicy': 'redirect-to-https',
        'CachePolicyId': '658327ea-f89d-4fab-a63d-7e88639e58f6',
        'Compress': True,              # Brotli/Gzip 自動圧縮
        'AllowedMethods': {'Quantity': 2, 'Items': ['GET', 'HEAD']},
    },
    'CacheBehaviors': {
        'Quantity': 1,
        'Items': [{
            'PathPattern': '/api/*',
            'TargetOriginId': 'ALB-api',
            'ViewerProtocolPolicy': 'https-only',
            'AllowedMethods': {
                'Quantity': 7,
                'Items': ['GET','HEAD','OPTIONS','PUT','POST','PATCH','DELETE'],
            },
            'CachePolicyId': '4135ea2d-6df8-44a3-9df3-4b5a84be39ad',
            'OriginRequestPolicyId':
                '216adef6-5c7f-47e4-b989-5492eafa07d3',  # AllViewer
            'ForwardedValues': {
                'QueryString': True,
                'Headers': {'Quantity': 3, 'Items': [
                    'Authorization', 'Accept', 'Content-Type'
                ]},
                'Cookies': {'Forward': 'none'},
            },
        }],
    },
    'Enabled': True,
    'PriceClass': 'PriceClass_200',    # 北米・欧州・アジア
    'HttpVersion': 'http2and3',        # HTTP/2 + HTTP/3 (QUIC) 有効化
    'Comment': 'Production distribution with S3 + ALB origins',
}
 
response = cf.create_distribution(DistributionConfig=distribution_config)
print(f"Distribution ID: {response['Distribution']['Id']}")
print(f"Domain: {response['Distribution']['DomainName']}")

// CloudFront Functions: URL リライト + セキュリティヘッダー追加
// 実行環境: JavaScript (ES 5.1), 最大実行時間: 1ms, 最大メモリ: 2MB
 
function handler(event) {
    var request = event.request;
    var uri = request.uri;
 
    // 1. SPA のフォールバック: /app/* → /app/index.html
    if (uri.startsWith('/app/') && !uri.includes('.')) {
        request.uri = '/app/index.html';
    }
 
    // 2. 拡張子がない場合は .html を追加
    if (!uri.includes('.') && uri !== '/') {
        request.uri = uri + '/index.html';
    }
 
    // 3. セキュリティヘッダーの追加 (レスポンスイベントの場合)
    if (event.response) {
        var response = event.response;
        response.headers['strict-transport-security'] = {
            value: 'max-age=63072000; includeSubDomains; preload'
        };
        response.headers['x-content-type-options'] = {
            value: 'nosniff'
        };
        response.headers['x-frame-options'] = {
            value: 'DENY'
        };
        response.headers['x-xss-protection'] = {
            value: '1; mode=block'
        };
        response.headers['content-security-policy'] = {
            value: "default-src 'self'; script-src 'self' 'unsafe-inline'; " +
                   "style-src 'self' 'unsafe-inline'; img-src 'self' data: https:;"
        };
        return response;
    }
 
    return request;
}

# Lambda@Edge: A/B テスト + 画像最適化ルーティング
# 実行環境: Node.js / Python, 最大実行時間: 5s (viewer) / 30s (origin)
 
import json
import hashlib
 
def viewer_request_handler(event, context):
    """ビューワーリクエスト: A/B テスト振り分け"""
    request = event['Records'][0]['cf']['request']
    headers = request['headers']
 
    # Cookie から A/B テストグループを判定
    cookies = headers.get('cookie', [{}])[0].get('value', '')
    ab_group = None
 
    for cookie in cookies.split(';'):
        cookie = cookie.strip()
        if cookie.startswith('ab_group='):
            ab_group = cookie.split('=')[1]
            break
 
    # 未割り当てならハッシュベースで振り分け
    if not ab_group:
        client_ip = event['Records'][0]['cf']['request']['clientIp']
        hash_val = int(hashlib.md5(client_ip.encode()).hexdigest(), 16)
        ab_group = 'A' if hash_val % 100 < 50 else 'B'
 
    # カスタムヘッダーでオリジンに伝達
    request['headers']['x-ab-group'] = [{'key': 'X-AB-Group', 'value': ab_group}]
 
    return request
 
def origin_response_handler(event, context):
    """オリジンレスポンス: A/B グループ Cookie を設定"""
    response = event['Records'][0]['cf']['response']
    request = event['Records'][0]['cf']['request']
 
    ab_group = request['headers'].get('x-ab-group', [{}])[0].get('value', 'A')
 
    # Set-Cookie で A/B グループを永続化
    response['headers']['set-cookie'] = [{
        'key': 'Set-Cookie',
        'value': f'ab_group={ab_group}; Path=/; Max-Age=604800; SameSite=Lax'
    }]
 
    return response

# CloudFront キャッシュ無効化（パージ）
# 注意: 1,000パス/月まで無料、超過は $0.005/パス
 
# 特定パスのパージ
aws cloudfront create-invalidation \
  --distribution-id E1234567890 \
  --paths "/index.html" "/api/*"
 
# 全キャッシュのパージ (緊急時のみ)
aws cloudfront create-invalidation \
  --distribution-id E1234567890 \
  --paths "/*"
 
# パージ状況の確認
aws cloudfront get-invalidation \
  --distribution-id E1234567890 \
  --id I1234567890

# Python でのキャッシュバスティング実装
import boto3
import hashlib
import json
from datetime import datetime
 
class CDNInvalidator:
    """CloudFront キャッシュ無効化マネージャー"""
 
    def __init__(self, distribution_id: str):
        self.cf = boto3.client('cloudfront')
        self.distribution_id = distribution_id
 
    def invalidate_paths(self, paths: list[str]) -> str:
        """指定パスのキャッシュを無効化"""
        caller_ref = f"inv-{datetime.now().strftime('%Y%m%d%H%M%S')}"
 
        response = self.cf.create_invalidation(
            DistributionId=self.distribution_id,
            InvalidationBatch={
                'Paths': {
                    'Quantity': len(paths),
                    'Items': paths,
                },
                'CallerReference': caller_ref,
            }
        )
        invalidation_id = response['Invalidation']['Id']
        print(f"Invalidation created: {invalidation_id}")
        print(f"Status: {response['Invalidation']['Status']}")
        return invalidation_id
 
    def wait_for_invalidation(self, invalidation_id: str):
        """無効化完了まで待機"""
        waiter = self.cf.get_waiter('invalidation_completed')
        print(f"Waiting for invalidation {invalidation_id}...")
        waiter.wait(
            DistributionId=self.distribution_id,
            Id=invalidation_id,
            WaiterConfig={'Delay': 10, 'MaxAttempts': 30}
        )
        print("Invalidation completed.")
 
    def deploy_with_cache_busting(
        self, s3_bucket: str, local_dir: str, html_files: list[str]
    ):
        """
        キャッシュバスティングを含むデプロイ:
        1. ハッシュ付きアセットをアップロード (パージ不要)
        2. HTML ファイルをアップロード (短 TTL)
        3. HTML のみパージ
        """
        s3 = boto3.client('s3')
        html_paths = []
 
        for html_file in html_files:
            s3.upload_file(
                Filename=f"{local_dir}/{html_file}",
                Bucket=s3_bucket,
                Key=html_file,
                ExtraArgs={
                    'ContentType': 'text/html',
                    'CacheControl': 'public, s-maxage=60, max-age=0, must-revalidate',
                }
            )
            html_paths.append(f"/{html_file}")
 
        # HTML のみパージ (ハッシュ付きアセットはパージ不要)
        if html_paths:
            inv_id = self.invalidate_paths(html_paths)
            self.wait_for_invalidation(inv_id)
 
# 使用例
invalidator = CDNInvalidator("E1234567890")
 
# デプロイ
invalidator.deploy_with_cache_busting(
    s3_bucket="my-static-assets",
    local_dir="./dist",
    html_files=["index.html", "about/index.html"]
)

// Cloudflare Workers: エッジでの API レスポンスキャッシュ
// V8 isolate ベース: 起動時間 < 1ms, CPU 50ms/リクエスト
 
export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    const cache = caches.default;
 
    // GET リクエストのみキャッシュ
    if (request.method !== 'GET') {
      return fetch(request);
    }
 
    // 1. キャッシュチェック
    const cacheKey = new Request(url.toString(), request);
    let response = await cache.match(cacheKey);
 
    if (response) {
      // Cache HIT: ヘッダーを追加して返却
      const newResponse = new Response(response.body, response);
      newResponse.headers.set('X-Cache-Status', 'HIT');
      return newResponse;
    }
 
    // 2. オリジンからフェッチ
    response = await fetch(request);
 
    // 3. 成功レスポンスのみキャッシュ
    if (response.ok) {
      const cacheResponse = new Response(response.body, response);
      cacheResponse.headers.set(
        'Cache-Control', 'public, s-maxage=300, stale-while-revalidate=600'
      );
      cacheResponse.headers.set('X-Cache-Status', 'MISS');
 
      // 非同期でキャッシュ書き込み（レスポンスを遅延させない）
      const ctx = { waitUntil: (p) => p };
      ctx.waitUntil(cache.put(cacheKey, cacheResponse.clone()));
 
      return cacheResponse;
    }
 
    return response;
  }
};

// Cloudflare Workers: 地理情報に基づくオリジン選択
export default {
  async fetch(request, env) {
    const cf = request.cf;  // Cloudflare のリクエストメタデータ
 
    // 地理情報の取得
    const country = cf.country;       // "JP"
    const continent = cf.continent;   // "AS"
    const city = cf.city;             // "Tokyo"
    const latitude = cf.latitude;
    const longitude = cf.longitude;
    const asn = cf.asn;               // ISP の AS 番号
 
    // 大陸ごとに最寄りのオリジンサーバーを選択
    const origins = {
      'AS': 'https://api-ap.example.com',    // アジア
      'NA': 'https://api-us.example.com',    // 北米
      'EU': 'https://api-eu.example.com',    // ヨーロッパ
      'SA': 'https://api-us.example.com',    // 南米 → 北米にフォールバック
      'AF': 'https://api-eu.example.com',    // アフリカ → 欧州にフォールバック
      'OC': 'https://api-ap.example.com',    // オセアニア → アジアにフォールバック
    };
 
    const originUrl = origins[continent] || origins['NA'];
 
    // オリジンにリクエスト転送
    const url = new URL(request.url);
    const originRequest = new Request(
      `${originUrl}${url.pathname}${url.search}`,
      request
    );
 
    const response = await fetch(originRequest);
 
    // レスポンスにルーティング情報を追加 (デバッグ用)
    const newResponse = new Response(response.body, response);
    newResponse.headers.set('X-Origin-Region', continent);
    newResponse.headers.set('X-Client-Country', country);
 
    return newResponse;
  }
};

// Workers KV を使ったエッジでの設定管理
// KV: 結果整合性、読み取り最適化 (60秒以内に伝播)
 
export default {
  async fetch(request, env) {
    const url = new URL(request.url);
 
    // 機能フラグの取得 (エッジで即座にレスポンス)
    if (url.pathname === '/api/feature-flags') {
      const flags = await env.FEATURE_FLAGS.get('current', { type: 'json' });
      return new Response(JSON.stringify(flags), {
        headers: {
          'Content-Type': 'application/json',
          'Cache-Control': 'public, max-age=30',
        }
      });
    }
 
    // レートリミット (エッジで実装)
    if (url.pathname.startsWith('/api/')) {
      const clientIp = request.headers.get('CF-Connecting-IP');
      const rateLimitKey = `ratelimit:${clientIp}`;
 
      // 現在のリクエスト数を取得
      const current = parseInt(await env.RATE_LIMITS.get(rateLimitKey) || '0');
 
      if (current >= 100) {  // 100 req/min
        return new Response('Rate limit exceeded', {
          status: 429,
          headers: {
            'Retry-After': '60',
            'X-RateLimit-Limit': '100',
            'X-RateLimit-Remaining': '0',
          }
        });
      }
 
      // カウントを更新 (TTL: 60秒)
      await env.RATE_LIMITS.put(rateLimitKey, String(current + 1), {
        expirationTtl: 60
      });
    }
 
    return fetch(request);
  }
};

"""キャッシュキー設計のベストプラクティス"""
from urllib.parse import urlparse, parse_qs, urlencode
 
class CacheKeyBuilder:
    """CDN キャッシュキーの設計と正規化"""
 
    # キャッシュに影響しないパラメータ (除外対象)
    EXCLUDED_PARAMS = {
        'utm_source', 'utm_medium', 'utm_campaign', 'utm_term',
        'utm_content',    # UTM トラッキングパラメータ
        'fbclid',         # Facebook クリック ID
        'gclid',          # Google クリック ID
        '_ga',            # Google Analytics
        'ref', 'source',  # リファラルパラメータ
        '_t', 'timestamp', 'nocache',  # キャッシュバスティング
    }
 
    # キャッシュキーに含めるパラメータ (ホワイトリスト方式)
    INCLUDED_PARAMS = {
        '/api/products': {'page', 'sort', 'category', 'limit'},
        '/api/search': {'q', 'page', 'sort', 'filters'},
    }
 
    @classmethod
    def normalize_cache_key(cls, url: str) -> str:
        """URL を正規化してキャッシュキーを生成"""
        parsed = urlparse(url)
        params = parse_qs(parsed.query, keep_blank_values=True)
 
        # 方法1: 除外リスト方式
        filtered_params = {
            k: v for k, v in params.items()
            if k.lower() not in cls.EXCLUDED_PARAMS
        }
 
        # パラメータをソートして正規化
        sorted_params = urlencode(
            sorted(filtered_params.items()), doseq=True
        )
 
        return f"{parsed.scheme}://{parsed.netloc}{parsed.path}" + \
               (f"?{sorted_params}" if sorted_params else "")
 
    @classmethod
    def normalize_with_whitelist(cls, url: str, path: str) -> str:
        """ホワイトリスト方式でキャッシュキーを生成"""
        parsed = urlparse(url)
        params = parse_qs(parsed.query, keep_blank_values=True)
 
        allowed = cls.INCLUDED_PARAMS.get(path, set())
        filtered_params = {
            k: v for k, v in params.items() if k in allowed
        }
 
        sorted_params = urlencode(
            sorted(filtered_params.items()), doseq=True
        )
        return f"{parsed.path}?{sorted_params}" if sorted_params else parsed.path
 
# テスト
urls = [
    "https://example.com/api/products?page=1&sort=price&utm_source=google",
    "https://example.com/api/products?utm_source=twitter&page=1&sort=price",
    "https://example.com/api/products?page=1&sort=price&fbclid=abc123",
]
 
for url in urls:
    normalized = CacheKeyBuilder.normalize_cache_key(url)
    print(f"Original: {url}")
    print(f"Cache Key: {normalized}")
    print()
 
# 出力:
# 3つのURLが全て同じキャッシュキーに正規化される:
# https://example.com/api/products?page=1&sort=price

"""Vary ヘッダーによるキャッシュバリエーション管理"""
from fastapi import FastAPI, Request, Response
 
app = FastAPI()
 
@app.get("/api/products")
async def get_products(request: Request):
    """デバイス・言語に応じたレスポンスバリエーション"""
    # Accept-Language からロケール判定
    lang = request.headers.get("Accept-Language", "en")
    locale = "ja" if "ja" in lang else "en"
 
    # User-Agent からデバイス判定
    ua = request.headers.get("User-Agent", "")
    is_mobile = "Mobile" in ua
 
    products = await fetch_products(locale=locale, mobile=is_mobile)
 
    return Response(
        content=json.dumps(products),
        headers={
            "Content-Type": "application/json",
            "Cache-Control": "public, s-maxage=300",
            # Vary: CDN に「これらのヘッダーが異なれば別キャッシュ」と指示
            "Vary": "Accept-Language, Accept-Encoding",
            # 注意: Vary: User-Agent は絶対に使わない！
            # → 無数のバリエーションが生成されキャッシュが効かなくなる
            # 代わりにCDN のデバイス検出機能を利用
        }
    )
 
# Vary ヘッダーの効果:
#
# Vary: Accept-Encoding の場合:
#   GET /api/products (Accept-Encoding: gzip)  → キャッシュ A (gzip版)
#   GET /api/products (Accept-Encoding: br)    → キャッシュ B (brotli版)
#   GET /api/products (Accept-Encoding: なし)   → キャッシュ C (非圧縮版)
#
# Vary: Accept-Language の場合:
#   GET /api/products (Accept-Language: ja)     → キャッシュ D (日本語版)
#   GET /api/products (Accept-Language: en)     → キャッシュ E (英語版)

"""Fastly の Surrogate-Key を使ったタグベースキャッシュパージ"""
from fastapi import FastAPI, Response
 
app = FastAPI()
 
@app.get("/api/products/{product_id}")
async def get_product(product_id: str):
    product = await fetch_product(product_id)
 
    return Response(
        content=json.dumps(product),
        headers={
            "Content-Type": "application/json",
            "Cache-Control": "public, s-maxage=3600",
            # Surrogate-Key: 複数のタグを設定
            # → タグ単位でパージ可能
            "Surrogate-Key": f"product-{product_id} "
                             f"category-{product['category']} "
                             f"all-products",
        }
    )
 
@app.get("/api/categories/{category_id}")
async def get_category(category_id: str):
    products = await fetch_products_by_category(category_id)
 
    return Response(
        content=json.dumps(products),
        headers={
            "Content-Type": "application/json",
            "Cache-Control": "public, s-maxage=3600",
            "Surrogate-Key": f"category-{category_id} all-products",
        }
    )
 
# パージの例:
# 1. 特定商品を更新した場合:
#    POST /service/{id}/purge/product-123
#    → product-123 のキャッシュのみパージ
#
# 2. カテゴリ全体を更新した場合:
#    POST /service/{id}/purge/category-electronics
#    → electronics カテゴリの全商品 + カテゴリ一覧をパージ
#
# 3. 全商品をパージ:
#    POST /service/{id}/purge/all-products
#
# メリット:
# - URL ベースのパージより柔軟
# - ワイルドカード不要で正確なパージが可能
# - 関連コンテンツを一括パージ

"""CDN エッジでの画像最適化パイプライン"""
from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class ImageFormat(Enum):
    WEBP = "webp"
    AVIF = "avif"
    JPEG = "jpeg"
    PNG = "png"
 
@dataclass
class ImageTransform:
    width: Optional[int] = None
    height: Optional[int] = None
    quality: int = 80
    format: ImageFormat = ImageFormat.WEBP
    fit: str = "cover"  # cover, contain, fill
 
class ImageOptimizationCDN:
    """
    CDN エッジでの画像最適化設計
 
    URL パターン: /images/{transforms}/{original_path}
    例: /images/w_800,h_600,q_80,f_webp/photos/hero.jpg
    """
 
    # レスポンシブ画像のプリセット
    PRESETS = {
        'thumbnail': ImageTransform(width=150, height=150, quality=60),
        'card':      ImageTransform(width=400, height=300, quality=75),
        'hero':      ImageTransform(width=1920, height=1080, quality=80),
        'og':        ImageTransform(width=1200, height=630, quality=85),
    }
 
    @staticmethod
    def generate_srcset(image_path: str, widths: list[int]) -> str:
        """レスポンシブ画像の srcset を生成"""
        srcset_entries = []
        for w in widths:
            url = f"/images/w_{w},f_auto/{image_path}"
            srcset_entries.append(f"{url} {w}w")
        return ",\n  ".join(srcset_entries)
 
    @staticmethod
    def generate_picture_tag(image_path: str) -> str:
        """<picture> タグを生成 (AVIF > WebP > JPEG フォールバック)"""
        return f"""<picture>
  <!-- AVIF (最高圧縮率) -->
  <source
    type="image/avif"
    srcset="{ImageOptimizationCDN.generate_srcset(image_path, [400, 800, 1200])}"
    sizes="(max-width: 768px) 100vw, 50vw" />
  <!-- WebP (広いブラウザ対応) -->
  <source
    type="image/webp"
    srcset="{ImageOptimizationCDN.generate_srcset(image_path, [400, 800, 1200])}"
    sizes="(max-width: 768px) 100vw, 50vw" />
  <!-- JPEG フォールバック -->
  <img
    src="/images/w_800,f_jpeg/{image_path}"
    loading="lazy"
    decoding="async"
    alt="" />
</picture>"""
 
# Cloudflare Workers での画像最適化リクエスト例:
#
# export default {
#   async fetch(request) {
#     const url = new URL(request.url);
#     const accept = request.headers.get('Accept') || '';
#
#     // ブラウザの対応フォーマットを判定
#     let format = 'jpeg';
#     if (accept.includes('image/avif')) format = 'avif';
#     else if (accept.includes('image/webp')) format = 'webp';
#
#     // Cloudflare Image Resizing を利用
#     return fetch(url.toString(), {
#       cf: {
#         image: {
#           width: 800,
#           height: 600,
#           quality: 80,
#           format: format,
#           fit: 'cover',
#         }
#       }
#     });
#   }
# };

"""CDN セキュリティ設計: DDoS 防御 + WAF + Bot 管理"""
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class CDNSecurityConfig:
    """CDN セキュリティ設定の設計"""
 
    # 1. TLS / SSL 設定
    tls_config: dict = field(default_factory=lambda: {
        'min_version': 'TLSv1.2',
        'preferred_ciphers': [
            'TLS_AES_256_GCM_SHA384',
            'TLS_CHACHA20_POLY1305_SHA256',
        ],
        'hsts': {
            'enabled': True,
            'max_age': 63072000,  # 2年
            'include_subdomains': True,
            'preload': True,
        },
        'ocsp_stapling': True,
    })
 
    # 2. WAF ルール
    waf_rules: dict = field(default_factory=lambda: {
        'managed_rules': [
            'OWASP Top 10',           # SQL Injection, XSS 等
            'Known Bad Inputs',        # 既知の攻撃パターン
            'Bot Management',          # 悪質ボット検出
        ],
        'custom_rules': [
            {
                'name': 'Block non-standard methods',
                'condition': 'http.request.method not in {"GET" "POST" "PUT" "DELETE" "PATCH" "OPTIONS"}',
                'action': 'block',
            },
            {
                'name': 'Rate limit login',
                'condition': 'http.request.uri.path eq "/api/auth/login"',
                'action': 'rate_limit',
                'rate': '10 per minute per ip',
            },
        ],
    })
 
    # 3. オリジン保護
    origin_protection: dict = field(default_factory=lambda: {
        # オリジンは CDN からのリクエストのみ受け付ける
        'allowed_ips': 'CDN IP ranges only',
        'origin_secret_header': {
            'name': 'X-Origin-Verify',
            'value': 'shared-secret-value',  # CDN → Origin の認証
        },
        # CloudFront: OAC (Origin Access Control)
        'oac_enabled': True,
    })
 
    # 4. DDoS 防御レイヤー
    ddos_protection: dict = field(default_factory=lambda: {
        'layer_3_4': {
            'provider': 'CDN built-in',  # TCP/UDP フラッド
            'capacity': '100+ Tbps',
        },
        'layer_7': {
            'rate_limiting': True,        # HTTP フラッド
            'challenge_page': True,       # JS チャレンジ
            'geo_blocking': ['KP', 'IR'], # 特定国のブロック
        },
    })
 
# オリジン保護の実装例 (Nginx)
NGINX_ORIGIN_PROTECTION = """
# CDN の IP レンジのみ許可
# CloudFront IP ranges: https://d7uri8nf7uskq.cloudfront.net/tools/list-cloudfront-ips
geo $is_cdn {
    default         0;
    13.224.0.0/14   1;  # CloudFront
    52.84.0.0/15    1;  # CloudFront
    99.84.0.0/16    1;  # CloudFront
    # ... (全レンジを記載)
}
 
server {
    # CDN 以外からのアクセスを拒否
    if ($is_cdn = 0) {
        return 403;
    }
 
    # CDN からのシークレットヘッダーを検証
    if ($http_x_origin_verify != "shared-secret-value") {
        return 403;
    }
}
"""

"""CDN パフォーマンスモニタリング"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
 
@dataclass
class CDNMetrics:
    """CDN モニタリングで追跡すべきメトリクス"""
 
    # キャッシュ効率
    cache_hit_ratio: float       # 目標: 静的 95%+, 動的 60%+
    cache_miss_ratio: float
    cache_expired_ratio: float   # TTL 切れの割合
 
    # パフォーマンス
    ttfb_p50_ms: float           # 目標: < 50ms
    ttfb_p95_ms: float           # 目標: < 200ms
    ttfb_p99_ms: float           # 目標: < 500ms
    total_latency_ms: float
 
    # オリジン健全性
    origin_request_count: int    # 目標: 全リクエストの 5-20%
    origin_error_rate: float     # 目標: < 0.1%
    origin_response_time_ms: float
 
    # 帯域・コスト
    bandwidth_gb: float
    request_count: int
    cost_usd: float
 
    # エラー
    http_4xx_rate: float         # 目標: < 1%
    http_5xx_rate: float         # 目標: < 0.01%
 
class CDNMonitor:
    """CDN メトリクスの収集と分析"""
 
    # アラート閾値
    ALERT_THRESHOLDS = {
        'cache_hit_ratio_low': 0.80,       # 80% 未満で警告
        'origin_error_rate_high': 0.01,     # 1% 以上で警告
        'ttfb_p95_high_ms': 200,            # 200ms 以上で警告
        'http_5xx_rate_high': 0.001,        # 0.1% 以上で緊急
    }
 
    @staticmethod
    def analyze_cache_efficiency(metrics: CDNMetrics) -> dict:
        """キャッシュ効率の分析とアドバイス"""
        analysis = {'status': 'healthy', 'recommendations': []}
 
        if metrics.cache_hit_ratio < 0.80:
            analysis['status'] = 'degraded'
            analysis['recommendations'].append(
                "キャッシュヒット率が低い。"
                "以下を確認: (1) TTL が短すぎないか、"
                "(2) Vary ヘッダーが過剰でないか、"
                "(3) クエリパラメータがキャッシュキーに不要に含まれていないか"
            )
 
        if metrics.cache_expired_ratio > 0.30:
            analysis['recommendations'].append(
                "キャッシュ期限切れが多い。TTL の延長または "
                "stale-while-revalidate の導入を検討"
            )
 
        if metrics.origin_error_rate > 0.01:
            analysis['status'] = 'critical'
            analysis['recommendations'].append(
                "オリジンエラー率が高い。"
                "stale-if-error ヘッダーの設定でユーザー影響を軽減"
            )
 
        return analysis
 
    @staticmethod
    def calculate_cost_savings(
        total_requests: int,
        cache_hit_ratio: float,
        avg_origin_cost_per_request: float = 0.00001,
    ) -> dict:
        """CDN によるコスト削減効果を計算"""
        requests_served_by_cache = int(total_requests * cache_hit_ratio)
        requests_to_origin = total_requests - requests_served_by_cache
        origin_cost = requests_to_origin * avg_origin_cost_per_request
        saved_cost = requests_served_by_cache * avg_origin_cost_per_request
 
        return {
            'total_requests': total_requests,
            'cache_hits': requests_served_by_cache,
            'origin_requests': requests_to_origin,
            'origin_cost_usd': round(origin_cost, 2),
            'saved_cost_usd': round(saved_cost, 2),
            'saving_ratio': f"{cache_hit_ratio * 100:.1f}%",
        }

# AWS CDK v2: CloudFront + S3 + WAF の本番構成
from aws_cdk import (
    Stack, Duration, RemovalPolicy,
    aws_s3 as s3,
    aws_cloudfront as cloudfront,
    aws_cloudfront_origins as origins,
    aws_certificatemanager as acm,
    aws_route53 as route53,
    aws_route53_targets as targets,
    aws_wafv2 as wafv2,
)
from constructs import Construct
 
class CDNStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
 
        # 1. S3 バケット (静的アセット)
        bucket = s3.Bucket(
            self, "StaticAssets",
            removal_policy=RemovalPolicy.RETAIN,
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            encryption=s3.BucketEncryption.S3_MANAGED,
            versioned=True,  # バージョニング有効
        )
 
        # 2. ACM 証明書 (us-east-1 で作成が必須)
        certificate = acm.Certificate(
            self, "Certificate",
            domain_name="cdn.example.com",
            validation=acm.CertificateValidation.from_dns(),
        )
 
        # 3. キャッシュポリシー
        cache_policy = cloudfront.CachePolicy(
            self, "CachePolicy",
            cache_policy_name="OptimizedCaching",
            default_ttl=Duration.hours(24),
            max_ttl=Duration.days(365),
            min_ttl=Duration.seconds(0),
            header_behavior=cloudfront.CacheHeaderBehavior.allow_list(
                "Accept-Encoding", "Accept-Language"
            ),
            query_string_behavior=cloudfront.CacheQueryStringBehavior.allow_list(
                "page", "sort", "category"  # 必要なパラメータのみ
            ),
            cookie_behavior=cloudfront.CacheCookieBehavior.none(),
            enable_accept_encoding_gzip=True,
            enable_accept_encoding_brotli=True,
        )
 
        # 4. CloudFront ディストリビューション
        distribution = cloudfront.Distribution(
            self, "Distribution",
            default_behavior=cloudfront.BehaviorOptions(
                origin=origins.S3Origin(
                    bucket,
                    origin_shield_region="ap-northeast-1",
                ),
                viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
                cache_policy=cache_policy,
                compress=True,
                allowed_methods=cloudfront.AllowedMethods.ALLOW_GET_HEAD,
            ),
            domain_names=["cdn.example.com"],
            certificate=certificate,
            http_version=cloudfront.HttpVersion.HTTP2_AND_3,
            price_class=cloudfront.PriceClass.PRICE_CLASS_200,
            error_responses=[
                cloudfront.ErrorResponse(
                    http_status=404,
                    response_http_status=200,
                    response_page_path="/index.html",  # SPA フォールバック
                    ttl=Duration.seconds(10),
                ),
            ],
        )
 
        # 5. Route 53 レコード
        zone = route53.HostedZone.from_lookup(
            self, "Zone", domain_name="example.com"
        )
        route53.ARecord(
            self, "CDNRecord",
            zone=zone,
            record_name="cdn",
            target=route53.RecordTarget.from_alias(
                targets.CloudFrontTarget(distribution)
            ),
        )

# NG: 全 URL に一律の Cache-Control
class BadCacheMiddleware:
    """全てのレスポンスに同じ TTL を適用"""
    async def __call__(self, request, call_next):
        response = await call_next(request)
        # 全てに 1日キャッシュ
        response.headers["Cache-Control"] = "public, max-age=86400"
        return response
 
# 問題:
# - index.html を更新しても24時間反映されない
# - API レスポンスが古いまま返される
# - 個人情報が CDN にキャッシュされるリスク
# - ハッシュ付き JS は1日で無駄に再取得
 
# OK: アセット種別ごとに TTL を最適化
class GoodCacheMiddleware:
    """アセット種別に応じた Cache-Control 設定"""
 
    CACHE_RULES = [
        # (パスパターン, Cache-Control)
        (r'^/api/user/', 'private, no-store'),
        (r'^/api/', 'public, s-maxage=10, stale-while-revalidate=60, stale-if-error=300'),
        (r'\.[a-f0-9]{8,}\.(js|css)$', 'public, max-age=31536000, immutable'),
        (r'\.(jpg|png|webp|avif)$', 'public, max-age=2592000'),
        (r'\.html$', 'public, s-maxage=60, max-age=0, must-revalidate'),
    ]
 
    async def __call__(self, request, call_next):
        response = await call_next(request)
        path = request.url.path
 
        for pattern, cache_control in self.CACHE_RULES:
            if re.match(pattern, path):
                response.headers["Cache-Control"] = cache_control
                break
        else:
            # デフォルト: キャッシュなし (安全側に倒す)
            response.headers["Cache-Control"] = "private, no-cache"
 
        return response

# NG: Vary: User-Agent でキャッシュキーを生成
class BadVaryConfig:
    """Vary: User-Agent を使う"""
    def get_response_headers(self):
        return {
            "Cache-Control": "public, s-maxage=300",
            "Vary": "User-Agent",  # 数千種類の UA → 数千のキャッシュバリエーション
        }
    # 結果: キャッシュヒット率が 5% 以下に低下
 
# NG: クエリパラメータ全てをキャッシュキーに含める
# /products?utm_source=google&utm_medium=cpc → キャッシュ MISS
# /products?utm_source=twitter              → キャッシュ MISS
# /products                                 → キャッシュ MISS
# → 同じコンテンツなのに3つの別キャッシュエントリ
 
# OK: 必要なパラメータのみキャッシュキーに含める
class GoodVaryConfig:
    """デバイス判定は CDN の機能を利用"""
    def get_response_headers(self, request):
        return {
            "Cache-Control": "public, s-maxage=300",
            "Vary": "Accept-Encoding, Accept-Language",
            # デバイス判定: CDN のヘッダーを利用
            # CloudFront: CloudFront-Is-Mobile-Viewer
            # Cloudflare: CF-Device-Type
        }
 
    def configure_cache_policy(self):
        """CloudFront キャッシュポリシーで UTM パラメータを除外"""
        return {
            'QueryStringBehavior': 'whitelist',
            'QueryStrings': ['page', 'sort', 'category', 'q'],
            # utm_*, fbclid, gclid は自動的に除外
        }

# NG: デプロイのたびに全キャッシュをパージ
class BadDeployment:
    """デプロイ時に /* をパージ"""
    def deploy(self):
        # 1. ファイルをアップロード
        upload_files()
 
        # 2. 全キャッシュをパージ
        invalidate_all("/*")
 
        # 問題:
        # - パージ完了まで数秒〜数十秒のラグ
        # - パージ中に古いHTMLが新しいJSを参照 → エラー
        # - 全POPで一貫性が保たれない瞬間がある
        # - パージ費用が発生 (CloudFront: 1000パス/月超で課金)
 
# OK: キャッシュバスティングでパージを不要にする
class GoodDeployment:
    """ハッシュ付きファイル名でパージ不要のデプロイ"""
    def deploy(self):
        # 1. アセットはハッシュ付きファイル名 (パージ不要)
        #    main.js → main.a1b2c3d4.js (新ファイル = 新キャッシュ)
        upload_hashed_assets()
 
        # 2. HTML だけ短 TTL (s-maxage=60) で自然更新
        upload_html(cache_control="s-maxage=60, must-revalidate")
 
        # 3. 緊急時のみ HTML をパージ
        invalidate_paths(["/index.html"])
 
        # メリット:
        # - 一貫性の問題なし (HTMLが新JSを参照する時点でJSは既にCDNにある)
        # - パージコスト最小 (HTMLのみ)
        # - 古いバージョンへのロールバックも即座 (HTMLのパスを戻すだけ)

access_logs = [
    {"url": "/products?page=1&utm_source=google", "cache": "MISS"},
    {"url": "/products?page=1&utm_source=twitter", "cache": "MISS"},
    {"url": "/products?page=1", "cache": "MISS"},
    {"url": "/products?page=1&fbclid=abc", "cache": "MISS"},
    {"url": "/api/user/profile", "cache": "MISS", "vary": "User-Agent"},
    {"url": "/api/user/profile", "cache": "MISS", "vary": "User-Agent"},
    {"url": "/images/hero.jpg", "cache": "MISS"},
    {"url": "/images/hero.jpg?v=1", "cache": "MISS"},
    {"url": "/images/hero.jpg?v=2", "cache": "MISS"},
]
 
# 課題:
# 1. 現在のキャッシュヒット率を計算せよ
# 2. 各 MISS の原因を特定せよ
# 3. キャッシュヒット率を改善する施策を3つ以上提案せよ
# 4. 改善後の期待キャッシュヒット率を見積もれ