イベントドリブンなワークフローエンジンをゼロから実装して仕組みを理解する

def process_order(order):
    validate_order(order)        # ステップ1: 注文の検証
    charge_payment(order)        # ステップ2: 決済
    ship_order(order)            # ステップ3: 発送
    send_confirmation(order)     # ステップ4: 確認メール

from __future__ import annotations
 
import time
from dataclasses import dataclass
from typing import Any
 
 
# ──────────────────────────────────────────────
# イベント：「起きたこと」を表すイミュータブルなデータ
# frozen=True により、生成後のフィールド変更を禁止する
# ──────────────────────────────────────────────
 
@dataclass(frozen=True)
class WorkflowStarted:
    workflow_id: str
    input_data: Any
    timestamp: float
 
@dataclass(frozen=True)
class TaskScheduled:
    task_id: str
    task_name: str
    input_data: Any
    timestamp: float
 
@dataclass(frozen=True)
class TaskCompleted:
    task_id: str
    result: Any
    timestamp: float
 
@dataclass(frozen=True)
class TaskFailed:
    task_id: str
    error: str
    timestamp: float
 
@dataclass(frozen=True)
class TimerScheduled:
    timer_id: str
    fire_at: float          # Unix タイムスタンプ（秒）
    timestamp: float
 
@dataclass(frozen=True)
class TimerFired:
    timer_id: str
    timestamp: float
 
@dataclass(frozen=True)
class ExternalEventReceived:
    event_name: str
    payload: Any
    timestamp: float
 
@dataclass(frozen=True)
class WorkflowCompleted:
    result: Any
    timestamp: float
 
@dataclass(frozen=True)
class WorkflowFailed:
    error: str
    timestamp: float
 
 
# すべてのイベント型をまとめた型エイリアス（Union 型）
WorkflowEvent = (
    WorkflowStarted | TaskScheduled | TaskCompleted | TaskFailed
    | TimerScheduled | TimerFired | ExternalEventReceived
    | WorkflowCompleted | WorkflowFailed
)

from collections import defaultdict
 
 
class EventStore:
    """追記専用（append-only）のイベントストア。
 
    ワークフロー ID ごとにイベントのリストを保持する。
    一度記録されたイベントは変更も削除もしない——
    これがイベントソーシングの根幹ルール。
    """
 
    def __init__(self) -> None:
        # dict のキーが workflow_id、値がイベントのリスト
        self._streams: dict[str, list[WorkflowEvent]] = defaultdict(list)
 
    def append(self, workflow_id: str, events: list[WorkflowEvent]) -> None:
        """イベントを末尾に追記する。既存イベントには触れない。"""
        self._streams[workflow_id].extend(events)
 
    def get_events(self, workflow_id: str) -> list[WorkflowEvent]:
        """指定ワークフローの全イベントを時系列順で返す。"""
        return list(self._streams[workflow_id])
 
    def get_events_since(
        self, workflow_id: str, after_index: int
    ) -> list[WorkflowEvent]:
        """指定インデックス以降のイベントのみ返す（スナップショット最適化用）。"""
        return list(self._streams[workflow_id][after_index:])

from abc import ABC, abstractmethod
 
 
# ──────────────────────────────────────────────
# コマンド：ワークフロー定義が発行する「意図」
# ──────────────────────────────────────────────
 
@dataclass(frozen=True)
class ScheduleTask:
    task_id: str
    task_name: str
    input_data: Any
 
@dataclass(frozen=True)
class StartTimer:
    timer_id: str
    delay_seconds: float
 
@dataclass(frozen=True)
class WaitForEvent:
    event_name: str
 
@dataclass(frozen=True)
class CompleteWorkflow:
    result: Any
 
@dataclass(frozen=True)
class FailWorkflow:
    error: str
 
 
WorkflowCommand = (
    ScheduleTask | StartTimer | WaitForEvent
    | CompleteWorkflow | FailWorkflow
)
 
 
# ──────────────────────────────────────────────
# ワークフロー定義の抽象インターフェース
# ──────────────────────────────────────────────
 
class WorkflowDefinition(ABC):
    """ワークフローのビジネスロジックを表す抽象基底クラス。
 
    サブクラスは decide() メソッドを実装する。
    decide() は「イベント履歴を受け取り、次のコマンドのリストを返す」
    純粋関数でなければならない。
    """
 
    @abstractmethod
    def decide(self, events: list[WorkflowEvent]) -> list[WorkflowCommand]:
        """イベント履歴から次に実行すべきコマンドを決定する。
 
        重要: このメソッドは **純粋関数** でなければならない。
        同じイベント列を入力すれば、必ず同じコマンド列を返すこと。
        """
        ...

