"""
一貫性モデルシミュレーター
==========================================
結果整合性と強い一貫性の振る舞いの違いを
スレッドを使って再現する。
 
実行方法: python consistency_simulator.py
依存: 標準ライブラリのみ
"""
 
import threading
import time
import random
from typing import Dict, List, Optional, Tuple
 
 
class Node:
    """分散システムの1ノードを表現するクラス。
 
    なぜロックが必要か:
    Pythonのdictはスレッドセーフではないため、
    複数スレッドからの同時アクセスでデータ破損が起きうる。
    """
 
    def __init__(self, node_id: str, latency_ms: float = 0):
        self.node_id = node_id
        self.data: Dict[str, Tuple[str, int]] = {}  # key -> (value, version)
        self.lock = threading.Lock()
        self.latency_ms = latency_ms  # ネットワーク遅延をシミュレート
 
    def write(self, key: str, value: str, version: int) -> bool:
        """データを書き込む。versionが古い場合は拒否する。"""
        if self.latency_ms > 0:
            time.sleep(self.latency_ms / 1000.0)
        with self.lock:
            current = self.data.get(key)
            if current is None or current[1] < version:
                self.data[key] = (value, version)
                return True
            return False
 
    def read(self, key: str) -> Optional[Tuple[str, int]]:
        """データを読み取る。"""
        if self.latency_ms > 0:
            time.sleep(self.latency_ms / 1000.0)
        with self.lock:
            return self.data.get(key)
 
 
class EventuallyConsistentStore:
    """結果整合性を持つ分散データストア。
 
    書き込みは1ノードに即座に反映し、他ノードには非同期で伝播する。
    なぜこの方式を採用するか:
    → 書き込みレイテンシを最小化できる
    → 一時的にノード間でデータが不一致になることを許容する
    """
 
    def __init__(self, node_count: int = 3, replication_delay_ms: float = 100):
        self.nodes = [
            Node(f"node-{i}", latency_ms=random.uniform(1, 5))
            for i in range(node_count)
        ]
        self.replication_delay_ms = replication_delay_ms
        self.version_counter = 0
        self.counter_lock = threading.Lock()
 
    def _get_next_version(self) -> int:
        with self.counter_lock:
            self.version_counter += 1
            return self.version_counter
 
    def write(self, key: str, value: str) -> str:
        """プライマリノードに書き込み、バックグラウンドで複製する。"""
        version = self._get_next_version()
        primary = self.nodes[0]
        primary.write(key, value, version)
 
        # 非同期レプリケーション（バックグラウンドスレッド）
        def replicate():
            time.sleep(self.replication_delay_ms / 1000.0)
            for node in self.nodes[1:]:
                delay = random.uniform(0, self.replication_delay_ms / 1000.0)
                time.sleep(delay)
                node.write(key, value, version)
 
        thread = threading.Thread(target=replicate, daemon=True)
        thread.start()
        return f"Written to {primary.node_id}: {key}={value} (v{version})"
 
    def read(self, key: str, node_index: Optional[int] = None) -> str:
        """指定ノード（またはランダムノード）から読み取る。"""
        if node_index is not None:
            node = self.nodes[node_index]
        else:
            node = random.choice(self.nodes)
        result = node.read(key)
        if result is None:
            return f"[{node.node_id}] {key} = (not found)"
        return f"[{node.node_id}] {key} = {result[0]} (v{result[1]})"
 
 
