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

# 카트에 대한 이벤트 생성과 처리 예제
from typing import List

class Event:
    """이벤트 소싱의 핵심: 상태 변화를 기록하는 이벤트 객체"""
    pass

class AddItemEvent(Event):
    """상품 추가 이벤트"""
    def __init__(self, item):
        self.item = item

class RemoveItemEvent(Event):
    """상품 제거 이벤트"""
    def __init__(self, item):
        self.item = item

class Cart:
    """카트 상태는 이벤트 Replay로 도출"""
    def __init__(self):
        self.items = []
        self.events: List[Event] = []
    
    def apply(self, event: Event):
        """이벤트를 적용하여 상태 변경"""
        if isinstance(event, AddItemEvent):
            self.items.append(event.item)
        elif isinstance(event, RemoveItemEvent):
            self.items.remove(event.item)
        self.events.append(event)
    
    def replay(self, events: List[Event]):
        """저장된 모든 이벤트를 순서대로 적용"""
        self.items = []
        for event in events:
            self.apply(event)

# 사용 예시
cart = Cart()
cart.apply(AddItemEvent('Kimchi'))
cart.apply(RemoveItemEvent('Kimchi'))
print(cart.items)  # 이벤트 Replay 후 현재 상태

 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

# 결제 시스템 핵심 상태 변화를 이벤트로 관리
class PaymentEvent(Event):
    """결제 이벤트 기본 클래스"""
    pass

class PaymentCreated(PaymentEvent): ...
class PaymentApproved(PaymentEvent): ...
class PaymentRefunded(PaymentEvent): ...

class PaymentOrder:
    """결제건 상태를 이벤트로 관리"""
    def __init__(self):
        self.state = "Created"
        self.events: List[PaymentEvent] = []
    
    def apply(self, event: PaymentEvent):
        if isinstance(event, PaymentCreated):
            self.state = "Created"
        elif isinstance(event, PaymentApproved):
            self.state = "Approved"
        elif isinstance(event, PaymentRefunded):
            self.state = "Refunded"
        self.events.append(event)
    
    def replay(self, events: List[PaymentEvent]):
        self.state = None
        for event in events:
            self.apply(event)

# 예시 적용
order = PaymentOrder()
order.apply(PaymentCreated())
order.apply(PaymentApproved())
print(order.state)  # "Approved"

  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

import json
from typing import List, Dict, Any

# ----- 이벤트 정의 -----
class Event:
    """이벤트 소싱의 핵심: 상태 변화를 기록하는 불변 객체"""
    def to_dict(self):
        return self.__dict__

class AddItemEvent(Event):
    def __init__(self, item):
        self.type = "ADD_ITEM"
        self.item = item

class RemoveItemEvent(Event):
    def __init__(self, item):
        self.type = "REMOVE_ITEM"
        self.item = item

# ----- 저장소 구현 -----
class EventStore:
    """이벤트 저장소 - 모든 상태 변화를 기록"""
    def __init__(self):
        self.events: List[Event] = []

    def save(self, event: Event):
        self.events.append(event)

    def get_events_since(self, index: int) -> List[Event]:
        """지정 인덱스 이후의 이벤트 반환"""
        return self.events[index:]

class SnapshotStore:
    """스냅샷 저장소 - 특정 시점의 전체 상태 저장"""
    def __init__(self):
        self.snapshots: Dict[int, Any] = {}  # key: 이벤트 인덱스

    def save_snapshot(self, index: int, state: Any):
        self.snapshots[index] = json.dumps(state)  # JSON 직렬화 저장

    def get_latest_snapshot(self):
        if not self.snapshots:
            return None, None
        latest_index = max(self.snapshots.keys())
        return latest_index, json.loads(self.snapshots[latest_index])  # JSON 역직렬화 반환

