# === スケール見積もり計算 ===
 
# 前提
daily_url_creation = 100_000_000    # 1日1億 URL 作成
read_write_ratio = 100              # 読み取り:書き込み = 100:1
daily_redirects = daily_url_creation * read_write_ratio  # 100億リダイレクト/日
retention_years = 5                 # 5年間保持
 
# QPS (Queries Per Second)
write_qps = daily_url_creation / 86400           # ≈ 1,160 QPS
write_qps_peak = write_qps * 2                   # ≈ 2,320 QPS (ピーク)
read_qps = daily_redirects / 86400               # ≈ 115,740 QPS
read_qps_peak = read_qps * 2                     # ≈ 231,480 QPS (ピーク)
 
# ストレージ
total_urls = daily_url_creation * 365 * retention_years  # 182.5B (1,825億)
bytes_per_record = 500                                    # URL + メタデータ
total_storage_tb = total_urls * bytes_per_record / (1024**4)  # ≈ 91.25 TB
 
# 帯域幅
avg_redirect_size_bytes = 500           # リダイレクトレスポンスのサイズ
bandwidth_mbps = (read_qps * avg_redirect_size_bytes * 8) / (1024**2)  # ≈ 442 Mbps
 
# キャッシュ (80-20 の法則)
daily_unique_urls = daily_redirects * 0.2         # 20% がユニーク
cache_memory_gb = (daily_unique_urls * bytes_per_record) / (1024**3)  # ≈ 931 GB
 
print(f"""
=== スケール見積もり結果 ===
書き込み QPS:     {write_qps:,.0f} (ピーク: {write_qps_peak:,.0f})
読み取り QPS:     {read_qps:,.0f} (ピーク: {read_qps_peak:,.0f})
5年間の総URL数:   {total_urls / 1e9:.1f}B ({total_urls / 1e9:.1f}億)
ストレージ:       {total_storage_tb:.1f} TB
帯域幅:          {bandwidth_mbps:.0f} Mbps
キャッシュ必要量: {cache_memory_gb:.0f} GB
""")

# === 方式1: Base62 エンコーディング ===
import string
 
CHARSET = string.digits + string.ascii_lowercase + string.ascii_uppercase
BASE = len(CHARSET)  # 62
 
 
def encode_base62(num: int) -> str:
    """整数を Base62 文字列に変換する
 
    62進法でエンコード:
    0-9 → '0'-'9'
    10-35 → 'a'-'z'
    36-61 → 'A'-'Z'
    """
    if num == 0:
        return CHARSET[0]
    result = []
    while num > 0:
        result.append(CHARSET[num % BASE])
        num //= BASE
    return ''.join(reversed(result))
 
 
def decode_base62(s: str) -> int:
    """Base62 文字列を整数に戻す"""
    num = 0
    for char in s:
        num = num * BASE + CHARSET.index(char)
    return num
 
 
# 7文字の Base62 で表現可能な範囲
# 62^7 = 3,521,614,606,208 (約3.5兆通り)
assert encode_base62(123456789) == "8M0kX"
assert decode_base62("8M0kX") == 123456789
 
 
# === 方式2: MD5 ハッシュの先頭7文字 ===
import hashlib
 
 
def generate_hash_key(url: str, attempt: int = 0) -> str:
    """URL の MD5 ハッシュから短縮キーを生成
 
    衝突時は attempt をインクリメントして再試行
    """
    input_str = url if attempt == 0 else f"{url}:{attempt}"
    hash_hex = hashlib.md5(input_str.encode()).hexdigest()
 
    # 16進数を整数に変換し、Base62 エンコード
    hash_int = int(hash_hex[:12], 16)  # 先頭12文字を使用
    key = encode_base62(hash_int)
    return key[:7]  # 7文字に切り詰め
 
 
