Recorded, Not Injected

A domain event announces that something happened, for instance "the booking was cancelled". Publishing it while the aggregate is still handling the cancellation means announcing it before knowing whether it will actually be stored in the database.

The fix isn't a cleverer configuration: you have to wait until the database has confirmed the cancellation before publishing anything.

This article shows why publishing from the aggregate amounts to lying, how to make it record its events instead of publishing them, and where to publish them: in the use case, once save() has succeeded.

The context: a cancellation the rest of the system must hear about

This series uses one small hotel-booking domain. Its aggregate root, Booking, owns a list of charges and one business rule: cancelling less than 48 hours before check-in adds a penalty worth half the room rate. Amounts are Money, the Value Object from the first article.

When Booking::cancel() runs and applies a penalty, the rest of the system (sending emails, accounting) has to be told.

The problem: publishing too early

The most intuitive fix is also the most dangerous one: inject an event publisher (EventPublisher) into the aggregate's constructor, and call it directly in cancel().

final class BookingWithInjectedPublisher
{
    private BookingStatus $status;

    public function __construct(
        private readonly EventPublisher $events,
        public readonly BookingId $id,
    ) {
    }

    public function cancel(\DateTimeImmutable $now): void
    {
        // ... penalty logic ...

        $this->events->publish(new BookingCancelled(EventId::generate(), $this->id, $now));
        $this->status = BookingStatus::Cancelled;
    }
}

At first glance, the code reads clean. In reality, it has three flaws:

1. The transactional lie: if the database throws right after (lost connection, constraint violation), the application fails, but the event has already gone out. A cancellation email has just been sent to the guest for a booking that is still active in the database.
2. The ORM trap: if the aggregate is mapped directly by Doctrine, the ORM reloads it from the database without ever calling its constructor. The $events property is then never initialized: on any booking loaded from the database, the first call to cancel() throws an Error ("must not be accessed before initialization").
3. The layering violation: EventPublisher is a port of the Application layer. Injecting it into the aggregate makes the domain depend on the rest of the application, when it shouldn't depend on anything.

The fix? The aggregate stops publishing anything. It just records the facts in memory.

The solution: the aggregate records, it doesn't publish

So that the aggregate no longer needs a publisher, give it an abstract base class. Its only job is to remember the events that happened, without knowing what they mean.

<?php

declare(strict_types=1);

namespace App\Booking\Domain;

use App\Booking\Domain\Event\DomainEvent;

abstract class AggregateRoot
{
    /** @var DomainEvent[] */
    private array $recordedEvents = [];

    protected function recordThat(DomainEvent $event): void
    {
        $this->recordedEvents[] = $event;
    }

    /** @return DomainEvent[] */
    public function pullEvents(): array
    {
        $events = $this->recordedEvents;
        $this->recordedEvents = [];

        return $events;
    }
}

Events implement DomainEvent, an empty interface used only as a label: thanks to it, recordThat() rejects anything that isn't an event.

pullEvents() returns the recorded events and empties the internal array: a second call returns nothing. So the same use case can't publish the same event twice. On the other hand, the code reacting to the event (sending the email, for instance) can still receive it twice: that's the subject of article 4.

With this base, Booking no longer depends on anything outside itself. It just calls recordThat() at the moment a fact becomes true. In cancel(), that happens twice: when a penalty is added, then when the booking switches to cancelled.

public function cancel(\DateTimeImmutable $now): void
{
    if ($this->status === BookingStatus::Cancelled) {
        return;
    }

    if ($this->stay->hoursUntilCheckIn($now) < 48) {
        $penalty = new Charge(
            ChargeId::generate(),
            ChargeType::CancellationPenalty,
            $this->roomRateCharge()->percentageOfAmount(50),
            'Late cancellation penalty (less than 48h before check-in)',
        );
        $this->charges[] = $penalty;

        $this->recordThat(new CancellationPenaltyCharged(
            EventId::generate(),
            $this->id,
            $penalty->id,
            $penalty->amount,
            $now,
        ));
    }

    $this->status = BookingStatus::Cancelled;

    $this->recordThat(new BookingCancelled(EventId::generate(), $this->id, $now));
}

Nothing in this code talks to Symfony, Messenger or an SMTP server. The aggregate knows nothing about infrastructure.

The use case: where the truth gets confirmed

The use case is what runs the steps in order. It's also the only place that knows two things at once: whether the save succeeded, and which events the aggregate recorded.

This is the one place where the EventPublisher interface belongs:

interface EventPublisher
{
    public function publish(DomainEvent $event): void;
}

final readonly class CancelBooking
{
    public function __construct(
        private BookingRepository $bookings,
        private EventPublisher $events,
    ) {
    }

    public function __invoke(CancelBookingCommand $command): void
    {
        $booking = $this->bookings->get($command->bookingId);

        $booking->cancel($command->now);

        $this->bookings->save($booking);

        foreach ($booking->pullEvents() as $event) {
            $this->events->publish($event);
        }
    }
}

