From a935088216a45b72f73eadd1d520e3c8b23f20a0 Mon Sep 17 00:00:00 2001
From: Daniel Badura <daniel.badura@patchlevel.de>
Date: Mon, 29 Dec 2025 09:23:21 +0100
Subject: [PATCH] Add `StoreMigrateCommand` from integration repos to ease
 maintainance

---
 src/Console/Command/StoreMigrateCommand.php   |  88 +++++
 .../Command/StoreMigrateCommandTest.php       | 330 ++++++++++++++++++
 2 files changed, 418 insertions(+)
 create mode 100644 src/Console/Command/StoreMigrateCommand.php
 create mode 100644 tests/Unit/Console/Command/StoreMigrateCommandTest.php

diff --git a/src/Console/Command/StoreMigrateCommand.php b/src/Console/Command/StoreMigrateCommand.php
new file mode 100644
index 00000000..b91daa3d
--- /dev/null
+++ b/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;
+    }
+}
diff --git a/tests/Unit/Console/Command/StoreMigrateCommandTest.php b/tests/Unit/Console/Command/StoreMigrateCommandTest.php
new file mode 100644
index 00000000..15e497f6
--- /dev/null
+++ b/tests/Unit/Console/Command/StoreMigrateCommandTest.php
@@ -0,0 +1,330 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Unit\Console\Command;
+
+use Patchlevel\EventSourcing\Console\Command\StoreMigrateCommand;
+use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Message\Translator\ExcludeEventTranslator;
+use Patchlevel\EventSourcing\Store\InMemoryStore;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Output\BufferedOutput;
+
+use function iterator_to_array;
+
+#[CoversClass(StoreMigrateCommand::class)]
+final class StoreMigrateCommandTest extends TestCase
+{
+    public function testNoMessages(): void
+    {
+        $fromStore = new InMemoryStore();
+        $toStore = new InMemoryStore();
+
+        $command = new StoreMigrateCommand($fromStore, $toStore, []);
+
+        $input = new ArrayInput([]);
+        $output = new BufferedOutput();
+
+        $exitCode = $command->run($input, $output);
+
+        self::assertSame(0, $exitCode);
+
+        $content = $output->fetch();
+
+        self::assertStringContainsString('Migration initialization...', $content);
+        self::assertStringContainsString('0', $content);
+        self::assertStringContainsString('Migration finished', $content);
+    }
+
+    public function testOneMessage(): void
+    {
+        $fromStore = new InMemoryStore([
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('1'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+        ]);
+        $toStore = new InMemoryStore();
+
+        $command = new StoreMigrateCommand($fromStore, $toStore, []);
+
+        $input = new ArrayInput([]);
+        $output = new BufferedOutput();
+
+        $exitCode = $command->run($input, $output);
+
+        self::assertSame(0, $exitCode);
+
+        $content = $output->fetch();
+
+        self::assertStringContainsString('Migration initialization...', $content);
+        self::assertStringContainsString('1', $content);
+        self::assertStringContainsString('Migration finished', $content);
+
+        self::assertCount(1, iterator_to_array($toStore->load()->getIterator()));
+    }
+
+    public function testTenMessages(): void
+    {
+        $fromStore = new InMemoryStore([
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('1'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('2'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('3'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('4'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('5'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('6'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('7'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('8'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('9'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('10'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+        ]);
+        $toStore = new InMemoryStore();
+
+        $command = new StoreMigrateCommand($fromStore, $toStore, []);
+
+        $input = new ArrayInput([]);
+        $output = new BufferedOutput();
+
+        $exitCode = $command->run($input, $output);
+
+        self::assertSame(0, $exitCode);
+
+        $content = $output->fetch();
+
+        self::assertStringContainsString('Migration initialization...', $content);
+        self::assertStringContainsString('10', $content);
+        self::assertStringContainsString('Migration finished', $content);
+
+        self::assertCount(10, iterator_to_array($toStore->load()->getIterator()));
+    }
+
+    public function testTenMessagesWithBufferAt2(): void
+    {
+        $fromStore = new InMemoryStore([
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('1'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('2'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('3'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('4'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('5'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('6'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('7'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('8'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('9'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('10'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+        ]);
+        $toStore = new InMemoryStore();
+
+        $command = new StoreMigrateCommand($fromStore, $toStore, []);
+
+        $input = new ArrayInput(['--buffer' => 10]);
+        $output = new BufferedOutput();
+
+        $exitCode = $command->run($input, $output);
+
+        self::assertSame(0, $exitCode);
+
+        $content = $output->fetch();
+
+        self::assertStringContainsString('Migration initialization...', $content);
+        self::assertStringContainsString('10', $content);
+        self::assertStringContainsString('Migration finished', $content);
+
+        self::assertCount(10, iterator_to_array($toStore->load()->getIterator()));
+    }
+
+    public function testTenMessagesWithDroppingTranslator(): void
+    {
+        $fromStore = new InMemoryStore([
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('1'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('2'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('3'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('4'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('5'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('6'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('7'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('8'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileCreated(
+                    ProfileId::fromString('9'),
+                    Email::fromString('info@patchlevel.de'),
+                ),
+            ),
+            new Message(
+                new ProfileVisited(
+                    ProfileId::fromString('1'),
+                ),
+            ),
+        ]);
+        $toStore = new InMemoryStore();
+
+        $command = new StoreMigrateCommand(
+            $fromStore,
+            $toStore,
+            [new ExcludeEventTranslator([ProfileCreated::class])],
+        );
+
+        $input = new ArrayInput([]);
+        $output = new BufferedOutput();
+
+        $exitCode = $command->run($input, $output);
+
+        self::assertSame(0, $exitCode);
+
+        $content = $output->fetch();
+
+        self::assertStringContainsString('Migration initialization...', $content);
+        self::assertStringContainsString('10', $content);
+        self::assertStringContainsString('Migration finished', $content);
+
+        self::assertCount(1, iterator_to_array($toStore->load()->getIterator()));
+    }
+}