Add pipeline docs

README.md

@@ -4,7 +4,21 @@
 
 # Event-Sourcing-Bundle
 
-a symfony integration of a small lightweight [event-sourcing](https://github.com/patchlevel/event-sourcing) library.
+A lightweight but also all-inclusive event sourcing bundle 
+with a focus on developer experience and based on doctrine dbal.
+This bundle is a [symfony](https://symfony.com/) integration 
+for [event-sourcing](https://github.com/patchlevel/event-sourcing) library.
+
+## Features
+
+* Everything is included in the package for event sourcing
+* Based on [doctrine dbal](https://github.com/doctrine/dbal) and their ecosystem
+* Developer experience oriented and fully typed
+* [Snapshots](docs/snapshots.md) system to quickly rebuild the aggregates
+* [Pipeline](docs/pipeline.md) to build new [projections](docs/projection.md) or to migrate events
+* [Scheme management](docs/store.md) and [doctrine migration](docs/store.md) support
+* Dev [tools](docs/tools.md) such as a realtime event watcher
+* Built in [cli commands](docs/cli.md) with [symfony](https://symfony.com/)
 
 ## Installation
 
@@ -28,6 +42,7 @@ as this documentation only deals with bundle integration.
 * [Store](docs/store.md)
 * [Pipeline](docs/pipeline.md)
 * [Tools](docs/tools.md)
+* [CLI](docs/cli.md)
 
 ## Integration
 
@@ -120,8 +135,9 @@ final class GuestIsCheckedOut extends AggregateChanged
 
 ### Define aggregates
 
-Next we need to define the aggregate. So the hotel and how the hotel should behave. We have also defined the `create`
-, `checkIn` and `checkOut` methods accordingly. These events are thrown here and the state of the hotel is also changed.
+Next we need to define the aggregate. So the hotel and how the hotel should behave. 
+We have also defined the `create`, `checkIn` and `checkOut` methods accordingly. 
+These events are thrown here and the state of the hotel is also changed.
 
 ```php
 namespace App\Domain\Hotel;
@@ -215,7 +231,8 @@ final class Hotel extends AggregateRoot
 }
 ```
 
-> :book: You can find out more about aggregates and events [here](./docs/aggregate.md).
+> :book: You can find out more about aggregates 
+> and events in the library [documentation](https://github.com/patchlevel/event-sourcing#documentation).
 
 Next we have to make our aggregate known:
 
@@ -228,8 +245,9 @@ patchlevel_event_sourcing:
 
 ### Define projections
 
-So that we can see all the hotels on our website and also see how many guests are currently visiting the hotels, we need
-a projection for it.
+So that we can see all the hotels on our website 
+and also see how many guests are currently visiting the hotels, 
+we need a projection for it.
 
 ```php
 namespace App\Projection;

docs/cli.md

@@ -0,0 +1,32 @@
+# CLI
+
+The bundle also offers `cli commands` to create or delete `databases`. 
+It is also possible to manage the `schema` and `projections`.
+
+## Database commands
+
+There are two commands for creating and deleting a database.
+
+* `event-sourcing:database:create`
+* `event-sourcing:database:drop`
+
+## Schema commands
+
+The database schema can also be created, updated and dropped.
+
+* `event-sourcing:schema:create`
+* `event-sourcing:schema:update`
+* `event-sourcing:schema:drop`
+
+> :book: You can also register doctrine migration commands,
+> see the [store](./store.md#Migration) documentation for this.
+
+## Projection commands
+
+The creation, deletion and rebuilding of the projections is also possible via the cli.
+
+* `event-sourcing:projection:create`
+* `event-sourcing:projection:drop`
+* `event-sourcing:projection:rebuild`
+
+> :book: The [pipeline](./pipeline.md) will be used to rebuild the projection.

docs/pipeline.md

@@ -0,0 +1,92 @@
+# Pipeline
+
+A store is immutable, i.e. it cannot be changed afterwards. This includes both manipulating events and deleting them.
+
+Instead, you can duplicate the store and manipulate the events in the process. Thus the old store remains untouched and
+you can test the new store beforehand, whether the migration worked.
+
+In this example the event `PrivacyAdded` is removed and the event `OldVisited` is replaced by `NewVisited`:
+
+```php
+namespace App\Command;
+
+use Doctrine\DBAL\Connection;
+use Patchlevel\EventSourcing\Pipeline\Middleware\ExcludeEventMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Middleware\RecalculatePlayheadMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Middleware\ReplaceEventMiddleware;
+use Patchlevel\EventSourcing\Pipeline\Pipeline;
+use Patchlevel\EventSourcing\Pipeline\Source\StoreSource;
+use Patchlevel\EventSourcing\Pipeline\Target\StoreTarget;
+use Patchlevel\EventSourcing\Store\MultiTableStore;
+use Patchlevel\EventSourcing\Store\Store;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+
+class CreateUserCommand extends Command
+{
+    protected static $defaultName = 'app:event-sourcing:migrate-db-v2';
+
+    private Store $oldStore;
+    private Connection $newConnection;
+    private array $aggregates;
+
+    public function __construct(Store $oldStore, Connection $newConnection, array $aggregates) 
+    {
+        $this->oldStore = $oldStore;
+        $this->newConnection = $newConnection;
+        $this->aggregates = $aggregates;
+
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $console = new SymfonyStyle($input, $output);
+        $newStore = new MultiTableStore($this->newConnection, $this->aggregates);
+
+        $pipeline = new Pipeline(
+            new StoreSource($oldStore),
+            new StoreTarget($newStore),
+            [
+                new ExcludeEventMiddleware([PrivacyAdded::class]),
+                new ReplaceEventMiddleware(OldVisited::class, static function (OldVisited $oldVisited) {
+                    return NewVisited::raise($oldVisited->profileId());
+                }),
+                new RecalculatePlayheadMiddleware(),
+            ]
+        );
+
+        $console->progressStart($pipeline->count());
+
+        $pipeline->run(static function () use ($console): void {
+            $console->progressAdvance();
+        });
+
+        $console->progressFinish();
+        $console->success('finish');
+
+        return Command::SUCCESS;
+    }
+}
+```
+
+The whole thing just has to be plugged together.
+
+```yaml
+services:
+  App\Command\CreateUserCommand:
+    arguments:
+      aggregates: '%event_sourcing.aggregates%'
+      newConnection: '@doctrine.dbal.new_connection'
+```
+
+> :book: If you have the doctrine bundle for the dbal connections, 
+> then you can [autowiren](https://symfony.com/bundles/DoctrineBundle/current/configuration.html#autowiring-multiple-connections) it.
+
+

docs/store.md

@@ -53,6 +53,29 @@ bin/console event-sourcing:schema:udapte
 bin/console event-sourcing:schema:drop
 ```
 
+## Use doctrine connection
+
+If you have installed the [doctrine bundle](https://github.com/doctrine/DoctrineBundle), 
+you can also define the connection via doctrine and then use it in the store.
+
+```yaml
+doctrine:
+    dbal:
+        connections:
+            eventstore:
+                url: '%env(EVENTSTORE_URL)%'
+
+patchlevel_event_sourcing:
+    connection:
+        service: doctrine.dbal.eventstore_connection
+```
+
+> :warning: You should avoid that this connection or database is used by other tools or libraries. 
+> Create for e.g. doctrine orm its own database and connection.
+
+> :book: You can find out more about the dbal configuration 
+> [here](https://symfony.com/bundles/DoctrineBundle/current/configuration.html).
+
 ## Migration
 
 You can also manage your schema with doctrine migrations.

src/DependencyInjection/PatchlevelEventSourcingExtension.php

@@ -179,7 +179,7 @@ private function configureStorage(array $config, ContainerBuilder $container): v
             $container->register(SingleTableStore::class)
                 ->setArguments([
                     new Reference('event_sourcing.dbal_connection'),
-                    $this->aggregateHashMap($config['aggregates']),
+                    '%event_sourcing.aggregates%',
                     $config['store']['options']['table_name'] ?? 'eventstore',
                 ]);
 
@@ -195,7 +195,7 @@ private function configureStorage(array $config, ContainerBuilder $container): v
         $container->register(MultiTableStore::class)
             ->setArguments([
                 new Reference('event_sourcing.dbal_connection'),
-                $this->aggregateHashMap($config['aggregates']),
+                '%event_sourcing.aggregates%',
                 $config['store']['options']['table_name'] ?? 'eventstore',
             ]);
 
@@ -233,7 +233,7 @@ private function configureSnapshots(array $config, ContainerBuilder $container):
      */
     private function configureAggregates(array $config, ContainerBuilder $container): void
     {
-        $container->setParameter('event_sourcing.aggregates', $config['aggregates']);
+        $container->setParameter('event_sourcing.aggregates', $this->aggregateHashMap($config['aggregates']));
 
         foreach ($config['aggregates'] as $aggregateName => $definition) {
             $id = sprintf('event_sourcing.repository.%s', $aggregateName);
@@ -327,7 +327,7 @@ private function configureCommands(array $config, ContainerBuilder $container):
         $container->register(ShowCommand::class)
             ->setArguments([
                 new Reference(Store::class),
-                $this->aggregateHashMap($config['aggregates']),
+                '%event_sourcing.aggregates%',
             ])
             ->addTag('console.command');
     }