Vengala Vinay

Having 12+ Years of Experience in Software Development

Home » Beyond the Basics: Mastering Laravel’s Event Sourcing for Scalable Microservices

Beyond the Basics: Mastering Laravel’s Event Sourcing for Scalable Microservices

Event Store Implementation with PostgreSQL

For robust event sourcing in a microservices architecture, a dedicated event store is paramount. PostgreSQL, with its JSONB capabilities and transactional integrity, offers a compelling foundation. We’ll define a simple yet effective schema for storing events.

Consider a table named event_stream. Each row represents a single event. Key columns include:

  • id: A UUID for unique event identification.
  • aggregate_type: The type of the entity the event pertains to (e.g., ‘Order’, ‘User’).
  • aggregate_id: The UUID of the specific entity instance.
  • sequence_number: An integer representing the order of events for a given aggregate. This is crucial for replaying events correctly.
  • event_type: The specific type of event that occurred (e.g., ‘OrderPlaced’, ‘UserRegistered’).
  • payload: A JSONB column storing the event’s data. This allows for flexible schema evolution.
  • occurred_at: A timestamp indicating when the event happened.

Here’s the SQL DDL for creating this table:

CREATE TABLE event_stream (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id UUID NOT NULL,
    sequence_number BIGINT NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    occurred_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (aggregate_type, aggregate_id, sequence_number)
);

CREATE INDEX idx_event_stream_aggregate ON event_stream (aggregate_type, aggregate_id);
CREATE INDEX idx_event_stream_sequence ON event_stream (aggregate_type, aggregate_id, sequence_number);

Laravel Integration: Event Repository and Domain Events

In your Laravel microservice, you’ll need a repository to interact with this event store. We’ll also define domain events that can be dispatched.

First, let’s define a base `DomainEvent` class. This will serve as a contract for all events within your system.

<?php

namespace App\Domain\Events;

use Carbon\Carbon;
use Ramsey\Uuid\UuidInterface;

abstract class DomainEvent
{
    public readonly UuidInterface $eventId;
    public readonly Carbon $occurredAt;

    public function __construct(
        public readonly string $aggregateType,
        public readonly UuidInterface $aggregateId,
        public readonly int $sequenceNumber
    ) {
        $this->eventId = \Ramsey\Uuid\Uuid::uuid4();
        $this->occurredAt = Carbon::now();
    }

    abstract public function eventType(): string;
    abstract public function payload(): array;

    public function toArray(): array
    {
        return [
            'id' => $this->eventId->toString(),
            'aggregate_type' => $this->aggregateType,
            'aggregate_id' => $this->aggregateId->toString(),
            'sequence_number' => $this->sequenceNumber,
            'event_type' => $this->eventType(),
            'payload' => $this->payload(),
            'occurred_at' => $this->occurredAt->toIso8601String(),
        ];
    }
}

Now, create a concrete event, for example, `OrderPlaced`.

<?php

namespace App\Domain\Events;

use Ramsey\Uuid\UuidInterface;

class OrderPlaced extends DomainEvent
{
    public function __construct(
        UuidInterface $orderId,
        int $sequenceNumber,
        public readonly array $items,
        public readonly float $totalAmount
    ) {
        parent::__construct('Order', $orderId, $sequenceNumber);
    }

    public function eventType(): string
    {
        return 'OrderPlaced';
    }

    public function payload(): array
    {
        return [
            'items' => $this->items,
            'total_amount' => $this->totalAmount,
        ];
    }
}

Next, the `EventRepository` interface and its PostgreSQL implementation.

<?php

namespace App\Domain\Repositories;

use App\Domain\Events\DomainEvent;
use Illuminate\Support\Collection;
use Ramsey\Uuid\UuidInterface;

interface EventRepository
{
    public function append(DomainEvent $event): void;
    public function appendMany(array $events): void;
    public function getStream(string $aggregateType, UuidInterface $aggregateId): Collection;
    public function getNextSequenceNumber(string $aggregateType, UuidInterface $aggregateId): int;
}

