CQRS와 이벤트 소싱 실전 구현: Axon Framework와 EventStoreDB 2026

# 전통적 상태 저장
orders 테이블: id=123, status=SHIPPING, total=50000

# 이벤트 소싱
이벤트 스트림 [order-123]:
  1. OrderCreated    (total=50000, items=[...])
  2. PaymentCompleted(method=CARD, amount=50000)
  3. ShippingStarted (trackingNo=KR123456)

현재 상태 = fold(events) => Order(status=SHIPPING, total=50000)

// Command 정의
data class CreateOrderCommand(
    @TargetAggregateIdentifier
    val orderId: String,
    val customerId: String,
    val items: List<OrderItem>,
    val totalAmount: Long
)

data class ConfirmPaymentCommand(
    @TargetAggregateIdentifier
    val orderId: String,
    val paymentMethod: String,
    val transactionId: String
)

data class CancelOrderCommand(
    @TargetAggregateIdentifier
    val orderId: String,
    val reason: String
)

// Event 정의
data class OrderCreatedEvent(
    val orderId: String,
    val customerId: String,
    val items: List<OrderItem>,
    val totalAmount: Long,
    val createdAt: Instant = Instant.now()
)

data class PaymentConfirmedEvent(
    val orderId: String,
    val paymentMethod: String,
    val transactionId: String,
    val confirmedAt: Instant = Instant.now()
)

data class OrderCancelledEvent(
    val orderId: String,
    val reason: String,
    val cancelledAt: Instant = Instant.now()
)

// Aggregate 구현
@Aggregate
class OrderAggregate {

    @AggregateIdentifier
    private lateinit var orderId: String
    private lateinit var customerId: String
    private var totalAmount: Long = 0
    private var status: OrderStatus = OrderStatus.DRAFT
    private var items: List<OrderItem> = emptyList()

    // 기본 생성자 (Axon 프레임워크가 리플렉션으로 사용)
    constructor()

    // 생성 커맨드 핸들러 — 새로운 Aggregate 인스턴스를 만든다
    @CommandHandler
    constructor(cmd: CreateOrderCommand) {
        require(cmd.items.isNotEmpty()) { "주문 항목이 비어있을 수 없다" }
        require(cmd.totalAmount > 0) { "주문 금액은 0보다 커야 한다" }

        AggregateLifecycle.apply(
            OrderCreatedEvent(
                orderId = cmd.orderId,
                customerId = cmd.customerId,
                items = cmd.items,
                totalAmount = cmd.totalAmount
            )
        )
    }

    // 결제 확인 커맨드 핸들러
    @CommandHandler
    fun handle(cmd: ConfirmPaymentCommand) {
        check(status == OrderStatus.CREATED) {
            "결제 확인은 CREATED 상태에서만 가능하다. 현재 상태: $status"
        }
        AggregateLifecycle.apply(
            PaymentConfirmedEvent(
                orderId = cmd.orderId,
                paymentMethod = cmd.paymentMethod,
                transactionId = cmd.transactionId
            )
        )
    }

    // 주문 취소 커맨드 핸들러
    @CommandHandler
    fun handle(cmd: CancelOrderCommand) {
        check(status != OrderStatus.CANCELLED) {
            "이미 취소된 주문은 다시 취소할 수 없다"
        }
        check(status != OrderStatus.SHIPPED) {
            "배송 시작된 주문은 취소할 수 없다"
        }
        AggregateLifecycle.apply(
            OrderCancelledEvent(
                orderId = cmd.orderId,
                reason = cmd.reason
            )
        )
    }

    // 이벤트 소싱 핸들러 — 상태를 변경한다 (부수효과 없이 순수하게)
    @EventSourcingHandler
    fun on(event: OrderCreatedEvent) {
        orderId = event.orderId
        customerId = event.customerId
        items = event.items
        totalAmount = event.totalAmount
        status = OrderStatus.CREATED
    }

    @EventSourcingHandler
    fun on(event: PaymentConfirmedEvent) {
        status = OrderStatus.PAID
    }

    @EventSourcingHandler
    fun on(event: OrderCancelledEvent) {
        status = OrderStatus.CANCELLED
    }
}

enum class OrderStatus {
    DRAFT, CREATED, PAID, SHIPPED, DELIVERED, CANCELLED
}