# 例
print(generate_hash_key("https://example.com/very/long/path"))  # => "aBcD123"
 
 
# === 方式3: Snowflake ID + Base62 ===
import time
import threading
 
 
class SnowflakeIDGenerator:
    """Twitter Snowflake 方式の分散 ID 生成
 
    64bit の ID:
    - 1bit: 符号（常に0）
    - 41bit: タイムスタンプ（ミリ秒）→ 約69年
    - 10bit: マシン ID（1024台）
    - 12bit: シーケンス番号（4096/ms/マシン）
 
    特徴:
    - 時系列順のID生成
    - 分散環境で衝突なし
    - 高スループット（1マシンで409万ID/秒）
    """
 
    def __init__(self, machine_id: int):
        self._machine_id = machine_id & 0x3FF  # 10bit
        self._sequence = 0
        self._last_timestamp = 0
        self._lock = threading.Lock()
        self._epoch = 1704067200000  # 2024-01-01 00:00:00 UTC
 
    def next_id(self) -> int:
        with self._lock:
            timestamp = int(time.time() * 1000) - self._epoch
 
            if timestamp == self._last_timestamp:
                self._sequence = (self._sequence + 1) & 0xFFF
                if self._sequence == 0:
                    # 同一ミリ秒でシーケンス上限 → 次のミリ秒まで待機
                    while timestamp <= self._last_timestamp:
                        timestamp = int(time.time() * 1000) - self._epoch
            else:
                self._sequence = 0
 
            self._last_timestamp = timestamp
 
            return (
                (timestamp << 22) |
                (self._machine_id << 12) |
                self._sequence
            )
 
 
# Snowflake ID → Base62 短縮キー
generator = SnowflakeIDGenerator(machine_id=1)
snowflake_id = generator.next_id()
short_key = encode_base62(snowflake_id)[:7]

# infrastructure/key_generation_service.py
import redis
import random
import string
import logging
from typing import Optional
 
logger = logging.getLogger(__name__)
 
 
class KeyGenerationService:
    """短縮URLキーを事前生成し、Redis のキューに格納する。
 
    設計:
    1. バックグラウンドで大量のキーを事前生成
    2. Redis Set に格納（未使用キープール）
    3. API サーバーは spop でアトミックにキーを取得
    4. キー残量が閾値を下回ったら自動補充

"""
 
    UNUSED_KEY = "kgs:unused_keys"
    USED_KEY = "kgs:used_keys"
    MIN_POOL_SIZE = 100_000
    BATCH_GENERATE_SIZE = 500_000
 
    def __init__(
        self,
        redis_client: redis.Redis,
        key_length: int = 7,
    ):
        self._redis = redis_client
        self._key_length = key_length
 
    def generate_keys(self, count: int = BATCH_GENERATE_SIZE) -> int:
        """キーを一括生成してプールに追加する
 
        Returns:
            実際に追加されたキー数
        """
        charset = string.ascii_letters + string.digits
        generated = 0
        pipe = self._redis.pipeline()
 
        for _ in range(count):
            key = ''.join(random.choices(charset, k=self._key_length))
            pipe.sadd(self.UNUSED_KEY, key)
            generated += 1
 
            # パイプラインのバッファが大きくなりすぎないように
            if generated % 10_000 == 0:
                pipe.execute()
                pipe = self._redis.pipeline()
 
        pipe.execute()
        pool_size = self.pool_size()
        logger.info(
            f"キー生成完了: {generated}件追加, "
            f"プールサイズ: {pool_size}"
        )
        return generated
 
    def get_key(self) -> str:
        """未使用キーをプールから取得する（アトミック操作）
 
        Raises:
            RuntimeError: キープールが枯渇した場合
        """
        key = self._redis.spop(self.UNUSED_KEY)
        if key is None:
            raise RuntimeError(
                "キープール枯渇! generate_keys() を実行してください"
            )
        # 使用済みセットに追加（重複チェック用）
        self._redis.sadd(self.USED_KEY, key)
        return key.decode()
 
    def return_key(self, key: str) -> None:
        """キーをプールに返却する（URL 作成が失敗した場合）"""
        self._redis.srem(self.USED_KEY, key)
        self._redis.sadd(self.UNUSED_KEY, key)
 
    def pool_size(self) -> int:
        """未使用キーの残数を取得"""
        return self._redis.scard(self.UNUSED_KEY)
 
    def ensure_pool_size(self) -> None:
        """プールサイズが閾値を下回ったら自動補充"""
        current = self.pool_size()
        if current < self.MIN_POOL_SIZE:
            logger.warning(
                f"キープール残量低下: {current}, 補充開始"
            )
            self.generate_keys(self.BATCH_GENERATE_SIZE)
 
    def is_key_used(self, key: str) -> bool:
        """キーが使用済みか確認（カスタムエイリアスの重複チェック用）"""
        return self._redis.sismember(self.USED_KEY, key)