<?php

namespace App\Infrastructure\Persistence\EventStore;

use App\Domain\Events\DomainEvent;
use App\Domain\Repositories\EventRepository;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\DB;
use Ramsey\Uuid\UuidInterface;

class PostgresEventRepository implements EventRepository
{
    public function append(DomainEvent $event): void
    {
        $this->appendMany([$event]);
    }

    public function appendMany(array $events): void
    {
        if (empty($events)) {
            return;
        }

        $data = array_map(fn(DomainEvent $event) => $event->toArray(), $events);

        // Ensure all events for the same aggregate have sequential sequence numbers
        $groupedEvents = collect($data)->groupBy(['aggregate_type', 'aggregate_id']);
        foreach ($groupedEvents as $aggregateType => $ids) {
            foreach ($ids as $aggregateId => $eventList) {
                $sortedEvents = $eventList->sortBy('sequence_number');
                $expectedSequence = $this->getNextSequenceNumber($aggregateType, \Ramsey\Uuid\Uuid::fromString($aggregateId));
                foreach ($sortedEvents as $index => $eventData) {
                    if ($eventData['sequence_number'] !== $expectedSequence) {
                        throw new \DomainException("Sequence number mismatch for aggregate {$aggregateType}:{$aggregateId}. Expected {$expectedSequence}, got {$eventData['sequence_number']}.");
                    }
                    $data[$index]['sequence_number'] = $expectedSequence; // Correct sequence number
                    $expectedSequence++;
                }
            }
        }

        DB::beginTransaction();
        try {
            foreach ($data as $eventData) {
                DB::table('event_stream')->insert($eventData);
            }
            DB::commit();
        } catch (\Exception $e) {
            DB::rollBack();
            // Log the error and potentially re-throw a more specific exception
            throw $e;
        }
    }

    public function getStream(string $aggregateType, UuidInterface $aggregateId): Collection
    {
        $events = DB::table('event_stream')
            ->where('aggregate_type', $aggregateType)
            ->where('aggregate_id', $aggregateId->toString())
            ->orderBy('sequence_number')
            ->get();

        return $events->map(function ($event) {
            // Hydrate DomainEvent objects here. This is a simplified example.
            // In a real application, you'd use an event factory or similar.
            return [
                'event_type' => $event->event_type,
                'payload' => (array) $event->payload,
                'sequence_number' => $event->sequence_number,
                'aggregate_id' => $event->aggregate_id,
                'aggregate_type' => $event->aggregate_type,
            ];
        });
    }

    public function getNextSequenceNumber(string $aggregateType, UuidInterface $aggregateId): int
    {
        $latestEvent = DB::table('event_stream')
            ->where('aggregate_type', $aggregateType)
            ->where('aggregate_id', $aggregateId->toString())
            ->orderByDesc('sequence_number')
            ->first();

        return $latestEvent ? (int) $latestEvent->sequence_number + 1 : 1;
    }
}

Register the repository in your service provider:

<?php

namespace App\Providers;

use App\Domain\Repositories\EventRepository;
use App\Infrastructure\Persistence\EventStore\PostgresEventRepository;
use Illuminate\Support\ServiceProvider;

class AppServiceProvider extends ServiceProvider
{
    public function register(): void
    {
        $this->app->bind(EventRepository::class, PostgresEventRepository::class);
    }

    // ...
}

Command Bus and Event Dispatching

A command bus pattern is essential for decoupling command execution from the application’s core logic. When a command is executed and results in domain events, these events should be dispatched to an event bus for asynchronous processing by other microservices or internal listeners.

Let’s assume you have a simple `CommandBus` interface and implementation.

<?php

namespace App\Domain\Bus;

interface CommandBus
{
    public function dispatch(object $command): mixed;
}

<?php

namespace App\Infrastructure\Bus;

use App\Domain\Bus\CommandBus;
use Illuminate\Contracts\Bus\Dispatcher;
use Illuminate\Support\Facades\App;

