from dataclasses import dataclass, field
from typing import Optional, Set, Dict
import time
 
@dataclass
class Node:
    name: str
    data: dict
    reachable: set  # 通信可能なノード名の集合
 
class DistributedStore:
    """ネットワーク分断を再現する分散ストア"""
 
    def __init__(self):
        self.nodes = {
            "node1": Node("node1", {"key": "value_v1"}, {"node2", "node3"}),
            "node2": Node("node2", {"key": "value_v1"}, {"node1", "node3"}),
            "node3": Node("node3", {"key": "value_v1"}, {"node1", "node2"}),
        }
 
    def partition(self, group_a: set, group_b: set):
        """ネットワーク分断を発生させる"""
        print(f"[PARTITION] {group_a} <-X-> {group_b}")
        for name in group_a:
            self.nodes[name].reachable -= group_b
        for name in group_b:
            self.nodes[name].reachable -= group_a
 
    def heal(self):
        """分断を解消する"""
        all_names = set(self.nodes.keys())
        for name, node in self.nodes.items():
            node.reachable = all_names - {name}
        print("[HEAL] ネットワーク分断を解消")
 
    def write_cp(self, node_name: str, key: str, value: str) -> bool:
        """CP方式: 過半数に書き込めなければ拒否（クォーラム）"""
        node = self.nodes[node_name]
        reachable_count = 1 + len(node.reachable)  # 自分自身 + 到達可能ノード
        quorum = len(self.nodes) // 2 + 1
 
        if reachable_count >= quorum:
            # 過半数に到達可能 → 書き込み成功
            node.data[key] = value
            for peer_name in node.reachable:
                self.nodes[peer_name].data[key] = value
            print(f"[CP] 書き込み成功 ({reachable_count}/{len(self.nodes)} >= quorum {quorum})")
            return True
        else:
            # 過半数に到達不可 → 書き込み拒否（一貫性を優先）
            print(f"[CP] 書き込み拒否 ({reachable_count}/{len(self.nodes)} < quorum {quorum})")
            return False
 
    def write_ap(self, node_name: str, key: str, value: str) -> bool:
        """AP方式: 到達可能なノードにのみ書き込み、常に成功"""
        node = self.nodes[node_name]
        node.data[key] = value
        for peer_name in node.reachable:
            self.nodes[peer_name].data[key] = value
        unreachable = set(self.nodes.keys()) - {node_name} - node.reachable
        if unreachable:
            print(f"[AP] 書き込み成功（{unreachable} は古いデータのまま）")
        else:
            print(f"[AP] 書き込み成功（全ノード同期済み）")
        return True
 
    def read_all(self, key: str) -> dict:
        """全ノードのデータを表示"""
        result = {}
        for name, node in self.nodes.items():
            result[name] = node.data.get(key, "NOT_FOUND")
        return result
 
 
# === デモ ===
print("=== ネットワーク分断のシミュレーション ===\n")
 
# 初期状態
store = DistributedStore()
print(f"初期状態: {store.read_all('key')}")
 
# 分断を発生させる
store.partition({"node1", "node2"}, {"node3"})
 
print("\n--- CP方式 ---")
store.write_cp("node1", "key", "value_v2")  # 成功 (2/3 >= 2)
store.write_cp("node3", "key", "value_v3")  # 拒否 (1/3 < 2)
print(f"データ状態: {store.read_all('key')}")
# → node1,node2 = "value_v2", node3 = "value_v1"（拒否されたので古いまま）
 
print("\n--- AP方式 ---")
store2 = DistributedStore()
store2.partition({"node1", "node2"}, {"node3"})
store2.write_ap("node1", "key", "value_v2")  # 成功 (node3は古い)
store2.write_ap("node3", "key", "value_v3")  # 成功 (node1,2とは異なる値)
print(f"データ状態: {store2.read_all('key')}")
# → node1,node2 = "value_v2", node3 = "value_v3" → 矛盾（コンフリクト）！

import time
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional, Dict, List
 
@dataclass
class VersionedEntry:
    """バージョン付きデータエントリ"""
    value: str
    vector_clock: Dict[str, int]
    timestamp: float
    node_id: str
 
