이벤트 드리븐 아키텍처 + CQRS + Saga 패턴: MSA 실전 설계 가이드

graph LR
    A[주문 서비스] -->|HTTP 동기 호출| B[결제 서비스]
    B -->|HTTP 동기 호출| C[재고 서비스]
    C -->|HTTP 동기 호출| D[배송 서비스]
    style A fill:#f96,stroke:#333
    style D fill:#6f9,stroke:#333

graph LR
    A[주문 서비스] -->|이벤트 발행| K[Kafka]
    K -->|이벤트 소비| B[결제 서비스]
    K -->|이벤트 소비| C[재고 서비스]
    K -->|이벤트 소비| D[배송 서비스]
    style K fill:#ff9,stroke:#333

graph TB
    Client -->|Command: 주문 생성| CmdAPI[Command API]
    Client -->|Query: 주문 조회| QryAPI[Query API]

    CmdAPI --> WriteDB[(Write DB<br/>PostgreSQL)]
    WriteDB -->|이벤트 발행| EventBus[Kafka]
    EventBus -->|이벤트 소비| Projector[Projector]
    Projector --> ReadDB[(Read DB<br/>Elasticsearch)]
    QryAPI --> ReadDB

    style CmdAPI fill:#f96,stroke:#333
    style QryAPI fill:#6f9,stroke:#333
    style EventBus fill:#ff9,stroke:#333

# Command Side
from dataclasses import dataclass
from datetime import datetime
import uuid

@dataclass
class CreateOrderCommand:
    customer_id: str
    items: list[dict]
    total_amount: float

class OrderCommandHandler:
    def __init__(self, repo, event_publisher):
        self.repo = repo
        self.event_publisher = event_publisher

    def handle_create_order(self, cmd: CreateOrderCommand):
        order = Order(
            id=str(uuid.uuid4()),
            customer_id=cmd.customer_id,
            items=cmd.items,
            total_amount=cmd.total_amount,
            status="PENDING",
            created_at=datetime.utcnow()
        )

        # Write DB에 저장
        self.repo.save(order)

        # 이벤트 발행
        self.event_publisher.publish("order.created", {
            "order_id": order.id,
            "customer_id": order.customer_id,
            "total_amount": order.total_amount,
            "items": order.items,
        })
        return order.id

# Query Side
class OrderQueryService:
    def __init__(self, read_repo):
        self.read_repo = read_repo  # Elasticsearch

    def get_order(self, order_id: str):
        return self.read_repo.find_by_id(order_id)

    def search_orders(self, customer_id: str, status: str = None):
        return self.read_repo.search(customer_id=customer_id, status=status)

# Projector (이벤트 → Read Model 동기화)
class OrderProjector:
    def __init__(self, read_repo):
        self.read_repo = read_repo

    def on_order_created(self, event):
        self.read_repo.upsert({
            "id": event["order_id"],
            "customer_id": event["customer_id"],
            "total_amount": event["total_amount"],
            "status": "PENDING",
            "items": event["items"],
        })

    def on_order_paid(self, event):
        self.read_repo.update(event["order_id"], {"status": "PAID"})

graph TB
    subgraph "Choreography (이벤트 기반)"
        O1[주문] -->|OrderCreated| P1[결제]
        P1 -->|PaymentCompleted| S1[재고]
        S1 -->|StockReserved| D1[배송]
        D1 -->|DeliveryStarted| O1
    end

graph TB
    subgraph "Orchestration (중앙 조율)"
        Orch[Saga Orchestrator]
        Orch -->|1. 결제 요청| P2[결제]
        P2 -->|결제 완료| Orch
        Orch -->|2. 재고 예약| S2[재고]
        S2 -->|예약 완료| Orch
        Orch -->|3. 배송 요청| D2[배송]
        D2 -->|배송 시작| Orch
    end

from enum import Enum
from kafka import KafkaProducer, KafkaConsumer
import json

class SagaStep(Enum):
    PAYMENT = "payment"
    INVENTORY = "inventory"
    DELIVERY = "delivery"

