patchlevel · DavidBadura · Oct 6, 2025 · Jul 30, 2025
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
@@ -78,6 +78,7 @@
  *      },
  *      clock: array{freeze: ?string, service: ?string},
  *      aggregate_handlers: array{enabled: bool, bus: string|null},
+ *      dcb: array{enabled: bool},
  * }
  */
 final class Configuration implements ConfigurationInterface
@@ -105,7 +106,7 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->addDefaultsIfNotSet()
                 ->children()
                     ->enumNode('type')
-                        ->values(['dbal_stream', 'in_memory', 'custom'])
+                        ->values(['dbal_stream', 'dbal_taggable', 'in_memory', 'custom'])
                         ->defaultValue('dbal_stream')
                     ->end()
                     ->scalarNode('service')->defaultNull()->end()
@@ -117,7 +118,7 @@ public function getConfigTreeBuilder(): TreeBuilder
                         ->addDefaultsIfNotSet()
                         ->children()
                             ->enumNode('type')
-                                ->values(['dbal_stream', 'in_memory', 'custom'])
+                                ->values(['dbal_stream', 'dbal_taggable', 'in_memory', 'custom'])
                             ->end()
                             ->scalarNode('service')->defaultNull()->end()
                             ->arrayNode('options')->variablePrototype()->end()->end()
@@ -334,6 +335,10 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
             ->end()
 
+            ->arrayNode('dcb')
+                ->canBeEnabled()
+            ->end()
+
             ->arrayNode('aggregate_handlers')
                 ->canBeEnabled()
                 ->addDefaultsIfNotSet()

diff --git a/src/DependencyInjection/PatchlevelEventSourcingExtension.php b/src/DependencyInjection/PatchlevelEventSourcingExtension.php
@@ -45,6 +45,10 @@
 use Patchlevel\EventSourcing\Console\Command\WatchCommand;
 use Patchlevel\EventSourcing\Console\DoctrineHelper;
 use Patchlevel\EventSourcing\Cryptography\DoctrineCipherKeyStore;
+use Patchlevel\EventSourcing\DecisionModel\DecisionModelBuilder;
+use Patchlevel\EventSourcing\DecisionModel\EventAppender;
+use Patchlevel\EventSourcing\DecisionModel\StoreDecisionModelBuilder;
+use Patchlevel\EventSourcing\DecisionModel\StoreEventAppender;
 use Patchlevel\EventSourcing\EventBus\AttributeListenerProvider;
 use Patchlevel\EventSourcing\EventBus\Consumer;
 use Patchlevel\EventSourcing\EventBus\DefaultConsumer;
@@ -70,6 +74,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;
@@ -80,20 +85,23 @@
 use Patchlevel\EventSourcing\Schema\DoctrineSchemaListener;
 use Patchlevel\EventSourcing\Schema\DoctrineSchemaProvider;
 use Patchlevel\EventSourcing\Schema\SchemaDirector;
+use Patchlevel\EventSourcing\Serializer\AttributeEventTagExtractor;
 use Patchlevel\EventSourcing\Serializer\DefaultEventSerializer;
 use Patchlevel\EventSourcing\Serializer\Encoder\Encoder;
 use Patchlevel\EventSourcing\Serializer\Encoder\JsonEncoder;
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
+use Patchlevel\EventSourcing\Serializer\EventTagExtractor;
 use Patchlevel\EventSourcing\Serializer\Upcast\Upcaster;
 use Patchlevel\EventSourcing\Serializer\Upcast\UpcasterChain;
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr16SnapshotAdapter;
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr6SnapshotAdapter;
 use Patchlevel\EventSourcing\Snapshot\DefaultSnapshotStore;
 use Patchlevel\EventSourcing\Snapshot\SnapshotStore;
 use Patchlevel\EventSourcing\Store\InMemoryStore;
+use Patchlevel\EventSourcing\Store\ReadOnlyStore;
 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;
@@ -188,6 +196,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 */
@@ -230,6 +239,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 */
@@ -627,6 +639,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');
 