class EventuallyConsistentStore:
    """結果整合性を持つAPストア（ゴシッププロトコルベース）
 
    WHY ゴシッププロトコル:
      全ノードに一斉ブロードキャストすると、ネットワーク負荷が O(N^2) になる。
      ゴシッププロトコルでは各ノードがランダムに選んだ隣接ノードに情報を伝播
      するため、O(N log N) ラウンドで全ノードに伝わる（疫学モデルに由来）。
    """
 
    def __init__(self, node_id: str, peers: list = None):
        self.node_id = node_id
        self.data: Dict[str, VersionedEntry] = {}
        self.vector_clock: Dict[str, int] = defaultdict(int)
        self.peers: List['EventuallyConsistentStore'] = peers or []
        self.gossip_interval = 1.0
 
    def write(self, key: str, value: str):
        """ローカルに書き込み、バックグラウンドで同期"""
        self.vector_clock[self.node_id] += 1
        entry = VersionedEntry(
            value=value,
            vector_clock=dict(self.vector_clock),
            timestamp=time.time(),
            node_id=self.node_id,
        )
        self.data[key] = entry
        print(f"[{self.node_id}] Write: {key}={value} "
              f"clock={dict(self.vector_clock)}")
 
    def read(self, key: str) -> Optional[str]:
        """ローカルデータを即座に返す（古い可能性あり）"""
        entry = self.data.get(key)
        if entry:
            return entry.value
        return None
 
    def merge(self, key: str, remote_entry: VersionedEntry):
        """リモートデータとマージ（Last-Writer-Wins）
 
        WHY LWW:
          最も単純なコンフリクト解決戦略。タイムスタンプが新しい方を採用する。
          問題点: 時計のずれ（clock skew）により不正確になる可能性がある。
          代替: ベクタークロック比較、CRDTs、アプリケーション固有のマージロジック。
        """
        local_entry = self.data.get(key)
        if local_entry is None:
            self.data[key] = remote_entry
            return
 
        # ベクタークロックで因果関係を判定
        relation = self._compare_clocks(local_entry.vector_clock,
                                        remote_entry.vector_clock)
        if relation == "before":
            # ローカルが古い → リモートで上書き
            self.data[key] = remote_entry
            print(f"[{self.node_id}] Merge: {key} updated to {remote_entry.value}")
        elif relation == "concurrent":
            # 並行書き込み → LWW で解決
            if remote_entry.timestamp > local_entry.timestamp:
                self.data[key] = remote_entry
                print(f"[{self.node_id}] Merge (LWW): {key} = {remote_entry.value}")
 
    def _compare_clocks(self, clock_a: dict, clock_b: dict) -> str:
        """ベクタークロックの比較
 
        Returns: "before" | "after" | "concurrent"
        """
        all_keys = set(clock_a.keys()) | set(clock_b.keys())
        a_lte_b = all(clock_a.get(k, 0) <= clock_b.get(k, 0) for k in all_keys)
        b_lte_a = all(clock_b.get(k, 0) <= clock_a.get(k, 0) for k in all_keys)
 
        if a_lte_b and not b_lte_a:
            return "before"   # A は B より前
        elif b_lte_a and not a_lte_b:
            return "after"    # A は B より後
        else:
            return "concurrent"  # 並行（因果関係なし）
 
    def gossip(self):
        """ゴシッププロトコルで隣接ノードにデータを伝播"""
        for peer in self.peers:
            for key, entry in self.data.items():
                peer.merge(key, entry)
 
 
# === デモ ===
node_a = EventuallyConsistentStore("nodeA")
node_b = EventuallyConsistentStore("nodeB")
node_a.peers = [node_b]
node_b.peers = [node_a]
 
# 正常時: 書き込み → ゴシップで同期
node_a.write("user:1", "Alice")
node_a.gossip()
print(f"nodeB read: {node_b.read('user:1')}")  # → "Alice"
 
# 分断時: 両ノードが独立に書き込み
node_a.peers = []  # 分断シミュレーション
node_b.peers = []
node_a.write("user:1", "Alice_updated_by_A")
time.sleep(0.01)  # わずかな時間差
node_b.write("user:1", "Alice_updated_by_B")
 
# 分断解消 → ゴシップでマージ
node_a.peers = [node_b]
node_b.peers = [node_a]
node_a.gossip()
node_b.gossip()
# LWW により timestamp が新しい方が勝つ

import time
from typing import Optional, List, Dict
 
