feat: split message queue process

src/Controller/StatusController.php

@@ -397,4 +397,17 @@ public function isResettingChecksums(Context $context): JsonResponse
             $settings->isResettingChecksums()
         );
     }
+
+    #[Route(
+        path: '/api/_action/migration/resume-after-fixes',
+        name: 'api.admin.migration.resume-after-fixes',
+        defaults: ['_acl' => ['admin']],
+        methods: [Request::METHOD_POST]
+    )]
+    public function resumeAfterFixes(Context $context): Response
+    {
+        $this->runService->resumeAfterFixes($context);
+
+        return new Response(null, Response::HTTP_NO_CONTENT);
+    }
 }

src/Exception/MigrationException.php

@@ -106,6 +106,8 @@ class MigrationException extends HttpException
 
     public const MISSING_MIGRATION_FIX_KEY = 'SWAG_MIGRATION__MISSING_MIGRATION_FIX_KEY';
 
+    public const MIGRATION_NOT_IN_STEP = 'SWAG_MIGRATION__MIGRATION_NOT_IN_STEP';
+
     public static function associationEntityRequiredMissing(string $entity, string $missingEntity): self
     {
         return new AssociationEntityRequiredMissingException(
@@ -557,4 +559,14 @@ public static function couldNotConvertFix(string $missingKey): self
             ['missingKey' => $missingKey]
         );
     }
+
+    public static function migrationNotInStep(string $runUuid, string $step): self
+    {
+        return new NoRunningMigrationException(
+            Response::HTTP_BAD_REQUEST,
+            self::MIGRATION_NOT_IN_STEP,
+            'Migration with id: "{{ runUuid }}" is not in step "{{ step }}".',
+            ['runUuid' => $runUuid, 'step' => $step]
+        );
+    }
 }

src/Migration/MessageQueue/Handler/MigrationProcessHandler.php

@@ -54,7 +54,7 @@ public function __invoke(MigrationProcessMessage $message): void
         }
 
         $processor = $this->processorRegistry->getProcessor($run->getStep());
-        $processor->process($migrationContext, $context, $run, $progress);
+        $processor?->process($migrationContext, $context, $run, $progress);
     }
 
     private function getCurrentRun(MigrationProcessMessage $message, Context $context): SwagMigrationRunEntity

src/Migration/MessageQueue/Handler/MigrationProcessorRegistry.php

@@ -25,8 +25,12 @@ public function __construct(private readonly iterable $processors)
     {
     }
 
-    public function getProcessor(MigrationStep $step): MigrationProcessorInterface
+    public function getProcessor(MigrationStep $step): ?MigrationProcessorInterface
     {
+        if (!$step->needsProcessor()) {
+            return null;
+        }
+
         foreach ($this->processors as $processor) {
             if ($processor->supports($step)) {
                 return $processor;

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

@@ -63,7 +63,7 @@ protected function changeProgressToNextEntity(SwagMigrationRunEntity $run, Migra
 
         if ($nextEntity === null && $run->getStep() === MigrationStep::FETCHING) {
             $nextEntity = \current($dataSets);
-            $this->runTransitionService->transitionToRunStep($run->getId(), MigrationStep::WRITING);
+            $this->runTransitionService->transitionToRunStep($run->getId(), MigrationStep::APPLY_FIXES);
             $progress->setProgress(0);
             $progress->setTotal($this->getWriteTotal($context));
         } elseif ($nextEntity === null && $run->getStep() === MigrationStep::WRITING) {

src/Migration/Run/MigrationStep.php

@@ -16,6 +16,8 @@ enum MigrationStep: string
 
     case FETCHING = 'fetching';
 
+    case APPLY_FIXES = 'apply-fixes';
+
     case WRITING = 'writing';
 
     case MEDIA_PROCESSING = 'media-processing';
@@ -39,4 +41,12 @@ public function isRunning(): bool
             self::ABORTED,
         ], true);
     }
+
+    public function needsProcessor(): bool
+    {
+        return !\in_array($this, [
+            self::APPLY_FIXES,
+            self::WAITING_FOR_APPROVE,
+        ], true);
+    }
 }

src/Migration/Run/RunService.php

@@ -276,6 +276,25 @@ public function assignThemeToSalesChannel(string $runUuid, Context $context): vo
         $this->loggingService->saveLogging($context);
     }
 
