pekko-cqrs-es-implementation

Pekko CQRS/ES Implementation Guide

A collection of concrete patterns for implementing CQRS/Event Sourcing with Apache Pekko + Scala 3.

Applicability Conditions

Use this skill only when all of the following are satisfied:

The programming language is Scala (3.x recommended)
The CQRS/Event Sourcing architecture is adopted
Apache Pekko (or Akka) is used as the actor framework

If the above are not satisfied,cqrs-tradeoffs、cqrs-to-event-sourcing、cqrs-aggregate-modelinguse a language-agnostic skill such as ...

Architecture Overview

Data Flow of the Entire System

クライアント
    │
    ▼
コマンドAPI（GraphQL Mutation）
    │
    ▼
ユースケース層（ZIO）
    │
    ▼
集約アクター（Pekko Typed）
    │
    ▼
ドメインモデル（純粋Scala）→ イベント生成
    │
    ▼
イベントストア（DynamoDB）
    │
    ▼
DynamoDB Streams
    │
    ▼
リードモデルアップデータ（AWS Lambda）
    │
    ▼
リードモデル（PostgreSQL）
    │
    ▼
クエリAPI（GraphQL Query）

Separation of the Command Side and the Query Side

	Command Side	Query Side
Purpose	Business Rule Execution	Efficient Data Retrieval
Data Store	DynamoDB (Event Store)	PostgreSQL (Read Model)
API	GraphQL Mutation	GraphQL Query
Layer	Domain → Use Case → Interface Adapter	Interface Adapter Only
Consistency	Strong Consistency (Within an Aggregate)	Eventual Consistency

There is no use case layer on the query side. GraphQL itself corresponds to the use case.

Module Structure

modules/
├── command/
│   ├── domain/                          # ドメイン層（純粋Scala）
│   ├── use-case/                        # ユースケース層（ZIO）
│   ├── interface-adapter/               # アクター、GraphQL、シリアライザ
│   ├── interface-adapter-contract/      # コマンド/リプライのプロトコル定義
│   └── interface-adapter-event-serializer/  # Protocol Buffersシリアライザ
├── query/
│   ├── interface-adapter/               # DAO（Slick）、GraphQL
│   └── flyway-migration/               # DBマイグレーション
└── infrastructure/                      # 共有ユーティリティ

apps/
├── command-api/                         # コマンド側HTTPサーバー
├── query-api/                           # クエリ側HTTPサーバー
└── read-model-updater/                  # AWS Lambda

Dependency Direction

domain ← use-case ← interface-adapter ← apps
  │                       ↑
  │               interface-adapter-contract
  │               interface-adapter-event-serializer
  └── Scalaのみに依存。フレームワーク非依存。

Command-Side Implementation Patterns

1. Domain Model

Principle: The domain model is pure Scala code and does not depend on Pekko at all.

Aggregate (trait + private case class)

trait UserAccount extends Entity {
  override type IdType = UserAccountId
  def id: UserAccountId
  def name: UserAccountName
  def emailAddress: EmailAddress
  def createdAt: DateTime
  def updatedAt: DateTime

  def rename(newName: UserAccountName): Either[RenameError, (UserAccount, UserAccountEvent)]
  def delete: Either[DeleteError, (UserAccount, UserAccountEvent)]
}

object UserAccount {
  // ファクトリ: 新しい状態とイベントのペアを返す
  def apply(
    id: UserAccountId,
    name: UserAccountName,
    emailAddress: EmailAddress,
    createdAt: DateTime = DateTime.now(),
    updatedAt: DateTime = DateTime.now()
  ): (UserAccount, UserAccountEvent) =
    (
      UserAccountImpl(id, false, name, emailAddress, createdAt, updatedAt),
      UserAccountEvent.Created_V1(
        id = DomainEventId.generate(),
        entityId = id,
        name = name,
        emailAddress = emailAddress,
        occurredAt = DateTime.now()
      ))

