Pdo Event Sourcing Laravel Package

Getting Started

1. Install and confirm what you actually got

composer require ecotone/pdo-event-sourcing

This is a split, read-only repository — it ships as part of the ecotone/ecotone monorepo and is published here only so Composer can install the PDO event store independently. Pulling it in also gives you the core ecotone/ecotone package and ecotone/dbal (the DBAL/PDO layer the store is built on). Contributions and issues go to ecotoneframework/ecotone-dev, not this repo — do not waste time opening PRs here.

Requirements to check before writing a line of code:

• PHP 8.1+ (attributes are the whole API surface, so no annotations fallback).
• PostgreSQL or MySQL/MariaDB only. SQLite is not a supported backend for the event store — it relies on DB-specific SQL and row locking.
• A framework integration: ecotone/symfony-bundle or ecotone/laravel, or EcotoneLite for standalone scripts and tests.

2. Where to look first

Start by grepping vendor/ecotone/ecotone/src/EventSourcing/Attribute/ — that directory is the public API. Attribute names have shifted between minor versions (e.g. identifier/stream attributes), so confirm against your installed version rather than the blog post you're reading.

3. First use case: a single aggregate

Do not start with sagas or streaming projections. Start with one event-sourced aggregate and one projection, and prove the replay works.

#[EventSourcingAggregate]
final class Wallet
{
    use WithAggregateVersioning;              // provided by the package

    #[AggregateIdentifier]
    private string $id;
    private int $balance = 0;

    #[CommandHandler]
    public static function open(OpenWallet $command): self
    {
        $wallet = new self();
        $wallet->recordThat(new WalletOpened($command->walletId));
        return $wallet;
    }

    #[EventSourcingHandler]
    public function applyWalletOpened(WalletOpened $event): void
    {
        $this->id = $event->walletId;
    }
}

The key mental shift from a normal Ecotone handler: the command handler returns the aggregate, and Ecotone records everything you passed to recordThat() and persists it against the stream. You do not call a repository yourself — this is the pattern the package is built around, and fighting it (injecting EventSourcingRepository and calling save() manually) usually means you've mis-modelled the aggregate.

Then one projection:

#[Projection('wallet_balance')]
final class WalletBalance
{
    public function __construct(private Connection $connection) {}

    #[FromAggregateStream(Wallet::class)]
    #[ProjectionEventHandler]
    public function on(WalletOpened $event): void
    {
        $this->connection->insert('wallet_balance', ['wallet_id' => $event->walletId, 'balance' => 0]);
    }

    #[ProjectionInitialization]
    public function init(): void { /* DDL for wallet_balance */ }
}

Run the projection init command, fire a command, and check both the event_store table and wallet_balance. If both are right, you understand the package.

Implementation Patterns

Aggregate = write model, projection = read model, and never the twain shall meet

#[EventSourcingHandler] methods are replayed from the beginning of the stream every time an aggregate is loaded. That means they must be pure state application: no HTTP calls, no emitting other messages, no logging side effects that matter. Anything with an external effect belongs in a projection, which is a separate replayable, restartable consumer.

A useful rule: if you'd be upset about it happening 10,000 times during a replay, it does not go in an event sourcing handler.

Projection lifecycle is explicit, not automatic

Projections are ordinary consumers that track a position. The day-to-day loop is:

bin/console ecotone:es:projection:init     # create tables / seed
bin/console ecotone:es:projection:play     # catch up from current position
bin/console ecotone:es:projection:reset    # wipe and set position back to start

(Exact command names vary by bundle version — bin/console list ecotone is authoritative.) Run init and migrations as an explicit deploy step, never on application boot, or you'll get concurrent DDL on every rolling restart.

Rebuilding projections is your normal debugging tool

The single biggest win of this architecture: when a projection has a bug, you fix the handler, reset, and play. Make this cheap and boring. Two things make it cheap:

• Projection handlers must be idempotent — playback is at-least-once, and a crash mid-replay resumes from the last committed position.
• Keep projection DDL in #[ProjectionInitialization] so reset can recreate state without a migration.

I typically add an artisan/console wrapper in the project that does reset && play for one projection name, because you'll run it dozens of times.

Event-sourced sagas for durable workflows

#[EventSourcingSaga] gives you a process manager whose state is an event stream — so a multi-step workflow (payment → shipment → notification) survives deploys, restarts, and partial failures, and you can inspect exactly why it's stuck by reading its stream. Use a saga when the workflow spans days or external systems; use an in-aggregate command handler when it doesn't. Sagas are the hardest thing in this package to retrofit, so model the workflow explicitly and early if you know you need one.

Upcasting is how you refactor events

Events are immutable and permanent. When you rename OrderPlaced → OrderWasPlaced, or split one event into two, you register an upcaster that converts the stored representation into your current event class on read. Old rows keep their original payload; new code reads them as current classes. The practical consequence: never delete or rename an event class without an upcaster, and keep event classes in a stable, boring namespace. If you find yourself wanting to "fix" historical events, you want an upcaster.

Streaming projections for events that leave the database

With ecotone/kafka or ecotone/amqp installed, you can define a projection that consumes an external stream and maintains a read model, persisting its consumed position so restarts resume rather than reprocess. Two integration tips:

• Keep the external-stream projection's read model in the same DB as everything else so replay semantics are consistent.
• Version your event payloads before they leave the boundary — once events are on Kafka, upcasting is the only lever you have.

Transactional consistency by construction

Because the event store lives in the same relational database as your other data (via the DBAL connection), the write of aggregate events, the outbox rows, and any same-connection projection update can commit atomically. That's the main reason to pick this package over an external event store. Exploit it: put your outbox and your business tables on the same connection, and let Ecotone's DBAL integration handle the transaction boundary.

Testing: two tiers

• Fast tier: EcotoneLite::bootstrapFlowTesting([Wallet::class, WalletBalance::class]) with an in-memory event store. Aggregates, projections, and sagas run in-process — most of your tests live here and run in milliseconds.
• Integration tier: a real Postgres/MySQL instance (testcontainer or a dedicated CI database) for anything touching concurrency, locking, the actual SQL schema, or streaming.

Do not try to run the whole suite against in-memory; you'll miss exactly the bugs (optimistic locking, DDL, replay ordering) that matter.

Gotchas and Tips

Split repo confusion. Bug reports and PRs belong in ecotone-dev. If you file an issue here it will be ignored. Also: this package's version tracks ecotone/ecotone, so pin the core and let the split packages follow — don't pin ecotone/pdo-event-sourcing independently and expect a clean graph.

No SQLite, even in tests. This bites people who use :memory: SQLite for a fast suite and then discover the event store's SQL isn't portable. Use EcotoneLite for fast tests and a real DB for integration tests. Don't hack the store to support SQLite.

Optimistic concurrency is your problem to handle. WithAggregateVersioning gives you a version column and the store will throw a concurrency exception when two writers race. That's correct behaviour, not a bug — wrap command handling in a retry (with jitter) at the bus boundary, and make sure the retry re-fetches the aggregate rather than replaying a half-built in-memory one.

Projections that reference events already in the past. Adding a #[ProjectionEventHandler] to an existing projection does nothing until you reset and replay from the start of the stream. New projections start at position 0 by default, so this is usually right; modified projections are where people forget and end up with a subtly wrong read model. Treat every projection change as a replay.

Event class serialisation is a contract. Events are stored as typed JSON payloads. Rename a class, move a namespace, or change a constructor signature, and deserialisation breaks at replay time — potentially weeks later, in production, during an unrelated incident. Enforce a review checklist item: does this change the event class shape? If yes, ship an upcaster in the same PR.

Don't query the event store directly in application code. Read from projections. The store's schema is an implementation detail. The one legitimate exception is a read-only diagnostics page, which is genuinely useful during incidents.

Console commands fail silently-ish in CI. Projection play commands are long-running consumers. In CI, use the flow-testing bootstrap rather than shelling out to a consumer that never exits. In deploy scripts, make play a background worker with a supervisor, not a && in your release pipeline.

Stream naming and renames. Aggregates map to a stream, and projections subscribe by aggregate class or stream name. Renaming an aggregate class means its stream identity changes — which means existing events aren't found. If you must rename, use the stream-name attribute to pin the old identity explicitly rather than relying on the class name default.

GDPR / deletion. Events are immutable. You cannot delete a customer by deleting rows without corrupting the stream. Plan for this from the start: either a per-user stream you can drop wholesale, or crypto-shredding (encrypt PII with a per-subject key and destroy the key). Retrofitting either into a system with 50M events is a month of work.

Extension points, when attributes run out. The EventStore interface and EventSourcingConfiguration builder are the seams for custom store implementations or stream configuration. For serialisation quirks, custom message converters/upcasters are the right hook. Keep these thin — the value of this package is that it's declarative, and every custom implementation you add is a thing that breaks on upgrade.

Maturity caveat. Eight stars, zero dependents, and a README that's mostly a redirect to the parent project — this is a sub-package of a mature framework, not an independent library. Judge it by ecotone/ecotone's health (active, well-tested monorepo) rather than by this repo's metrics. But that also means: the split package has no independent release cadence or issue history, so when something breaks, debug against the monorepo source and tests, and expect to read vendor code rather than find a StackOverflow answer.