Message processor

config/parameters.yml.dist

@@ -68,3 +68,16 @@ parameters:
 
     graylog_host: 'graylog.example.com'
     graylog_port: 12201
+
+    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'
+    messaging.max_process_time: '%%env(MESSAGING_MAX_PROCESS_TIME)%%'
+    env(MESSAGING_MAX_PROCESS_TIME): '600'

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,13 @@ services:
   PhpList\Core\Domain\Subscription\Service\Provider\SubscriberProvider:
     autowire: true
     autoconfigure: true
+
+  PhpList\Core\Core\ConfigProvider:
+    arguments:
+      $config: '%app.config%'
+
+  PhpList\Core\Domain\Common\IspRestrictionsProvider:
+    autowire: true
+    autoconfigure: true
+    arguments:
+      $confPath: '%app.phplist_isp_conf_path%'

config/services/services.yml

@@ -36,6 +36,14 @@ services:
     autoconfigure: true
     public: true
 
+  PhpList\Core\Domain\Messaging\Service\SendRateLimiter:
+    autowire: true
+    autoconfigure: 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
     autoconfigure: true
@@ -44,6 +52,10 @@ services:
     autowire: true
     autoconfigure: true
 
+  PhpList\Core\Domain\Messaging\Service\RateLimitedCampaignMailer:
+    autowire: true
+    autoconfigure: true
+
   PhpList\Core\Domain\Messaging\Service\ConsecutiveBounceHandler:
     autowire: true
     autoconfigure: true
@@ -108,6 +120,13 @@ services:
     arguments:
       - !tagged_iterator { tag: 'phplist.bounce_action_handler' }
 
+  PhpList\Core\Domain\Messaging\Service\MaxProcessTimeLimiter:
+    autowire: true
+    autoconfigure: true
+    arguments:
+      $maxSeconds: '%messaging.max_process_time%'
+
+
   PhpList\Core\Domain\Identity\Service\PermissionChecker:
     autowire: true
     autoconfigure: true

src/Domain/Common/IspRestrictionsProvider.php

@@ -0,0 +1,137 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Common;
+
+use PhpList\Core\Domain\Common\Model\IspRestrictions;
+use Psr\Log\LoggerInterface;
+
+class IspRestrictionsProvider
+{
+    public function __construct(
+        private readonly string $confPath,
+        private readonly LoggerInterface $logger,
+    ) {
+    }
+
+    public function load(): IspRestrictions
+    {
+        $contents = $this->readConfigFile();
+        if ($contents === null) {
+            return new IspRestrictions(null, null, null);
+        }
+
+        [$raw, $maxBatch, $minBatchPeriod, $lockFile] = $this->parseContents($contents);
+
+        $this->logIfDetected($maxBatch, $minBatchPeriod, $lockFile);
+
+        return new IspRestrictions($maxBatch, $minBatchPeriod, $lockFile, $raw);
+    }
+
+    private function readConfigFile(): ?string
+    {
+        if (!is_file($this->confPath) || !is_readable($this->confPath)) {
+            return null;
+        }
+        $contents = file_get_contents($this->confPath);
+        if ($contents === false) {
+            $this->logger->warning('Cannot read ISP restrictions file', ['path' => $this->confPath]);
+            return null;
+        }
+        return $contents;
+    }
+
+    /**
+     * @return array{0: array<string,string>, 1: ?int, 2: ?int, 3: ?string}
+     */
+    private function parseContents(string $contents): array
+    {
+        $maxBatch = null;
+        $minBatchPeriod = null;
+        $lockFile = null;
+        $raw = [];
+
+        foreach (preg_split('/\R/', $contents) as $line) {
+            [$key, $val] = $this->parseLine($line);
+            if ($key === null) {
+                continue;
+            }
+            $raw[$key] = $val;
+            [$maxBatch, $minBatchPeriod, $lockFile] = $this->applyKeyValue(
+                $key,
+                $val,
+                $maxBatch,
+                $minBatchPeriod,
+                $lockFile
+            );
+        }
+
+        return [$raw, $maxBatch, $minBatchPeriod, $lockFile];
+    }
+
+    /**
+     * @return array{0: ?string, 1: string}
+     */
+    private function parseLine(string $line): array
+    {
+        $line = trim($line);
+        if ($line === '' || str_starts_with($line, '#') || str_starts_with($line, ';')) {
+            return [null, ''];
+        }
+        $parts = explode('=', $line, 2);
+        if (\count($parts) !== 2) {
+            return [null, ''];
+        }
+
+        return array_map('trim', $parts);
+    }
+
+    /**
+     * @param string $key
+     * @param string $val
+     * @param ?int $maxBatch
+     * @param ?int $minBatchPeriod
+     * @param ?string $lockFile
+     * @return array{0: ?int, 1: ?int, 2: ?string}
+     */
+    private function applyKeyValue(
+        string $key,
+        string $val,
+        ?int $maxBatch,
+        ?int $minBatchPeriod,
+        ?string $lockFile
+    ): array {
+        if ($key === 'maxbatch') {
+            if ($val !== '' && ctype_digit($val)) {
+                $maxBatch = (int) $val;
+            }
+            return [$maxBatch, $minBatchPeriod, $lockFile];
+        }
+        if ($key === 'minbatchperiod') {
+            if ($val !== '' && ctype_digit($val)) {
+                $minBatchPeriod = (int) $val;
+            }
+            return [$maxBatch, $minBatchPeriod, $lockFile];
+        }
+        if ($key === 'lockfile') {
+            if ($val !== '') {
+                $lockFile = $val;
+            }
+            return [$maxBatch, $minBatchPeriod, $lockFile];
+        }
+        return [$maxBatch, $minBatchPeriod, $lockFile];
+    }
+
+    private function logIfDetected(?int $maxBatch, ?int $minBatchPeriod, ?string $lockFile): void
+    {
+        if ($maxBatch !== null || $minBatchPeriod !== null || $lockFile !== null) {
+            $this->logger->info('ISP restrictions detected', [
+                'path' => $this->confPath,
+                'maxbatch' => $maxBatch,
+                'minbatchperiod' => $minBatchPeriod,
+                'lockfile' => $lockFile,
+            ]);
+        }
+    }
+}