class LaravelCommandBus implements CommandBus
{
    public function dispatch(object $command): mixed
    {
        // Laravel's Bus facade can be used directly, but this abstraction
        // allows for more control, like custom middleware or logging.
        return App::make(Dispatcher::class)->dispatchNow($command); // Use dispatchNow for synchronous execution within the current service
    }
}

And a simple `EventBus` interface.

<?php

namespace App\Domain\Bus;

use App\Domain\Events\DomainEvent;

interface EventBus
{
    public function publish(DomainEvent $event): void;
    public function publishMany(array $events): void;
}

<?php

namespace App\Infrastructure\Bus;

use App\Domain\Bus\EventBus;
use App\Domain\Events\DomainEvent;
use Illuminate\Contracts\Events\Dispatcher;
use Illuminate\Support\Facades\App;

class LaravelEventBus implements EventBus
{
    public function publish(DomainEvent $event): void
    {
        $this->publishMany([$event]);
    }

    public function publishMany(array $events): void
    {
        $dispatcher = App::make(Dispatcher::class);
        foreach ($events as $event) {
            // Laravel's built-in event dispatcher can be used for intra-service communication.
            // For inter-service communication, this would publish to a message queue.
            $dispatcher->dispatch($event);
        }
    }
}

Register these in your service provider:

<?php

namespace App\Providers;

use App\Domain\Bus\CommandBus;
use App\Domain\Bus\EventBus;
use App\Infrastructure\Bus\LaravelCommandBus;
use App\Infrastructure\Bus\LaravelEventBus;
use Illuminate\Support\ServiceProvider;

class AppServiceProvider extends ServiceProvider
{
    public function register(): void
    {
        // ... other bindings
        $this->app->singleton(CommandBus::class, LaravelCommandBus::class);
        $this->app->singleton(EventBus::class, LaravelEventBus::class);
    }

    // ...
}

Aggregate Root and Event Replay

Aggregate roots are the core of event sourcing. They encapsulate business logic and are responsible for producing domain events. They must be able to replay their history to reconstruct their current state.

Here’s an example of an `Order` aggregate root.

<?php

namespace App\Domain\Aggregates;

use App\Domain\Events\DomainEvent;
use App\Domain\Events\OrderPlaced;
use App\Domain\Events\OrderItemAdded;
use App\Domain\Events\OrderShipped;
use App\Domain\Repositories\EventRepository;
use Illuminate\Support\Collection;
use Ramsey\Uuid\UuidInterface;

class Order
{
    private UuidInterface $id;
    private int $sequenceNumber = 0;
    private array $items = [];
    private float $totalAmount = 0.0;
    private bool $shipped = false;

    private array $recordedEvents = [];

    public function __construct(
        private readonly EventRepository $eventRepository
    ) {}

    public static function create(UuidInterface $orderId, array $items, float $totalAmount, EventRepository $eventRepository): self
    {
        $order = new self($eventRepository);
        $order->id = $orderId;
        $order->apply(new OrderPlaced($orderId, 1, $items, $totalAmount));
        return $order;
    }

    public static function replay(UuidInterface $orderId, EventRepository $eventRepository): self
    {
        $order = new self($eventRepository);
        $order->id = $orderId;
        $stream = $eventRepository->getStream('Order', $orderId);

        if ($stream->isEmpty()) {
            throw new \RuntimeException("Order with ID {$orderId->toString()} not found.");
        }

        $stream->each(fn($eventData) => $order->applyEventData($eventData));
        return $order;
    }

    private function applyEventData(array $eventData): void
    {
        $event = $this->hydrateEvent($eventData);
        $this->apply($event);
    }

