j5ik2o/okite-ai/cqrs-aggregate-modeling

// 従来型: 集約がすべてのデータを保持
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に更新
   → どのフィールドが更新されたか不明なため、全情報を更新する必要がある

// 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

// イベントを消費してリードモデルを構築
consumeEventsByThreadIdFromDDBStreams.foreach {
  case ev: MemberAdded   => insertMember(ev)
  case ev: MessageAdded  => insertMessage(ev)
  case ev: MessageUpdated => updateMessage(ev)
}
 
// リードモデルはクエリに最適化されたDTO
case class MessageDto(id: Long, threadId: Long, text: String,
                     senderId: Long, createdAt: Instant, updatedAt: Instant)
 
// 部分取得が可能（ページネーション等）
val messages: Seq[MessageDto] =
  MessageDao.findAllByThreadIdWithOffsetLimit(threadId, 0, 100)

集約が現在保持しているデータ
    ↓
各フィールドについて:
    「このデータはコマンドの検証に使われるか？」
    ├─ YES → 集約に残す
    └─ NO → クエリ専用データ → リードモデルへ移動

集約が大きすぎる
    ↓
1. 各フィールドを「コマンド検証用」と「クエリ用」に分類
    ↓
2. クエリ用データをリードモデルへ移動（CQRSの導入）
    ↓
3. 集約はIDリストや状態フラグなど最小限の状態のみ保持
    ↓
4. イベントで状態変更を記録し、リードモデルはイベントから構築