PHP 的 CQRS 与事件溯源：命令总线、事件存储与投影

final readonly class PlaceOrder
{
    public function __construct(
        public string $orderId,
        public string $customerId,
        public array $items,        // [['sku' => 'A', 'qty' => 2, 'price' => 1000]]
    ) {}
}

final class SimpleCommandBus
{
    private array $handlers = [];            // 命令类 => 处理器
    public function register(string $c, callable $h): void { $this->handlers[$c] = $h; }

    public function dispatch(object $command): mixed
    {
        $handler = $this->handlers[$command::class]
            ?? throw new RuntimeException('未注册处理器: ' . $command::class);
        return $handler($command);
    }
}

final class PlaceOrderHandler
{
    public function __construct(
        private OrderRepository $orders,
        private EventBus $events,
    ) {}

    public function __invoke(PlaceOrder $command): void
    {
        $order = Order::place($command->orderId, $command->customerId, $command->items);
        $this->orders->save($order);            // 追加事件到事件存储
        $this->events->publish(...$order->releaseEvents());
    }
}

final readonly class OrderPlaced
{
    public function __construct(
        public string $orderId,
        public string $customerId,
        public array $items,
        public int $total,
        public DateTimeImmutable $occurredAt,
    ) {}
}

abstract class AggregateRoot
{
    protected int $version = 0;
    private array $events = [];

    protected function apply(object $event): void
    {
        $this->applyEvent($event);      // 纯状态变换
        $this->events[] = $event;       // 记录待发布
        $this->version++;
    }

    abstract protected function applyEvent(object $event): void;

    public function releaseEvents(): array
    {
        return tap($this->events, fn () => $this->events = []);
    }
}

final class Order extends AggregateRoot
{
    private string $status = 'new';

    public static function place(string $id, string $customerId, array $items): self
    {
        if ($items === []) { throw new DomainException('订单不能为空'); }   // 不变量校验
        $order = new self();
        $order->apply(new OrderPlaced($id, $customerId, $items, 0, new DateTimeImmutable()));
        return $order;
    }

    public function confirm(): void
    {
        if ($this->status !== 'new') {
            throw new DomainException('只有新订单可以确认');   // 状态不变量
        }
        $this->apply(new OrderConfirmed($this->id, new DateTimeImmutable()));
    }

    protected function applyEvent(object $event): void
    {
        if ($event instanceof OrderConfirmed) { $this->status = 'confirmed'; }
    }
}

CREATE TABLE event_store (
    aggregate_id CHAR(36)     NOT NULL,
    version      INT          NOT NULL,
    event_type   VARCHAR(100) NOT NULL,
    payload      JSON         NOT NULL,
    occurred_at  DATETIME(3)  NOT NULL,
    PRIMARY KEY (aggregate_id, version)     -- 复合主键 = 乐观并发控制
);

public function append(string $aggregateId, int $expectedVersion, array $events): void
{
    $this->pdo->beginTransaction();
    try {
        $stmt = $this->pdo->prepare(
            'INSERT INTO event_store (aggregate_id, version, event_type, payload, occurred_at)
             VALUES (?, ?, ?, ?, ?)'
        );
        foreach ($events as $i => $event) {
            $stmt->execute([
                $aggregateId, $expectedVersion + $i + 1, $event::class,
                json_encode($event, JSON_THROW_ON_ERROR), date('Y-m-d H:i:s.v'),
            ]);
        }
        $this->pdo->commit();
    } catch (PDOException $e) {
        $this->pdo->rollBack();
        if ($e->getCode() === '23000') {         // 唯一键冲突 = 并发修改
            throw new ConcurrencyException('聚合已被并发修改，请重试');
        }
        throw $e;
    }
}

final class OrderSummaryProjector
{
    public function __construct(private PDO $pdo) {}

    public function handle(object $event): void
    {
        match (true) {
            $event instanceof OrderPlaced => $this->pdo->prepare(
                'INSERT INTO order_summary (id, customer_id, status, total) VALUES (?, ?, "new", ?)'
            )->execute([$event->orderId, $event->customerId, $event->total]),

            $event instanceof OrderConfirmed => $this->pdo->prepare(
                'UPDATE order_summary SET status = "confirmed" WHERE id = ?'
            )->execute([$event->orderId]),

            default => null,
        };
    }
}

$snap  = $snapshots->latestFor($aggregateId);         // {version: 500, state: {...}}
$order = Order::fromSnapshot($snap->state);
foreach ($events->since($aggregateId, $snap->version) as $e) { $order->apply($e); }