class StronglyConsistentStore:
    """強い一貫性（線形化可能性）を持つ分散データストア。
 
    書き込みは過半数のノードへの反映を待ってから完了とする。
    読み取りも過半数のノードから取得し、最新バージョンを返す。
 
    なぜ過半数か:
    → N=3ノード、W=2（書き込みクオラム）、R=2（読み取りクオラム）
    → W + R > N を満たすため、読み書きのクオラムが必ず重なる
    → 重なったノードが最新データを持つことが保証される
    """
 
    def __init__(self, node_count: int = 3):
        self.nodes = [
            Node(f"node-{i}", latency_ms=random.uniform(1, 5))
            for i in range(node_count)
        ]
        self.quorum = node_count // 2 + 1
        self.version_counter = 0
        self.counter_lock = threading.Lock()
 
    def _get_next_version(self) -> int:
        with self.counter_lock:
            self.version_counter += 1
            return self.version_counter
 
    def write(self, key: str, value: str) -> str:
        """クオラム数のノードに同期的に書き込む。"""
        version = self._get_next_version()
        success_count = 0
        lock = threading.Lock()
 
        def write_to_node(node: Node):
            nonlocal success_count
            if node.write(key, value, version):
                with lock:
                    success_count += 1
 
        threads = []
        for node in self.nodes:
            t = threading.Thread(target=write_to_node, args=(node,))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
 
        if success_count >= self.quorum:
            return (
                f"Written to {success_count}/{len(self.nodes)} nodes: "
                f"{key}={value} (v{version}) [COMMITTED]"
            )
        else:
            return (
                f"Write failed: only {success_count}/{len(self.nodes)} "
                f"nodes responded (need {self.quorum}) [ABORTED]"
            )
 
    def read(self, key: str) -> str:
        """クオラム数のノードから読み取り、最新バージョンを返す。"""
        results: List[Optional[Tuple[str, int]]] = []
        read_nodes: List[str] = []
        lock = threading.Lock()
 
        def read_from_node(node: Node):
            result = node.read(key)
            with lock:
                results.append(result)
                read_nodes.append(node.node_id)
 
        threads = []
        for node in self.nodes[:self.quorum]:
            t = threading.Thread(target=read_from_node, args=(node,))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
 
        valid_results = [r for r in results if r is not None]
        if not valid_results:
            return f"[quorum={read_nodes}] {key} = (not found)"
        latest = max(valid_results, key=lambda x: x[1])
        return f"[quorum={read_nodes}] {key} = {latest[0]} (v{latest[1]})"
 
 
def demo():
    """結果整合性と強い一貫性の違いをデモする。"""
    print("=" * 60)
    print("結果整合性デモ")
    print("=" * 60)
    store = EventuallyConsistentStore(node_count=3, replication_delay_ms=200)
    print(store.write("user:1:name", "Alice"))
    print("\n--- 書き込み直後（レプリケーション完了前）---")
    for i in range(3):
        print(store.read("user:1:name", node_index=i))
    print("\n--- 500ms後（レプリケーション完了後）---")
    time.sleep(0.5)
    for i in range(3):
        print(store.read("user:1:name", node_index=i))
 
    print("\n" + "=" * 60)
    print("強い一貫性デモ")
    print("=" * 60)
    store2 = StronglyConsistentStore(node_count=3)
    print(store2.write("user:1:name", "Bob"))
    print("\n--- 書き込み直後（クオラム読み取り）---")
    for _ in range(3):
        print(store2.read("user:1:name"))
 
 
if __name__ == "__main__":
    demo()

"""
Raftリーダー選出シミュレーター
==========================================
5ノードのRaftクラスタでリーダー選出が行われる様子を
スレッドベースでシミュレートする。
 
実行方法: python raft_election.py
依存: 標準ライブラリのみ
"""
 
import threading
import time
import random
from enum import Enum
from typing import Dict, Optional
 
 
class Role(Enum):
    FOLLOWER = "Follower"
    CANDIDATE = "Candidate"
    LEADER = "Leader"
 
 
