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

# commands.py
class CreatePostCommand:
    """게시글 생성 명령 객체 - CQRS 패턴의 명령 모델 역할"""
    def __init__(self, title, content):
        self.title = title
        self.content = content

# handlers.py
class CommandHandler:
    """명령 처리 핸들러 - 게시글 생성 로직 분리"""
    def handle(self, command):
        # 비즈니스 로직 수행, DB에 저장, 이벤트 발행
        save_to_write_db(command)
        publish_event("PostCreated", command)

# queries.py
class GetPostQuery:
    """게시글 조회 쿼리 객체 - CQRS 패턴의 조회 모델 역할"""
    def __init__(self, post_id):
        self.post_id = post_id

class QueryHandler:
    """조회 처리 핸들러 - 읽기 전용 DB 또는 캐시 조회"""
    def handle(self, query):
        return fetch_from_read_db(query.post_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
46
47
48
49
50
51

# command_models.py
class CreateOrderCommand:
    """주문 생성 명령 - CQRS의 Command 역할"""
    def __init__(self, order_id, items):
        self.order_id = order_id
        self.items = items

# event_models.py
class OrderCreatedEvent:
    """주문 생성 이벤트 - Event Sourcing 핵심 역할"""
    def __init__(self, order_id, items):
        self.order_id = order_id
        self.items = items

# handlers.py (Command Side)
def handle_create_order(command: CreateOrderCommand):
    """명령 처리 -> Write DB 저장 + 이벤트 발행"""
    save_order_to_write_db(command)
    event = OrderCreatedEvent(command.order_id, command.items)
    publish_event_to_kafka("order_created", event.__dict__)

# handlers.py (Query Side)
def handle_order_created_event(event: OrderCreatedEvent):
    """이벤트 처리 -> Read DB 갱신"""
    update_order_in_read_db(event.order_id, event.items)

# message_broker.py (Kafka Publisher & Consumer)
from kafka import KafkaProducer, KafkaConsumer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def publish_event_to_kafka(topic, event_data):
    """이벤트 발행"""
    producer.send(topic, event_data)
    producer.flush()

def consume_order_created_events():
    """이벤트 수신 후 Query 모델 갱신"""
    consumer = KafkaConsumer(
        'order_created',
        bootstrap_servers='localhost:9092',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    for message in consumer:
        event_data = message.value
        event = OrderCreatedEvent(**event_data)
        handle_order_created_event(event)

 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

# saga_coordinator.py
class SagaCoordinator:
    """사가 패턴 중심 주문 트랜잭션 처리"""
    def __init__(self):
        self.state = {}

    def handle_order_created(self, event):
        # 결제 서비스 명령 전송
        self.publish_command('payment', {'order_id': event.order_id, 'amount': event.amount})

    def handle_payment_approved(self, event):
        # 재고 서비스 명령 전송
        self.publish_command('inventory', {'order_id': event.order_id, 'items': event.items})

    def handle_inventory_reserved(self, event):
        # 배송 서비스 명령 전송
        self.publish_command('delivery', {'order_id': event.order_id, 'address': event.address})

    def handle_delivery_scheduled(self, event):
        # 주문 완료 이벤트 발행
        self.publish_event('order_completed', {'order_id': event.order_id})

    def handle_failure(self, stage, event):
        # 실패 단계에 따라 보상(Compensation) 트랜잭션 실행
        if stage == 'payment':
            # 결제 취소 로직
            self.publish_command('payment_compensate', {'order_id': event.order_id})
        elif stage == 'inventory':
            # 재고 복원 로직
            self.publish_command('inventory_compensate', {'order_id': event.order_id})
        elif stage == 'delivery':
            # 배송 예약 취소 로직
            self.publish_command('delivery_compensate', {'order_id': event.order_id})

    def publish_command(self, service, payload):
        # 각 서비스에 명령 메시지 전달
        pass  # 실제 구현은 Kafka/RabbitMQ 등 사용

    def publish_event(self, event_type, payload):
        # 마무리 이벤트 송신 (Kafka 등)
        pass

1
2
3
4
5
6
7
8

# event_handler.py
def handle_order_completed_event(event):
    """주문 완료 이벤트 수신 후 Read DB 갱신"""
    update_order_status_in_read_db(event['order_id'], status='COMPLETED')

def handle_payment_approved_event(event):
    """결제 승인 이벤트 수신 후 Read DB 반영"""
    update_payment_status_in_read_db(event['order_id'], status='APPROVED')

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

# payment_command.py
class ApprovePaymentCommand:
    """결제 승인 명령 - CQRS 명령 모델"""
    def __init__(self, order_id, amount):
        self.order_id = order_id
        self.amount = amount

# payment_handler.py
def handle_approve_payment(command: ApprovePaymentCommand):
    """결제 승인 처리 - 결제 DB 업데이트 + 이벤트 발행"""
    save_payment_to_db(command.order_id, command.amount)
    payment_event = {"order_id": command.order_id, "status": "APPROVED"}
    publish_event("payment_approved", payment_event)

# payment_query.py
class GetPaymentStatusQuery:
    """결제 상태 조회 쿼리 - CQRS 조회 모델"""
    def __init__(self, order_id):
        self.order_id = order_id

def handle_get_payment_status(query: GetPaymentStatusQuery):
    """결제 상태 조회 실행"""
    return fetch_payment_status_from_read_db(query.order_id)

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

# inventory_command.py
class ReserveInventoryCommand:
    """재고 예약 명령 - CQRS 명령 모델"""
    def __init__(self, order_id, items):
        self.order_id = order_id
        self.items = items

def handle_reserve_inventory(command: ReserveInventoryCommand):
    """재고 예약 처리 - 재고 DB 업데이트 + 이벤트 발행"""
    for item in command.items:
        update_inventory_count(item["product_id"], -item["quantity"])
    inventory_event = {"order_id": command.order_id, "status": "RESERVED"}
    publish_event("inventory_reserved", inventory_event)

# inventory_query.py
class GetInventoryStatusQuery:
    """재고 상태 조회 쿼리"""
    def __init__(self, product_id):
        self.product_id = product_id

def handle_get_inventory_status(query: GetInventoryStatusQuery):
    """조회 DB에서 재고 상태 반환"""
    return fetch_inventory_status_from_read_db(query.product_id)

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

# delivery_command.py
class ScheduleDeliveryCommand:
    """배송 예약 명령 모델"""
    def __init__(self, order_id, address):
        self.order_id = order_id
        self.address = address

def handle_schedule_delivery(command: ScheduleDeliveryCommand):
    """배송 예약 처리 - 배송 DB 기록 + 이벤트 발행"""
    save_delivery_schedule_to_db(command.order_id, command.address)
    delivery_event = {"order_id": command.order_id, "status": "SCHEDULED"}
    publish_event("delivery_scheduled", delivery_event)

# delivery_query.py
class GetDeliveryStatusQuery:
    """배송 상태 조회 쿼리"""
    def __init__(self, order_id):
        self.order_id = order_id

def handle_get_delivery_status(query: GetDeliveryStatusQuery):
    """조회 DB에서 배송 상태 반환"""
    return fetch_delivery_status_from_read_db(query.order_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

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-command-service
spec:
  replicas: 3  # 오토스케일링 고려
  selector:
    matchLabels:
      app: order-command-service
  template:
    metadata:
      labels:
        app: order-command-service
    spec:
      containers:
      - name: order-command
        image: myregistry/order-command:latest
        ports:
        - containerPort: 8080
        env:
        - name: KAFKA_BROKER
          value: kafka:9092
---
apiVersion: v1
kind: Service
metadata:
  name: order-command-service
spec:
  selector:
    app: order-command-service
  ports:
  - port: 80
    targetPort: 8080

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

# replay_script.py - Kafka 이벤트 재생 예시
from kafka import KafkaConsumer
import json

def replay_events():
    consumer = KafkaConsumer(
        'order_events',
        bootstrap_servers='kafka:9092',
        auto_offset_reset='earliest',
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    for message in consumer:
        event = message.value
        # 이벤트 타입에 따른 처리
        if event['type'] == 'OrderCreatedEvent':
            rebuild_order_in_read_db(event)
        elif event['type'] == 'PaymentApprovedEvent':
            rebuild_payment_in_read_db(event)

1
2
3
4
5
6
7
8

processed_events = set()

def process_event(event):
    event_key = f"{event['type']}:{event['order_id']}"
    if event_key in processed_events:
        return  # 이미 처리됨
    # 이벤트 처리 로직
    processed_events.add(event_key)

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

# PrometheusRule 예시 (Kafka Consumer Lag 경고)
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: kafka-alert-rules
spec:
  groups:
  - name: kafka.rules
    rules:
    - alert: HighConsumerLag
      expr: kafka_consumer_lag > 1000
      for: 5m
      labels:
        severity: warning
      annotations:
        summary: "Kafka Consumer Lag 지속 발생"
        description: "Kafka Consumer Lag가 5분 이상 1000건 초과"

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

# lambda_handler_order.py
import boto3
import json
sns = boto3.client('sns')

def lambda_handler(event, context):
    """주문 생성 처리 후 이벤트 발행"""
    order_id = event["order_id"]
    amount = event["amount"]
    
    # Write Model 저장 (DynamoDB)
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('OrderWriteModel')
    table.put_item(Item={"order_id": order_id, "amount": amount, "status": "CREATED"})

    # OrderCreatedEvent 발행 (SNS)
    sns.publish(
        TopicArn="arn:aws:sns:ap-northeast-2:123456789012:OrderEvents",
        Message=json.dumps({"event_type": "OrderCreatedEvent", "order_id": order_id, "amount": amount})
    )

    return {"statusCode": 200, "body": "Order created"}

 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

{
  "Comment": "Order Saga Coordination",
  "StartAt": "PaymentService",
  "States": {
    "PaymentService": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789012:function:PaymentHandler",
      "Next": "InventoryService",
      "Catch": [
        { "ErrorEquals": ["States.ALL"], "Next": "CompensatePayment" }
      ]
    },
    "InventoryService": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789012:function:InventoryHandler",
      "Next": "DeliveryService",
      "Catch": [
        { "ErrorEquals": ["States.ALL"], "Next": "CompensateInventory" }
      ]
    },
    "DeliveryService": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789012:function:DeliveryHandler",
      "End": true
    },
    "CompensatePayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789012:function:PaymentCompensate",
      "End": true
    },
    "CompensateInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789012:function:InventoryCompensate",
      "Next": "CompensatePayment"
    }
  }
}

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