class OrderWorkflow(WorkflowDefinition):
    """注文処理ワークフロー。
 
    検証 → 決済 → 発送 → 完了 の順に進む。
    途中でタスクが失敗したらワークフロー全体を失敗にする。
    """
 
    def decide(self, events: list[WorkflowEvent]) -> list[WorkflowCommand]:
        # まずイベント履歴から「今どのフェーズにいるか」を計算する
        state = self._build_state(events)
 
        # フェーズに応じて次のコマンドを返す
        if state["phase"] == "started":
            return [ScheduleTask(
                task_id="validate-1",
                task_name="validate_order",
                input_data=state["order_data"],
            )]
 
        if state["phase"] == "validated":
            return [ScheduleTask(
                task_id="payment-1",
                task_name="charge_payment",
                input_data=state["order_data"],
            )]
 
        if state["phase"] == "charged":
            return [ScheduleTask(
                task_id="ship-1",
                task_name="ship_order",
                input_data=state["order_data"],
            )]
 
        if state["phase"] == "shipped":
            return [CompleteWorkflow(result={
                "order_id": state["order_id"],
                "status": "delivered",
            })]
 
        if state["phase"] == "failed":
            return [FailWorkflow(error=state["error_message"])]
 
        # タスク実行中や外部イベント待ちの場合はコマンドなし
        return []
 
    def _build_state(self, events: list[WorkflowEvent]) -> dict:
        """イベント履歴を先頭から順に処理して現在の状態を構築する。
 
        この操作はイベントソーシングにおいて rehydration（再水和）と呼ばれる。
        関数型プログラミングでは fold（畳み込み）に相当する。
        """
        phase = "initial"
        order_data: Any = None
        order_id = ""
        error_message = ""
 
        for event in events:
            if isinstance(event, WorkflowStarted):
                phase = "started"
                order_data = event.input_data
                if isinstance(event.input_data, dict):
                    order_id = event.input_data.get("order_id", "")
 
            elif isinstance(event, TaskCompleted):
                if event.task_id == "validate-1":
                    phase = "validated"
                elif event.task_id == "payment-1":
                    phase = "charged"
                elif event.task_id == "ship-1":
                    phase = "shipped"
 
            elif isinstance(event, TaskFailed):
                phase = "failed"
                error_message = event.error
 
        return {
            "phase": phase,
            "order_data": order_data,
            "order_id": order_id,
            "error_message": error_message,
        }

