cqrs-aggregate-modeling

CQRSによる集約の境界再定義

CQRSを導入すると集約のモデリングが変わる。集約はコマンド実行に必要な最小限の状態のみ保持し、読み取り責務はリードモデルに委譲する。

問題: 肥大化した集約

典型例: Thread集約が1000件のメッセージを保持

// 従来型: 集約がすべてのデータを保持
case class Message(id: MessageId, text: MessageText, senderId: AccountId,
                   createdAt: Instant, updatedAt: Instant)
case class Messages(values: List[Message])

class Thread(id: ThreadId, members: Members, messages: Messages,
             createdAt: Instant)

更新時の問題

1. threadRepository.findById(threadId)
   → 1000件のメッセージを含むスレッド全体をDBから取得

2. thread.addMessage(...)
   → メッセージを1件追加

3. threadRepository.store(newThread)
   → 1001件全体をDBに更新
   → どのフィールドが更新されたか不明なため、全情報を更新する必要がある

1件のメッセージ追加のために1001件を更新する。 これは集約が「コマンドに必要なデータ」と「クエリに必要なデータ」を区別せずに保持していることが原因。

差分更新の誘惑

差分更新を実装しようとすると、集約の内部実装が複雑化する。どのフィールドが変更されたかを追跡する仕組みが必要になり、ドメインロジックとインフラの関心が混在する。

解決: CQRSによる集約の再設計

核心原則

CQRSを導入すると、集約はコマンド実行に必要な最小限の状態だけ持てばよい。

読み取り責務（クエリ）を集約から完全に除去し、リードモデルに委譲する。その結果、集約はコマンドの検証に必要な情報のみ保持する。

問い: このコマンドの検証に何が必要か？

Thread集約の場合、「メッセージ追加」コマンドの検証に必要なのは：

• 送信者がスレッドのメンバーであること → メンバーIDのリストが必要
• メッセージIDの重複がないこと → メッセージIDのリストが必要

メッセージの本文は不要。 本文は表示（クエリ）のために必要であり、コマンドの検証には関係ない。

再設計後の集約

// CQRS/ES: 集約はコマンド検証に必要な最小限の状態のみ保持
class Thread(id: ThreadId, memberIds: MemberIds, messageIds: MessageIds,
             createdAt: Instant) {

  def addMessage(messageId: MessageId, messageText: MessageText,
                 senderId: AccountId): Either[ThreadError, Thread] =
    if (memberIds.contains(senderId)) {
      // イベントを追記するだけ。1001件の更新は発生しない
      persistEvent(MessageAdded(id, messageId, messageText, senderId, Instant.now))
      Right(copy(messageIds = messageIds.add(messageId)))  // IDのみ追加
    } else {
      Left(new AddMessageError)
    }
}

メッセージ本文を持たないため、集約は大幅に軽量化される。

イベントの設計

sealed trait ThreadEvent

case class MemberAdded(threadId: ThreadId, accountId: AccountId,
                       occurredAt: Instant) extends ThreadEvent

case class MessageAdded(threadId: ThreadId, messageId: MessageId,
                       messageText: MessageText, senderId: AccountId,
                       occurredAt: Instant) extends ThreadEvent

case class MessageUpdated(threadId: ThreadId, messageId: MessageId,
                         messageText: MessageText, senderId: AccountId,
                         occurredAt: Instant) extends ThreadEvent