CQRS

 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

# pip install fastapi uvicorn sqlalchemy psycopg2-binary pydantic
# 핵심: 동일 트랜잭션 내에 '도메인 쓰기'와 'outbox 이벤트 기록'을 함께 커밋
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, JSON, text
from sqlalchemy.orm import declarative_base, sessionmaker

engine = create_engine("postgresql://user:pass@localhost/app", future=True)
Session = sessionmaker(bind=engine, expire_on_commit=False, future=True)
Base = declarative_base()

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    customer_id = Column(String, nullable=False)
    status = Column(String, default="CREATED")

class Outbox(Base):
    __tablename__ = "outbox"
    id = Column(Integer, primary_key=True)
    aggregate_id = Column(String, nullable=False)
    type = Column(String, nullable=False)          # e.g. "OrderCreated"
    payload = Column(JSON, nullable=False)

class CreateOrderCommand(BaseModel):
    customer_id: str

app = FastAPI()

@app.post("/orders")
def create_order(cmd: CreateOrderCommand):
    with Session.begin() as s:  # 1) 트랜잭션 시작
        order = Order(customer_id=cmd.customer_id)
        s.add(order)

        # 2) 같은 트랜잭션에서 Outbox 레코드 삽입 (정합성 보장)
        evt = Outbox(
            aggregate_id=str(order.id),
            type="OrderCreated",
            payload={"orderId": order.id, "customerId": order.customer_id},
        )
        s.add(evt)

    # 3) Debezium Outbox Router가 outbox 테이블 CDC→Kafka로 게시(비동기)
    return {"orderId": order.id, "status": "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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

// npm i kafkajs mongodb
// 핵심: (1) Producer가 "주문ID"를 파티션 키로 사용 → 파티션 내 순서 보장
//      (2) Consumer가 이벤트를 받아 MongoDB 읽기 모델을 업데이트(프로젝션)

import { Kafka } from "kafkajs";
import { MongoClient } from "mongodb";

// 1) Producer: 파티션 키=orderId
async function publishOrderEvent(broker: string, orderId: string, payload: any) {
  const kafka = new Kafka({ clientId: "orders", brokers: [broker] });
  const producer = kafka.producer();
  await producer.connect();
  await producer.send({
    topic: "order.events",
    messages: [{ key: orderId, value: JSON.stringify(payload) }], // 파티션 키
  });
  await producer.disconnect();
}

// 2) Consumer/Projection: MongoDB 읽기 모델 업데이트
async function startProjection(broker: string, mongoUri: string) {
  const kafka = new Kafka({ clientId: "orders-projection", brokers: [broker] });
  const consumer = kafka.consumer({ groupId: "orders-read-model" });
  const mongo = new MongoClient(mongoUri);
  await mongo.connect();
  const view = mongo.db("read").collection("order_views");

  await consumer.connect();
  await consumer.subscribe({ topic: "order.events", fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ message }) => {
      if (!message.value) return;
      const event = JSON.parse(message.value.toString());
      // 이벤트 타입에 따라 읽기 모델 비정규화/업데이트
      if (event.type === "OrderCreated") {
        await view.updateOne(
          { orderId: event.orderId },
          { $set: { orderId: event.orderId, customerId: event.customerId, status: "CREATED" } },
          { upsert: true }
        );
      }
      if (event.type === "OrderPaid") {
        await view.updateOne({ orderId: event.orderId }, { $set: { status: "PAID" } });
      }
    },
  });
}

// 사용 예시
// publishOrderEvent("localhost:9092", "order-123", { type: "OrderCreated", orderId: "order-123", customerId: "c-1" });
// startProjection("localhost:9092", "mongodb://localhost:27017").then(() => console.log("projection running"));

 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

# Command API 예시
from fastapi import FastAPI
from kafka import KafkaProducer
import json

app = FastAPI()
producer = KafkaProducer(bootstrap_servers='localhost:9092')

