IspRestrictions

Author: tatevikg1

config/parameters.yml.dist

@@ -71,3 +71,11 @@ parameters:
 
     app.phplist_isp_conf_path: '%%env(APP_PHPLIST_ISP_CONF_PATH)%%'
     env(APP_PHPLIST_ISP_CONF_PATH): '/etc/phplist.conf'
+
+    # Message sending
+    messaging.mail_queue_batch_size: '%%env(MAILQUEUE_BATCH_SIZE)%%'
+    env(MAILQUEUE_BATCH_SIZE): '5'
+    messaging.mail_queue_period: '%%env(MAILQUEUE_BATCH_PERIOD)%%'
+    env(MAILQUEUE_BATCH_PERIOD): '5'
+    messaging.mail_queue_throttle: '%%env(MAILQUEUE_THROTTLE)%%'
+    env(MAILQUEUE_THROTTLE): '5'

config/services.yml

@@ -7,10 +7,6 @@ services:
     autoconfigure: true
     public: false
 
-  PhpList\Core\Core\ConfigProvider:
-    arguments:
-      $config: '%app.config%'
-
   PhpList\Core\Core\ApplicationStructure:
     public: true
 

config/services/providers.yml

@@ -2,3 +2,7 @@ services:
   PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider:
     autowire: true
     autoconfigure: true
+
+  PhpList\Core\Core\ConfigProvider:
+    arguments:
+      $config: '%app.config%'

config/services/services.yml

@@ -35,6 +35,10 @@ services:
     autowire: true
     autoconfigure: true
     public: true
+    arguments:
+      $mailqueueBatchSize: '%messaging.mail_queue_batch_size%'
+      $mailqueueBatchPeriod: '%messaging.mail_queue_period%'
+      $mailqueueThrottle: '%messaging.mail_queue_throttle%'
 
   PhpList\Core\Domain\Common\ClientIpResolver:
     autowire: true

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

@@ -7,6 +7,7 @@
 use Doctrine\ORM\EntityManagerInterface;
 use PhpList\Core\Domain\Messaging\Model\Message;
 use PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator;
+use PhpList\Core\Domain\Common\IspRestrictionsProvider;
 use PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Console\Output\OutputInterface;
@@ -21,34 +22,78 @@ class CampaignProcessor
     private SubscriberProvider $subscriberProvider;
     private MessageProcessingPreparator $messagePreparator;
     private LoggerInterface $logger;
+    private ?IspRestrictionsProvider $ispRestrictionsProvider;
+    private ?int $mailqueueBatchSize;
+    private ?int $mailqueueBatchPeriod;
+    private ?int $mailqueueThrottle;
 
     public function __construct(
         MailerInterface $mailer,
         EntityManagerInterface $entityManager,
         SubscriberProvider $subscriberProvider,
         MessageProcessingPreparator $messagePreparator,
         LoggerInterface $logger,
+        ?IspRestrictionsProvider $ispRestrictionsProvider = null,
+        ?int $mailqueueBatchSize = null,
+        ?int $mailqueueBatchPeriod = null,
+        ?int $mailqueueThrottle = null,
     ) {
         $this->mailer = $mailer;
         $this->entityManager = $entityManager;
         $this->subscriberProvider = $subscriberProvider;
         $this->messagePreparator = $messagePreparator;
         $this->logger = $logger;
+        $this->ispRestrictionsProvider = $ispRestrictionsProvider;
+        $this->mailqueueBatchSize = $mailqueueBatchSize;
+        $this->mailqueueBatchPeriod = $mailqueueBatchPeriod;
+        $this->mailqueueThrottle = $mailqueueThrottle;
     }
 
     public function process(Message $campaign, ?OutputInterface $output = null): void
     {
-        $campaign->getMetadata()->setStatus(Message\MessageStatus::Prepared);
-        $this->entityManager->flush();
-
+        $this->updateMessageStatus($campaign, Message\MessageStatus::Prepared);
+        $ispRestrictions = $this->ispRestrictionsProvider->load();
         $subscribers = $this->subscriberProvider->getSubscribersForMessage($campaign);
 
-        $campaign->getMetadata()->setStatus(Message\MessageStatus::InProcess);
-        $this->entityManager->flush();
+        $cfgBatch = ($this->mailqueueBatchSize ?? 0);
+        $ispMax = isset($ispRestrictions->maxBatch) ? (int)$ispRestrictions->maxBatch : null;
+
+        $cfgPeriod = ($this->mailqueueBatchPeriod ?? 0);
+        $ispMinPeriod = ($ispRestrictions->minBatchPeriod ?? 0);
+
+        $cfgThrottle = ($this->mailqueueThrottle ?? 0);
+        $ispMinThrottle = (int)($ispRestrictions->minThrottle ?? 0);
+
+        if ($cfgBatch <= 0) {
+            $batchSize = $ispMax !== null ? max(0, $ispMax) : 0;
+        } else {
+            $batchSize = $ispMax !== null ? min($cfgBatch, max(1, $ispMax)) : $cfgBatch;
+        }
+
+        $batchPeriod = max(0, $cfgPeriod, $ispMinPeriod);
+
+        $throttleSec = max(0, $cfgThrottle, $ispMinThrottle);
+
+        $sentInBatch = 0;
+        $batchStart = microtime(true);
+
+        $this->updateMessageStatus($campaign, Message\MessageStatus::InProcess);
 
-        // phpcs:ignore Generic.Commenting.Todo
-        // @todo check $ISPrestrictions logic
         foreach ($subscribers as $subscriber) {
+            if ($batchSize > 0 && $batchPeriod > 0 && $sentInBatch >= $batchSize) {
+                $elapsed = microtime(true) - $batchStart;
+                $remaining = (int)ceil($batchPeriod - $elapsed);
+                if ($remaining > 0) {
+                    $output?->writeln(sprintf(
+                        'Batch limit reached, sleeping %ds to respect MAILQUEUE_BATCH_PERIOD',
+                        $remaining
+                    ));
+                    sleep($remaining);
+                }
+                $batchStart = microtime(true);
+                $sentInBatch = 0;
+            }
+
             if (!filter_var($subscriber->getEmail(), FILTER_VALIDATE_EMAIL)) {
                 continue;
             }
@@ -62,9 +107,7 @@ public function process(Message $campaign, ?OutputInterface $output = null): voi
 
             try {
                 $this->mailer->send($email);
-
-                // phpcs:ignore Generic.Commenting.Todo
-                // @todo log somewhere that this subscriber got email
+                $sentInBatch++;
             } catch (Throwable $e) {
                 $this->logger->error($e->getMessage(), [
                     'subscriber_id' => $subscriber->getId(),
@@ -73,10 +116,17 @@ public function process(Message $campaign, ?OutputInterface $output = null): voi
                 $output?->writeln('Failed to send to: ' . $subscriber->getEmail());
             }
 
-            usleep(100000);
+            if ($throttleSec > 0) {
+                sleep($throttleSec);
+            }
         }
 
-        $campaign->getMetadata()->setStatus(Message\MessageStatus::Sent);
+        $this->updateMessageStatus($campaign, Message\MessageStatus::Sent);
+    }
+
+    private function updateMessageStatus(Message $message, Message\MessageStatus $status): void
+    {
+        $message->getMetadata()->setStatus($status);
         $this->entityManager->flush();
     }
 }