-- URL マッピングテーブル（メインテーブル）
CREATE TABLE url_mappings (
    id            BIGINT PRIMARY KEY AUTO_INCREMENT,
    short_key     VARCHAR(7) NOT NULL UNIQUE,
    original_url  TEXT NOT NULL,
    user_id       BIGINT NULL,
    custom_alias  BOOLEAN DEFAULT FALSE,
    created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    expires_at    TIMESTAMP NULL,
    click_count   BIGINT DEFAULT 0,
    is_active     BOOLEAN DEFAULT TRUE,
 
    -- インデックス
    INDEX idx_short_key (short_key),             -- リダイレクト時のルックアップ
    INDEX idx_user_id (user_id),                 -- ユーザーの URL 一覧
    INDEX idx_expires_at (expires_at),           -- 期限切れクリーンアップ
    INDEX idx_created_at (created_at)            -- 最新順の一覧
) ENGINE=InnoDB;
 
-- アクセスログ（分析用、別テーブル/別 DB）
CREATE TABLE click_events (
    id            BIGINT PRIMARY KEY AUTO_INCREMENT,
    short_key     VARCHAR(7) NOT NULL,
    clicked_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    ip_address    VARCHAR(45),                    -- IPv6 対応
    user_agent    VARCHAR(500),
    referer       VARCHAR(2048),
    country_code  CHAR(2),
    device_type   VARCHAR(20),                    -- mobile / desktop / tablet
    os            VARCHAR(50),
    browser       VARCHAR(50),
 
    -- パーティション: 月単位（大量データの管理用）
    INDEX idx_short_key_time (short_key, clicked_at)
) ENGINE=InnoDB
PARTITION BY RANGE (UNIX_TIMESTAMP(clicked_at)) (
    PARTITION p2026_01 VALUES LESS THAN (UNIX_TIMESTAMP('2026-02-01')),
    PARTITION p2026_02 VALUES LESS THAN (UNIX_TIMESTAMP('2026-03-01')),
    PARTITION p_future VALUES LESS THAN MAXVALUE
);
 
-- ユーザーテーブル（認証機能がある場合）
CREATE TABLE users (
    id            BIGINT PRIMARY KEY AUTO_INCREMENT,
    email         VARCHAR(255) NOT NULL UNIQUE,
    api_key       VARCHAR(64) NOT NULL UNIQUE,
    plan          ENUM('free', 'pro', 'enterprise') DEFAULT 'free',
    rate_limit    INT DEFAULT 100,                -- 1時間あたりの URL 作成上限
    created_at    TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 
    INDEX idx_api_key (api_key)
) ENGINE=InnoDB;

# DynamoDB スキーマ設計
 
"""
DynamoDB テーブル設計（NoSQL 代替案）
 
テーブル: url_mappings
  パーティションキー: short_key (String)
  属性:
    - original_url (String)
    - user_id (String, optional)
    - created_at (Number, Unix timestamp)
    - expires_at (Number, Unix timestamp, optional)
    - click_count (Number)
    - is_active (Boolean)
 
  GSI: user_id-index
    パーティションキー: user_id
    ソートキー: created_at
 
メリット:
  - short_key でのルックアップが O(1) → 超低レイテンシ
  - 水平スケールが自動（シャーディング不要）
  - TTL 機能で期限切れ URL を自動削除
 
デメリット:
  - 複雑なクエリが困難（SQL の柔軟性がない）
  - コストが読み書きキャパシティに依存
"""
 
import boto3
from datetime import datetime, timezone, timedelta
 
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('url_mappings')
 
 
def create_short_url(short_key: str, original_url: str, ttl_days: int = 0):
    """URL マッピングを作成"""
    item = {
        'short_key': short_key,
        'original_url': original_url,
        'created_at': int(datetime.now(timezone.utc).timestamp()),
        'click_count': 0,
        'is_active': True,
    }
    if ttl_days > 0:
        expires_at = datetime.now(timezone.utc) + timedelta(days=ttl_days)
        item['expires_at'] = int(expires_at.timestamp())
        item['ttl'] = int(expires_at.timestamp())  # DynamoDB TTL
 
    table.put_item(
        Item=item,
        ConditionExpression='attribute_not_exists(short_key)',  # 重複防止
    )
 
 