import json
import uuid
from datetime import datetime, timezone
from esdbclient import EventStoreDBClient, NewEvent, StreamState

# EventStoreDB 연결
client = EventStoreDBClient(uri="esdb://localhost:2113?tls=false")

# 이벤트 생성 및 스트림에 추가
stream_name = f"order-{uuid.uuid4()}"

events = [
    NewEvent(
        type="OrderCreated",
        data=json.dumps({
            "orderId": stream_name.split("-", 1)[1],
            "customerId": "customer-001",
            "items": [
                {"productId": "prod-a", "quantity": 2, "price": 15000},
                {"productId": "prod-b", "quantity": 1, "price": 20000},
            ],
            "totalAmount": 50000,
            "createdAt": datetime.now(timezone.utc).isoformat(),
        }).encode("utf-8"),
        metadata=json.dumps({"correlationId": str(uuid.uuid4())}).encode("utf-8"),
        content_type="application/json",
    ),
    NewEvent(
        type="PaymentConfirmed",
        data=json.dumps({
            "paymentMethod": "CARD",
            "transactionId": f"txn-{uuid.uuid4()}",
            "confirmedAt": datetime.now(timezone.utc).isoformat(),
        }).encode("utf-8"),
        content_type="application/json",
    ),
]

# 스트림이 아직 존재하지 않을 때만 쓰기 (낙관적 동시성 제어)
commit_position = client.append_to_stream(
    stream_name=stream_name,
    current_version=StreamState.NO_STREAM,
    events=events,
)
print(f"이벤트 기록 완료. Commit position: {commit_position}")

# 스트림에서 이벤트 읽기
recorded_events = client.get_stream(stream_name)
for event in recorded_events:
    data = json.loads(event.data.decode("utf-8"))
    print(f"[{event.type}] stream_position={event.stream_position}, data={data}")

import { EventStoreDBClient, jsonEvent, NO_STREAM, JSONEventType } from '@eventstore/db-client'

// 이벤트 타입 정의
interface OrderCreated extends JSONEventType {
  type: 'OrderCreated'
  data: {
    orderId: string
    customerId: string
    items: Array<{ productId: string; quantity: number; price: number }>
    totalAmount: number
  }
}

interface PaymentConfirmed extends JSONEventType {
  type: 'PaymentConfirmed'
  data: {
    paymentMethod: string
    transactionId: string
  }
}

// 클라이언트 생성 (싱글턴으로 사용)
const client = EventStoreDBClient.connectionString('esdb://localhost:2113?tls=false')

async function appendAndReadEvents(): Promise<void> {
  const streamName = `order-${crypto.randomUUID()}`

  // 이벤트 추가
  const orderCreated = jsonEvent<OrderCreated>({
    type: 'OrderCreated',
    data: {
      orderId: streamName.replace('order-', ''),
      customerId: 'customer-001',
      items: [{ productId: 'prod-a', quantity: 2, price: 15000 }],
      totalAmount: 30000,
    },
  })

  const paymentConfirmed = jsonEvent<PaymentConfirmed>({
    type: 'PaymentConfirmed',
    data: {
      paymentMethod: 'CARD',
      transactionId: `txn-${crypto.randomUUID()}`,
    },
  })

  // 낙관적 동시성 제어: 스트림이 없을 때만 쓰기
  await client.appendToStream(streamName, [orderCreated, paymentConfirmed], {
    expectedRevision: NO_STREAM,
  })

  // 스트림에서 이벤트 읽기
  const events = client.readStream(streamName)
  for await (const resolvedEvent of events) {
    console.log(`[${resolvedEvent.event?.type}]`, JSON.stringify(resolvedEvent.event?.data))
  }
}

appendAndReadEvents().catch(console.error)

// 읽기 모델 엔티티
@Entity
@Table(name = "order_summary")
data class OrderSummaryEntity(
    @Id
    val orderId: String,
    val customerId: String,
    val totalAmount: Long,
    val status: String,
    val itemCount: Int,
    val createdAt: Instant,
    var updatedAt: Instant = Instant.now()
)

