ProcessQueueCommand

Author: tatevikg1

src/Domain/Analytics/Service/LinkTrackService.php

@@ -72,7 +72,7 @@ public function extractAndSaveLinks(Message $message, int $userId): array
             $linkTrack->setUserId($userId);
             $linkTrack->setUrl($url);
 
-            $this->linkTrackRepository->save($linkTrack);
+            $this->linkTrackRepository->persist($linkTrack);
             $savedLinks[] = $linkTrack;
         }
 

src/Domain/Messaging/Command/ProcessQueueCommand.php

@@ -5,6 +5,7 @@
 namespace PhpList\Core\Domain\Messaging\Command;
 
 use DateTimeImmutable;
+use Doctrine\ORM\EntityManagerInterface;
 use PhpList\Core\Domain\Configuration\Model\ConfigOption;
 use PhpList\Core\Domain\Configuration\Service\Provider\ConfigProvider;
 use PhpList\Core\Domain\Messaging\Model\Message\MessageStatus;
@@ -31,14 +32,16 @@ class ProcessQueueCommand extends Command
     private CampaignProcessor $campaignProcessor;
     private ConfigProvider $configProvider;
     private TranslatorInterface $translator;
+    private EntityManagerInterface $entityManager;
 
     public function __construct(
         MessageRepository $messageRepository,
         LockFactory $lockFactory,
         MessageProcessingPreparator $messagePreparator,
         CampaignProcessor $campaignProcessor,
         ConfigProvider $configProvider,
-        TranslatorInterface $translator
+        TranslatorInterface $translator,
+        EntityManagerInterface $entityManager,
     ) {
         parent::__construct();
         $this->messageRepository = $messageRepository;
@@ -47,6 +50,7 @@ public function __construct(
         $this->campaignProcessor = $campaignProcessor;
         $this->configProvider = $configProvider;
         $this->translator = $translator;
+        $this->entityManager = $entityManager;
     }
 
     /**
@@ -73,11 +77,21 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             $this->messagePreparator->ensureSubscribersHaveUuid($output);
             $this->messagePreparator->ensureCampaignsHaveUuid($output);
 
-            $campaigns = $this->messageRepository->getByStatusAndEmbargo(
-                status: MessageStatus::Submitted,
-                embargo: new DateTimeImmutable()
-            );
+            $this->entityManager->flush();
+        } catch (Throwable $throwable) {
+            $output->writeln($throwable->getMessage());
+
+            return Command::FAILURE;
+        } finally {
+            $lock->release();
+        }
 
+        $campaigns = $this->messageRepository->getByStatusAndEmbargo(
+            status: MessageStatus::Submitted,
+            embargo: new DateTimeImmutable()
+        );
+
+        try {
             foreach ($campaigns as $campaign) {
                 $this->campaignProcessor->process($campaign, $output);
             }

src/Domain/Messaging/Service/Handler/RequeueHandler.php

@@ -46,7 +46,6 @@ public function handle(Message $campaign, ?OutputInterface $output = null): bool
         $schedule->setEmbargo($next);
         $campaign->setSchedule($schedule);
         $campaign->getMetadata()->setStatus(MessageStatus::Submitted);
-        $this->entityManager->flush();
 
         $output?->writeln($this->translator->trans(
             'Requeued campaign; next embargo at %time%',

src/Domain/Messaging/Service/MessageProcessingPreparator.php

@@ -49,7 +49,6 @@ public function ensureSubscribersHaveUuid(OutputInterface $output): void
             foreach ($subscribersWithoutUuid as $subscriber) {
                 $subscriber->setUniqueId(bin2hex(random_bytes(16)));
             }
-            $this->entityManager->flush();
         }
     }
 
@@ -65,7 +64,6 @@ public function ensureCampaignsHaveUuid(OutputInterface $output): void
             foreach ($campaignsWithoutUuid as $campaign) {
                 $campaign->setUuid(bin2hex(random_bytes(18)));
             }
-            $this->entityManager->flush();
         }
     }
 

src/Domain/Messaging/Service/Processor/CampaignProcessor.php

@@ -103,6 +103,7 @@ public function process(Message $campaign, ?OutputInterface $output = null): voi
                         ['%message_id%' => $campaign->getId()]
                     )
                 );
+                $this->entityManager->flush();
                 continue;
             }
 
@@ -125,6 +126,7 @@ public function process(Message $campaign, ?OutputInterface $output = null): voi
         }
 
         if ($stoppedEarly && $this->requeueHandler->handle($campaign, $output)) {
+            $this->entityManager->flush();
             return;
         }