diff --git a/.gitattributes b/.gitattributes index d2f9d2cb0..7bdfb803e 100644 --- a/.gitattributes +++ b/.gitattributes @@ -5,7 +5,6 @@ /.gitignore export-ignore /composer.lock export-ignore /demo/ export-ignore -/docs/ export-ignore /phpbench.json export-ignore /phpcs.xml.dist export-ignore /phpstan.neon export-ignore diff --git a/.github/workflows/backward-compatibility-check.yml b/.github/workflows/backward-compatibility-check.yml index 0a98d7837..ebfc8ab25 100644 --- a/.github/workflows/backward-compatibility-check.yml +++ b/.github/workflows/backward-compatibility-check.yml @@ -29,7 +29,6 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_sqlite, bcmath, intl, sodium diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 23f877cf5..afb84d9ef 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -14,7 +14,7 @@ jobs: services: postgres: # Docker Hub image - image: "postgres:18.4" + image: "postgres:18.6" # Provide the password for postgres env: POSTGRES_PASSWORD: postgres diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index eb84cdcae..6759acac9 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -53,7 +53,6 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_pgsql @@ -108,7 +107,6 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_mysql @@ -162,7 +160,6 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_mysql @@ -196,7 +193,6 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_sqlite diff --git a/.github/workflows/mutation-tests-diff.yml b/.github/workflows/mutation-tests-diff.yml index c38555ac7..98555ab71 100644 --- a/.github/workflows/mutation-tests-diff.yml +++ b/.github/workflows/mutation-tests-diff.yml @@ -29,7 +29,7 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" + coverage: "xdebug" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_sqlite, pdo_pgsql diff --git a/.github/workflows/mutation-tests.yml b/.github/workflows/mutation-tests.yml index e4ee6a896..9619c5ff7 100644 --- a/.github/workflows/mutation-tests.yml +++ b/.github/workflows/mutation-tests.yml @@ -31,7 +31,7 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" + coverage: "xdebug" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_sqlite, pdo_pgsql diff --git a/.github/workflows/unit.yml b/.github/workflows/unit.yml index 749c003b7..269c23682 100644 --- a/.github/workflows/unit.yml +++ b/.github/workflows/unit.yml @@ -43,7 +43,7 @@ jobs: - name: "Install PHP" uses: "shivammathur/setup-php@2.37.2" with: - coverage: "pcov" + coverage: "xdebug" php-version: "${{ matrix.php-version }}" ini-values: memory_limit=-1 extensions: pdo_sqlite, pdo_pgsql diff --git a/composer.lock b/composer.lock index 871ffa04d..14e8b126c 100644 --- a/composer.lock +++ b/composer.lock @@ -3800,16 +3800,16 @@ }, { "name": "league/commonmark", - "version": "2.8.2", + "version": "2.9.0", "source": { "type": "git", "url": "https://github.com/thephpleague/commonmark.git", - "reference": "59fb075d2101740c337c7216e3f32b36c204218b" + "reference": "5703d83ba3da3b2e356a5fedc848ed6d8ffb6529" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/thephpleague/commonmark/zipball/59fb075d2101740c337c7216e3f32b36c204218b", - "reference": "59fb075d2101740c337c7216e3f32b36c204218b", + "url": "https://api.github.com/repos/thephpleague/commonmark/zipball/5703d83ba3da3b2e356a5fedc848ed6d8ffb6529", + "reference": "5703d83ba3da3b2e356a5fedc848ed6d8ffb6529", "shasum": "" }, "require": { @@ -3831,8 +3831,8 @@ "github/gfm": "0.29.0", "michelf/php-markdown": "^1.4 || ^2.0", "nyholm/psr7": "^1.5", - "phpstan/phpstan": "^1.8.2", - "phpunit/phpunit": "^9.5.21 || ^10.5.9 || ^11.0.0", + "phpstan/phpstan": "^2.0.0", + "phpunit/phpunit": "^9.5.21 || ^10.5.9 || ^11.0.0 || ^12.0.0 || ^13.0.0", "scrutinizer/ocular": "^1.8.1", "symfony/finder": "^5.3 | ^6.0 | ^7.0 || ^8.0", "symfony/process": "^5.4 | ^6.0 | ^7.0 || ^8.0", @@ -3846,7 +3846,7 @@ "type": "library", "extra": { "branch-alias": { - "dev-main": "2.9-dev" + "dev-main": "2.10-dev" } }, "autoload": { @@ -3903,7 +3903,7 @@ "type": "tidelift" } ], - "time": "2026-03-19T13:16:38+00:00" + "time": "2026-08-03T13:42:31+00:00" }, { "name": "league/config", diff --git a/docs/aggregate.md b/docs/aggregate.md index 83e2e501d..9014ec5d1 100644 --- a/docs/aggregate.md +++ b/docs/aggregate.md @@ -846,6 +846,11 @@ use Patchlevel\EventSourcing\Metadata\AggregateRoot\AttributeAggregateRootRegist $aggregateRegistry = (new AttributeAggregateRootRegistryFactory())->create([/* paths... */]); ``` +:::tip +Scanning the paths on every request costs time. In production you can wrap the factory +in a [metadata cache](metadata-cache.md). +::: + ## Learn more * [How to create own aggregate id](identifier.md) @@ -853,3 +858,4 @@ $aggregateRegistry = (new AttributeAggregateRootRegistryFactory())->create([/* p * [How to snapshot aggregates](snapshots.md) * [How to create Projections](subscription.md) * [How to split streams](split-stream.md) +* [How to cache metadata](metadata-cache.md) diff --git a/docs/cli.md b/docs/cli.md index 0607e678d..277a73802 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -7,6 +7,8 @@ You can: * Create and delete `databases` * Create, update and delete `schemas` * Manage `subscriptions` +* Inspect your `aggregates`, `events` and `subscribers` +* Migrate events from one `store` to another ## Database commands @@ -53,6 +55,88 @@ The inspector is a tool to inspect the event streams. * ShowAggregateCommand: `event-sourcing:show-aggregate` * WatchCommand: `event-sourcing:watch` +## Debug command + +The debug command prints everything the library knows about your application: +all registered aggregates, all registered events and all subscribers with their subscribe methods. +It is the fastest way to check if a class was picked up by the attribute scanning. + +* DebugCommand: `event-sourcing:debug` (alias `debug:event-sourcing`) + +```php +use Patchlevel\EventSourcing\Console\Command\DebugCommand; +use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootRegistry; +use Patchlevel\EventSourcing\Metadata\Event\EventRegistry; +use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository; +use Symfony\Component\Console\Application; + +/** + * @var Application $cli + * @var AggregateRootRegistry $aggregateRootRegistry + * @var EventRegistry $eventRegistry + * @var SubscriberAccessorRepository $subscriberAccessorRepository + */ +$cli->add( + new DebugCommand( + $aggregateRootRegistry, + $eventRegistry, + $subscriberAccessorRepository, + ), +); +``` +:::note +The subscriber repository is optional. If you don't pass it, the subscriber section is skipped. +::: + +## Store migration command + +The store migration command copies all events from one store into another one. +You need it when you switch the store implementation, +for example from the [StreamDoctrineDbalStore](store.md#streamdoctrinedbalstore) +to the [TaggableDoctrineDbalStore](store.md#taggabledoctrinedbalstore). + +* StoreMigrateCommand: `event-sourcing:store:migrate` + +```php +use Patchlevel\EventSourcing\Console\Command\StoreMigrateCommand; +use Patchlevel\EventSourcing\Message\Translator\ExtractEventTagTranslator; +use Patchlevel\EventSourcing\Store\Store; +use Symfony\Component\Console\Application; + +/** + * @var Application $cli + * @var Store $oldStore + * @var Store $newStore + */ +$cli->add( + new StoreMigrateCommand( + $oldStore, + $newStore, + [new ExtractEventTagTranslator()], + ), +); +``` +The third constructor argument is a list of [translators](message.md#translator) +that are applied to every message before it is written into the new store. +The `ExtractEventTagTranslator` reads the tags from the events and adds them as `TagsHeader`, +which is what you need for a migration to the `TaggableDoctrineDbalStore`. +If both stores work with the same headers, you can leave the list empty. + +Events are written in batches. You can control the batch size with the `buffer` option: + +```bash +bin/console event-sourcing:store:migrate --buffer=5000 +``` +:::danger +The command writes into the target store, it does not clean it up first. +Make sure the target store is empty and create a backup before you run the migration. +::: + +:::note +The schema of the new store has to exist before you run the command. +You can create it with the [schema commands](#schema-commands). +::: + ## CLI example A cli php file can look like this: diff --git a/docs/dynamic-consistency-boundary.md b/docs/dynamic-consistency-boundary.md index 7b750f3bd..0605d0695 100644 --- a/docs/dynamic-consistency-boundary.md +++ b/docs/dynamic-consistency-boundary.md @@ -46,7 +46,6 @@ final class HotelCreated } } ``` - :::note A tagged property that is `null` is skipped, so no tag is generated for it. This lets you tag optional properties without having to guard against missing values. ::: @@ -423,7 +422,6 @@ $provider = new ServiceHandlerProvider([ $commandBus = new SyncCommandBus($provider); ``` - :::tip For tests you can replace the `TaggableDoctrineDbalStore` with an `InMemoryStore`. It implements the same tag query and conditional append, so the decision model and the append condition behave the same diff --git a/docs/events.md b/docs/events.md index dd3d50bef..ef1fae60e 100644 --- a/docs/events.md +++ b/docs/events.md @@ -40,7 +40,7 @@ Here are some examples: * `profile.created` * `profile.name_changed` * `hotel.guest_checked_out` - + ::: ## Alias @@ -83,6 +83,68 @@ The serializer needs the path information where the event classes are located so that it can instantiate the correct classes. Internally, an EventRegistry is used, which is described in the [Event Registry](#event-registry) section below. +## Encoder + +The serializer turns an event into an array first and then encodes that array into a string. +The encoding is done by an `Encoder`. By default the `JsonEncoder` is used, which encodes the payload as JSON. + +If you want to change how the payload is encoded, you can pass your own encoder to the serializer. + +```php +use Patchlevel\EventSourcing\Metadata\Event\AttributeEventRegistryFactory; +use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer; +use Patchlevel\EventSourcing\Serializer\Encoder\JsonEncoder; + +$serializer = new DefaultEventSerializer( + (new AttributeEventRegistryFactory())->create(['src/Domain']), + encoder: new JsonEncoder(), +); +``` +The `Encoder` interface has two methods: + +```php +use Patchlevel\EventSourcing\Serializer\Encoder\Encoder; + +final class MyEncoder implements Encoder +{ + /** + * @param array $data + * @param array $options + */ + public function encode(array $data, array $options = []): string + { + // your encoding + } + + /** + * @param array $options + * + * @return array + */ + public function decode(string $data, array $options = []): array + { + // your decoding + } +} +``` +An `encode` call that fails must throw an `EncodeNotPossible` exception, +a failing `decode` call a `DecodeNotPossible` exception. + +The options are passed through from the serializer. The `JsonEncoder` understands +`Encoder::OPTION_PRETTY_PRINT`, which is used by the [cli](cli.md) to print readable payloads. + +```php +use Patchlevel\EventSourcing\Serializer\Encoder\Encoder; +use Patchlevel\EventSourcing\Serializer\EventSerializer; + +/** @var EventSerializer $serializer */ +$data = $serializer->serialize($event, [Encoder::OPTION_PRETTY_PRINT => true]); +``` +:::warning +The encoder decides the format of everything that is already in your store. +If you change it, old events can no longer be decoded. +::: + ## Normalizer Sometimes you also want to add more complex data as a payload. For example DateTime or value objects. @@ -141,6 +203,11 @@ use Patchlevel\EventSourcing\Metadata\Event\AttributeEventRegistryFactory; $eventRegistry = (new AttributeEventRegistryFactory())->create([/* paths... */]); ``` +:::tip +Scanning the paths on every request costs time. In production you can wrap the factory +in a [metadata cache](metadata-cache.md). +::: + ## Learn more * [How to normalize events](normalizer.md) @@ -148,3 +215,4 @@ $eventRegistry = (new AttributeEventRegistryFactory())->create([/* paths... */]) * [How to store events](store.md) * [How to upcast events](upcasting.md) * [How to use messages](message.md) +* [How to cache metadata](metadata-cache.md) diff --git a/docs/message.md b/docs/message.md index efe05643e..54b631cd2 100644 --- a/docs/message.md +++ b/docs/message.md @@ -59,6 +59,27 @@ The message object has some built-in headers which are used internally. * `ArchivedHeader` - Flag if the message is archived. * `StreamStartHeader` - Flag if the message is the first message in a new stream. +```php +use Patchlevel\EventSourcing\Message\Message; +use Patchlevel\EventSourcing\Store\Header\EventIdHeader; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; +use Patchlevel\EventSourcing\Store\Header\PlayheadHeader; +use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader; +use Patchlevel\EventSourcing\Store\Header\StreamNameHeader; + +/** @var Message $message */ +$message->header(IndexHeader::class)->index; // 42 +$message->header(StreamNameHeader::class)->streamName; // 'profile-e3e3e3e3-...' +$message->header(PlayheadHeader::class)->playhead; // 2 +$message->header(RecordedOnHeader::class)->recordedOn; // DateTimeImmutable +$message->header(EventIdHeader::class)->eventId; // 'a4a4a4a4-4a4a-...' +``` +:::warning +The `PlayheadHeader` is only added if the stream is playhead based. +Streams that are written without a playhead, for example custom streams, +do not have this header. Use `hasHeader` before you access it. +::: + ## Custom headers You can also add custom headers to the message object. For example, you can add an application id. diff --git a/docs/metadata-cache.md b/docs/metadata-cache.md new file mode 100644 index 000000000..cd77bcbff --- /dev/null +++ b/docs/metadata-cache.md @@ -0,0 +1,185 @@ +# Metadata Cache + +The library reads a lot of information from attributes: which classes are aggregates and events, +which apply method belongs to which event, which subscriber listens to what. +This information is called metadata, and collecting it means scanning directories +and reflecting over classes. + +In development that is exactly what you want, because every change is picked up immediately. +In production the classes never change while the process runs, so the same work is repeated on every request. +For this the library ships decorators that cache the metadata in a +[PSR-6](https://www.php-fig.org/psr/psr-6/) or [PSR-16](https://www.php-fig.org/psr/psr-16/) cache. + +:::note +The library only provides the decorators, not the cache implementation itself. +You can use [symfony cache](https://symfony.com/doc/current/components/cache.html) or any other +PSR-6 or PSR-16 compatible cache. +::: + +## Registry cache + +Registries are the hashmaps between names and classes. +Building them means scanning all configured paths, which is the most expensive part of the metadata handling. + +The `AggregateRootRegistryFactory` and the `EventRegistryFactory` both have a caching decorator. +You wrap the attribute based factory and pass a cache to it. + +```php +use Patchlevel\EventSourcing\Metadata\AggregateRoot\AttributeAggregateRootRegistryFactory; +use Patchlevel\EventSourcing\Metadata\AggregateRoot\Psr6AggregateRootRegistryFactory; +use Psr\Cache\CacheItemPoolInterface; + +/** @var CacheItemPoolInterface $cache */ +$aggregateRegistryFactory = new Psr6AggregateRootRegistryFactory( + new AttributeAggregateRootRegistryFactory(), + $cache, +); + +$aggregateRegistry = $aggregateRegistryFactory->create(['src/Domain']); +``` +```php +use Patchlevel\EventSourcing\Metadata\Event\AttributeEventRegistryFactory; +use Patchlevel\EventSourcing\Metadata\Event\Psr6EventRegistryFactory; +use Psr\Cache\CacheItemPoolInterface; + +/** @var CacheItemPoolInterface $cache */ +$eventRegistryFactory = new Psr6EventRegistryFactory( + new AttributeEventRegistryFactory(), + $cache, +); + +$eventRegistry = $eventRegistryFactory->create(['src/Domain']); +``` +If you use a PSR-16 cache, take the `Psr16` variants instead. They work the same way. + +```php +use Patchlevel\EventSourcing\Metadata\Event\AttributeEventRegistryFactory; +use Patchlevel\EventSourcing\Metadata\Event\Psr16EventRegistryFactory; +use Psr\SimpleCache\CacheInterface; + +/** @var CacheInterface $cache */ +$eventRegistryFactory = new Psr16EventRegistryFactory( + new AttributeEventRegistryFactory(), + $cache, +); +``` +:::warning +The registry factories cache under a fixed key, `aggregate_root_registry` and `event_registry`. +The paths you pass to `create` are not part of that key. +If you call the same factory with different paths, you get the result of the first call back. +Use a separate factory instance with its own cache for each set of paths. +::: + +## Metadata cache + +Next to the registries, there is the metadata of a single class. +These factories are called with a class name and return the metadata for it, +so they cache per class name. + +```php +use Patchlevel\EventSourcing\Metadata\AggregateRoot\AttributeAggregateRootMetadataFactory; +use Patchlevel\EventSourcing\Metadata\AggregateRoot\Psr6AggregateRootMetadataFactory; +use Psr\Cache\CacheItemPoolInterface; + +/** @var CacheItemPoolInterface $cache */ +$aggregateMetadataFactory = new Psr6AggregateRootMetadataFactory( + new AttributeAggregateRootMetadataFactory(), + $cache, +); +``` +```php +use Patchlevel\EventSourcing\Metadata\Event\AttributeEventMetadataFactory; +use Patchlevel\EventSourcing\Metadata\Event\Psr6EventMetadataFactory; +use Psr\Cache\CacheItemPoolInterface; + +/** @var CacheItemPoolInterface $cache */ +$eventMetadataFactory = new Psr6EventMetadataFactory( + new AttributeEventMetadataFactory(), + $cache, +); +``` +```php +use Patchlevel\EventSourcing\Metadata\Subscriber\AttributeSubscriberMetadataFactory; +use Patchlevel\EventSourcing\Metadata\Subscriber\Psr6SubscriberMetadataFactory; +use Psr\Cache\CacheItemPoolInterface; + +/** @var CacheItemPoolInterface $cache */ +$subscriberMetadataFactory = new Psr6SubscriberMetadataFactory( + new AttributeSubscriberMetadataFactory(), + $cache, +); +``` +For all three there is a `Psr16` variant that takes a PSR-16 cache instead. + +:::note +The attribute based metadata factories already keep an in memory cache for the current process. +The PSR decorators add a cache that survives the process. +::: + +## Usage + +The cached factories are drop-in replacements, so you pass them wherever the library asks +for a factory or a registry. + +The registries go into the [repository manager](repository.md) and the serializer, +the aggregate metadata factory is the seventh argument of the `DefaultRepositoryManager`. + +```php +use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager; +use Patchlevel\EventSourcing\Repository\MessageDecorator\SplitStreamDecorator; +use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer; +use Patchlevel\EventSourcing\Store\Store; + +/** @var Store $store */ +$repositoryManager = new DefaultRepositoryManager( + $aggregateRegistryFactory->create(['src/Domain']), + $store, + null, + null, + new SplitStreamDecorator($eventMetadataFactory), + null, + $aggregateMetadataFactory, +); + +$serializer = new DefaultEventSerializer( + $eventRegistryFactory->create(['src/Domain']), +); +``` +The subscriber metadata factory goes into the `MetadataSubscriberAccessorRepository`. + +```php +use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository; + +$subscriberRepository = new MetadataSubscriberAccessorRepository( + [/* subscribers... */], + $subscriberMetadataFactory, +); +``` +:::note +Without an explicit factory, the `DefaultRepositoryManager` asks the aggregate class itself +for its metadata. Pass the cached factory as shown above if you want the repository manager to use it. +::: + +## Deployment + +The cache is keyed by class name and by a fixed registry key, not by the content of your classes. +Nothing invalidates it when you change an attribute. + +:::danger +You have to clear the cache on every deployment. +A stale registry makes the library load the wrong class for an event name, +a stale metadata entry makes it call the wrong apply method. +::: + +:::tip +Use a separate cache pool for the metadata and clear it as part of your deployment, +in the same step where you warm up your other caches. +Do not enable the cache in development, otherwise new events and subscribers are not picked up. +::: + +## Learn more + +* [How to register aggregates](aggregate.md) +* [How to register events](events.md) +* [How to create subscribers](subscription.md) +* [How to store events](store.md) diff --git a/docs/project.json b/docs/project.json index dc95ee61d..447e91cb3 100644 --- a/docs/project.json +++ b/docs/project.json @@ -88,6 +88,10 @@ "title": "Time / Clock", "file": "clock.md" }, + { + "title": "Metadata Cache", + "file": "metadata-cache.md" + }, { "title": "Testing", "file": "testing.md" diff --git a/docs/repository.md b/docs/repository.md index 70b1d37d2..0e58b0eee 100644 --- a/docs/repository.md +++ b/docs/repository.md @@ -176,7 +176,7 @@ An `AggregateOutdated` exception is thrown if a conflict occurs. ::: :::tip -If you use the Command Bus, you can use the [instant retry](command-bus.md#instant-retry) decorator +If you use the Command Bus, you can use the [InstantRetryCommandBus](command-bus.md#instant-retry) decorator to retry the command automatically when an `AggregateOutdated` exception occurs. ::: diff --git a/docs/store.md b/docs/store.md index 35212a286..859113410 100644 --- a/docs/store.md +++ b/docs/store.md @@ -335,15 +335,26 @@ use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion; use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion; use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion; use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion; +use Patchlevel\EventSourcing\Store\Criteria\ToIndexCriterion; $criteria = new Criteria( new StreamCriterion('profile-e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e'), new FromPlayheadCriterion(2), new FromIndexCriterion(100), + new ToIndexCriterion(200), new ArchivedCriterion(true), new EventsCriterion(['profile.created', 'profile.name_changed']), ); ``` +The `StreamCriterion` is variadic, so you can pass multiple stream names. +There is also a `startWith` named constructor that appends the wildcard for you. + +```php +use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion; + +$criterion = new StreamCriterion('profile-*', 'hotel-*'); +$criterion = StreamCriterion::startWith('profile-'); +``` Or you can the criteria builder to create the criteria. ```php @@ -447,7 +458,7 @@ In event sourcing, the events are immutable. ### Remove -You can remove streams with the `remove` method by passing a criteria. +You can remove events with the `remove` method. It takes the same criteria as the `load` method. ```php use Patchlevel\EventSourcing\Store\Criteria\Criteria; @@ -457,6 +468,51 @@ use Patchlevel\EventSourcing\Store\Store; /** @var Store $store */ $store->remove(new Criteria(StreamCriterion::startWith('profile-'))); ``` +:::danger +Without criteria the method removes every event in the store. +Deleted events cannot be restored, all subscriptions built from them become inconsistent. +::: + +### Archive + +You can archive events with the `archive` method. +Archived events are still in the store, but they are skipped when an aggregate is loaded, +which keeps the loading of long living aggregates fast. + +```php +use Patchlevel\EventSourcing\Store\Criteria\Criteria; +use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion; +use Patchlevel\EventSourcing\Store\Criteria\ToPlayheadCriterion; +use Patchlevel\EventSourcing\Store\Store; + +/** @var Store $store */ +$store->archive( + new Criteria( + new StreamCriterion('profile-e3e3e3e3-3e3e-3e3e-3e3e-3e3e3e3e3e3e'), + new ToPlayheadCriterion(100), + ), +); +``` +Archived events get the `ArchivedHeader` when they are loaded again. +You can include or exclude them explicitly with the `ArchivedCriterion`. + +```php +use Patchlevel\EventSourcing\Store\Criteria\ArchivedCriterion; +use Patchlevel\EventSourcing\Store\Criteria\Criteria; +use Patchlevel\EventSourcing\Store\Store; + +/** @var Store $store */ +$stream = $store->load(new Criteria(new ArchivedCriterion(false))); +``` +:::note +This method is used when a [split stream](split-stream.md) event is saved. +::: + +:::tip +Archiving is the non destructive alternative to `remove`. The events stay readable, +so you can still replay them by passing an `ArchivedCriterion(true)`. +::: + ### List Streams You can list all streams with the `streams` method. diff --git a/docs/subscription.md b/docs/subscription.md index 725e0960b..d2388bf9e 100644 --- a/docs/subscription.md +++ b/docs/subscription.md @@ -241,7 +241,7 @@ final class PublicProfileProjection public function onPublished(Lookup $lookup): void { $messages = $lookup - ->currentAggregate() // or ->currentStream() for StreamStore + ->currentStream() ->events( ProfileCreated::class, ProfileNameChanged::class, @@ -1139,6 +1139,58 @@ use Patchlevel\EventSourcing\Subscription\RetryStrategy\NoRetryStrategy; $retryStrategy = new NoRetryStrategy(); ``` +#### Custom Retry Strategy + +You can write your own strategy by implementing the `RetryStrategy` interface. +The `shouldRetry` method is asked on every run whether the errored subscription should be picked up again. + +```php +use Patchlevel\EventSourcing\Subscription\RetryStrategy\RetryStrategy; +use Patchlevel\EventSourcing\Subscription\Subscription; + +final class AlwaysRetryStrategy implements RetryStrategy +{ + public function shouldRetry(Subscription $subscription): bool + { + return true; + } +} +``` +A plain `RetryStrategy` can never mark a subscription as failed. +The subscription stays in the error status and is offered to `shouldRetry` again on every run. +To give up at some point, implement `ConditionalRetryStrategy` instead. +It adds a `canRetry` method that answers whether a retry is possible at all. +As soon as `canRetry` returns `false`, the subscription is set to failed and is skipped in all future runs. + +```php +use Patchlevel\EventSourcing\Subscription\RetryStrategy\ConditionalRetryStrategy; +use Patchlevel\EventSourcing\Subscription\Subscription; + +final class TenAttemptsRetryStrategy implements ConditionalRetryStrategy +{ + public function canRetry(Subscription $subscription): bool + { + return $subscription->retryAttempt() < 10; + } + + public function shouldRetry(Subscription $subscription): bool + { + return $this->canRetry($subscription); + } +} +``` +:::note +Both built-in strategies implement `ConditionalRetryStrategy`. +The `ClockBasedRetryStrategy` returns `false` in `canRetry` once `maxAttempts` is reached, +the `NoRetryStrategy` always returns `false`. +::: + +:::warning +`shouldRetry` is called on every run of the subscription engine, so keep it cheap. +Use `canRetry` for the "is there any point in trying again" decision and `shouldRetry` +for the "is it time to try again" decision. +::: + #### Retry Strategy Repository You can define multiple retry strategies and select them by name in the subscriber. diff --git a/src/CommandBus/CommandBus.php b/src/CommandBus/CommandBus.php index f47ee9c28..c4c3323d5 100644 --- a/src/CommandBus/CommandBus.php +++ b/src/CommandBus/CommandBus.php @@ -4,8 +4,13 @@ namespace Patchlevel\EventSourcing\CommandBus; +use Throwable; + interface CommandBus { - /** @throws HandlerNotFound */ + /** + * @throws HandlerNotFound + * @throws Throwable + */ public function dispatch(object $command): void; }