cqrs-aggregate-modeling

Redefining aggregate boundaries with CQRS

Introducing CQRS changes aggregate modeling. Aggregates hold only the minimal state required for command execution, and read responsibilities are delegated to the read model.

Problem: Bloated aggregates

Typical example: A Thread aggregate holds 1,000 messages

// 従来型: 集約がすべてのデータを保持
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)

Problem during updates

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

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

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

To add one message, 1,001 records are updated. This is caused by the aggregate holding both "data needed for commands" and "data needed for queries" without distinguishing between them.

The temptation of differential updates

Trying to implement differential updates complicates the internal implementation of the aggregate. A mechanism is needed to track which fields have changed, mixing domain logic and infrastructure concerns.

Solution: Redesigning aggregates with CQRS

Core principle

When CQRS is introduced, aggregates only need to hold the minimal state required for command execution.

Read responsibilities (queries) are completely removed from the aggregate and delegated to the read model. As a result, the aggregate holds only the information needed to validate commands.

Question: What is needed to validate this command?

In the case of the Thread aggregate, what is needed to validate the "add message" command is:

• The sender is a member of the thread → List of member IDsis needed
• There is no duplicate message ID → List of message IDsis needed

The message body is not needed. The body is needed for display (queries) and is unrelated to command validation.

Aggregate after redesign

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

Because it does not hold message bodies, the aggregate is significantly lighter.

Event design

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

Events include the message body (because it is needed to build the read model). However, when restoring the aggregate's state, only the ID is reflected.

Read model (Q side)

// イベントを消費してリードモデルを構築
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)

Before / After comparison

Aspect	Conventional (non-CQRS)	CQRS/ES
Aggregate state	Holds full message text	Holds only message IDs
Adding a message	Updates all records	Appends one event
Reading	Retrieved directly from the aggregate	Retrieved from the read model
Memory usage	Grows proportionally to the number of messages	Proportional to the number of IDs (lightweight)
Pagination	Implemented within the aggregate (complex)	Read model DAO (natural)

Approach to redefining aggregate boundaries

Criterion: Is it necessary for command validation?

To decide what data an aggregate should hold, analyze the validation logic of each command.

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

Concrete example: Analysis of the Thread aggregate

Data	Necessary for command validation	Decision
Member ID list	YES (check whether the sender is a member)	Keep in the aggregate
Message ID list	YES (duplicate check)	Keep in the aggregate
Message body	NO (display only)	To lead model
Sender name	NO (display only)	To lead model

Reconsidering strong consistency

Questions to ask when introducing CQRS:

Is strong consistency necessary for the relationship between threads and messages?

• For adding and displaying messages, "immediate consistency of the message body" is not necessary
• Strong consistency is necessary only for membership verification
• If you can picture the behavior, the structure of the aggregate becomes clear

Signs of an overly large aggregate and how to deal with them

Signs

Sign	Cause
Loading the aggregate is slow	Holding a large amount of unnecessary data
An all-rows SQL query occurs on update	Differences cannot be tracked
Pagination logic inside the aggregate	Query responsibilities are mixed in
The structure of the DTO and the aggregate are very similar	Query data is used as-is in the aggregate

Remediation flow

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

Relationship with related skills

Skills	Relationship
aggregate-design	Internal design principles of aggregates. This skill covers redefinition of boundaries through CQRSboundary redefinition
cqrs-to-event-sourcing	Why ES is necessary. This skill covers modeling transformation premised on ESmodeling transformation
cqrs-tradeoffs	Trade-offs between consistency and availability. This skill covers impact on modelingimpact on modeling

Review checklist

Aggregate bloat

• Is the aggregate holding query-only data (display names, calculation results, etc.)?
• Are there performance problems when loading the aggregate?
• Are unnecessary full-row updates occurring on update?

Redesign with CQRS/ES

• Is each field classified by "whether it is necessary for command validation"?
• Is query-only data delegated to the read model?
• Does the aggregate hold only minimal state such as ID lists?
• Do the events contain all the information necessary for building the read model?

Validity of boundaries

• Have you reconsidered whether strong consistency is truly necessary for all data in the aggregate?
• Are you determining the aggregate boundaries based on behavior (commands)?
• Are sufficient eventually consistent data separated from the aggregate?

Related skills (recommended for concurrent reading)

When using this skill, also refer to the following skills:

• cqrs-to-event-sourcing: Why event sourcing changes aggregate modeling
• aggregate-design: Basic aggregate design rules before applying CQRS
• cqrs-tradeoffs: Trade-off analysis for adopting CQRS