def get_original_url(short_key: str) -> str | None:
    """短縮キーから元 URL を取得"""
    response = table.get_item(
        Key={'short_key': short_key},
        ProjectionExpression='original_url, is_active, expires_at',
    )
    item = response.get('Item')
    if not item or not item.get('is_active', True):
        return None
 
    # 期限切れチェック
    expires_at = item.get('expires_at')
    if expires_at and expires_at < int(datetime.now(timezone.utc).timestamp()):
        return None
 
    return item['original_url']
 
 
def increment_click_count(short_key: str):
    """クリックカウントをアトミックにインクリメント"""
    table.update_item(
        Key={'short_key': short_key},
        UpdateExpression='ADD click_count :inc',
        ExpressionAttributeValues={':inc': 1},
    )

# application/api/url_shortener_api.py
from fastapi import FastAPI, HTTPException, Request, Depends, Header
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, HttpUrl, Field
from typing import Optional
import redis
import databases
import logging
import asyncio
from datetime import datetime, timezone
 
logger = logging.getLogger(__name__)
 
app = FastAPI(title="URL Shortener API")
cache = redis.Redis(host="redis-host", port=6379, db=0, decode_responses=True)
db = databases.Database("mysql://user:pass@db-host/url_shortener")
 
CACHE_TTL = 86400  # 24時間
BASE_URL = "https://short.ly"
 
 
class ShortenRequest(BaseModel):
    """URL 短縮リクエスト"""
    url: HttpUrl
    custom_alias: Optional[str] = Field(
        None, min_length=4, max_length=30, pattern=r'^[a-zA-Z0-9_-]+$'
    )
    expires_in_days: Optional[int] = Field(None, ge=1, le=3650)
 
 
class ShortenResponse(BaseModel):
    """URL 短縮レスポンス"""
    short_url: str
    short_key: str
    original_url: str
    expires_at: Optional[str] = None
 
 
class UrlStatsResponse(BaseModel):
    """URL 統計レスポンス"""
    short_key: str
    original_url: str
    click_count: int
    created_at: str
    expires_at: Optional[str] = None
 
 
# === URL 短縮 API ===
 
@app.post("/api/v1/shorten", response_model=ShortenResponse)
async def shorten_url(
    request: ShortenRequest,
    api_key: str = Header(..., alias="X-API-Key"),
):
    """長い URL を短縮する"""
    # 1. API キーの認証（省略: 実際は認証ミドルウェアで処理）
    user = await authenticate_api_key(api_key)
 
    # 2. レート制限チェック
    await check_rate_limit(user.id)
 
    # 3. URL の安全性チェック（フィッシング等）
    if not await is_safe_url(str(request.url)):
        raise HTTPException(
            status_code=400,
            detail="URL が安全でない可能性があります"
        )
 
    # 4. 短縮キーの取得
    if request.custom_alias:
        short_key = request.custom_alias
        # カスタムエイリアスの重複チェック
        existing = await db.fetch_one(
            "SELECT id FROM url_mappings WHERE short_key = :key",
            {"key": short_key}
        )
        if existing:
            raise HTTPException(
                status_code=409,
                detail=f"エイリアス '{short_key}' は既に使用されています"
            )
    else:
        kgs = KeyGenerationService(redis_client=cache)
        short_key = kgs.get_key()
 
    # 5. 有効期限の計算
    expires_at = None
    if request.expires_in_days:
        expires_at = datetime.now(timezone.utc) + timedelta(
            days=request.expires_in_days
        )
 
    # 6. DB に保存
    try:
        await db.execute(
            """
            INSERT INTO url_mappings
                (short_key, original_url, user_id, custom_alias, expires_at)
            VALUES (:key, :url, :user_id, :custom, :expires)
            """,
            {
                "key": short_key,
                "url": str(request.url),
                "user_id": user.id,
                "custom": request.custom_alias is not None,
                "expires": expires_at,
            }
        )
    except Exception as e:
        # DB 保存失敗時はキーをプールに返却
        if not request.custom_alias:
            kgs.return_key(short_key)
        raise HTTPException(status_code=500, detail="URL の作成に失敗しました")
 
    # 7. キャッシュにも保存
    cache.setex(f"url:{short_key}", CACHE_TTL, str(request.url))
 
    return ShortenResponse(
        short_url=f"{BASE_URL}/{short_key}",
        short_key=short_key,
        original_url=str(request.url),
        expires_at=expires_at.isoformat() if expires_at else None,
    )
 
 
# === リダイレクト API ===
 
