Add query bus

baseline.xml

@@ -1,51 +1,19 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<files psalm-version="6.7.1@a2f190972555ea01b0cfcc1913924d6c5fc1a64e">
-  <file src="src/Attribute/AsListener.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/Command/StoreMigrateCommand.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/CommandBus/SymfonyCommandBus.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/CommandBus/SymfonyParameterResolver.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/DataCollector/EventSourcingCollector.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-      <code><![CDATA[getAggregates]]></code>
-      <code><![CDATA[getEvents]]></code>
-      <code><![CDATA[getMessages]]></code>
-    </PossiblyUnusedMethod>
-  </file>
+<files psalm-version="6.8.8@1361cd33008feb3ae2b4a93f1860e14e538ec8c2">
   <file src="src/DataCollector/MessageCollectorEventBus.php">
     <ClassMustBeFinal>
       <code><![CDATA[MessageCollectorEventBus]]></code>
     </ClassMustBeFinal>
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-      <code><![CDATA[clear]]></code>
-    </PossiblyUnusedMethod>
   </file>
   <file src="src/DependencyInjection/Configuration.php">
     <UndefinedInterfaceMethod>
       <code><![CDATA[scalarNode]]></code>
     </UndefinedInterfaceMethod>
   </file>
-  <file src="src/DependencyInjection/PatchlevelEventSourcingExtension.php">
-    <UnusedClass>
-      <code><![CDATA[PatchlevelEventSourcingExtension]]></code>
-    </UnusedClass>
+  <file src="src/DependencyInjection/QueryHandlerCompilerPass.php">
+    <ArgumentTypeCoercion>
+      <code><![CDATA[$subscriberClass]]></code>
+    </ArgumentTypeCoercion>
   </file>
   <file src="src/DependencyInjection/SubscriberGuardCompilePass.php">
     <InvalidOperand>
@@ -55,50 +23,4 @@
       <code><![CDATA[$argument]]></code>
     </MixedAssignment>
   </file>
-  <file src="src/Doctrine/DbalConnectionFactory.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[createConnection]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/EventBus/SymfonyEventBus.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/Normalizer/SymfonyUuidNormalizer.php">
-    <UnusedClass>
-      <code><![CDATA[SymfonyUuidNormalizer]]></code>
-    </UnusedClass>
-  </file>
-  <file src="src/PatchlevelEventSourcingBundle.php">
-    <UnusedClass>
-      <code><![CDATA[PatchlevelEventSourcingBundle]]></code>
-    </UnusedClass>
-  </file>
-  <file src="src/RequestListener/AutoSetupListener.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-      <code><![CDATA[onKernelRequest]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/RequestListener/SubscriptionRebuildAfterFileChangeListener.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-      <code><![CDATA[onKernelRequest]]></code>
-    </PossiblyUnusedMethod>
-  </file>
-  <file src="src/Subscription/ResetServicesListener.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[__construct]]></code>
-      <code><![CDATA[onWorkerRunningEvent]]></code>
-    </PossiblyUnusedMethod>
-    <UnusedParam>
-      <code><![CDATA[$event]]></code>
-    </UnusedParam>
-  </file>
-  <file src="src/Subscription/StaticInMemorySubscriptionStoreFactory.php">
-    <PossiblyUnusedMethod>
-      <code><![CDATA[create]]></code>
-    </PossiblyUnusedMethod>
-  </file>
 </files>

composer.json

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

phpstan-baseline.neon

@@ -1,2 +1,7 @@
 parameters:
 	ignoreErrors:
+		-
+			message: '#^Parameter \#1 \$classString of static method Patchlevel\\EventSourcing\\QueryBus\\HandlerFinder\:\:findInClass\(\) expects class\-string, string given\.$#'
+			identifier: argument.type
+			count: 1
+			path: src/DependencyInjection/QueryHandlerCompilerPass.php

psalm.xml

@@ -6,6 +6,7 @@
     xmlns="https://getpsalm.org/schema/config"
     xsi:schemaLocation="https://getpsalm.org/schema/config vendor/vimeo/psalm/config.xsd"
     errorBaseline="baseline.xml"
+    findUnusedCode="false"
     ensureOverrideAttribute="false"
 >
     <projectFiles>

