Merge release 3.15.0 into 4.0.x

composer.json

@@ -38,7 +38,7 @@
   "require-dev": {
     "ext-pdo_sqlite": "~8.2.0 || ~8.3.0 || ~8.4.0 || ~8.5.0",
     "doctrine/orm": "^2.18.0 || ^3.0.0",
-    "infection/infection": "^0.31.9",
+    "infection/infection": "^0.32.0",
     "league/commonmark": "^2.6.1",
     "patchlevel/coding-standard": "^1.3.0",
     "patchlevel/event-sourcing-phpstan-extension": "dev-event-sourcing-4.0",

docs/getting-started.md

@@ -163,12 +163,12 @@ use Patchlevel\EventSourcing\Attribute\Projector;
 use Patchlevel\EventSourcing\Attribute\Setup;
 use Patchlevel\EventSourcing\Attribute\Subscribe;
 use Patchlevel\EventSourcing\Attribute\Teardown;
-use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberUtil;
 
-#[Projector('hotel')]
+#[Projector(self::TABLE)]
 final class HotelProjector
 {
-    use SubscriberUtil;
+    // use a const for easier access in the projector & to keep projector id and table name in sync
+    private const TABLE = 'hotel';
 
     public function __construct(
         private readonly Connection $db,
@@ -178,14 +178,14 @@ final class HotelProjector
     /** @return list<array{id: string, name: string, guests: int}> */
     public function getHotels(): array
     {
-        return $this->db->fetchAllAssociative("SELECT id, name, guests FROM {$this->table()};");
+        return $this->db->fetchAllAssociative(sprintf('SELECT id, name, guests FROM %s;'), self::TABLE);
     }
 
     #[Subscribe(HotelCreated::class)]
     public function handleHotelCreated(HotelCreated $event): void
     {
         $this->db->insert(
-            $this->table(),
+            self::TABLE,
             [
                 'id' => $event->hotelId->toString(),
                 'name' => $event->hotelName,
@@ -198,7 +198,7 @@ final class HotelProjector
     public function handleGuestIsCheckedIn(GuestIsCheckedIn $event): void
     {
         $this->db->executeStatement(
-            "UPDATE {$this->table()} SET guests = guests + 1 WHERE id = ?;",
+            sprintf('UPDATE %s SET guests = guests + 1 WHERE id = ?;', self::TABLE),
             [$event->hotelId->toString()],
         );
     }
@@ -207,26 +207,21 @@ final class HotelProjector
     public function handleGuestIsCheckedOut(GuestIsCheckedOut $event): void
     {
         $this->db->executeStatement(
-            "UPDATE {$this->table()} SET guests = guests - 1 WHERE id = ?;",
+            sprintf('UPDATE %s SET guests = guests - 1 WHERE id = ?;', self::TABLE),
             [$event->hotelId->toString()],
         );
     }
 
     #[Setup]
     public function create(): void
     {
-        $this->db->executeStatement("CREATE TABLE IF NOT EXISTS {$this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR, guests INTEGER);");
+        $this->db->executeStatement(sprintf('CREATE TABLE IF NOT EXISTS %s (id VARCHAR PRIMARY KEY, name VARCHAR, guests INTEGER);', self::TABLE));
     }
 
     #[Teardown]
     public function drop(): void
     {
-        $this->db->executeStatement("DROP TABLE IF EXISTS {$this->table()};");
-    }
-
-    private function table(): string
-    {
-        return 'projection_' . $this->subscriberId();
+        $this->db->executeStatement(sprintf('DROP TABLE IF EXISTS %s;', self::TABLE));
     }
 }
 ```

docs/subscription.md

@@ -234,8 +234,6 @@ use Patchlevel\EventSourcing\Subscription\Lookup;
 #[Projector('public_profile')]
 final class PublicProfileProjection
 {
-    use SubscriberUtil;
-
     // ... constructor
 
     #[Subscribe(Published::class)]
@@ -310,32 +308,26 @@ use Doctrine\DBAL\Connection;
 use Patchlevel\EventSourcing\Attribute\Projector;
 use Patchlevel\EventSourcing\Attribute\Setup;
 use Patchlevel\EventSourcing\Attribute\Teardown;
-use Patchlevel\EventSourcing\Subscription\Subscriber\SubscriberUtil;
 
-#[Projector('profile_1')]
+#[Projector(self::TABLE)]
 final class ProfileProjector
 {
-    use SubscriberUtil;
+    private const TABLE = 'profile_v1';
 
     private Connection $connection;
 
     #[Setup]
     public function create(): void
     {
         $this->connection->executeStatement(
-            "CREATE TABLE IF NOT EXISTS {$this->table()} (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL);",
+            sprintf('CREATE TABLE IF NOT EXISTS %s (id VARCHAR PRIMARY KEY, name VARCHAR NOT NULL);', self::TABLE),
         );
     }
 
     #[Teardown]
     public function drop(): void
     {
-        $this->connection->executeStatement("DROP TABLE IF EXISTS {$this->table()};");
-    }
-
-    private function table(): string
-    {
-        return 'projection_' . $this->subscriberId();
+        $this->connection->executeStatement(sprintf('DROP TABLE IF EXISTS %s;', self::TABLE));
     }
 }
 ```
@@ -345,12 +337,11 @@ MySQL and MariaDB don't support transactions for DDL statements.
 So you must use a different database connection in your projectors, 
 otherwise you will get an error when the subscription tries to create the table.
 :::
-    
+
 :::warning
 If you change the subscriber id, you must also change the table/collection name.
 The subscription engine will create a new subscription with the new subscriber id.
 That means the setup method will be called again and the table/collection will conflict with the old existing projection.
-You can use the `SubscriberUtil` to build the table/collection name.
 :::
 
 :::note

phpstan-baseline.neon

@@ -42,12 +42,6 @@ parameters:
 			count: 1
 			path: src/Message/Serializer/DefaultHeadersSerializer.php
 
-		-
-			message: '#^Call to an undefined method Patchlevel\\EventSourcing\\Store\\Store\:\:archive\(\)\.$#'
-			identifier: method.notFound
-			count: 1
-			path: src/Repository/DefaultRepository.php
-
 		-
 			message: '#^Property Patchlevel\\EventSourcing\\Serializer\\Normalizer\\IdNormalizer\:\:\$identifierClass \(class\-string\<Patchlevel\\EventSourcing\\Identifier\\Identifier\>\|null\) does not accept string\.$#'
 			identifier: assign.propertyType
@@ -168,6 +162,24 @@ parameters:
 			count: 3
 			path: src/Subscription/ThrowableToErrorContextTransformer.php
 
+		-
+			message: '#^Cannot access offset ''name'' on array\<string, mixed\>\|false\.$#'
+			identifier: offsetAccess.nonOffsetAccessible
+			count: 1
+			path: tests/Benchmark/BasicImplementation/Projection/ProfileProjector.php
+
+		-
+			message: '#^Method Patchlevel\\EventSourcing\\Tests\\Benchmark\\BasicImplementation\\Projection\\ProfileProjector\:\:getProfileName\(\) should return string but returns mixed\.$#'
+			identifier: return.type
+			count: 1
+			path: tests/Benchmark/BasicImplementation/Projection/ProfileProjector.php
+
+		-
+			message: '#^Parameter \#1 \$messageLoader of class Patchlevel\\EventSourcing\\Subscription\\Engine\\DefaultSubscriptionEngine constructor expects Patchlevel\\EventSourcing\\Subscription\\Engine\\MessageLoader, Patchlevel\\EventSourcing\\Store\\StreamDoctrineDbalStore given\.$#'
+			identifier: argument.type
+			count: 1
+			path: tests/Benchmark/CommandToQueryBench.php
+
 		-
 			message: '#^Call to static method PHPUnit\\Framework\\Assert\:\:assertInstanceOf\(\) with ''Patchlevel\\\\EventSourcing\\\\Tests\\\\Integration\\\\BankAccountSplitStream\\\\BankAccount'' and Patchlevel\\EventSourcing\\Tests\\Integration\\BankAccountSplitStream\\BankAccount will always evaluate to true\.$#'
 			identifier: staticMethod.alreadyNarrowedType
@@ -216,6 +228,42 @@ parameters:
 			count: 1
 			path: tests/Integration/BasicImplementation/BasicIntegrationTest.php
 
+		-
+			message: '#^Instantiated class Patchlevel\\EventSourcing\\Tests\\Integration\\BasicImplementation\\DoctrineDbalStore not found\.$#'
+			identifier: class.notFound
+			count: 1
+			path: tests/Integration/BasicImplementation/BasicIntegrationTest.php
+
+		-
+			message: '#^Parameter \#1 \$messageLoader of class Patchlevel\\EventSourcing\\Subscription\\Engine\\DefaultSubscriptionEngine constructor expects Patchlevel\\EventSourcing\\Subscription\\Engine\\MessageLoader, Patchlevel\\EventSourcing\\Tests\\Integration\\BasicImplementation\\DoctrineDbalStore given\.$#'
+			identifier: argument.type
+			count: 1
+			path: tests/Integration/BasicImplementation/BasicIntegrationTest.php
+
+		-
+			message: '#^Parameter \#2 \$schemaConfigurator of class Patchlevel\\EventSourcing\\Schema\\DoctrineSchemaDirector constructor expects Patchlevel\\EventSourcing\\Schema\\DoctrineSchemaConfigurator, Patchlevel\\EventSourcing\\Tests\\Integration\\BasicImplementation\\DoctrineDbalStore given\.$#'
+			identifier: argument.type
+			count: 1
+			path: tests/Integration/BasicImplementation/BasicIntegrationTest.php
+
+		-
+			message: '#^Parameter \#2 \$store of class Patchlevel\\EventSourcing\\Repository\\DefaultRepositoryManager constructor expects Patchlevel\\EventSourcing\\Store\\Store, Patchlevel\\EventSourcing\\Tests\\Integration\\BasicImplementation\\DoctrineDbalStore given\.$#'
+			identifier: argument.type
+			count: 1
+			path: tests/Integration/BasicImplementation/BasicIntegrationTest.php
+
+		-
+			message: '#^Cannot access offset ''name'' on array\<string, mixed\>\|false\.$#'
+			identifier: offsetAccess.nonOffsetAccessible
+			count: 1
+			path: tests/Integration/BasicImplementation/Projection/ProfileProjector.php
+
+		-
+			message: '#^Method Patchlevel\\EventSourcing\\Tests\\Integration\\BasicImplementation\\Projection\\ProfileProjector\:\:getProfileName\(\) should return string but returns mixed\.$#'
+			identifier: return.type
+			count: 1
+			path: tests/Integration/BasicImplementation/Projection/ProfileProjector.php
+
 		-
 			message: '#^Call to static method PHPUnit\\Framework\\Assert\:\:assertInstanceOf\(\) with ''Patchlevel\\\\EventSourcing\\\\Tests\\\\Integration\\\\MicroAggregate\\\\Profile'' and Patchlevel\\EventSourcing\\Tests\\Integration\\MicroAggregate\\Profile will always evaluate to true\.$#'
 			identifier: staticMethod.alreadyNarrowedType
@@ -276,12 +324,6 @@ parameters:
 			count: 1
 			path: tests/Integration/Subscription/SubscriptionTest.php
 
-		-
-			message: '#^Cannot use array destructuring on list\<mixed\>\|null\.$#'
-			identifier: offsetAccess.nonArray
-			count: 1
-			path: tests/ReturnCallback.php
-
 		-
 			message: '#^Call to static method PHPUnit\\Framework\\Assert\:\:assertInstanceOf\(\) with ''Patchlevel\\\\EventSourcing\\\\Metadata\\\\AggregateRoot\\\\AggregateRootMetadata'' and Patchlevel\\EventSourcing\\Metadata\\AggregateRoot\\AggregateRootMetadata\<Patchlevel\\EventSourcing\\Aggregate\\BasicAggregateRoot\> will always evaluate to true\.$#'
 			identifier: staticMethod.alreadyNarrowedType

src/Console/Command/StoreMigrateCommand.php

@@ -0,0 +1,88 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Console\Command;
+
+use Patchlevel\EventSourcing\Console\InputHelper;
+use Patchlevel\EventSourcing\Console\OutputStyle;
+use Patchlevel\EventSourcing\Message\Pipe;
+use Patchlevel\EventSourcing\Message\Translator\Translator;
+use Patchlevel\EventSourcing\Store\Store;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+use function count;
+
+#[AsCommand(
+    'event-sourcing:store:migrate',
+    'migrate events from one store to another',
+)]
+final class StoreMigrateCommand extends Command
+{
+    /** @param iterable<int, Translator> $translators */
+    public function __construct(
+        private readonly Store $store,
+        private readonly Store $newStore,
+        private readonly iterable $translators = [],
+    ) {
+        parent::__construct();
+    }
+
+    protected function configure(): void
+    {
+        $this
+            ->addOption(
+                'buffer',
+                null,
+                InputOption::VALUE_REQUIRED,
+                'How many messages should be buffered',
+                1_000,
+            );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $buffer = InputHelper::positiveIntOrZero($input->getOption('buffer'));
+        $style = new OutputStyle($input, $output);
+
+        $style->info('Migration initialization...');
+
+        $count = $this->store->count();
+        $messages = $this->store->load();
+
+        $style->progressStart($count);
+
+        $bufferedMessages = [];
+
+        $pipe = new Pipe(
+            $messages,
+            ...$this->translators,
+        );
+
+        foreach ($pipe as $message) {
+            $bufferedMessages[] = $message;
+
+            if (count($bufferedMessages) < $buffer) {
+                continue;
+            }
+
+            $this->newStore->save(...$bufferedMessages);
+            $bufferedMessages = [];
+            $style->progressAdvance($buffer);
+        }
+
+        if (count($bufferedMessages) !== 0) {
+            $this->newStore->save(...$bufferedMessages);
+            $style->progressAdvance(count($bufferedMessages));
+        }
+
+        $style->progressFinish();
+        $style->success('Migration finished');
+
+        return 0;
+    }
+}

src/Subscription/Engine/SubscriptionManager.php

@@ -77,48 +77,48 @@ public function find(SubscriptionCriteria $criteria): array
     public function add(Subscription ...$subscriptions): void
     {
         foreach ($subscriptions as $sub) {
-            $this->forAdd->attach($sub);
+            $this->forAdd->offsetSet($sub);
         }
     }
 
     public function update(Subscription ...$subscriptions): void
     {
         foreach ($subscriptions as $sub) {
-            $this->forUpdate->attach($sub);
+            $this->forUpdate->offsetSet($sub);
         }
     }
 
     public function remove(Subscription ...$subscriptions): void
     {
         foreach ($subscriptions as $sub) {
-            $this->forRemove->attach($sub);
+            $this->forRemove->offsetSet($sub);
         }
     }
 
     public function flush(): void
     {
         foreach ($this->forAdd as $subscription) {
-            if ($this->forRemove->contains($subscription)) {
+            if ($this->forRemove->offsetExists($subscription)) {
                 continue;
             }
 
             $this->subscriptionStore->add($subscription);
         }
 
         foreach ($this->forUpdate as $subscription) {
-            if ($this->forAdd->contains($subscription)) {
+            if ($this->forAdd->offsetExists($subscription)) {
                 continue;
             }
 
-            if ($this->forRemove->contains($subscription)) {
+            if ($this->forRemove->offsetExists($subscription)) {
                 continue;
             }
 
             $this->subscriptionStore->update($subscription);
         }
 
         foreach ($this->forRemove as $subscription) {
-            if ($this->forAdd->contains($subscription)) {
+            if ($this->forAdd->offsetExists($subscription)) {
                 continue;
             }
 

src/Subscription/Subscriber/SubscriberHelper.php

@@ -8,6 +8,7 @@
 use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadata;
 use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadataFactory;
 
+/** @deprecated since 3.15.0 will be removed with 4.0.0 */
 final class SubscriberHelper
 {
     public function __construct(

src/Subscription/Subscriber/SubscriberUtil.php

@@ -7,6 +7,7 @@
 use Patchlevel\EventSourcing\Metadata\Subscriber\AttributeSubscriberMetadataFactory;
 use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadataFactory;
 
+/** @deprecated since 3.15.0 will be removed with 4.0.0 */
 trait SubscriberUtil
 {
     private static SubscriberMetadataFactory|null $metadataFactory = null;