@app.get("/{short_key}")
async def redirect(short_key: str, request: Request):
    """短縮 URL から元 URL へリダイレクトする"""
    # 1. キャッシュを確認
    cached_url = cache.get(f"url:{short_key}")
    if cached_url:
        # 非同期でクリックイベントを記録（リダイレクトをブロックしない）
        asyncio.create_task(
            record_click_event(short_key, request)
        )
        return RedirectResponse(
            url=cached_url,
            status_code=301,  # 永続リダイレクト
        )
 
    # 2. DB を確認
    row = await db.fetch_one(
        """
        SELECT original_url, expires_at, is_active
        FROM url_mappings
        WHERE short_key = :key
        """,
        {"key": short_key}
    )
 
    if not row:
        raise HTTPException(status_code=404, detail="URL が見つかりません")
 
    if not row["is_active"]:
        raise HTTPException(status_code=410, detail="URL は無効化されています")
 
    # 3. 期限切れチェック
    if row["expires_at"] and row["expires_at"] < datetime.now(timezone.utc):
        raise HTTPException(status_code=410, detail="URL の有効期限が切れています")
 
    original_url = row["original_url"]
 
    # 4. キャッシュに保存
    cache.setex(f"url:{short_key}", CACHE_TTL, original_url)
 
    # 5. 非同期でクリックイベントを記録
    asyncio.create_task(record_click_event(short_key, request))
 
    return RedirectResponse(url=original_url, status_code=301)
 
 
# === 統計 API ===
 
@app.get("/api/v1/stats/{short_key}", response_model=UrlStatsResponse)
async def get_stats(
    short_key: str,
    api_key: str = Header(..., alias="X-API-Key"),
):
    """URL のアクセス統計を取得する"""
    user = await authenticate_api_key(api_key)
 
    row = await db.fetch_one(
        """
        SELECT short_key, original_url, click_count, created_at, expires_at
        FROM url_mappings
        WHERE short_key = :key AND user_id = :user_id
        """,
        {"key": short_key, "user_id": user.id}
    )
    if not row:
        raise HTTPException(status_code=404, detail="URL が見つかりません")
 
    return UrlStatsResponse(
        short_key=row["short_key"],
        original_url=row["original_url"],
        click_count=row["click_count"],
        created_at=row["created_at"].isoformat(),
        expires_at=row["expires_at"].isoformat() if row["expires_at"] else None,
    )
 
 
# === ヘルパー関数 ===
 
async def record_click_event(short_key: str, request: Request) -> None:
    """クリックイベントを非同期で記録する
 
    注: 本番環境ではメッセージキュー（Kafka 等）に
    発行して別サービスで処理するのが推奨
    """
    try:
        await db.execute(
            """
            INSERT INTO click_events
                (short_key, ip_address, user_agent, referer)
            VALUES (:key, :ip, :ua, :referer)
            """,
            {
                "key": short_key,
                "ip": request.client.host,
                "ua": request.headers.get("User-Agent", "")[:500],
                "referer": request.headers.get("Referer", "")[:2048],
            }
        )
        # クリックカウントのインクリメント
        await db.execute(
            "UPDATE url_mappings SET click_count = click_count + 1 WHERE short_key = :key",
            {"key": short_key}
        )
    except Exception as e:
        # クリック記録の失敗はリダイレクトに影響させない
        logger.error(f"クリックイベント記録失敗: {short_key}, {e}")
 
 
async def is_safe_url(url: str) -> bool:
    """URL の安全性チェック（Google Safe Browsing API 等）"""
    # 実装省略: 実際は外部 API で確認
    dangerous_patterns = [
        "phishing.example.com",
        "malware.example.com",
    ]
    return not any(pattern in url for pattern in dangerous_patterns)
 
 
async def check_rate_limit(user_id: int) -> None:
    """レート制限チェック"""
    key = f"rate_limit:{user_id}"
    current = cache.incr(key)
    if current == 1:
        cache.expire(key, 3600)  # 1時間のウィンドウ
    if current > 100:  # 1時間あたり100件
        raise HTTPException(
            status_code=429,
            detail="レート制限を超過しました。1時間後に再試行してください"
        )
 
 
async def authenticate_api_key(api_key: str):
    """API キーの認証"""
    row = await db.fetch_one(
        "SELECT id, email, plan, rate_limit FROM users WHERE api_key = :key",
        {"key": api_key}
    )
    if not row:
        raise HTTPException(status_code=401, detail="無効な API キーです")
    return row

