実践ユースケース ── DCB が解決する5つの典型問題
ユースケース1: 在庫予約（複数エンティティの同時制約）
EC サイトで「同じ商品を同時に複数ユーザーが購入する」場合。
不変条件:
- 在庫数を超えて販売してはならない
- 1ユーザーあたりの購入上限は5個
集約ベースの問題:
Product 集約で在庫を管理、User 集約で購入上限を管理
→ 2つの集約にまたがる → Saga で結果整合性
→ 一時的に在庫を超えて販売される可能性DCB での解決
def reserve_stock(event_store, user_id: str, product_id: str, quantity: int):
# Decision Model: 2つのクエリで必要な情報を収集
stock_events = event_store.query(
event_types=["StockAdded", "StockReserved"],
tags=[f"product:{product_id}"]
)
available = sum(e.data["quantity"] for e in stock_events if e.type == "StockAdded") \
- sum(e.data["quantity"] for e in stock_events if e.type == "StockReserved")
user_events = event_store.query(
event_type="StockReserved",
tags=[f"user:{user_id}", f"product:{product_id}"]
)
user_purchased = sum(e.data["quantity"] for e in user_events)
# ビジネスルール検証
if available < quantity:
raise DomainError(f"在庫不足（残り{available}個）")
if user_purchased + quantity > 5:
raise DomainError(f"購入上限超過（現在{user_purchased}個）")
# 条件付き書き込み（タグに product と user の両方を付与）
event_store.append(
event=DcbEvent(
type="StockReserved",
tags=[f"product:{product_id}", f"user:{user_id}"],
data={"product_id": product_id, "user_id": user_id, "quantity": quantity}
),
preconditions=[
{"query": {"types": ["StockAdded", "StockReserved"], "tags": [f"product:{product_id}"]},
"last_known_position": max_position(stock_events)},
{"query": {"type": "StockReserved", "tags": [f"user:{user_id}", f"product:{product_id}"]},
"last_known_position": max_position(user_events)},
]
)DCB の効果: 在庫制約とユーザー購入上限の両方を1つのトランザクションで保証。Saga 不要。一時的な不整合なし。
ユースケース2: 一意性制約（メールアドレスの重複防止）
不変条件:
メールアドレスはシステム全体で一意
集約ベースの問題:
User 集約は1人のユーザーの中だけ → 全ユーザーのメールアドレスを知らない
→ 別途一意性チェック用のリードモデル + 結果整合性
→ 競合時にメールが重複する可能性DCB での解決
def register_user(event_store, email: str, name: str):
# メールアドレスのタグでクエリ
email_tag = f"email:{email.lower()}"
existing = event_store.query(
event_type="UserRegistered",
tags=[email_tag]
)
if len(existing) > 0:
raise DomainError(f"メールアドレス {email} は既に使用されています")
user_id = generate_id()
event_store.append(
event=DcbEvent(
type="UserRegistered",
tags=[f"user:{user_id}", email_tag],
data={"user_id": user_id, "email": email, "name": name}
),
preconditions=[
# 「このメールアドレスで登録されたイベントがないこと」を保証
{"query": {"type": "UserRegistered", "tags": [email_tag]},
"last_known_position": 0, # 0件であること
"expected_count": 0}
]
)DCB の効果: メールの一意性を即時一貫性で保証。リードモデルとの遅延による競合がない。
ユースケース3: 予算管理（累積制約）
不変条件:
部署の月間予算を超えて承認してはならない
各承認は個別の経費申請（異なるエンティティ）
集約ベースの問題:
ExpenseRequest 集約は個別の申請
Department 集約に全承認を集約? → 全申請がシリアライズされるDCB での解決
def approve_expense(event_store, department_id: str, request_id: str, amount: int):
month_tag = f"budget:{department_id}:2026-04"
approved = event_store.query(
event_type="ExpenseApproved",
tags=[month_tag]
)
total_approved = sum(e.data["amount"] for e in approved)
budget_limit = get_monthly_budget(department_id) # 例: 500万円
if total_approved + amount > budget_limit:
raise DomainError(
f"月間予算超過（使用済み: {total_approved}円、"
f"残り: {budget_limit - total_approved}円）"
)
event_store.append(
event=DcbEvent(
type="ExpenseApproved",
tags=[month_tag, f"request:{request_id}", f"dept:{department_id}"],
data={"request_id": request_id, "amount": amount, "department_id": department_id}
),
preconditions=[
{"query": {"type": "ExpenseApproved", "tags": [month_tag]},
"last_known_position": max_position(approved)}
]
)DCB の効果: 月のタグで自然に時間範囲が区切られる。異なる月の承認は互いに競合しない。
ユースケース4: ダブルブッキング防止（時間帯の重複）
不変条件:
同じ会議室の同じ時間帯に重複する予約を許可しない
集約ベースの問題:
MeetingRoom 集約に全予約を集約 → 全予約がシリアライズ
→ 異なる日の予約まで競合するDCB での解決
def book_room(event_store, room_id: str, date: str, start: str, end: str, user_id: str):
# その日のその部屋の予約だけクエリ
day_tag = f"room:{room_id}:date:{date}"
bookings = event_store.query(
event_type="RoomBooked",
tags=[day_tag]
)
# 時間帯の重複チェック
for booking in bookings:
if times_overlap(start, end, booking.data["start"], booking.data["end"]):
raise DomainError(
f"時間帯が重複しています: "
f"{booking.data['start']}〜{booking.data['end']}"
)
event_store.append(
event=DcbEvent(
type="RoomBooked",
tags=[day_tag, f"room:{room_id}", f"user:{user_id}"],
data={"room_id": room_id, "date": date, "start": start, "end": end}
),
preconditions=[
{"query": {"type": "RoomBooked", "tags": [day_tag]},
"last_known_position": max_position(bookings)}
]
)DCB の効果: タグに日付を含めることで、異なる日の予約が競合しない。同じ日の同じ部屋の予約だけが前提条件で競合する。
ユースケース5: ワークフロー承認（状態遷移の制約）
不変条件:
承認フローは「申請→上長承認→部長承認→完了」の順序を守る
既に承認済みのステップを巻き戻してはならない
同じステップを2回承認してはならないDCB での解決
def approve_step(event_store, workflow_id: str, step: str, approver_id: str):
VALID_TRANSITIONS = {
"manager_approved": ["submitted"],
"director_approved": ["manager_approved"],
"completed": ["director_approved"],
}
workflow_events = event_store.query(
tags=[f"workflow:{workflow_id}"]
)
current_state = workflow_events[-1].type if workflow_events else None
required_previous = VALID_TRANSITIONS.get(step, [])
if current_state not in required_previous:
raise DomainError(
f"無効な状態遷移: {current_state} → {step}"
)
# 同じステップの重複承認を防止
already_approved = any(e.type == step for e in workflow_events)
if already_approved:
raise DomainError(f"ステップ '{step}' は既に承認済みです")
event_store.append(
event=DcbEvent(
type=step,
tags=[f"workflow:{workflow_id}", f"approver:{approver_id}"],
data={"workflow_id": workflow_id, "approver_id": approver_id}
),
preconditions=[
{"query": {"tags": [f"workflow:{workflow_id}"]},
"last_known_position": max_position(workflow_events)}
]
)ユースケースの共通パターン
全てのユースケースに共通するのは「判断に必要な情報の範囲と、一貫性を保つ範囲を一致させる」ことだ。
集約: 「構造的に」境界を引く → ユースケースと一致しないことがある
DCB: 「操作ごとに」境界を引く → ユースケースと自然に一致する| ユースケース | 集約ベースの問題 | DCB での解決 |
|---|---|---|
| 在庫予約 | 2集約 → Saga | タグで両制約を同時検証 |
| 一意性制約 | 集約の外 → リードモデル | メールタグで即時一貫性 |
| 予算管理 | 全承認がシリアライズ | 月タグで範囲を限定 |
| ダブルブッキング | 全予約がシリアライズ | 日タグで競合を最小化 |
| ワークフロー | 状態管理が複雑 | イベント履歴で自然に遷移 |