feat: checksum and reset via mq

UPGRADE.md

@@ -67,6 +67,10 @@
     - Added methods `setProfile()`, `getGateway()`,  `setGateway()` and `setConnection()` to `SwagMigrationAssistant\Migration\MigrationContext`
     - Added null checks to methods `getProfile()` and `getGateway()` in `SwagMigrationAssistant\Migration\MigrationContext` to ensure that a profile and gateway is set before usage
 
+- [BREAKING] [#57](https://github.com/shopware/SwagMigrationAssistant/pull/57) feat!: checksum and reset via mq
+    - [BREAKING] Renamed method `cleanupMappingChecksums()` to `startCleanupMappingChecksums()` in `SwagMigrationAssistant\Migration\Run\RunServiceInterface` and implementation `SwagMigrationAssistant\Migration\Run\RunService`
+    - [BREAKING] Renamed method `cleanupMigrationData()` to `startTruncateMigrationData()` in `SwagMigrationAssistant\Migration\Run\RunServiceInterface` and implementation `SwagMigrationAssistant\Migration\Run\RunService`
+
 # 14.0.0
 - [BREAKING] MIG-1053 - Removed ability to set the `verify` flag for the guzzle API client. This is now always true by default.
 - [BREAKING] MIG-1053 - Refactored both Shopware 5 and Shopware 6 EnvironmentReader classes to provide more information about exceptions.

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,18 +354,18 @@ public function resetChecksums(Request $request, Context $context): Response
     )]
     public function cleanupMigrationData(Context $context): Response
     {
-        $this->runService->cleanupMigrationData($context);
+        $this->runService->startTruncateMigrationData($context);
 
         return new Response();
     }
 
     #[Route(
-        path: '/api/_action/migration/get-reset-status',
-        name: 'api.admin.migration.get-reset-status',
+        path: '/api/_action/migration/is-truncating-migration-data',
+        name: 'api.admin.migration.is-truncating-migration-data',
         defaults: ['_acl' => ['admin']],
         methods: [Request::METHOD_GET]
     )]
-    public function getResetStatus(Context $context): JsonResponse
+    public function isTruncatingMigrationData(Context $context): JsonResponse
     {
         $settings = $this->generalSettingRepo->search(new Criteria(), $context)->getEntities()->first();
 
@@ -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,37 @@
+<?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
+    {
+        $this->addColumn(
+            connection: $connection,
+            table: self::TABLE,
+            column: self::COLUMN,
+            type: 'TINYINT(1)',
+            nullable: false,
+            default: '0'
+        );
+    }
+}

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"/>
@@ -312,7 +311,7 @@
             <tag name="messenger.message_handler"/>
         </service>
 
-        <service id="SwagMigrationAssistant\Migration\MessageQueue\Handler\CleanupMigrationHandler">
+        <service id="SwagMigrationAssistant\Migration\MessageQueue\Handler\TruncateMigrationHandler">
             <argument type="service" id="Doctrine\DBAL\Connection"/>
             <argument type="service" id="messenger.default_bus"/>
             <tag name="messenger.message_handler"/>
@@ -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/Exception/MigrationException.php

@@ -32,6 +32,10 @@ class MigrationException extends HttpException
 
     public const MIGRATION_IS_ALREADY_RUNNING = 'SWAG_MIGRATION__MIGRATION_IS_ALREADY_RUNNING';
 
+    public const MIGRATION_IS_RESETTING_CHECKSUMS = 'SWAG_MIGRATION__MIGRATION_IS_RESETTING_CHECKSUMS';
+
+    public const MIGRATION_IS_TRUNCATING_DATA = 'SWAG_MIGRATION__MIGRATION_IS_TRUNCATING_DATA';
+
     public const NO_CONNECTION_IS_SELECTED = 'SWAG_MIGRATION__NO_CONNECTION_IS_SELECTED';
 
     public const NO_CONNECTION_FOUND = 'SWAG_MIGRATION__NO_CONNECTION_FOUND';
@@ -294,6 +298,24 @@ public static function migrationIsAlreadyRunning(): self
         );
     }
 
+    public static function checksumResetRunning(): self
+    {
+        return new MigrationIsAlreadyRunningException(
+            Response::HTTP_BAD_REQUEST,
+            self::MIGRATION_IS_RESETTING_CHECKSUMS,
+            'Checksum reset is running.',
+        );
+    }
+
+    public static function truncatingDataRunning(): self
+    {
+        return new MigrationIsAlreadyRunningException(
+            Response::HTTP_BAD_REQUEST,
+            self::MIGRATION_IS_TRUNCATING_DATA,
+            'Data truncation is running.',
+        );
+    }
+
     public static function noConnectionIsSelected(): self
     {
         return new self(

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,14 @@ 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,
+            $run->getId(),
+            $progress->getCurrentEntity(),
+            null,
+            0,
+            true // abort flow flag
+        ));
     }
 }

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

@@ -9,11 +9,9 @@
 
 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 +25,8 @@
 #[Package('fundamentals@after-sales')]
 class CleanUpProcessor extends AbstractProcessor
 {
+    public const BATCH_SIZE = 250;
+
     /**
      * @param EntityRepository<SwagMigrationRunCollection> $migrationRunRepo
      * @param EntityRepository<SwagMigrationDataCollection> $migrationDataRepo
@@ -37,7 +37,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 +59,51 @@ 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()));
+        $this->updateProgress(
+            $migrationContext->getRunUuid(),
+            $progress,
+            $context
+        );
+
+        $this->bus->dispatch(new MigrationProcessMessage(
+            $context,
+            $migrationContext->getRunUuid()
+        ));
     }
 
-    private function removeMigrationData(): int|string
+    private function removeMigrationData(): int
     {
-        return (new QueryBuilder($this->dbalConnection))
-            ->delete(SwagMigrationDataDefinition::ENTITY_NAME)
-            ->andWhere('written = 1')
-            ->setMaxResults(1000)
+        return (int) $this->connection->createQueryBuilder()
+            ->delete('swag_migration_data')
+            ->setMaxResults(self::BATCH_SIZE)
             ->executeStatement();
     }
+
+    private function getMigrationDataTotal(): int
+    {
+        return (int) $this->connection->createQueryBuilder()
+            ->select('COUNT(id)')->from('swag_migration_data')
+            ->executeQuery()
+            ->fetchOne();
+    }
 }