# ----- Cart Aggregate -----
class Cart:
    """Cart 상태는 이벤트 Replay 또는 Snapshot + Replay로 복원"""
    def __init__(self):
        self.items: List[str] = []

    def apply(self, event: Event):
        """이벤트를 적용하여 상태 변경"""
        if event.type == "ADD_ITEM":
            self.items.append(event.item)
        elif event.type == "REMOVE_ITEM":
            if event.item in self.items:
                self.items.remove(event.item)

    def get_state(self):
        return {"items": self.items}

    def set_state(self, state: Dict):
        self.items = state["items"]

# ----- 서비스 로직 (핵심 비즈니스 + 스냅샷) -----
class CartService:
    SNAPSHOT_THRESHOLD = 5  # N개 이벤트마다 스냅샷 저장

    def __init__(self, event_store: EventStore, snapshot_store: SnapshotStore):
        self.event_store = event_store
        self.snapshot_store = snapshot_store
        self.cart = Cart()

    def handle_event(self, event: Event):
        self.event_store.save(event)
        self.cart.apply(event)

        # 스냅샷 조건 충족 시 저장
        if len(self.event_store.events) % self.SNAPSHOT_THRESHOLD == 0:
            self.snapshot_store.save_snapshot(len(self.event_store.events), self.cart.get_state())
            print(f"[스냅샷 저장] {len(self.event_store.events)}개 이벤트 시점")

    def restore_state(self):
        """스냅샷 + 남은 이벤트로 상태 복원"""
        last_snapshot_index, snapshot_state = self.snapshot_store.get_latest_snapshot()
        if snapshot_state:
            self.cart.set_state(snapshot_state)
            start_index = last_snapshot_index
        else:
            start_index = 0

        remaining_events = self.event_store.get_events_since(start_index)
        for ev in remaining_events:
            self.cart.apply(ev)

        print(f"[복원 완료] 총 {len(remaining_events)}개의 이벤트 재생")

# ----- 테스트 시나리오 -----
event_store = EventStore()
snapshot_store = SnapshotStore()
service = CartService(event_store, snapshot_store)

# 이벤트 발생
service.handle_event(AddItemEvent("사과"))
service.handle_event(AddItemEvent("바나나"))
service.handle_event(RemoveItemEvent("사과"))
service.handle_event(AddItemEvent("키위"))
service.handle_event(AddItemEvent("포도"))  # 여기서 스냅샷 저장
service.handle_event(AddItemEvent("수박"))

# 상태 확인
print("[현재 상태]", service.cart.items)

# 복구 테스트
new_service = CartService(event_store, snapshot_store)
new_service.restore_state()
print("[복구된 상태]", new_service.cart.items)

 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

from kafka import KafkaProducer
import json
from typing import List, Dict

# ----- 이벤트 정의 -----
class Event:
    def to_dict(self):
        return self.__dict__

class AddItemEvent(Event):
    def __init__(self, item):
        self.type = "ADD_ITEM"
        self.item = item

class RemoveItemEvent(Event):
    def __init__(self, item):
        self.type = "REMOVE_ITEM"
        self.item = item

# ----- 저장소 -----
class EventStore:
    def __init__(self):
        self.events: List[Event] = []
    def save(self, event: Event):
        self.events.append(event)
    def get_events(self):
        return self.events

class SnapshotStore:
    def __init__(self):
        self.snapshots: Dict[int, Dict] = {}
    def save_snapshot(self, index: int, state: Dict):
        self.snapshots[index] = state
    def get_latest_snapshot(self):
        if not self.snapshots:
            return None, None
        idx = max(self.snapshots.keys())
        return idx, self.snapshots[idx]

# ----- Cart Aggregate -----
class Cart:
    def __init__(self):
        self.items = []
    def apply(self, event: Event):
        if event.type == "ADD_ITEM":
            self.items.append(event.item)
        elif event.type == "REMOVE_ITEM" and event.item in self.items:
            self.items.remove(event.item)
    def get_state(self):
        return {"items": self.items}
    def set_state(self, state: Dict):
        self.items = state["items"]

