Filter migrated and processing subscriptions in Mollie subscription repository queries (#347)

Author: TheMilek

UPGRADE-3.3.md

@@ -24,3 +24,17 @@
    ```bash
    bin/console doctrine:migrations:migrate
    ```
+
+   To support the transition to 4.0, two interfaces gained methods used by the migration process. If you provide
+   your own implementations, add them (note that both interfaces are themselves deprecated and will be removed in
+   4.0):
+
+   `Sylius\MolliePlugin\Repository\MollieSubscriptionRepositoryInterface`:
+   - `findScheduledSubscriptionsForMigration(): array`
+   - `iterateToMigrate(int $batchSize): iterable`
+   - `findMigrated(int $limit): array`
+
+   `Sylius\MolliePlugin\Entity\MollieSubscriptionInterface`:
+   - `getMigratedAt(): ?\DateTime`
+   - `setMigratedAt(?\DateTimeInterface $migratedAt): void`
+   - `isMigrated(): bool`

src/Repository/MollieSubscriptionRepository.php

@@ -72,19 +72,21 @@ public function findScheduledSubscriptions(): array
         $qb->andWhere('s.scheduledDate < :date');
         $qb->setParameter('date', new \DateTime());
         $qb->andWhere('s.fulfilledDate IS NULL');
+        $qb->andWhere('q.migratedAt IS NULL');
 
         return $qb->getQuery()->getResult();
     }
 
     public function findScheduledSubscriptionsForMigration(): array
     {
         $qb = $this->createQueryBuilder('q');
-        $qb->andWhere('q.state IN (:states)');
-        $qb->setParameter('states', [MollieSubscriptionInterface::STATE_ACTIVE, MollieSubscriptionInterface::STATE_PROCESSING]);
+        $qb->andWhere('q.state = :state');
+        $qb->setParameter('state', MollieSubscriptionInterface::STATE_ACTIVE);
         $qb->leftJoin('q.schedules', 's');
         $qb->andWhere('s.scheduledDate < :date');
         $qb->setParameter('date', new \DateTime());
         $qb->andWhere('s.fulfilledDate IS NULL');
+        $qb->andWhere('q.migratedAt IS NULL');
 
         return $qb->getQuery()->getResult();
     }
@@ -96,6 +98,7 @@ public function findProcessableSubscriptions(): array
         $qb->setParameter('state', MollieSubscriptionInterface::STATE_PROCESSING);
         $qb->andWhere('q.processingState = :processingState');
         $qb->setParameter('processingState', MollieSubscriptionInterface::PROCESSING_STATE_PENDING);
+        $qb->andWhere('q.migratedAt IS NULL');
 
         return $qb->getQuery()->getResult();
     }
@@ -104,8 +107,11 @@ public function iterateToMigrate(int $batchSize): iterable
     {
         $qb = $this->createQueryBuilder('q');
         $qb->andWhere('q.migratedAt IS NULL');
-        $qb->andWhere('q.state != :paused');
-        $qb->setParameter('paused', MollieSubscriptionInterface::STATE_PAUSED);
+        $qb->andWhere('q.state NOT IN (:excludedStates)');
+        $qb->setParameter('excludedStates', [
+            MollieSubscriptionInterface::STATE_PAUSED,
+            MollieSubscriptionInterface::STATE_PROCESSING,
+        ]);
         $qb->setMaxResults($batchSize);
 
         return $qb->getQuery()->toIterable();