upgrade event-sourcing to 3.12

Author: DavidBadura

composer.json

@@ -19,7 +19,7 @@
   ],
   "require": {
     "php": "~8.2.0 || ~8.3.0 || ~8.4.0",
-    "patchlevel/event-sourcing": "^3.11.0",
+    "patchlevel/event-sourcing": "dev-DCB as 3.12.0",
     "symfony/cache": "^6.4.0|^7.0.0",
     "symfony/config": "^6.4.0|^7.0.0",
     "symfony/console": "^6.4.1|^7.0.1",

composer.lock

@@ -6,11 +6,19 @@
 
 use Symfony\Component\Config\Definition\Builder\TreeBuilder;
 use Symfony\Component\Config\Definition\ConfigurationInterface;
+use Throwable;
 
 /**
  * @psalm-type Config = array{
  *      event_bus: array{enabled: bool, type: string, service: string},
- *      command_bus: array{enabled: bool, service: string},
+ *      command_bus: array{
+ *          enabled: bool,
+ *          service: string,
+ *          instant_retry: array{
+ *              default_max_retries: positive-int|0,
+ *              default_exceptions: list<class-string<Throwable>>
+ *          },
+ *      },
  *      query_bus: array{enabled: bool, service: string},
  *      subscription: array{
  *          store: array{type: string, service: string|null},
@@ -69,6 +77,7 @@
  *      },
  *      clock: array{freeze: ?string, service: ?string},
  *      aggregate_handlers: array{enabled: bool, bus: string|null},
+ *      dcb: array{enabled: bool},
  * }
  */
 final class Configuration implements ConfigurationInterface
@@ -96,7 +105,7 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->addDefaultsIfNotSet()
                 ->children()
                     ->enumNode('type')
-                        ->values(['dbal_aggregate', 'dbal_stream', 'in_memory', 'custom'])
+                        ->values(['dbal_aggregate', 'dbal_stream', 'dbal_taggable', 'in_memory', 'custom'])
                         ->defaultValue('dbal_aggregate')
                     ->end()
                     ->scalarNode('service')->defaultNull()->end()
@@ -108,7 +117,7 @@ public function getConfigTreeBuilder(): TreeBuilder
                         ->addDefaultsIfNotSet()
                         ->children()
                             ->enumNode('type')
-                                ->values(['dbal_aggregate', 'dbal_stream', 'in_memory', 'custom'])
+                                ->values(['dbal_aggregate', 'dbal_stream', 'dbal_taggable', 'in_memory', 'custom'])
                             ->end()
                             ->scalarNode('service')->defaultNull()->end()
                             ->arrayNode('options')->variablePrototype()->end()->end()
@@ -302,6 +311,18 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->children()
                     ->scalarNode('service')->isRequired()->end()
                     ->booleanNode('register_aggregate_handlers')->defaultTrue()->end()
+                    ->arrayNode('instant_retry')
+                        ->addDefaultsIfNotSet()
+                        ->children()
+                            ->integerNode('default_max_retries')
+                                ->defaultValue(3)
+                            ->end()
+                            ->arrayNode('default_exceptions')
+                                ->defaultValue([])
+                                ->scalarPrototype()->end()
+                            ->end()
+                        ->end()
+                    ->end()
                 ->end()
             ->end()
 
@@ -313,6 +334,10 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
             ->end()
 
+            ->arrayNode('dcb')
+                ->canBeEnabled()
+            ->end()
+
             ->arrayNode('aggregate_handlers')
                 ->canBeEnabled()
                 ->addDefaultsIfNotSet()

src/DependencyInjection/Configuration.php

@@ -25,6 +25,7 @@
 use Patchlevel\EventSourcing\Clock\FrozenClock;
 use Patchlevel\EventSourcing\Clock\SystemClock;
 use Patchlevel\EventSourcing\CommandBus\CommandBus;
+use Patchlevel\EventSourcing\CommandBus\InstantRetryCommandBus;
 use Patchlevel\EventSourcing\Console\Command\DatabaseCreateCommand;
 use Patchlevel\EventSourcing\Console\Command\DatabaseDropCommand;
 use Patchlevel\EventSourcing\Console\Command\DebugCommand;
@@ -44,6 +45,12 @@
 use Patchlevel\EventSourcing\Console\Command\WatchCommand;
 use Patchlevel\EventSourcing\Console\DoctrineHelper;
 use Patchlevel\EventSourcing\Cryptography\DoctrineCipherKeyStore;
+use Patchlevel\EventSourcing\DCB\AttributeEventTagExtractor;
+use Patchlevel\EventSourcing\DCB\DecisionModelBuilder;
+use Patchlevel\EventSourcing\DCB\EventAppender;
+use Patchlevel\EventSourcing\DCB\EventTagExtractor;
+use Patchlevel\EventSourcing\DCB\StoreDecisionModelBuilder;
+use Patchlevel\EventSourcing\DCB\StoreEventAppender;
 use Patchlevel\EventSourcing\EventBus\AttributeListenerProvider;
 use Patchlevel\EventSourcing\EventBus\Consumer;
 use Patchlevel\EventSourcing\EventBus\DefaultConsumer;
@@ -69,6 +76,7 @@
 use Patchlevel\EventSourcing\QueryBus\QueryBus;
 use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager;
 use Patchlevel\EventSourcing\Repository\MessageDecorator\ChainMessageDecorator;
+use Patchlevel\EventSourcing\Repository\MessageDecorator\EventTagDecorator;
 use Patchlevel\EventSourcing\Repository\MessageDecorator\MessageDecorator;
 use Patchlevel\EventSourcing\Repository\MessageDecorator\SplitStreamDecorator;
 use Patchlevel\EventSourcing\Repository\RepositoryManager;
@@ -95,6 +103,7 @@
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore;
 use Patchlevel\EventSourcing\Store\StreamReadOnlyStore;
+use Patchlevel\EventSourcing\Store\TaggableDoctrineDbalStore;
 use Patchlevel\EventSourcing\Subscription\Engine\CatchUpSubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Engine\DefaultSubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Engine\GapResolverStoreMessageLoader;
@@ -189,6 +198,7 @@ public function load(array $configs, ContainerBuilder $container): void
         $this->configureMigration($config, $container);
         $this->configureValueResolver($container);
         $this->configureStoreMigration($config, $container);
+        $this->configureDCB($config, $container);
     }
 
     /** @param Config $config */