    private function hydrateEvent(array $eventData): DomainEvent
    {
        $className = '\\App\\Domain\\Events\\' . $eventData['event_type'];
        if (!class_exists($className)) {
            throw new \RuntimeException("Event class {$className} not found.");
        }

        // This is a simplified hydration. In a real app, you'd map payload keys to constructor arguments.
        // For example, using reflection or a dedicated event hydration service.
        $event = new $className(
            \Ramsey\Uuid\Uuid::fromString($eventData['aggregate_id']),
            $eventData['sequence_number'],
            ...array_values($eventData['payload']) // Assuming payload keys match constructor order
        );

        // Manually set event ID and occurred at if needed for specific logic,
        // though usually not required for state reconstruction.
        // $event->eventId = Uuid::fromString($eventData['id']);
        // $event->occurredAt = Carbon::parse($eventData['occurred_at']);

        return $event;
    }

    public function addOrderItem(string $productId, int $quantity, float $price): void
    {
        if ($this->shipped) {
            throw new \DomainException("Cannot add items to a shipped order.");
        }
        $newSequenceNumber = $this->getNextSequenceNumber();
        $this->apply(new OrderItemAdded($this->id, $newSequenceNumber, $productId, $quantity, $price));
    }

    public function ship(): void
    {
        if ($this->shipped) {
            throw new \DomainException("Order is already shipped.");
        }
        $newSequenceNumber = $this->getNextSequenceNumber();
        $this->apply(new OrderShipped($this->id, $newSequenceNumber));
    }

    private function apply(DomainEvent $event): void
    {
        $method = 'when' . $event->eventType();
        if (method_exists($this, $method)) {
            $this->$method($event);
        }
        $this->recordEvent($event);
    }

    private function recordEvent(DomainEvent $event): void
    {
        $this->recordedEvents[] = $event;
        $this->sequenceNumber = $event->sequenceNumber; // Keep track of the latest sequence number
    }

    // Event handlers
    protected function whenOrderPlaced(OrderPlaced $event): void
    {
        $this->id = $event->aggregateId;
        $this->items = $event->payload()['items']; // Simplified, should be structured
        $this->totalAmount = $event->payload()['total_amount'];
        $this->sequenceNumber = $event->sequenceNumber;
    }

    protected function whenOrderItemAdded(OrderItemAdded $event): void
    {
        $this->items[] = $event->payload(); // Simplified
        $this->totalAmount += $event->payload()['price'] * $event->payload()['quantity'];
        $this->sequenceNumber = $event->sequenceNumber;
    }

    protected function whenOrderShipped(OrderShipped $event): void
    {
        $this->shipped = true;
        $this->sequenceNumber = $event->sequenceNumber;
    }

    public function getRecordedEvents(): array
    {
        return $this->recordedEvents;
    }

    public function getId(): UuidInterface
    {
        return $this->id;
    }

    public function getSequenceNumber(): int
    {
        return $this->sequenceNumber;
    }

    private function getNextSequenceNumber(): int
    {
        // This logic needs to be robust. It should query the event store
        // to get the *current* highest sequence number for this aggregate
        // and return that + 1.
        // For simplicity here, we're assuming it's managed by the repository.
        // A better approach is to fetch the current sequence number *before*
        // creating a new event.
        return $this->sequenceNumber + 1;
    }

    // Method to save events to the store
    public function save(): void
    {
        if (!empty($this->recordedEvents)) {
            $this->eventRepository->appendMany($this->recordedEvents);
            $this->recordedEvents = []; // Clear after saving
        }
    }
}

Service Layer and Event Publishing

The service layer orchestrates the process: it receives commands, uses aggregates to perform actions, records events, and then publishes them.

<?php

namespace App\Services;

use App\Domain\Aggregates\Order;
use App\Domain\Bus\CommandBus;
use App\Domain\Bus\EventBus;
use App\Domain\Repositories\EventRepository;
use Ramsey\Uuid\UuidInterface;

class OrderService
{
    public function __construct(
        private readonly EventRepository $eventRepository,
        private readonly EventBus $eventBus,
        private readonly CommandBus $commandBus // Potentially for internal command dispatching
    ) {}

    public function placeOrder(UuidInterface $orderId, array $items, float $totalAmount): Order
    {
        $order = Order::create($orderId, $items, $totalAmount, $this->eventRepository);
        $events = $order->getRecordedEvents();
        $order->save(); // Persist events to the event store

        $this->eventBus->publishMany($events); // Publish events for other services/listeners

        return $order;
    }