  // 実装はprivateで隠蔽
  private final case class UserAccountImpl(
    id: UserAccountId,
    deleted: Boolean,
    name: UserAccountName,
    emailAddress: EmailAddress,
    createdAt: DateTime,
    updatedAt: DateTime
  ) extends UserAccount {

    override def rename(
      newName: UserAccountName): Either[RenameError, (UserAccount, UserAccountEvent)] =
      if (name == newName) {
        Left(RenameError.FamilyNameSame)
      } else {
        val updated = this.copy(name = newName, updatedAt = DateTime.now())
        val event = UserAccountEvent.Renamed_V1(
          id = DomainEventId.generate(),
          entityId = id,
          oldName = name,
          newName = newName,
          occurredAt = DateTime.now()
        )
        Right((updated, event))
      }

    override def delete: Either[DeleteError, (UserAccount, UserAccountEvent)] =
      if (deleted) {
        Left(DeleteError.AlreadyDeleted)
      } else {
        val updated = copy(deleted = true, updatedAt = DateTime.now())
        val event = UserAccountEvent.Deleted_V1(
          id = DomainEventId.generate(),
          entityId = id,
          occurredAt = DateTime.now()
        )
        Right((updated, event))
      }
  }
}

Important Patterns:

State-changing methods return Either[Error, (NewState, Event)] returns
Factories also return (State, Event) pair
The implementation class is private cannot be constructed directly from outside
There are no Pekko imports whatsoever within the domain model

Event Definition (enum + versioning)

enum UserAccountEvent extends DomainEvent {
  override type EntityIdType = UserAccountId

  case Created_V1(
    id: DomainEventId,
    entityId: UserAccountId,
    name: UserAccountName,
    emailAddress: EmailAddress,
    occurredAt: DateTime
  )

  case Renamed_V1(
    id: DomainEventId,
    entityId: UserAccountId,
    oldName: UserAccountName,
    newName: UserAccountName,
    occurredAt: DateTime
  )

  case Deleted_V1(
    id: DomainEventId,
    entityId: UserAccountId,
    occurredAt: DateTime
  )
}

Event Design Rules:

Rule	Description	Example
Named in the past tense	Represents "what happened"	`Created`, `Renamed`, `Deleted`
`_V1`Suffix	Supports schema evolution	`Renamed_V1`, `Renamed_V2`
Immutable	Automatically guaranteed by case class
Self-contained	Includes values before and after the change	`oldName`, `newName`
Required Fields	`id`, `entityId`, `occurredAt`	Common to all events

2. Aggregate State (enum)

enum UserAccountAggregateState {
  case NotCreated(id: UserAccountId)
  case Created(user: UserAccount)
  case Deleted(user: UserAccount)

  def applyEvent(event: UserAccountEvent): UserAccountAggregateState = (this, event) match {
    case (NotCreated(id), UserAccountEvent.Created_V1(_, entityId, name, emailAddress, _))
        if id == entityId =>
      Created(UserAccount(entityId, name, emailAddress)._1)

    case (Created(user), UserAccountEvent.Renamed_V1(_, entityId, _, newName, _))
        if user.id == entityId =>
      Created(user.rename(newName) match {
        case Right((u, _)) => u
        case Left(error) =>
          throw new IllegalStateException(s"Failed to rename user: $error")
      })

    case (Created(user), UserAccountEvent.Deleted_V1(_, entityId, _))
        if user.id == entityId =>
      Deleted(user.delete match {
        case Right((deletedUser, _)) => deletedUser
        case Left(error) =>
          throw new IllegalStateException(s"Failed to delete user: $error")
      })

    case _ =>
      throw new IllegalStateException(s"Cannot apply event $event to state $this")
  }
}

Type Safety of State Transitions:

NotCreated → Created(Created_V1 event only)
Created → Created（Renamed_V1）/ Deleted（Deleted_V1）
Deleted → No transition (does not accept any event)

3. Protocol definition (commands/replies)

object UserAccountProtocol {
  // コマンド: すべてidを持つ
  sealed trait Command { def id: UserAccountId }
  final case class Create(
    id: UserAccountId, name: UserAccountName,
    emailAddress: EmailAddress, replyTo: ActorRef[CreateReply]) extends Command
  final case class Rename(
    id: UserAccountId, newName: UserAccountName,
    replyTo: ActorRef[RenameReply]) extends Command
  final case class Delete(
    id: UserAccountId, replyTo: ActorRef[DeleteReply]) extends Command
  final case class Get(
    id: UserAccountId, replyTo: ActorRef[GetReply]) extends Command

  // リプライ: コマンドごとに専用の型
  sealed trait CreateReply
  final case class CreateSucceeded(id: UserAccountId) extends CreateReply

