feat: delay step transition until all media is processed

src/DependencyInjection/migration.xml

@@ -323,6 +323,7 @@
             <argument type="service" id="SwagMigrationAssistant\Migration\Media\MediaFileProcessorRegistry"/>
             <argument type="service" id="SwagMigrationAssistant\Migration\Logging\LoggingService"/>
             <argument type="service" id="SwagMigrationAssistant\Migration\MigrationContextFactory"/>
+            <argument type="service" id="messenger.default_bus"/>
             <tag name="messenger.message_handler"/>
         </service>
 

src/Migration/MessageQueue/Handler/ProcessMediaHandler.php

@@ -19,12 +19,15 @@
 use SwagMigrationAssistant\Migration\Media\MediaFileProcessorInterface;
 use SwagMigrationAssistant\Migration\Media\MediaFileProcessorRegistryInterface;
 use SwagMigrationAssistant\Migration\Media\MediaProcessWorkloadStruct;
+use SwagMigrationAssistant\Migration\MessageQueue\Message\MigrationProcessMessage;
 use SwagMigrationAssistant\Migration\MessageQueue\Message\ProcessMediaMessage;
 use SwagMigrationAssistant\Migration\MigrationContextFactoryInterface;
 use SwagMigrationAssistant\Migration\MigrationContextInterface;