class QuorumStore:
    """クォーラムベースの読み書き (Dynamo/Cassandra風)
 
    WHY クォーラム:
      N個のレプリカに対して、W個の書き込み確認 + R個の読み込み確認を要求する。
      W + R > N のとき、読み書きの「重なり」が保証され、
      最新のデータを必ず読める（強い一貫性）。
 
    トレードオフ:
      W=1, R=1 → 最速だが一貫性なし（結果整合性）
      W=N, R=1 → 書き込みが遅いが読み取りが速い
      W=1, R=N → 書き込みが速いが読み取りが遅い
      W=⌊N/2⌋+1, R=⌊N/2⌋+1 → バランス（最も一般的）
    """
 
    def __init__(self, n: int, w: int, r: int):
        self.n = n
        self.w = w
        self.r = r
        self.replicas: List[Dict] = [{} for _ in range(n)]
        self.is_strong = (w + r) > n
        print(f"Quorum: N={n}, W={w}, R={r}")
        print(f"Strong consistency: W+R > N → {w}+{r} > {n} = {self.is_strong}")
 
    def write(self, key: str, value: str) -> bool:
        """Wノードに書き込み成功で完了"""
        version = time.time()
        success = 0
        for i, replica in enumerate(self.replicas):
            replica[key] = {"value": value, "version": version}
            success += 1
            if success >= self.w:
                remaining = self.n - success
                print(f"Write OK: {key}={value} "
                      f"(ack: {success}/{self.n}, async: {remaining})")
                # 残りのレプリカは非同期で伝播
                return True
        return False
 
    def read(self, key: str) -> Optional[str]:
        """Rノードから読み込み、最新版を返す"""
        responses = []
        for i, replica in enumerate(self.replicas):
            entry = replica.get(key)
            if entry:
                responses.append(entry)
            if len(responses) >= self.r:
                break
 
        if not responses:
            return None
 
        # 最新のバージョンを返す（Read Repair の基礎）
        latest = max(responses, key=lambda x: x["version"])
        return latest["value"]
 
    def read_repair(self, key: str):
        """Read Repair: 読み取り時に古いレプリカを更新
 
        WHY Read Repair:
          クォーラム読み取りで最新値を取得した後、古い値を持つ
          レプリカを最新値で更新する。これにより、結果整合性が
          より速く収束する。Cassandraの重要な機能。
        """
        # 全レプリカから読み取り
        entries = [(i, r.get(key)) for i, r in enumerate(self.replicas)]
        valid = [(i, e) for i, e in entries if e is not None]
 
        if not valid:
            return
 
        # 最新バージョンを特定
        latest_idx, latest_entry = max(valid, key=lambda x: x[1]["version"])
 
        # 古いレプリカを更新
        for i, entry in valid:
            if entry["version"] < latest_entry["version"]:
                self.replicas[i][key] = latest_entry
                print(f"[Read Repair] Replica {i}: "
                      f"{entry['value']} → {latest_entry['value']}")
 
 
# === デモ ===
print("=== 強い一貫性: W + R > N ===")
strong = QuorumStore(n=3, w=2, r=2)  # 2+2 > 3 → True
strong.write("x", "100")
print(f"Read: {strong.read('x')}")
 
print("\n=== 結果整合性: W + R <= N ===")
eventual = QuorumStore(n=3, w=1, r=1)  # 1+1 > 3 → False
eventual.write("x", "200")
print(f"Read: {eventual.read('x')}")
 
print("\n=== 書き込み重視: W=1, R=N ===")
write_fast = QuorumStore(n=3, w=1, r=3)  # 1+3 > 3 → True
write_fast.write("x", "300")
print(f"Read: {write_fast.read('x')}")

from enum import Enum
 
class ConsistencyLevel(Enum):
    ONE = 1           # 1ノードの応答で完了（最速・最弱）
    QUORUM = "quorum" # 過半数の応答で完了（バランス）
    ALL = "all"       # 全ノードの応答で完了（最強・最遅）
    LOCAL_QUORUM = "local_quorum"  # ローカルDC内の過半数
    EACH_QUORUM = "each_quorum"    # 各DC内の過半数
 
