本文へ移動
cccskills
無料GitHub で公開

pekko-cqrs-es-implementation

Apache Pekko + Scala 3によるCQRS/Event Sourcing実装ガイド。

PersistenceEffectorを用いた集約アクター、ドメインモデルとアクターの分離、 状態遷移の型安全な表現、イベント設計、Protocol Buffersシリアライズ、 ZIOベースのユースケース層、リードモデルアップデータの実装パターンを提供する。対象言語: Scala 3限定。

CQRS/Event Sourcingアーキテクチャが前提の場合のみ使用。トリガー条件: 「Scala」かつ「CQRS」または「Event Sourcing」または「Pekko」が リクエストに含まれる場合のみ起動。Scala以外の言語やCQRS/ES以外のアーキテクチャでは このスキルを使用してはならない。

続きを読む（あと約 280 字）

トリガー：「PekkoでCQRS/ESを実装したい」「Scalaで集約アクターを書きたい」 「PersistenceEffectorの使い方」「Pekkoのイベントソーシング」 「ScalaでCQRSのコマンド側を実装」「Pekkoで状態遷移を管理」 といったPekko + Scala + CQRS/ES実装リクエストで起動。

非トリガー：「CQRSのトレードオフ」「イベントソーシングとは」「JavaでCQRS」 「GoでEvent Sourcing」「CQRSの概念を教えて」など、Scala/Pekko以外や概念的な質問では 起動してはならない。

j5ik2o/okite-ai
インストール方法を見る

含まれるファイル（1）

  • SKILL.md19.3 KB

SKILL.md（原文）

インストールする前に、エージェントに与えられる指示の中身を確認できます。

Pekko CQRS/ES 実装ガイド

Apache Pekko + Scala 3でCQRS/Event Sourcingを実装する際の具体的なパターン集。

適用条件

このスキルは以下のすべてを満たす場合にのみ使用する：

  1. プログラミング言語がScala（3.x推奨）であること
  2. CQRS/Event Sourcingアーキテクチャを採用していること
  3. Apache Pekko（またはAkka）をアクターフレームワークとして使用していること

上記を満たさない場合は、cqrs-tradeoffs、cqrs-to-event-sourcing、cqrs-aggregate-modelingなど 言語非依存のスキルを使用すること。

アーキテクチャ概要

システム全体のデータフロー

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

コマンド側とクエリ側の分離

コマンド側クエリ側
目的ビジネスルール実行効率的なデータ取得
データストアDynamoDB（イベントストア）PostgreSQL（リードモデル）
APIGraphQL MutationGraphQL Query
レイヤードメイン → ユースケース → インターフェースアダプタインターフェースアダプタのみ
整合性強い整合性（集約内）結果整合性

クエリ側にユースケース層はない。 GraphQL自体がユースケースに相当する。

モジュール構成

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

依存方向

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

コマンド側実装パターン

1. ドメインモデル

原則: ドメインモデルは純粋なScalaコードで、Pekkoに一切依存しない。

集約（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))
      }
  }
}

重要なパターン:

  • 状態変更メソッドは Either[Error, (NewState, Event)] を返す
  • ファクトリも (State, Event) のペアを返す
  • 実装クラスは private で外部から直接構築できない
  • ドメインモデル内でPekkoのimportは一切ない

イベント定義（enum + バージョニング）

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

イベント設計ルール:

ルール説明例
過去形で命名「何が起きたか」を表すCreated, Renamed, Deleted
_V1サフィックススキーマ進化に対応Renamed_V1, Renamed_V2
不変case classで自動的に保証
自己完結変更前後の値を含むoldName, newName
必須フィールドid, entityId, occurredAtすべてのイベントに共通

2. 集約の状態（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")
  }
}

状態遷移の型安全性:

  • NotCreated → Created（Created_V1イベントのみ）
  • Created → Created（Renamed_V1）/ Deleted（Deleted_V1）
  • Deleted → 遷移なし（どのイベントも受け付けない）

3. プロトコル定義（コマンド/リプライ）

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
}

プロトコル設計ルール:

  • コマンドはすべて sealed trait Command を継承し、id を持つ
  • リプライはコマンドごとに専用の sealed trait を定義する（CreateReply, RenameReply等）
  • replyTo: ActorRef[XxxReply] で型安全な応答を保証
  • 成功/失敗をcase classで表現し、パターンマッチで網羅性チェック

