Separate commands and queries: Use CQRS to decouple write operations (such as CreateUser) from read operations (such as GetUserById), the command side processes business logic and saves events, and the query side quickly obtains data from optimized read tables (such as users_view); 2. Implement event traceability: Record and playback events (such as UserCreated) through the aggregate root (Aggregate Root), state changes are persisted in the form of events, supporting audit, debugging and state reconstruction; 3. Build event storage: Use MySQL and other databases to store event streams, load event sequences according to the aggregate ID to reconstruct aggregate instances; 4. Update the read model: apply events synchronously or asynchronously to the read model table through the projector to keep the query view updated; 5. Gradually iterated: no framework is required, start with a simple implementation, focus on event serialization, version control and aggregation testing, and ensure that the logic of the core field is reliable. After correct implementation, the system will have high scalability, traceability and flexible evolution capabilities.

Building a CQRS and Event Sourcing system in PHP might sound intimidating at first, but once you break it down, it becomes a powerful way to design scalable and maintainable applications—especially when dealing with complex business logic. Here's a practical guide to help you get started, without drowning in theory.

What You're Actually Building

CQRS (Command Query Responsibility Segregation) means separating write operations (commands) from read operations (queries).
Event Sourcing means storing state changes as a sequence of events, rather than just the current state.

Together, they allow you to:

• Track why and how data changed (audit trail by design)
• Rebuild state from events (great for debugging or analytics)
• Scale reads and writes independently

Let's walk through a simplified but real-world-like implementation in PHP.

1. Structure Your Application with CQRS Boundaries

Start by splitting your app into two sides:

• Command Side : Handles actions like "CreateUser", "UpdateEmail"
• Query Side : Handles data fetching like "GetUserById", "ListUsers"

 // Example: Command
class CreateUserCommand
{
    public function __construct(
        public readonly string $userId,
        public readonly string $name,
        public readonly string $email
    ) {}
}

// Example: Command Handler
class CreateUserHandler
{
    private EventStore $eventStore;

    public function __construct(EventStore $eventStore)
    {
        $this->eventStore = $eventStore;
    }

    public function __invoke(CreateUserCommand $command): void
    {
        $user = User::fromCommand($command);
        $this->eventStore->save($user->getUncommittedEvents());
        $user->clearEvents(); // Important: reset after saving
    }
}

On the query side, keep it simple and fast:

 class UserQueryHandler
{
    private PDO $pdo;

    public function getUserById(string $id): ?array
    {
        $stmt = $this->pdo->prepare("SELECT * FROM users_view WHERE id = ?");
        $stmt->execute([$id]);
        return $stmt->fetch(PDO::FETCH_ASSOC) ?: null;
    }
}

? Key idea: The query side reads from a denormalized table (like users_view ) optimized for reads. The command side updates the event store.

2. Implement Event Sourcing with an Aggregate Root

An aggregate is a cluster of domain objects treated as a single unit. For example, User is an aggregate root.

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

    protected function recordEvent(DomainEvent $event): void
    {
        $this->events[] = $event;
        $this->apply($event);
    }

    abstract protected function apply(DomainEvent $event): void;

    public function getUncommittedEvents(): array
    {
        return $this->events;
    }

    public function clearEvents(): void
    {
        $this->events = [];
    }

    public function getVersion(): int
    {
        return $this->version;
    }
}

Now, build a User aggregate:

 class User extends AggregateRoot
{
    private string $id;
    private string $name;
    private string $email;
    private bool $active = true;

    public static function create(CreateUserCommand $command): self
    {
        $user = new self();
        $user->recordEvent(new UserCreated(
            $command->userId,
            $command->name,
            $command->email
        ));
        return $user;
    }

    protected function apply(DomainEvent $event): void
    {
        switch ($event) {
            case $event instance of UserCreated:
                $this->id = $event->userId;
                $this->name = $event->name;
                $this->email = $event->email;
                $this->version ;
                break;
        }
    }
}

And the event:

 class UserCreated implements DomainEvent
{
    public function __construct(
        public readonly string $userId,
        public readonly string $name,
        public readonly string $email,
        public readonly DateTimeImmutable $occurredAt = new DateTimeImmutable()
    ) {}
}

