feat: checksum and reset via mq

src/Controller/StatusController.php

@@ -341,13 +341,7 @@ public function resetChecksums(Request $request, Context $context): Response
             throw RoutingException::missingRequestParameter('connectionId');
         }
 
-        $connection = $this->migrationConnectionRepo->search(new Criteria([$connectionId]), $context)->getEntities()->first();
-
-        if ($connection === null) {
-            throw MigrationException::noConnectionFound();
-        }
-
-        $this->runService->cleanupMappingChecksums($connectionId, $context);
+        $this->runService->startCleanupMappingChecksums($connectionId, $context);
 
         return new Response();
     }
@@ -360,7 +354,7 @@ public function resetChecksums(Request $request, Context $context): Response
     )]
     public function cleanupMigrationData(Context $context): Response
     {
-        $this->runService->cleanupMigrationData($context);
+        $this->runService->startCleanupMigrationData($context);
 
         return new Response();
     }
@@ -381,4 +375,26 @@ public function getResetStatus(Context $context): JsonResponse
 
         return new JsonResponse($settings->isReset());
     }
+
+    #[Route(
+        path: '/api/_action/migration/is-resetting-checksums',
+        name: 'api.admin.migration.is-resetting-checksums',
+        defaults: ['_acl' => ['admin']],
+        methods: [Request::METHOD_GET]
+    )]
+    public function isResettingChecksums(Context $context): JsonResponse
+    {
+        $settings = $this->generalSettingRepo
+            ->search(new Criteria(), $context)
+            ->getEntities()
+            ->first();
+
+        if ($settings === null) {
+            return new JsonResponse(false);
+        }
+
+        return new JsonResponse(
+            $settings->isResettingChecksums()
+        );
+    }
 }

src/Core/Migration/Migration1759000000AddIsResettingChecksumsToSetting.php

@@ -0,0 +1,41 @@
+<?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\Core\Migration;
+
+use Doctrine\DBAL\Connection;
+use Shopware\Core\Framework\Log\Package;
+use Shopware\Core\Framework\Migration\MigrationStep;
+
+#[Package('fundamentals@after-sales')]
+class Migration1759000000AddIsResettingChecksumsToSetting extends MigrationStep
+{
+    public const TABLE = 'swag_migration_general_setting';
+
+    public const COLUMN = 'is_resetting_checksums';
+
+    public function getCreationTimestamp(): int
+    {
+        return 1759000000;
+    }
+
+    public function update(Connection $connection): void
+    {
+        $schemaManager = $connection->createSchemaManager();
+        $columns = $schemaManager->listTableColumns(self::TABLE);
+
+        if (isset($columns[self::COLUMN])) {
+            return;
+        }
+
+        $connection->executeStatement(\sprintf(
+            'ALTER TABLE %s ADD COLUMN %s TINYINT(1) NOT NULL DEFAULT 0',
+            self::TABLE,
+            self::COLUMN
+        ));
+    }
+}

src/DependencyInjection/migration.xml

@@ -185,7 +185,6 @@
             <argument type="service" id="swag_migration_general_setting.repository"/>
             <argument type="service" id="Shopware\Storefront\Theme\ThemeService"/>
             <argument type="service" id="SwagMigrationAssistant\Migration\Mapping\MappingService"/>
-            <argument type="service" id="SwagMigrationAssistant\Migration\Data\SwagMigrationDataDefinition"/>
             <argument type="service" id="Doctrine\DBAL\Connection"/>
             <argument type="service" id="SwagMigrationAssistant\Migration\Logging\LoggingService"/>
             <argument type="service" id="Shopware\Core\Framework\Store\Services\TrackingEventClient"/>
@@ -328,7 +327,14 @@
             <argument type="service" id="SwagMigrationAssistant\Migration\MigrationContextFactory"/>
             <argument type="service"
                       id="SwagMigrationAssistant\Migration\MessageQueue\Handler\MigrationProcessorRegistry"/>
+            <tag name="messenger.message_handler"/>
+        </service>
 