4. 集約アクター（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)
    }
  }
}

アクター実装ルール:

  • ドメインロジックをアクターに書かない。 アクターは永続化とライフサイクル管理に徹する
  • 状態ごとにハンドラ関数を分離し、受け付けるコマンドを限定する
  • effector.persistEvent(event) { _ => ... } でイベント永続化後にリプライ
  • ドメインモデルが Left を返したらリプライのみ（永続化しない）
  • ドメインモデルが Right を返したらイベントを永続化してからリプライ

5. ユースケース層（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) }
}

ユースケース層のルール:

  • ビジネスロジックを書く場所ではない。 処理ステップの調整役に徹する
  • ZIOのfor式でアクターとの通信を型安全に記述
  • askActor ヘルパーでPekko Ask → ZIO Task変換を共通化
  • エラーは UserAccountUseCaseError にマッピング

シリアライズ

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;
}

シリアライズ方針:

  • イベントとスナップショットの両方をProtocol Buffersで定義
  • ScalaPBでScalaコードを自動生成
  • カスタムシリアライザでドメインイベント ↔ Protobufメッセージを変換
  • バージョニングはProtobufのフィールド番号で後方互換性を確保

スナップショット戦略

設定値理由
保存頻度1000イベントごと起動時間とストレージのバランス
保持数最新2つリカバリ安全性の確保

テスト戦略

ドメインモデルテスト（Pekko非依存）

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

アクターテスト（ActorTestKit）

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

テストの分離

レベル対象ツールPekko依存
ドメインビジネスロジックScalaTestなし
アクターメッセージング、永続化ActorTestKitあり
シリアライザイベント/スナップショットの変換ScalaTestなし
統合エンドツーエンドLocalStackあり

関連スキルとの使い分け

スキルフォーカス使うタイミング
本スキルPekko + Scala 3での具体的実装CQRS/ESをScalaで実装するとき
cqrs-tradeoffsCQRS採用判断のトレードオフ分析CQRS導入の是非を検討するとき
cqrs-to-event-sourcingなぜESが必要になるかの論理的説明ESの必然性を理解したいとき
cqrs-aggregate-modelingCQRS導入時の集約境界再定義集約の粒度を見直すとき
aggregate-design集約設計ルール全般集約の新規設計やレビューのとき
domain-building-blocksVO/Entity/Aggregate等の設計ドメインモデル全体を設計するとき

参考文献

関連スキル（併読推奨）

このスキルを使用する際は、以下のスキルも併せて参照すること：

  • cqrs-tradeoffs: CQRS/ES採用判断のトレードオフ分析
  • cqrs-to-event-sourcing: CQRSからイベントソーシングへの必然性
  • aggregate-design: ドメインモデルとしての集約設計ルール

レビュー

まだレビューはありません。使ってみた感想をお寄せください。

ログインしてレビューを書く

同じリポジトリのスキル

概要と使いどころ

aggregate-design

無料日本語概要

DDDの集約(Aggregate)設計ルールに基づくコードレビュー・設計支援・リファクタリングを行う。 Evans Rules、Vernon's 4 Rules、Design by Contractに基づき、集約の境界定義、不変条件の検証、 不変(Immutable)設計、ID参照、結果整合性、ドメインイベント連携を包括的にガイドする。 以下のいずれかに該当する場合は必ずこのスキルを使用すること： - 集約(Aggregate)の新規設計・実装・リファクタリング（どの言語でも） - 既存の集約やエンティティクラスのDDD観点でのコードレビュー - 集約の境界決定（「AとBは同じ集約にすべきか？」「この集約は大きすぎるか？」） - 集約内の不変条件・整合性境界の設計 - 集約間の連携方式の判断（ドメインイベント、結果整合性、Sagaパターン） - 可変(Mutable)な集約コードを不変(Immutable)設計にリファクタリングする - publicフィールド、直接参照、push/appendなどカプセル化違反の検出・修正 キーワード例：集約、Aggregate、aggregate boundary、集約ルート、AggregateRoot、 エンティティ設計、DDD実装、Vernon Rules、Evans Rules、集約の分割、真の不変条件

j5ik2o/okite-ai802026年4月25日 更新

aggregate-transaction-boundary

無料日本語概要