3. Build a Simple Event Store

The event store saves events and loads them to rebuild aggregates.

 interface EventStore
{
    public function save(string $aggregateType, string $aggregateId, array $events, int $expectedVersion): void;
    public function load(string $aggregateId): array;
}

class MySqlEventStore implements EventStore
{
    private PDO $pdo;

    public function __construct(PDO $pdo)
    {
        $this->pdo = $pdo;
    }

    public function save(string $aggregateType, string $aggregateId, array $events, int $expectedVersion): void
    {
        $this->pdo->beginTransaction();

        foreach ($events as $event) {
            $stmt = $this->pdo->prepare(
                "INSERT INTO events (aggregate_type, aggregate_id, type, payload, version, occurred_at)
                 VALUES (?, ?, ?, ?, ?, ?, ?)"
            );
            $stmt->execute([
                $aggregateType,
                $aggregateId,
                get_class($event),
                json_encode($event),
                $this->version , // Increment per event
                $event->occurredAt->format('Ymd H:i:s.u')
            ]);
        }

        $this->pdo->commit();
    }

    public function load(string $aggregateId): array
    {
        $stmt = $this->pdo->prepare("SELECT * FROM events WHERE aggregate_id = ? ORDER BY version");
        $stmt->execute([$aggregateId]);

        $events = [];
        foreach ($stmt as $row) {
            $payload = json_decode($row['payload'], true);
            // In real apps, use a mapper or serializer
            $events[] = match ($row['type']) {
                UserCreated::class => new UserCreated(
                    $payload['userId'],
                    $payload['name'],
                    $payload['email'],
                    new DateTimeImmutable($payload['occurredAt'])
                ),
                default => throw new \Exception("Unknown event: " . $row['type'])
            };
        }

        return $events;
    }
}

? Tip: You'll want to version your events and handle schema changes later (eg, using upcasters).

4. Rebuild State from Events

When loading a User , replay all events:

 class User extends AggregateRoot
{
    // ... previous code

    public static function fromHistory(array $events): self
    {
        $user = new self();
        foreach ($events as $event) {
            $user->apply($event);
            $user->version ;
        }
        return $user;
    }
}

Usage in handler:

 public function __invoke(CreateUserCommand $command): void
{
    $events = $this->eventStore->load($command->userId);
    if (!empty($events)) {
        throw new \Exception("User already exists");
    }

    $user = User::create($command);
    $this->eventStore->save('User', $command->userId, $user->getUncommittedEvents(), $user->getVersion());
    $user->clearEvents();
}

5. Update the Read Model (Projections)

After an event is saved, update your read-optimized tables.

 interface Projector
{
    public function project(DomainEvent $event): void;
}

class UserProjector implements Projector
{
    private PDO $pdo;

    public function project(DomainEvent $event): void
    {
        if ($event instanceof UserCreated) {
            $this->pdo->prepare(
                "INSERT INTO users_view (id, name, email, created_at) VALUES (?, ?, ?, ?)"
            )->execute([
                $event->userId,
                $event->name,
                $event->email,
                $event->occurredAt->format('Ymd H:i:s')
            ]);
        }
    }
}

You can run projects:

• Synchronously (simple, consistent)
• Via a message queue (scalable, eventual consistency)

Final Notes

• Use dependency injection (eg, Symfony, Laravel, or plain container)
• Serialize events carefully—prefer JSON or MessagePack over PHP serialization
• Consider using UUIDs for aggregate IDs
• Add metadata to events (eg, user IP, trace ID)
• Test your aggregates heavily—they're the core of your domain

It's not trivial, but you don't need a framework to get started. With plain PHP, interfaces, and good separation, you can build a solid foundation. The real power shows up when your business rules grow and you need to audit, replay, or extend behavior without breaking things.

Basically: start small, get the flow right, and let the events do the talking.

The above is the detailed content of Building a CQRS and Event Sourcing System with PHP. For more information, please follow other related articles on the PHP Chinese website!