イベント駆動ワークフローの構築

イベント駆動ワークフロー

コレオグラフィーとオーケストレーション

イベントとコマンド

<?php
final class OrderPlaced {
    public function __construct(
        public readonly string $orderId,
        public readonly string $customerId,
        public readonly int $amountCents,
        public readonly string $occurredAt,
    ) {}
}

$e = new OrderPlaced('o-42', 'c-7', 1990, gmdate('c'));
echo json_encode($e), "\n";

デュアルライト問題

<?php
// BROKEN: not atomic. A crash between the two lines corrupts state.
function placeOrder(PDO $db, $broker, array $o): void {
    $db->prepare('INSERT INTO orders ...')->execute($o);
    // <-- crash here = row exists but no event ever published
    $broker->publish('OrderPlaced', json_encode($o));
}

トランザクショナルOutbox

<?php
function placeOrder(PDO $db, array $o): void {
    $db->beginTransaction();
    $db->prepare('INSERT INTO orders (id, total) VALUES (?, ?)')
       ->execute([$o['id'], $o['total']]);
    // Same transaction -> atomic with the business write
    $db->prepare('INSERT INTO outbox (id, type, payload) VALUES (?, ?, ?)')
       ->execute([bin2hex(random_bytes(8)), 'OrderPlaced', json_encode($o)]);
    $db->commit();
}

リレー（パブリッシャー）

<?php
function relayOutbox(PDO $db, $broker): void {
    $rows = $db->query(
        'SELECT id, type, payload FROM outbox
         WHERE published_at IS NULL ORDER BY created_at LIMIT 100'
    )->fetchAll(PDO::FETCH_ASSOC);

    foreach ($rows as $r) {
        $broker->publish($r['type'], $r['payload'], messageId: $r['id']);
        $db->prepare('UPDATE outbox SET published_at = now() WHERE id = ?')
           ->execute([$r['id']]);
    }
}

冪等なコンシューマー（再び）

<?php
function onOrderPlaced(PDO $db, string $messageId, array $data): void {
    $db->beginTransaction();
    try {
        $db->prepare('INSERT INTO inbox (message_id) VALUES (?)')
           ->execute([$messageId]); // unique index = dedup
    } catch (PDOException $e) {
        $db->rollBack();
        return; // already handled this message
    }
    reserveStock($data['orderId']);
    $db->commit();
}
function reserveStock(string $id): void {}

Saga：分散ロールバック

オーケストレーションされたSaga

<?php
function handleStepResult(array $saga, string $step, bool $ok, $bus): array {
    if ($ok) {
        $next = ['reserveStock' => 'chargeCard', 'chargeCard' => 'ship'][$step] ?? null;
        if ($next) { $bus->send($next, $saga['orderId']); $saga['state'] = $next; }
        else { $saga['state'] = 'completed'; }
    } else {
        // Run compensations in reverse for whatever already succeeded
        foreach (array_reverse($saga['done']) as $s) {
            $bus->send('compensate.' . $s, $saga['orderId']);
        }
        $saga['state'] = 'compensating';
    }
    return $saga;
}

長時間フローのタイムアウト

<?php
function reapTimedOutSagas(PDO $db, $bus): void {
    $rows = $db->query(
        "SELECT order_id, state FROM sagas
         WHERE state NOT IN ('completed','compensating')
           AND deadline_at < now()"
    )->fetchAll(PDO::FETCH_ASSOC);

    foreach ($rows as $r) {
        echo "Saga {$r['order_id']} timed out at step {$r['state']}\n";
        $bus->send('saga.compensate', $r['order_id']); // trigger rollback
    }
}

バージョニングと可観測性

<?php
$envelope = [
    'type'          => 'OrderPlaced',
    'version'       => 2,
    'correlationId' => $incoming['correlationId'] ?? bin2hex(random_bytes(8)),
    'occurredAt'    => gmdate('c'),
    'data'          => ['orderId' => 'o-42'],
];
echo json_encode($envelope, JSON_PRETTY_PRINT), "\n";

クイックチェック

まとめ