集約とトランザクション境界の関係を明確化し、複数集約を単一トランザクションに含めるアンチパターンを 検出・是正する。集約は強い整合性境界であり、ユースケースで複数集約を更新する場合は結果整合性を 使うべきという原則を適用する。コードレビュー、ユースケース設計、リファクタリング時に トランザクション境界の問題を検出する場合に使用。 対象言語: 言語非依存（Java, Kotlin, Scala, TypeScript, Go, Rust, Python等すべて）。 トリガー：「複数集約を同じトランザクションで更新している」「ユースケースに@Transactionalがある」 「集約間の整合性をどう取るか」「Sagaパターンを使うべきか」「トランザクション境界の設計」 「1トランザクション1集約」「結果整合性の実装」「集約をまたぐトランザクション」 といったトランザクション境界関連リクエストで起動。

j5ik2o/okite-ai802026年4月25日 更新

backward-compat-governance

無料日本語概要

後方互換性がゴミコードを量産する構造を検出し、互換性を「契約と撤去計画」として管理する ガバナンスを支援するスキル。公開API境界の明確化、非推奨化サイクル（deprecation cycle）の 制度化、互換層の局所化（Adapter/Strangler Fig）、契約テスト（CDC）による互換性検証、 AI生成コードの互換性ゲート設計を含む。コードレビュー、API設計、リファクタリング、 レガシー移行時に互換性起因の技術的負債を防ぐために使用。 対象言語: 言語非依存（Java, TypeScript, Go, Python, Rust等すべて）。 トリガー：「後方互換性を保ちたい」「非推奨APIをどうする」「互換層が増えてきた」 「レガシー移行の戦略」「API設計レビュー」「互換性のためのコードが多い」 「deprecation policyを作りたい」「破壊的変更の管理」といった互換性管理関連リクエストで起動。

j5ik2o/okite-ai802026年4月25日 更新

breach-encapsulation-naming

無料日本語概要

getterの濫用を防ぐための命名規約スキル。ドメインモデルでgetterが必要な場合（永続化、JSON変換など）に `breachEncapsulationOf` プレフィックスを付与することで、カプセル化を破っていることを明示する。 これにより、Tell Don't Ask原則の違反を未然に防ぎ、getterの意図しない使用を抑制する。 コードレビュー、新規実装、リファクタリング時にgetter設計が必要な場合に使用。 対象言語: Java, Kotlin, Scala, TypeScript, Python, Go, Rust。 トリガー：「getterの命名規約」「カプセル化を破るgetter」「永続化用のgetter」 「breachEncapsulation」「getterを作りたいが濫用を防ぎたい」といったgetter命名関連リクエストで起動。

j5ik2o/okite-ai802026年4月25日 更新

clean-architecture

無料日本語概要

クリーンアーキテクチャを採用しているプロジェクト向けの設計・レビュー支援。4層構造（ドメイン層、 ユースケース層、インターフェースアダプタ層、インフラストラクチャ層）に基づく。特にインフラ層は 横断的関心事（ロギング、設定管理）のみ、永続化やRPCはインターフェースアダプタ層に配置すべき という原則を適用する。トリガー：「クリーンアーキテクチャで」「クリーンアーキテクチャに従って」 「クリーンアーキテクチャのレビュー」など、クリーンアーキテクチャを明示的に指定した場合のみ起動。 一般的な「設計レビュー」「アーキテクチャ相談」では起動しない。

j5ik2o/okite-ai802026年4月25日 更新

cqrs-aggregate-modeling

無料日本語概要

CQRS/ESが集約の境界定義とモデリングに与える影響を解説する。CQRSを導入すると集約は コマンド実行に必要な最小限の状態のみ保持すればよくなり、読み取り責務はリードモデルに 委譲できる。大きすぎる集約の軽量化、集約境界の再定義、イベントによる状態管理を支援する。 集約設計、CQRS導入時のモデリング見直し、パフォーマンス問題の解決時に使用。 対象言語: 言語非依存。 トリガー：「CQRSで集約が変わる」「集約が大きすぎる」「集約にメッセージ1000件」 「集約の更新が重い」「CQRS導入で集約を見直す」「集約を軽量化したい」 「集約にクエリ用データが混ざっている」「集約の境界を再定義」 といったCQRS/モデリング関連リクエストで起動。

j5ik2o/okite-ai802026年4月25日 更新

j5ik2o のスキルをすべて見る

このスキルの問題を報告する