単一データベーステーブルでのイベントソーシング：簡略化されたアプローチ

はじめに

現代のソフトウェアアーキテクチャの進化する状況において、堅牢で監査可能でスケーラブルなシステムの必要性は最優先事項です。アプリケーションの状態のすべての変更を不変のイベントのシーケンスとして記録する強力なパターンであるイベントソーシングは、大きな注目を集めています。監査、デバッグ、および複雑なビジネスルールを持つシステムの構築に比類のないメリットを提供します。従来、イベントソーシングの実装は、イベントログとして機能するためにKafkaのような洗練されたメッセージングシステムを必要とすることがよくありました。これは非常に効果的ですが、Kafkaの統合と管理は、特に小規模なプロジェクトや専任のDevOpsリソースを持たないチームにとって、かなりの運用オーバーヘッドと複雑さを導入する可能性があります。この記事では、代替の簡略化されたアプローチ、つまり「イベントログ」として単一のリレーショナルデータベーステーブルのみを使用してイベントソーシングを実装することについて掘り下げます。この方法は、イベントソーシングをわかりやすくし、アクセスしやすくすることを目的としており、外部メッセージブローカーのオーバーヘッドなしでそのコアメリットを活用できることを証明しています。これは、運用のシンプルさが優先される場合に特に魅力的なアプローチです。

コアコンセプトの理解

実装に飛び込む前に、イベントソーシングの理解に不可欠ないくつかの基本的な概念を明確にしましょう。

イベントソーシング: このアーキテクチャパターンは、エンティティの現在の状態を保存するのではなく、その状態のすべての変更を不変のイベントとして保存することを規定します。現在の状態は、これらのイベントを時系列順に再生することによって導き出されます。
イベント: 過去に発生したことの記録。イベントは不変の事実です。過去形で名前が付けられています（例：OrderPlaced、InventoryAdjusted、UserRegistered）。各イベントには通常、timestamp、event_type、aggregate_id、versionのようなメタデータと、その特定の変更に関連するデータを示すpayloadが含まれます。
集約 (Aggregate): データ変更の単一の単位として扱えるドメインオブジェクトのクラスター。ドメイン駆動設計における一貫性の境界です。イベントは常に特定の集約インスタンス（aggregate_idで識別）に関連付けられます。
状態の再構築: 集約の過去のイベントを再生して現在の状態を再構築するプロセス。
スナップショット: 特定の時点での集約の事前計算された状態。スナップショットは、特に長期間のイベント履歴を持つ集約の場合、最初からすべての過去のイベントを再生する必要を回避することで、状態再構築を最適化するために使用されます。

単一データベーステーブルでのイベントソーシングの実装

コアアイデアはシンプルです。すべてのイベントは、そのタイプや属する集約に関係なく、単一の追記専用データベーステーブルに保存されます。このテーブルは、私たちの中心的で権威のあるイベントログとして機能します。

イベントログテーブルのスキーマ設計

すべてのドメインイベントを格納する events テーブルを検討してみましょう。

CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    version INT NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    timestamp TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    metadata JSONB,
    UNIQUE (aggregate_id, version) -- Ensures event order and prevents duplicate events for an aggregate
);

CREATE INDEX idx_events_aggregate_id ON events (aggregate_id);
CREATE INDEX idx_events_timestamp ON events (timestamp);

id: イベント自体のユニークな識別子（例：シリアルプライマリキー）。
aggregate_id: このイベントが適用される集約インスタンスのID。フィルタリングに重要です。
aggregate_type: 集約のタイプまたは名前（例：「Order」、「UserAccount」）。クエリとイベントストリームの理解に役立ちます。
version: このイベントが適用された後の集約のバージョン。楽観的同時実行制御に不可欠です。
event_type: イベントの特定のタイプ（例：OrderCreated、ItemAddedToOrder）。
payload: イベントの実際のデータを格納するためのJSONB列。JSONBは、効率的なストレージとクエリ機能のために推奨されます。
timestamp: イベントが発生した時刻。
metadata: 追加の運用メタデータ（例：アクションを開始したuser_id、ip_address）のためのオプションのJSONB列。

コア操作

イベントソーシングシステムには、主に2つの操作があります。

イベントの追加: ビジネス操作が発生すると、新しいイベントが生成され、イベントログに追加されます。
集約状態の読み込み: 集約に対して操作を実行するには、イベントを再生してその現在の状態を再構築する必要があります。

Pythonでの概念的な Order 集約を使用してこれらを説明しましょう。

1. イベントと集約の定義

import uuid
import datetime
import json
from typing import List, Dict, Any, Type

# --- Event Definitions ---
class Event:
    def __init__(self, aggregate_id: uuid.UUID, version: int, timestamp: datetime.datetime, payload: Dict[str, Any]):
        self.aggregate_id = aggregate_id
        self.version = version
        self.timestamp = timestamp
        self.payload = payload

    @property
    def event_type(self) -> str:
        return self.__class__.__name__

    def to_dict(self) -> Dict[str, Any]:
        return {