class OrderSagaOrchestrator:
    STEPS = [SagaStep.PAYMENT, SagaStep.INVENTORY, SagaStep.DELIVERY]

    def __init__(self, producer: KafkaProducer):
        self.producer = producer
        self.saga_state = {}  # order_id -> {step, status}

    def start_saga(self, order_id: str, order_data: dict):
        self.saga_state[order_id] = {
            "current_step": 0,
            "data": order_data,
            "completed_steps": [],
        }
        self._execute_step(order_id)

    def _execute_step(self, order_id: str):
        state = self.saga_state[order_id]
        step_idx = state["current_step"]

        if step_idx >= len(self.STEPS):
            # 모든 단계 완료
            self._publish(f"order.completed", {"order_id": order_id})
            return

        step = self.STEPS[step_idx]
        self._publish(f"{step.value}.request", {
            "order_id": order_id,
            "saga_id": order_id,
            **state["data"]
        })

    def handle_step_success(self, order_id: str, step: SagaStep):
        state = self.saga_state[order_id]
        state["completed_steps"].append(step)
        state["current_step"] += 1
        self._execute_step(order_id)

    def handle_step_failure(self, order_id: str, failed_step: SagaStep):
        """보상 트랜잭션 실행 (역순)"""
        state = self.saga_state[order_id]

        for step in reversed(state["completed_steps"]):
            self._publish(f"{step.value}.compensate", {
                "order_id": order_id,
            })

        self._publish("order.failed", {
            "order_id": order_id,
            "failed_at": failed_step.value,
        })

    def _publish(self, topic: str, data: dict):
        self.producer.send(topic, json.dumps(data).encode())

sequenceDiagram
    participant O as Orchestrator
    participant P as 결제 서비스
    participant I as 재고 서비스
    participant D as 배송 서비스

    O->>P: 1. 결제 요청
    P-->>O: 결제 성공
    O->>I: 2. 재고 예약
    I-->>O: 재고 예약 성공
    O->>D: 3. 배송 요청
    D-->>O: ❌ 배송 실패 (주소 오류)

    Note over O: 보상 트랜잭션 시작 (역순)
    O->>I: 재고 예약 취소
    I-->>O: 취소 완료
    O->>P: 결제 환불
    P-->>O: 환불 완료
    O->>O: 주문 실패 처리

전통적 방식:
  orders 테이블: {id: 1, status: "SHIPPED", amount: 50000}

Event Sourcing:
  events 테이블:
    1. OrderCreated {amount: 50000}
    2. PaymentReceived {payment_id: "pay_123"}
    3. StockReserved {warehouse: "seoul"}
    4. OrderShipped {tracking: "KR12345"}

  → 이벤트를 순서대로 리플레이하면 현재 상태 복원

class EventStore:
    def __init__(self, db):
        self.db = db

    def append(self, aggregate_id: str, event_type: str, data: dict, version: int):
        self.db.execute("""
            INSERT INTO events (aggregate_id, event_type, data, version, created_at)
            VALUES (%s, %s, %s, %s, NOW())
        """, (aggregate_id, event_type, json.dumps(data), version))

    def get_events(self, aggregate_id: str, after_version: int = 0):
        return self.db.query("""
            SELECT event_type, data, version
            FROM events
            WHERE aggregate_id = %s AND version > %s
            ORDER BY version
        """, (aggregate_id, after_version))

class Order:
    def __init__(self):
        self.id = None
        self.status = None
        self.version = 0
        self._pending_events = []

    def apply_event(self, event_type: str, data: dict):
        if event_type == "OrderCreated":
            self.id = data["order_id"]
            self.status = "PENDING"
        elif event_type == "PaymentReceived":
            self.status = "PAID"
        elif event_type == "OrderShipped":
            self.status = "SHIPPED"
        self.version += 1

    @classmethod
    def from_events(cls, events):
        order = cls()
        for e in events:
            order.apply_event(e["event_type"], e["data"])
        return order

graph TB
    Client[클라이언트] --> Gateway[API Gateway]

    Gateway --> OrderCmd[주문 Command API]
    Gateway --> OrderQry[주문 Query API]

    OrderCmd --> WriteDB[(PostgreSQL)]
    WriteDB --> CDC[Debezium CDC]
    CDC --> Kafka[Apache Kafka]

    Kafka --> Saga[Saga Orchestrator]
    Kafka --> Projector[CQRS Projector]

    Saga --> PaymentSvc[결제 서비스]
    Saga --> InventorySvc[재고 서비스]
    Saga --> DeliverySvc[배송 서비스]

    Projector --> ReadDB[(Elasticsearch)]
    OrderQry --> ReadDB

    PaymentSvc --> Kafka
    InventorySvc --> Kafka
    DeliverySvc --> Kafka

    style Kafka fill:#ff9,stroke:#333
    style Saga fill:#f96,stroke:#333

# 이벤트 중복 처리 방지
class IdempotentConsumer:
    def __init__(self, redis_client):
        self.redis = redis_client

    def process_event(self, event_id: str, handler):
        key = f"processed:{event_id}"
        if self.redis.setnx(key, "1"):
            self.redis.expire(key, 86400)  # 24시간 TTL
            handler()
        else:
            pass  # 이미 처리됨, 스킵

# Kafka 파티션 키로 순서 보장
producer.send(
    "order.events",
    key=order_id.encode(),  # 같은 주문 → 같은 파티션 → 순서 보장
    value=json.dumps(event).encode()
)