refactor: cleanup

Author: larskemper

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

@@ -354,18 +354,18 @@ public function resetChecksums(Request $request, Context $context): Response
     )]
     public function cleanupMigrationData(Context $context): Response
     {
-        $this->runService->startCleanupMigrationData($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();
 

src/Core/Migration/Migration1759000000AddIsResettingChecksumsToSetting.php

@@ -25,17 +25,13 @@ public function getCreationTimestamp(): int
 
     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
-        ));
+        $this->addColumn(
+            connection: $connection,
+            table: self::TABLE,
+            column: self::COLUMN,
+            type: 'TINYINT(1)',
+            nullable: false,
+            default: '0'
+        );
     }
 }

src/DependencyInjection/migration.xml

@@ -311,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"/>

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

@@ -58,7 +58,6 @@ public function process(
         $this->bus->dispatch(new ResetChecksumMessage(
             $migrationContext->getConnection()->getId(),
             $context,
-            true,
             $run->getId(),
             $progress->getCurrentEntity(),
             null,

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

@@ -7,7 +7,6 @@
 
 namespace SwagMigrationAssistant\Migration\MessageQueue\Handler\Processor;
 
-use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
 use Shopware\Core\Framework\Context;
 use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
@@ -94,23 +93,10 @@ public function process(
 
     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;
-        }
-
-        $query = $this->connection->createQueryBuilder()
+        return (int) $this->connection->createQueryBuilder()
             ->delete('swag_migration_data')
-            ->where('id IN (:ids)')
-            ->setParameter('ids', $ids, ArrayParameterType::BINARY);
-
-        return (int) $query->executeStatement();
+            ->setMaxResults(self::BATCH_SIZE)
+            ->executeStatement();
     }
 
     private function getMigrationDataTotal(): int

src/Migration/MessageQueue/Handler/ResetChecksumHandler.php

@@ -7,8 +7,8 @@
 
 namespace SwagMigrationAssistant\Migration\MessageQueue\Handler;
 
-use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\ParameterType;
 use Shopware\Core\Framework\Context;
 use Shopware\Core\Framework\DataAbstractionLayer\EntityRepository;
 use Shopware\Core\Framework\Log\Package;
@@ -51,52 +51,27 @@ public function __invoke(ResetChecksumMessage $message): void
         $progress = null;
 
         if ($totalMappings === null) {
-            $totalMappings = $this->getTotalMappingsCount($connectionId, $message->isResettingAll());
+            $totalMappings = $this->getTotalMappingsCount($connectionId);
 
             if ($message->getRunId() !== null && $totalMappings > 0) {
-                $progress = $this->initializeProgress(
+                $progress = $this->updateProgress(
                     $message,
+                    0,
                     $totalMappings,
                     $message->getContext()
                 );
             }
         }
 
-        $query = $this->connection->createQueryBuilder()
-            ->select('m.id')
-            ->from('swag_migration_mapping', 'm')
-            ->where('m.checksum IS NOT NULL')
-            ->andWhere('m.connection_id = :connectionId')
-            ->setParameter('connectionId', Uuid::fromHexToBytes($connectionId))
-            ->setMaxResults(self::BATCH_SIZE);
-
-        if (!$message->isResettingAll()) {
-            $query->innerJoin(
-                'm',
-                'swag_migration_data',
-                'd',
-                'd.mapping_uuid = m.id AND d.written = 0'
-            );
-        }
-
-        $ids = $query->executeQuery()->fetchFirstColumn();
-        $batchSize = \count($ids);
+        $affectedRows = $this->resetChecksums($connectionId);
 
-        if ($batchSize === 0) {
-            $this->clearResettingChecksumsFlag();
+        if ($affectedRows === 0) {
             $this->handleCompletion($message, $progress);
 
             return;
         }
 
-        $this->connection->createQueryBuilder()
-            ->update('swag_migration_mapping')
-            ->set('checksum', 'NULL')
-            ->where('id IN (:ids)')
-            ->setParameter('ids', $ids, ArrayParameterType::BINARY)
-            ->executeStatement();
-
-        $newProcessedCount = $message->getProcessedMappings() + $batchSize;
+        $newProcessedCount = $message->getProcessedMappings() + $affectedRows;
 
         if ($message->getRunId() !== null) {
             $progress = $this->updateProgress(
@@ -107,8 +82,7 @@ public function __invoke(ResetChecksumMessage $message): void
             );
         }
 
-        if ($batchSize < self::BATCH_SIZE) {
-            $this->clearResettingChecksumsFlag();
+        if ($affectedRows < self::BATCH_SIZE) {
             $this->handleCompletion($message, $progress);
 
             return;
@@ -117,7 +91,6 @@ public function __invoke(ResetChecksumMessage $message): void
         $this->messageBus->dispatch(new ResetChecksumMessage(
             $message->getConnectionId(),
             $message->getContext(),
-            $message->isResettingAll(),
             $message->getRunId(),
             $message->getEntity(),
             $totalMappings,
@@ -128,6 +101,8 @@ public function __invoke(ResetChecksumMessage $message): void
 
     private function handleCompletion(ResetChecksumMessage $message, ?MigrationProgress $progress): void
     {
+        $this->clearResettingChecksumsFlag();
+
         if (!$message->isPartOfAbort() || $message->getRunId() === null) {
             return;
         }
@@ -152,7 +127,7 @@ private function handleCompletion(ResetChecksumMessage $message, ?MigrationProgr
         $this->migrationRunRepo->upsert([
             [
                 'id' => $runId,
-                'progress' => $finalProgress,
+                'progress' => $finalProgress->jsonSerialize(),
             ],
         ], $context);
 
@@ -162,43 +137,35 @@ private function handleCompletion(ResetChecksumMessage $message, ?MigrationProgr
         ));
     }
 
-    private function getTotalMappingsCount(string $connectionId, bool $resetAll): int
+    private function resetChecksums(string $connectionId): int
     {
-        $query = $this->connection->createQueryBuilder()
+        return (int) $this->connection->executeStatement(
+            'UPDATE swag_migration_mapping
+            SET checksum = NULL
+            WHERE checksum IS NOT NULL
+              AND connection_id = :connectionId
+            LIMIT :limit',
+            [
+                'connectionId' => Uuid::fromHexToBytes($connectionId),
+                'limit' => self::BATCH_SIZE,
+            ],
+            [
+                'connectionId' => ParameterType::BINARY,
+                'limit' => ParameterType::INTEGER,
+            ]
+        );
+    }
+
+    private function getTotalMappingsCount(string $connectionId): int
+    {
+        return (int) $this->connection->createQueryBuilder()
             ->select('COUNT(m.id)')
             ->from('swag_migration_mapping', 'm')
             ->where('m.checksum IS NOT NULL')
             ->andWhere('m.connection_id = :connectionId')
-            ->setParameter('connectionId', Uuid::fromHexToBytes($connectionId));
-
-        if (!$resetAll) {
-            $query->innerJoin(
-                'm',
-                'swag_migration_data',
-                'd',
-                'd.mapping_uuid = m.id AND d.written = 0'
-            );
-        }
-
-        return (int) $query->executeQuery()->fetchOne();
-    }
-
-    private function initializeProgress(ResetChecksumMessage $message, int $total, Context $context): MigrationProgress
-    {
-        $progress = new MigrationProgress(
-            0,
-            $total,
-            new ProgressDataSetCollection(),
-            $message->getEntity() ?? DefaultEntities::RULE,
-            0
-        );
-
-        $this->migrationRunRepo->update([[
-            'id' => $message->getRunId(),
-            'progress' => $progress->jsonSerialize(),
-        ]], $context);
-
-        return $progress;
+            ->setParameter('connectionId', Uuid::fromHexToBytes($connectionId))
+            ->executeQuery()
+            ->fetchOne();
     }
 
     private function updateProgress(ResetChecksumMessage $message, int $processed, int $total, Context $context): MigrationProgress

…ueue/Handler/CleanupMigrationHandler.php …eue/Handler/TruncateMigrationHandler.phpsrc/Migration/MessageQueue/Handler/CleanupMigrationHandler.php renamed to src/Migration/MessageQueue/Handler/TruncateMigrationHandler.php src/Migration/MessageQueue/Handler/CleanupMigrationHandler.php renamed to src/Migration/MessageQueue/Handler/TruncateMigrationHandler.php

@@ -9,7 +9,7 @@
 
 use Doctrine\DBAL\Connection;
 use Shopware\Core\Framework\Log\Package;
-use SwagMigrationAssistant\Migration\MessageQueue\Message\CleanupMigrationMessage;
+use SwagMigrationAssistant\Migration\MessageQueue\Message\TruncateMigrationMessage;
 use Symfony\Component\Messenger\Attribute\AsMessageHandler;
 use Symfony\Component\Messenger\MessageBusInterface;
 
@@ -18,15 +18,17 @@
 /**
  * @internal
  */
-final class CleanupMigrationHandler
+final class TruncateMigrationHandler
 {
+    private const BATCH_SIZE = 250;
+
     public function __construct(
         private readonly Connection $connection,
         private readonly MessageBusInterface $bus,
     ) {
     }
 
-    public function __invoke(CleanupMigrationMessage $message): void
+    public function __invoke(TruncateMigrationMessage $message): void
     {
         $currentStep = 0;
         $tablesToReset = [
@@ -48,15 +50,30 @@ public function __invoke(CleanupMigrationMessage $message): void
             $currentStep = $step;
         }
 
+        $affectedRows = (int) $this->connection->executeStatement(
+            'DELETE FROM ' . $tablesToReset[$currentStep] . ' LIMIT ' . self::BATCH_SIZE
+        );
+
+        if ($affectedRows >= self::BATCH_SIZE) {
+            $this->bus->dispatch(new TruncateMigrationMessage(
+                $tablesToReset[$currentStep]
+            ));
+
+            return;
+        }
+
         $nextStep = $currentStep + 1;
 
         if (isset($tablesToReset[$nextStep])) {
-            $nextMessage = new CleanupMigrationMessage($tablesToReset[$nextStep]);
-            $this->bus->dispatch($nextMessage);
+            $this->bus->dispatch(new TruncateMigrationMessage(
+                $tablesToReset[$nextStep]
+            ));
+
+            return;
         }
 
         $this->connection->executeStatement(
-            'DELETE FROM ' . $tablesToReset[$currentStep] . ';'
+            'UPDATE swag_migration_general_setting SET `is_reset` = 0;'
         );
     }
 }