최종 일관성

# 최종 일관성 시스템에서의 충돌 해결 전략
from datetime import datetime
from dataclasses import dataclass
from typing import List, Any
import hashlib

# 1. Last-Write-Wins (LWW) - 가장 간단
@dataclass
class LWWValue:
    value: Any
    timestamp: datetime

    def merge(self, other: 'LWWValue') -> 'LWWValue':
        """타임스탬프가 더 최신인 값을 선택"""
        return self if self.timestamp > other.timestamp else other

# 사용 예시
replica1 = LWWValue("value_A", datetime(2024, 1, 1, 10, 0))
replica2 = LWWValue("value_B", datetime(2024, 1, 1, 10, 5))
resolved = replica1.merge(replica2)  # "value_B" 선택


# 2. G-Counter CRDT (Conflict-free Replicated Data Type)
# 증가만 가능한 카운터 - 최종 일관성에서도 충돌 없이 병합
class GCounter:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counts = {}  # {node_id: count}

    def increment(self, amount: int = 1):
        # 자신의 노드 카운트만 증가
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount

    def value(self) -> int:
        # 모든 노드 카운트의 합
        return sum(self.counts.values())

    def merge(self, other: 'GCounter') -> 'GCounter':
        """각 노드의 max 값을 취해 병합 - 충돌 없음"""
        merged = GCounter(self.node_id)
        all_nodes = set(self.counts.keys()) | set(other.counts.keys())
        for node in all_nodes:
            merged.counts[node] = max(
                self.counts.get(node, 0),
                other.counts.get(node, 0)
            )
        return merged

# 사용 예시: 좋아요 카운터
node_a = GCounter("node-a")
node_b = GCounter("node-b")

node_a.increment(5)   # 노드 A에서 5번 클릭
node_b.increment(3)   # 노드 B에서 3번 클릭

# 네트워크 복구 후 병합 - 순서 무관, 멱등성 보장
merged = node_a.merge(node_b)
print(f"총 좋아요: {merged.value()}")  # 8


# 3. Vector Clock - 인과관계 추적
class VectorClock:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.clock = {}

    def tick(self):
        self.clock[self.node_id] = self.clock.get(self.node_id, 0) + 1

    def compare(self, other: 'VectorClock') -> str:
        """인과관계 판단: 'before', 'after', 'concurrent'"""
        less = equal = greater = False
        all_keys = set(self.clock.keys()) | set(other.clock.keys())
        for key in all_keys:
            v1 = self.clock.get(key, 0)
            v2 = other.clock.get(key, 0)
            if v1 < v2: less = True
            elif v1 > v2: greater = True
            else: equal = True

        if less and not greater: return 'before'
        if greater and not less: return 'after'
        return 'concurrent'  # 충돌 발생!

# Saga 패턴: 분산 트랜잭션에서의 최종 일관성
# 각 서비스는 로컬 트랜잭션 + 보상 트랜잭션으로 일관성 유지
from enum import Enum
from dataclasses import dataclass
from typing import Callable, List
import asyncio

class SagaState(Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"

@dataclass
class SagaStep:
    name: str
    action: Callable           # 실행할 작업
    compensation: Callable     # 롤백 작업

class OrderSaga:
    """주문 처리 Saga 예시"""

    def __init__(self, order_id: str):
        self.order_id = order_id
        self.state = SagaState.PENDING
        self.completed_steps: List[SagaStep] = []

    async def execute(self):
        steps = [
            SagaStep(
                name="재고 확보",
                action=lambda: self._reserve_inventory(),
                compensation=lambda: self._release_inventory()
            ),
            SagaStep(
                name="결제 처리",
                action=lambda: self._process_payment(),
                compensation=lambda: self._refund_payment()
            ),
            SagaStep(
                name="배송 예약",
                action=lambda: self._schedule_delivery(),
                compensation=lambda: self._cancel_delivery()
            ),
        ]

        try:
            for step in steps:
                print(f"[Saga] 실행: {step.name}")
                await step.action()
                self.completed_steps.append(step)

            self.state = SagaState.COMPLETED
            print(f"[Saga] 완료: 주문 {self.order_id}")

        except Exception as e:
            print(f"[Saga] 실패: {e}")
            await self._compensate()

    async def _compensate(self):
        """완료된 단계를 역순으로 롤백"""
        self.state = SagaState.COMPENSATING
        for step in reversed(self.completed_steps):
            try:
                print(f"[Saga] 보상: {step.name} 롤백")
                await step.compensation()
            except Exception as e:
                print(f"[Saga] 보상 실패: {e}")
        self.state = SagaState.FAILED

    # 각 단계 구현 (실제로는 마이크로서비스 호출)
    async def _reserve_inventory(self):
        # 재고 서비스에 재고 확보 요청
        await asyncio.sleep(0.1)  # 시뮬레이션

    async def _release_inventory(self):
        # 재고 서비스에 재고 해제 요청
        await asyncio.sleep(0.1)

    async def _process_payment(self):
        # 결제 서비스에 결제 요청
        await asyncio.sleep(0.1)
        # raise Exception("결제 실패")  # 테스트용

    async def _refund_payment(self):
        # 결제 서비스에 환불 요청
        await asyncio.sleep(0.1)

    async def _schedule_delivery(self):
        # 배송 서비스에 배송 예약
        await asyncio.sleep(0.1)

    async def _cancel_delivery(self):
        # 배송 서비스에 배송 취소
        await asyncio.sleep(0.1)

# 실행
async def main():
    saga = OrderSaga("order-12345")
    await saga.execute()

asyncio.run(main())