class WorkflowEngine:
    """ワークフローエンジン本体。
 
    責務:
    1. ワークフローの開始（start_workflow）
    2. イベント履歴からの状態復元と次コマンドの決定（process_workflow）
    3. コマンドの実行とイベントの記録（_execute_command）
    4. タスク完了/失敗/タイマー発火のコールバック
    """
 
    def __init__(
        self,
        event_store: EventStore,
        task_queue: "TaskQueue",
        timer_queue: "TimerQueue",
        registry: dict[str, WorkflowDefinition],
    ) -> None:
        self.event_store = event_store
        self._task_queue = task_queue
        self._timer_queue = timer_queue
        self._registry = registry
 
    # ─── ワークフローの開始 ───
 
    def start_workflow(
        self, workflow_id: str, workflow_type: str, input_data: Any
    ) -> None:
        """新しいワークフローを開始する。"""
        start_event = WorkflowStarted(
            workflow_id=workflow_id,
            input_data=input_data,
            timestamp=time.time(),
        )
        self.event_store.append(workflow_id, [start_event])
        self.process_workflow(workflow_id, workflow_type)
 
    # ─── コアループ ───
 
    def process_workflow(self, workflow_id: str, workflow_type: str) -> None:
        """イベント列から次のコマンドを決定し実行する。
 
        このメソッドが「決定論的リプレイ」の実体。
        同じイベント列に対して decide() は同じコマンド列を返すため、
        クラッシュ後でも正しい地点から再開できる。
        """
        definition = self._registry.get(workflow_type)
        if definition is None:
            raise ValueError(f"Unknown workflow type: {workflow_type}")
 
        events = self.event_store.get_events(workflow_id)
 
        # ① 決定論的リプレイ：同じイベント列 → 同じコマンド列
        commands = definition.decide(events)
 
        # ② 既にスケジュール済みのコマンドをスキップ（冪等性の確保）
        new_commands = self._filter_already_processed(events, commands)
 
        # ③ 新しいコマンドだけを実行
        for command in new_commands:
            self._execute_command(workflow_id, workflow_type, command)
 
    # ─── コマンドの実行 ───
 
    def _execute_command(
        self, workflow_id: str, workflow_type: str, command: WorkflowCommand
    ) -> None:
        """コマンドを実行し、対応するイベントを記録する。"""
        now = time.time()
 
        if isinstance(command, ScheduleTask):
            event = TaskScheduled(
                task_id=command.task_id,
                task_name=command.task_name,
                input_data=command.input_data,
                timestamp=now,
            )
            self.event_store.append(workflow_id, [event])
            self._task_queue.enqueue(TaskEntry(
                workflow_id=workflow_id,
                workflow_type=workflow_type,
                task_id=command.task_id,
                task_name=command.task_name,
                input_data=command.input_data,
            ))
 
        elif isinstance(command, StartTimer):
            fire_at = now + command.delay_seconds
            event = TimerScheduled(
                timer_id=command.timer_id,
                fire_at=fire_at,
                timestamp=now,
            )
            self.event_store.append(workflow_id, [event])
            self._timer_queue.schedule(TimerEntry(
                workflow_id=workflow_id,
                workflow_type=workflow_type,
                timer_id=command.timer_id,
                fire_at=fire_at,
            ))
 
        elif isinstance(command, CompleteWorkflow):
            event = WorkflowCompleted(result=command.result, timestamp=now)
            self.event_store.append(workflow_id, [event])
 
        elif isinstance(command, FailWorkflow):
            event = WorkflowFailed(error=command.error, timestamp=now)
            self.event_store.append(workflow_id, [event])
 
        # WaitForEvent はエンジン側では何もしない——
        # 外部イベントの到着を待つだけ
 
    # ─── コールバック ───
 
    def on_task_completed(
        self, workflow_id: str, workflow_type: str,
        task_id: str, result: Any,
    ) -> None:
        """ワーカーからタスク完了の通知を受け取る。"""
        event = TaskCompleted(
            task_id=task_id, result=result, timestamp=time.time()
        )
        self.event_store.append(workflow_id, [event])
        # ワークフローを再駆動して次のコマンドを決定
        self.process_workflow(workflow_id, workflow_type)
 
    def on_task_failed(
        self, workflow_id: str, workflow_type: str,
        task_id: str, error: str,
    ) -> None:
        """ワーカーからタスク失敗の通知を受け取る。"""
        event = TaskFailed(
            task_id=task_id, error=error, timestamp=time.time()
        )
        self.event_store.append(workflow_id, [event])
        self.process_workflow(workflow_id, workflow_type)
 
    def on_timer_fired(
        self, workflow_id: str, workflow_type: str,
        timer_id: str, now: float,
    ) -> None:
        """タイマーキューからタイマー発火の通知を受け取る。"""
        event = TimerFired(timer_id=timer_id, timestamp=now)
        self.event_store.append(workflow_id, [event])
        self.process_workflow(workflow_id, workflow_type)
 
    # ─── 冪等性フィルター ───
 
    def _filter_already_processed(
        self,
        events: list[WorkflowEvent],
        commands: list[WorkflowCommand],
    ) -> list[WorkflowCommand]:
        """既にスケジュール済みのコマンドをスキップする。
 
        リプレイ時に同じコマンドを二重実行しないための冪等性ガード。
        """
        scheduled_task_ids: set[str] = set()
        scheduled_timer_ids: set[str] = set()
        is_completed = False
 
        for e in events:
            if isinstance(e, TaskScheduled):
                scheduled_task_ids.add(e.task_id)
            elif isinstance(e, TimerScheduled):
                scheduled_timer_ids.add(e.timer_id)
            elif isinstance(e, (WorkflowCompleted, WorkflowFailed)):
                is_completed = True
 
        if is_completed:
            return []
 
        result: list[WorkflowCommand] = []
        for cmd in commands:
            if isinstance(cmd, ScheduleTask) and cmd.task_id in scheduled_task_ids:
                continue    # 既にスケジュール済み → スキップ
            if isinstance(cmd, StartTimer) and cmd.timer_id in scheduled_timer_ids:
                continue
            result.append(cmd)
 
        return result

