1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 결제 상태를 나타내는 State 클래스
class PaymentState:
    def __init__(self):
        self.status = "created"
        self.history = []
        
# 이벤트 객체 정의(이벤트 타입, 데이터 포함)
class Event:
    def __init__(self, event_type, data):
        self.event_type = event_type
        self.data = data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import json

# 간단한 이벤트 저장소(메모리)
event_store = []

# 이벤트 저장 함수
def persist(event):
    event_store.append(json.dumps(event.__dict__))

# 이벤트 리플레이(상태 재구성)
def replay_events():
    state = PaymentState()
    for e_str in event_store:
        e = json.loads(e_str)
        state.history.append(e)
        if e['event_type'] == "pay":
            state.status = "paid"
        elif e['event_type'] == "refund":
            state.status = "refunded"
    return state

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 명령(Command)에 따른 이벤트 생성
def command_handler(command):
    if command["type"] == "pay":
        event = Event("pay", {"amount": command["amount"]})
        persist(event)
    elif command["type"] == "refund":
        event = Event("refund", {"amount": command["amount"]})
        persist(event)

# 실행 예시
command_handler({"type": "pay", "amount": 100})
command_handler({"type": "refund", "amount": 50})

# 상태 조회
state = replay_events()
print(state.status)   # "refunded"
print(state.history)  # 모든 이벤트 내역

1

# 이벤트 DB는 Cassandra 등 NoSQL 기반, persist 이벤트 → 상태 재구성, 조회용 뷰 Projection 구축

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