# ----- Cart Service -----
class CartService:
    SNAPSHOT_THRESHOLD = 5
    def __init__(self, event_store, snapshot_store):
        self.cart = Cart()
        self.event_store = event_store
        self.snapshot_store = snapshot_store
        self.producer = KafkaProducer(
            bootstrap_servers='localhost:9092',
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
    def handle_event(self, event: Event):
        self.event_store.save(event)
        self.cart.apply(event)
        # Kafka 발행
        self.producer.send('cart-events', event.to_dict())
        print(f"[Kafka 발행] {event.to_dict()}")
        # 스냅샷 조건 검사
        if len(self.event_store.events) % self.SNAPSHOT_THRESHOLD == 0:
            self.snapshot_store.save_snapshot(len(self.event_store.events), self.cart.get_state())
            print(f"[스냅샷 저장] {self.cart.get_state()}")

# 테스트 실행
if __name__ == "__main__":
    event_store = EventStore()
    snapshot_store = SnapshotStore()
    service = CartService(event_store, snapshot_store)

    service.handle_event(AddItemEvent("사과"))
    service.handle_event(AddItemEvent("바나나"))
    service.handle_event(RemoveItemEvent("사과"))
    service.handle_event(AddItemEvent("키위"))
    service.handle_event(AddItemEvent("포도"))   # 스냅샷 저장 시점

 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

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'cart-events',
    bootstrap_servers='localhost:9092',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='analytics-service'
)

print("[Analytics] Kafka 구독 시작")
item_counts = {}

for message in consumer:
    event = message.value
    if event["type"] == "ADD_ITEM":
        item = event["item"]
        item_counts[item] = item_counts.get(item, 0) + 1
    elif event["type"] == "REMOVE_ITEM":
        item = event["item"]
        if item in item_counts:
            item_counts[item] -= 1
            if item_counts[item] <= 0:
                del item_counts[item]
    print("[Analytics] 현재 집계:", item_counts)

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

# 주문 도메인의 이벤트 모델링
class OrderEvent:
    def __init__(self, aggregate_id, timestamp, user_id):
        self.aggregate_id = aggregate_id
        self.timestamp = timestamp
        self.user_id = user_id

class OrderPlaced(OrderEvent):
    def __init__(self, aggregate_id, timestamp, user_id, items, total_amount):
        super().__init__(aggregate_id, timestamp, user_id)
        self.items = items
        self.total_amount = total_amount

class OrderShipped(OrderEvent):
    def __init__(self, aggregate_id, timestamp, user_id, tracking_number):
        super().__init__(aggregate_id, timestamp, user_id)
        self.tracking_number = tracking_number

 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

# PostgreSQL 기반 이벤트 스토어 구현
class PostgreSQLEventStore:
    def __init__(self, connection):
        self.connection = connection
    
    def append_events(self, aggregate_id, expected_version, events):
        """이벤트를 append-only 방식으로 저장"""
        cursor = self.connection.cursor()
        try:
            # 낙관적 동시성 제어
            cursor.execute(
                "SELECT version FROM aggregates WHERE id = %s FOR UPDATE",
                (aggregate_id,)
            )
            current_version = cursor.fetchone()
            
            if current_version and current_version[0] != expected_version:
                raise ConcurrencyError("Aggregate version mismatch")
            
            # 이벤트 저장
            for i, event in enumerate(events):
                cursor.execute("""
                    INSERT INTO events (aggregate_id, version, event_type, event_data, timestamp)
                    VALUES (%s, %s, %s, %s, %s)
                """, (
                    aggregate_id,
                    expected_version + i + 1,
                    event.__class__.__name__,
                    json.dumps(event.__dict__),
                    datetime.utcnow()
                ))
            
            # 집계 버전 업데이트
            new_version = expected_version + len(events)
            cursor.execute("""
                INSERT INTO aggregates (id, version) VALUES (%s, %s)
                ON CONFLICT (id) DO UPDATE SET version = %s
            """, (aggregate_id, new_version, new_version))
            
            self.connection.commit()
        except Exception as e:
            self.connection.rollback()
            raise e
    
    def get_events(self, aggregate_id, from_version=0):
        """집계의 이벤트 시퀀스 조회"""
        cursor = self.connection.cursor()
        cursor.execute("""
            SELECT event_type, event_data, version, timestamp
            FROM events
            WHERE aggregate_id = %s AND version > %s
            ORDER BY version
        """, (aggregate_id, from_version))
        
        events = []
        for row in cursor.fetchall():
            event_type, event_data, version, timestamp = row
            # 이벤트 객체 재구성 (event_type을 통한 동적 생성)
            event_class = globals()[event_type]
            event_dict = json.loads(event_data)
            event = event_class(**event_dict)
            events.append(event)
        
        return events

 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