@@ -721,7 +737,7 @@ private function configureStore(array $config, ContainerBuilder $container): voi
             $container->setAlias(Store::class, StreamDoctrineDbalStore::class);
 
             if ($config['store']['read_only']) {
-                $container->register(StreamReadOnlyStore::class)
+                $container->register(ReadOnlyStore::class)
                     ->setDecoratedService(Store::class)
                     ->setArguments([
                         new Reference('.inner'),
@@ -731,6 +747,27 @@ 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(EventRegistry::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']));
     }
 
@@ -784,6 +821,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']));
     }
 
@@ -1135,6 +1186,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(IdentifierValueResolver::class)

diff --git a/tests/Unit/PatchlevelEventSourcingBundleTest.php b/tests/Unit/PatchlevelEventSourcingBundleTest.php
@@ -34,6 +34,12 @@
 use Patchlevel\EventSourcing\Console\Command\SubscriptionStatusCommand;
 use Patchlevel\EventSourcing\Console\Command\SubscriptionTeardownCommand;
 use Patchlevel\EventSourcing\Console\Command\WatchCommand;
+use Patchlevel\EventSourcing\Serializer\AttributeEventTagExtractor;
+use Patchlevel\EventSourcing\DecisionModel\DecisionModelBuilder;
+use Patchlevel\EventSourcing\DecisionModel\EventAppender;
+use Patchlevel\EventSourcing\Serializer\EventTagExtractor;
+use Patchlevel\EventSourcing\DecisionModel\StoreDecisionModelBuilder;
+use Patchlevel\EventSourcing\DecisionModel\StoreEventAppender;
 use Patchlevel\EventSourcing\EventBus\DefaultEventBus;
 use Patchlevel\EventSourcing\EventBus\EventBus;
 use Patchlevel\EventSourcing\EventBus\Psr14EventBus;
@@ -61,8 +67,7 @@
 use Patchlevel\EventSourcing\Store\InMemoryStore;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Store\StreamDoctrineDbalStore;
-use Patchlevel\EventSourcing\Store\StreamReadOnlyStore;
-use Patchlevel\EventSourcing\Store\StreamStore;
+use Patchlevel\EventSourcing\Store\ReadOnlyStore;
 use Patchlevel\EventSourcing\Subscription\Engine\CatchUpSubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Engine\DefaultSubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Engine\GapResolverStoreMessageLoader;
@@ -154,6 +159,7 @@ public function testMinimalConfig(): void
         self::assertInstanceOf(EventRegistry::class, $container->get(EventRegistry::class));
         self::assertInstanceOf(SystemClock::class, $container->get('event_sourcing.clock'));
         self::assertInstanceOf(DefaultSubscriptionEngine::class, $container->get(SubscriptionEngine::class));
+        self::assertInstanceOf(AttributeEventTagExtractor::class, $container->get(EventTagExtractor::class));
 
         self::assertFalse($container->has(EventBus::class));
 
@@ -278,7 +284,7 @@ public function testStreamStore(): void
             ]
         );
 
-        self::assertInstanceOf(StreamStore::class, $container->get(Store::class));
+        self::assertInstanceOf(StreamDoctrineDbalStore::class, $container->get(Store::class));
     }
 
     public function testInMemoryStore(): void
@@ -318,7 +324,7 @@ public function testReadOnlyStore(): void
             ]
         );
 
-        self::assertInstanceOf(StreamReadOnlyStore::class, $container->get(Store::class));
+        self::assertInstanceOf(ReadOnlyStore::class, $container->get(Store::class));
     }
 
     public function testMigrateStore(): void
@@ -365,27 +371,6 @@ public function testMigrateStore(): void
         );
     }
 
-    public function testStreamReadOnlyStore(): void
-    {
-        $container = new ContainerBuilder();
-        $this->compileContainer(
-            $container,
-            [
-                'patchlevel_event_sourcing' => [
-                    'connection' => [
-                        'service' => 'doctrine.dbal.eventstore_connection',
-                    ],
-                    'store' => [
-                        'type' => 'dbal_stream',
-                        'read_only' => true,
-                    ]
-                ],
-            ]
-        );
-
-        self::assertInstanceOf(StreamReadOnlyStore::class, $container->get(Store::class));
-    }
-
     public function testSymfonyEventBus(): void
     {
         $eventBus = $this->prophesize(MessageBusInterface::class)->reveal();
@@ -662,6 +647,30 @@ public function testQueryBus(): void
         self::assertEquals('foo', $handler->{$tag['method']}(new QueryFoo('foo')));
     }
 
+
+    public function testDCB(): void
+    {
+        $container = new ContainerBuilder();
+
+        $this->compileContainer(
+            $container,
+            [
+                'patchlevel_event_sourcing' => [
+                    'connection' => [
+                        'service' => 'doctrine.dbal.eventstore_connection',
+                    ],
+                    'store' => [
+                        'type' => 'dbal_taggable',
+                    ],
+                    'dcb' => true,
+                ],
+            ]
+        );
+
+        self::assertInstanceOf(StoreEventAppender::class, $container->get(EventAppender::class));
+        self::assertInstanceOf(StoreDecisionModelBuilder::class, $container->get(DecisionModelBuilder::class));
+    }
+
     public function testMessageLoader(): void
     {
         $container = new ContainerBuilder();