+    public function resumeAfterFixes(Context $context): void
+    {
+        $run = $this->getActiveRun($context);
+
+        if ($run === null) {
+            throw MigrationException::noRunningMigration();
+        }
+
+        $runId = $run->getId();
+
+        if ($run->getStepValue() !== MigrationStep::APPLY_FIXES->value) {
+            throw MigrationException::migrationNotInStep($runId, MigrationStep::APPLY_FIXES->value);
+        }
+
+        $this->runTransitionService->transitionToRunStep($runId, MigrationStep::WRITING);
+
+        $this->bus->dispatch(new MigrationProcessMessage($context, $runId));
+    }
+
     /**
      * @param array<int, string> $dataSelectionIds
      */

src/Migration/Run/RunServiceInterface.php

@@ -19,10 +19,10 @@ interface RunServiceInterface
      * If no migration run is running, it returns the progress with the step status IDLE.
      *
      * After starting the migration run, the steps are as follows, if the migration run is not aborted:
-     * IDLE -> FETCHING -> WRITING -> MEDIA_PROCESSING -> CLEANUP -> INDEXING -> WAITING_FOR_APPROVE -> IDLE
+     * IDLE -> FETCHING -> APPLY_FIXES -> WRITING -> MEDIA_PROCESSING -> CLEANUP -> INDEXING -> WAITING_FOR_APPROVE -> IDLE
      *
      * If the migration run is aborted, the steps are as follows:
-     * IDLE -> [FETCHING || WRITING || MEDIA_PROCESSING] -> ABORTING -> CLEANUP -> INDEXING -> IDLE
+     * IDLE -> [FETCHING || APPLY_FIXES || WRITING || MEDIA_PROCESSING] -> ABORTING -> CLEANUP -> INDEXING -> IDLE
      */
     public function getRunStatus(Context $context): MigrationState;
 
@@ -51,4 +51,6 @@ public function updateConnectionCredentials(Context $context, string $connection
     public function approveFinishingMigration(Context $context): void;
 
     public function assignThemeToSalesChannel(string $runUuid, Context $context): void;
+
+    public function resumeAfterFixes(Context $context): void;
 }

src/Resources/app/administration/src/core/service/api/swag-migration.api.service.ts

@@ -28,6 +28,7 @@ export const MIGRATION_API_SERVICE = 'migrationApiService';
 export const MIGRATION_STEP = {
     IDLE: 'idle',
     FETCHING: 'fetching',
+    APPLY_FIXES: 'apply-fixes',
     WRITING: 'writing',
     MEDIA_PROCESSING: 'media-processing',
     CLEANUP: 'cleanup',
@@ -247,6 +248,26 @@ export default class MigrationApiService extends ApiService {
             });
     }
 
