Store

In the end, the messages have to be saved somewhere. Each message contains an event and the associated headers.

More information can be found in the message documentation.

The store is optimized to efficiently store and load events for aggregates.

Configure Store

We offer different stores to store the messages. Two stores based on doctrine dbal and one in-memory store for testing purposes.

DoctrineDbalStore

This is the current default store for event sourcing. You can create a store with the DoctrineDbalStore class. The store needs a dbal connection, an event serializer and has some optional parameters like options.

use Doctrine\DBAL\DriverManager;
use Doctrine\DBAL\Tools\DsnParser;
use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
use Patchlevel\EventSourcing\Store\DoctrineDbalStore;

$connection = DriverManager::getConnection(
    (new DsnParser())->parse('pdo-pgsql://user:secret@localhost/app'),
);

$store = new DoctrineDbalStore(
    $connection,
    DefaultEventSerializer::createFromPaths(['src/Event']),
);

You can find out more about how to create a connection in the doctrine dbal documentation.

Following options are available in DoctrineDbalStore:

Option	Type	Default	Description
table_name	string	eventstore	The name of the table in the database
aggregate_id_type	"uuid"/"string"	uuid	The type of the `aggregate_id` column
locking	bool	true	If the store should use locking for writing
lock_id	int	133742	The id of the lock
lock_timeout	int	-1	The timeout of the lock. -1 means no timeout

The table structure of the DoctrineDbalStore looks like this:

Column	Type	Description
id	bigint	The index of the whole stream (autoincrement)
aggregate	string	The name of the aggregate
aggregate_id	uuid/string	The id of the aggregate
playhead	int	The current playhead of the aggregate
event	string	The name of the event
payload	json	The payload of the event
recorded_on	datetime	The date when the event was recorded
new_stream_start	bool	If the event is the first event of the aggregate
archived	bool	If the event is archived
custom_headers	json	Custom headers for the event

The default type of the aggregate_id column is uuid if the database supports it and string if not. You can change the type with the aggregate_id_type option to string if you want to use a custom id.

StreamDoctrineDbalStore

We offer a new store called StreamDoctrineDbalStore. This store is decoupled from the aggregate and can be used to store events from other sources. The difference to the DoctrineDbalStore is that the StreamDoctrineDbalStore merges the aggregate id and the aggregate name into one column named stream. Additionally, the column playhead is nullable. This store introduces two new methods streams and remove.

The store needs a dbal connection, an event serializer and has some optional parameters like options.

use Doctrine\DBAL\DriverManager;
use Doctrine\DBAL\Tools\DsnParser;
use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore;

$connection = DriverManager::getConnection(
    (new DsnParser())->parse('pdo-pgsql://user:secret@localhost/app'),
);

$store = new StreamDoctrineDbalStore(
    $connection,
    DefaultEventSerializer::createFromPaths(['src/Event']),
);

You can find out more about how to create a connection in the doctrine dbal documentation.

Following options are available in StreamDoctrineDbalStore:

Option	Type	Default	Description
table_name	string	event_store	The name of the table in the database
locking	bool	true	If the store should use locking for writing
lock_id	int	133742	The id of the lock
lock_timeout	int	-1	The timeout of the lock. -1 means no timeout
keep_index	bool	false	If enabled, the index header is kept on save

The table structure of the StreamDoctrineDbalStore looks like this:

Column	Type	Description
id	bigint	The index of the whole stream (autoincrement)
stream	string	The name of the stream
playhead	?int	The current playhead of the aggregate
event_id	string	The id of the event
event_name	string	The name of the event
event_payload	json	The payload of the event
recorded_on	datetime	The date when the event was recorded
new_stream_start	bool	If the event is the first event of the aggregate
archived	bool	If the event is archived
custom_headers	json	Custom headers for the event

InMemoryStore

We also offer an in-memory store for testing purposes.

use Patchlevel\EventSourcing\Store\InMemoryStore;

$store = new InMemoryStore();

You can pass messages to the constructor to initialize the store with some events.

ReadOnlyStore & StreamReadOnlyStore