  sealed trait RenameReply
  final case class RenameSucceeded(id: UserAccountId) extends RenameReply
  final case class RenameFailed(id: UserAccountId, reason: RenameError) extends RenameReply

  sealed trait DeleteReply
  final case class DeleteSucceeded(id: UserAccountId) extends DeleteReply
  final case class DeleteFailed(id: UserAccountId, reason: DeleteError) extends DeleteReply

  sealed trait GetReply
  final case class GetSucceeded(value: UserAccount) extends GetReply
  final case class GetNotFoundFailed(id: UserAccountId) extends GetReply
}

Protocol design rules:

All commands sealed trait Command inherit, andid have
Define a dedicated sealed trait for each command reply (CreateReply, RenameReplyetc.)
replyTo: ActorRef[XxxReply] Guarantees type-safe responses
Express success/failure as case classes, and check exhaustiveness with pattern matching

4. Aggregate actor (PersistenceEffector)

object UserAccountAggregate {

  // 状態ごとにハンドラ関数を分離
  private def handleNotCreated(
    state: UserAccountAggregateState.NotCreated,
    effector: PersistenceEffector[UserAccountAggregateState, UserAccountEvent, Command]
  ): Behavior[Command] = Behaviors.receiveMessagePartial {
    case Create(id, name, emailAddress, replyTo) if state.id == id =>
      val (newState, event) = UserAccount(id, name, emailAddress)
      effector.persistEvent(event) { _ =>
        replyTo ! CreateSucceeded(id)
        handleCreated(UserAccountAggregateState.Created(newState), effector)
      }
    case Get(id, replyTo) if state.id == id =>
      replyTo ! GetNotFoundFailed(id)
      Behaviors.same
  }

  private def handleCreated(
    state: UserAccountAggregateState.Created,
    effector: PersistenceEffector[UserAccountAggregateState, UserAccountEvent, Command]
  ): Behavior[Command] = Behaviors.receiveMessagePartial {
    case Rename(id, newName, replyTo) if state.user.id == id =>
      // ドメインモデルに委譲
      state.user.rename(newName) match {
        case Left(reason) =>
          replyTo ! RenameFailed(id, reason)
          Behaviors.same
        case Right((newUser, event)) =>
          effector.persistEvent(event) { _ =>
            replyTo ! RenameSucceeded(id)
            handleCreated(state.copy(user = newUser), effector)
          }
      }
    // ... Delete, Get も同様
  }

  // エントリポイント
  def apply(id: UserAccountId): Behavior[Command] = {
    val config = PersistenceEffectorConfig
      .create[UserAccountAggregateState, UserAccountEvent, Command](
        persistenceId = s"${id.entityTypeName}-${id.asString}",
        initialState = UserAccountAggregateState.NotCreated(id),
        applyEvent = (state, event) => state.applyEvent(event)
      )
      .withPersistenceMode(PersistenceMode.Persisted)
      .withSnapshotCriteria(SnapshotCriteria.every(1000))
      .withRetentionCriteria(RetentionCriteria.snapshotEvery(2))

    Behaviors.setup[Command] { implicit ctx =>
      Behaviors
        .supervise(
          PersistenceEffector.fromConfig(config) {
            case (state: UserAccountAggregateState.NotCreated, effector) =>
              handleNotCreated(state, effector)
            case (state: UserAccountAggregateState.Created, effector) =>
              handleCreated(state, effector)
            case (state: UserAccountAggregateState.Deleted, effector) =>
              handleDeleted(state, effector)
          })
        .onFailure[IllegalArgumentException](SupervisorStrategy.restart)
    }
  }
}

Actor implementation rules:

Do not write domain logic in actors. Actors focus on persistence and lifecycle management
Separate handler functions per state, and limit the commands accepted
effector.persistEvent(event) { _ => ... } Reply after event persistence with
When the domain model returns Left reply only (do not persist)
When the domain model returns Right persist the event and then reply

5. Use case layer (ZIO)

