Add event emitter, GIN index for tags and DCB with any append store

docs/configuration.md

@@ -178,14 +178,22 @@ patchlevel_event_sourcing:
 Following store types are available:
 
 - `dbal_stream` *default*
-- `dbal_taggable` *required for DCB*
+- `dbal_taggable`
 - `in_memory`
 - `custom`
 
 :::note
 If you use `custom` store type, you need to set the service id under `patchlevel_event_sourcing.store.service`.
 :::
 
+:::tip
+With `dbal_taggable` the bundle registers the `PostgreSQLPlatformMiddleware` on the event store connection,
+so a GIN index on the `tags` column is created on PostgreSQL.
+This works automatically if you configure the connection via `url` or use a doctrine bundle connection
+like `doctrine.dbal.eventstore_connection`.
+For any other connection service you have to register the middleware yourself.
+:::
+
 ### Change table Name
 
 You can change the table name of the event store.
@@ -550,6 +558,23 @@ final class SubscriptionErrorListener
 You can find all available events in the [library documentation](/docs/event-sourcing/latest/subscription).
 :::
 
+### Event Emitter
+
+Subscribers can emit new events into their own `subscription_<id>` stream with the `EventEmitter`.
+This is disabled by default and can be enabled like this:
+
+```yaml
+patchlevel_event_sourcing:
+    subscription:
+        event_emitter: true
+```
+:::warning
+If a subscription is removed, its `subscription_<id>` stream is removed from the event store as well.
+:::
+
+:::note
+The event emitter does not work with a read only store.
+:::
 ## Command Bus
 
 You can enable the command bus integration to use your aggregates as command handlers.
@@ -869,4 +894,25 @@ You can then specify this service here:
 patchlevel_event_sourcing:
     clock:
         service: 'my_own_clock_service'
-```
+```
+
+## Dynamic Consistency Boundary
+
+You can enable the experimental dynamic consistency boundary (DCB).
+This registers the `DecisionModelBuilder` and the `EventAppender` services.
+
+```yaml
+patchlevel_event_sourcing:
+    store:
+        type: 'dbal_taggable'
+    dcb: true
+```
+:::note
+DCB needs a store that supports appending, like `dbal_taggable` or `in_memory` for tests.
+:::
+
+:::tip
+We recommend PostgreSQL for DCB, because only there the tag queries can use an index.
+:::
+
+If you want to learn more about DCB, read the [library documentation](/docs/event-sourcing/latest/dynamic-consistency-boundary).

src/DependencyInjection/Configuration.php

@@ -52,7 +52,8 @@
  *              enabled: bool,
  *              retries_in_ms: list<int>,
  *              detection_window: string
- *          }
+ *          },
+ *          event_emitter: array{enabled: bool}
  *      },
  *      connection: ?array{
  *          service: ?string,
@@ -184,7 +185,7 @@
                         ->addDefaultsIfNotSet()
                         ->children()
                             ->enumNode('type')
                                 ->values(['dbal', 'in_memory', 'static_in_memory', 'custom'])
                                 ->defaultValue('dbal')
                             ->end()
                             ->scalarNode('service')->defaultNull()->end()
@@ -206,13 +207,13 @@
                                 ->arrayNode('options')->variablePrototype()->end()->end()
                             ->end()
                         ->end()
                         ->defaultValue([
                             'default' => [
                                 'type' => 'clock_based',
                                 'options' => [
                                     'base_delay' => 5,
                                     'delay_factor' => 2,
                                     'max_attempts' => 5,
                                 ],
                             ],
                             'no_retry' => [
@@ -264,6 +265,10 @@
                             ->scalarNode('detection_window')->defaultValue('PT5M')->end()
                         ->end()
                     ->end()
+
+                    ->arrayNode('event_emitter')
+                        ->canBeEnabled()
+                    ->end()
                 ->end()
             ->end()
 

src/DependencyInjection/PatchlevelEventSourcingExtension.php

@@ -92,6 +92,7 @@
 use Patchlevel\EventSourcing\Snapshot\Adapter\Psr6SnapshotAdapter;
 use Patchlevel\EventSourcing\Snapshot\DefaultSnapshotStore;
 use Patchlevel\EventSourcing\Snapshot\SnapshotStore;
+use Patchlevel\EventSourcing\Store\Dbal\PostgreSQLPlatformMiddleware;
 use Patchlevel\EventSourcing\Store\InMemoryStore;
 use Patchlevel\EventSourcing\Store\ReadOnlyStore;
 use Patchlevel\EventSourcing\Store\Store;
@@ -102,7 +103,9 @@
 use Patchlevel\EventSourcing\Subscription\Cleanup\DefaultCleaner;
 use Patchlevel\EventSourcing\Subscription\Engine\CatchUpSubscriptionEngine;
 use Patchlevel\EventSourcing\Subscription\Engine\DefaultSubscriptionEngine;
+use Patchlevel\EventSourcing\Subscription\Engine\Event\OnSubscriptionRemoved;
 use Patchlevel\EventSourcing\Subscription\Engine\GapResolverStoreMessageLoader;
+use Patchlevel\EventSourcing\Subscription\Engine\Listener\RemoveSubscriptionStreamListener;
 use Patchlevel\EventSourcing\Subscription\Engine\MessageLoader;
 use Patchlevel\EventSourcing\Subscription\Engine\StoreMessageLoader;
 use Patchlevel\EventSourcing\Subscription\Engine\SubscriptionEngine;
@@ -115,6 +118,7 @@
 use Patchlevel\EventSourcing\Subscription\Store\InMemorySubscriptionStore;
 use Patchlevel\EventSourcing\Subscription\Store\SubscriptionStore;
 use Patchlevel\EventSourcing\Subscription\Subscriber\ArgumentResolver\ArgumentResolver;
+use Patchlevel\EventSourcing\Subscription\Subscriber\ArgumentResolver\EventEmitterResolver;
 use Patchlevel\EventSourcing\Subscription\Subscriber\ArgumentResolver\LookupResolver;
 use Patchlevel\EventSourcing\Subscription\Subscriber\MetadataSubscriberAccessorRepository;
 use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberAccessorRepository;
@@ -156,6 +160,7 @@
 use Symfony\Component\EventDispatcher\EventDispatcher;
 
 use function class_exists;
+use function preg_match;
 use function sprintf;
 
 /** @psalm-import-type Config from Configuration */
@@ -533,6 +538,7 @@
             ]);
 
         $this->configureSyncSubscription($config, $container);
+        $this->configureEventEmitter($config, $container);
 
         if ($config['subscription']['auto_setup']['enabled']) {
             $container->register(AutoSetupListener::class)
@@ -599,6 +605,33 @@
             ]);
     }
 