src/DependencyInjection/HandlerCompilerPass.php → src/DependencyInjection/CommandHandlerCompilerPass.php

@@ -17,7 +17,7 @@
 use function strtolower;
 
 /** @internal */
-final class HandlerCompilerPass implements CompilerPassInterface
+final class CommandHandlerCompilerPass implements CompilerPassInterface
 {
     public function process(ContainerBuilder $container): void
     {

src/DependencyInjection/Configuration.php

@@ -11,6 +11,7 @@
  * @psalm-type Config = array{
  *      event_bus: array{enabled: bool, type: string, service: string},
  *      command_bus: array{enabled: bool, service: string},
+ *      query_bus: array{enabled: bool, service: string},
  *      subscription: array{
  *          store: array{type: string, service: string|null},
  *          retry_strategy?: array{base_delay: int, delay_factor: int, max_attempts: int},
@@ -280,6 +281,14 @@ public function getConfigTreeBuilder(): TreeBuilder
                 ->end()
             ->end()
 
+            ->arrayNode('query_bus')
+                ->canBeEnabled()
+                ->addDefaultsIfNotSet()
+                ->children()
+                    ->scalarNode('service')->isRequired()->end()
+                ->end()
+            ->end()
+
             ->arrayNode('aggregate_handlers')
                 ->canBeEnabled()
                 ->addDefaultsIfNotSet()

src/DependencyInjection/PatchlevelEventSourcingExtension.php

@@ -65,6 +65,7 @@
 use Patchlevel\EventSourcing\Metadata\Message\MessageHeaderRegistryFactory;
 use Patchlevel\EventSourcing\Metadata\Subscriber\AttributeSubscriberMetadataFactory;
 use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadataFactory;
+use Patchlevel\EventSourcing\QueryBus\QueryBus;
 use Patchlevel\EventSourcing\Repository\DefaultRepositoryManager;
 use Patchlevel\EventSourcing\Repository\MessageDecorator\ChainMessageDecorator;
 use Patchlevel\EventSourcing\Repository\MessageDecorator\MessageDecorator;
@@ -117,6 +118,7 @@
 use Patchlevel\EventSourcingBundle\DataCollector\MessageCollectorEventBus;
 use Patchlevel\EventSourcingBundle\Doctrine\DbalConnectionFactory;
 use Patchlevel\EventSourcingBundle\EventBus\SymfonyEventBus;
+use Patchlevel\EventSourcingBundle\QueryBus\SymfonyQueryBus;
 use Patchlevel\EventSourcingBundle\RequestListener\AutoSetupListener;
 use Patchlevel\EventSourcingBundle\RequestListener\SubscriptionRebuildAfterFileChangeListener;
 use Patchlevel\EventSourcingBundle\Subscription\ResetServicesListener;
@@ -168,6 +170,7 @@ public function load(array $configs, ContainerBuilder $container): void
         $this->configureMessageDecorator($container);
         $this->configureCommandBus($config, $container);
         $this->configureEventBus($config, $container);
+        $this->configureQueryBus($config, $container);
         $this->configureConnection($config, $container);
         $this->configureStore($config, $container);
         $this->configureSnapshots($config, $container);
@@ -324,6 +327,26 @@ static function (ChildDefinition $definition, AsListener $attribute): void {
         $container->setAlias(EventBus::class, $config['event_bus']['service']);
     }
 
+    /** @param Config $config */
+    private function configureQueryBus(array $config, ContainerBuilder $container): void
+    {
+        if (!$config['query_bus']['enabled']) {
+            return;
+        }
+
+        $container->register(SymfonyQueryBus::class)
+            ->setArguments([
+                new Reference($config['query_bus']['service']),
+            ]);
+
+        $container->setAlias(QueryBus::class, SymfonyQueryBus::class);
+
+        $container->setParameter(
+            'patchlevel_event_sourcing.query_handlers.bus',
+            $config['query_bus']['service'],
+        );
+    }
+
     /** @param Config $config */
     private function configureSubscription(array $config, ContainerBuilder $container): void
     {

src/DependencyInjection/QueryHandlerCompilerPass.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcingBundle\DependencyInjection;
+
+use Patchlevel\EventSourcing\QueryBus\HandlerFinder;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+
+use function array_keys;
+
+/** @internal */
+final class QueryHandlerCompilerPass implements CompilerPassInterface
+{
+    public function process(ContainerBuilder $container): void
+    {
+        if (!$container->hasParameter('patchlevel_event_sourcing.query_handlers.bus')) {
+            return;
+        }
+
+        $bus = $container->getParameter('patchlevel_event_sourcing.query_handlers.bus');
+        $subscribers = $container->findTaggedServiceIds('event_sourcing.subscriber');
+
+        foreach (array_keys($subscribers) as $subscriberServiceName) {
+            $subscriberDefinition = $container->getDefinition($subscriberServiceName);
+            $subscriberClass = $subscriberDefinition->getClass();
+
+            if ($subscriberClass === null) {
+                continue;
+            }
+
+            foreach (HandlerFinder::findInClass($subscriberClass) as $queryHandler) {
+                $subscriberDefinition->addTag(
+                    'messenger.message_handler',
+                    [
+                        'method' => $queryHandler->method,
+                        'handles' => $queryHandler->queryClass,
+                        'bus' => $bus,
+                    ],
+                );
+            }
+        }
+    }
+}

src/PatchlevelEventSourcingBundle.php

@@ -4,8 +4,9 @@
 
 namespace Patchlevel\EventSourcingBundle;
 
-use Patchlevel\EventSourcingBundle\DependencyInjection\HandlerCompilerPass;
+use Patchlevel\EventSourcingBundle\DependencyInjection\CommandHandlerCompilerPass;
 use Patchlevel\EventSourcingBundle\DependencyInjection\HandlerServiceLocatorCompilerPass;
+use Patchlevel\EventSourcingBundle\DependencyInjection\QueryHandlerCompilerPass;
 use Patchlevel\EventSourcingBundle\DependencyInjection\RepositoryCompilerPass;
 use Patchlevel\EventSourcingBundle\DependencyInjection\SubscriberGuardCompilePass;
 use Patchlevel\EventSourcingBundle\DependencyInjection\TranslatorCompilerPass;
@@ -18,7 +19,8 @@ public function build(ContainerBuilder $container): void
     {
         $container->addCompilerPass(new RepositoryCompilerPass());
         $container->addCompilerPass(new SubscriberGuardCompilePass());
-        $container->addCompilerPass(new HandlerCompilerPass(), priority: 100);
+        $container->addCompilerPass(new CommandHandlerCompilerPass(), priority: 100);
+        $container->addCompilerPass(new QueryHandlerCompilerPass(), priority: 100);
         $container->addCompilerPass(new HandlerServiceLocatorCompilerPass(), priority: -100);
         $container->addCompilerPass(new TranslatorCompilerPass());
     }

src/QueryBus/SymfonyQueryBus.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcingBundle\QueryBus;
+
+use Patchlevel\EventSourcing\QueryBus\QueryBus;
+use RuntimeException;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Messenger\Stamp\HandledStamp;
+
+final class SymfonyQueryBus implements QueryBus
+{
+    public function __construct(
+        private readonly MessageBusInterface $messageBus,
+    ) {
+    }
+
+    public function dispatch(object $query): mixed
+    {
+        $handledStamp = $this->messageBus->dispatch($query)->last(HandledStamp::class);
+
+        if ($handledStamp === null) {
+            throw new RuntimeException('No message handled yet');
+        }
+
+        return $handledStamp->getResult();
+    }
+}

tests/Fixtures/ProfileProjector.php

@@ -2,9 +2,15 @@
 
 namespace Patchlevel\EventSourcingBundle\Tests\Fixtures;
 
+use Patchlevel\EventSourcing\Attribute\Answer;
 use Patchlevel\EventSourcing\Attribute\Projector;
 
 #[Projector('profile')]
 class ProfileProjector
 {
+    #[Answer]
+    public function query(QueryFoo $query): string
+    {
+        return $query->result;
+    }
 }

tests/Fixtures/QueryFoo.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcingBundle\Tests\Fixtures;
+
+final readonly class QueryFoo
+{
+    public function __construct(public string $result)
+    {
+    }
+}