class CassandraClient:
    """Cassandra の一貫性レベル分析ツール
 
    WHY クエリごとに一貫性レベルを変えるのか:
      Cassandra はAPシステムだが、クエリ単位で一貫性レベルを指定できる。
      これにより、同じクラスタ内で「口座残高はQUORUM読み書き」
      「アクセスログはONE書き込み」と使い分けられる。
    """
 
    def __init__(self, replication_factor: int = 3):
        self.rf = replication_factor
 
    def required_responses(self, level: ConsistencyLevel) -> int:
        if level == ConsistencyLevel.ONE:
            return 1
        elif level == ConsistencyLevel.QUORUM:
            return self.rf // 2 + 1
        elif level == ConsistencyLevel.ALL:
            return self.rf
        elif level == ConsistencyLevel.LOCAL_QUORUM:
            return self.rf // 2 + 1
        elif level == ConsistencyLevel.EACH_QUORUM:
            return self.rf // 2 + 1
 
    def is_strongly_consistent(self, write_cl: ConsistencyLevel,
                                read_cl: ConsistencyLevel) -> bool:
        """W + R > N なら強い一貫性"""
        w = self.required_responses(write_cl)
        r = self.required_responses(read_cl)
        return (w + r) > self.rf
 
    def analyze(self, write_cl: ConsistencyLevel, read_cl: ConsistencyLevel):
        w = self.required_responses(write_cl)
        r = self.required_responses(read_cl)
        strong = self.is_strongly_consistent(write_cl, read_cl)
        latency = "低" if w == 1 and r == 1 else "中" if w + r <= self.rf + 1 else "高"
        print(f"W={write_cl.name}({w}) + R={read_cl.name}({r}) "
              f"{'>' if strong else '<='} N={self.rf} "
              f"→ {'強い一貫性' if strong else '結果整合性'} "
              f"(レイテンシ: {latency})")
 
    def recommend(self, use_case: str):
        """ユースケースに応じた推奨設定"""
        recommendations = {
            "balance": (ConsistencyLevel.QUORUM, ConsistencyLevel.QUORUM,
                       "口座残高: 強い一貫性が必須"),
            "log": (ConsistencyLevel.ONE, ConsistencyLevel.ONE,
                   "アクセスログ: レイテンシ重視、多少の損失は許容"),
            "session": (ConsistencyLevel.LOCAL_QUORUM, ConsistencyLevel.LOCAL_QUORUM,
                       "セッション: DCローカルの一貫性で十分"),
            "config": (ConsistencyLevel.ALL, ConsistencyLevel.ONE,
                      "設定データ: 全レプリカに確実に書き込み、読み取りは速く"),
        }
        if use_case in recommendations:
            w_cl, r_cl, description = recommendations[use_case]
            print(f"\n{description}")
            self.analyze(w_cl, r_cl)
 
 
# === デモ ===
client = CassandraClient(replication_factor=3)
 
print("=== 一貫性レベルの分析 ===")
client.analyze(ConsistencyLevel.ONE, ConsistencyLevel.ONE)
# W=ONE(1) + R=ONE(1) <= N=3 → 結果整合性 (レイテンシ: 低)
 
client.analyze(ConsistencyLevel.QUORUM, ConsistencyLevel.QUORUM)
# W=QUORUM(2) + R=QUORUM(2) > N=3 → 強い一貫性 (レイテンシ: 中)
 
client.analyze(ConsistencyLevel.ALL, ConsistencyLevel.ONE)
# W=ALL(3) + R=ONE(1) > N=3 → 強い一貫性 (レイテンシ: 高)
 
print("\n=== ユースケース別推奨 ===")
client.recommend("balance")
client.recommend("log")
client.recommend("session")

from dataclasses import dataclass
from typing import Any, Callable
import time
 
@dataclass
class VersionedValue:
    value: Any
    timestamp: float
    node_id: str
    vector_clock: dict
 
