DCB の仕組み ── タグ・クエリ・楽観的ロック

---

イベントの構造：タグ付きイベント

DCB のイベントストアでは、イベントはグローバルに順序付けられた1本のストリームに格納される。集約ごとのストリーム分割は行わない。

interface DcbEvent {
  // グローバル位置（単調増加）
  position: number;

  // イベントタイプ
  type: string;

  // タグ（複数可）
  tags: string[];

  // イベントデータ
  data: Record<string, unknown>;

  // メタデータ
  metadata: {
    timestamp: string;
    causation_id?: string;
    correlation_id?: string;
  };
}

実際のイベント例（講座予約）:

Position 1: { type: "CourseCreated",    tags: ["course:C1"],               data: { name: "DDD入門", capacity: 30 } }
Position 2: { type: "StudentCreated",   tags: ["student:S1"],              data: { name: "田中太郎" } }
Position 3: { type: "StudentRegistered", tags: ["course:C1", "student:S1"], data: { course_id: "C1", student_id: "S1" } }
Position 4: { type: "StudentRegistered", tags: ["course:C1", "student:S2"], data: { course_id: "C1", student_id: "S2" } }
Position 5: { type: "StudentCreated",   tags: ["student:S3"],              data: { name: "鈴木花子" } }
Position 6: { type: "StudentRegistered", tags: ["course:C2", "student:S1"], data: { course_id: "C2", student_id: "S1" } }

---

イベントの読み取り：動的クエリ

「講座 C1 の登録者数」を知りたい場合

クエリ:
  type = "StudentRegistered" AND tags CONTAINS "course:C1"

結果:
  Position 3: { student_id: "S1" }
  Position 4: { student_id: "S2" }
  → 2名が登録済み（最終位置: 4）

「受講者 S1 の登録講座数」を知りたい場合

クエリ:
  type = "StudentRegistered" AND tags CONTAINS "student:S1"

結果:
  Position 3: { course_id: "C1" }
  Position 6: { course_id: "C2" }
  → 2講座に登録済み（最終位置: 6）

「受講者 S3 が講座 C1 に登録できるか」を判断する場合

2つのクエリを組み合わせる：

# Decision Model の構築
def build_registration_decision(event_store, student_id: str, course_id: str):
    # クエリ1: 講座の登録者数
    course_events = event_store.query(
        event_type="StudentRegistered",
        tags=[f"course:{course_id}"]
    )
    course_count = len(course_events)
    course_last_position = course_events[-1].position if course_events else 0

    # クエリ2: 受講者の登録講座数
    student_events = event_store.query(
        event_type="StudentRegistered",
        tags=[f"student:{student_id}"]
    )
    student_count = len(student_events)
    student_last_position = student_events[-1].position if student_events else 0

    return {
        "course_count": course_count,
        "student_count": student_count,
        "last_known_position": max(course_last_position, student_last_position),
        "queries": [
            {"type": "StudentRegistered", "tags": [f"course:{course_id}"]},
            {"type": "StudentRegistered", "tags": [f"student:{student_id}"]},
        ]
    }

---

一貫性の保証：条件付き書き込み

DCB の楽観的ロックは、集約のストリームバージョンではなく、クエリ結果に基づく前提条件で行われる。

def register_student(event_store, student_id: str, course_id: str):
    # 1. Decision Model を構築
    decision = build_registration_decision(event_store, student_id, course_id)

    # 2. ビジネスルールを検証
    if decision["course_count"] >= 30:
        raise DomainError("講座が満員です")
    if decision["student_count"] >= 10:
        raise DomainError("受講上限に達しています")

    # 3. 条件付き書き込み
    new_event = DcbEvent(
        type="StudentRegistered",
        tags=[f"course:{course_id}", f"student:{student_id}"],
        data={"course_id": course_id, "student_id": student_id}
    )

    try:
        event_store.append(
            event=new_event,
            preconditions=[
                # 「私がクエリした後に、同じ条件のイベントが追加されていないこと」
                {
                    "query": {"type": "StudentRegistered", "tags": [f"course:{course_id}"]},
                    "last_known_position": decision["last_known_position"]
                },
                {
                    "query": {"type": "StudentRegistered", "tags": [f"student:{student_id}"]},
                    "last_known_position": decision["last_known_position"]
                },
            ]
        )
    except PreconditionFailedError:
        # 他のリクエストが先に書き込んだ → リトライ
        return register_student(event_store, student_id, course_id)

競合の検出メカニズム

シナリオ: 2つのリクエストがほぼ同時に到着

Request A: 受講者 S3 を講座 C1 に登録（現在 29名）
Request B: 受講者 S4 を講座 C1 に登録（現在 29名）

Request A のフロー:
  1. クエリ → course:C1 の登録数 = 29（最終位置 P99）
  2. 29 < 30 → OK
  3. 条件付き書き込み → P99 以降に course:C1 の StudentRegistered はない → 成功
  4. 新イベント書き込み（Position P100）

Request B のフロー:
  1. クエリ → course:C1 の登録数 = 29（最終位置 P99）
  2. 29 < 30 → OK
  3. 条件付き書き込み → P99 以降に course:C1 の StudentRegistered がある（P100）
  4. → PreconditionFailedError！
  5. リトライ → 再クエリ → 登録数 = 30 → 「講座が満員です」

→ 30名の上限が確実に守られる

---

集約の楽観的ロック vs DCB の楽観的ロック

集約ベース:
  ストリーム "course-C1" のバージョン = 29
  → 書き込み時に「バージョンが 29 であること」を前提条件にする
  → course-C1 への全てのイベント（登録以外も含む）が競合対象

DCB ベース:
  「type=StudentRegistered AND tags=course:C1」のクエリ結果の最終位置
  → 書き込み時に「このクエリの最終位置が変わっていないこと」を前提条件にする
  → 登録に関係するイベントだけが競合対象
  → 講座名の変更や説明文の更新は競合しない

DCB の方が競合が少ない。集約ベースでは同じストリーム内の全操作がシリアライズされるが、DCB では同じ判断に影響するイベントだけが競合する。

---

書き込みの並行性

集約ベース:
  course-C1 ストリームへの書き込みは全てシリアル
  → 登録、講座名変更、説明文更新が全て直列化

DCB ベース:
  「登録」と「講座名変更」は異なるタグ/タイプ → 並行して書き込み可能
  「登録」同士のみが前提条件で競合する

→ 不要なシリアライゼーションが排除される

---

まとめ

DCB の構成要素	役割
タグ付きイベント	1つのイベントを複数の次元でスライス可能にする
動的クエリ	判断に必要なイベントだけを実行時に選択する
Decision Model	クエリ結果からビジネスルールを検証する
条件付き書き込み	クエリ後にイベントが追加されていないことを保証する
リトライ	前提条件が失敗したらクエリからやり直す

次章では、これらの仕組みが実際のユースケースでどう活きるかを具体的に見ていく。

＋ － ⟳ ✕