Process queue command

composer.json

@@ -66,7 +66,8 @@
         "symfony/mailchimp-mailer": "^6.4",
         "symfony/sendgrid-mailer": "^6.4",
         "symfony/twig-bundle": "^6.4",
-        "symfony/messenger": "^6.4"
+        "symfony/messenger": "^6.4",
+        "symfony/lock": "^6.4"
     },
     "require-dev": {
         "phpunit/phpunit": "^9.5",

config/services/managers.yml

@@ -55,8 +55,3 @@ services:
   PhpList\Core\Domain\Configuration\Service\Manager\ConfigManager:
     autowire: true
     autoconfigure: true
-
-  PhpList\Core\Domain\Subscription\Service\SubscriberDeletionService:
-    autowire: true
-    autoconfigure: true
-    public: true

config/services/providers.yml

@@ -0,0 +1,4 @@
+services:
+  PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider:
+    autowire: true
+    autoconfigure: true

config/services/repositories.yml

@@ -100,3 +100,8 @@ services:
         parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
         arguments:
             - PhpList\Core\Domain\Subscription\Model\SubscriberHistory
+
+    PhpList\Core\Domain\Messaging\Repository\ListMessageRepository:
+        parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
+        arguments:
+            - PhpList\Core\Domain\Subscription\Model\ListMessage

config/services/services.yml

@@ -14,3 +14,13 @@ services:
         autoconfigure: true
         arguments:
             $defaultFromEmail: '%app.mailer_from%'
+
+    PhpList\Core\Domain\Subscription\Service\SubscriberDeletionService:
+        autowire: true
+        autoconfigure: true
+        public: true
+
+    PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator:
+        autowire: true
+        autoconfigure: true
+        public: true

src/Domain/Messaging/Command/ProcessQueueCommand.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Messaging\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Doctrine\ORM\EntityManagerInterface;
+use PhpList\Core\Domain\Messaging\Model\Message;
+use PhpList\Core\Domain\Messaging\Repository\MessageRepository;
+use PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator;
+use PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Mailer\MailerInterface;
+use Symfony\Component\Mime\Email;
+use Throwable;
+
+class ProcessQueueCommand extends Command
+{
+    protected static $defaultName = 'phplist:process-queue';
+
+    private MessageRepository $messageRepository;
+    private MailerInterface $mailer;
+    private LockFactory $lockFactory;
+    private EntityManagerInterface $entityManager;
+    private SubscriberProvider $subscriberProvider;
+    private MessageProcessingPreparator $messageProcessingPreparator;
+
+    public function __construct(
+        MessageRepository $messageRepository,
+        MailerInterface   $mailer,
+        LockFactory       $lockFactory,
+        EntityManagerInterface $entityManager,
+        SubscriberProvider $subscriberProvider,
+        MessageProcessingPreparator $messageProcessingPreparator
+    ) {
+        parent::__construct();
+        $this->messageRepository = $messageRepository;
+        $this->mailer = $mailer;
+        $this->lockFactory = $lockFactory;
+        $this->entityManager = $entityManager;
+        $this->subscriberProvider = $subscriberProvider;
+        $this->messageProcessingPreparator = $messageProcessingPreparator;
+    }
+
+    /**
+     * @SuppressWarnings("PHPMD.UnusedFormalParameter")
+     */
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $lock = $this->lockFactory->createLock('queue_processor');
+        if (!$lock->acquire()) {
+            $output->writeln('Queue is already being processed by another instance.');
+
+            return Command::FAILURE;
+        }
+
+        try {
+            $this->messageProcessingPreparator->ensureSubscribersHaveUuid($output);
+            $this->messageProcessingPreparator->ensureCampaignsHaveUuid($output);
+
+            $campaigns = $this->messageRepository->findBy(['status' => 'submitted']);
+
+            foreach ($campaigns as $campaign) {
+                $this->processCampaign($campaign, $output);
+            }
+        } finally {
+            $lock->release();
+        }
+
+        return Command::SUCCESS;
+    }
+
+    private function processCampaign(Message $campaign, OutputInterface $output): void
+    {
+        $subscribers = $this->subscriberProvider->getSubscribersForMessage($campaign);
+        // todo: check $ISPrestrictions logic
+        foreach ($subscribers as $subscriber) {
+            if (!filter_var($subscriber->getEmail(), FILTER_VALIDATE_EMAIL)) {
+                continue;
+            }
+
+            $email = (new Email())
+                ->from('[email protected]')
+                ->to($subscriber->getEmail())
+                ->subject($campaign->getContent()->getSubject())
+                ->text($campaign->getContent()->getTextMessage())
+                ->html($campaign->getContent()->getText());
+
+            try {
+                $this->mailer->send($email);
+
+                // todo: log somewhere that this subscriber got email
+            } catch (Throwable $e) {
+                $output->writeln('Failed to send to: ' . $subscriber->getEmail());
+            }
+
+            usleep(100000);
+        }
+
+        $campaign->getMetadata()->setStatus('sent');
+        $this->entityManager->flush();
+    }
+}

src/Domain/Messaging/Model/Message.php

@@ -125,6 +125,12 @@ public function getUuid(): ?string
         return $this->uuid;
     }
 