+    async resumeMigrationAfterFixes(): Promise<ApiResponse<unknown>> {
+        // @ts-ignore
+        const headers = this.getBasicHeaders();
+
+        // @ts-ignore
+        return this.httpClient
+            .post(
+                // @ts-ignore
+                `_action/${this.getApiBasePath()}/resume-after-fixes`,
+                {},
+                {
+                    ...this.basicConfig,
+                    headers,
+                },
+            )
+            .then((response: AxiosResponse) => {
+                return ApiService.handleResponse(response);
+            });
+    }
+
     async getProfiles(): Promise<MigrationProfile[]> {
         // @ts-ignore
         const headers = this.getBasicHeaders();

src/Resources/app/administration/src/module/swag-migration/component/loading-screen/index.ts

@@ -4,3 +4,4 @@
  */
 Shopware.Component.register('swag-migration-loading-screen', () => import('./swag-migration-loading-screen'));
 Shopware.Component.register('swag-migration-result-screen', () => import('./swag-migration-result-screen'));
+Shopware.Component.register('swag-migration-apply-fixes', () => import('./swag-migration-apply-fixes'));

src/Resources/app/administration/src/module/swag-migration/component/loading-screen/swag-migration-apply-fixes/index.ts

@@ -0,0 +1,20 @@
+import template from './swag-migration-apply-fixes.html.twig';
+import { MIGRATION_API_SERVICE } from '../../../../../core/service/api/swag-migration.api.service';
+
+/**
+ * @private
+ * @sw-package fundamentals@after-sales
+ */
+export default Shopware.Component.wrapComponentConfig({
+    template,
+
+    inject: [
+        MIGRATION_API_SERVICE,
+    ],
+
+    methods: {
+        async resumeMigration() {
+            await this.migrationApiService.resumeMigrationAfterFixes();
+        },
+    },
+});

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

@@ -0,0 +1,32 @@
+{% block swag_migration_apply_fixes %}
+    <div class="swag-migration-loading-screen">
+        {% block swag_migration_apply_fixes_card %}
+            <mt-card
+                class="swag-migration-loading-screen__card"
+                position-identifier="swag-migration-loading-screen__card"
+            >
+                {% block swag_migration_apply_fixes_content %}
+                    <div class="swag-migration-loading-screen__content">
+                        {% block swag_migration_apply_fixes_header %}
+                            <div class="swag-migration-loading-screen__header">
+                                {% block swag_migration_apply_fixes_title %}
+                                    <div class="swag-migration-loading-screen__title">
+                                        {{ $tc('swag-migration.index.loadingScreenCard.applyFixes.title') }}
+                                    </div>
+                                {% endblock %}
+                            </div>
+                        {% endblock %}
+
+                        <mt-button
+                            size="small"
+                            variant="secondary"
+                            @click="resumeMigration"
+                        >
+                            {{ $tc('swag-migration.index.loadingScreenCard.applyFixes.resumeButton') }}
+                        </mt-button>
+                    </div>
+                {% endblock %}
+            </mt-card>
+        {% endblock %}
+    </div>
+{% endblock %}

src/Resources/app/administration/src/module/swag-migration/page/swag-migration-process-screen/index.ts

@@ -13,17 +13,19 @@ const MIGRATION_STATE_POLLING_INTERVAL = 1000 as const; // 1 second
 const MIGRATION_STEP_DISPLAY_INDEX = {
     [MIGRATION_STEP.IDLE]: 0,
     [MIGRATION_STEP.FETCHING]: 0,
-    [MIGRATION_STEP.WRITING]: 1,
-    [MIGRATION_STEP.MEDIA_PROCESSING]: 2,
-    [MIGRATION_STEP.ABORTING]: 3,
-    [MIGRATION_STEP.CLEANUP]: 3,
-    [MIGRATION_STEP.INDEXING]: 4,
-    [MIGRATION_STEP.WAITING_FOR_APPROVE]: 5,
+    [MIGRATION_STEP.APPLY_FIXES]: 1,
+    [MIGRATION_STEP.WRITING]: 2,
+    [MIGRATION_STEP.MEDIA_PROCESSING]: 3,
+    [MIGRATION_STEP.ABORTING]: 4,
+    [MIGRATION_STEP.CLEANUP]: 4,
+    [MIGRATION_STEP.INDEXING]: 5,
+    [MIGRATION_STEP.WAITING_FOR_APPROVE]: 6,
 } as const;
 
 const UI_COMPONENT_INDEX = {
     LOADING_SCREEN: 0,
-    RESULT_SUCCESS: 1,
+    APPLY_FIXES: 1,
+    RESULT_SUCCESS: 2,
 } as const;
 
 /**
@@ -213,6 +215,9 @@ export default Shopware.Component.wrapComponentConfig({
             ) {
                 this.componentIndex = UI_COMPONENT_INDEX.LOADING_SCREEN;
                 this.flowChartItemIndex = MIGRATION_STEP_DISPLAY_INDEX[state.step];
+            } else if (state.step === MIGRATION_STEP.APPLY_FIXES) {
+                this.componentIndex = UI_COMPONENT_INDEX.APPLY_FIXES;
+                this.flowChartItemIndex = MIGRATION_STEP_DISPLAY_INDEX[state.step];
             } else if (state.step === MIGRATION_STEP.WAITING_FOR_APPROVE || state.step === MIGRATION_STEP.IDLE) {
                 this.componentIndex = UI_COMPONENT_INDEX.RESULT_SUCCESS;
                 this.flowChartItemIndex = MIGRATION_STEP_DISPLAY_INDEX[state.step];

src/Resources/app/administration/src/module/swag-migration/page/swag-migration-process-screen/swag-migration-process-screen.html.twig

@@ -36,6 +36,9 @@
                     <sw-step-item>
                         {{ $tc('swag-migration.index.loadingScreenCard.flowChart.fetchData') }}
                     </sw-step-item>
+                    <sw-step-item>
+                        {{ $tc('swag-migration.index.loadingScreenCard.flowChart.applyFixes') }}
+                    </sw-step-item>
                     <sw-step-item>
                         {{ $tc('swag-migration.index.loadingScreenCard.flowChart.writeData') }}
                     </sw-step-item>
@@ -62,6 +65,12 @@
                     :total="total" />
             {% endblock %}
 
+            {% block swag_migration_index_main_page_card_applyFixes %}
+                <div v-if="componentIndex === UI_COMPONENT_INDEX.APPLY_FIXES">
+                    <swag-migration-apply-fixes/>
+                </div>
+            {% endblock %}
+
             {% block swag_migration_index_main_page_card_loadingScreen_success %}
                 <swag-migration-result-screen v-if="componentIndex === UI_COMPONENT_INDEX.RESULT_SUCCESS" />
             {% endblock %}

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

@@ -376,6 +376,7 @@
         "flowChart": {
           "premapping": "Prüfen",
           "fetchData": "Lesen",
+          "applyFixes": "Fehler beheben",
           "writeData": "Schreiben",
           "processMedia": "Download",
           "cleanup": "Aufräumen",
@@ -421,6 +422,10 @@
           "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."
+        },
+        "applyFixes": {
+          "title": "Fehler beheben",
+          "resumeButton": "Migration fortsetzen"
         }
       },
       "confirmAbortDialog": {

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

@@ -397,6 +397,7 @@
         "flowChart": {
           "premapping": "Checking",
           "fetchData": "Reading",
+          "applyFixes": "Applying fixes",
           "writeData": "Writing",
           "processMedia": "Downloading",
           "cleanup": "Cleanup",
@@ -442,6 +443,10 @@
           "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."
+        },
+        "applyFixes": {
+          "title": "Apply fixes",
+          "resumeButton": "Continue migration"
         }
       },
       "confirmAbortDialog": {

src/Resources/app/administration/src/type/types.d.ts

@@ -16,6 +16,7 @@ type TRepository<T> = Repository<T>;
 type MigrationStep =
     | 'idle'
     | 'fetching'
+    | 'apply-fixes'
     | 'writing'
     | 'media-processing'
     | 'cleanup'

tests/Migration/Controller/StatusControllerTest.php

@@ -612,6 +612,46 @@ public function testFinishMigration(): void
         static::assertSame(MigrationStep::FINISHED, $run->getStep());
     }
 
+    public function testResumeMigrationWithIncorrectStep(): void
+    {
+        $this->runRepo->update(
+            [
+                [
+                    'id' => $this->runUuid,
+                    'step' => MigrationStep::FETCHING->value,
+                ],
+            ],
+            $this->context
+        );
+
+        try {
+            $this->controller->resumeAfterFixes($this->context);
+        } catch (MigrationException $e) {
+            static::assertSame(Response::HTTP_BAD_REQUEST, $e->getStatusCode());
+            static::assertSame(MigrationException::MIGRATION_NOT_IN_STEP, $e->getErrorCode());
+        }
+    }
+
+    public function testResumeMigration(): void
+    {
+        $this->runRepo->update(
+            [
+                [
+                    'id' => $this->runUuid,
+                    'step' => MigrationStep::APPLY_FIXES->value,
+                ],
+            ],
+            $this->context
+        );
+
+        $response = $this->controller->resumeAfterFixes($this->context);
+        static::assertSame(Response::HTTP_NO_CONTENT, $response->getStatusCode());
+
+        $run = $this->runRepo->search(new Criteria([$this->runUuid]), $this->context)->getEntities()->first();
+        static::assertNotNull($run);
+        static::assertSame(MigrationStep::WRITING, $run->getStep());
+    }
+
     private function createConnection(string $connectionId, string $profileName, string $connectionName): void
     {
         $this->context->scope(MigrationContext::SOURCE_CONTEXT, function (Context $context) use ($connectionId, $profileName, $connectionName): void {