config:
  target: "https://api.example.com/orders"
  phases:
    - duration: 60
      arrivalRate: 200   # 초당 200건 API 호출
scenarios:
  - flow:
      - post:
          url: "/create"
          json:
            order_id: "{{ $randomInt(1000000) }}"
            amount: 10000

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

# CloudWatch에서 단계별 Metric 수집 예시 (Python, Boto3)
import boto3

client = boto3.client('cloudwatch')

def get_step_function_metrics(state_machine_arn):
    response = client.get_metric_data(
        MetricDataQueries=[{
            'Id': 'sfExecSuccess',
            'MetricStat': {
                'Metric': {
                    'Namespace': 'AWS/States',
                    'MetricName': 'ExecutionsSucceeded',
                    'Dimensions': [{'Name': 'StateMachineArn', 'Value': state_machine_arn}]
                },
                'Period': 300,
                'Stat': 'Sum',
            }
        }]
    )
    return response['MetricDataResults']

1
2
3
4
5

# Chaos Lambda 예시: 실패 상황을 강제로 발생
def payment_handler(event, context):
    if event.get("chaos_mode"):
        raise Exception("Payment error for chaos test")
    # 정상 결제 처리 로직

1
2
3
4
5
6
7

def event_replay(event_stream):
    for event in event_stream:
        if event['type'] == "OrderCreatedEvent":
            rebuild_order(event)
        elif event['type'] == "PaymentApprovedEvent":
            rebuild_payment(event)
        # ... 단계별 복구 처리 추가