# 주문 프로젝션 구현
class OrderProjection:
    def __init__(self, read_db):
        self.read_db = read_db
        self.event_handlers = {
            'OrderPlaced': self.handle_order_placed,
            'OrderShipped': self.handle_order_shipped,
            'OrderCancelled': self.handle_order_cancelled
        }
    
    def handle_event(self, event):
        """이벤트 타입에 따른 적절한 핸들러 호출"""
        handler = self.event_handlers.get(event.__class__.__name__)
        if handler:
            handler(event)
    
    def handle_order_placed(self, event):
        """주문 생성 이벤트 처리"""
        self.read_db.execute("""
            INSERT INTO order_summary (
                order_id, user_id, status, total_amount, created_at
            ) VALUES (%s, %s, %s, %s, %s)
        """, (
            event.aggregate_id,
            event.user_id,
            'PLACED',
            event.total_amount,
            event.timestamp
        ))
        
        # 주문 항목 저장
        for item in event.items:
            self.read_db.execute("""
                INSERT INTO order_items (order_id, product_id, quantity, price)
                VALUES (%s, %s, %s, %s)
            """, (event.aggregate_id, item.product_id, item.quantity, item.price))
    
    def handle_order_shipped(self, event):
        """주문 배송 이벤트 처리"""
        self.read_db.execute("""
            UPDATE order_summary 
            SET status = 'SHIPPED', tracking_number = %s, shipped_at = %s
            WHERE order_id = %s
        """, (event.tracking_number, event.timestamp, event.aggregate_id))
    
    def handle_order_cancelled(self, event):
        """주문 취소 이벤트 처리"""
        self.read_db.execute("""
            UPDATE order_summary 
            SET status = 'CANCELLED', cancelled_at = %s
            WHERE order_id = %s
        """, (event.timestamp, event.aggregate_id))

 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

# 스냅샷 기반 집계 로딩
class SnapshotRepository:
    def __init__(self, snapshot_store, event_store):
        self.snapshot_store = snapshot_store
        self.event_store = event_store
    
    def load_aggregate(self, aggregate_id):
        """스냅샷과 이후 이벤트를 결합하여 집계 로딩"""
        # 최신 스냅샷 로드
        snapshot = self.snapshot_store.get_latest_snapshot(aggregate_id)
        
        if snapshot:
            aggregate = self.deserialize_aggregate(snapshot.data)
            from_version = snapshot.version
        else:
            aggregate = Order(aggregate_id)  # 새 집계 생성
            from_version = 0
        
        # 스냅샷 이후 이벤트 적용
        events = self.event_store.get_events(aggregate_id, from_version)
        for event in events:
            aggregate.apply_event(event)
        
        return aggregate
    
    def save_aggregate(self, aggregate):
        """집계 저장 및 스냅샷 생성 결정"""
        # 새 이벤트 저장
        new_events = aggregate.get_uncommitted_events()
        self.event_store.append_events(
            aggregate.id,
            aggregate.version - len(new_events),
            new_events
        )
        
        # 스냅샷 생성 조건 확인 (예: 100개 이벤트마다)
        if aggregate.version % 100 == 0:
            snapshot_data = self.serialize_aggregate(aggregate)
            self.snapshot_store.save_snapshot(
                aggregate.id,
                aggregate.version,
                snapshot_data
            )
        
        aggregate.mark_events_as_committed()

  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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

