ref: email sending

.coderabbit.yaml

@@ -35,7 +35,22 @@ reviews:
           - Still prefer keeping this logic well-encapsulated (e.g. in dedicated services/repos),
             not scattered across unrelated domain objects.
 
-        - ⚠️ For non-DynamicListAttr code:
+        - ✅ **Relaxed rule for atomic conditional-UPDATE reservation/claim patterns**:
+          - Some Domain repositories intentionally use `$connection->executeStatement(...)` to run a
+            conditional `UPDATE ... WHERE <guard>` (checking the affected-row count) or a guarded
+            `INSERT` (catching `UniqueConstraintViolationException`) as a portable, lock-free way to
+            atomically claim/reserve a row across concurrent workers - e.g.
+            `MessageRepository::tryClaimForProcessing` and
+            `DomainThrottleStateRepository::tryReserveSlot`/`resetBlockedCount`.
+          - This is not a domain-purity violation: it involves no `flush()`, transaction, or DDL, and
+            avoids vendor-specific upsert/locking syntax so it stays portable across MySQL/PostgreSQL/SQLite.
+          - Do *not* flag `executeStatement()` DML calls that follow this guarded-UPDATE/affected-rows
+            or catch-`UniqueConstraintViolationException`-on-INSERT pattern against the repository's own
+            entity table.
+          - Still flag any DBAL write that is unconditional, unrelated to claim/reservation semantics, or
+            that mutates a different entity's table than the repository owns.
+
+        - ⚠️ For other non-DynamicListAttr code:
           - If code is invoking actual table-creation, DDL execution, or schema synchronization,
             then request moving that to the Infrastructure or Application layer (e.g. MessageHandler).
           - Repositories in Domain should be abstractions without side effects; they should express *intent*,

.env.dist

@@ -78,9 +78,15 @@ MAILQUEUE_BATCH_SIZE=5
 MAILQUEUE_BATCH_PERIOD=5
 MAILQUEUE_THROTTLE=5
 MESSAGING_MAX_PROCESS_TIME=600
+MESSAGING_STUCK_CAMPAIGN_THRESHOLD=1800
 MAX_MAILSIZE=209715200
 DEFAULT_MESSAGEAGE=691200
 USE_MANUAL_TEXT_PART=0
+USE_LIST_EXCLUDE=0
+USE_DOMAIN_THROTTLE=0
+DOMAIN_BATCH_SIZE=1
+DOMAIN_BATCH_PERIOD=120
+DOMAIN_AUTO_THROTTLE=0
 MESSAGING_BLACKLIST_GRACE_TIME=600
 GOOGLE_SENDERID=
 USE_AMAZONSES=0

.env.test

@@ -1,3 +1,5 @@
 PHPLIST_DATABASE_DRIVER=pdo_sqlite
 PHPLIST_DATABASE_PATH=:memory:
 SEARCH_TRANSPORT_DSN=sync://
+ELASTICSEARCH_ENABLED=false
+

config/parameters.yml

@@ -76,9 +76,15 @@ parameters:
     messaging.mail_queue_period: '%env(MAILQUEUE_BATCH_PERIOD)%'
     messaging.mail_queue_throttle: '%env(MAILQUEUE_THROTTLE)%'
     messaging.max_process_time: '%env(MESSAGING_MAX_PROCESS_TIME)%'
+    messaging.stuck_campaign_threshold: '%env(int:MESSAGING_STUCK_CAMPAIGN_THRESHOLD)%'
     messaging.max_mail_size: '%env(MAX_MAILSIZE)%'
     messaging.default_message_age: '%env(DEFAULT_MESSAGEAGE)%'
     messaging.use_manual_text_part: '%env(USE_MANUAL_TEXT_PART)%'
+    messaging.use_list_exclude: '%env(bool:USE_LIST_EXCLUDE)%'
+    messaging.use_domain_throttle: '%env(bool:USE_DOMAIN_THROTTLE)%'
+    messaging.domain_batch_size: '%env(int:DOMAIN_BATCH_SIZE)%'
+    messaging.domain_batch_period: '%env(int:DOMAIN_BATCH_PERIOD)%'
+    messaging.domain_auto_throttle: '%env(bool:DOMAIN_AUTO_THROTTLE)%'
     messaging.blacklist_grace_time: '%env(MESSAGING_BLACKLIST_GRACE_TIME)%'
     messaging.google_sender_id: '%env(GOOGLE_SENDERID)%'
     messaging.use_amazon_ses: '%env(USE_AMAZONSES)%'