class RaftNode:
    """Raftノードのリーダー選出ロジックを実装するクラス。
 
    なぜ各ノードが独自のタイムアウトを持つか:
    → 全ノードが同時にCandidateになると投票が分裂する
    → ランダムなタイムアウトにより、1つのノードが先に
      選挙を開始する確率が高くなる
    """
 
    def __init__(self, node_id: int, cluster: "RaftCluster"):
        self.node_id = node_id
        self.cluster = cluster
        self.role = Role.FOLLOWER
        self.current_term = 0
        self.voted_for: Optional[int] = None
        self.leader_id: Optional[int] = None
        self.lock = threading.Lock()
        self.election_timeout = random.uniform(0.15, 0.30)
        self.last_heartbeat = time.time()
        self.alive = True
        self.votes_received = 0
 
    def request_vote(self, candidate_id: int, term: int) -> bool:
        """投票リクエストに応答する。
 
        なぜTermの比較が必要か:
        → 古いTermのCandidateに投票すると、
          既に新しいTermで選出されたLeaderと競合する
        """
        with self.lock:
            if not self.alive:
                return False
            if term > self.current_term:
                self.current_term = term
                self.role = Role.FOLLOWER
                self.voted_for = None
            if term == self.current_term and (
                self.voted_for is None or self.voted_for == candidate_id
            ):
                self.voted_for = candidate_id
                self.last_heartbeat = time.time()
                return True
            return False
 
    def receive_heartbeat(self, leader_id: int, term: int):
        """Leaderからのハートビートを受信する。"""
        with self.lock:
            if not self.alive:
                return
            if term >= self.current_term:
                self.current_term = term
                self.role = Role.FOLLOWER
                self.leader_id = leader_id
                self.voted_for = leader_id
                self.last_heartbeat = time.time()
 
    def start_election(self):
        """選挙を開始する。"""
        with self.lock:
            if not self.alive or self.role == Role.LEADER:
                return
            self.current_term += 1
            self.role = Role.CANDIDATE
            self.voted_for = self.node_id
            self.votes_received = 1  # 自分に投票
            term = self.current_term
            print(
                f"  [Node-{self.node_id}] 選挙開始 "
                f"(Term={term})"
            )
 
        # 他の全ノードに投票をリクエスト
        for node in self.cluster.nodes.values():
            if node.node_id != self.node_id:
                if node.request_vote(self.node_id, term):
                    with self.lock:
                        self.votes_received += 1
                        if (
                            self.votes_received
                            > len(self.cluster.nodes) // 2
                            and self.role == Role.CANDIDATE
                        ):
                            self.role = Role.LEADER
                            self.leader_id = self.node_id
                            print(
                                f"  [Node-{self.node_id}] *** "
                                f"Leader当選 *** "
                                f"(Term={self.current_term}, "
                                f"得票={self.votes_received}/"
                                f"{len(self.cluster.nodes)})"
                            )
                            self._send_heartbeats()
                            return
 
    def _send_heartbeats(self):
        """全Followerにハートビートを送信する。"""
        for node in self.cluster.nodes.values():
            if node.node_id != self.node_id:
                node.receive_heartbeat(self.node_id, self.current_term)
 
    def run_election_timer(self):
        """選挙タイマーを実行する（バックグラウンドスレッド）。"""
        while self.alive:
            time.sleep(0.05)
            with self.lock:
                if not self.alive:
                    break
                if self.role == Role.LEADER:
                    continue
                elapsed = time.time() - self.last_heartbeat
                if elapsed > self.election_timeout:
                    pass
                else:
                    continue
            self.start_election()
            self.election_timeout = random.uniform(0.15, 0.30)
            with self.lock:
                self.last_heartbeat = time.time()
 
 
class RaftCluster:
    """Raftクラスタ全体を管理するクラス。"""
 
    def __init__(self, node_count: int = 5):
        self.nodes: Dict[int, RaftNode] = {}
        for i in range(node_count):
            self.nodes[i] = RaftNode(i, self)
 
    def start(self):
        """全ノードの選挙タイマーを開始する。"""
        threads = []
        for node in self.nodes.values():
            t = threading.Thread(
                target=node.run_election_timer, daemon=True
            )
            threads.append(t)
            t.start()
        return threads
 
    def kill_node(self, node_id: int):
        """ノードを停止する（障害シミュレーション）。"""
        self.nodes[node_id].alive = False
        print(f"\n  [Node-{node_id}] === 障害発生 ===")
 
    def status(self) -> str:
        """クラスタの現在の状態を返す。"""
        lines = []
        for nid, node in sorted(self.nodes.items()):
            status = "DEAD" if not node.alive else node.role.value
            leader_info = (
                f" (leader=Node-{node.leader_id})"
                if node.leader_id is not None
                else ""
            )
            lines.append(
                f"    Node-{nid}: {status:10s} "
                f"Term={node.current_term}{leader_info}"
            )
        return "\n".join(lines)
 
 
def demo():
    """リーダー選出のデモ。"""
    print("=" * 60)
    print("Raft リーダー選出シミュレーション（5ノード）")
    print("=" * 60)
 
    cluster = RaftCluster(node_count=5)
    print("\n--- 初期状態 ---")
    print(cluster.status())
 
    print("\n--- 選挙タイマー開始 ---")
    cluster.start()
    time.sleep(0.5)  # 選挙完了を待つ
 
    print("\n--- 選挙後の状態 ---")
    print(cluster.status())
 
    # Leaderを障害にする
    leader_id = None
    for nid, node in cluster.nodes.items():
        if node.role == Role.LEADER:
            leader_id = nid
            break
 
    if leader_id is not None:
        cluster.kill_node(leader_id)
        print("\n--- 新しいLeaderの選出を待機中... ---")
        time.sleep(0.8)  # 再選挙を待つ
        print("\n--- 再選挙後の状態 ---")
        print(cluster.status())
 
    # 全ノード停止
    for node in cluster.nodes.values():
        node.alive = False
 
 