# 주문 집계 (Order Aggregate)
class Order:
    def __init__(self, order_id):
        self.id = order_id  # 집계 식별자
        self.version = 0    # 낙관적 동시성 제어를 위한 버전
        self.status = None
        self.items = []
        self.total_amount = 0
        self.uncommitted_events = []  # 저장되지 않은 이벤트 목록
    
    def place_order(self, user_id, items):
        """주문 생성 비즈니스 로직"""
        if self.status is not None:
            raise ValueError("Order already exists")
        
        # 비즈니스 규칙 검증
        if not items or len(items) == 0:
            raise ValueError("Order must have at least one item")
        
        total = sum(item.price * item.quantity for item in items)
        
        # 도메인 이벤트 생성
        event = OrderPlaced(
            aggregate_id=self.id,
            timestamp=datetime.utcnow(),
            user_id=user_id,
            items=items,
            total_amount=total
        )
        
        # 이벤트 적용 및 미커밋 목록에 추가
        self.apply_event(event)
        self.uncommitted_events.append(event)
    
    def ship_order(self, tracking_number):
        """주문 배송 처리"""
        if self.status != 'PLACED':
            raise ValueError("Cannot ship order in current status")
        
        event = OrderShipped(
            aggregate_id=self.id,
            timestamp=datetime.utcnow(),
            user_id=self.user_id,
            tracking_number=tracking_number
        )
        
        self.apply_event(event)
        self.uncommitted_events.append(event)
    
    def apply_event(self, event):
        """이벤트를 집계에 적용하여 상태 변경"""
        if isinstance(event, OrderPlaced):
            self.status = 'PLACED'
            self.user_id = event.user_id
            self.items = event.items
            self.total_amount = event.total_amount
        elif isinstance(event, OrderShipped):
            self.status = 'SHIPPED'
            self.tracking_number = event.tracking_number
        elif isinstance(event, OrderCancelled):
            self.status = 'CANCELLED'
        
        self.version += 1
    
    def get_uncommitted_events(self):
        """저장되지 않은 이벤트 목록 반환"""
        return self.uncommitted_events.copy()
    
    def mark_events_as_committed(self):
        """이벤트가 저장되었음을 표시"""
        self.uncommitted_events.clear()

# Command Handler 구현
class OrderCommandHandler:
    def __init__(self, repository, event_bus):
        self.repository = repository  # Order Repository
        self.event_bus = event_bus    # Event Bus for publishing
    
    def handle_place_order(self, command):
        """주문 생성 명령 처리"""
        # 새로운 주문 집계 생성
        order = Order(command.order_id)
        
        # 비즈니스 로직 실행
        order.place_order(command.user_id, command.items)
        
        # 집계 저장 (이벤트 저장소에 이벤트 저장)
        self.repository.save(order)
        
        # 이벤트 발행
        for event in order.get_uncommitted_events():
            self.event_bus.publish(event)
        
        order.mark_events_as_committed()
        
        return order.id
    
    def handle_ship_order(self, command):
        """주문 배송 명령 처리"""
        # 기존 주문 로드
        order = self.repository.get_by_id(command.order_id)
        
        # 비즈니스 로직 실행
        order.ship_order(command.tracking_number)
        
        # 변경사항 저장
        self.repository.save(order)
        
        # 이벤트 발행
        for event in order.get_uncommitted_events():
            self.event_bus.publish(event)
        
        order.mark_events_as_committed()

# Repository 구현 (Event Store 인터페이스)
class OrderRepository:
    def __init__(self, event_store):
        self.event_store = event_store
    
    def get_by_id(self, order_id):
        """주문 ID로 집계 로드 (이벤트 재생)"""
        events = self.event_store.get_events(order_id)
        
        if not events:
            raise ValueError(f"Order {order_id} not found")
        
        order = Order(order_id)
        for event in events:
            order.apply_event(event)
        
        return order
    
    def save(self, order):
        """집계의 미커밋 이벤트를 저장"""
        uncommitted_events = order.get_uncommitted_events()
        if uncommitted_events:
            expected_version = order.version - len(uncommitted_events)
            self.event_store.append_events(
                order.id, 
                expected_version, 
                uncommitted_events
            )