The use case loads the aggregate, runs the business logic, saves, and only then publishes. If save() throws, the foreach loop is never reached. The implementation of EventPublisher, MessengerEventPublisher, sits in the Infrastructure layer. It hands each event to Messenger's bus, as in any API built on Symfony. The bus dispatches the event, but never decides whether it happened.

In practice: proving it with tests

Checking that cancel() records the right events, in the right order, needs nothing but the aggregate itself:

#[Test]
public function cancelRecordsThePenaltyThenTheCancellationEvent(): void
{
    $checkIn = new \DateTimeImmutable('2027-01-10 15:00:00');
    $booking = BookingBuilder::aBooking()->checkingInOn($checkIn)->build();

    $booking->cancel($checkIn->modify('-24 hours'));

    $events = $booking->pullEvents();
    self::assertCount(2, $events);
    self::assertInstanceOf(CancellationPenaltyCharged::class, $events[0]);
    self::assertInstanceOf(BookingCancelled::class, $events[1]);
}

No publisher to fake, no bus to mock, no Symfony container. That's the whole point of keeping Messenger out of the domain.

At the use case level, two fakes are enough to prove the article's thesis: publish() never runs if save() failed. The first is a publisher that refuses to be called:

final class EventPublisherThatMustNotBeCalled implements EventPublisher
{
    public function publish(DomainEvent $event): void
    {
        throw new \LogicException('publish() must not run unless save() already succeeded.');
    }
}

The second, FailingBookingRepository, returns the requested booking but throws \RuntimeException('Connection lost.') on every save(). The test:

#[Test]
public function publishNeverRunsUnlessSaveDid(): void
{
    $booking = BookingBuilder::aBooking()->build();
    $useCase = new CancelBooking(
        new FailingBookingRepository($booking),
        new EventPublisherThatMustNotBeCalled(),
    );
    $this->expectException(\RuntimeException::class);
    $this->expectExceptionMessage('Connection lost.');

    $useCase(new CancelBookingCommand($booking->id, new \DateTimeImmutable()));
}

cancel() did record the events in memory, but they never reach a publisher. If CancelBooking published before save() finished, this publisher would throw a \LogicException instead of the expected \RuntimeException, and the test would fail. No extra assertion needed: the publisher does the checking.

This test doesn't see everything. If the use case forgot to publish, it would still pass. Hence a second test, checking that after a successful save(), every recorded event does get published:

#[Test]
public function publishRunsForEveryRecordedEventOnceSaveSucceeded(): void
{
    $checkIn = new \DateTimeImmutable('2027-01-10 15:00:00');
    $booking = BookingBuilder::aBooking()->checkingInOn($checkIn)->build();
    $publisher = new RecordingEventPublisher();
    $useCase = new CancelBooking(new InMemoryBookingRepository($booking), $publisher);

    $useCase(new CancelBookingCommand($booking->id, $checkIn->modify('-24 hours')));

    self::assertCount(2, $publisher->published());
}

Things to watch

Publishing from the use case fixes the original problem, but creates three smaller risks. They share one cause: nothing forces the code to be written in the right order.

• Forgetting to call pullEvents(). The booking is cancelled in the database, but no email goes out and accounting is never told. No exception, no warning: only the second test above catches it.
• Calling pullEvents() before save(). Swap those two lines and you're right back to the original problem: the events go out before the save is confirmed. That's what the first test catches: the test publisher would be called before save() fails.
• Calling pullEvents() in the repository. Moving publication into BookingRepository::save() means it can't be forgotten in the use case, which is exactly why it's tempting. But then the repository decides when the application announces what happened, when its only job is writing to the database. And the day it's swapped for another implementation, an in-memory one for tests say, event publication silently goes with it. A Doctrine postFlush listener publishing the saved aggregates' events has exactly the same flaw, only better hidden: nothing in the use case shows that publication happens.

The limit: what if publishing fails after save()?

Everything above assumes CancelBooking's publishing loop runs in the same request as save(). If the process stops between the two, replaying the command recovers nothing: cancel() does nothing when the booking is already cancelled, so no event is recorded, or published. No email will ever go out, and no error will report it.

The same goes if publish() fails halfway through the loop: pullEvents() has already emptied the aggregate's events, so the ones not yet published are lost, and replaying the command won't find them.

Finally, this whole article assumes save() commits its transaction. If the use case runs inside a transactional middleware, like Messenger's doctrine_transaction, the transaction only commits after the use case returns: the events then go out before the commit. Article 4 covers that case.

A transactional outbox solves exactly these problems. Does this system need one? That's the subject of Does This Need an Outbox?

Conclusion

The EventPublisher interface was never a bad idea in itself. The mistake was only when, and from where, it got called.

The aggregate records, the use case publishes once the save is confirmed: the application no longer announces anything too early. A domain event should only be announced once the database has confirmed what it claims.

If your domain still depends on Symfony (its bus, its services, its container), that's one of the first things I look at in a code audit.

💬 Working on something similar?

Send me an email Discuss your project on LinkedIn See my profile on Malt

Comments