config/services/repositories.yml

@@ -185,6 +185,10 @@ services:
     arguments:
       - PhpList\Core\Domain\Messaging\Model\Attachment
 
+  PhpList\Core\Domain\Messaging\Repository\DomainThrottleStateRepository:
+    parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
+    arguments:
+      - PhpList\Core\Domain\Messaging\Model\DomainThrottleState
   PhpList\Core\Domain\Messaging\Repository\MessageAttachmentRepository:
     parent: PhpList\Core\Domain\Common\Repository\AbstractRepository
     arguments:

config/services/services.yml

@@ -73,6 +73,15 @@ services:
       $mailqueueBatchPeriod: '%messaging.mail_queue_period%'
       $mailqueueThrottle: '%messaging.mail_queue_throttle%'
 
+  PhpList\Core\Domain\Messaging\Service\DomainRateLimiter:
+    autowire: true
+    autoconfigure: true
+    arguments:
+      $enabled: '%messaging.use_domain_throttle%'
+      $domainBatchSize: '%messaging.domain_batch_size%'
+      $domainBatchPeriod: '%messaging.domain_batch_period%'
+      $autoThrottle: '%messaging.domain_auto_throttle%'
+
   PhpList\Core\Domain\Common\SystemInfoCollector:
     autowire: true
     autoconfigure: true

src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php

@@ -25,6 +25,7 @@
 use PhpList\Core\Domain\Messaging\Repository\UserMessageRepository;
 use PhpList\Core\Domain\Messaging\Service\Builder\EmailBuilder;
 use PhpList\Core\Domain\Messaging\Service\Builder\SystemEmailBuilder;
+use PhpList\Core\Domain\Messaging\Service\DomainRateLimiter;
 use PhpList\Core\Domain\Messaging\Service\Handler\RequeueHandler;
 use PhpList\Core\Domain\Messaging\Service\MailSizeChecker;
 use PhpList\Core\Domain\Messaging\Service\MaxProcessTimeLimiter;