# infrastructure/cache/redis_cache.py
import redis
from redis.sentinel import Sentinel
import json
import logging
 
logger = logging.getLogger(__name__)
 
 
class UrlCache:
    """URL キャッシュ（Redis Cluster / Sentinel 対応）
 
    設計判断:
    - Redis Cluster: 水平分散（16384 スロット）
    - Sentinel: 高可用性（マスター障害時の自動フェイルオーバー）
    - TTL: 24時間（ホットな URL は自然にキャッシュに残る）
    """
 
    DEFAULT_TTL = 86400  # 24時間
 
    def __init__(self, redis_client: redis.Redis):
        self._redis = redis_client
 
    def get_url(self, short_key: str) -> str | None:
        """キャッシュから URL を取得"""
        result = self._redis.get(f"url:{short_key}")
        if result:
            logger.debug(f"キャッシュヒット: {short_key}")
            return result
        logger.debug(f"キャッシュミス: {short_key}")
        return None
 
    def set_url(
        self, short_key: str, original_url: str, ttl: int = DEFAULT_TTL
    ) -> None:
        """URL をキャッシュに保存"""
        self._redis.setex(f"url:{short_key}", ttl, original_url)
 
    def delete_url(self, short_key: str) -> None:
        """キャッシュから URL を削除（URL 無効化時）"""
        self._redis.delete(f"url:{short_key}")
 
    def get_cache_stats(self) -> dict:
        """キャッシュの統計情報"""
        info = self._redis.info("stats")
        hits = info.get("keyspace_hits", 0)
        misses = info.get("keyspace_misses", 0)
        total = hits + misses
        return {
            "hits": hits,
            "misses": misses,
            "hit_rate": f"{(hits / total * 100):.1f}%" if total > 0 else "N/A",
            "total_keys": self._redis.dbsize(),
        }

# infrastructure/database/shard_router.py
import hashlib
 
 
class ConsistentHashShardRouter:
    """一貫性ハッシュに基づくシャードルーティング
 
    設計判断:
    - short_key のハッシュでシャードを決定
    - 一貫性ハッシュリングでシャード追加/削除時の影響を最小化
    - 仮想ノードで負荷の均等分散を実現

"""
 
    def __init__(
        self,
        shard_configs: dict[int, str],
        virtual_nodes: int = 150,
    ):
        self._shard_configs = shard_configs
        self._virtual_nodes = virtual_nodes
        self._ring: dict[int, int] = {}
        self._sorted_keys: list[int] = []
        self._build_ring()
 
    def _build_ring(self) -> None:
        """ハッシュリングを構築"""
        for shard_id in self._shard_configs:
            for i in range(self._virtual_nodes):
                key = f"shard-{shard_id}-vnode-{i}"
                hash_val = self._hash(key)
                self._ring[hash_val] = shard_id
        self._sorted_keys = sorted(self._ring.keys())
 
    def get_shard(self, short_key: str) -> int:
        """短縮キーからシャード番号を決定"""
        hash_val = self._hash(short_key)
        # 二分探索で最初の仮想ノードを見つける
        for ring_key in self._sorted_keys:
            if hash_val <= ring_key:
                return self._ring[ring_key]
        # リングの末尾を超えた場合は先頭に戻る
        return self._ring[self._sorted_keys[0]]
 
    def get_connection_string(self, short_key: str) -> str:
        """シャードの接続文字列を取得"""
        shard_id = self.get_shard(short_key)
        return self._shard_configs[shard_id]
 
    @staticmethod
    def _hash(key: str) -> int:
        """キーのハッシュ値を計算"""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
 
 
# 使用例
router = ConsistentHashShardRouter(
    shard_configs={
        0: "mysql://user:pass@shard-0.db/urls",
        1: "mysql://user:pass@shard-1.db/urls",
        2: "mysql://user:pass@shard-2.db/urls",
        3: "mysql://user:pass@shard-3.db/urls",
    },
    virtual_nodes=150,
)
 
# 短縮キーからシャードを決定
shard_id = router.get_shard("abc123")
conn_str = router.get_connection_string("abc123")

