Message Repository For Illuminate Laravel Package

Getting Started

This is a storage adapter, not a framework. It implements EventSauce's MessageRepository interface on top of Laravel's Illuminate\Database\ConnectionInterface, which means your job is: install it, create a table, wire a serializer, and hand it to EventSauce's aggregate repository. There is no service provider, no config file, and no published migration — that's all on you.

1. Install the real dependency chain. The adapter alone does nothing:

composer require eventsauce/eventsauce eventsauce/message-repository-for-illuminate

eventsauce/eventsauce brings in AggregateRoot, ConstructingAggregateRootRepository, ConstructingMessageSerializer, and the MessageRepository interface. The Illuminate package supplies only the read/write adapter.

2. Read three things before writing code. This repo is a [READ ONLY] sub-split of the EventSauce monorepo, so Composer is your source of truth:

• vendor/eventsauce/message-repository/src/MessageRepository.php — the three-method contract (persist, retrieveAll, retrieveAllAfterVersion) you are programming against.
• vendor/eventsauce/message-repository-for-illuminate/src/ — the concrete repository class. Read the constructor signature and, critically, the exact column names it selects and inserts. Do not guess these.
• The parent monorepo's tests for this adapter — that's where behavior is actually specified, not here.

3. Write the migration yourself. A canonical shape looks like:

Schema::create('event_messages', function (Blueprint $table) {
    $table->uuid('aggregate_root_id');
    $table->unsignedInteger('aggregate_root_version');
    $table->json('payload');
    $table->json('headers')->nullable();
    $table->timestamp('created_at')->nullable();

    // NOT optional. This is your optimistic-concurrency guard.
    $table->unique(['aggregate_root_id', 'aggregate_root_version'], 'uq_stream_version');
    $table->index('aggregate_root_id');
});

4. Wire it up. In a service provider, bind the interface and construct the concrete repository with your connection, serializer, and table name (verify the exact argument order from the class):

$this->app->singleton(MessageRepository::class, function ($app) {
    return new IlluminateMessageRepository(
        $app['db']->connection(),
        new ConstructingMessageSerializer(),
        'event_messages',
    );
});

Then build the aggregate repository on top of it:

$repository = new ConstructingAggregateRepository(
    Order::class,
    $messageRepository,
    $messageDecorator, // optional
);

First use case to prove it: create one aggregate, record one event in a transaction, reload it by id, and confirm you get the same event back with the right version. If that round-trips on SQLite in a test, the wiring is correct.

Implementation Patterns

Persist inside DB::transaction() — always. The adapter does not manage transactions for you. If you save an aggregate and the write fails halfway, you have a corrupt stream. The standard pattern:

DB::transaction(function () use ($repository, $aggregate) {
    $repository->persist($aggregate);
});

Use the same connection for the message table and any read-model writes you want to be atomic with it.

Treat the repository as the aggregate repository's engine, not the thing you call directly. In practice you rarely call persist/retrieveAll yourself — you use AggregateRootRepository:

$order = $orderRepository->retrieve($id);
$order->ship($address);
$orderRepository->persist($order);

This is important because AggregateRootRepository handles the reconciliation of expected versions and raises EventSauce's own concurrency exception. Calling the message repository directly bypasses that.

Guard against concurrent writes with a retry loop. Two processes loading version 5 and both persisting version 6 will produce a duplicate-key violation on the unique index. Catch it and retry the whole load→mutate→persist cycle:

use EventSauce\EventSauce\ExactQuerySpecification\...; // whichever your version exposes

$attempts = 0;
retry:
try {
    DB::transaction(fn () => $repo->persist($aggregate));
} catch (QueryException $e) {
    if (++$attempts > 3) throw $e;
    // reload the aggregate fresh, re-apply the intent, then goto retry
}

The retry must reload the aggregate — you cannot resume from the stale in-memory instance. eventsauce/backoff pairs well here if you add jitter.

Bind against the interface, not the concrete class. Your domain service should type-hint MessageRepository, never IlluminateMessageRepository. This makes the eventual swap to the Doctrine adapter or a bespoke store a one-line binding change.

Multi-tenant isolation: one connection or one database per tenant. Bind a per-tenant repository via a scoped container binding that resolves the tenant's connection name and passes it to the adapter. Don't rely on a global scope — the adapter uses the raw query builder and Eloquent's global scopes won't apply.

Test with SQLite for speed, but run at least one CI job against Postgres/MySQL. The query builder emits portable SQL, but version-conflict behavior and lock semantics differ by engine. The concurrency guard is exactly the thing you want tested against your production engine.

Read-model projections: subscribe to the same stream via retrieveAll (streaming generator) and write projections inside their own transaction. Do not attempt to write the projection in the same transaction as the aggregate persist unless the projection is trivially idempotent — a failed projection should not roll back a committed event.

Gotchas and Tips

The unique index is not optional. Without unique(aggregate_root_id, aggregate_root_version), optimistic concurrency is advisory only and silent stream corruption is possible. This is the single most common mistake with hand-rolled migrations. Build it in the very first migration and never remove it.

Column names are load-bearing. The adapter selects and inserts specific column names. If your migration guesses event_type instead of headers, or version instead of aggregate_root_version, you'll get cryptic SQL errors, not friendly ones. Read the class source before writing the migration.

retrieveAll returns a Generator. Do not do iterator_to_array() on it unless you want to load the whole stream into memory. On a long-lived aggregate this is a memory exhaustion waiting to happen. Stream, filter, and aggregate as you go.

This repo is a [READ ONLY] sub-split. Zero stars and zero dependents are packaging artifacts — the real activity is at EventSaucePHP/EventSauce. Consequences: you cannot file issues or PRs here; patch bugs via cweagans/composer-patches or fork the parent; pin the version and commit the lockfile; watch the parent monorepo's releases, not this page.

Duplicate-key errors are domain conflicts, not infrastructure failures. Map them to a retry in your command handler or surface them to the user as "this record changed while you were editing." Do not let them bubble as a 500.

Serializer changes are data migrations. If you change how payloads are serialized (e.g., adding an event-type discriminator, changing JSON shape), every historical row is affected. Version your serializer, migrate the payload shape forward in a dedicated migration, and never assume the old bytes still deserialize.

The adapter does not manage transactions, does not snapshot, and does not do retention. Every load replays the full stream from version 0. Once streams exceed a few hundred events, you need a snapshot repository (layered on top of the same adapter) and an archival or partitioning strategy keyed by aggregate_root_id.

No Eloquent, no model events, no observers, no HasUuids, no soft deletes. If your codebase leans on those, implement the equivalent at the repository boundary. Do not try to make the event table an Eloquent model — you'll fight the adapter and lose.

persist on a fresh aggregate starts at version 0 and increments per event; persist on a loaded aggregate continues from the loaded version. If you see version conflicts on first-ever persist, check whether you are accidentally reusing an aggregate instance across requests (e.g., as a container singleton).

Deadlocks under MySQL write concurrency are expected. Wrap your persist in a retry with exponential backoff. Postgres handles this better; if you're MySQL and hitting it hard, consider isolating hot aggregates to a dedicated connection.

Debugging tip: when an aggregate won't reload, select * from event_messages where aggregate_root_id = ? and check aggregate_root_version is contiguous starting at 0. Gaps mean an interrupted write outside a transaction — that's the failure mode the transaction wrapping prevents.

Extension points are thin by design. If you need per-event metadata (tenant id, correlation id, causation id), put it in the message headers via an EventMessageHeaderDecorator or a MessageDecorator on the aggregate repository — not by adding columns to the table, which the adapter won't populate.