// 프로젝션 (Event Handler)
@Component
@ProcessingGroup("order-summary")
class OrderSummaryProjection(
    private val repository: OrderSummaryRepository
) {

    @EventHandler
    fun on(event: OrderCreatedEvent) {
        val summary = OrderSummaryEntity(
            orderId = event.orderId,
            customerId = event.customerId,
            totalAmount = event.totalAmount,
            status = "CREATED",
            itemCount = event.items.size,
            createdAt = event.createdAt
        )
        repository.save(summary)
    }

    @EventHandler
    fun on(event: PaymentConfirmedEvent) {
        repository.findById(event.orderId).ifPresent { summary ->
            repository.save(summary.copy(
                status = "PAID",
                updatedAt = Instant.now()
            ))
        }
    }

    @EventHandler
    fun on(event: OrderCancelledEvent) {
        repository.findById(event.orderId).ifPresent { summary ->
            repository.save(summary.copy(
                status = "CANCELLED",
                updatedAt = Instant.now()
            ))
        }
    }

    // 리플레이 시 기존 데이터를 정리하는 리셋 핸들러
    @ResetHandler
    fun onReset() {
        repository.deleteAll()
    }
}

// 쿼리 핸들러
@Component
class OrderQueryHandler(
    private val repository: OrderSummaryRepository
) {

    @QueryHandler
    fun handle(query: FindOrderByIdQuery): OrderSummaryEntity? {
        return repository.findById(query.orderId).orElse(null)
    }

    @QueryHandler
    fun handle(query: FindOrdersByCustomerQuery): List<OrderSummaryEntity> {
        return repository.findByCustomerIdOrderByCreatedAtDesc(query.customerId)
    }
}

# application.yml — Axon 스냅샷 설정
axon:
  aggregate:
    order:
      snapshot-trigger:
        type: event-count
        threshold: 100 # 100개 이벤트마다 스냅샷 생성

@Configuration
class AxonSnapshotConfig {

    @Bean
    fun snapshotTriggerDefinition(snapshotter: Snapshotter): SnapshotTriggerDefinition {
        // 50개 이벤트마다 스냅샷 생성
        return EventCountSnapshotTriggerDefinition(snapshotter, 50)
    }
}

// V1: 최초 버전
@Revision("1.0")
data class OrderCreatedEvent(
    val orderId: String,
    val customerId: String,
    val items: List<OrderItem>,
    val totalAmount: Long,
    val createdAt: Instant = Instant.now()
)

// V2: couponCode 필드 추가, currency 필드 추가
@Revision("2.0")
data class OrderCreatedEvent(
    val orderId: String,
    val customerId: String,
    val items: List<OrderItem>,
    val totalAmount: Long,
    val currency: String = "KRW",
    val couponCode: String? = null,
    val createdAt: Instant = Instant.now()
)

// Upcaster 구현: V1 -> V2
class OrderCreatedEventUpcaster : SingleEventUpcaster() {

    override fun canUpcast(intermediateRepresentation: IntermediateEventRepresentation): Boolean {
        return intermediateRepresentation.type.name == OrderCreatedEvent::class.qualifiedName
            && intermediateRepresentation.type.revision == "1.0"
    }

    override fun doUpcast(intermediateRepresentation: IntermediateEventRepresentation): IntermediateEventRepresentation {
        return intermediateRepresentation.upcast(
            SimpleSerializedType(
                OrderCreatedEvent::class.qualifiedName,
                "2.0"  // 목표 리비전
            ),
            JsonNode::class.java
        ) { jsonNode ->
            // V1 이벤트에 없는 필드에 기본값 추가
            val objectNode = jsonNode as ObjectNode
            objectNode.put("currency", "KRW")
            objectNode.putNull("couponCode")
            objectNode
        }
    }
}

// Upcaster 등록
@Configuration
class UpcasterConfig {

    @Bean
    fun orderUpcasterChain(): JpaEventStorageEngine {
        // Upcaster는 체인으로 구성 — V1->V2->V3 순서로 실행
        return JpaEventStorageEngine.builder()
            .upcasterChain(DefaultUpcasterChain(listOf(
                OrderCreatedEventUpcaster(),
                // 추후 V2->V3 Upcaster 추가
            )))
            .build()
    }
}

axon:
  eventhandling:
    processors:
      order-summary:
        mode: tracking
        batch-size: 256 # 한 번에 처리할 이벤트 수
        thread-count: 2 # 병렬 처리 스레드 수
        token-claim-interval: 5000 # 토큰 갱신 간격(ms)