def determine_redirect_status(url_mapping: dict) -> int:
    """リダイレクトのステータスコードを判断する
 
    判断基準:
    1. アクセス統計が有効 → 302（毎回サーバーを経由させる）
    2. カスタムエイリアス → 302（リダイレクト先の変更可能性）
    3. それ以外 → 301（パフォーマンス最優先）
    """
    if url_mapping.get('analytics_enabled'):
        return 302  # 毎回サーバーを経由 → 統計取得可能
    if url_mapping.get('custom_alias'):
        return 302  # リダイレクト先が変更される可能性
    if url_mapping.get('expires_at'):
        return 302  # 有効期限あり → 期限切れの検知が必要
    return 301      # 永続リダイレクト → 最高のパフォーマンス

# tests/test_url_shortener_api.py
import pytest
from httpx import AsyncClient
from unittest.mock import AsyncMock, patch
 
 
class TestShortenURL:
    """URL 短縮 API のテスト"""
 
    @pytest.fixture
    def mock_kgs(self):
        kgs = AsyncMock()
        kgs.get_key.return_value = "abc1234"
        return kgs
 
    @pytest.mark.asyncio
    async def test_URL短縮_正常(self, client: AsyncClient, mock_kgs):
        """有効な URL を短縮できる"""
        response = await client.post(
            "/api/v1/shorten",
            json={"url": "https://example.com/very/long/path"},
            headers={"X-API-Key": "valid-api-key"},
        )
 
        assert response.status_code == 200
        data = response.json()
        assert "short_url" in data
        assert data["short_url"].startswith("https://short.ly/")
        assert len(data["short_key"]) == 7
 
    @pytest.mark.asyncio
    async def test_カスタムエイリアス(self, client: AsyncClient):
        """カスタムエイリアスで短縮 URL を作成できる"""
        response = await client.post(
            "/api/v1/shorten",
            json={
                "url": "https://example.com/event",
                "custom_alias": "my-event",
            },
            headers={"X-API-Key": "valid-api-key"},
        )
 
        assert response.status_code == 200
        data = response.json()
        assert data["short_key"] == "my-event"
 
    @pytest.mark.asyncio
    async def test_カスタムエイリアス重複(self, client: AsyncClient):
        """既存のエイリアスと重複する場合は 409 エラー"""
        # 1回目: 成功
        await client.post(
            "/api/v1/shorten",
            json={
                "url": "https://example.com/1",
                "custom_alias": "duplicate",
            },
            headers={"X-API-Key": "valid-api-key"},
        )
 
        # 2回目: 409 Conflict
        response = await client.post(
            "/api/v1/shorten",
            json={
                "url": "https://example.com/2",
                "custom_alias": "duplicate",
            },
            headers={"X-API-Key": "valid-api-key"},
        )
        assert response.status_code == 409
 
    @pytest.mark.asyncio
    async def test_不正なURL(self, client: AsyncClient):
        """不正な URL は 422 エラー"""
        response = await client.post(
            "/api/v1/shorten",
            json={"url": "not-a-valid-url"},
            headers={"X-API-Key": "valid-api-key"},
        )
        assert response.status_code == 422
 
    @pytest.mark.asyncio
    async def test_レート制限(self, client: AsyncClient):
        """レート制限を超えると 429 エラー"""
        for _ in range(101):  # 100 + 1 回
            response = await client.post(
                "/api/v1/shorten",
                json={"url": "https://example.com/test"},
                headers={"X-API-Key": "valid-api-key"},
            )
 
        assert response.status_code == 429
 
 
class TestRedirect:
    """リダイレクト API のテスト"""
 
    @pytest.mark.asyncio
    async def test_リダイレクト_正常(self, client: AsyncClient):
        """有効な短縮 URL でリダイレクトされる"""
        # URL を作成
        create_resp = await client.post(
            "/api/v1/shorten",
            json={"url": "https://example.com/target"},
            headers={"X-API-Key": "valid-api-key"},
        )
        short_key = create_resp.json()["short_key"]
 
        # リダイレクト
        response = await client.get(
            f"/{short_key}",
            follow_redirects=False,
        )
        assert response.status_code in (301, 302)
        assert response.headers["location"] == "https://example.com/target"
 
    @pytest.mark.asyncio
    async def test_存在しないキー(self, client: AsyncClient):
        """存在しない短縮キーは 404 エラー"""
        response = await client.get("/nonexistent", follow_redirects=False)
        assert response.status_code == 404
 
    @pytest.mark.asyncio
    async def test_期限切れURL(self, client: AsyncClient):
        """期限切れの URL は 410 エラー"""
        # 期限切れ URL を直接 DB に作成（テスト用）
        # ... 省略
        response = await client.get("/expired-key", follow_redirects=False)
        assert response.status_code == 410
 
 