# Event Bus 구현 (Redis 기반)
class RedisEventBus:
    def __init__(self, redis_client):
        self.redis_client = redis_client
    
    def publish(self, event):
        """이벤트를 Redis 채널에 발행"""
        channel = f"events.{event.__class__.__name__}"
        event_data = {
            'event_type': event.__class__.__name__,
            'aggregate_id': event.aggregate_id,
            'timestamp': event.timestamp.isoformat(),
            'data': event.__dict__
        }
        
        self.redis_client.publish(channel, json.dumps(event_data))
    
    def subscribe(self, event_type, handler):
        """특정 이벤트 타입에 대한 핸들러 등록"""
        channel = f"events.{event_type}"
        pubsub = self.redis_client.pubsub()
        pubsub.subscribe(channel)
        
        for message in pubsub.listen():
            if message['type'] == 'message':
                event_data = json.loads(message['data'])
                handler(event_data)

  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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

// 시청 이벤트 정의
public class ViewingEvent {
    private String userId;
    private String contentId;
    private long timestamp;
    private int watchDurationSeconds;
    private String deviceType;
    
    // Netflix 특화: 시청 품질 메트릭
    private double bufferingRatio;
    private String videoQuality;
    
    // 생성자, getter/setter 생략
}

// Kafka Streams를 활용한 실시간 집계
@Service
public class ViewingAnalyticsProcessor {
    
    @Autowired
    private StreamsBuilder streamsBuilder;
    
    @PostConstruct
    public void buildTopology() {
        // 시청 이벤트 스트림 생성
        KStream<String, ViewingEvent> viewingEvents = 
            streamsBuilder.stream("viewing-events");
        
        // 사용자별 실시간 집계
        KTable<String, UserViewingStats> userStats = viewingEvents
            .filter((key, event) -> event.getWatchDurationSeconds() > 30) // 30초 이상 시청만
            .groupBy((key, event) -> event.getUserId()) // 사용자별 그룹화
            .aggregate(
                UserViewingStats::new, // 초기값
                (userId, event, stats) -> {
                    // 시청 통계 업데이트
                    stats.addWatchTime(event.getWatchDurationSeconds());
                    stats.addContentView(event.getContentId());
                    stats.updateLastActivity(event.getTimestamp());
                    
                    // Netflix 특화: 시청 품질 추적
                    stats.updateQualityMetrics(
                        event.getBufferingRatio(), 
                        event.getVideoQuality()
                    );
                    
                    return stats;
                },
                Materialized.<String, UserViewingStats, KeyValueStore<Bytes, byte[]>>as("user-stats-store")
                    .withValueSerde(Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(UserViewingStats.class)))
            );
        
        // 콘텐츠별 인기도 실시간 계산
        KTable<String, ContentPopularity> contentPopularity = viewingEvents
            .filter((key, event) -> event.getWatchDurationSeconds() > 60) // 1분 이상 시청
            .groupBy((key, event) -> event.getContentId())
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5))) // 5분 윈도우
            .aggregate(
                ContentPopularity::new,
                (contentId, event, popularity) -> {
                    popularity.incrementViewCount();
                    popularity.addWatchTime(event.getWatchDurationSeconds());
                    
                    // Netflix 특화: 디바이스별 시청 패턴
                    popularity.trackDeviceUsage(event.getDeviceType());
                    
                    return popularity;
                },
                Materialized.as("content-popularity-store")
            );
        
        // 실시간 추천을 위한 이벤트 발행
        userStats.toStream()
            .filter((userId, stats) -> stats.shouldUpdateRecommendations()) // 추천 업데이트 조건
            .to("recommendation-updates");
        
        // 이상 패턴 감지 (사기 탐지)
        viewingEvents
            .filter(this::detectAnomalousViewing) // 비정상 시청 패턴 감지
            .to("fraud-detection");
    }
    
    private boolean detectAnomalousViewing(String key, ViewingEvent event) {
        // Netflix 특화: 비정상 시청 패턴 감지 로직
        // 예: 짧은 시간에 다수 콘텐츠 시청, 비정상적인 시청 시간 등
        return event.getWatchDurationSeconds() > 86400 || // 24시간 초과 시청
               event.getBufferingRatio() > 0.8; // 버퍼링이 80% 이상
    }
}