from collections import deque
from collections.abc import Awaitable, Callable
 
 
@dataclass
class TaskEntry:
    """タスクキューに投入されるエントリ。"""
    workflow_id: str
    workflow_type: str
    task_id: str
    task_name: str
    input_data: Any
 
 
class TaskQueue:
    """シンプルなインメモリ FIFO タスクキュー。
 
    本番では RabbitMQ、Amazon SQS、Redis Streams 等に置き換える。
    """
 
    def __init__(self) -> None:
        self._queue: deque[TaskEntry] = deque()
 
    def enqueue(self, task: TaskEntry) -> None:
        """タスクをキューの末尾に追加する。"""
        self._queue.append(task)
 
    def dequeue(self) -> TaskEntry | None:
        """キューの先頭からタスクを取得する（FIFO）。
 
        キューが空の場合は None を返す。
        """
        if self._queue:
            return self._queue.popleft()
        return None
 
    def __len__(self) -> int:
        return len(self._queue)
 
 
# アクティビティ関数の型：入力を受け取り結果を返す非同期関数
ActivityFn = Callable[[Any], Awaitable[Any]]
 
 
class Worker:
    """ワーカー：タスクキューからタスクを取得し実行する。
 
    アクティビティ（副作用のある処理）を実行し、
    完了/失敗をエンジンに通知する。
    """
 
    def __init__(
        self,
        task_queue: TaskQueue,
        engine: WorkflowEngine,
        activities: dict[str, ActivityFn],
    ) -> None:
        self._task_queue = task_queue
        self._engine = engine
        self._activities = activities
 
    async def poll(self) -> None:
        """キューからタスクを1つ取得して実行する。
 
        本番ではこれをループで回し続ける（ロングポーリング）。
        """
        task = self._task_queue.dequeue()
        if task is None:
            return
 
        activity = self._activities.get(task.task_name)
        if activity is None:
            self._engine.on_task_failed(
                task.workflow_id, task.workflow_type,
                task.task_id, f"Unknown activity: {task.task_name}",
            )
            return
 
        try:
            result = await activity(task.input_data)
            self._engine.on_task_completed(
                task.workflow_id, task.workflow_type,
                task.task_id, result,
            )
        except Exception as exc:
            self._engine.on_task_failed(
                task.workflow_id, task.workflow_type,
                task.task_id, str(exc),
            )

@dataclass
class TimerEntry:
    """タイマーキューに登録されるエントリ。"""
    workflow_id: str
    workflow_type: str
    timer_id: str
    fire_at: float   # Unix タイムスタンプ（秒）
 
 
class TimerQueue:
    """永続タイマーキュー。
 
    発火時刻順にソートされたリストを保持し、
    定期的に check_and_fire() を呼ぶことで
    期限到来のタイマーを発火する。
 
    本番では Redis の Sorted Set や
    データベースのスケジューラに置き換える。
    """
 
    def __init__(self) -> None:
        self._timers: list[TimerEntry] = []
 
    def schedule(self, timer: TimerEntry) -> None:
        """タイマーを登録し、発火時刻順にソートする。"""
        self._timers.append(timer)
        self._timers.sort(key=lambda t: t.fire_at)
 
    def check_and_fire(
        self, engine: WorkflowEngine, now: float | None = None
    ) -> None:
        """発火時刻を過ぎたタイマーをすべて発火する。"""
        if now is None:
            now = time.time()
 
        while self._timers and self._timers[0].fire_at <= now:
            timer = self._timers.pop(0)
            engine.on_timer_fired(
                timer.workflow_id, timer.workflow_type,
                timer.timer_id, now,
            )

import json
 
 
@dataclass
class OutboxEntry:
    """Outbox テーブルのエントリ。
 
    イベントストアと同じトランザクションで書き込まれ、
    Outbox リレーが非同期でタスクキュー/タイマーキューに配信する。
    """
    id: int
    workflow_id: str
    workflow_type: str
    message_type: str      # "task" または "timer"
    payload: str           # JSON 文字列
    processed: bool = False
 
 