if __name__ == "__main__":
    demo()

"""
コンシステントハッシュリングの実装
==========================================
仮想ノード付きのコンシステントハッシュを実装し、
ノード追加/削除時のデータ移動量を確認する。
 
実行方法: python consistent_hash.py
依存: 標準ライブラリのみ
"""
 
import hashlib
from bisect import bisect_right
from collections import defaultdict
from typing import Dict, List, Optional
 
 
class ConsistentHashRing:
    """コンシステントハッシュリングの実装。
 
    なぜbisectを使うか:
    → リング上のノード探索は二分探索で O(log N) にできる
    → 線形探索だと O(N) になり、ノード数が多いと遅い
 
    なぜMD5を使うか:
    → ハッシュ値の均一分布が重要
    → MD5は暗号学的には破られているが、
      ハッシュの均一分布性能は十分
    → SHA-256でも良いが、速度面でMD5が有利
    """
 
    def __init__(self, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: Dict[int, str] = {}  # hash -> physical_node_name
        self.sorted_keys: List[int] = []
        self.nodes: set = set()
 
    def _hash(self, key: str) -> int:
        """キーからハッシュ値を計算する。"""
        digest = hashlib.md5(key.encode()).hexdigest()
        return int(digest, 16)
 
    def add_node(self, node: str):
        """ノードをリングに追加する。
 
        仮想ノードを使う理由:
        物理ノード3台をそのまま配置すると、リング上の間隔が
        偏り、特定ノードにデータが集中する。仮想ノードで
        リング上にまんべんなく分布させることで解決する。
        """
        self.nodes.add(node)
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}:vn{i}"
            h = self._hash(virtual_key)
            self.ring[h] = node
            self.sorted_keys.append(h)
        self.sorted_keys.sort()
 
    def remove_node(self, node: str):
        """ノードをリングから削除する。"""
        self.nodes.discard(node)
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}:vn{i}"
            h = self._hash(virtual_key)
            if h in self.ring:
                del self.ring[h]
                self.sorted_keys.remove(h)
 
    def get_node(self, key: str) -> Optional[str]:
        """キーを担当するノードを返す。
 
        リング上でキーのハッシュ値から時計回りに探索し、
        最初に見つかるノードを返す。
        """
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0  # リングの末端を超えたら先頭に戻る
        return self.ring[self.sorted_keys[idx]]
 
    def get_distribution(self, keys: List[str]) -> Dict[str, int]:
        """キーの分布を計算する。"""
        dist: Dict[str, int] = defaultdict(int)
        for key in keys:
            node = self.get_node(key)
            if node:
                dist[node] += 1
        return dict(dist)
 
 