// 사용자 시청 통계 집계 객체
public class UserViewingStats {
    private String userId;
    private long totalWatchTimeSeconds;
    private Set<String> uniqueContent;
    private long lastActivityTimestamp;
    private int totalViews;
    
    // Netflix 특화 메트릭
    private double averageBufferingRatio;
    private Map<String, Integer> qualityPreferences; // 화질별 시청 횟수
    private Map<String, Integer> deviceUsage; // 디바이스별 사용 패턴
    private List<String> genrePreferences; // 장르 선호도
    
    public void addWatchTime(int seconds) {
        this.totalWatchTimeSeconds += seconds;
        this.totalViews++;
    }
    
    public void addContentView(String contentId) {
        if (this.uniqueContent == null) {
            this.uniqueContent = new HashSet<>();
        }
        this.uniqueContent.add(contentId);
    }
    
    public void updateQualityMetrics(double bufferingRatio, String quality) {
        // 버퍼링 비율 평균 계산
        this.averageBufferingRatio = 
            (this.averageBufferingRatio * (totalViews - 1) + bufferingRatio) / totalViews;
        
        // 화질 선호도 추적
        qualityPreferences.merge(quality, 1, Integer::sum);
    }
    
    public boolean shouldUpdateRecommendations() {
        // 추천 업데이트가 필요한 조건
        // 예: 새로운 콘텐츠 5개 이상 시청, 1시간 이상 시청 등
        return uniqueContent.size() % 5 == 0 || 
               (System.currentTimeMillis() - lastActivityTimestamp) > 3600000; // 1시간
    }
}

// Cassandra 기반 이벤트 저장소
@Repository
public class CassandraEventStore {
    
    @Autowired
    private CassandraTemplate cassandraTemplate;
    
    public void saveEvent(ViewingEvent event) {
        // 시간 기반 파티셔닝으로 확장성 확보
        String partitionKey = generatePartitionKey(event.getTimestamp());
        
        String cql = """
            INSERT INTO events.viewing_events (
                partition_key, event_id, user_id, content_id, 
                timestamp, watch_duration, device_type, 
                buffering_ratio, video_quality, event_data
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """;
        
        cassandraTemplate.execute(cql, 
            partitionKey,
            UUID.randomUUID(),
            event.getUserId(),
            event.getContentId(),
            new Date(event.getTimestamp()),
            event.getWatchDurationSeconds(),
            event.getDeviceType(),
            event.getBufferingRatio(),
            event.getVideoQuality(),
            JsonUtils.toJson(event) // 전체 이벤트 데이터 JSON 저장
        );
    }
    
    public List<ViewingEvent> getEventsForUser(String userId, long fromTimestamp) {
        // 사용자의 특정 시점 이후 이벤트 조회
        String cql = """
            SELECT * FROM events.viewing_events 
            WHERE user_id = ? AND timestamp >= ?
            ORDER BY timestamp ASC
            """;
        
        return cassandraTemplate.select(cql, ViewingEvent.class, userId, new Date(fromTimestamp));
    }
    
    private String generatePartitionKey(long timestamp) {
        // 시간 기반 파티셔닝 (예: 일별)
        LocalDate date = LocalDate.ofEpochSecond(timestamp / 1000, 0, ZoneOffset.UTC);
        return date.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
    }
    
    // Netflix 특화: 롤업 데이터 저장
    public void saveRollupData(String timeWindow, String aggregationType, Object data) {
        String cql = """
            INSERT INTO events.rollup_data (
                time_window, aggregation_type, data, created_at
            ) VALUES (?, ?, ?, ?)
            """;
        
        cassandraTemplate.execute(cql,
            timeWindow,
            aggregationType, 
            JsonUtils.toJson(data),
            new Date()
        );
    }
}