Last but not least, we offer two read-only stores. One for the DoctrineDbalStore and one for the StreamDoctrineDbalStore. It passes all methods to the underlying store, but throws a StoreIsReadOnly exception when trying to execute write operations.

use Patchlevel\EventSourcing\Store\ReadOnlyStore;
use Patchlevel\EventSourcing\Store\Store;
use Patchlevel\EventSourcing\Store\StreamReadOnlyStore;
use Patchlevel\EventSourcing\Store\StreamStore;

/** @var Store $store */
$readOnlyStore = new ReadOnlyStore($store);

/** @var StreamStore $store */
$readOnlyStore = new StreamReadOnlyStore($store);

Schema

With the help of the SchemaDirector, the database structure can be created, updated and deleted.

You can also use doctrine migration to create and keep your schema in sync.

Doctrine Schema Director

The SchemaDirector is responsible for creating, updating and deleting the database schema. The DoctrineSchemaDirector is a concrete implementation of the SchemaDirector for doctrine dbal. Additionally, it implements the DryRunSchemaDirector interface, to show the sql statements that would be executed.

use Doctrine\DBAL\Connection;
use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
use Patchlevel\EventSourcing\Store\Store;

/**
 * @var Connection $connection
 * @var Store $store
 */
$schemaDirector = new DoctrineSchemaDirector(
    $connection,
    $store,
);

How to setup cli commands for the schema director can be found in the cli documentation.

Create schema

You can create the table from scratch using the create method.

use Patchlevel\EventSourcing\Schema\SchemaDirector;

/** @var SchemaDirector $schemaDirector */
$schemaDirector->create();

Or can give you back which SQL statements would be necessary for this. Either for a dry run, or to define your own migrations.

use Patchlevel\EventSourcing\Schema\DryRunSchemaDirector;

/** @var DryRunSchemaDirector $schemaDirector */
$sql = $schemaDirector->dryRunCreate();

Update schema

The update method compares the current state in the database and how the table should be structured. As a result, the diff is executed to bring the table to the desired state.

use Patchlevel\EventSourcing\Schema\SchemaDirector;

/** @var SchemaDirector $schemaDirector */
$schemaDirector->update();

Or can give you back which SQL statements would be necessary for this.

use Patchlevel\EventSourcing\Schema\DryRunSchemaDirector;

/** @var DryRunSchemaDirector $schemaDirector */
$sql = $schemaDirector->dryRunUpdate();

Drop schema

You can also delete the table with the drop method.

use Patchlevel\EventSourcing\Schema\SchemaDirector;

/** @var SchemaDirector $schemaDirector */
$schemaDirector->drop();

Or can give you back which SQL statements would be necessary for this.

use Patchlevel\EventSourcing\Schema\DryRunSchemaDirector;

/** @var DryRunSchemaDirector $schemaDirector */
$sql = $schemaDirector->dryRunDrop();

Doctrine Migrations

You can use doctrine migration, which is known from doctrine orm, to create your schema and keep it in sync. We have added a DoctrineMigrationSchemaProvider for doctrine migrations so that you just have to plug the whole thing together.

use Doctrine\DBAL\Connection;
use Doctrine\Migrations\Configuration\Connection\ExistingConnection;
use Doctrine\Migrations\Configuration\Migration\ConfigurationLoader;
use Doctrine\Migrations\DependencyFactory;
use Doctrine\Migrations\Provider\SchemaProvider;
use Patchlevel\EventSourcing\Schema\DoctrineMigrationSchemaProvider;
use Patchlevel\EventSourcing\Schema\DoctrineSchemaDirector;
use Patchlevel\EventSourcing\Store\Store;

// event sourcing schema director configuration

/**
 * @var Connection $connection
 * @var Store $store
 */
$schemaDirector = new DoctrineSchemaDirector(
    $connection,
    $store,
);

$schemaProvider = new DoctrineMigrationSchemaProvider($schemaDirector);

// doctrine migration configuration

/** @var ConfigurationLoader $configLoader */
$dependencyFactory = DependencyFactory::fromConnection(
    $configLoader,
    new ExistingConnection($connection),
);

