Posted inDevelopment

Node.js + PostgreSQL で実現する Event Sourcing & CQRS：結果だけでなく「過程」を記録せよ

Development tutorial - IT technology blog
Development tutorial - IT technology blog

なぜあえて Event Sourcing と CQRS を採用するのか？

CRUD（Create, Read, Update, Delete）という伝統的な考え方は非常に馴染み深いものです。ユーザーが名前を変更したとき、単に UPDATE コマンドを実行すれば完了です。しかし、この操作は意図せず古い痕跡を消し去ってしまいます。顧客から「なぜ午前2時に残高が50万ドン減っているのか？」と聞かれた際、DBに最終的な数字しか残っていないという事態は避けたいはずです。

私は以前、1日5万件以上のトランザクションを処理する電子マネーシステムの再構築に携わりました。その時の教訓は、適切な監査ログ（audit log）の仕組みがなければ、デバッグやデータの照合が悲劇に変わるということです。そこで Event Sourcing と CQRS が真価を発揮します。静的な状態を保存する代わりに、発生したすべての行動を記録します。同時に、読み取りと書き込みのモデルを分離することで、アクセスが急増してもシステムがボトルネックにならないようにします。

クイック実装：PostgreSQL を Event Store に変える

最初から Kafka や EventStoreDB を使う必要はありません。PostgreSQL の JSONB 型は非常に強力で、中規模から大規模のプロジェクトにおいてこの役割を十分に果たすことができます。

ステップ1：イベント保存用テーブルの設計

過去を再現するために、テーブル構造は最小限かつ十分な情報を持つ必要があります。

CREATE TABLE events (
    id BIGSERIAL PRIMARY KEY,
    aggregate_id UUID NOT NULL, -- 例：Order ID または User ID
    event_type VARCHAR(50) NOT NULL, -- MoneyDeposited, OrderCancelled
    payload JSONB NOT NULL, -- イベントの生データ
    version INT NOT NULL, -- データ競合の防止（並行性制御）
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

ステップ2：Node.js での Command Side（書き込み側）の処理

アカウントへの入金イベントを記録する方法を見てみましょう。

async function depositMoney(userId, amount) {
    const event = {
        aggregate_id: userId,
        event_type: 'MoneyDeposited',
        payload: { amount, currency: 'VND' },
        version: await getNextVersion(userId)
    };

    await db.query(
        'INSERT INTO events (aggregate_id, event_type, payload, version) VALUES ($1, $2, $3, $4)',
        [event.aggregate_id, event.event_type, event.payload, event.version]
    );
    
    console.log(`[Event Saved] ユーザー ${userId} が ${amount} VND の入金に成功しました`);
}

組み合わせの解明：過去が現在を定義する

Event Sourcing：過去こそが唯一の真実

この世界では、口座残高は DB 内の固定された数値ではありません。それは全取引履歴を再実行（replay）した結果です。

  • 取引 1: +1,000,000đ
  • 取引 2: -200,000đ
  • 取引 3: +50,000đ
  • => 現在の状態: 850,000đ

計算ロジックにエラーがあった場合でも、コードを修正して再実行（replay）するだけで済みます。データは上書きされて消える心配がなく、常に絶対的に正確な状態を保てます。

CQRS：分離によるブレイクスルー

CQRS (Command Query Responsibility Segregation) は、アプリケーションを2つの独立したブランチに分けます：

  1. Write Side (Command): リクエストを受け取り、ロジックを検証し、イベントを書き込みます。このブランチはデータの整合性を優先します。
  2. Read Side (Query): 高速なクエリのために非正規化（denormalized）されたデータベースを使用します。

実際には、ユーザーがアプリを開くたびに100万件のイベントを再実行して残高を確認するのは不可能です。そのため、最新のイベントを監視して user_balances テーブルを更新するワーカー（Worker）が必要です。これにより、フロントエンド側は単純な SELECT 命令を実行するだけで済みます。

Read Model（プロジェクション）の最適化

これは、システムが最終的な整合性（Eventual Consistency）を達成するための架け橋となります。

async function projectMoneyDeposited(event) {
    // ユーザーが新しい残高を確認できるよう、Read Model テーブルを即座に更新する
    await db.query(
        'UPDATE user_balances SET balance = balance + $1, updated_at = NOW() WHERE user_id = $2',
        [event.payload.amount, event.aggregate_id]
    );
}

システムをよりスムーズにするために、Redis Pub/Sub や RabbitMQ の使用を検討してください。Write Side がイベントを保存するとすぐに、Read Side が信号を受け取り、数ミリ秒以内にデータを更新します。

実践経験：回避すべき落とし穴

この2つを実際のプロジェクトに適用するのは、理論ほど簡単ではありません。私が学んだ3つの重要な注意点を以下に挙げます：

1. 最初からオーバースペックにしない

Todo リストの管理アプリやニュースサイトを作っているだけなら、CRUD を使ってシンプルに進めましょう。Event Sourcing が真に価値を発揮するのは、ビジネスロジックが非常に複雑な場合や、厳格な監査（Audit）が必要な場合のみです。

2. スナップショット – パフォーマンスのための解決策

ロイヤルカスタマーが5,000件以上の取引を行っている場合、再実行（replay）は遅くなり始めます。解決策はスナップショット（Snapshot）を使用することです。100イベントごとに、その時点の状態を保存します。残高を計算する必要があるときは、最新のスナップショットを取得し、その後に発生したいくつかの新しいイベントを加算するだけで済みます。

3. バージョニングと不変性

一度書き込まれたイベントデータは変更できません（Immutable）。しかし、ビジネス要件は常に変化します。新しいイベントに必須フィールドを追加する場合は、古いコードがバージョン 1.0 のイベントを処理してもシステムがクラッシュしないように注意してください。

結び

Node.js と PostgreSQL を使った Event Sourcing と CQRS の実装はそれほど難しくありません。最も難しいのは、思考を「状態の保存」から「行動의記録」へと転換することです。この技術をマスターすれば、高い信頼性が求められる大規模なシステムの設計に自信が持てるようになるでしょう。

すべてを INSERT するために UPDATE を捨てる準備はできていますか？ 次のプロジェクトの小さなモジュールで試して、その違いを実感してみてください。

Share:
Tags: