Event Store Bus Bridge Laravel Package

Getting Started

What it actually gives you. Two (sometimes three) glue classes between prooph/event-store and a message bus:

• Prooph\EventStoreBusBridge\TransactionManager — a message bus plugin. It opens a transaction on the event store when a command starts being handled and commits it when the handler returns; on an exception it rolls back.
• Prooph\EventStoreBusBridge\EventPublisher — an event store plugin. It takes the events recorded during an append and hands them to a MessageProducer (usually your EventBus) after the write succeeds.
• Prooph\EventStoreBusBridge\AggregateTypePlugin — present in the 1.x line; stamps/reads _aggregate_type metadata so events can be routed back to the right stream.

Minimal setup (event-store v7 / service-bus v6 era):

composer require prooph/event-store-bus-bridge

use Prooph\EventStore\ActionEventEmitterEventStore;
use Prooph\EventStore\InMemoryEventStore;
use Prooph\EventStoreBusBridge\EventPublisher;
use Prooph\EventStoreBusBridge\TransactionManager;

$eventStore = new ActionEventEmitterEventStore(new InMemoryEventStore());
$eventBus   = new EventBus();   // or a MessageProducer of your own
$commandBus = new CommandBus();

// 1. Transaction around command handling
$commandBus->utilize(new TransactionManager($eventStore));

// 2. Publish committed events onto the bus
$eventPublisher = new EventPublisher($eventBus);
$eventPublisher->attachTo($eventStore->getActionEventEmitter());

// 3. Same $eventStore instance must be injected everywhere
$commandBus->attach(
    RegisterUser::class,
    new RegisterUserHandler($eventStore)
);

Look first at:

1. src/TransactionManager.php — read the attachTo()/utilize() method to see exactly which MessageBus actions it hooks (action / finalize) and in what order it commits.
2. src/EventPublisher.php — see whether it listens to commit, appendTo, or create post-actions; that determines when your listeners run relative to your DB commit.
3. The Plugin / MessageBus\Plugin interface versions you have installed (prooph/event-store ^7 vs the newer event-bus split). The wiring differs between the 1.x and later lines — check composer show prooph/event-store-bus-bridge before copying any blog post.

First useful use case: a RegisterUser command whose handler loads the user stream, calls a domain method that records UserRegistered, and calls $eventStore->appendTo($streamName, $user->uncommittedEvents()). The TransactionManager guarantees the write is atomic with anything else the handler touched, and EventPublisher fires UserRegistered to the event bus for projectors/read models — you never dispatch() the event yourself.

Implementation Patterns

Command handler as the unit of work. The canonical shape under this bridge:

final class RegisterUserHandler
{
    public function __construct(private EventStore $eventStore) {}

    public function __invoke(RegisterUser $command): void
    {
        $stream = new StreamName('user', $command->userId());

        $user = User::reconstituteFromHistory(
            $this->eventStore->load($stream) // empty history is fine
        );

        $user->register($command->userId(), $command->email());

        $this->eventStore->appendTo($stream, ...$user->uncommittedEvents());
    }
}

The handler does not wrap itself in beginTransaction(). Let the bus plugin own that boundary; nesting it manually is the fastest way to break rollback semantics.

Projectors/read models as event-bus listeners. Because EventPublisher pushes to the event bus after the store accepts the write, projectors receive only events that survived the transaction. That is the whole value proposition — keep projectors free of try/catch-and-swallow, since a thrown exception there surfaces from commit().

Route events back to writes only through the store. Do not let an event-bus listener call $eventStore->appendTo() for the same aggregate. The bridge is one-directional (store → bus). If you need "process managers", treat them as new commands dispatched to the command bus, so the TransactionManager wraps them too.

Multiple buses, multiple publishers. One EventPublisher per bus is fine — attach the same instance to several emitters if you fan out, or attach two publishers with different producers (one sync, one queued). Keep the transaction manager attached exactly once per command bus.

Custom producers for async. EventPublisher only needs something implementing the producer/event-bus contract. Wrap your real transport:

final class OutboxProducer implements MessageProducer
{
    public function __construct(private Connection $db) {}

    public function __invoke(Message $message): void
    {
        // written in the same DB transaction the TransactionManager opened
        $this->db->insert('outbox', ['payload' => serialize($message)]);
    }
}

Then a worker drains outbox to the real broker. This is the pattern that survives production; the naive "publish straight to RabbitMQ inside the commit" variant does not.

Testing pattern. Use InMemoryEventStore + MessageBus ->utilize() with a spy listener; assert on the stream contents, not on bus dispatches, so you actually verify the transaction manager committed.

Gotchas and Tips

• Shared event store instance. TransactionManager and every handler must hold the same ActionEventEmitterEventStore object. Two instances over one DB connection means the transaction is opened on a connection nothing writes to, and rollback silently does nothing. Bind it as a singleton in your container.
• Publish-after-commit is not atomic with the broker. If the event bus is remote, the DB commit happens first; a broker failure leaves committed events unpublished. Use the outbox producer above, or you will be reconciling by hand forever.
• Nested command dispatch. A handler that dispatches another command opens a second transaction. Many stores either error on nested beginTransaction() or autocommit the inner one, defeating the outer rollback. Track a depth counter in a decorator, or enforce "handlers never dispatch commands synchronously".
• Listener ordering matters. TransactionManager must be first to begin and last to commit. With prooph/service-bus plugins this is usually handled by the plugin's own priorities; with plain attach() you must attach it before handlers. Verify with a debug listener printing action names.
• Throwing inside an event-bus listener (e.g. a projector failing) will bubble out of commit() after the data is durable. You cannot roll back a committed stream. Decide up front: log-and-continue, or retry queue.
• _aggregate_type / _aggregate_version metadata. The 1.x AggregateTypePlugin and the event store's stream resolution depend on this metadata being present on messages when publishing. A hand-rolled event class without withAddedMetadata(['_aggregate_type' => ...]) will route to the wrong stream or throw RuntimeException from the publisher. Use the aggregate's popRecordedEvents() + withAddedMetadata() helper rather than new-ing messages inline.
• In-memory event store hides transaction bugs. InMemoryEventStore has no real transactions; a missing rollback passes all unit tests. Always keep one integration test against the real store (PDO/Mongo) asserting that a handler exception leaves zero rows in event_stream.
• Debugging. Attach a catch-all listener to $eventStore->getActionEventEmitter()->attachListener('commit', fn($e) => dump($e->getParam('stream'))); — the action-event API is the only place you can observe what the bridge is doing, since both classes are intentionally thin.
• Legacy status is the biggest gotcha. Last release was 2021, prooph/* is archived, and the code targets PHP 7.x / prooph/event-store v7. Expect Composer conflicts on PHP 8.1+ and no upstream security fixes. For new builds, copy the ~200 lines of TransactionManager + EventPublisher into your own src/ and depend directly on event-store — you keep the exact behavior with a maintainable surface. Plan a migration to patchlevel/event-sourcing or a doctrine-based outbox if the project is long-lived.
• Extension points worth knowing: implement Prooph\Common\Event\ActionEventListener to observe store actions, implement Prooph\EventStore\Plugin\Plugin (attachTo(ActionEventEmitter)) to add your own behavior on appendTo/create, and decorate the MessageProducer to add tracing/correlation IDs to metadata before dispatch — the bridge itself stays untouched.