diff --git a/README.md b/README.md index 24e485455..8da12a21d 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,8 @@ # Event-Sourcing -A lightweight but also all-inclusive event sourcing library with a focus on developer experience. +An event sourcing library, complete with all the essential features, +powered by the reliable Doctrine ecosystem and focused on developer experience. ## Features @@ -14,10 +15,10 @@ A lightweight but also all-inclusive event sourcing library with a focus on deve * Developer experience oriented and fully typed * Automatic [snapshot](https://patchlevel.github.io/event-sourcing-docs/latest/snapshots/)-system to boost your performance * [Split](https://patchlevel.github.io/event-sourcing-docs/latest/split_stream/) big aggregates into multiple streams -* Build-in [pipeline](https://patchlevel.github.io/event-sourcing-docs/latest/pipeline/) to export, import and migrate event streams -* Versioned and managed lifecycle of [projections](https://patchlevel.github.io/event-sourcing-docs/latest/projection/) +* Versioned and managed lifecycle of [subscriptions](https://patchlevel.github.io/event-sourcing-docs/latest/subscription/) like projections and processors +* Safe usage of [Personal Data](https://patchlevel.github.io/event-sourcing-docs/latest/personal_data/) with crypto-shredding * Smooth [upcasting](https://patchlevel.github.io/event-sourcing-docs/latest/upcasting/) of old events -* Simple setup with [scheme management](https://patchlevel.github.io/event-sourcing-docs/latest/store/) and [doctrine migration](https://patchlevel.github.io/event-sourcing-docs/latest/migration/) +* Simple setup with [scheme management](https://patchlevel.github.io/event-sourcing-docs/latest/store/) and [doctrine migration](https://patchlevel.github.io/event-sourcing-docs/latest/store/) * Built in [cli commands](https://patchlevel.github.io/event-sourcing-docs/latest/cli/) with [symfony](https://symfony.com/) * and much more... @@ -30,6 +31,7 @@ composer require patchlevel/event-sourcing ## Documentation * Latest [Docs](https://patchlevel.github.io/event-sourcing-docs/latest) +* 3.0 [Docs](https://patchlevel.github.io/event-sourcing-docs/3.0) (preview) ## Integration @@ -42,6 +44,9 @@ We officially only support the databases and versions listed in the table, as th Since the package is based on doctrine dbal, other databases such as OracleDB and MSSQL may also work. But we can only really support the databases if we can also automatically ensure that they don't break due to changes. +> [!TIP] +> We recommend using PostgresSQL. + | Database | Version | |-------------|---------------------------------| | PostgresSQL | 12.17, 13.13, 14.10, 15.5, 16.1 | diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index a427d9f7a..a5d6c06d2 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -83,15 +83,14 @@ nav: - Repository: repository.md - Message: message.md - Store: store.md - - Event Bus: event_bus.md - Subscription: subscription.md + - Event Bus: event_bus.md - Advanced: - Aggregate ID: aggregate_id.md - Normalizer: normalizer.md - Snapshots: snapshots.md - Personal Data: personal_data.md - Upcasting: upcasting.md - - Outbox: outbox.md - Message Decorator: message_decorator.md - Split Stream: split_stream.md - Time / Clock: clock.md diff --git a/docs/pages/cli.md b/docs/pages/cli.md index a1ac7320d..0c272a6bf 100644 --- a/docs/pages/cli.md +++ b/docs/pages/cli.md @@ -145,4 +145,8 @@ $cli->addCommands([ Here you can find more information on how to [configure doctrine migration](https://www.doctrine-project.org/projects/doctrine-migrations/en/3.3/reference/custom-configuration.html). - \ No newline at end of file + +## Learn more + +* [How to configure store](store.md) +* [How to configure subscription engine](subscription.md) diff --git a/docs/pages/clock.md b/docs/pages/clock.md index 65c6992bd..4939c3e67 100644 --- a/docs/pages/clock.md +++ b/docs/pages/clock.md @@ -69,5 +69,4 @@ $clock->sleep(10); // sleep 10 seconds * [How to test with datetime](testing.md) * [How to normalize datetime](normalizer.md) -* [How to use messages](event_bus.md) -* [How to decorate messages](message_decorator.md) +* [How to use messages](message.md) diff --git a/docs/pages/event_bus.md b/docs/pages/event_bus.md index e660bd900..9294561c7 100644 --- a/docs/pages/event_bus.md +++ b/docs/pages/event_bus.md @@ -139,6 +139,7 @@ $eventBus = new Psr14EventBus($psr14EventDispatcher); ## Learn more * [How to use messages](message.md) +* [How to use events](events.md) * [How to use the subscription engine](subscription.md) * [How to use repositories](repository.md) * [How to use decorate messages](message_decorator.md) diff --git a/docs/pages/events.md b/docs/pages/events.md index 3a995aabe..14e5196e1 100644 --- a/docs/pages/events.md +++ b/docs/pages/events.md @@ -113,8 +113,7 @@ $eventRegistry = (new AttributeEventRegistryFactory())->create([/* paths... */]) ## Learn more * [How to normalize events](normalizer.md) -* [How to dispatch events](event_bus.md) -* [How to listen on events](processor.md) +* [How to subscribe on events](subscription.md) * [How to store events](store.md) -* [How to split streams](split_stream.md) * [How to upcast events](upcasting.md) +* [How to use messages](message.md) diff --git a/docs/pages/getting_started.md b/docs/pages/getting_started.md index ff492aea5..c40060684 100644 --- a/docs/pages/getting_started.md +++ b/docs/pages/getting_started.md @@ -295,17 +295,17 @@ $eventStore = new DoctrineDbalStore( $hotelProjector = new HotelProjector($projectionConnection); -$projectorRepository = new MetadataSubscriberAccessorRepository([ +$subscriberRepository = new MetadataSubscriberAccessorRepository([ $hotelProjector, new SendCheckInEmailProcessor($mailer), ]); -$projectionStore = new DoctrineSubscriptionStore($connection); +$subscriptionStore = new DoctrineSubscriptionStore($connection); $engine = new DefaultSubscriptionEngine( $eventStore, - $projectionStore, - $projectorRepository, + $subscriptionStore, + $subscriberRepository, ); $repositoryManager = new DefaultRepositoryManager( @@ -382,7 +382,7 @@ $hotels = $hotelProjection->getHotels(); ``` !!! warning - You need to run the subscription engine to update the projections. + You need to run the subscription engine to update the projections and execute the processors. !!! note @@ -403,6 +403,5 @@ $hotels = $hotelProjection->getHotels(); * [How to create an aggregate](aggregate.md) * [How to create an event](events.md) * [How to store aggregates](repository.md) -* [How to process events](subscription.md) -* [How to create a projection](subscription.md) +* [How to create a projection and processors](subscription.md) * [How to setup the database](store.md) diff --git a/docs/pages/index.md b/docs/pages/index.md index b5789df1b..99a5e8bd5 100644 --- a/docs/pages/index.md +++ b/docs/pages/index.md @@ -1,6 +1,7 @@ # Event-Sourcing -A lightweight but also all-inclusive event sourcing library with a focus on developer experience. +An event sourcing library, complete with all the essential features, +powered by the reliable Doctrine ecosystem and focused on developer experience. ## Features @@ -9,10 +10,10 @@ A lightweight but also all-inclusive event sourcing library with a focus on deve * Developer experience oriented and fully typed * Automatic [snapshot](snapshots.md)-system to boost your performance * [Split](split_stream.md) big aggregates into multiple streams -* Build-in [pipeline](pipeline.md) to export, import and migrate event streams * Versioned and managed lifecycle of [subscriptions](subscription.md) like projections and processors +* Safe usage of [Personal Data](personal_data.md) with crypto-shredding * Smooth [upcasting](upcasting.md) of old events -* Simple setup with [scheme management](store.md) and [doctrine migration](migration.md) +* Simple setup with [scheme management](store.md) and [doctrine migration](store.md) * Built in [cli commands](cli.md) with [symfony](https://symfony.com/) * and much more... diff --git a/docs/pages/message.md b/docs/pages/message.md index 864bc4fdb..398ddb170 100644 --- a/docs/pages/message.md +++ b/docs/pages/message.md @@ -262,9 +262,8 @@ final class SplitProfileCreatedTranslator implements Translator ## Learn more +* [How to decorate messages](message_decorator.md) +* [How to load aggregates](repository.md) +* [How to store messages](store.md) * [How to use subscriptions](subscription.md) -* [How to use the repository](repository.md) * [How to use the event bus](event_bus.md) -* [How to decorate messages](message_decorator.md) -* [How to use the normalizer](normalizer.md) -* [How to use the upcasting](upcasting.md) diff --git a/docs/pages/message_decorator.md b/docs/pages/message_decorator.md index 58c0a9d29..3bb7c7c50 100644 --- a/docs/pages/message_decorator.md +++ b/docs/pages/message_decorator.md @@ -107,7 +107,7 @@ final class OnSystemRecordedDecorator implements MessageDecorator ## Learn more +* [How to create messages](message.md) * [How to define events](events.md) -* [How to use the event bus](event_bus.md) * [How to configure repositories](repository.md) * [How to upcast events](upcasting.md) diff --git a/docs/pages/normalizer.md b/docs/pages/normalizer.md index 6caa1d150..47e655e5c 100644 --- a/docs/pages/normalizer.md +++ b/docs/pages/normalizer.md @@ -37,6 +37,10 @@ final class DTO } } ``` +!!! tip + + If you have personal data, you can use [crypto-shredding](personal_data.md). + ### Event For the event, the properties are normalized to a payload and saved in the DB at the end. @@ -382,6 +386,6 @@ final class DTO * [How to use the Hydrator](https://github.com/patchlevel/hydrator) * [How to define aggregates](aggregate.md) -* [How to snapshot aggregates](snapshots.md) -* [How to create own aggregate id](aggregate_id.md) * [How to define events](events.md) +* [How to snapshot aggregates](snapshots.md) +* [How to work with personal data](personal_data.md) diff --git a/docs/pages/outbox.md b/docs/pages/outbox.md deleted file mode 100644 index 141b612b0..000000000 --- a/docs/pages/outbox.md +++ /dev/null @@ -1,85 +0,0 @@ -# Outbox - -There is the problem that errors can occur when saving an aggregate or in the individual event listeners. -This means that you either saved an aggregate, but an error occurred in the email listener, so that no email went out. -Or that an email was sent but the aggregate could not be saved. - -Both cases are very bad and can only be solved if both the saving of an aggregate -and the dispatching of the events are in a transaction. - -The best way to ensure this is to store the events to be dispatched together -with the aggregate in a transaction in the same database. - -After the transaction becomes successful, the events can be loaded from the outbox table with a worker -and then dispatched into the correct event bus. As soon as the events have been dispatched, -they are deleted from the outbox table. If an error occurs when dispatching, the whole thing will be retrieved later. - -## Configuration - -First you have to replace the correct event bus with an outbox event bus. -This stores the events to be dispatched in the database. - -```php -use Patchlevel\EventSourcing\Outbox\OutboxEventBus; -use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager; - -$eventBus = new OutboxEventBus($store); - -$repositoryManager = new DefaultRepositoryManager( - $aggregateRootRegistry, - $store, - $eventBus, -); -``` -And then you have to define the consumer. This gets the right event bus. -It is used to load the events to be dispatched from the database, dispatch the events and then empty the outbox table. - -```php -use Patchlevel\EventSourcing\EventBus\DefaultConsumer; -use Patchlevel\EventSourcing\Outbox\EventBusPublisher; -use Patchlevel\EventSourcing\Outbox\StoreOutboxProcessor; - -$consumer = DefaultConsumer::create([$mailListener]); - -$processor = new StoreOutboxProcessor( - $store, - new EventBusPublisher($consumer), -); - -$processor->process(); -``` -## Using outbox - -So that this is also executed in a transaction, you have to make sure that a transaction has also been started. - -```php -$store->transactional(static function () use ($command, $profileRepository): void { - $profile = Profile::register( - $command->id(), - $command->email(), - ); - - $profileRepository->save($profile); -}); -``` -!!! note - - You can find out more about transaction [here](store.md#transaction). - -You can also interact directly with the outbox store. - -```php -$store->saveOutboxMessage($message); -$store->markOutboxMessageConsumed($message); - -$store->retrieveOutboxMessages(); -$store->countOutboxMessages(); -``` -!!! note - - Both single table store and multi table store implement the outbox store. - -!!! tip - - Interacting with the outbox store is also possible via the [cli](cli.md). - \ No newline at end of file diff --git a/docs/pages/personal_data.md b/docs/pages/personal_data.md index 62083087f..6d2032ea0 100644 --- a/docs/pages/personal_data.md +++ b/docs/pages/personal_data.md @@ -219,4 +219,10 @@ final class DeletePersonalDataProcessor $this->cipherKeyStore->remove($event->personId); } } -``` \ No newline at end of file +``` +## Learn more + +* [How to use the hydrator](https://github.com/patchlevel/hydrator) +* [How to define aggregates](aggregate.md) +* [How to define events](events.md) +* [How to normalize data](normalizer.md) diff --git a/docs/pages/repository.md b/docs/pages/repository.md index 838c42d5b..beceec9e2 100644 --- a/docs/pages/repository.md +++ b/docs/pages/repository.md @@ -1,7 +1,7 @@ # Repository A `repository` takes care of storing and loading the `aggregates`. -He is also responsible for building [messages](event_bus.md) from the events +He is also responsible for building [messages](message.md) from the events and optionally dispatching them to the event bus. ## Create a repository @@ -63,7 +63,7 @@ $repository = $repositoryManager->get(Profile::class); !!! warning If you use the event bus, you should be aware that the events are dispatched synchronously. - You may encounter "at least once" problems. + You may encounter [at least once](https://softwaremill.com/message-delivery-and-deduplication-strategies/) problems. !!! note @@ -138,6 +138,10 @@ $repository = $repositoryManager->get(Profile::class); You can find out more about message decorator [here](message_decorator.md). +!!! tip + + If you have multiple decorators, you can use the `ChainMessageDecorator` to chain them. + ## Use the repository Each `repository` has three methods that are responsible for loading an `aggregate`, @@ -159,9 +163,17 @@ $profile = Profile::create($id, 'david.badura@patchlevel.de'); /** @var Repository $repository */ $repository->save($profile); ``` -!!! note +!!! Warning All events are written to the database with one transaction in order to ensure data consistency. + If an exception occurs during the save process, + the transaction is rolled back and the aggregate is not valid anymore. + You can not save the aggregate again and you need to load it again. + +!!! note + + Due to the nature of the aggregate having a playhead, + we have a unique constraint that ensures that no race condition happens here. ### Load an aggregate @@ -246,4 +258,13 @@ class ProfileRepository return $this->repository->has($id); } } -``` \ No newline at end of file +``` +## Learn more + +* [How to create an aggregate](aggregate.md) +* [How to create an event](events.md) +* [How to work with the store](store.md) +* [How to use snapshots](snapshots.md) +* [How to split streams](split_stream.md) +* [How to use the event bus](event_bus.md) +* [How to create messages](message.md) diff --git a/docs/pages/snapshots.md b/docs/pages/snapshots.md index ad5853432..5b8d2e65e 100644 --- a/docs/pages/snapshots.md +++ b/docs/pages/snapshots.md @@ -5,12 +5,20 @@ This is not a problem if there are a few hundred. But if the number gets bigger at some point, then loading and rebuilding can become slow. The `snapshot` system can be used to control this. -Normally, the events are all executed again on the aggregate in order to rebuild the current state. +!!! tip + + Use snapshots only if you have a performance problems, + because it introduces additional complexity. + + In our benchmarks we can load 10 000 events for one aggregate in 50ms. + Of course, this can vary from system to system. + +Normally, the events are all applied again on the aggregate in order to rebuild the current state. With a `snapshot`, we can shorten the way in which we temporarily save the current state of the aggregate. When loading it is checked whether the snapshot exists. -If a hit exists, the aggregate is built up with the help of the snapshot. +If a hit exists, the aggregate is created with the help of the snapshot. A check is then made to see whether further events have existed since the snapshot -and these are then also executed on the aggregate. +and these are then also applied on the aggregate. Here, however, only the last events are loaded from the database and not all. ## Configuration @@ -106,13 +114,14 @@ final class Profile extends BasicAggregateRoot !!! note - You can find more about normalizer [here](normalizer.md). + The [hydrator](https://github.com/patchlevel/hydrator) is used internally and you can use all of its features. + You can find more about normalizer also [here](normalizer.md). ### Snapshot batching Since the loading of events in itself is quite fast and only becomes noticeably slower with thousands of events, we do not need to create a snapshot after each event. That would also have a negative impact on performance. -Instead, we can also create a snapshot after `N` events. +Instead, we can also create a snapshot after `n` events. The remaining events that are not in the snapshot are then loaded from store. ```php @@ -132,9 +141,10 @@ final class Profile extends BasicAggregateRoot Whenever something changes on the aggregate, the previous snapshot must be discarded. You can do this by removing the entire snapshot cache when deploying. But that can be quickly forgotten. It is much easier to specify a snapshot version. -This snapshot version is also saved. When loading, the versions are compared and if they do not match, +This snapshot version is also saved in the snapshot cache. +When loading, the versions are compared and if they do not match, the snapshot is discarded and the aggregate is rebuilt from scratch. -The new aggregate is then saved again as a snapshot. +The new snapshot is then created automatically. ```php use Patchlevel\EventSourcing\Aggregate\BasicAggregateRoot; @@ -151,10 +161,14 @@ final class Profile extends BasicAggregateRoot !!! warning If the snapshots are discarded, a load peak can occur since the aggregates have to be rebuilt. + You should update the snapshot version only when necessary. !!! tip - You can also use uuids for the snapshot version. + If you have aggregates with a lot of events, + you should consider using [split streams](split_stream.md) if it make sense in your domain. + Then the load peak is not so high anymore, + because only the events from new stream start are loaded to rebuild the aggregate. ## Adapter @@ -169,7 +183,7 @@ Here are a few listed: * [laminas cache](https://docs.laminas.dev/laminas-cache/) * [scrapbook](https://www.scrapbook.cash/) -### psr6 +### psr-6 A `Psr6SnapshotAdapter`, the associated documentation can be found [here](https://www.php-fig.org/psr/psr-6/). @@ -180,7 +194,7 @@ use Psr\Cache\CacheItemPoolInterface; /** @var CacheItemPoolInterface $cache */ $adapter = new Psr6SnapshotAdapter($cache); ``` -### psr16 +### psr-16 A `Psr16SnapshotAdapter`, the associated documentation can be found [here](https://www.php-fig.org/psr/psr-16/). @@ -245,4 +259,10 @@ And if the version is no longer correct and the snapshot is therefore invalid, t The aggregate may be in an old state as the snapshot may lag behind. You still have to bring the aggregate up to date by loading the missing events from the event store. - \ No newline at end of file + +## Learn more + +* [How to define aggregates](aggregate.md) +* [How to store and load aggregates](repository.md) +* [How to split streams](split_stream.md) +* [How to work with personal data](personal_data.md) diff --git a/docs/pages/split_stream.md b/docs/pages/split_stream.md index 6ec833a90..648ba0a69 100644 --- a/docs/pages/split_stream.md +++ b/docs/pages/split_stream.md @@ -1,34 +1,71 @@ -# Splitting the eventstream +# Split Stream -In some cases the business has rules which implies an restart of the event stream for an aggregate since the past events -are not relevant for the current state. For example a user decides to end his active subscription and the business rules -says if the user start a new subscription all past events should not be considered anymore. Another case could be a -banking scenario. There the business decides to save the current state every quarter for each banking account. +In some cases the business has rules which implies a restart of the event stream for an aggregate +since the past events are not relevant for the current state. +A bank is often used as an example. A bank account has hundreds of transactions, +but every bank makes a balance report at the end of the year. +In this step the current account balance is persisted. +This event is perfect to split the stream and start aggregating from this point. -Not only that some businesses requires such an action it also increases the performance for aggregate which would have a -really long event stream. +Not only that some businesses requires such an action +it also increases the performance for aggregate which would have a really long event stream. -## Flagging an event to split the stream +In the background the library will mark all past events as archived +and will not load them anymore for building the aggregate. +It will only load the events from the split event and onwards. +But subscriptions will still receive all events. +So you can create projections which are based on the full event stream. -To use this feature you need to add the `SplitStreamDecorator`. You will also need events which will trigger this -action. For that you can use the `#[SplitStream]` attribute. We decided that we are not literallty splitting the stream, -instead we are marking all past events as archived as soon as this event is saved. Then the past events will not be -loaded anymore for building the aggregate. This means that all needed data has to be present in these events which -should trigger the event split. +## Configuration + +To use this feature you need to add the `SplitStreamDecorator` in the repository manager. + +```php +use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootRegistry; +use Patchlevel\EventSourcing\Metadata\Event\EventMetadataFactory; +use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager; +use Patchlevel\EventSourcing\Repository\MessageDecorator\SplitStreamDecorator; +use Patchlevel\EventSourcing\Store\Store; + +/** + * @var AggregateRootRegistry $aggregateRootRegistry + * @var Store $store + * @var EventMetadataFactory $eventMetadataFactory + */ +$repositoryManager = new DefaultRepositoryManager( + $aggregateRootRegistry, + $store, + null, + null, + new SplitStreamDecorator($eventMetadataFactory), +); +``` +!!! note + + You can find out more about decorator [here](./message_decorator.md). + +!!! tip + + You can use multiple decorators with the `ChainMessageDecorator`. + +## Usage + +To use this feature you need to mark the event which should split the stream. +For that you can use the `#[SplitStream]` attribute. ```php use Patchlevel\EventSourcing\Attribute\Event; use Patchlevel\EventSourcing\Attribute\SplitStream; use Patchlevel\EventSourcing\Serializer\Normalizer\IdNormalizer; -#[Event('bank_account.month_passed')] +#[Event('bank_account.balance_reported')] #[SplitStream] -final class MonthPassed +final class BalanceReported { public function __construct( #[IdNormalizer] - public AccountId $accountId, - public string $name, + public BankAccountId $bankAccountId, + public int $year, public int $balanceInCents, ) { } @@ -36,10 +73,21 @@ final class MonthPassed ``` !!! warning - The event needs all data which is relevant the aggregate to be used since all past event will not be loaded! Keep - this in mind if you want to use this feature. + The event needs all data which is relevant the aggregate to be used since all past event will not be loaded! + Keep this in mind if you want to use this feature. !!! note - This archive flag only impacts the Store::load method which is used the build the aggregate from the stream. - \ No newline at end of file + This impacts only the aggregate loaded by the repository. Subscriptions will still receive all events. + +!!! tip + + You can combine this feature with the snapshot feature to increase the performance even more. + +## Learn more + +* [How to use message decorator](message_decorator.md) +* [How to define events](events.md) +* [How to define aggregates](aggregate.md) +* [How to store and load aggregates](repository.md) +* [How to use snapshots](snapshots.md) diff --git a/docs/pages/store.md b/docs/pages/store.md index ad38a072f..b6cfd9f68 100644 --- a/docs/pages/store.md +++ b/docs/pages/store.md @@ -343,4 +343,12 @@ $store->transactional(static function () use ($command, $bankAccountRepository): $bankAccountRepository->save($accountFrom); $bankAccountRepository->save($accountTo); }); -``` \ No newline at end of file +``` +## Learn more + +* [How to create events](events.md) +* [How to use repositories](repository.md) +* [How to create message](message.md) +* [How to create projections](subscription.md) +* [How to upcast events](upcasting.md) +* [How configure cli commands](cli.md) diff --git a/docs/pages/subscription.md b/docs/pages/subscription.md index 9e57c9ba3..0f64fb034 100644 --- a/docs/pages/subscription.md +++ b/docs/pages/subscription.md @@ -803,6 +803,5 @@ foreach ($subscriptions as $subscription) { ## Learn more * [How to use CLI commands](./cli.md) -* [How to use Pipeline](./pipeline.md) -* [How to use Event Bus](./event_bus.md) +* [How to create Messages](./message.md) * [How to Test](./testing.md) diff --git a/docs/pages/upcasting.md b/docs/pages/upcasting.md index edd55401c..877bc2211 100644 --- a/docs/pages/upcasting.md +++ b/docs/pages/upcasting.md @@ -1,15 +1,17 @@ # Upcasting -There are cases where the already have events in our stream but there is data missing or not in the right format for our -new usecase. Normally you would need to create versioned events for this. This can lead to many versions of the same -event which could lead to some chaos. To prevent this we offer `Upcaster`, which can operate on the payload before -denormalizing to an event object. There you can change the event name and adjust the payload of the event. +There are cases where the already have events in our stream but there is data missing +or not in the right format for our new usecase. Normally you would need to create versioned events for this. +This can lead to many versions of the same event which could lead to some chaos. +To prevent this we offer `Upcaster`, which can operate on the payload before denormalizing to an event object. +There you can change the event name and adjust the payload of the event. ## Adjust payload -Let's assume we have an `ProfileCreated` event which holds an email. Now the business needs to have all emails to be in -lower case. For that we could adjust the aggregate and the projections to take care of that. Or we can do this -beforehand so we don't need to maintain two different places. +Let's assume we have an `ProfileCreated` event which holds an email. +Now the business needs to have all emails to be in lower case. +For that we could adjust the aggregate and the projections to take care of that. +Or we can do this beforehand so we don't need to maintain two different places. ```php use Patchlevel\EventSourcing\Serializer\Upcast\Upcast; @@ -20,7 +22,7 @@ final class ProfileCreatedEmailLowerCastUpcaster implements Upcaster public function __invoke(Upcast $upcast): Upcast { // ignore if other event is processed - if ($upcast->eventName !== 'profile_created') { + if ($upcast->eventName !== 'profile.created') { return $upcast; } @@ -38,15 +40,14 @@ final class ProfileCreatedEmailLowerCastUpcaster implements Upcaster ## Adjust event name -For the upgrade to 2.0.0 this feature is also really handy since we adjusted the event value from FQCN to an unique -name which the user needs to choose. This opens up for moving or renaming the events at code level. Here an example for -the upgrade path. +Sometimes your event name was not the best choice and you want to change it. +For this we can use the `Upcaster` to change the event name. ```php use Patchlevel\EventSourcing\Serializer\Upcast\Upcast; use Patchlevel\EventSourcing\Serializer\Upcast\Upcaster; -final class LegacyEventNameUpaster implements Upcaster +final class EventNameRenameUpcaster implements Upcaster { /** @param array $eventNameMap */ public function __construct( @@ -64,7 +65,7 @@ final class LegacyEventNameUpaster implements Upcaster } } ``` -## Use upcasting +## Configure After we have defined the upcasting rules, we also have to pass the whole thing to the serializer. Since we have multiple upcasters, we use a chain here. @@ -77,7 +78,7 @@ use Patchlevel\EventSourcing\Serializer\Upcast\UpcasterChain; /** @var EventRegistry $eventRegistry */ $upcaster = new UpcasterChain([ new ProfileCreatedEmailLowerCastUpcaster(), - new LegacyEventNameUpaster(['old_event_name' => 'new_event_name']), + new EventNameRenameUpcaster(['old_event_name' => 'new_event_name']), ]); $serializer = DefaultEventSerializer::createFromPaths( @@ -85,54 +86,8 @@ $serializer = DefaultEventSerializer::createFromPaths( $upcaster, ); ``` -## Update event stream +## Learn more -But what if we need it also in our stream because some other applications has also access on it? Or want to cleanup our -Upcasters since we have collected alot of them over the time? Then we can use our pipeline feature without any -middlewares to achive a complete rebuild of our stream with adjusted event data. - -```php -use Patchlevel\EventSourcing\Pipeline\Pipeline; -use Patchlevel\EventSourcing\Pipeline\Source\StoreSource; -use Patchlevel\EventSourcing\Pipeline\Target\StoreTarget; -use Patchlevel\EventSourcing\Store\Store; -use Symfony\Component\Console\Attribute\AsCommand; -use Symfony\Component\Console\Command\Command; -use Symfony\Component\Console\Input\InputInterface; -use Symfony\Component\Console\Output\OutputInterface; - -#[AsCommand( - name: 'event-stream:cleanup', - description: 'rebuild event stream', -)] -final class EventStreamCleanupCommand extends Command -{ - public function __construct( - private readonly Store $sourceStore, - private readonly Store $targetStore, - ) { - parent::__construct(); - } - - protected function execute(InputInterface $input, OutputInterface $output): int - { - $pipeline = new Pipeline( - new StoreSource($this->sourceStore), - new StoreTarget($this->targetStore), - ); - - $pipeline->run(); - - return Command::SUCCESS; - } -} -``` -!!! danger - - Under no circumstances may the same store be used that is used for the source. - Otherwise the store will be broken afterwards! - -!!! note - - You can find out more about the pipeline [here](pipeline.md). - \ No newline at end of file +* [How to create messages](message.md) +* [How to define events](events.md) +* [How to configure store](store.md)