+    /** @param Config $config */
+    private function configureEventEmitter(array $config, ContainerBuilder $container): void
+    {
+        if (!$config['subscription']['event_emitter']['enabled']) {
+            return;
+        }
+
+        if ($config['store']['read_only']) {
+            throw new InvalidArgumentException('Event emitter does not support a read only store');
+        }
+
+        $container->register(EventEmitterResolver::class)
+            ->setArguments([new Reference(Store::class)])
+            ->addTag('event_sourcing.argument_resolver');
+
+        $container->register(RemoveSubscriptionStreamListener::class)
+            ->setArguments([
+                new Reference(Store::class),
+                new Reference('logger', ContainerInterface::NULL_ON_INVALID_REFERENCE),
+            ])
+            ->addTag('kernel.event_listener', [
+                'event' => OnSubscriptionRemoved::class,
+                'dispatcher' => 'event_sourcing.subscription.event_dispatcher',
+            ])
+            ->addTag('monolog.logger', ['channel' => 'event_sourcing']);
+    }
+
     /** @param Config $config */
     private function configureHydrator(array $config, ContainerBuilder $container): void
     {
@@ -681,11 +714,19 @@
             return;
         }
 
+        $middlewares = [];
+
+        if ($this->usesTaggableStore($config)) {
+            $container->register(PostgreSQLPlatformMiddleware::class);
+            $middlewares[] = new Reference(PostgreSQLPlatformMiddleware::class);
+        }
+
         if ($config['connection']['url'] !== null) {
             $container->register('event_sourcing.dbal_connection', Connection::class)
                 ->setFactory([DbalConnectionFactory::class, 'createConnection'])
                 ->setArguments([
                     $config['connection']['url'],
+                    $middlewares,
                 ]);
 
             if ($config['connection']['provide_dedicated_connection']) {
@@ -710,6 +751,26 @@
         }
 
         $container->setAlias('event_sourcing.dbal_connection', $config['connection']['service']);
+
+        if (
+            $middlewares === []
+            || !preg_match('/^doctrine\.dbal\.(.+)_connection$/', $config['connection']['service'], $matches)
+        ) {
+            return;
+        }
+
+        $container->getDefinition(PostgreSQLPlatformMiddleware::class)
+            ->addTag('doctrine.middleware', ['connection' => $matches[1]]);
+    }
+
+    /** @param Config $config */
+    private function usesTaggableStore(array $config): bool
+    {
+        return $config['store']['type'] === 'dbal_taggable'
+            || (
+                $config['store']['migrate_to_new_store']['enabled']
+                && $config['store']['migrate_to_new_store']['type'] === 'dbal_taggable'
+            );
     }
 
     /** @param Config $config */
@@ -1196,19 +1257,19 @@
             return;
         }
 
-        if ($config['store']['type'] !== 'dbal_taggable') {
+        if ($config['store']['type'] === 'dbal_stream') {
             throw new InvalidArgumentException(
-                'DCB requires a taggable store, please use "dbal_taggable" as store type.',
+                'DCB requires a store that supports appending, please use "dbal_taggable", "in_memory" or a custom store.',
             );
         }
 
         $container->register(StoreDecisionModelBuilder::class)
-            ->setArguments([new Reference(TaggableDoctrineDbalStore::class)]);
+            ->setArguments([new Reference(Store::class)]);
         $container->setAlias(DecisionModelBuilder::class, StoreDecisionModelBuilder::class);
 
         $container->register(StoreEventAppender::class)
             ->setArguments([
-                new Reference(TaggableDoctrineDbalStore::class),
+                new Reference(Store::class),
                 new Reference(EventTagExtractor::class),
             ]);
         $container->setAlias(EventAppender::class, StoreEventAppender::class);

src/Doctrine/DbalConnectionFactory.php

@@ -4,7 +4,9 @@
 
 namespace Patchlevel\EventSourcingBundle\Doctrine;
 
+use Doctrine\DBAL\Configuration;
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Driver\Middleware;
 use Doctrine\DBAL\DriverManager;
 use Doctrine\DBAL\Tools\DsnParser;
 
@@ -27,10 +29,12 @@ final class DbalConnectionFactory
         'sqlite3'    => 'pdo_sqlite',
     ];
 
-    public static function createConnection(string $url): Connection
+    /** @param list<Middleware> $middlewares */
+    public static function createConnection(string $url, array $middlewares = []): Connection
     {
         return DriverManager::getConnection(
             (new DsnParser(self::DEFAULT_SCHEME_MAP))->parse($url),
+            (new Configuration())->setMiddlewares($middlewares),
         );
     }
 }