@@ -72,13 +73,20 @@ public function __construct(
         private readonly EmailBuilder $campaignEmailBuilder,
         private readonly MailSizeChecker $mailSizeChecker,
         private readonly ConfigProvider $configProvider,
+        private readonly DomainRateLimiter $domainRateLimiter,
         #[Autowire('%imap_bounce.email%')] private readonly string $bounceEmail,
+        #[Autowire('%messaging.use_list_exclude%')] private readonly bool $useListExclude = false,
+        #[Autowire('%messaging.stuck_campaign_threshold%')] private readonly int $stuckCampaignThresholdSeconds = 0,
     ) {
     }
 
     public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $data): void
     {
-        $campaign = $this->messageRepository->findByIdAndStatus($data->getMessageId(), MessageStatus::Submitted);
+        // todo: recheck this stuckCampaignThresholdSeconds logic
+        $campaign = $this->messageRepository->tryClaimForProcessing(
+            $data->getMessageId(),
+            $this->stuckCampaignThresholdSeconds
+        );
         if (!$campaign) {
             $this->logger->warning(
                 $this->translator->trans('Campaign not found or not in submitted status'),
@@ -121,32 +129,17 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $
 
         $this->handleAdminNotifications($campaign, $loadedMessageData, $data->getMessageId());
 
-        $this->updateMessageStatus($campaign, MessageStatus::Prepared);
-        $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists($data, $campaign);
+        // Campaign was already atomically claimed into Prepared status above.
+        $excludeListIds = $this->getExcludeListIds($loadedMessageData);
+        $this->markExcludedSubscribers($campaign, $data, $excludeListIds);
+        $subscribers = $this->subscriberProvider->getSubscribersForMessageOrLists(
+            $data,
+            $campaign,
+            $excludeListIds
+        );
 
         $this->updateMessageStatus($campaign, MessageStatus::InProcess);
 
-//        if (USE_LIST_EXCLUDE) {
-//            if (VERBOSE) {
-//                processQueueOutput(s('looking for users who can be excluded from this mailing'));
-//            }
-//            if (count($msgdata['excludelist'])) {
-//                $query
-//                    = ' select userid'
-//                    .' from '.$GLOBALS['tables']['listuser']
-//                    .' where listid in ('.implode(',', $msgdata['excludelist']).')';
-//                if (VERBOSE) {
-//                    processQueueOutput('Exclude query '.$query);
-//                }
-//                $req = Sql_Query($query);
-//                while ($row = Sql_Fetch_Row($req)) {
-//                    $um = Sql_Query(sprintf('replace into %s (entered,userid,messageid,status)
-//                           values(now(),%d,%d,"excluded")',
-//                        $tables['usermessage'], $row[0], $messageid));
-//                }
-//            }
-//        }
-
         $stoppedEarly = $this->processSubscribersForCampaign($campaign, $subscribers, $cacheKey);
 
         if ($stoppedEarly && $this->requeueHandler->handle($campaign)) {
@@ -157,6 +150,72 @@ public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $
         $this->updateMessageStatus($campaign, MessageStatus::Sent);
     }
 
+    /**
+     * Exclude-list IDs are stored via MessageData as an array keyed by list ID  e.g. [3 => 1, 7 => 1].
+     *
+     * @return int[]
+     */
+    private function getExcludeListIds(array $loadedMessageData): array
+    {
+        if (!$this->useListExclude) {
+            return [];
+        }
+
+        $excludeList = $loadedMessageData['excludelist'] ?? [];
+        if (!is_array($excludeList) || $excludeList === []) {
+            return [];
+        }
+
+        return array_values(array_filter(array_map(
+            static fn (mixed $key): ?int => is_numeric($key) ? (int) $key : null,
+            array_keys($excludeList)
+        ), static fn (?int $id): bool => $id !== null));
+    }
+
+    /**
+     * pre-marking of exclude-list members as "excluded" in usermessage before the main send loop runs,
+     * so there's a persisted audit trail for why a subscriber wasn't sent to. Skips
+     * subscribers who already have a nontodo UserMessage for this campaign, so a later run
+     * can't clobber an already-recorded Sent/NotSent/etc. status from an earlier partial run.
+     * Only campaign recipients (i.e. subscribers who'd otherwise be sent this campaign) are
+     * marked, since a subscriber on an exclude list who isn't a campaign recipient anyway
+     * shouldn't get an exclusion record.
+     */
+    private function markExcludedSubscribers(
+        Message $campaign,
+        CampaignProcessorMessage|SyncCampaignProcessorMessage $data,
+        array $excludeListIds,
+    ): void {
+        if ($excludeListIds === []) {
+            return;
+        }
+
+        $excludedSubscribers = $this->subscriberProvider->getExcludedSubscribers($excludeListIds);
+        if ($excludedSubscribers === []) {
+            return;
+        }
+
+        $sendableSubscribers = $this->subscriberProvider->getSendableSubscribersForMessageOrLists(
+            $data,
+            $campaign
+        );
+
+        foreach ($excludedSubscribers as $subscriber) {
+            if (!isset($sendableSubscribers[$subscriber->getEmail()])) {
+                continue;
+            }
+
+            $existing = $this->userMessageRepository->findByUserAndMessage($subscriber, $campaign);
+            if ($existing && $existing->getStatus() !== UserMessageStatus::Todo) {
+                continue;
+            }
+
+            $userMessage = $existing ?? new UserMessage($subscriber, $campaign);
+            $userMessage->setStatus(UserMessageStatus::Excluded);
+            $this->userMessageRepository->save($userMessage);
+        }
+    }
+
     private function unconfirmSubscriber(Subscriber $subscriber): void
     {
         if ($subscriber->isConfirmed()) {
@@ -170,6 +229,9 @@ private function updateMessageStatus(Message $message, MessageStatus $status): v
         if ($status === MessageStatus::InProcess && $message->getMetadata()->getSendStart() === null) {
             $message->getMetadata()->setSendStart(new DateTime());
         }
+        if ($status === MessageStatus::Sent) {
+            $message->getMetadata()->setSent(new DateTime());
+        }
         $message->getMetadata()->setStatus($status);
         $this->entityManager->flush();
     }
@@ -220,6 +282,9 @@ private function handleEmailSending(
                 htmlPref: $subscriber->hasHtmlEmail(),
             );
             if ($result === null) {
+                $status = $subscriber->isBlacklisted() ? UserMessageStatus::Excluded : UserMessageStatus::NotSent;
+                $this->updateUserMessageStatus($userMessage, $status);
+
                 return;
             }
             [$email, $sentAs] = $result;
@@ -228,7 +293,7 @@ private function handleEmailSending(
             $this->rateLimitedCampaignMailer->send($email);
             ($this->mailSizeChecker)($campaign, $email, $subscriber->hasHtmlEmail());
             $this->updateUserMessageStatus($userMessage, UserMessageStatus::Sent);
-            $campaign->incrementSentCount($sentAs);
+            $this->messageRepository->incrementSentCounts($campaign->getId(), $sentAs);
         } catch (MessageSizeLimitExceededException $e) {
             // stop after the first message if size is exceeded
             $this->updateMessageStatus($campaign, MessageStatus::Suspended);
@@ -330,6 +395,13 @@ private function processSubscribersForCampaign(Message $campaign, array $subscri
                 continue;
             }
 
+            if (!$this->domainRateLimiter->attemptSend($subscriber->getEmail())->allowed) {
+                // Leave no UserMessage record so this subscriber is picked up again on a
+                // later run, once their domain's throttle window has passed.
+                $stoppedEarly = true;
+                continue;
+            }
+
             $userMessage = $existing ?? new UserMessage($subscriber, $campaign);
             $userMessage->setStatus(UserMessageStatus::Active);
             $this->userMessageRepository->save($userMessage);

src/Domain/Messaging/Model/DomainThrottleState.php

@@ -0,0 +1,62 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Messaging\Model;
+
+use Doctrine\ORM\Mapping as ORM;
+use PhpList\Core\Domain\Common\Model\Interfaces\DomainModel;
+use PhpList\Core\Domain\Messaging\Repository\DomainThrottleStateRepository;
+
+/**
+ * Per-domain send counters for a single fixed throttle window, persisted so that
+ * DomainRateLimiter enforces DOMAIN_BATCH_SIZE/DOMAIN_BATCH_PERIOD consistently across
+ * concurrent queue-processing workers instead of each worker keeping its own count.
+ * Rows are read/written exclusively via DomainThrottleStateRepository's atomic
+ * UPDATE/INSERT statements, not through the entity manager's persist/flush.
+ */
+#[ORM\Entity(repositoryClass: DomainThrottleStateRepository::class)]
+#[ORM\Table(name: 'domain_throttle')]
+class DomainThrottleState implements DomainModel
+{
+    #[ORM\Id]
+    #[ORM\Column(name: 'domain', type: 'string', length: 255)]
+    private string $domain;
+
+    #[ORM\Column(name: 'window_start', type: 'integer')]
+    private int $windowStart;
+
+    #[ORM\Column(name: 'sent_count', type: 'integer')]
+    private int $sentCount;
+
+    #[ORM\Column(name: 'blocked_count', type: 'integer')]
+    private int $blockedCount;
+
+    public function __construct(string $domain, int $windowStart, int $sentCount = 0, int $blockedCount = 0)
+    {
+        $this->domain = $domain;
+        $this->windowStart = $windowStart;
+        $this->sentCount = $sentCount;
+        $this->blockedCount = $blockedCount;
+    }
+
+    public function getDomain(): string
+    {
+        return $this->domain;
+    }
+
+    public function getWindowStart(): int
+    {
+        return $this->windowStart;
+    }
+
+    public function getSentCount(): int
+    {
+        return $this->sentCount;
+    }
+
+    public function getBlockedCount(): int
+    {
+        return $this->blockedCount;
+    }
+}

src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Messaging\Model\Dto;
+
+/**
+ * Outcome of an atomic slot reservation attempt in DomainThrottleStateRepository.
+ */
+final class DomainThrottleReservation
+{
+    public function __construct(
+        public readonly bool $allowed,
+        public readonly int $blockedAttempts = 0,
+    ) {
+    }
+}

src/Domain/Messaging/Model/Dto/DomainThrottleResult.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PhpList\Core\Domain\Messaging\Model\Dto;
+
+/**
+ * Outcome of DomainRateLimiter::attemptSend() for a single recipient.
+ */
+final class DomainThrottleResult
+{
+    public function __construct(
+        public readonly bool $allowed,
+        public readonly ?string $domain,
+        public readonly int $blockedAttempts = 0,
+        public readonly bool $backoffApplied = false,
+        public readonly int $backoffSeconds = 0,
+    ) {
+    }
+}

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

@@ -131,7 +131,6 @@ public function setEntered(?DateTime $entered): self
         return $this;
     }
 
-    // todo: set sent to the time when it was sent
     public function setSent(?DateTime $sent): self
     {
         $this->sent = $sent;

src/Domain/Messaging/Model/UserMessage.php

@@ -66,9 +66,9 @@ public function getViewed(): ?DateTime
         return $this->viewed;
     }
 
-    public function getStatus(): ?UserMessageStatus
+    public function getStatus(): UserMessageStatus
     {
-        return UserMessageStatus::from($this->status);
+        return $this->status !== null ? UserMessageStatus::from($this->status) : UserMessageStatus::Todo;
     }
 
     public function setViewed(?DateTime $viewed): self