class TestKeyGenerationService:
    """KGS のテスト"""
 
    def test_キー生成(self):
        """指定数のキーが生成される"""
        kgs = KeyGenerationService(redis_client=fake_redis)
        generated = kgs.generate_keys(count=1000)
        assert generated == 1000
        assert kgs.pool_size() >= 1000
 
    def test_キー取得はアトミック(self):
        """同じキーが2回取得されない"""
        kgs = KeyGenerationService(redis_client=fake_redis)
        kgs.generate_keys(count=100)
 
        keys = set()
        for _ in range(100):
            key = kgs.get_key()
            assert key not in keys, f"重複キー: {key}"
            keys.add(key)
 
    def test_プール枯渇時はエラー(self):
        """プールが空の場合は RuntimeError"""
        kgs = KeyGenerationService(redis_client=fake_redis)
        # プールは空
        with pytest.raises(RuntimeError, match="キープール枯渇"):
            kgs.get_key()
 
    def test_キー返却(self):
        """失敗時にキーをプールに返却できる"""
        kgs = KeyGenerationService(redis_client=fake_redis)
        kgs.generate_keys(count=10)
        initial_size = kgs.pool_size()
 
        key = kgs.get_key()
        assert kgs.pool_size() == initial_size - 1
 
        kgs.return_key(key)
        assert kgs.pool_size() == initial_size

# tests/test_base62.py
import pytest
 
 
class TestBase62:
    """Base62 エンコーディングのテスト"""
 
    def test_エンコード_デコードの往復(self):
        """エンコード → デコードで元の値に戻る"""
        for num in [0, 1, 61, 62, 100, 123456789, 2**40]:
            encoded = encode_base62(num)
            decoded = decode_base62(encoded)
            assert decoded == num, f"num={num}, encoded={encoded}"
 
    def test_キー長の検証(self):
        """7文字以内で十分な範囲をカバーできる"""
        # 62^7 = 3,521,614,606,208
        max_7char = 62**7 - 1
        encoded = encode_base62(max_7char)
        assert len(encoded) == 7
 
    def test_ゼロ(self):
        """0 のエンコード"""
        assert encode_base62(0) == "0"
 
    def test_全文字が使われる(self):
        """Base62 の全文字（0-9, a-z, A-Z）が使われる"""
        chars_used = set()
        for i in range(62):
            chars_used.add(encode_base62(i))
        assert len(chars_used) == 62

assert encode_base62(0) == "0"
assert encode_base62(61) == "Z"
assert encode_base62(62) == "10"
assert encode_base62(123456789) == "8M0kX"
assert decode_base62("8M0kX") == 123456789
 
# 往復テスト
for i in range(10000):
    assert decode_base62(encode_base62(i)) == i

# 並行テスト
import concurrent.futures
 
kgs = KeyGenerationService(redis_client)
kgs.generate_keys(count=1000)
 
all_keys = []
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
    futures = [executor.submit(kgs.get_key) for _ in range(1000)]
    for future in concurrent.futures.as_completed(futures):
        all_keys.append(future.result())
 
# 全キーがユニーク
assert len(set(all_keys)) == 1000

# 戦略1: Lazy Deletion（リダイレクト時に確認）
@app.get("/{short_key}")
async def redirect(short_key):
    url_data = await get_url_data(short_key)
    if url_data['expires_at'] < datetime.now():
        raise HTTPException(status_code=410, detail="期限切れ")
    return RedirectResponse(url_data['original_url'])
 
# 戦略2: Background Cleanup（定期バッチ）
# Cron: 毎日 3:00 AM に実行
async def cleanup_expired_urls():
    deleted = await db.execute(
        "DELETE FROM url_mappings WHERE expires_at < NOW() LIMIT 10000"
    )
    # Redis キャッシュからも削除
    # ...
 
# 戦略3: TTL in Cache
# Redis の TTL で自動期限切れ
cache.setex(f"url:{short_key}", ttl_seconds, original_url)
 
# DynamoDB の TTL（NoSQL の場合）
# expires_at カラムを TTL として設定 → 自動削除