{
  "eventId": "uuid",
  "eventType": "OrderPlaced",
  "eventVersion": "1.0",
  "timestamp": "2024-01-15T10:30:00Z",
  "aggregateId": "order-123",
  "aggregateVersion": 1,
  "causationId": "command-456",
  "correlationId": "session-789",
  "data": {
    "customerId": "cust-001",
    "items": [...],
    "totalAmount": 150.00
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict, Any
import uuid

@dataclass
class DomainEvent:
    """모든 도메인 이벤트의 기본 클래스"""
    event_id: str
    event_type: str
    aggregate_id: str
    aggregate_version: int
    timestamp: datetime
    data: Dict[str, Any]
    
    def __post_init__(self):
        if not self.event_id:
            self.event_id = str(uuid.uuid4())
        if not self.timestamp:
            self.timestamp = datetime.utcnow()

@dataclass
class OrderPlaced(DomainEvent):
    """주문 생성 이벤트 - 고객이 주문을 완료했을 때 발생"""
    def __init__(self, aggregate_id: str, customer_id: str, items: List[Dict], total_amount: float):
        super().__init__(
            event_id="",
            event_type="OrderPlaced",
            aggregate_id=aggregate_id,
            aggregate_version=1,
            timestamp=datetime.utcnow(),
            data={
                "customer_id": customer_id,
                "items": items,
                "total_amount": total_amount
            }
        )

@dataclass
class OrderShipped(DomainEvent):
    """주문 배송 이벤트 - 주문이 배송 시작되었을 때 발생"""
    def __init__(self, aggregate_id: str, tracking_number: str, aggregate_version: int):
        super().__init__(
            event_id="",
            event_type="OrderShipped",
            aggregate_id=aggregate_id,
            aggregate_version=aggregate_version,
            timestamp=datetime.utcnow(),
            data={
                "tracking_number": tracking_number
            }
        )

@dataclass
class OrderCancelled(DomainEvent):
    """주문 취소 이벤트 - 주문이 취소되었을 때 발생"""
    def __init__(self, aggregate_id: str, reason: str, aggregate_version: int):
        super().__init__(
            event_id="",
            event_type="OrderCancelled",
            aggregate_id=aggregate_id,
            aggregate_version=aggregate_version,
            timestamp=datetime.utcnow(),
            data={
                "reason": reason
            }
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

from enum import Enum
from typing import List, Optional

class OrderStatus(Enum):
    """주문 상태 열거형 - 비즈니스 상태를 명확히 정의"""
    PLACED = "placed"
    SHIPPED = "shipped"
    CANCELLED = "cancelled"

class Order:
    """주문 애그리게이트 - 주문 관련 비즈니스 로직과 상태 관리"""
    
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.customer_id: Optional[str] = None
        self.items: List[Dict] = []
        self.total_amount: float = 0.0
        self.status: OrderStatus = None
        self.tracking_number: Optional[str] = None
        self.version: int = 0
        self._uncommitted_events: List[DomainEvent] = []
    
    @classmethod
    def create(cls, order_id: str, customer_id: str, items: List[Dict], total_amount: float) -> 'Order':
        """새로운 주문 생성 - 팩토리 메서드 패턴으로 객체 생성과 이벤트 발생을 분리"""
        order = cls(order_id)
        event = OrderPlaced(order_id, customer_id, items, total_amount)
        order._apply_event(event)
        order._uncommitted_events.append(event)
        return order
    
    def ship(self, tracking_number: str) -> None:
        """주문 배송 처리 - 비즈니스 규칙 검증 후 이벤트 발생"""
        if self.status != OrderStatus.PLACED:
            raise ValueError(f"주문 상태가 'PLACED'가 아닙니다: {self.status}")
        
        event = OrderShipped(self.order_id, tracking_number, self.version + 1)
        self._apply_event(event)
        self._uncommitted_events.append(event)
    
    def cancel(self, reason: str) -> None:
        """주문 취소 처리 - 비즈니스 규칙에 따른 취소 가능 여부 검증"""
        if self.status == OrderStatus.SHIPPED:
            raise ValueError("배송된 주문은 취소할 수 없습니다")
        
        event = OrderCancelled(self.order_id, reason, self.version + 1)
        self._apply_event(event)
        self._uncommitted_events.append(event)
    
    def _apply_event(self, event: DomainEvent) -> None:
        """이벤트를 애그리게이트에 적용하여 상태 변경"""
        if event.event_type == "OrderPlaced":
            self.customer_id = event.data["customer_id"]
            self.items = event.data["items"]
            self.total_amount = event.data["total_amount"]
            self.status = OrderStatus.PLACED
        elif event.event_type == "OrderShipped":
            self.tracking_number = event.data["tracking_number"]
            self.status = OrderStatus.SHIPPED
        elif event.event_type == "OrderCancelled":
            self.status = OrderStatus.CANCELLED
        
        self.version = event.aggregate_version
    
    def get_uncommitted_events(self) -> List[DomainEvent]:
        """저장되지 않은 이벤트 반환 - 이벤트 스토어에 저장할 이벤트 목록"""
        return self._uncommitted_events.copy()
    
    def mark_events_as_committed(self) -> None:
        """이벤트를 커밋됨으로 표시 - 저장 완료 후 호출"""
        self._uncommitted_events.clear()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class InMemoryEventStore:
    """메모리 기반 이벤트 스토어 - 실습용 간단한 구현"""
    
    def __init__(self):
        self._events: Dict[str, List[DomainEvent]] = {}
    
    def save_events(self, aggregate_id: str, events: List[DomainEvent], expected_version: int) -> None:
        """이벤트를 스토어에 저장 - 동시성 제어를 위한 버전 검증 포함"""
        if aggregate_id not in self._events:
            self._events[aggregate_id] = []
        
        current_version = len(self._events[aggregate_id])
        if current_version != expected_version:
            raise ValueError(f"동시성 충돌: 예상 버전 {expected_version}, 실제 버전 {current_version}")
        
        self._events[aggregate_id].extend(events)
        print(f"이벤트 저장 완료: {aggregate_id}, 이벤트 수: {len(events)}")
    
    def get_events(self, aggregate_id: str) -> List[DomainEvent]:
        """애그리게이트의 모든 이벤트 조회 - 상태 재구성을 위한 이벤트 스트림 반환"""
        return self._events.get(aggregate_id, [])
    
    def get_all_events(self) -> List[DomainEvent]:
        """모든 이벤트 조회 - 프로젝션 구성을 위한 전체 이벤트 스트림"""
        all_events = []
        for events in self._events.values():
            all_events.extend(events)
        return sorted(all_events, key=lambda e: e.timestamp)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class OrderRepository:
    """주문 애그리게이트 레포지토리 - 도메인과 인프라 계층 연결"""
    
    def __init__(self, event_store: InMemoryEventStore):
        self.event_store = event_store
    
    def save(self, order: Order) -> None:
        """주문 애그리게이트 저장 - 이벤트 기반 저장"""
        events = order.get_uncommitted_events()
        if events:
            expected_version = order.version - len(events)
            self.event_store.save_events(order.order_id, events, expected_version)
            order.mark_events_as_committed()
    
    def get_by_id(self, order_id: str) -> Optional[Order]:
        """ID로 주문 조회 - 이벤트 재생을 통한 상태 재구성"""
        events = self.event_store.get_events(order_id)
        if not events:
            return None
        
        order = Order(order_id)
        for event in events:
            order._apply_event(event)
        
        return order

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

class OrderSummaryProjection:
    """주문 요약 프로젝션 - 읽기 최적화된 뷰 모델"""
    
    def __init__(self):
        self.orders: Dict[str, Dict] = {}
        self.customer_orders: Dict[str, List[str]] = {}
        self.total_revenue: float = 0.0
    
    def handle_event(self, event: DomainEvent) -> None:
        """이벤트 처리를 통한 프로젝션 업데이트"""
        if event.event_type == "OrderPlaced":
            self._handle_order_placed(event)
        elif event.event_type == "OrderShipped":
            self._handle_order_shipped(event)
        elif event.event_type == "OrderCancelled":
            self._handle_order_cancelled(event)
    
    def _handle_order_placed(self, event: DomainEvent) -> None:
        """주문 생성 이벤트 처리"""
        order_summary = {
            "order_id": event.aggregate_id,
            "customer_id": event.data["customer_id"],
            "total_amount": event.data["total_amount"],
            "status": "placed",
            "item_count": len(event.data["items"]),
            "placed_at": event.timestamp
        }
        
        self.orders[event.aggregate_id] = order_summary
        
        # 고객별 주문 목록 업데이트
        customer_id = event.data["customer_id"]
        if customer_id not in self.customer_orders:
            self.customer_orders[customer_id] = []
        self.customer_orders[customer_id].append(event.aggregate_id)
        
        # 총 매출 업데이트
        self.total_revenue += event.data["total_amount"]
    
    def _handle_order_shipped(self, event: DomainEvent) -> None:
        """주문 배송 이벤트 처리"""
        if event.aggregate_id in self.orders:
            self.orders[event.aggregate_id]["status"] = "shipped"
            self.orders[event.aggregate_id]["tracking_number"] = event.data["tracking_number"]
            self.orders[event.aggregate_id]["shipped_at"] = event.timestamp
    
    def _handle_order_cancelled(self, event: DomainEvent) -> None:
        """주문 취소 이벤트 처리"""
        if event.aggregate_id in self.orders:
            order = self.orders[event.aggregate_id]
            order["status"] = "cancelled"
            order["cancelled_at"] = event.timestamp
            
            # 취소된 주문은 매출에서 제외
            self.total_revenue -= order["total_amount"]
    
    def get_order_summary(self, order_id: str) -> Optional[Dict]:
        """주문 요약 정보 조회"""
        return self.orders.get(order_id)
    
    def get_customer_orders(self, customer_id: str) -> List[Dict]:
        """고객의 모든 주문 조회"""
        order_ids = self.customer_orders.get(customer_id, [])
        return [self.orders[order_id] for order_id in order_ids if order_id in self.orders]
    
    def get_revenue_report(self) -> Dict:
        """매출 리포트 생성"""
        return {
            "total_revenue": self.total_revenue,
            "total_orders": len(self.orders),
            "orders_by_status": self._get_orders_by_status()
        }
    
    def _get_orders_by_status(self) -> Dict[str, int]:
        """상태별 주문 수 집계"""
        status_count = {}
        for order in self.orders.values():
            status = order["status"]
            status_count[status] = status_count.get(status, 0) + 1
        return status_count

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# 실습 시나리오 실행
def main():
    # 인프라 설정
    event_store = InMemoryEventStore()
    order_repo = OrderRepository(event_store)
    projection = OrderSummaryProjection()
    
    # 1. 주문 생성
    order = Order.create(
        order_id="order-001",
        customer_id="customer-123",
        items=[
            {"product_id": "product-1", "quantity": 2, "price": 50.0},
            {"product_id": "product-2", "quantity": 1, "price": 30.0}
        ],
        total_amount=130.0
    )
    order_repo.save(order)
    
    # 2. 프로젝션 업데이트
    for event in event_store.get_all_events():
        projection.handle_event(event)
    
    # 3. 주문 배송
    order.ship("TRACK-12345")
    order_repo.save(order)
    
    # 4. 프로젝션 재업데이트
    for event in event_store.get_all_events():
        projection.handle_event(event)
    
    # 결과 출력
    print("=== 주문 요약 ===")
    summary = projection.get_order_summary("order-001")
    print(f"주문 ID: {summary['order_id']}")
    print(f"상태: {summary['status']}")
    print(f"금액: ${summary['total_amount']}")
    print(f"추적번호: {summary.get('tracking_number', 'N/A')}")
    
    print("\n=== 매출 리포트 ===")
    revenue_report = projection.get_revenue_report()
    print(f"총 매출: ${revenue_report['total_revenue']}")
    print(f"총 주문수: {revenue_report['total_orders']}")
    print(f"상태별 주문: {revenue_report['orders_by_status']}")

if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

이벤트 저장 완료: order-001, 이벤트 수: 1
이벤트 저장 완료: order-001, 이벤트 수: 1
=== 주문 요약 ===
주문 ID: order-001
상태: shipped
금액: $130.0
추적번호: TRACK-12345

=== 매출 리포트 ===
총 매출: $130.0
총 주문수: 1
상태별 주문: {'shipped': 1}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# Netflix 스타일 콘텐츠 이벤트 (단순화)
class ContentMetadataUpdated:
    """콘텐츠 메타데이터 업데이트 이벤트"""
    def __init__(self, content_id: str, field: str, old_value: Any, new_value: Any, locale: str = "en-US"):
        self.content_id = content_id
        self.field = field  # title, description, genre 등
        self.old_value = old_value
        self.new_value = new_value
        self.locale = locale
        self.timestamp = datetime.utcnow()
        self.event_id = str(uuid.uuid4())

class LocalizationAdded:
    """지역화 정보 추가 이벤트"""
    def __init__(self, content_id: str, locale: str, localized_data: Dict):
        self.content_id = content_id
        self.locale = locale
        self.localized_data = localized_data
        self.timestamp = datetime.utcnow()
        self.event_id = str(uuid.uuid4())

# 검색 최적화 프로젝션
class SearchProjection:
    """Elasticsearch용 검색 최적화 프로젝션"""
    
    def __init__(self, elasticsearch_client):
        self.es = elasticsearch_client
    
    def handle_content_metadata_updated(self, event: ContentMetadataUpdated):
        """메타데이터 업데이트를 검색 인덱스에 반영"""
        doc_update = {
            "doc": {
                event.field: event.new_value,
                "last_updated": event.timestamp.isoformat()
            },
            "doc_as_upsert": True
        }
        
        # 지역별 인덱스 업데이트
        index_name = f"content-{event.locale.lower()}"
        self.es.update(
            index=index_name,
            id=event.content_id,
            body=doc_update
        )
    
    def handle_localization_added(self, event: LocalizationAdded):
        """새로운 지역화 정보를 검색 인덱스에 추가"""
        index_name = f"content-{event.locale.lower()}"
        
        # 기존 문서가 있으면 업데이트, 없으면 생성
        self.es.update(
            index=index_name,
            id=event.content_id,
            body={
                "doc": event.localized_data,
                "doc_as_upsert": True
            }
        )

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

# 은행 계좌 도메인 이벤트
class MoneyDeposited:
    """입금 이벤트 - 계좌에 돈이 입금되었을 때 발생"""
    def __init__(self, account_id: str, amount: Decimal, source: str, reference: str):
        self.account_id = account_id
        self.amount = amount
        self.source = source  # "ATM", "TRANSFER", "SALARY" 등
        self.reference = reference  # 거래 참조번호
        self.timestamp = datetime.utcnow()
        self.event_id = str(uuid.uuid4())

class MoneyWithdrawn:
    """출금 이벤트 - 계좌에서 돈이 출금되었을 때 발생"""
    def __init__(self, account_id: str, amount: Decimal, destination: str, reference: str):
        self.account_id = account_id
        self.amount = amount
        self.destination = destination
        self.reference = reference
        self.timestamp = datetime.utcnow()
        self.event_id = str(uuid.uuid4())

class TransactionBlocked:
    """거래 차단 이벤트 - 사기 의심 거래가 차단되었을 때 발생"""
    def __init__(self, account_id: str, attempted_amount: Decimal, reason: str, risk_score: float):
        self.account_id = account_id
        self.attempted_amount = attempted_amount
        self.reason = reason
        self.risk_score = risk_score
        self.timestamp = datetime.utcnow()
        self.event_id = str(uuid.uuid4())

# 계좌 애그리게이트
class BankAccount:
    """은행 계좌 애그리게이트 - 계좌의 비즈니스 로직과 상태 관리"""
    
    def __init__(self, account_id: str):
        self.account_id = account_id
        self.balance = Decimal('0.00')
        self.daily_withdrawal_limit = Decimal('1000.00')
        self.daily_withdrawn = Decimal('0.00')
        self.is_frozen = False
        self.version = 0
        self._uncommitted_events = []
        self._last_reset_date = datetime.utcnow().date()
    
    def deposit(self, amount: Decimal, source: str, reference: str) -> None:
        """입금 처리 - 비즈니스 규칙 검증 후 이벤트 발생"""
        if amount <= 0:
            raise ValueError("입금액은 0보다 커야 합니다")
        
        if self.is_frozen:
            raise ValueError("동결된 계좌에는 입금할 수 없습니다")
        
        event = MoneyDeposited(self.account_id, amount, source, reference)
        self._apply_event(event)
        self._uncommitted_events.append(event)
    
    def withdraw(self, amount: Decimal, destination: str, reference: str, fraud_detector) -> None:
        """출금 처리 - 잔액, 일일 한도, 사기 탐지 검증"""
        if amount <= 0:
            raise ValueError("출금액은 0보다 커야 합니다")
        
        if self.is_frozen:
            raise ValueError("동결된 계좌에서는 출금할 수 없습니다")
        
        if self.balance < amount:
            raise ValueError("잔액이 부족합니다")
        
        # 일일 출금 한도 검증 (날짜가 바뀌면 리셋)
        today = datetime.utcnow().date()
        if today > self._last_reset_date:
            self.daily_withdrawn = Decimal('0.00')
            self._last_reset_date = today
        
        if self.daily_withdrawn + amount > self.daily_withdrawal_limit:
            raise ValueError("일일 출금 한도를 초과했습니다")
        
        # 사기 탐지 검증
        risk_score = fraud_detector.calculate_risk(self.account_id, amount, destination)
        if risk_score > 0.8:  # 80% 이상 위험도면 차단
            block_event = TransactionBlocked(self.account_id, amount, "High fraud risk", risk_score)
            self._apply_event(block_event)
            self._uncommitted_events.append(block_event)
            raise ValueError(f"보안상의 이유로 거래가 차단되었습니다 (위험도: {risk_score:.2f})")
        
        # 정상 출금 처리
        event = MoneyWithdrawn(self.account_id, amount, destination, reference)
        self._apply_event(event)
        self._uncommitted_events.append(event)
    
    def _apply_event(self, event) -> None:
        """이벤트를 계좌 상태에 적용"""
        if isinstance(event, MoneyDeposited):
            self.balance += event.amount
        elif isinstance(event, MoneyWithdrawn):
            self.balance -= event.amount
            self.daily_withdrawn += event.amount
        elif isinstance(event, TransactionBlocked):
            # 차단 이벤트는 상태 변경 없음 (로깅 목적)
            pass
        
        self.version += 1

# 실시간 사기 탐지 프로젝션
class FraudDetectionProjection:
    """사기 탐지를 위한 실시간 프로젝션"""
    
    def __init__(self):
        self.account_patterns = {}  # 계좌별 거래 패턴
        self.suspicious_destinations = set()  # 의심스러운 목적지
    
    def handle_money_withdrawn(self, event: MoneyWithdrawn):
        """출금 이벤트 처리 - 거래 패턴 학습"""
        account_id = event.account_id
        
        if account_id not in self.account_patterns:	        
	        self.account_patterns[account_id] = {
	            'daily_withdrawals': 0,
	            'total_amount': Decimal('0.00'),
	            'destinations': set(),
	            'last_withdrawal': None
	        }
        
        pattern = self.account_patterns[account_id]
        pattern['daily_withdrawals'] += 1
        pattern['total_amount'] += event.amount
        pattern['destinations'].add(event.destination)
        pattern['last_withdrawal'] = event.timestamp
        
        # 의심 패턴 탐지
        if pattern['daily_withdrawals'] > 5:  # 하루 5회 이상 출금
            self._flag_suspicious_activity(account_id, "High frequency withdrawals")
        
        if event.amount > Decimal('5000.00'):  # 대액 출금
            self._flag_suspicious_activity(account_id, "Large amount withdrawal")
    
    def handle_transaction_blocked(self, event: TransactionBlocked):
        """차단된 거래 패턴 학습"""
        # 차단된 목적지를 의심 목록에 추가
        # 실제로는 더 정교한 ML 알고리즘 적용
        pass
    
    def _flag_suspicious_activity(self, account_id: str, reason: str):
        """의심 활동 플래그"""
        print(f"[FRAUD ALERT] 계좌 {account_id}: {reason}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

class OrderSaga:
    """주문 처리 사가 - 여러 서비스 간 분산 트랜잭션 조율"""
    
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.state = {}
    
    def handle_order_placed(self, event):
        """1단계: 주문 생성 → 결제 요청"""
        order_id = event.aggregate_id
        self.state[order_id] = {"step": "payment_requested"}
        
        payment_command = RequestPayment(
            order_id=order_id,
            amount=event.data["total_amount"],
            customer_id=event.data["customer_id"]
        )
        self.event_bus.send_command("payment-service", payment_command)
    
    def handle_payment_completed(self, event):
        """2단계: 결제 완료 → 배송 요청"""
        order_id = event.data["order_id"]
        if order_id in self.state:
            self.state[order_id]["step"] = "shipping_requested"
            
            shipping_command = RequestShipping(
                order_id=order_id,
                address=event.data["shipping_address"]
            )
            self.event_bus.send_command("shipping-service", shipping_command)
    
    def handle_payment_failed(self, event):
        """보상 트랜잭션: 결제 실패 → 주문 취소"""
        order_id = event.data["order_id"]
        
        cancel_command = CancelOrder(
            order_id=order_id,
            reason="Payment failed"
        )
        self.event_bus.send_command("order-service", cancel_command)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38