+use SwagMigrationAssistant\Migration\Run\MigrationProgress;
 use SwagMigrationAssistant\Migration\Run\SwagMigrationRunCollection;
 use SwagMigrationAssistant\Migration\Run\SwagMigrationRunEntity;
 use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+use Symfony\Component\Messenger\MessageBusInterface;
 
 /**
  * @internal
@@ -43,6 +46,7 @@ public function __construct(
         private readonly MediaFileProcessorRegistryInterface $mediaFileProcessorRegistry,
         private readonly LoggingServiceInterface $loggingService,
         private readonly MigrationContextFactoryInterface $migrationContextFactory,
+        private readonly MessageBusInterface $messageBus,
     ) {
     }
 
@@ -101,6 +105,12 @@ public function __invoke(ProcessMediaMessage $message): void
 
             $this->loggingService->saveLogging($context);
         }
+
+        if ($run->getProgress() !== null) {
+            $this->updateProgress($message, $run->getProgress(), $context);
+        }
+
+        $this->messageBus->dispatch(new MigrationProcessMessage($context, $migrationContext->getRunUuid()));
     }
 
     /**
@@ -128,4 +138,18 @@ private function processFailures(
             $workload = $processor->process($migrationContext, $context, $errorWorkload);
         }
     }
+
+    private function updateProgress(
+        ProcessMediaMessage $message,
+        MigrationProgress $progress,
+        Context $context
+    ): void {
+        $progress->setCurrentEntityProgress($progress->getCurrentEntityProgress() + \count($message->getMediaFileIds()));
+        $progress->setProgress($progress->getProgress() + \count($message->getMediaFileIds()));
+
+        $this->migrationRunRepo->update([[
+            'id' => $message->getRunId(),
+            'progress' => $progress->jsonSerialize(),
+        ]], $context);
+    }
 }

src/Migration/MessageQueue/Handler/Processor/MediaProcessingProcessor.php

@@ -9,6 +9,9 @@
 
 use Shopware\Core\Framework\Context;
 use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
+use Shopware\Core\Framework\DataAbstractionLayer\Search\Criteria;
+use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\EqualsFilter;
+use Shopware\Core\Framework\DataAbstractionLayer\Search\Filter\MultiFilter;
 use Shopware\Core\Framework\Log\Package;
 use SwagMigrationAssistant\Migration\Data\SwagMigrationDataCollection;
 use SwagMigrationAssistant\Migration\Media\SwagMigrationMediaFileCollection;
@@ -57,19 +60,33 @@ public function process(
         SwagMigrationRunEntity $run,
         MigrationProgress $progress,
     ): void {
-        $fileCount = $this->mediaFileProcessorService->processMediaFiles($migrationContext, $context);
+        $this->mediaFileProcessorService->processMediaFiles($migrationContext, $context);
 
-        if ($fileCount <= 0) {
+        if ($this->isAllMediaProcessed($context, $migrationContext->getRunUuid())) {
             $this->runTransitionService->transitionToRunStep($migrationContext->getRunUuid(), MigrationStep::CLEANUP);
             $this->updateProgress($migrationContext->getRunUuid(), $progress, $context);
             $this->bus->dispatch(new MigrationProcessMessage($context, $migrationContext->getRunUuid()));
-
-            return;
         }
+    }
+
+    private function isAllMediaProcessed(Context $context, string $runId): bool
+    {
+        $criteria = new Criteria();
+        $criteria->addFilter(
+            new EqualsFilter('runId', $runId)
+        );
+        $criteria->addFilter(
+            new MultiFilter(
+                MultiFilter::CONNECTION_AND,
+                [
+                    new EqualsFilter('processed', false),
+                    new EqualsFilter('processFailure', false),
+                ]
+            )
+        );
+
+        $unprocessedCount = $this->migrationMediaFileRepo->search($criteria, $context)->getTotal();
 
-        $progress->setCurrentEntityProgress($progress->getCurrentEntityProgress() + $fileCount);
-        $progress->setProgress($progress->getProgress() + $fileCount);
-        $this->updateProgress($migrationContext->getRunUuid(), $progress, $context);
-        $this->bus->dispatch(new MigrationProcessMessage($context, $migrationContext->getRunUuid()));
+        return $unprocessedCount === 0;
     }
 }

src/Migration/Service/MediaFileProcessorService.php

@@ -23,7 +23,7 @@
 #[Package('fundamentals@after-sales')]
 class MediaFileProcessorService implements MediaFileProcessorServiceInterface
 {
-    final public const MESSAGE_SIZE = 5;
+    final public const MESSAGE_SIZE = 20;
 
     public function __construct(
         private readonly MessageBusInterface $messageBus,
@@ -33,7 +33,7 @@ public function __construct(
     ) {
     }
 
-    public function processMediaFiles(MigrationContextInterface $migrationContext, Context $context): int
+    public function processMediaFiles(MigrationContextInterface $migrationContext, Context $context): void
     {
         $mediaFiles = $this->getMediaFiles($migrationContext);
 
@@ -53,16 +53,9 @@ public function processMediaFiles(MigrationContextInterface $migrationContext, C
 
             if ($currentDataSet::getEntity() !== $mediaFile['entity']) {
                 $this->addMessageToBus($migrationContext->getRunUuid(), $context, $currentDataSet, $messageMediaUuids);
+                $this->loggingService->saveLogging($context);
 
-                try {
-                    $messageMediaUuids = [];
-                    $currentCount = 0;
-                    $currentDataSet = $this->dataSetRegistry->getDataSet($migrationContext, $mediaFile['entity']);
-                } catch (DataSetNotFoundException $e) {
-                    $this->logDataSetNotFoundException($migrationContext, $mediaFile);
-
-                    continue;
-                }
+                return;
             }
 
             ++$currentCount;
@@ -73,17 +66,16 @@ public function processMediaFiles(MigrationContextInterface $migrationContext, C
             }
 
             $this->addMessageToBus($migrationContext->getRunUuid(), $context, $currentDataSet, $messageMediaUuids);
-            $messageMediaUuids = [];
-            $currentCount = 0;
+            $this->loggingService->saveLogging($context);
+
+            return;
         }
 
         if ($currentCount > 0 && $currentDataSet !== null) {
             $this->addMessageToBus($migrationContext->getRunUuid(), $context, $currentDataSet, $messageMediaUuids);
         }
 
         $this->loggingService->saveLogging($context);
-
-        return \count($mediaFiles);
     }
 
     /**

src/Migration/Service/MediaFileProcessorServiceInterface.php

@@ -14,5 +14,5 @@
 #[Package('fundamentals@after-sales')]
 interface MediaFileProcessorServiceInterface
 {
-    public function processMediaFiles(MigrationContextInterface $migrationContext, Context $context): int;
+    public function processMediaFiles(MigrationContextInterface $migrationContext, Context $context): void;
 }

src/Resources/app/administration/src/module/swag-migration/component/loading-screen/swag-migration-result-screen/swag-migration-result-screen.html.twig

@@ -21,9 +21,6 @@
                             <div class="swag-migration-result-screen__title">
                                 {{ $tc('swag-migration.index.loadingScreenCard.result.title') }}
                             </div>
-                            <div class="swag-migration-result-screen__caption">
-                                {{ $tc('swag-migration.index.loadingScreenCard.result.caption') }}
-                            </div>
                         {% endblock %}
                     </div>
                 {% endblock %}

src/Resources/app/administration/src/module/swag-migration/snippet/de.json

@@ -426,9 +426,8 @@
         },
         "result": {
           "title": "Der Migrations-Assistent ist fertig",
-          "caption": "Die Medien werden weiter im Hintergrund heruntergeladen. Große Dateien können etwas Zeit in Anspruch nehmen.",
           "logSummary": "Logbuch",
-          "historyHint": "Du kannst diese Informationen jederzeit in der Migrations-Historie einsehen. Es kann sein, dass zu einem späteren Zeitpunkt noch mehr Log-Nachrichten hinzukommen, weil der Mediendownload noch im Hintergrund läuft."
+          "historyHint": "Du kannst diese Informationen jederzeit in der Migrations-Historie einsehen."
         }
       },
       "confirmAbortDialog": {

src/Resources/app/administration/src/module/swag-migration/snippet/en.json

@@ -426,9 +426,8 @@
         },
         "result": {
           "title": "The Migration Assistant is done",
-          "caption": "The media download continues in the background. Large files can take some time.",
           "logSummary": "Logbook",
-          "historyHint": "You can view this information any time in the migration history. It is possible that more log messages will be added at a later time because the media download is still running in the background."
+          "historyHint": "You can view this information any time in the migration history."
         }
       },
       "confirmAbortDialog": {

tests/Migration/MessageQueue/Handler/ProcessMediaHandlerTest.php

@@ -0,0 +1,112 @@
+<?php declare(strict_types=1);
+/*
+ * (c) shopware AG <[email protected]>
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace SwagMigrationAssistant\Test\Migration\MessageQueue\Handler;
+
+use PHPUnit\Framework\TestCase;
+use Shopware\Core\Content\Product\ProductDefinition;
+use Shopware\Core\Framework\Context;
+use Shopware\Core\Framework\DataAbstractionLayer\EntityCollection;
+use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
+use Shopware\Core\Framework\DataAbstractionLayer\Search\Criteria;
+use Shopware\Core\Framework\DataAbstractionLayer\Search\EntitySearchResult;
+use Shopware\Core\Framework\Log\Package;
+use Shopware\Core\Framework\Uuid\Uuid;
+use Shopware\Core\Test\Stub\MessageBus\CollectingMessageBus;
+use SwagMigrationAssistant\Migration\Connection\SwagMigrationConnectionEntity;
+use SwagMigrationAssistant\Migration\Logging\LoggingServiceInterface;
+use SwagMigrationAssistant\Migration\Media\MediaFileProcessorInterface;
+use SwagMigrationAssistant\Migration\Media\MediaFileProcessorRegistryInterface;
+use SwagMigrationAssistant\Migration\MessageQueue\Handler\ProcessMediaHandler;
+use SwagMigrationAssistant\Migration\MessageQueue\Message\ProcessMediaMessage;
+use SwagMigrationAssistant\Migration\MigrationContext;
+use SwagMigrationAssistant\Migration\MigrationContextFactoryInterface;
+use SwagMigrationAssistant\Migration\Run\MigrationProgress;
+use SwagMigrationAssistant\Migration\Run\ProgressDataSetCollection;
+use SwagMigrationAssistant\Migration\Run\SwagMigrationRunDefinition;
+use SwagMigrationAssistant\Migration\Run\SwagMigrationRunEntity;
+use SwagMigrationAssistant\Profile\Shopware55\Shopware55Profile;
+
+#[Package('fundamentals@after-sales')]
+class ProcessMediaHandlerTest extends TestCase
+{
+    public function testProcess(): void
+    {
+        $mediaFileIds = [
+            Uuid::randomHex(),
+            Uuid::randomHex(),
+            Uuid::randomHex(),
+        ];
+
+        $connection = new SwagMigrationConnectionEntity();
+        $connection->setId(Uuid::randomHex());
+
+        $progress = new MigrationProgress(
+            100,
+            200,
+            new ProgressDataSetCollection(),
+            ProductDefinition::ENTITY_NAME,
+            10
+        );
+
+        $migrationRun = new SwagMigrationRunEntity();
+        $migrationRun->setId(Uuid::randomHex());
+        $migrationRun->setConnection($connection);
+        $migrationRun->setProgress($progress);
+
+        $repository = $this->createMock(EntityRepository::class);
+        $repository->method('search')->willReturn(
+            new EntitySearchResult(
+                SwagMigrationRunDefinition::ENTITY_NAME,
+                1,
+                new EntityCollection([$migrationRun]),
+                null,
+                new Criteria(),
+                Context::createDefaultContext()
+            )
+        );
+
+        $repository->expects(static::once())->method('update');
+
+        $processMediaMessage = new ProcessMediaMessage(
+            $mediaFileIds,
+            $migrationRun->getId(),
+            ProductDefinition::ENTITY_NAME,
+            Context::createDefaultContext()
+        );
+
+        $processor = $this->createMock(MediaFileProcessorInterface::class);
+        $processor->expects(static::once())->method('process')->willReturn([]);
+
+        $registry = $this->createMock(MediaFileProcessorRegistryInterface::class);
+        $registry->method('getProcessor')->willReturn($processor);
+
+        $logger = $this->createMock(LoggingServiceInterface::class);
+
+        $migrationContext = new MigrationContext(new Shopware55Profile());
+
+        $migrationContextFactory = $this->createMock(MigrationContextFactoryInterface::class);
+        $migrationContextFactory->method('create')->willReturn($migrationContext);
+
+        $messageBus = new CollectingMessageBus();
+
+        $handler = new ProcessMediaHandler(
+            $repository,
+            $registry,
+            $logger,
+            $migrationContextFactory,
+            $messageBus
+        );
+
+        $handler->__invoke($processMediaMessage);
+
+        static::assertSame(103, $migrationRun->getProgress()?->getProgress());
+        static::assertSame(13, $migrationRun->getProgress()->getCurrentEntityProgress());
+
+        static::assertCount(1, $messageBus->getMessages());
+    }
+}

tests/Migration/MessageQueue/Handler/Processor/MediaProcessingProcessorTest.php

@@ -25,8 +25,6 @@
 use SwagMigrationAssistant\Migration\Service\MediaFileProcessorService;
 use SwagMigrationAssistant\Profile\Shopware55\Shopware55Profile;
 
-use function PHPUnit\Framework\once;
-
 #[Package('fundamentals@after-sales')]
 class MediaProcessingProcessorTest extends TestCase
 {
@@ -71,7 +69,7 @@ public function testProcessingWithoutMediaFiles(): void
 
         $runTransitionService = $this->createMock(RunTransitionServiceInterface::class);
         $runTransitionService
-            ->expects(once())
+            ->expects(static::once())
             ->method('transitionToRunStep')
             ->with($run->getId(), MigrationStep::CLEANUP);
 
@@ -116,23 +114,16 @@ public function testProcessing(): void
 
         $migrationContext = new MigrationContext(new Shopware55Profile(), $connection, $run->getId());
 
-        $runTransitionService = $this->createMock(RunTransitionServiceInterface::class);
-        $runTransitionService
-            ->expects(static::never())
-            ->method('transitionToRunStep')
-            ->with($run->getId(), MigrationStep::CLEANUP);
-
         $mediaFileProcessorService = $this->createMock(MediaFileProcessorService::class);
         $mediaFileProcessorService
             ->expects(static::once())
-            ->method('processMediaFiles')
-            ->willReturn(100);
+            ->method('processMediaFiles');
 
         $this->processor = new MediaProcessingProcessor(
             $this->createMock(EntityRepository::class),
             $this->createMock(EntityRepository::class),
             $this->createMock(EntityRepository::class),
-            $runTransitionService,
+            $this->createMock(RunTransitionServiceInterface::class),
             $mediaFileProcessorService,
             $this->bus
         );