$dependencyFactory->setService(
    SchemaProvider::class,
    $schemaProvider,
);

Here you can find more information on how to configure doctrine migration.

How to setup cli commands for doctrine migration can be found in the cli documentation.

Usage

The store has a few methods to interact with the database.

Load

You can load all events from an aggregate with the load method. This method returns a Stream object, which is a collection of events.

use Patchlevel\EventSourcing\Store\Store;

/** @var Store $store */
$stream = $store->load();

The load method also has a few parameters to filter, limit and sort the events.

use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Store;

/** @var Store $store */
$stream = $store->load(
    new Criteria(), // filter criteria
    100, // limit
    50, // offset
    true,  // latest first
);

Criteria

The Criteria object is used to filter the events.

use Patchlevel\EventSourcing\Store\Criteria\AggregateIdCriterion;
use Patchlevel\EventSourcing\Store\Criteria\AggregateNameCriterion;
use Patchlevel\EventSourcing\Store\Criteria\ArchivedCriterion;
use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion;
use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion;
use Patchlevel\EventSourcing\Store\Criteria\ToIndexCriterion;

$criteria = new Criteria(
    new AggregateNameCriterion('profile'),
    new AggregateIdCriterion('e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e'),
    new FromPlayheadCriterion(2),
    new FromIndexCriterion(100),
    new ToIndexCriterion(200),
    new ArchivedCriterion(true),
    new EventsCriterion(['profile.created', 'profile.name_changed']),
);

Or you can use the criteria builder to create the criteria.

use Patchlevel\EventSourcing\Store\Criteria\CriteriaBuilder;

$criteria = (new CriteriaBuilder())
    ->aggregateName('profile')
    ->aggregateId('e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e')
    ->fromPlayhead(2)
    ->fromIndex(100)
    ->archived(true)
    ->events(['profile.created', 'profile.name_changed'])
    ->build();

Criteria for Stream Stores

The StreamDoctrineDbalStore does not know about aggregates, it works with stream names. Instead of the aggregate criteria you use the StreamCriterion, which also supports a * wildcard. On top of that it offers the ToPlayheadCriterion and the EventIdCriterion.

use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Criteria\EventIdCriterion;
use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion;
use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
use Patchlevel\EventSourcing\Store\Criteria\ToPlayheadCriterion;

$criteria = new Criteria(
    new StreamCriterion('profile-e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e'),
    new FromPlayheadCriterion(2),
    new ToPlayheadCriterion(10),
);

$criteria = new Criteria(
    new EventIdCriterion('a4a4a4a4-4a4a-4a4a-4a4a-4a4a4a4a4a4a'),
);

The StreamCriterion is variadic, so you can pass multiple stream names. There is also a startWith named constructor that appends the wildcard for you.

use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;

$criterion = new StreamCriterion('profile-*', 'hotel-*');
$criterion = StreamCriterion::startWith('profile-');

The criteria builder covers this as well:

use Patchlevel\EventSourcing\Store\Criteria\CriteriaBuilder;

$criteria = (new CriteriaBuilder())
    ->streamName('profile-*')
    ->fromPlayhead(2)
    ->toPlayhead(10)
    ->build();

Not every store supports every criterion. If a store gets a criterion it cannot handle, it throws an UnsupportedCriterion exception. StreamCriterion, ToPlayheadCriterion and EventIdCriterion only work with the StreamDoctrineDbalStore, AggregateNameCriterion and AggregateIdCriterion only with the DoctrineDbalStore.

Stream

The load method returns a Stream object and is a generator. This means that the messages are only loaded when they are needed.

use Patchlevel\EventSourcing\Store\Stream;

/** @var Stream $stream */
$stream->index(); // get the index of the stream
$stream->position(); // get the current position of the stream
$stream->current(); // get the current event
$stream->next(); // move to the next event
$stream->end(); // check if the stream is at the end

foreach ($stream as $message) {
    $message->event(); // get the event
}

You can find more information about the message object in the message documentation.