+        <service id="SwagMigrationAssistant\Migration\MessageQueue\Handler\ResetChecksumHandler">
+            <argument type="service" id="Doctrine\DBAL\Connection"/>
+            <argument type="service" id="messenger.default_bus"/>
+            <argument id="swag_migration_run.repository" type="service"/>
+            <argument id="SwagMigrationAssistant\Migration\Run\RunTransitionService" type="service"/>
             <tag name="messenger.message_handler"/>
         </service>
 
@@ -368,7 +374,6 @@
 
         <service id="SwagMigrationAssistant\Migration\MessageQueue\Handler\Processor\AbortingProcessor"
                  parent="SwagMigrationAssistant\Migration\MessageQueue\Handler\Processor\AbstractProcessor">
-            <argument id="SwagMigrationAssistant\Migration\Run\RunService" type="service"/>
             <argument type="service" id="messenger.default_bus"/>
 
             <tag name="shopware.migration.processor"/>

src/Migration/MessageQueue/Handler/CleanupMigrationHandler.php

@@ -38,16 +38,25 @@ public function __invoke(CleanupMigrationMessage $message): void
             'swag_migration_connection',
         ];
 
-        $step = \array_search($message->getTableName(), $tablesToReset, true);
+        $step = \array_search(
+            $message->getTableName(),
+            $tablesToReset,
+            true
+        );
+
         if ($step !== false) {
             $currentStep = $step;
         }
 
         $nextStep = $currentStep + 1;
+
         if (isset($tablesToReset[$nextStep])) {
             $nextMessage = new CleanupMigrationMessage($tablesToReset[$nextStep]);
             $this->bus->dispatch($nextMessage);
         }
-        $this->connection->executeStatement('DELETE FROM ' . $tablesToReset[$currentStep] . ';');
+
+        $this->connection->executeStatement(
+            'DELETE FROM ' . $tablesToReset[$currentStep] . ';'
+        );
     }
 }

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

@@ -12,11 +12,10 @@
 use Shopware\Core\Framework\Log\Package;
 use SwagMigrationAssistant\Migration\Data\SwagMigrationDataCollection;
 use SwagMigrationAssistant\Migration\Media\SwagMigrationMediaFileCollection;
-use SwagMigrationAssistant\Migration\MessageQueue\Message\MigrationProcessMessage;
+use SwagMigrationAssistant\Migration\MessageQueue\Message\ResetChecksumMessage;
 use SwagMigrationAssistant\Migration\MigrationContextInterface;
 use SwagMigrationAssistant\Migration\Run\MigrationProgress;
 use SwagMigrationAssistant\Migration\Run\MigrationStep;
-use SwagMigrationAssistant\Migration\Run\RunServiceInterface;
 use SwagMigrationAssistant\Migration\Run\RunTransitionServiceInterface;
 use SwagMigrationAssistant\Migration\Run\SwagMigrationRunCollection;
 use SwagMigrationAssistant\Migration\Run\SwagMigrationRunEntity;
