Merge pull request #536 from patchlevel/improve-deployment

worker should not discover new subscriptions in running process

src/Console/Command/SubscriptionBootCommand.php

@@ -74,6 +74,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $setup = InputHelper::bool($input->getOption('setup'));
 
         $criteria = $this->subscriptionEngineCriteria($input);
+        $criteria = $this->resolveCriteriaIntoCriteriaWithOnlyIds($criteria);
 
         if ($setup) {
             $this->engine->setup($criteria);

src/Console/Command/SubscriptionCommand.php

@@ -11,6 +11,8 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 
+use function array_map;
+
 /** @interal */
 abstract class SubscriptionCommand extends Command
 {
@@ -44,4 +46,17 @@ protected function subscriptionEngineCriteria(InputInterface $input): Subscripti
             InputHelper::nullableStringList($input->getOption('group')),
         );
     }
+
+    protected function resolveCriteriaIntoCriteriaWithOnlyIds(
+        SubscriptionEngineCriteria $criteria,
+    ): SubscriptionEngineCriteria {
+        $subscriptions = $this->engine->subscriptions($criteria);
+
+        return new SubscriptionEngineCriteria(
+            array_map(
+                static fn ($subscription) => $subscription->id(),
+                $subscriptions,
+            ),
+        );
+    }
 }

src/Console/Command/SubscriptionRunCommand.php

@@ -73,6 +73,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $rebuild = InputHelper::bool($input->getOption('rebuild'));
 
         $criteria = $this->subscriptionEngineCriteria($input);
+        $criteria = $this->resolveCriteriaIntoCriteriaWithOnlyIds($criteria);
 
         $logger = new ConsoleLogger($output);
 

src/Subscription/Store/DoctrineSubscriptionStore.php

@@ -74,7 +74,8 @@ public function find(SubscriptionCriteria|null $criteria = null): array
     {
         $qb = $this->connection->createQueryBuilder()
             ->select('*')
-            ->from($this->tableName);
+            ->from($this->tableName)
+            ->orderBy('id');
 
         if (!$this->connection->getDatabasePlatform() instanceof SQLitePlatform) {
             $qb->forUpdate();

tests/Integration/Subscription/Subscriber/ProfileNewProjection.php

@@ -0,0 +1,66 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Schema\Table;
+use Patchlevel\EventSourcing\Attribute\Projector;
+use Patchlevel\EventSourcing\Attribute\Setup;
+use Patchlevel\EventSourcing\Attribute\Subscribe;
+use Patchlevel\EventSourcing\Attribute\Teardown;
+use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberUtil;
+use Patchlevel\EventSourcing\Tests\Integration\Subscription\Events\ProfileCreated;
+
+use function assert;
+
+#[Projector('profile_2')]
+final class ProfileNewProjection
+{
+    use SubscriberUtil;
+
+    public function __construct(
+        private Connection $connection,
+    ) {
+    }
+
+    #[Setup]
+    public function create(): void
+    {
+        $table = new Table($this->tableName());
+        $table->addColumn('id', 'string')->setLength(36);
+        $table->addColumn('firstname', 'string')->setLength(255);
+        $table->setPrimaryKey(['id']);
+
+        $this->connection->createSchemaManager()->createTable($table);
+    }
+
+    #[Teardown]
+    public function drop(): void
+    {
+        $this->connection->createSchemaManager()->dropTable($this->tableName());
+    }
+
+    #[Subscribe(ProfileCreated::class)]
+    public function handleProfileCreated(Message $message): void
+    {
+        $profileCreated = $message->event();
+
+        assert($profileCreated instanceof ProfileCreated);
+
+        $this->connection->executeStatement(
+            'INSERT INTO ' . $this->tableName() . ' (id, firstname) VALUES(:id, :firstname);',
+            [
+                'id' => $profileCreated->profileId->toString(),
+                'firstname' => $profileCreated->name,
+            ],
+        );
+    }
+
+    private function tableName(): string
+    {
+        return 'projection_' . $this->subscriberId();
+    }
+}