@@ -231,6 +241,9 @@ private function configureSerializer(array $config, ContainerBuilder $container)
             ]);
 
         $container->setAlias(HeadersSerializer::class, DefaultHeadersSerializer::class);
+
+        $container->register(AttributeEventTagExtractor::class);
+        $container->setAlias(EventTagExtractor::class, AttributeEventTagExtractor::class);
     }
 
     /** @param Config $config */
@@ -246,7 +259,14 @@ private function configureCommandBus(array $config, ContainerBuilder $container)
                     new Reference($config['command_bus']['service']),
                 ]);
 
-            $container->setAlias(CommandBus::class, SymfonyCommandBus::class);
+            $container->register(InstantRetryCommandBus::class)
+                ->setArguments([
+                    new Reference(SymfonyCommandBus::class),
+                    $config['command_bus']['instant_retry']['default_max_retries'],
+                    $config['command_bus']['instant_retry']['default_exceptions'],
+                ]);
+
+            $container->setAlias(CommandBus::class, InstantRetryCommandBus::class);
 
             $container->setParameter(
                 'patchlevel_event_sourcing.aggregate_handlers.bus',
@@ -621,6 +641,10 @@ private function configureMessageDecorator(ContainerBuilder $container): void
             ->setArguments([new Reference(EventMetadataFactory::class)])
             ->addTag('event_sourcing.message_decorator');
 
+        $container->register(EventTagDecorator::class)
+            ->setArguments([new Reference(EventTagExtractor::class)])
+            ->addTag('event_sourcing.message_decorator');
+
         $container->registerForAutoconfiguration(MessageDecorator::class)
             ->addTag('event_sourcing.message_decorator');
 
@@ -748,6 +772,26 @@ private function configureStore(array $config, ContainerBuilder $container): voi
             return;
         }
 
+        if ($config['store']['type'] === 'dbal_taggable') {
+            $container->register(TaggableDoctrineDbalStore::class)
+                ->setArguments([
+                    new Reference('event_sourcing.dbal_connection'),
+                    new Reference(EventSerializer::class),
+                    new Reference(HeadersSerializer::class),
+                    new Reference('event_sourcing.clock'),
+                    $config['store']['options'],
+                ])
+                ->addTag('event_sourcing.doctrine_schema_configurator');
+
+            $container->setAlias(Store::class, TaggableDoctrineDbalStore::class);
+
+            if ($config['store']['read_only']) {
+                throw new InvalidArgumentException('Taggable store does not support read only');
+            }
+
+            return;
+        }
+
         throw new InvalidArgumentException(sprintf('Unknown store type "%s"', $config['store']['type']));
     }
 
@@ -814,6 +858,20 @@ private function configureStoreMigration(array $config, ContainerBuilder $contai
             return;
         }
 
+        if ($config['store']['migrate_to_new_store']['type'] === 'dbal_taggable') {
+            $container->register($id, TaggableDoctrineDbalStore::class)
+                ->setArguments([
+                    new Reference('event_sourcing.dbal_connection'),
+                    new Reference(EventSerializer::class),
+                    new Reference(HeadersSerializer::class),
+                    new Reference('event_sourcing.clock'),
+                    $config['store']['migrate_to_new_store']['options'],
+                ])
+                ->addTag('event_sourcing.doctrine_schema_configurator');
+
+            return;
+        }
+
         throw new InvalidArgumentException(sprintf('Unknown store type "%s"', $config['store']['type']));
     }
 
@@ -1165,6 +1223,31 @@ private function configureCryptography(array $config, ContainerBuilder $containe
         $container->setAlias(PayloadCryptographer::class, PersonalDataPayloadCryptographer::class);
     }
 
+    /** @param Config $config */
+    private function configureDCB(array $config, ContainerBuilder $container): void
+    {
+        if (!$config['dcb']['enabled']) {
+            return;
+        }
+
+        if ($config['store']['type'] !== 'dbal_taggable') {
+            throw new InvalidArgumentException(
+                'DCB requires a taggable store, please use "dbal_taggable" as store type.',
+            );
+        }
+
+        $container->register(StoreDecisionModelBuilder::class)
+            ->setArguments([new Reference(TaggableDoctrineDbalStore::class)]);
+        $container->setAlias(DecisionModelBuilder::class, StoreDecisionModelBuilder::class);
+
+        $container->register(StoreEventAppender::class)
+            ->setArguments([
+                new Reference(TaggableDoctrineDbalStore::class),
+                new Reference(EventTagExtractor::class),
+            ]);
+        $container->setAlias(EventAppender::class, StoreEventAppender::class);
+    }
+
     private function configureValueResolver(ContainerBuilder $container): void
     {
         $container->register(AggregateRootIdValueResolver::class)