The stream cannot rewind, so you can only iterate over it once. If you want to iterate over it again, you have to call the load method again.

Count

You can count the number of events in the store with the count method.

use Patchlevel\EventSourcing\Store\Store;

/** @var Store $store */
$count = $store->count();

The count method also has the possibility to filter the events.

use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Store;

/** @var Store $store */
$count = $store->count(
    new Criteria(), // filter criteria
);

Save

You can save a message with the save method.

use Patchlevel\EventSourcing\Message\Message;
use Patchlevel\EventSourcing\Store\Store;

/**
 * @var Store $store
 * @var Message $message
 * @var Message $message1
 * @var Message $message2
 * @var Message $message3
 * @var list<Message> $messages
 */
$store->save($message);
$store->save($message1, $message2, $message3);
$store->save(...$messages);

The saving happens in a transaction, so all messages are saved or none. The store locks the table for writing during each save by default.

Use the transactional method if you want to call multiple save methods in one transaction.

Custom streams

The StreamDoctrineDbalStore is not limited to aggregates. You can save any event into your own stream, for example integration events that are derived from your domain events and consumed by other bounded contexts. For this, the message needs a StreamNameHeader with the name of the stream.

use Patchlevel\EventSourcing\Attribute\Event;

#[Event('hotel.integration.guest_checked_in')]
final class GuestCheckedInIntegrationEvent
{
    public function __construct(
        public readonly string $hotelId,
        public readonly string $guestName,
    ) {
    }
}

use Patchlevel\EventSourcing\Attribute\Processor;
use Patchlevel\EventSourcing\Attribute\Subscribe;
use Patchlevel\EventSourcing\Message\Message;
use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
use Patchlevel\EventSourcing\Store\Store;

#[Processor('hotel_integration_publisher')]
final class HotelIntegrationPublisher
{
    public function __construct(
        private readonly Store $store,
    ) {
    }

    #[Subscribe(GuestIsCheckedIn::class)]
    public function onGuestIsCheckedIn(GuestIsCheckedIn $event): void
    {
        $this->store->save(
            Message::create(new GuestCheckedInIntegrationEvent(
                $event->hotelId->toString(),
                $event->guestName,
            ))->withHeader(new StreamNameHeader('integration-hotel')),
        );
    }
}

These messages are part of the store like any other message. Every subscriber can subscribe to them, and you can load them with the StreamCriterion.

use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
use Patchlevel\EventSourcing\Store\Store;

/** @var Store $store */
$stream = $store->load(
    new Criteria(new StreamCriterion('integration-hotel')),
);

Only the StreamNameHeader is required, the other headers are optional:

Header	If missing
`StreamNameHeader`	A `MissingDataForStorage` exception is thrown
`PlayheadHeader`	The playhead is saved as `null`
`EventIdHeader`	A new UUIDv7 is generated
`RecordedOnHeader`	The current time of the clock is used

Choose a stream name that cannot collide with your aggregate streams, which are named like profile-e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e by default. A prefix like integration- helps.

Without a PlayheadHeader there is no optimistic locking for this stream, because the unique index on stream and playhead does not apply to null values.

Custom streams are only supported by the StreamDoctrineDbalStore. The event class still needs the #[Event] attribute and must be known by the event serializer.

Update

It is not possible to update events. In event sourcing, the events are immutable.

Remove

You can remove events with the remove method. It takes the same criteria as the load method.

use Patchlevel\EventSourcing\Store\Criteria\Criteria;
use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
use Patchlevel\EventSourcing\Store\StreamStore;

/** @var StreamStore $store */
$store->remove(new Criteria(StreamCriterion::startWith('profile-')));

Without criteria the method removes every event in the store. Deleted events cannot be restored, all subscriptions built from them become inconsistent.

The method is only available in the StreamStore like StreamDoctrineDbalStore.

List Streams

You can list all streams with the streams method.

use Patchlevel\EventSourcing\Store\StreamStore;

/** @var StreamStore $store */
$streams = $store->streams(); // ['profile-1', 'profile-2', 'profile-3']