def demo():
    """コンシステントハッシュの動作デモ。"""
    print("=" * 60)
    print("コンシステントハッシュリング デモ")
    print("=" * 60)
 
    # テストデータ: 10000個のキー
    keys = [f"user:{i}" for i in range(10000)]
 
    # --- 3ノードでの分布 ---
    ring = ConsistentHashRing(virtual_nodes=150)
    for node in ["Node-A", "Node-B", "Node-C"]:
        ring.add_node(node)
 
    dist = ring.get_distribution(keys)
    print("\n--- 3ノード構成 ---")
    for node, count in sorted(dist.items()):
        bar = "#" * (count // 50)
        print(f"  {node}: {count:5d} keys {bar}")
 
    # データの配置を記録
    original_mapping = {key: ring.get_node(key) for key in keys}
 
    # --- ノード追加時の影響 ---
    ring.add_node("Node-D")
    new_mapping = {key: ring.get_node(key) for key in keys}
 
    moved = sum(
        1 for key in keys
        if original_mapping[key] != new_mapping[key]
    )
    print(f"\n--- Node-D追加後 ---")
    dist = ring.get_distribution(keys)
    for node, count in sorted(dist.items()):
        bar = "#" * (count // 50)
        print(f"  {node}: {count:5d} keys {bar}")
    print(f"\n  移動したキー数: {moved}/{len(keys)} "
          f"({moved/len(keys)*100:.1f}%)")
    print(f"  理想値: {100/4:.1f}% (1/N)")
 
    # --- ノード削除時の影響 ---
    original_mapping_4 = {key: ring.get_node(key) for key in keys}
    ring.remove_node("Node-B")
    after_removal = {key: ring.get_node(key) for key in keys}
 
    moved = sum(
        1 for key in keys
        if original_mapping_4[key] != after_removal[key]
    )
    print(f"\n--- Node-B削除後 ---")
    dist = ring.get_distribution(keys)
    for node, count in sorted(dist.items()):
        bar = "#" * (count // 50)
        print(f"  {node}: {count:5d} keys {bar}")
    print(f"\n  移動したキー数: {moved}/{len(keys)} "
          f"({moved/len(keys)*100:.1f}%)")
 
    # --- 仮想ノード数による分布の違い ---
    print(f"\n--- 仮想ノード数による分布の標準偏差 ---")
    for vn_count in [1, 10, 50, 150, 500]:
        r = ConsistentHashRing(virtual_nodes=vn_count)
        for node in ["Node-A", "Node-B", "Node-C"]:
            r.add_node(node)
        d = r.get_distribution(keys)
        values = list(d.values())
        mean = sum(values) / len(values)
        variance = sum((v - mean) ** 2 for v in values) / len(values)
        stddev = variance ** 0.5
        print(
            f"  vn={vn_count:4d}: "
            f"stddev={stddev:7.1f} "
            f"(偏りの小ささ: "
            f"{'##' * max(1, int(10 - stddev / 50))})"
        )
 
 
if __name__ == "__main__":
    demo()

"""
Sagaパターン（オーケストレーション型）の実装
==========================================
ECサイトの注文処理をSagaパターンで実装する。
障害発生時の補償トランザクションの動作を確認できる。
 
実行方法: python saga_pattern.py
依存: 標準ライブラリのみ
"""
 
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, List, Optional
import random
 
 
class StepStatus(Enum):
    PENDING = "PENDING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    COMPENSATED = "COMPENSATED"
 
 
@dataclass
class SagaStep:
    """Sagaの1ステップを表現する。
 
    なぜ正常操作と補償操作をペアにするか:
    → ステップが失敗した場合、それまでの全ステップを
      逆順に補償する必要がある
    → ペアにしておくことで、どの操作をどう取り消すか
      が明確になる
    """
    name: str
    action: Callable[[], bool]
    compensate: Callable[[], bool]
    status: StepStatus = StepStatus.PENDING
 
 
@dataclass
class SagaResult:
    success: bool
    steps_completed: int
    steps_compensated: int
    log: List[str] = field(default_factory=list)
 
 
class SagaOrchestrator:
    """Sagaオーケストレータ。
 
    ステップを順番に実行し、失敗時は逆順に補償する。
    なぜ逆順で補償するか:
    → 後のステップほど前のステップの結果に依存している
    → 依存関係を壊さないためには、後ろから取り消す必要がある
    """
 
    def __init__(self):
        self.steps: List[SagaStep] = []
        self.log: List[str] = []
 
    def add_step(
        self,
        name: str,
        action: Callable[[], bool],
        compensate: Callable[[], bool],
    ):
        """ステップを追加する。"""
        self.steps.append(SagaStep(name, action, compensate))
 
    def execute(self) -> SagaResult:
        """Sagaを実行する。"""
        completed_steps: List[SagaStep] = []
 
        self.log.append("=== Saga 実行開始 ===")
 
        for step in self.steps:
            self.log.append(f"  実行中: {step.name}")
            try:
                success = step.action()
            except Exception as e:
                self.log.append(f"  例外発生: {step.name} - {e}")
                success = False
 
            if success:
                step.status = StepStatus.COMPLETED
                completed_steps.append(step)
                self.log.append(f"  完了: {step.name}")
            else:
                step.status = StepStatus.FAILED
                self.log.append(f"  失敗: {step.name}")
                self.log.append("=== 補償トランザクション開始 ===")
 
                # 逆順に補償
                compensated = 0
                for completed in reversed(completed_steps):
                    self.log.append(f"  補償中: {completed.name}")
                    try:
                        completed.compensate()
                        completed.status = StepStatus.COMPENSATED
                        compensated += 1
                        self.log.append(
                            f"  補償完了: {completed.name}"
                        )
                    except Exception as e:
                        self.log.append(
                            f"  補償失敗: {completed.name} - {e}"
                        )
 
                self.log.append("=== Saga 失敗（ロールバック完了）===")
                return SagaResult(
                    success=False,
                    steps_completed=len(completed_steps),
                    steps_compensated=compensated,
                    log=self.log,
                )
 
        self.log.append("=== Saga 成功 ===")
        return SagaResult(
            success=True,
            steps_completed=len(completed_steps),
            steps_compensated=0,
            log=self.log,
        )
 
 
# --- ECサイト注文処理のシミュレーション ---
 
class InventoryService:
    """在庫サービス。"""
 
    def __init__(self):
        self.stock = {"item-A": 10, "item-B": 5}
        self.reserved = {}
 
    def reserve(self, item_id: str, qty: int) -> bool:
        """在庫を確保する。"""
        if self.stock.get(item_id, 0) >= qty:
            self.stock[item_id] -= qty
            self.reserved[item_id] = (
                self.reserved.get(item_id, 0) + qty
            )
            print(
                f"    [在庫] {item_id} x{qty} 確保 "
                f"(残: {self.stock[item_id]})"
            )
            return True
        print(f"    [在庫] {item_id} 在庫不足")
        return False
 
    def release(self, item_id: str, qty: int) -> bool:
        """在庫確保を取り消す。"""
        self.stock[item_id] = self.stock.get(item_id, 0) + qty
        self.reserved[item_id] = max(
            0, self.reserved.get(item_id, 0) - qty
        )
        print(
            f"    [在庫] {item_id} x{qty} 戻し "
            f"(残: {self.stock[item_id]})"
        )
        return True
 
 
class PaymentService:
    """決済サービス。"""
 
    def __init__(self, fail_probability: float = 0.0):
        self.transactions = []
        self.fail_probability = fail_probability
 
    def charge(self, user_id: str, amount: int) -> bool:
        """決済を実行する。"""
        if random.random() < self.fail_probability:
            print(f"    [決済] {user_id} ¥{amount} 失敗（残高不足）")
            return False
        self.transactions.append(
            {"user": user_id, "amount": amount, "type": "charge"}
        )
        print(f"    [決済] {user_id} ¥{amount} 成功")
        return True
 
    def refund(self, user_id: str, amount: int) -> bool:
        """返金を実行する。"""
        self.transactions.append(
            {"user": user_id, "amount": -amount, "type": "refund"}
        )
        print(f"    [決済] {user_id} ¥{amount} 返金完了")
        return True
 
 
class ShippingService:
    """配送サービス。"""
 
    def __init__(self, fail_probability: float = 0.0):
        self.shipments = []
        self.fail_probability = fail_probability
 
    def schedule(self, order_id: str, address: str) -> bool:
        """配送を手配する。"""
        if random.random() < self.fail_probability:
            print(
                f"    [配送] 注文{order_id} → {address} "
                f"手配失敗（配送業者エラー）"
            )
            return False
        self.shipments.append({"order": order_id, "address": address})
        print(f"    [配送] 注文{order_id} → {address} 手配完了")
        return True
 
    def cancel(self, order_id: str) -> bool:
        """配送をキャンセルする。"""
        self.shipments = [
            s for s in self.shipments if s["order"] != order_id
        ]
        print(f"    [配送] 注文{order_id} キャンセル完了")
        return True
 
 
def demo():
    """Sagaパターンのデモ。"""
    # --- 正常ケース ---
    print("=" * 60)
    print("Sagaパターン デモ: 正常ケース")
    print("=" * 60)
 
    inv = InventoryService()
    pay = PaymentService()
    ship = ShippingService()
 
    saga = SagaOrchestrator()
    saga.add_step(
        "在庫確保",
        lambda: inv.reserve("item-A", 2),
        lambda: inv.release("item-A", 2),
    )
    saga.add_step(
        "決済実行",
        lambda: pay.charge("user-1", 5000),
        lambda: pay.refund("user-1", 5000),
    )
    saga.add_step(
        "配送手配",
        lambda: ship.schedule("order-1", "東京都渋谷区..."),
        lambda: ship.cancel("order-1"),
    )
 
    result = saga.execute()
    print("\n".join(result.log))
 
    # --- 障害ケース（配送失敗） ---
    print("\n" + "=" * 60)
    print("Sagaパターン デモ: 配送失敗ケース")
    print("=" * 60)
 
    inv2 = InventoryService()
    pay2 = PaymentService()
    ship2 = ShippingService(fail_probability=1.0)  # 必ず失敗
 
    saga2 = SagaOrchestrator()
    saga2.add_step(
        "在庫確保",
        lambda: inv2.reserve("item-A", 2),
        lambda: inv2.release("item-A", 2),
    )
    saga2.add_step(
        "決済実行",
        lambda: pay2.charge("user-2", 3000),
        lambda: pay2.refund("user-2", 3000),
    )
    saga2.add_step(
        "配送手配",
        lambda: ship2.schedule("order-2", "大阪市北区..."),
        lambda: ship2.cancel("order-2"),
    )
 
    result2 = saga2.execute()
    print("\n".join(result2.log))
    print(f"\n  在庫状態: {inv2.stock}")
    print(f"  決済履歴: {pay2.transactions}")
 
 
if __name__ == "__main__":
    demo()

"""
ベクトル時計の実装と因果関係判定
==========================================
3プロセス間でのメッセージ送受信をシミュレートし、
ベクトル時計で因果関係を正確に判定する。
 
実行方法: python vector_clock.py
依存: 標準ライブラリのみ
"""
 
from typing import Dict, List, Tuple, Optional
from enum import Enum
from copy import deepcopy
 
 
class CausalRelation(Enum):
    """2つのイベント間の因果関係。"""
    BEFORE = "BEFORE"          # a → b（aがbの原因）
    AFTER = "AFTER"            # b → a（bがaの原因）
    CONCURRENT = "CONCURRENT"  # a || b（並行、因果関係なし）
    EQUAL = "EQUAL"            # a = b（同一イベント）
 
 
class VectorClock:
    """ベクトル時計の実装。
 
    なぜdictで実装するか:
    → プロセス数が動的に変化する場合に対応できる
    → 固定長配列だと、プロセス追加時に全ノードの
      配列を拡張する必要がある
    """
 
    def __init__(self, process_id: str):
        self.process_id = process_id
        self.clock: Dict[str, int] = {process_id: 0}
 
    def increment(self) -> "VectorClock":
        """内部イベント発生時にカウンタをインクリメント。"""
        self.clock[self.process_id] = (
            self.clock.get(self.process_id, 0) + 1
        )
        return self
 
    def send(self) -> Dict[str, int]:
        """メッセージ送信時: インクリメントしてクロックを返す。
 
        なぜ送信時にもインクリメントするか:
        → 送信イベント自体が1つのイベントだから
        → インクリメントしないと、内部イベントと送信の
          順序が区別できなくなる
        """
        self.increment()
        return deepcopy(self.clock)
 
    def receive(self, other_clock: Dict[str, int]) -> "VectorClock":
        """メッセージ受信時: 要素ごとのmaxを取る。
 
        なぜmaxを取るか:
        → 送信者が知っている全ての因果情報を取り込むため
        → maxを取ることで「送信者が見た全てのイベント」が
          受信者にも反映される
        """
        for pid, count in other_clock.items():
            self.clock[pid] = max(
                self.clock.get(pid, 0), count
            )
        self.clock[self.process_id] = (
            self.clock.get(self.process_id, 0) + 1
        )
        return self
 
    def snapshot(self) -> Dict[str, int]:
        """現在のクロック状態のスナップショットを返す。"""
        return deepcopy(self.clock)
 
    @staticmethod
    def compare(
        vc1: Dict[str, int], vc2: Dict[str, int]
    ) -> CausalRelation:
        """2つのベクトル時計の因果関係を判定する。
 
        判定ロジック:
        - 全要素で vc1[i] <= vc2[i] かつ vc1 != vc2 → BEFORE
        - 全要素で vc1[i] >= vc2[i] かつ vc1 != vc2 → AFTER
        - 上記どちらでもない → CONCURRENT
        """
        all_keys = set(vc1.keys()) | set(vc2.keys())
 
        leq = True   # vc1 <= vc2 ?
        geq = True   # vc1 >= vc2 ?
        equal = True  # vc1 == vc2 ?
 
        for key in all_keys:
            v1 = vc1.get(key, 0)
            v2 = vc2.get(key, 0)
            if v1 > v2:
                leq = False
                equal = False
            if v1 < v2:
                geq = False
                equal = False
 
        if equal:
            return CausalRelation.EQUAL
        if leq:
            return CausalRelation.BEFORE
        if geq:
            return CausalRelation.AFTER
        return CausalRelation.CONCURRENT
 
    def __repr__(self) -> str:
        items = sorted(self.clock.items())
        return "[" + ", ".join(
            f"{pid}:{count}" for pid, count in items
        ) + "]"
 
 
class DistributedSystem:
    """ベクトル時計を使った分散システムのシミュレータ。"""
 
    def __init__(self, process_ids: List[str]):
        self.processes: Dict[str, VectorClock] = {
            pid: VectorClock(pid) for pid in process_ids
        }
        self.events: List[Tuple[str, str, Dict[str, int]]] = []
 
    def local_event(self, pid: str, description: str):
        """ローカルイベントを発生させる。"""
        vc = self.processes[pid]
        vc.increment()
        snapshot = vc.snapshot()
        self.events.append((pid, description, snapshot))
        print(f"  [{pid}] {description:30s} clock={vc}")
 
    def send_message(
        self, from_pid: str, to_pid: str, description: str
    ):
        """メッセージを送受信する。"""
        sender = self.processes[from_pid]
        receiver = self.processes[to_pid]
 
        msg_clock = sender.send()
        send_snapshot = sender.snapshot()
        self.events.append(
            (from_pid, f"send({description})", send_snapshot)
        )
        print(
            f"  [{from_pid}] send({description:22s}) "
            f"clock={sender}"
        )
 
        receiver.receive(msg_clock)
        recv_snapshot = receiver.snapshot()
        self.events.append(
            (to_pid, f"recv({description})", recv_snapshot)
        )
        print(
            f"  [{to_pid}] recv({description:22s}) "
            f"clock={receiver}"
        )
 
    def analyze_causality(self):
        """全イベント間の因果関係を分析する。"""
        print("\n--- 因果関係分析 ---")
        for i in range(len(self.events)):
            for j in range(i + 1, len(self.events)):
                pid_i, desc_i, vc_i = self.events[i]
                pid_j, desc_j, vc_j = self.events[j]
                relation = VectorClock.compare(vc_i, vc_j)
                if relation != CausalRelation.EQUAL:
                    symbol = {
                        CausalRelation.BEFORE: "-->",
                        CausalRelation.AFTER: "<--",
                        CausalRelation.CONCURRENT: "|||",
                    }[relation]
                    print(
                        f"  {pid_i}:{desc_i:20s} "
                        f"{symbol} "
                        f"{pid_j}:{desc_j:20s}"
                    )
 
 
def demo():
    """ベクトル時計のデモ。"""
    print("=" * 60)
    print("ベクトル時計デモ")
    print("=" * 60)
    print()
 
    sys = DistributedSystem(["A", "B", "C"])
 
    # シナリオ: SNSの投稿とコメント
    sys.local_event("A", "Aが投稿を作成")
    sys.send_message("A", "B", "投稿を通知")
    sys.local_event("C", "Cが別の投稿を作成")
    sys.send_message("B", "C", "コメントを送信")
    sys.local_event("A", "Aが投稿を編集")
 
    sys.analyze_causality()
 
    # 競合検出のデモ
    print("\n" + "=" * 60)
    print("競合検出デモ（同時編集）")
    print("=" * 60)
    print()
 
    sys2 = DistributedSystem(["Editor-A", "Editor-B"])
    sys2.local_event("Editor-A", "ドキュメントを編集")
    sys2.local_event("Editor-B", "同じ箇所を編集")
 
    vc_a = sys2.events[0][2]
    vc_b = sys2.events[1][2]
    relation = VectorClock.compare(vc_a, vc_b)
    print(f"\n  判定: {relation.value}")
    if relation == CausalRelation.CONCURRENT:
        print("  → 競合が発生！マージまたはユーザーに選択を求める必要あり")
 
 
if __name__ == "__main__":
    demo()