    public function addOrderItemToOrder(UuidInterface $orderId, string $productId, int $quantity, float $price): Order
    {
        $order = Order::replay($orderId, $this->eventRepository);
        $order->addOrderItem($productId, $quantity, $price);
        $events = $order->getRecordedEvents();
        $order->save();

        $this->eventBus->publishMany($events);

        return $order;
    }

    public function shipOrder(UuidInterface $orderId): Order
    {
        $order = Order::replay($orderId, $this->eventRepository);
        $order->ship();
        $events = $order->getRecordedEvents();
        $order->save();

        $this->eventBus->publishMany($events);

        return $order;
    }
}

Handling Asynchronous Event Consumption

When events are published to the `LaravelEventBus`, they are dispatched using Laravel’s built-in `Dispatcher`. For inter-service communication, this dispatching mechanism needs to be replaced with a robust message queue system like RabbitMQ or Kafka. Laravel’s Queue system can be configured to use these backends.

Here’s how you might configure Laravel’s queue for RabbitMQ:

[queue]
; Default queue connection that will be used if none is specified when the
; command is sent to the queue.
default = rabbitmq

; The connections that are available to your application.
connections.rabbitmq.driver = rabbitmq
connections.rabbitmq.host = 127.0.0.1
connections.rabbitmq.port = 5672
connections.rabbitmq.username = guest
connections.rabbitmq.password = guest
connections.rabbitmq.queue = default
connections.rabbitmq.options.ssl_enabled = false
connections.rabbitmq.options.ssl_verify_peer = false
connections.rabbitmq.options.vhost = /
connections.rabbitmq.options.heartbeat = 60

With this configuration, the `LaravelEventBus` would publish messages to RabbitMQ. Other microservices would have listeners (Laravel Jobs) that consume these messages.

<?php

namespace App\Listeners;

use App\Domain\Events\OrderPlaced;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;

class HandleOrderPlacedForInventoryService implements ShouldQueue
{
    use InteractsWithQueue;

    public function handle(OrderPlaced $event): void
    {
        // This listener runs in a separate microservice or as a background job.
        // It consumes the OrderPlaced event and updates inventory.
        // Example:
        // $inventoryService = app(InventoryService::class);
        // $inventoryService->decreaseStock($event->payload()['items']);

        // Log or perform other actions
        \Log::info("Inventory service received OrderPlaced event for order: {$event->aggregateId->toString()}");
    }
}

To make this work, you’d register the listener in App\Providers\EventServiceProvider:

<?php

namespace App\Providers;

use App\Domain\Events\OrderPlaced;
use App\Listeners\HandleOrderPlacedForInventoryService;
use Illuminate\Auth\Events\Registered;
use Illuminate\Auth\Listeners\SendEmailVerificationNotification;
use Illuminate\Foundation\Support\Providers\EventServiceProvider as ServiceProvider;
use Illuminate\Support\Facades\Event;

class EventServiceProvider extends ServiceProvider
{
    protected $listen = [
        OrderPlaced::class => [
            HandleOrderPlacedForInventoryService::class,
        ],
        // ... other events
    ];

    public function boot(): void
    {
        parent::boot();
    }
}

And ensure the queue worker is running:

php artisan queue:work rabbitmq

CQRS Considerations

Event sourcing naturally lends itself to Command Query Responsibility Segregation (CQRS). The event store serves as the write model’s source of truth. For read models (queries), you can build projections by consuming events from the event bus and updating denormalized views in separate databases (e.g., Elasticsearch, or another relational database optimized for reads).

For instance, a `OrderReadModel` could be updated by a listener that consumes `OrderPlaced`, `OrderItemAdded`, and `OrderShipped` events. This read model would be optimized for querying order status and details, without needing to replay events every time.

The key is that the event store remains the single source of truth. Read models are derived from it and can be rebuilt if necessary.

Recent Posts

Top Categories

Our Products

Our Services