src/Domain/Common/Model/IspRestrictions.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Common\Model;
+
+final class IspRestrictions
+{
+    public function __construct(
+        public readonly ?int $maxBatch,
+        public readonly ?int $minBatchPeriod,
+        public readonly ?string $lockFile,
+        public readonly array $raw = [],
+    ) {
+    }
+
+    public function isEmpty(): bool
+    {
+        return $this->maxBatch === null && $this->minBatchPeriod === null && $this->lockFile === null;
+    }
+}

src/Domain/Configuration/Service/Manager/ConfigManager.php

@@ -17,6 +17,12 @@ public function __construct(ConfigRepository $configRepository)
         $this->configRepository = $configRepository;
     }
 
+    public function inMaintenanceMode(): bool
+    {
+        $config = $this->getByItem('maintenancemode');
+        return $config?->getValue() === '1';
+    }
+
     /**
      * Get a configuration item by its key
      */

src/Domain/Messaging/Command/ProcessQueueCommand.php

@@ -4,6 +4,9 @@
 
 namespace PhpList\Core\Domain\Messaging\Command;
 
+use DateTimeImmutable;
+use PhpList\Core\Domain\Configuration\Service\Manager\ConfigManager;
+use PhpList\Core\Domain\Messaging\Model\Message\MessageStatus;
 use PhpList\Core\Domain\Messaging\Repository\MessageRepository;
 use PhpList\Core\Domain\Messaging\Service\MessageProcessingPreparator;
 use PhpList\Core\Domain\Messaging\Service\Processor\CampaignProcessor;
@@ -24,18 +27,21 @@ class ProcessQueueCommand extends Command
     private LockFactory $lockFactory;
     private MessageProcessingPreparator $messagePreparator;
     private CampaignProcessor $campaignProcessor;
+    private ConfigManager $configManager;
 
     public function __construct(
         MessageRepository $messageRepository,
         LockFactory $lockFactory,
         MessageProcessingPreparator $messagePreparator,
         CampaignProcessor $campaignProcessor,
+        ConfigManager $configManager
     ) {
         parent::__construct();
         $this->messageRepository = $messageRepository;
         $this->lockFactory = $lockFactory;
         $this->messagePreparator = $messagePreparator;
         $this->campaignProcessor = $campaignProcessor;
+        $this->configManager = $configManager;
     }
 
     /**
@@ -50,11 +56,20 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             return Command::FAILURE;
         }
 
+        if ($this->configManager->inMaintenanceMode()) {
+            $output->writeln('The system is in maintenance mode, stopping. Try again later.');
+
+            return Command::FAILURE;
+        }
+
         try {
             $this->messagePreparator->ensureSubscribersHaveUuid($output);
             $this->messagePreparator->ensureCampaignsHaveUuid($output);
 
-            $campaigns = $this->messageRepository->findBy(['status' => 'submitted']);
+            $campaigns = $this->messageRepository->getByStatusAndEmbargo(
+                status: MessageStatus::Submitted,
+                embargo: new DateTimeImmutable()
+            );
 
             foreach ($campaigns as $campaign) {
                 $this->campaignProcessor->process($campaign, $output);

src/Domain/Messaging/Model/Dto/Message/MessageMetadataDto.php

@@ -4,10 +4,12 @@
 
 namespace PhpList\Core\Domain\Messaging\Model\Dto\Message;
 
+use PhpList\Core\Domain\Messaging\Model\Message\MessageStatus;
+
 class MessageMetadataDto
 {
     public function __construct(
-        public readonly string $status,
+        public readonly MessageStatus $status,
     ) {
     }
 }

src/Domain/Messaging/Model/Message/MessageMetadata.php

@@ -6,6 +6,7 @@
 
 use DateTime;
 use Doctrine\ORM\Mapping as ORM;
+use InvalidArgumentException;
 use PhpList\Core\Domain\Common\Model\Interfaces\EmbeddableInterface;
 
 #[ORM\Embeddable]
@@ -33,13 +34,13 @@ class MessageMetadata implements EmbeddableInterface
     private ?DateTime $sendStart;
 
     public function __construct(
-        ?string $status = null,
+        ?MessageStatus $status = null,
         int $bounceCount = 0,
         ?DateTime $entered = null,
         ?DateTime $sent = null,
         ?DateTime $sendStart = null,
     ) {
-        $this->status = $status;
+        $this->status = $status->value ?? null;
         $this->processed = false;
         $this->viewed = 0;
         $this->bounceCount = $bounceCount;
@@ -48,14 +49,21 @@ public function __construct(
         $this->sendStart = $sendStart;
     }
 
-    public function getStatus(): ?string
+    /**
+     * @SuppressWarnings("PHPMD.StaticAccess")
+     */
+    public function getStatus(): ?MessageStatus
     {
-        return $this->status;
+        return MessageStatus::from($this->status);
     }
 
-    public function setStatus(string $status): self
+    public function setStatus(MessageStatus $status): self
     {
-        $this->status = $status;
+        if (!$this->getStatus()->canTransitionTo($status)) {
+            throw new InvalidArgumentException('Invalid status transition');
+        }
+        $this->status = $status->value;
+
         return $this;
     }