class TransactionalEventStore:
    """Outbox パターン付きのトランザクショナルイベントストア。
 
    append_with_outbox() が「イベント追記 + Outbox メッセージ追加」を
    アトミックに行う。実際のシステムではデータベーストランザクションを使う。
    """
 
    def __init__(self) -> None:
        self._streams: dict[str, list[WorkflowEvent]] = defaultdict(list)
        self._outbox: list[OutboxEntry] = []
        self._next_outbox_id = 1
 
    def append_with_outbox(
        self,
        workflow_id: str,
        events: list[WorkflowEvent],
        messages: list[OutboxEntry],
    ) -> None:
        """アトミックにイベントを追記し Outbox メッセージを追加する。
 
        ──── ここがトランザクション境界 ────
        実際のシステムでは BEGIN / COMMIT で囲む：
          BEGIN;
            INSERT INTO event_store ...;
            INSERT INTO outbox ...;
          COMMIT;
        """
        self._streams[workflow_id].extend(events)
        for msg in messages:
            msg.id = self._next_outbox_id
            self._next_outbox_id += 1
            self._outbox.append(msg)
 
    def process_outbox(
        self,
        task_queue: TaskQueue,
        timer_queue: TimerQueue,
    ) -> None:
        """Outbox リレー：未処理メッセージをキューに配信する。"""
        for entry in self._outbox:
            if entry.processed:
                continue
 
            payload = json.loads(entry.payload)
            if entry.message_type == "task":
                task_queue.enqueue(TaskEntry(**payload))
            elif entry.message_type == "timer":
                timer_queue.schedule(TimerEntry(**payload))
 
            entry.processed = True
 
    def get_events(self, workflow_id: str) -> list[WorkflowEvent]:
        """指定ワークフローの全イベントを返す。"""
        return list(self._streams[workflow_id])

import asyncio
 
 
# ─── アクティビティ関数（副作用のある実際のビジネスロジック） ───
 
async def validate_order(input_data: Any) -> dict:
    """注文内容を検証する。"""
    order_id = input_data["order_id"]
    # 実際には在庫確認や入力バリデーションを行う
    return {"valid": True, "order_id": order_id}
 
 
async def charge_payment(input_data: Any) -> dict:
    """決済処理を行う。"""
    order_id = input_data["order_id"]
    # 実際には決済ゲートウェイ API を呼び出す
    return {"charged": True, "order_id": order_id}
 
 
async def ship_order(input_data: Any) -> dict:
    """出荷処理を行う。"""
    order_id = input_data["order_id"]
    # 実際には配送業者 API を呼び出す
    return {"tracking_id": "TRK-456", "order_id": order_id}
 
 
# ─── メイン関数 ───
 
async def main() -> None:
    # 1. インフラ層を初期化
    event_store = EventStore()
    task_queue = TaskQueue()
    timer_queue = TimerQueue()
 
    # 2. ワークフロー定義を登録
    #    キー "order" は start_workflow の workflow_type に対応
    registry: dict[str, WorkflowDefinition] = {
        "order": OrderWorkflow(),
    }
 
    # 3. エンジンを組み立て
    engine = WorkflowEngine(event_store, task_queue, timer_queue, registry)
 
    # 4. アクティビティ関数を登録してワーカーを作成
    #    キーはワークフロー定義の decide() が返す task_name に対応
    activities: dict[str, ActivityFn] = {
        "validate_order": validate_order,
        "charge_payment": charge_payment,
        "ship_order": ship_order,
    }
    worker = Worker(task_queue, engine, activities)
 
    # 5. ワークフローを開始
    engine.start_workflow(
        workflow_id="order-123",
        workflow_type="order",
        input_data={"order_id": "ORD-123", "items": ["item-A", "item-B"]},
    )
 
    # 6. ワーカーでタスクを処理（キューが空になるまで）
    while len(task_queue) > 0:
        await worker.poll()
 
    # 7. 結果を確認
    events = event_store.get_events("order-123")
    for i, event in enumerate(events):
        print(f"[{i}] {type(event).__name__}")
 
 
asyncio.run(main())

[0] WorkflowStarted      input={"order_id": "ORD-123", "items": [...]}
[1] TaskScheduled         task_id="validate-1"  task_name="validate_order"
[2] TaskCompleted         task_id="validate-1"  result={"valid": True, ...}
[3] TaskScheduled         task_id="payment-1"   task_name="charge_payment"
[4] TaskCompleted         task_id="payment-1"   result={"charged": True, ...}
[5] TaskScheduled         task_id="ship-1"      task_name="ship_order"
[6] TaskCompleted         task_id="ship-1"      result={"tracking_id": "TRK-456", ...}
[7] WorkflowCompleted     result={"order_id": "ORD-123", "status": "delivered"}