+    public function setUuid(string $uuid): self
+    {
+        $this->uuid = $uuid;
+        return $this;
+    }
+
     public function getOwner(): ?Administrator
     {
         return $this->owner;

src/Domain/Messaging/Repository/ListMessageRepository.php

@@ -11,4 +11,15 @@
 class ListMessageRepository extends AbstractRepository implements PaginatableRepositoryInterface
 {
     use CursorPaginationTrait;
+
+    /** @return int[] */
+    public function getListIdsByMessageId(int $messageId): array
+    {
+        return $this->createQueryBuilder('lm')
+            ->select('IDENTITY(lm.list)')
+            ->where('lm.messageId = :messageId')
+            ->setParameter('messageId', $messageId)
+            ->getQuery()
+            ->getSingleColumnResult();
+    }
 }

src/Domain/Messaging/Repository/MessageRepository.php

@@ -12,6 +12,18 @@
 
 class MessageRepository extends AbstractRepository implements PaginatableRepositoryInterface
 {
+    /**
+     * @return Message[]
+     */
+    public function findCampaignsWithoutUuid(): array
+    {
+        return $this->createQueryBuilder('m')
+            ->where('m.uuid IS NULL OR m.uuid = :emptyString')
+            ->setParameter('emptyString', '')
+            ->getQuery()
+            ->getResult();
+    }
+
     public function getByOwnerId(int $ownerId): array
     {
         return $this->createQueryBuilder('m')

src/Domain/Messaging/Service/MessageProcessingPreparator.php

@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Messaging\Service;
+
+use Doctrine\ORM\EntityManagerInterface;
+use PhpList\Core\Domain\Messaging\Repository\MessageRepository;
+use PhpList\Core\Domain\Subscription\Repository\SubscriberRepository;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class MessageProcessingPreparator
+{
+    private EntityManagerInterface $entityManager;
+    private SubscriberRepository $subscriberRepository;
+    private MessageRepository $messageRepository;
+
+    public function __construct(
+        EntityManagerInterface $entityManager,
+        SubscriberRepository $subscriberRepository,
+        MessageRepository $messageRepository
+    ) {
+        $this->entityManager = $entityManager;
+        $this->subscriberRepository = $subscriberRepository;
+        $this->messageRepository = $messageRepository;
+    }
+
+    public function ensureSubscribersHaveUuid(OutputInterface $output): void
+    {
+        $subscribersWithoutUuid = $this->subscriberRepository->findSubscribersWithoutUuid();
+
+        $numSubscribers = count($subscribersWithoutUuid);
+        if ($numSubscribers > 0) {
+            $output->writeln(sprintf('Giving a UUID to %d subscribers, this may take a while', $numSubscribers));
+            foreach ($subscribersWithoutUuid as $subscriber) {
+                $subscriber->setUniqueId(bin2hex(random_bytes(16)));
+            }
+            $this->entityManager->flush();
+        }
+    }
+
+    public function ensureCampaignsHaveUuid(OutputInterface $output): void
+    {
+        $campaignsWithoutUuid = $this->messageRepository->findCampaignsWithoutUuid();
+
+        $numCampaigns = count($campaignsWithoutUuid);
+        if ($numCampaigns > 0) {
+            $output->writeln(sprintf('Giving a UUID to %d campaigns', $numCampaigns));
+            foreach ($campaignsWithoutUuid as $campaign) {
+                $campaign->setUuid(bin2hex(random_bytes(18)));
+            }
+            $this->entityManager->flush();
+        }
+    }
+}

src/Domain/Subscription/Repository/SubscriberRepository.php

@@ -19,6 +19,18 @@
  */
 class SubscriberRepository extends AbstractRepository implements PaginatableRepositoryInterface
 {
+    /**
+     * @return Subscriber[]
+     */
+    public function findSubscribersWithoutUuid(): array
+    {
+        return $this->createQueryBuilder('s')
+            ->where('s.uniqueId IS NULL OR s.uniqueId = :emptyString')
+            ->setParameter('emptyString', '')
+            ->getQuery()
+            ->getResult();
+    }
+
     public function findOneByEmail(string $email): ?Subscriber
     {
         return $this->findOneBy(['email' => $email]);

src/Domain/Subscription/Service/Provider/SubscriberProvider.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Subscription\Service\Provider;
+
+use PhpList\Core\Domain\Messaging\Model\Message;
+use PhpList\Core\Domain\Messaging\Repository\ListMessageRepository;
+use PhpList\Core\Domain\Subscription\Model\Subscriber;
+use PhpList\Core\Domain\Subscription\Repository\SubscriberRepository;
+
+class SubscriberProvider
+{
+    private ListMessageRepository $listMessageRepository;
+    private SubscriberRepository $subscriberRepository;
+
+    public function __construct(
+        ListMessageRepository $listMessageRepository,
+        SubscriberRepository $subscriberRepository
+    ) {
+        $this->listMessageRepository = $listMessageRepository;
+        $this->subscriberRepository = $subscriberRepository;
+    }
+
+    /**
+     * Get subscribers for a message
+     *
+     * @param Message $message The message to get subscribers for
+     * @return Subscriber[] Array of subscribers
+     */
+    public function getSubscribersForMessage(Message $message): array
+    {
+        $listIds = $this->listMessageRepository->getListIdsByMessageId($message->getId());
+
+        $subscribers = [];
+        foreach ($listIds as $listId) {
+            $listSubscribers = $this->subscriberRepository->getSubscribersBySubscribedListId($listId);
+            foreach ($listSubscribers as $subscriber) {
+                $subscribers[$subscriber->getId()] = $subscriber;
+            }
+        }
+
+        return array_values($subscribers);
+    }
+}