class ConflictResolver:
    """分断解消後のコンフリクト解決戦略
 
    WHY 複数の解決戦略が必要か:
      データの性質によって最適な解決方法が異なる。
      - 価格データ → LWW（最新値が正しい）
      - ショッピングカート → マージ（両方の追加を保持）
      - カウンター → CRDTs（数学的にマージ可能な構造）
      - 注文データ → アプリケーション固有のロジック
    """
 
    @staticmethod
    def last_writer_wins(v1: VersionedValue, v2: VersionedValue) -> VersionedValue:
        """LWW: タイムスタンプが新しい方を採用
 
        メリット: シンプル、実装が容易
        デメリット: 時計のずれに脆弱、データが静かに失われる
        用途: セッション、キャッシュ、最終更新状態
        """
        winner = v1 if v1.timestamp > v2.timestamp else v2
        print(f"LWW: {winner.value} wins (from {winner.node_id})")
        return winner
 
    @staticmethod
    def merge_values(v1: VersionedValue, v2: VersionedValue) -> VersionedValue:
        """マージ: 両方の値を保持（ショッピングカート等で有用）
 
        メリット: データの損失がない
        デメリット: 削除操作の扱いが難しい（Tombstone が必要）
        用途: ショッピングカート、タグ、お気に入りリスト
        """
        if isinstance(v1.value, set) and isinstance(v2.value, set):
            merged = v1.value | v2.value
            print(f"Merge: {v1.value} | {v2.value} = {merged}")
            return VersionedValue(merged, time.time(), "merged", {})
        raise ValueError("Merge not supported for non-set values")
 
    @staticmethod
    def higher_value_wins(v1: VersionedValue, v2: VersionedValue) -> VersionedValue:
        """数値が大きい方を採用（カウンター等で有用）
 
        用途: monotonically increasing カウンター（いいね数等）
        """
        if isinstance(v1.value, (int, float)) and isinstance(v2.value, (int, float)):
            winner = v1 if v1.value >= v2.value else v2
            print(f"Higher wins: {winner.value}")
            return winner
        raise ValueError("Numeric values required")
 
    @staticmethod
    def application_level(v1: VersionedValue, v2: VersionedValue,
                          resolver: Callable) -> VersionedValue:
        """アプリケーション固有のロジックで解決
 
        用途: 複雑なビジネスルール（注文状態の遷移等）
        """
        return resolver(v1, v2)
 
 
# === デモ ===
v1 = VersionedValue("price=100", 1000.001, "node1", {"node1": 1})
v2 = VersionedValue("price=120", 1000.005, "node2", {"node2": 1})
ConflictResolver.last_writer_wins(v1, v2)
# LWW: price=120 wins (from node2)
 
cart1 = VersionedValue({"itemA", "itemB"}, 1000.0, "node1", {})
cart2 = VersionedValue({"itemB", "itemC"}, 1000.0, "node2", {})
ConflictResolver.merge_values(cart1, cart2)
# Merge: {'itemA', 'itemB'} | {'itemB', 'itemC'} = {'itemA', 'itemB', 'itemC'}
 
counter1 = VersionedValue(42, 1000.0, "node1", {})
counter2 = VersionedValue(45, 1000.0, "node2", {})
ConflictResolver.higher_value_wins(counter1, counter2)
# Higher wins: 45

from collections import defaultdict
 
class GCounter:
    """G-Counter (Grow-only Counter) - CRDT
 
    WHY CRDTs:
      コンフリクト解決を「データ構造レベル」で保証する手法。
      特別な調停ロジックなしに、任意の順序でマージしても
      全ノードが同じ値に収束する（数学的に証明済み）。
 
    G-Counter の仕組み:
      各ノードが自分のカウンターを持ち、インクリメントは
      自分のカウンターのみ行う。合計は全ノードのカウンターの和。
      マージは各ノードの max を取る → 順序に依存せず収束。
    """
 
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counters: dict[str, int] = defaultdict(int)
 
    def increment(self, amount: int = 1):
        """自分のカウンターのみインクリメント"""
        self.counters[self.node_id] += amount
 
    def value(self) -> int:
        """全ノードのカウンターの合計"""
        return sum(self.counters.values())
 
    def merge(self, other: 'GCounter'):
        """マージ: 各ノードの max を採用（順序に依存しない）"""
        all_keys = set(self.counters.keys()) | set(other.counters.keys())
        for key in all_keys:
            self.counters[key] = max(
                self.counters.get(key, 0),
                other.counters.get(key, 0)
            )
 
    def __repr__(self):
        return f"GCounter({dict(self.counters)}, total={self.value()})"
 
 
class PNCounter:
    """PN-Counter (Positive-Negative Counter) - CRDT
 
    G-Counter の拡張。増加用と減少用の2つのG-Counterを持つ。
    値 = P.value() - N.value()
    """
 
    def __init__(self, node_id: str):
        self.p = GCounter(node_id)  # 増加用
        self.n = GCounter(node_id)  # 減少用
 
    def increment(self, amount: int = 1):
        self.p.increment(amount)
 
    def decrement(self, amount: int = 1):
        self.n.increment(amount)
 
    def value(self) -> int:
        return self.p.value() - self.n.value()
 
    def merge(self, other: 'PNCounter'):
        self.p.merge(other.p)
        self.n.merge(other.n)
 
 
