patchlevel/event-sourcing

The patchlevel/event-sourcing library is a fully-typed event sourcing implementation built on the Doctrine ecosystem. Key features include:

• Performance: Automatic snapshot system and the ability to split large aggregates into multiple streams.
• Lifecycle Management: Versioned and managed subscriptions for projections and processors.
• Data Privacy: Safe handling of Personal Data using crypto-shredding.
• Evolution: Smooth upcasting of old events to maintain compatibility with new schemas.
• Tooling: Schema management, Doctrine migrations, and built-in Symfony CLI commands.

The patchlevel/event-sourcing library is a developer-experience-oriented event sourcing implementation powered by the Doctrine ecosystem. Key features include:

• Performance: Automatic snapshot system to boost performance.
• Scalability: Ability to split big aggregates into multiple streams.
• Lifecycle Management: Versioned and managed lifecycle for subscriptions (projections and processors).
• Data Privacy: Safe usage of Personal Data via crypto-shredding.
• Evolution: Smooth upcasting of old events.
• Infrastructure: Scheme management, Doctrine migration support, and built-in Symfony CLI commands.

Upcasting is a mechanism used to transform event data before it is denormalized into an event object. It is useful when you have existing events in a stream that are missing data or contain data in an outdated format. Instead of creating many versioned event classes (which can lead to complexity), you can use an Upcaster to:

1. Adjust the payload: Modify existing keys or values within the event data.
2. Adjust the event name: Rename an event if the original name was poorly chosen.

An Upcaster operates on an Upcast object, which contains the eventName and the payload (as an array).

An aggregate is the central entity in event-sourcing. Unlike traditional ORM entities that save current state, an aggregate saves only the individual events that led to that state. This allows the current state to be rebuilt at any time by replaying the events.

To function as an aggregate in this library, a class must:

1. Implement the AggregateRoot interface.
2. Have a unique identifier.
3. Provide the current playhead.
4. Make state changes available as events.
5. Be able to rebuild/catchup its state from those events.

You can implement these requirements manually or use the BasicAggregateRoot implementation, which uses attributes to handle event configuration and state application.

A Translator acts as middleware to manipulate, filter, or expand messages or events in an event stream. Translators can return zero, one, or multiple messages, allowing you to remove, swap, or enrich the stream. You can group multiple translators using a ChainTranslator.

use Patchlevel\
EventSourcing\\Message\\Translator\\ChainTranslator;
use Patchlevel\
EventSourcing\\Message\\Translator\\ExcludeEventTranslator;
use Patchlevel\
EventSourcing\\Message\\Translator\\RecalculatePlayheadTranslator;

$translator = new ChainTranslator([
    new ExcludeEventTranslator([EmailChanged::class]),
    new RecalculatePlayheadTranslator(),
]);

A Message is a construct that wraps an event with additional metadata in the form of headers. Messages are automatically created, saved, and dispatched by the repository whenever an aggregate is saved.

Note that the Message object is immutable; methods like withHeader return a new instance of the message.

To manually create a message or add headers:

use Patchlevel\
EventSourcing\\Message\\Message;
use Patchlevel\\EventSourcing\\Aggregate\\AggregateHeader;
use Patchlevel\\EventSourcing\\Clock\\SystemClock;

$clock = new SystemClock();
$message = Message::create(new NameChanged('foo'))
    ->withHeader(new AggregateHeader(
        aggregateName: 'profile',
        aggregateId: 'bca7576c-536f-4428-b694-7b1f00c714b7',
        playhead: 2,
        recordedOn: $clock->now(),
    ));

Split Stream allows you to restart the event stream for an aggregate at a specific point in time. This is useful when business rules imply that past events are no longer relevant for the current state (e.g., a bank account performing a year-end balance report) or when an aggregate's event stream becomes so long that it impacts performance.

How it works:

• When a 'split event' occurs, the library marks all preceding events as archived.
• When loading an aggregate, the repository will only load events from the split event onwards.
• Subscriptions are unaffected: Projections and subscribers will still receive the full history of events, allowing you to maintain projections based on the complete event stream while keeping aggregate loading efficient.

If you need to rename an event class or its primary name while maintaining compatibility with existing data in your database, use the aliases option in the #[Event] attribute.

When saving new events, the primary name is always used. When loading (deserializing) events, the library will check both the primary name and any provided aliases to find the correct class.

use Patchlevel\EventSourcing\Attribute\Event;

#[Event(name: 'profile.registered', aliases: ['profile.created'])]
final class ProfileRegistered
{
}

You can handle multiple commands within a single handler method using three different approaches:

1. Multiple Attributes: Use multiple #[Handle] attributes on a single method.
2. Union Types: Use PHP union types in the parameter hint; the library automatically detects the commands.
3. Inheritance: The handler will be invoked if the command implements an interface or extends a class that the handler expects.

// Multiple Attributes
#[Handle(CreateProfile::class)]
#[Handle(UpdateProfile::class)]
public function __invoke(object $command): void { ... }

// Union Types
#[Handle]
public function __invoke(CreateProfile|UpdateProfile $command): void { ... }

// Inheritance
#[Handle]
public function __invoke(CommandInterface $command): void { ... }

The Event Bus is an optional component used to dispatch events to listeners. When an event is persisted via the save method on a repository, it is wrapped in a Message and dispatched to the event bus. The bus then triggers all registered listeners for that message.

Note: For more powerful and flexible message processing, it is recommended to use the subscription engine instead of the event bus.