@app.post("/orders")
def create_order(order: dict):
    # 1. Write DB에 저장 (간략화 예시)
    print(f"Saving order: {order}")
    # 2. 이벤트 발행
    producer.send('order_created', json.dumps(order).encode())
    return {"status": "order created"}

# Query API 예시
from fastapi import FastAPI
import redis

query_app = FastAPI()
cache = redis.Redis(host='localhost', port=6379)

@query_app.get("/orders/{order_id}")
def get_order(order_id: str):
    # Redis 캐시 조회
    data = cache.get(order_id)
    return json.loads(data) if data else {"error": "Not found"}

 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

# FastAPI Command API: 트랜잭션 내 본데이터+Outbox에 함께 기록하여 이중 쓰기 문제 방지
from fastapi import FastAPI
import psycopg2, json, uuid, time

app = FastAPI()
conn = psycopg2.connect("dbname=orders user=app password=secret host=localhost")

@app.post("/orders")
def create_order(payload: dict):
    order_id = str(uuid.uuid4())
    evt = {
        "event_id": str(uuid.uuid4()),
        "type": "OrderCreated",
        "aggregate_id": order_id,
        "occurred_at": int(time.time()),
        "data": payload,
        "version": 1
    }
    with conn:
        with conn.cursor() as cur:
            # 1) 본 데이터 저장 (쓰기 모델)
            cur.execute("INSERT INTO orders(id, status, total_amount) VALUES (%s, %s, %s)",
                        (order_id, "CREATED", payload["total_amount"]))
            # 2) Outbox에 이벤트 기록 (동일 트랜잭션)
            cur.execute("""INSERT INTO outbox(event_id, aggregate_id, event_type, payload, occurred_at, status)
                           VALUES (%s,%s,%s,%s, to_timestamp(%s), 'PENDING')""",
                        (evt["event_id"], evt["aggregate_id"], evt["type"], json.dumps(evt), evt["occurred_at"]))
    return {"order_id": order_id, "status": "created"}

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

-- PostgreSQL 테이블 (쓰기 모델 + Outbox)
CREATE TABLE orders(
  id uuid PRIMARY KEY,
  status text NOT NULL,
  total_amount numeric NOT NULL
);

CREATE TABLE outbox(
  event_id uuid PRIMARY KEY,
  aggregate_id uuid NOT NULL,
  event_type text NOT NULL,
  payload jsonb NOT NULL,
  occurred_at timestamp NOT NULL,
  status text NOT NULL
);
-- Debezium은 outbox 테이블의 변경을 캡처하여 Kafka로 전달 (CDC)

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

# Projection Consumer (Kafka -> Redis): 동일 aggregate_id 키를 사용해 순서 보장(단일 파티션 가정)
from kafka import KafkaConsumer
import redis, json

r = redis.Redis(host="localhost", port=6379)
consumer = KafkaConsumer(
    "orders-events",
    bootstrap_servers=["localhost:9092"],
    group_id="projection-readmodel",
    enable_auto_commit=False)

for msg in consumer:
    evt = json.loads(msg.value)
    agg_id = evt["aggregate_id"]  # 파티션 키로 설정되어 있다고 가정
    # 이벤트 타입별 Projection 갱신
    if evt["type"] == "OrderCreated":
        r.hset(f"order:{agg_id}", mapping={"status":"CREATED", "total": evt["data"]["total_amount"]})
    elif evt["type"] == "OrderPaid":
        r.hset(f"order:{agg_id}", mapping={"status":"PAID"})
    elif evt["type"] == "OrderShipped":
        r.hset(f"order:{agg_id}", mapping={"status":"SHIPPED"})
    consumer.commit()  # at-least-once, idempotent 갱신 로직 권장

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# FastAPI Query API: 읽기 전용(Projection) 경로
from fastapi import FastAPI
import redis, json

q = FastAPI()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

@q.get("/orders/{order_id}")
def get_order(order_id: str):
    data = r.hgetall(f"order:{order_id}")
    return data if data else {"error": "not-found"}

 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
 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시간 전 데이터에 가중치