# === デモ ===
print("=== G-Counter (いいね数) ===")
node_a = GCounter("A")
node_b = GCounter("B")
 
node_a.increment(3)  # Aで3回いいね
node_b.increment(5)  # Bで5回いいね
 
# 分断中 → 各ノードは独立にカウント
print(f"A: {node_a}")  # GCounter({'A': 3}, total=3)
print(f"B: {node_b}")  # GCounter({'B': 5}, total=5)
 
# 分断解消 → マージ
node_a.merge(node_b)
node_b.merge(node_a)
print(f"After merge A: {node_a}")  # GCounter({'A': 3, 'B': 5}, total=8)
print(f"After merge B: {node_b}")  # GCounter({'A': 3, 'B': 5}, total=8)
# → 順序に関わらず、両方が 8 に収束する
 
print("\n=== PN-Counter (在庫数) ===")
stock_a = PNCounter("A")
stock_b = PNCounter("B")
 
stock_a.increment(100)  # 100個入荷
stock_a.decrement(3)    # 3個販売（Aで）
stock_b.decrement(5)    # 5個販売（Bで）
 
stock_a.merge(stock_b)
stock_b.merge(stock_a)
print(f"在庫: {stock_a.value()}")  # 92 (100 - 3 - 5)

# NG: 「CAを選ぶ」という誤った判断
class BadDistributedSystem:
    """
    誤解: 「うちはCAを選ぶ。一貫性と可用性を取って分断耐性は捨てる」
 
    問題:
    1. 分散システムでネットワーク分断は避けられない
    2. 分断耐性を「捨てる」ことはできない（物理的に不可避）
    3. CAは「分散でないシステム」（単一ノード）にしか成立しない
    """
    def __init__(self):
        # 単一ノードなら CA は成立するが、それは分散システムではない
        self.single_db = "postgresql://single-server:5432/db"
 
# OK: 正しい理解
class GoodDistributedSystem:
    """
    正しい理解:
    1. P は選択肢ではなく前提条件（分断は必ず発生する）
    2. 分断時: CP（一貫性優先）or AP（可用性優先）を選ぶ
    3. さらに、通常時も L vs C のトレードオフが存在する（PACELC）
    4. データの性質ごとに使い分けるのがベストプラクティス
    """
    def __init__(self):
        self.balance_db = "mongodb://..."       # CP: 口座残高
        self.session_cache = "cassandra://..."  # AP: セッション
        self.log_store = "cassandra://..."      # AP: ログ

# NG: 金融システムだから全データを強い一貫性で管理
class BadFinancialSystem:
    def __init__(self):
        # 全データに QUORUM 読み書き → レイテンシが全体的に悪化
        self.consistency_level = "QUORUM"
 
    def get_balance(self, user_id):
        # OK: 残高は強い一貫性が必要
        return self.read(f"balance:{user_id}", cl="QUORUM")
 
    def get_user_preferences(self, user_id):
        # NG: ユーザー設定にまで QUORUM は不要
        return self.read(f"prefs:{user_id}", cl="QUORUM")
 
    def write_access_log(self, log_entry):
        # NG: アクセスログにまで QUORUM は不要（レイテンシ悪化）
        return self.write("access_log", log_entry, cl="QUORUM")
 
 
# OK: データの性質ごとに一貫性レベルを使い分け
class GoodFinancialSystem:
    """
    データ分類:

"""
    def get_balance(self, user_id):
        return self.read(f"balance:{user_id}", cl="QUORUM")
 
    def get_user_preferences(self, user_id):
        return self.read(f"prefs:{user_id}", cl="ONE")
 
    def write_access_log(self, log_entry):
        return self.write("access_log", log_entry, cl="ONE")

"""
シナリオ: ECサイトのショッピングカート
 
分断中に2つのノードで同じユーザーのカートが更新された:
- Node A: ユーザーが「商品X」を追加、「商品Y」を削除
- Node B: ユーザーが「商品Z」を追加
 
分断解消後にどうマージするか？
 
要件:
1. 追加された商品は全て保持する（Union）
2. 削除された商品は正しく反映する（Tombstone）
3. 数量変更はLWWで解決する
"""
 
# ここに実装する

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]