private[users] final class UserAccountUseCaseImpl(
  userAccountAggregateRef: ActorRef[UserAccountProtocol.Command]
)(implicit
  timeout: Timeout,
  scheduler: Scheduler,
  ec: ExecutionContext
) extends UserAccountUseCase {

  override def createUserAccount(
    userAccountName: UserAccountName,
    emailAddress: EmailAddress
  ): IO[UserAccountUseCaseError, UserAccountId] =
    for {
      userAccountId <- ZIO.succeed(UserAccountId.generate())
      reply <- askActor[UserAccountProtocol.CreateReply] { replyTo =>
        UserAccountProtocol.Create(
          id = userAccountId,
          name = userAccountName,
          emailAddress = emailAddress,
          replyTo = replyTo
        )
      }.mapError(e => UserAccountUseCaseError.UnexpectedError(e.getMessage, Some(e)))
      result <- reply match {
        case UserAccountProtocol.CreateSucceeded(id) => ZIO.succeed(id)
      }
    } yield result

  private def askActor[R](
    createMessage: ActorRef[R] => UserAccountProtocol.Command
  ): Task[R] =
    PekkoInterop.fromFuture { userAccountAggregateRef.ask(createMessage) }
}

Rules for the use case layer:

This is not the place to write business logic. Focus on coordinating processing steps
Describe communication with actors type-safely using ZIO for-comprehension
askActor Standardize Pekko Ask → ZIO Task conversion with a helper
Errors are UserAccountUseCaseError mapped to

Serialization

Event serialization with Protocol Buffers

// event.proto
syntax = "proto3";

message UserAccountCreatedV1 {
  string id = 1;
  string entity_id = 2;
  string name = 3;
  string email_address = 4;
  string occurred_at = 5;
}

message UserAccountRenamedV1 {
  string id = 1;
  string entity_id = 2;
  string old_name = 3;
  string new_name = 4;
  string occurred_at = 5;
}

Serialization policy:

Define both events and snapshots with Protocol Buffers
Automatically generate Scala code with ScalaPB
Convert domain events ↔ Protobuf messages with a custom serializer
Ensure backward compatibility through Protobuf field numbers for versioning

Snapshot strategy

Setting	Value	Reason
Save frequency	Every 1000 events	Balance between startup time and storage
Retention count	Latest 2	Ensure recovery safety

Test strategy

Domain model tests (Pekko-independent)

test("ユーザー名を変更できる") {
  val (user, _) = UserAccount(id, name, email)
  val result = user.rename(newName)
  result match {
    case Right((updated, event)) =>
      updated.name shouldBe newName
      event shouldBe a[UserAccountEvent.Renamed_V1]
    case Left(error) => fail(s"Unexpected error: $error")
  }
}

Actor tests (ActorTestKit)

// PersistenceEffectorのテスト
val probe = testKit.createTestProbe[CreateReply]()
aggregateRef ! Create(id, name, email, probe.ref)
probe.expectMessage(CreateSucceeded(id))

Test isolation

Level	Target	Tool	Pekko dependency
Domain	Business logic	ScalaTest	None
Actor	Messaging, persistence	ActorTestKit	Yes
Serializer	Event/snapshot conversion	ScalaTest	None
Integration	End-to-end	LocalStack	Available

Proper use with related skills

Skills	Focus	When to use
This skill	Concrete implementation with Pekko + Scala 3	When implementing CQRS/ES in Scala
cqrs-tradeoffs	Trade-off analysis for deciding whether to adopt CQRS	When considering whether to introduce CQRS
cqrs-to-event-sourcing	Logical explanation of why ES becomes necessary	When you want to understand the inevitability of ES
cqrs-aggregate-modeling	Redefining aggregate boundaries when introducing CQRS	When reviewing the granularity of aggregates
aggregate-design	General aggregate design rules	When designing new aggregates or reviewing them
domain-building-blocks	Design of VO/Entity/Aggregate, etc.	When designing the domain model as a whole

References

Kato Jun "A Complete Development Guide to Designing CQRS/Event Sourcing Systems with Apache Pekko (or Akka)" - https://tech-book.precena.co.jp/entry/pekko-cqrs-es-guide
Reference implementation: https://github.com/j5ik2o/pekko-cqrs-es-example
pekko-persistence-effector: https://github.com/j5ik2o/pekko-persistence-effector

Related skills (recommended to read together)

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

cqrs-tradeoffs: Trade-off analysis for deciding whether to adopt CQRS/ES
cqrs-to-event-sourcing: The inevitability of moving from CQRS to event sourcing
aggregate-design: Aggregate design rules as a domain model