@@ -35,7 +34,6 @@ public function __construct(
         EntityRepository $migrationDataRepo,
         EntityRepository $migrationMediaFileRepo,
         RunTransitionServiceInterface $runTransitionService,
-        private readonly RunServiceInterface $runService,
         private readonly MessageBusInterface $bus,
     ) {
         parent::__construct(
@@ -57,12 +55,15 @@ public function process(
         SwagMigrationRunEntity $run,
         MigrationProgress $progress,
     ): void {
-        $connection = $migrationContext->getConnection();
-        $this->runService->cleanupMappingChecksums($connection->getId(), $context);
-
-        $this->runTransitionService->forceTransitionToRunStep($migrationContext->getRunUuid(), MigrationStep::CLEANUP);
-        $progress->setIsAborted(true);
-        $this->updateProgress($migrationContext->getRunUuid(), $progress, $context);
-        $this->bus->dispatch(new MigrationProcessMessage($context, $migrationContext->getRunUuid()));
+        $this->bus->dispatch(new ResetChecksumMessage(
+            $migrationContext->getConnection()->getId(),
+            $context,
+            true,
+            $run->getId(),
+            $progress->getCurrentEntity(),
+            null,
+            0,
+            true // abort flow flag
+        ));
     }
 }

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

@@ -7,13 +7,12 @@
 
 namespace SwagMigrationAssistant\Migration\MessageQueue\Handler\Processor;
 
+use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
 use Shopware\Core\Framework\Context;
-use Shopware\Core\Framework\DataAbstractionLayer\Dbal\QueryBuilder;
 use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
 use Shopware\Core\Framework\Log\Package;
 use SwagMigrationAssistant\Migration\Data\SwagMigrationDataCollection;
-use SwagMigrationAssistant\Migration\Data\SwagMigrationDataDefinition;
 use SwagMigrationAssistant\Migration\Media\SwagMigrationMediaFileCollection;
 use SwagMigrationAssistant\Migration\MessageQueue\Message\MigrationProcessMessage;
 use SwagMigrationAssistant\Migration\MigrationContextInterface;
@@ -27,6 +26,8 @@
 #[Package('fundamentals@after-sales')]
 class CleanUpProcessor extends AbstractProcessor
 {
+    public const BATCH_SIZE = 250;
+
     /**
      * @param EntityRepository<SwagMigrationRunCollection> $migrationRunRepo
      * @param EntityRepository<SwagMigrationDataCollection> $migrationDataRepo
@@ -37,7 +38,7 @@ public function __construct(
         EntityRepository $migrationDataRepo,
         EntityRepository $migrationMediaFileRepo,
         RunTransitionServiceInterface $runTransitionService,
-        private readonly Connection $dbalConnection,
+        private readonly Connection $connection,
         private readonly MessageBusInterface $bus,
     ) {
         parent::__construct(
@@ -59,22 +60,64 @@ public function process(
         SwagMigrationRunEntity $run,
         MigrationProgress $progress,
     ): void {
-        $deleteCount = (int) $this->removeMigrationData();
+        if ($progress->getTotal() === 0) {
+            $progress->setTotal($this->getMigrationDataTotal());
+            $progress->setProgress(0);
+        }
+
+        $deleteCount = $this->removeMigrationData();
+
+        if ($deleteCount > 0) {
+            $progress->setProgress(
+                $progress->getProgress() + $deleteCount
+            );
+        }
 
         if ($deleteCount <= 0) {
-            $this->runTransitionService->transitionToRunStep($migrationContext->getRunUuid(), MigrationStep::INDEXING);
+            $this->runTransitionService->transitionToRunStep(
+                $migrationContext->getRunUuid(),
+                MigrationStep::INDEXING
+            );
+        }
+
+        $this->updateProgress(
+            $migrationContext->getRunUuid(),
+            $progress,
+            $context
+        );
+
+        $this->bus->dispatch(new MigrationProcessMessage(
+            $context,
+            $migrationContext->getRunUuid()
+        ));
+    }
+
+    private function removeMigrationData(): int
+    {
+        $ids = $this->connection->createQueryBuilder()
+            ->select('id')
+            ->from('swag_migration_data')
+            ->setMaxResults(self::BATCH_SIZE)
+            ->executeQuery()
+            ->fetchFirstColumn();
+
+        if (empty($ids)) {
+            return 0;
         }
 
-        $this->updateProgress($migrationContext->getRunUuid(), $progress, $context);
-        $this->bus->dispatch(new MigrationProcessMessage($context, $migrationContext->getRunUuid()));
+        $query = $this->connection->createQueryBuilder()
+            ->delete('swag_migration_data')
+            ->where('id IN (:ids)')
+            ->setParameter('ids', $ids, ArrayParameterType::BINARY);
+
+        return (int) $query->executeStatement();
     }
 
-    private function removeMigrationData(): int|string
+    private function getMigrationDataTotal(): int
     {
-        return (new QueryBuilder($this->dbalConnection))
-            ->delete(SwagMigrationDataDefinition::ENTITY_NAME)
-            ->andWhere('written = 1')
-            ->setMaxResults(1000)
-            ->executeStatement();
+        return (int) $this->connection->createQueryBuilder()
+            ->select('COUNT(id)')->from('swag_migration_data')
+            ->executeQuery()
+            ->fetchOne();
     }
 }