1

Software Engineering > Design and Architecture > Architecture Patterns > Data Management Patterns > CQRS

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

# 명령 서비스 (Command Service)
class OrderCommandService:
    def create_order(self, command: CreateOrderCommand):
        # 비즈니스 로직 처리
        order = Order(command.customer_id, command.items)
        order.validate()  # 도메인 규칙 검증
        
        # 데이터 저장
        self.repository.save(order)
        
        # 이벤트 발행
        self.event_bus.publish(OrderCreatedEvent(order.id, order.total))

# 조회 서비스 (Query Service)  
class OrderQueryService:
    def get_order_summary(self, order_id: str):
        # 읽기 최적화된 뷰에서 조회
        return self.read_repository.get_order_summary(order_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

// 명령 측 - PostgreSQL 사용
class OrderCommandHandler {
    async handle(command) {
        // 트랜잭션 보장이 중요한 명령 처리
        const transaction = await this.postgres.beginTransaction();
        try {
            await this.orderRepository.save(order);
            await this.inventoryRepository.updateStock(items);
            await transaction.commit();
            
            // 이벤트 발행
            await this.eventBus.publish(new OrderProcessedEvent(order));
        } catch (error) {
            await transaction.rollback();
            throw error;
        }
    }
}

// 조회 측 - Redis/MongoDB 사용  
class OrderQueryHandler {
    async getOrderHistory(customerId) {
        // 읽기 성능에 최적화된 NoSQL에서 조회
        return await this.redis.get(`customer:${customerId}:orders`);
    }
}

 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

// 이벤트 소싱과 결합된 CQRS
type AccountAggregate struct {
    ID      string
    Balance int
    Version int
}

func (a *AccountAggregate) Withdraw(amount int) []Event {
    if a.Balance < amount {
        return []Event{WithdrawalRejected{Amount: amount, Reason: "Insufficient funds"}}
    }
    
    return []Event{MoneyWithdrawn{Amount: amount, NewBalance: a.Balance - amount}}
}

// 프로젝션 핸들러 (Projection Handler)
type AccountProjectionHandler struct {
    readStore ReadStore
}

func (h *AccountProjectionHandler) Handle(event Event) {
    switch e := event.(type) {
    case MoneyWithdrawn:
        // 읽기 모델 업데이트
        h.readStore.UpdateAccountBalance(e.AccountID, e.NewBalance)
        h.readStore.AddToTransactionHistory(e.AccountID, e.Amount, "withdrawal")
    }
}

  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

# 1. 도메인 모델 (Domain Model)
from dataclasses import dataclass
from typing import List
from datetime import datetime

@dataclass
class Order:
    """주문 애그리게이트 (Order Aggregate)"""
    id: str
    customer_id: str
    items: List['OrderItem']
    status: str
    created_at: datetime
    total_amount: float

    def calculate_total(self) -> float:
        """총 주문 금액 계산 - 도메인 로직"""
        return sum(item.price * item.quantity for item in self.items)

    def confirm(self):
        """주문 확정 - 비즈니스 규칙 적용"""
        if self.status != 'pending':
            raise ValueError("이미 처리된 주문입니다")
        self.status = 'confirmed'

# 2. 명령 핸들러 (Command Handler)
class CreateOrderCommandHandler:
    def __init__(self, repository, event_bus):
        self.repository = repository  # PostgreSQL 저장소
        self.event_bus = event_bus    # Kafka 이벤트 버스

    async def handle(self, command: 'CreateOrderCommand'):
        """주문 생성 명령 처리"""
        # 도메인 객체 생성 및 검증
        order = Order(
            id=command.order_id,
            customer_id=command.customer_id,
            items=command.items,
            status='pending',
            created_at=datetime.now(),
            total_amount=0
        )
        
        # 비즈니스 로직 실행
        order.total_amount = order.calculate_total()
        order.confirm()
        
        # 영속성 저장 (PostgreSQL)
        await self.repository.save(order)
        
        # 이벤트 발행 (비동기 처리를 위한 이벤트)
        await self.event_bus.publish(OrderCreatedEvent(
            order_id=order.id,
            customer_id=order.customer_id,
            total_amount=order.total_amount,
            items=order.items
        ))

# 3. 조회 핸들러 (Query Handler)
class OrderQueryHandler:
    def __init__(self, redis_client, elasticsearch_client):
        self.redis = redis_client           # 빠른 조회용
        self.elasticsearch = elasticsearch_client  # 검색용

    async def get_order_summary(self, order_id: str):
        """주문 요약 조회 - Redis에서 빠른 조회"""
        # Redis에서 비정규화된 주문 요약 데이터 조회
        order_data = await self.redis.hgetall(f"order:{order_id}")
        return {
            'order_id': order_data.get('id'),
            'customer_name': order_data.get('customer_name'),
            'total_amount': float(order_data.get('total_amount', 0)),
            'status': order_data.get('status'),
            'item_count': int(order_data.get('item_count', 0))
        }

    async def search_orders(self, customer_id: str, query: str):
        """주문 검색 - Elasticsearch에서 검색"""
        search_body = {
            "query": {
                "bool": {
                    "must": [
                        {"match": {"customer_id": customer_id}},
                        {"multi_match": {
                            "query": query,
                            "fields": ["items.name", "status"]
                        }}
                    ]
                }
            }
        }
        
        result = await self.elasticsearch.search(
            index="orders", 
            body=search_body
        )
        return [hit['_source'] for hit in result['hits']['hits']]

# 4. 프로젝션 핸들러 (Projection Handler)
class OrderProjectionHandler:
    def __init__(self, redis_client, elasticsearch_client, customer_service):
        self.redis = redis_client
        self.elasticsearch = elasticsearch_client
        self.customer_service = customer_service

    async def handle_order_created(self, event: 'OrderCreatedEvent'):
        """주문 생성 이벤트 처리 - 읽기 모델 업데이트"""
        
        # 고객 정보 조회 (외부 서비스 호출)
        customer = await self.customer_service.get_customer(event.customer_id)
        
        # Redis에 주문 요약 저장 (빠른 조회용)
        order_summary = {
            'id': event.order_id,
            'customer_id': event.customer_id,
            'customer_name': customer.name,
            'total_amount': str(event.total_amount),
            'status': 'confirmed',
            'item_count': str(len(event.items)),
            'created_at': datetime.now().isoformat()
        }
        
        await self.redis.hset(f"order:{event.order_id}", mapping=order_summary)
        
        # Elasticsearch에 검색용 문서 저장
        search_document = {
            'order_id': event.order_id,
            'customer_id': event.customer_id,
            'customer_name': customer.name,
            'total_amount': event.total_amount,
            'status': 'confirmed',
            'items': [{'name': item.name, 'category': item.category} 
                     for item in event.items],
            'created_at': datetime.now()
        }
        
        await self.elasticsearch.index(
            index="orders", 
            id=event.order_id, 
            body=search_document
        )

# 5. API 엔드포인트 (FastAPI 예시)
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 명령 API 엔드포인트
@app.post("/orders")
async def create_order(command: CreateOrderCommand):
    """주문 생성 - 명령 처리"""
    try:
        await order_command_handler.handle(command)
        return {"message": "주문이 성공적으로 생성되었습니다", "order_id": command.order_id}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

# 조회 API 엔드포인트  
@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    """주문 조회 - 쿼리 처리"""
    order = await order_query_handler.get_order_summary(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="주문을 찾을 수 없습니다")
    return order

@app.get("/customers/{customer_id}/orders/search")
async def search_customer_orders(customer_id: str, q: str = ""):
    """고객 주문 검색 - 복잡한 쿼리 처리"""
    orders = await order_query_handler.search_orders(customer_id, q)
    return {"orders": orders, "total": len(orders)}

  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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257

# 1. 사용자 행동 추적 명령 핸들러
import asyncio
from datetime import datetime
import json

class UserBehaviorTrackingHandler:
    def __init__(self, kafka_producer, cassandra_session):
        self.kafka_producer = kafka_producer
        self.cassandra = cassandra_session

    async def track_user_interaction(self, event: 'UserInteractionEvent'):
        """사용자 상호작용 추적 - Netflix의 시청/평가/검색 행동"""
        
        # 실시간 스트리밍을 위한 Kafka 이벤트 발행
        event_data = {
            'user_id': event.user_id,
            'content_id': event.content_id,
            'interaction_type': event.interaction_type,  # view, rate, search, click
            'interaction_value': event.value,  # 시청 시간, 평점 등
            'timestamp': datetime.now().isoformat(),
            'context': {
                'device_type': event.device_type,
                'session_id': event.session_id,
                'location': event.location
            }
        }
        
        # Kafka로 실시간 이벤트 스트리밍
        await self.kafka_producer.send(
            topic='user-interactions',
            key=event.user_id,
            value=json.dumps(event_data)
        )
        
        # Cassandra에 원시 데이터 저장 (대용량 시계열 데이터 처리)
        await self.cassandra.execute_async("""
            INSERT INTO user_interactions 
            (user_id, content_id, interaction_type, interaction_value, timestamp, context)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (
            event.user_id, 
            event.content_id, 
            event.interaction_type, 
            event.interaction_value,
            datetime.now(), 
            json.dumps(event_data['context'])
        ))

# 2. 추천 시스템 조회 핸들러
class RecommendationQueryHandler:
    def __init__(self, redis_client, mongodb_client):
        self.redis = redis_client      # 빠른 개인화 추천 캐시
        self.mongodb = mongodb_client  # 복잡한 추천 메타데이터

    async def get_personalized_recommendations(self, user_id: str, category: str = None):
        """개인화된 추천 목록 조회 - 사전 계산된 결과 반환"""
        
        # Redis에서 개인화된 추천 목록 조회 (밀리초 응답)
        cache_key = f"recommendations:{user_id}:{category or 'all'}"
        cached_recommendations = await self.redis.get(cache_key)
        
        if cached_recommendations:
            recommendations = json.loads(cached_recommendations)
            
            # MongoDB에서 상세 메타데이터 조회
            content_ids = [rec['content_id'] for rec in recommendations]
            content_details = await self.mongodb.find(
                "contents", 
                {"_id": {"$in": content_ids}},
                {"title": 1, "thumbnail": 1, "description": 1, "rating": 1}
            )
            
            # 추천 점수와 콘텐츠 상세 정보 결합
            detailed_recommendations = []
            for rec in recommendations:
                content_detail = next(
                    (c for c in content_details if c['_id'] == rec['content_id']), 
                    None
                )
                if content_detail:
                    detailed_recommendations.append({
                        **rec,
                        **content_detail,
                        'recommendation_reason': rec.get('reason', '당신을 위한 추천')
                    })
            
            return detailed_recommendations[:20]  # 상위 20개 반환
        
        # 캐시 미스인 경우 기본 추천 반환 (실제로는 fallback 로직)
        return await self._get_default_recommendations(category)

    async def get_trending_content(self, time_window: str = '24h'):
        """실시간 트렌딩 콘텐츠 조회"""
        trending_key = f"trending:{time_window}"
        trending_data = await self.redis.zrevrange(
            trending_key, 0, 19, withscores=True
        )
        
        return [
            {
                'content_id': content_id.decode(),
                'trending_score': score,
                'rank': idx + 1
            }
            for idx, (content_id, score) in enumerate(trending_data)
        ]

# 3. ML 기반 추천 파이프라인 (배치 처리)
from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALS
from pyspark.ml.feature import StringIndexer

class RecommendationPipelineHandler:
    def __init__(self, spark_session, cassandra_config, redis_client):
        self.spark = spark_session
        self.cassandra_config = cassandra_config
        self.redis = redis_client

    async def process_recommendations(self):
        """주기적 추천 모델 훈련 및 결과 생성"""
        
        # Cassandra에서 사용자 행동 데이터 로드
        user_interactions = self.spark.read \
            .format("org.apache.spark.sql.cassandra") \
            .options(table="user_interactions", keyspace="recommendations") \
            .load() \
            .filter("timestamp > date_sub(current_date(), 30)")  # 최근 30일 데이터
        
        # 협업 필터링을 위한 데이터 전처리
        indexer_user = StringIndexer(inputCol="user_id", outputCol="user_idx")
        indexer_content = StringIndexer(inputCol="content_id", outputCol="content_idx")
        
        df_indexed = indexer_user.fit(user_interactions).transform(user_interactions)
        df_indexed = indexer_content.fit(df_indexed).transform(df_indexed)
        
        # 암시적 평점 계산 (시청 시간, 완료율 등으로부터)
        df_ratings = df_indexed.withColumn("rating", 
            self._calculate_implicit_rating(df_indexed))
        
        # ALS 모델 훈련
        als = ALS(
            maxIter=10,
            regParam=0.1,
            userCol="user_idx",
            itemCol="content_idx", 
            ratingCol="rating",
            coldStartStrategy="drop"
        )
        
        model = als.fit(df_ratings)
        
        # 모든 사용자에 대한 추천 생성
        user_recommendations = model.recommendForAllUsers(20)
        
        # 추천 결과를 Redis에 저장
        for row in user_recommendations.collect():
            user_id = self._get_original_user_id(row.user_idx)
            recommendations = [
                {
                    'content_id': self._get_original_content_id(rec.content_idx),
                    'score': float(rec.rating),
                    'reason': self._generate_recommendation_reason(rec)
                }
                for rec in row.recommendations
            ]
            
            # Redis에 개인화 추천 저장 (24시간 TTL)
            await self.redis.setex(
                f"recommendations:{user_id}:all",
                86400,  # 24시간
                json.dumps(recommendations)
            )

    def _calculate_implicit_rating(self, df):
        """암시적 평점 계산 로직"""
        # 시청 시간, 완료율, 재시청 여부 등을 종합한 점수
        from pyspark.sql.functions import when, col
        
        return when(
            col("interaction_type") == "view",
            col("interaction_value") / 3600.0  # 시청 시간(초) -> 시간 단위 점수
        ).when(
            col("interaction_type") == "rate", 
            col("interaction_value")  # 직접 평점
        ).when(
            col("interaction_type") == "complete",
            5.0  # 완료 시청 = 높은 점수
        ).otherwise(1.0)

# 4. API 엔드포인트 (FastAPI)
from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI(title="Netflix-Style Recommendation API")

@app.post("/track")
async def track_interaction(interaction: UserInteractionEvent):
    """사용자 상호작용 추적 - 명령 API"""
    await behavior_handler.track_user_interaction(interaction)
    return {"status": "tracked"}

@app.get("/users/{user_id}/recommendations")
async def get_recommendations(
    user_id: str, 
    category: Optional[str] = Query(None, description="콘텐츠 카테고리")
):
    """개인화된 추천 조회 - 쿼리 API"""
    recommendations = await recommendation_handler.get_personalized_recommendations(
        user_id, category
    )
    return {
        "user_id": user_id,
        "recommendations": recommendations,
        "generated_at": datetime.now().isoformat()
    }

@app.get("/trending")
async def get_trending(time_window: str = Query("24h", regex="^(1h|24h|7d)$")):
    """트렌딩 콘텐츠 조회"""
    trending = await recommendation_handler.get_trending_content(time_window)
    return {
        "time_window": time_window,
        "trending": trending
    }

# 5. 이벤트 프로세서 (실시간 트렌딩 계산)
class TrendingEventProcessor:
    def __init__(self, kafka_consumer, redis_client):
        self.kafka_consumer = kafka_consumer
        self.redis = redis_client

    async def process_trending_events(self):
        """실시간 트렌딩 점수 계산"""
        async for message in self.kafka_consumer:
            event_data = json.loads(message.value)
            
            if event_data['interaction_type'] == 'view':
                # 실시간 트렌딩 점수 업데이트
                content_id = event_data['content_id']
                timestamp = datetime.fromisoformat(event_data['timestamp'])
                
                # 시간 기반 가중치 적용 (최근일수록 높은 가중치)
                weight = self._calculate_time_weight(timestamp)
                
                # Redis Sorted Set에 트렌딩 점수 누적
                await self.redis.zincrby("trending:24h", weight, content_id)
                await self.redis.zincrby("trending:7d", weight * 0.7, content_id)
                
                # TTL 설정으로 오래된 데이터 자동 정리
                await self.redis.expire("trending:24h", 86400)  # 24시간
                await self.redis.expire("trending:7d", 604800)  # 7일

    def _calculate_time_weight(self, timestamp):
        """시간 기반 가중치 계산 (최근일수록 높은 가중치)"""
        time_diff = datetime.now() - timestamp
        hours_ago = time_diff.total_seconds() / 3600
        return max(1.0, 24.0 - hours_ago) / 24.0  # 0-24시간 전 데이터에 가중치