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)

// コマンド定義
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
)

// イベント定義
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）