diff --git a/.env.dist b/.env.dist index e66a76e4..a9738072 100644 --- a/.env.dist +++ b/.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 diff --git a/.env.test b/.env.test index 88b6c338..a23613db 100644 --- a/.env.test +++ b/.env.test @@ -1,3 +1,5 @@ PHPLIST_DATABASE_DRIVER=pdo_sqlite PHPLIST_DATABASE_PATH=:memory: SEARCH_TRANSPORT_DSN=sync:// +ELASTICSEARCH_ENABLED=false + diff --git a/config/parameters.yml b/config/parameters.yml index 1edd106d..772774d1 100644 --- a/config/parameters.yml +++ b/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)%' diff --git a/config/services/repositories.yml b/config/services/repositories.yml index 5ee7eb40..4cb9d01b 100644 --- a/config/services/repositories.yml +++ b/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: diff --git a/config/services/services.yml b/config/services/services.yml index 9e527c13..4d0b4514 100644 --- a/config/services/services.yml +++ b/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 diff --git a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php b/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php index ad8d0f48..1cebf5a9 100644 --- a/src/Domain/Messaging/MessageHandler/CampaignProcessor/CampaignProcessorMessageHandler.php +++ b/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,15 @@ 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, ) { } public function __invoke(CampaignProcessorMessage|SyncCampaignProcessorMessage $data): void { - $campaign = $this->messageRepository->findByIdAndStatus($data->getMessageId(), MessageStatus::Submitted); + $campaign = $this->messageRepository->tryClaimForProcessing($data->getMessageId()); if (!$campaign) { $this->logger->warning( $this->translator->trans('Campaign not found or not in submitted status'), @@ -121,32 +124,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, $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 +145,52 @@ 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. + */ + private function markExcludedSubscribers(Message $campaign, array $excludeListIds): void + { + if ($excludeListIds === []) { + return; + } + + foreach ($this->subscriberProvider->getExcludedSubscribers($excludeListIds) as $subscriber) { + $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 +204,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 +257,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 +268,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 +370,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); diff --git a/src/Domain/Messaging/Model/DomainThrottleState.php b/src/Domain/Messaging/Model/DomainThrottleState.php new file mode 100644 index 00000000..13035049 --- /dev/null +++ b/src/Domain/Messaging/Model/DomainThrottleState.php @@ -0,0 +1,62 @@ +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; + } +} diff --git a/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php b/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php new file mode 100644 index 00000000..652286fb --- /dev/null +++ b/src/Domain/Messaging/Model/Dto/DomainThrottleReservation.php @@ -0,0 +1,17 @@ +sent = $sent; diff --git a/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php b/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php new file mode 100644 index 00000000..a1d38673 --- /dev/null +++ b/src/Domain/Messaging/Repository/DomainThrottleStateRepository.php @@ -0,0 +1,130 @@ +getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + + if ($this->incrementSentIfAllowed($connection, $table, $domain, $windowStart, $batchSize)) { + return new DomainThrottleReservation(allowed: true); + } + + if ($this->rolloverWindow($connection, $table, $domain, $windowStart)) { + return new DomainThrottleReservation(allowed: true); + } + + if ($this->insertFirstRow($connection, $table, $domain, $windowStart)) { + return new DomainThrottleReservation(allowed: true); + } + + // Lost the insert race to another worker; its row may already have room in this + // window, so give the increment one more try before concluding we're blocked. + if ($this->incrementSentIfAllowed($connection, $table, $domain, $windowStart, $batchSize)) { + return new DomainThrottleReservation(allowed: true); + } + + return new DomainThrottleReservation( + allowed: false, + blockedAttempts: $this->incrementBlocked($connection, $table, $domain, $windowStart), + ); + } + + public function resetBlockedCount(string $domain, int $windowStart): void + { + $connection = $this->getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + + $connection->executeStatement( + sprintf('UPDATE %s SET blocked_count = 0 WHERE domain = :domain AND window_start = :window', $table), + ['domain' => $domain, 'window' => $windowStart] + ); + } + + /** @phpstan-impure */ + private function incrementSentIfAllowed( + Connection $connection, + string $table, + string $domain, + int $windowStart, + int $batchSize + ): bool { + $affected = $connection->executeStatement( + sprintf( + 'UPDATE %s SET sent_count = sent_count + 1 + WHERE domain = :domain AND window_start = :window AND sent_count < :batchSize', + $table + ), + ['domain' => $domain, 'window' => $windowStart, 'batchSize' => $batchSize] + ); + + return $affected > 0; + } + + /** @phpstan-impure */ + private function rolloverWindow(Connection $connection, string $table, string $domain, int $windowStart): bool + { + $affected = $connection->executeStatement( + sprintf( + 'UPDATE %s SET window_start = :window, sent_count = 1, blocked_count = 0 + WHERE domain = :domain AND window_start < :window', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return $affected > 0; + } + + /** @phpstan-impure */ + private function insertFirstRow(Connection $connection, string $table, string $domain, int $windowStart): bool + { + try { + $connection->executeStatement( + sprintf( + 'INSERT INTO %s (domain, window_start, sent_count, blocked_count) VALUES (:domain, :window, 1, 0)', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return true; + } catch (UniqueConstraintViolationException) { + return false; + } + } + + /** @phpstan-impure */ + private function incrementBlocked(Connection $connection, string $table, string $domain, int $windowStart): int + { + $connection->executeStatement( + sprintf( + 'UPDATE %s SET blocked_count = blocked_count + 1 WHERE domain = :domain AND window_start = :window', + $table + ), + ['domain' => $domain, 'window' => $windowStart] + ); + + return (int) $connection->fetchOne( + sprintf('SELECT blocked_count FROM %s WHERE domain = :domain AND window_start = :window', $table), + ['domain' => $domain, 'window' => $windowStart] + ); + } +} diff --git a/src/Domain/Messaging/Repository/MessageRepository.php b/src/Domain/Messaging/Repository/MessageRepository.php index 13394794..414e27ed 100644 --- a/src/Domain/Messaging/Repository/MessageRepository.php +++ b/src/Domain/Messaging/Repository/MessageRepository.php @@ -4,6 +4,7 @@ namespace PhpList\Core\Domain\Messaging\Repository; +use DateTime; use DateTimeImmutable; use DateTimeInterface; use Doctrine\ORM\AbstractQuery; @@ -11,6 +12,7 @@ use PhpList\Core\Domain\Common\Model\PaginatedResult; use PhpList\Core\Domain\Common\Repository\AbstractRepository; use PhpList\Core\Domain\Common\Repository\Interfaces\PaginatableRepositoryInterface; +use PhpList\Core\Domain\Configuration\Model\OutputFormat; use PhpList\Core\Domain\Messaging\Model\Filter\MessageFilter; use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Subscription\Model\SubscriberList; @@ -158,6 +160,85 @@ public function findByIdAndStatus(int $id, Message\MessageStatus $status): ?Mess ->getOneOrNullResult(); } + /** + * Atomically claims a campaign for processing by flipping its status from Submitted to + * Prepared in a single UPDATE ... WHERE statement, so two concurrent workers can't both + * pass a check-then-act race and process the same campaign. + */ + public function tryClaimForProcessing(int $id): ?Message + { + $connection = $this->getEntityManager()->getConnection(); + $table = $connection->quoteIdentifier($this->getClassMetadata()->getTableName()); + $now = new DateTime(); + + $sql = sprintf('UPDATE %s SET status = :to, modified = :now WHERE id = :id AND status = :from', $table); + $params = [ + 'to' => Message\MessageStatus::Prepared->value, + 'now' => $now->format('Y-m-d H:i:s'), + 'id' => $id, + 'from' => Message\MessageStatus::Submitted->value, + ]; + + $affected = $connection->executeStatement($sql, $params); + + if ($affected === 0) { + return null; + } + + return $this->find($id); + } + + /** + * Returns campaigns stuck in Prepared/InProcess whose row hasn't been touched since + * $staleBefore, i.e. candidates for tryClaimForProcessing's stale-reclaim path. Callers + * are expected to re-dispatch a CampaignProcessorMessage for each, since nothing else + * automatically resumes a campaign that isn't in Submitted status. + * + * @return Message[] + */ + public function getStuckInProcessing(DateTimeImmutable $staleBefore): array + { + return $this->createQueryBuilder('m') + ->where('m.metadata.status IN (:statuses)') + ->andWhere('m.updatedAt < :staleBefore') + ->setParameter('statuses', [ + Message\MessageStatus::Prepared->value, + Message\MessageStatus::InProcess->value, + ]) + ->setParameter('staleBefore', $staleBefore) + ->getQuery() + ->getResult(); + } + + /** + * Atomically increments a campaign's processed/format-sent counters directly in the + * database (bypassing the entity's in-memory incrementSentCount()), so concurrent + * updates to the same campaign can't lose an update the way a read-modify-write via + * the entity manager could. Also bumps `modified`, since this is the liveness signal + * tryClaimForProcessing's stale-reclaim relies on. + */ + public function incrementSentCounts(int $messageId, OutputFormat $sentAs): void + { + $formatField = match ($sentAs) { + OutputFormat::Html => 'm.format.asHtml', + OutputFormat::Text => 'm.format.asText', + OutputFormat::Pdf => 'm.format.asPdf', + OutputFormat::TextAndHtml => 'm.format.asTextAndHtml', + OutputFormat::TextAndPdf => 'm.format.asTextAndPdf', + }; + + $this->createQueryBuilder('m') + ->update() + ->set('m.metadata.processed', 'm.metadata.processed + 1') + ->set($formatField, $formatField . ' + 1') + ->set('m.updatedAt', ':now') + ->where('m.id = :id') + ->setParameter('now', new DateTime()) + ->setParameter('id', $messageId) + ->getQuery() + ->execute(); + } + public function getNonEmptyFields(int $id): array { $message = $this->createQueryBuilder('m') diff --git a/src/Domain/Messaging/Service/DomainRateLimiter.php b/src/Domain/Messaging/Service/DomainRateLimiter.php new file mode 100644 index 00000000..900e2f81 --- /dev/null +++ b/src/Domain/Messaging/Service/DomainRateLimiter.php @@ -0,0 +1,110 @@ +enabled || $this->domainBatchSize <= 0 || $this->domainBatchPeriod <= 0) { + return new DomainThrottleResult(allowed: true, domain: null); + } + + $domain = $this->extractDomain($email); + if ($domain === null) { + return new DomainThrottleResult(allowed: true, domain: null); + } + + $windowStart = intdiv(time(), $this->domainBatchPeriod) * $this->domainBatchPeriod; + $reservation = $this->repository->tryReserveSlot($domain, $windowStart, $this->domainBatchSize); + + if ($reservation->allowed) { + return new DomainThrottleResult(allowed: true, domain: $domain); + } + + $this->logger->info('Send blocked by domain throttle', [ + 'domain' => $domain, + 'blocked_attempts' => $reservation->blockedAttempts, + 'domain_batch_size' => $this->domainBatchSize, + 'domain_batch_period' => $this->domainBatchPeriod, + ]); + + return $this->applyAutoThrottleIfDue($domain, $windowStart, $reservation->blockedAttempts); + } + + private function applyAutoThrottleIfDue( + string $domain, + int $windowStart, + int $blockedAttempts + ): DomainThrottleResult { + if (!$this->autoThrottle || $blockedAttempts <= self::AUTO_THROTTLE_ATTEMPT_THRESHOLD) { + return new DomainThrottleResult(allowed: false, domain: $domain, blockedAttempts: $blockedAttempts); + } + + // Reset the trigger counter so it takes another full run of blocked attempts + // before backoff fires again for this domain/window. + $this->repository->resetBlockedCount($domain, $windowStart); + $delaySeconds = max(1, intdiv($this->domainBatchPeriod, max(1, $this->domainBatchSize * 4))); + + $this->logger->info('Introducing extra delay to reduce domain throttle failures', [ + 'domain' => $domain, + 'delay_seconds' => $delaySeconds, + ]); + sleep($delaySeconds); + + return new DomainThrottleResult( + allowed: false, + domain: $domain, + blockedAttempts: $blockedAttempts, + backoffApplied: true, + backoffSeconds: $delaySeconds, + ); + } + + private function extractDomain(string $email): ?string + { + $atPosition = strrpos($email, '@'); + if ($atPosition === false) { + return null; + } + + return strtolower(substr($email, $atPosition + 1)); + } +} diff --git a/src/Domain/Messaging/Service/Handler/RequeueHandler.php b/src/Domain/Messaging/Service/Handler/RequeueHandler.php index 3fbca634..0d028023 100644 --- a/src/Domain/Messaging/Service/Handler/RequeueHandler.php +++ b/src/Domain/Messaging/Service/Handler/RequeueHandler.php @@ -14,6 +14,16 @@ class RequeueHandler { + /** + * Fallback delay (minutes) used when a campaign stops early (time limit, domain throttle, + * etc.) but has no explicit requeueInterval configured. requeueInterval/requeueUntil control + * *how long* to wait before resuming, not *whether* to resume: a campaign that stopped early + * must always be retried, mirroring phplist3's unconditional "don't mark sent while anything + * failed/was throttled" guard - it must never be silently marked Sent with recipients still + * unprocessed. requeueUntil remains a legitimate opt-out (a real deadline). + */ + private const DEFAULT_REQUEUE_INTERVAL_MINUTES = 1; + public function __construct( private readonly LoggerInterface $logger, private readonly TranslatorInterface $translator, @@ -24,11 +34,11 @@ public function handle(Message $campaign, ?OutputInterface $output = null): bool { $schedule = $campaign->getSchedule(); $interval = $schedule->getRequeueInterval() ?? 0; - $until = $schedule->getRequeueUntil(); - if ($interval <= 0) { - return false; + $interval = self::DEFAULT_REQUEUE_INTERVAL_MINUTES; } + $until = $schedule->getRequeueUntil(); + $now = new DateTime(); if ($until instanceof DateTime && $now > $until) { return false; diff --git a/src/Domain/Messaging/Service/Manager/MessageManager.php b/src/Domain/Messaging/Service/Manager/MessageManager.php index 7ed34594..bd7b9315 100644 --- a/src/Domain/Messaging/Service/Manager/MessageManager.php +++ b/src/Domain/Messaging/Service/Manager/MessageManager.php @@ -4,6 +4,7 @@ namespace PhpList\Core\Domain\Messaging\Service\Manager; +use DateTimeImmutable; use PhpList\Core\Domain\Identity\Model\Administrator; use PhpList\Core\Domain\Messaging\Model\Dto\MessageContext; use PhpList\Core\Domain\Messaging\Model\Dto\MessageDtoInterface; @@ -101,6 +102,12 @@ public function getMessagesByOwner(Administrator $owner): array return $this->messageRepository->getByOwnerId($owner->getId()); } + /** @return Message[] */ + public function getStuckCampaigns(DateTimeImmutable $staleBefore): array + { + return $this->messageRepository->getStuckInProcessing($staleBefore); + } + private function canBeSubmitted(Message $message): bool { return $message->getListMessages()->count() > 0 diff --git a/src/Domain/Subscription/Repository/SubscriberRepository.php b/src/Domain/Subscription/Repository/SubscriberRepository.php index e9def63e..2f20b334 100644 --- a/src/Domain/Subscription/Repository/SubscriberRepository.php +++ b/src/Domain/Subscription/Repository/SubscriberRepository.php @@ -73,6 +73,52 @@ public function getSubscribersBySubscribedListId(int $listId): array ->getResult(); } + /** + * Same as getSubscribersBySubscribedListId(), but restricted to subscribers who are + * confirmed and not disabled - i.e. eligible to receive a campaign. Blacklisting is + * intentionally not filtered here since it's checked live against UserBlacklistRepository + * at send time instead of the (potentially stale) Subscriber::$blacklisted flag. + * + * @return Subscriber[] + */ + public function getSendableSubscribersBySubscribedListId(int $listId): array + { + return $this->createQueryBuilder('s') + ->innerJoin('s.subscriptions', 'subscription') + ->innerJoin('subscription.subscriberList', 'list') + ->where('list.id = :listId') + ->andWhere('s.confirmed = :confirmed') + ->andWhere('s.disabled = :disabled') + ->setParameter('listId', $listId) + ->setParameter('confirmed', true) + ->setParameter('disabled', false) + ->getQuery() + ->getResult(); + } + + /** + * Returns all subscribers on any of the given lists, regardless of confirmed/disabled + * status - used to resolve campaign exclude-lists, where membership alone is enough + * to suppress a send. + * + * @param int[] $listIds + * @return Subscriber[] + */ + public function getSubscribersBySubscribedListIds(array $listIds): array + { + if ($listIds === []) { + return []; + } + + return $this->createQueryBuilder('s') + ->innerJoin('s.subscriptions', 'subscription') + ->innerJoin('subscription.subscriberList', 'list') + ->where('list.id IN (:listIds)') + ->setParameter('listIds', $listIds) + ->getQuery() + ->getResult(); + } + /** * @return PaginatedResult * @throws InvalidArgumentException diff --git a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php index 758db32e..31474ef6 100644 --- a/src/Domain/Subscription/Service/Provider/SubscriberProvider.php +++ b/src/Domain/Subscription/Service/Provider/SubscriberProvider.php @@ -29,10 +29,15 @@ public function __construct( * * @param CampaignProcessorMessageInterface $data * @param Message $campaign + * @param int[] $excludeListIds List IDs whose members should be suppressed from the send, + * regardless of their confirmed/disabled status. * @return Subscriber[] Array of subscribers */ - public function getSubscribersForMessageOrLists(CampaignProcessorMessageInterface $data, Message $campaign): array - { + public function getSubscribersForMessageOrLists( + CampaignProcessorMessageInterface $data, + Message $campaign, + array $excludeListIds = [], + ): array { if ($data instanceof TestCampaignProcessorMessage) { return $this->subscriberRepository->getByEmails($data->getSubscriberEmails()); } @@ -45,12 +50,32 @@ public function getSubscribersForMessageOrLists(CampaignProcessorMessageInterfac $subscribers = []; foreach ($listIds as $listId) { - $listSubscribers = $this->subscriberRepository->getSubscribersBySubscribedListId($listId); + $listSubscribers = $this->subscriberRepository->getSendableSubscribersBySubscribedListId($listId); foreach ($listSubscribers as $subscriber) { $subscribers[$subscriber->getEmail()] = $subscriber; } } + foreach ($this->getExcludedSubscribers($excludeListIds) as $excluded) { + unset($subscribers[$excluded->getEmail()]); + } + return array_values($subscribers); } + + /** + * Resolves the subscribers on the given exclude-lists, regardless of confirmed/disabled + * status - membership alone is enough to suppress a send. + * + * @param int[] $excludeListIds + * @return Subscriber[] + */ + public function getExcludedSubscribers(array $excludeListIds): array + { + if ($excludeListIds === []) { + return []; + } + + return $this->subscriberRepository->getSubscribersBySubscribedListIds($excludeListIds); + } } diff --git a/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php b/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php new file mode 100644 index 00000000..f1cddf57 --- /dev/null +++ b/src/Migrations/Version20260908130000MySqlCreateDomainThrottleTable.php @@ -0,0 +1,46 @@ +connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof MySQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql( + 'CREATE TABLE phplist_domain_throttle ( + domain VARCHAR(255) NOT NULL, + window_start INT NOT NULL, + sent_count INT NOT NULL DEFAULT 0, + blocked_count INT NOT NULL DEFAULT 0, + PRIMARY KEY (domain) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3' + ); + } + + public function down(Schema $schema): void + { + $platform = $this->connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof MySQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql('DROP TABLE phplist_domain_throttle'); + } +} \ No newline at end of file diff --git a/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php b/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php new file mode 100644 index 00000000..f7835465 --- /dev/null +++ b/src/Migrations/Version20260908130001PostGreCreateDomainThrottleTable.php @@ -0,0 +1,46 @@ +connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof PostgreSQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql( + 'CREATE TABLE phplist_domain_throttle ( + domain VARCHAR(255) NOT NULL, + window_start INT NOT NULL, + sent_count INT NOT NULL DEFAULT 0, + blocked_count INT NOT NULL DEFAULT 0, + PRIMARY KEY (domain) + )' + ); + } + + public function down(Schema $schema): void + { + $platform = $this->connection->getDatabasePlatform(); + $this->skipIf(!$platform instanceof PostgreSQLPlatform, sprintf( + 'Unsupported platform for this migration: %s', + get_class($platform) + )); + + $this->addSql('DROP TABLE phplist_domain_throttle'); + } +} \ No newline at end of file diff --git a/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php new file mode 100644 index 00000000..38f32fa8 --- /dev/null +++ b/tests/Integration/Domain/Messaging/Repository/DomainThrottleStateRepositoryTest.php @@ -0,0 +1,86 @@ +loadSchema(); + + $this->repository = self::getContainer()->get(DomainThrottleStateRepository::class); + } + + protected function tearDown(): void + { + $schemaTool = new SchemaTool($this->entityManager); + $schemaTool->dropDatabase(); + parent::tearDown(); + } + + public function testFirstReservationForNewDomainIsAllowed(): void + { + $reservation = $this->repository->tryReserveSlot('example.com', 1000, 1); + + $this->assertTrue($reservation->allowed); + $this->assertSame(0, $reservation->blockedAttempts); + } + + public function testReservationBlockedOnceQuotaReachedInSameWindow(): void + { + $this->assertTrue($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + + $second = $this->repository->tryReserveSlot('example.com', 1000, 1); + + $this->assertFalse($second->allowed); + $this->assertSame(1, $second->blockedAttempts); + + $third = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertFalse($third->allowed); + $this->assertSame(2, $third->blockedAttempts); + } + + public function testDomainsAreTrackedIndependently(): void + { + $this->assertTrue($this->repository->tryReserveSlot('a.com', 1000, 1)->allowed); + + $this->assertTrue($this->repository->tryReserveSlot('b.com', 1000, 1)->allowed); + $this->assertFalse($this->repository->tryReserveSlot('a.com', 1000, 1)->allowed); + } + + public function testReservationAllowedAgainAfterWindowRollsOver(): void + { + $this->assertTrue($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + $this->assertFalse($this->repository->tryReserveSlot('example.com', 1000, 1)->allowed); + + $nextWindow = $this->repository->tryReserveSlot('example.com', 1120, 1); + + $this->assertTrue($nextWindow->allowed); + } + + public function testResetBlockedCountClearsCounterForCurrentWindow(): void + { + $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->repository->tryReserveSlot('example.com', 1000, 1); + $blocked = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertSame(2, $blocked->blockedAttempts); + + $this->repository->resetBlockedCount('example.com', 1000); + + $afterReset = $this->repository->tryReserveSlot('example.com', 1000, 1); + $this->assertFalse($afterReset->allowed); + $this->assertSame(1, $afterReset->blockedAttempts); + } +} diff --git a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php index 7bd83207..baee85b2 100644 --- a/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php +++ b/tests/Integration/Domain/Messaging/Repository/MessageRepositoryTest.php @@ -221,6 +221,44 @@ public function testGetFilteredAfterIdDefaultsToAscendingOrder(): void self::assertSame($second->getId(), $result->getItems()[1]->getId()); } + public function testTryClaimForProcessingClaimsSubmittedCampaign(): void + { + $message = $this->persistMessage(Message\MessageStatus::Submitted, 'Ready to send'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + $claimed = $this->messageRepository->tryClaimForProcessing($id); + + self::assertNotNull($claimed); + self::assertSame($id, $claimed->getId()); + self::assertSame(Message\MessageStatus::Prepared, $claimed->getMetadata()->getStatus()); + } + + public function testTryClaimForProcessingReturnsNullWhenNotSubmitted(): void + { + $message = $this->persistMessage(Message\MessageStatus::Draft, 'Not ready yet'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + self::assertNull($this->messageRepository->tryClaimForProcessing($id)); + } + + public function testTryClaimForProcessingCannotClaimTwice(): void + { + $message = $this->persistMessage(Message\MessageStatus::Submitted, 'Only one winner'); + $this->entityManager->flush(); + $id = $message->getId(); + $this->entityManager->clear(); + + $firstClaim = $this->messageRepository->tryClaimForProcessing($id); + $secondClaim = $this->messageRepository->tryClaimForProcessing($id); + + self::assertNotNull($firstClaim); + self::assertNull($secondClaim); + } + public function testGetFilteredAfterIdSortsDescendingAndCursorsBackward(): void { $first = $this->persistMessage(Message\MessageStatus::Sent, 'First'); diff --git a/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php b/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php index 2aa89b25..1a091bae 100644 --- a/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php +++ b/tests/Integration/Domain/Subscription/Repository/SubscriberRepositoryTest.php @@ -8,6 +8,7 @@ use Doctrine\DBAL\Exception\UniqueConstraintViolationException; use Doctrine\ORM\Tools\SchemaTool; use PhpList\Core\Domain\Subscription\Model\Subscriber; +use PhpList\Core\Domain\Subscription\Model\SubscriberList; use PhpList\Core\Domain\Subscription\Model\Subscription; use PhpList\Core\Domain\Subscription\Repository\SubscriberRepository; use PhpList\Core\Domain\Subscription\Repository\SubscriptionRepository; @@ -239,4 +240,65 @@ public function testRemoveRemovesModel() $numberOfModelsAfterRemove = count($this->subscriberRepository->findAll()); self::assertSame(1, $numberOfModelsBeforeRemove - $numberOfModelsAfterRemove); } + + private function subscribe(Subscriber $subscriber, SubscriberList $list): void + { + $subscription = (new Subscription()) + ->setSubscriber($subscriber) + ->setSubscriberList($list); + $this->entityManager->persist($subscription); + } + + public function testGetSendableSubscribersBySubscribedListIdExcludesUnconfirmedAndDisabled(): void + { + $list = (new SubscriberList())->setName('list'); + $this->entityManager->persist($list); + + $confirmed = (new Subscriber('confirmed@example.com'))->setConfirmed(true); + $unconfirmed = (new Subscriber('unconfirmed@example.com'))->setConfirmed(false); + $disabled = (new Subscriber('disabled@example.com'))->setConfirmed(true)->setDisabled(true); + foreach ([$confirmed, $unconfirmed, $disabled] as $subscriber) { + $this->entityManager->persist($subscriber); + $this->subscribe($subscriber, $list); + } + $this->entityManager->flush(); + + $result = $this->subscriberRepository->getSendableSubscribersBySubscribedListId($list->getId()); + + self::assertTrue(in_array($confirmed, $result, true)); + self::assertFalse(in_array($unconfirmed, $result, true)); + self::assertFalse(in_array($disabled, $result, true)); + } + + public function testGetSubscribersBySubscribedListIdsReturnsMembersOfAnyGivenList(): void + { + $listA = (new SubscriberList())->setName('a'); + $listB = (new SubscriberList())->setName('b'); + $listC = (new SubscriberList())->setName('c'); + $this->entityManager->persist($listA); + $this->entityManager->persist($listB); + $this->entityManager->persist($listC); + + $inA = new Subscriber('in-a@example.com'); + $inB = new Subscriber('in-b@example.com'); + $inC = new Subscriber('in-c@example.com'); + $this->entityManager->persist($inA); + $this->entityManager->persist($inB); + $this->entityManager->persist($inC); + $this->subscribe($inA, $listA); + $this->subscribe($inB, $listB); + $this->subscribe($inC, $listC); + $this->entityManager->flush(); + + $result = $this->subscriberRepository->getSubscribersBySubscribedListIds([$listA->getId(), $listB->getId()]); + + self::assertTrue(in_array($inA, $result, true)); + self::assertTrue(in_array($inB, $result, true)); + self::assertFalse(in_array($inC, $result, true)); + } + + public function testGetSubscribersBySubscribedListIdsReturnsEmptyArrayForEmptyInput(): void + { + self::assertSame([], $this->subscriberRepository->getSubscribersBySubscribedListIds([])); + } } diff --git a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php index 683c215a..e9a9b477 100644 --- a/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php +++ b/tests/Unit/Domain/Messaging/MessageHandler/CampaignProcessorMessageHandlerTest.php @@ -10,15 +10,18 @@ use PhpList\Core\Domain\Configuration\Service\Provider\ConfigProvider; use PhpList\Core\Domain\Messaging\Message\CampaignProcessor\CampaignProcessorMessage; use PhpList\Core\Domain\Messaging\MessageHandler\CampaignProcessor\CampaignProcessorMessageHandler; +use PhpList\Core\Domain\Messaging\Model\Dto\DomainThrottleResult; use PhpList\Core\Domain\Messaging\Model\Dto\MessagePrecacheDto; use PhpList\Core\Domain\Messaging\Model\Message; use PhpList\Core\Domain\Messaging\Model\Message\MessageContent; use PhpList\Core\Domain\Messaging\Model\Message\MessageMetadata; -use PhpList\Core\Domain\Messaging\Model\Message\MessageStatus; +use PhpList\Core\Domain\Messaging\Model\Message\UserMessageStatus; +use PhpList\Core\Domain\Messaging\Model\UserMessage; use PhpList\Core\Domain\Messaging\Repository\MessageRepository; 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; @@ -52,6 +55,10 @@ class CampaignProcessorMessageHandlerTest extends TestCase private MessagePrecacheService|MockObject $precacheService; private CacheInterface|MockObject $cache; private MailerInterface|MockObject $symfonyMailer; + private UserMessageRepository|MockObject $userMessageRepository; + private MaxProcessTimeLimiter|MockObject $timeLimiter; + private RequeueHandler|MockObject $requeueHandler; + private DomainRateLimiter|MockObject $domainRateLimiter; protected function setUp(): void { @@ -72,7 +79,19 @@ protected function setUp(): void $timeLimiter->method('start'); $timeLimiter->method('shouldStop')->willReturn(false); - $this->handler = new CampaignProcessorMessageHandler( + $this->userMessageRepository = $userMessageRepository; + $this->timeLimiter = $timeLimiter; + $this->requeueHandler = $requeueHandler; + $this->domainRateLimiter = $this->createMock(DomainRateLimiter::class); + $this->domainRateLimiter->method('attemptSend') + ->willReturn(new DomainThrottleResult(allowed: true, domain: null)); + + $this->handler = $this->createHandler(); + } + + private function createHandler(bool $useListExclude = false): CampaignProcessorMessageHandler + { + return new CampaignProcessorMessageHandler( mailer: $this->symfonyMailer, rateLimitedCampaignMailer: $this->mailer, entityManager: $this->entityManager, @@ -80,9 +99,9 @@ protected function setUp(): void messagePreparator: $this->messagePreparator, logger: $this->logger, cache: $this->cache, - userMessageRepository: $userMessageRepository, - timeLimiter: $timeLimiter, - requeueHandler: $requeueHandler, + userMessageRepository: $this->userMessageRepository, + timeLimiter: $this->timeLimiter, + requeueHandler: $this->requeueHandler, translator: $this->translator, subscriberHistoryManager: $this->createMock(SubscriberHistoryManager::class), messageRepository: $this->messageRepository, @@ -92,7 +111,9 @@ protected function setUp(): void campaignEmailBuilder: $this->createMock(EmailBuilder::class), mailSizeChecker: $this->createMock(MailSizeChecker::class), configProvider: $this->createMock(ConfigProvider::class), + domainRateLimiter: $this->domainRateLimiter, bounceEmail: 'bounce@email.com', + useListExclude: $useListExclude, ); } @@ -101,8 +122,8 @@ public function testInvokeWhenCampaignNotFound(): void $message = new CampaignProcessorMessage(999); $this->messageRepository->expects($this->once()) - ->method('findByIdAndStatus') - ->with(999, MessageStatus::Submitted) + ->method('tryClaimForProcessing') + ->with(999) ->willReturn(null); $this->translator->method('trans')->willReturnCallback(fn(string $msg) => $msg); @@ -122,8 +143,8 @@ public function testInvokeWithNoSubscribers(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -148,6 +169,192 @@ public function testInvokeWithNoSubscribers(): void ($this->handler)($data); } + public function testInvokePassesExcludeListIdsFromMessageDataToSubscriberProviderWhenEnabled(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1, 66 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, [55, 66]) + ->willReturn([]); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeIgnoresExcludeListWhenUseListExcludeDisabled(): void + { + $handler = $this->createHandler(useListExclude: false); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1, 66 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, []) + ->willReturn([]); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeMarksExcludedSubscribersAsExcludedInUserMessage(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $excludedSubscriber = $this->createMock(Subscriber::class); + $excludedSubscriber->method('getEmail')->willReturn('excluded@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getExcludedSubscribers') + ->with([55]) + ->willReturn([$excludedSubscriber]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->with($data, $campaign, [55]) + ->willReturn([]); + + $this->userMessageRepository->expects($this->once()) + ->method('findByUserAndMessage') + ->with($excludedSubscriber, $campaign) + ->willReturn(null); + + $this->userMessageRepository->expects($this->once()) + ->method('save') + ->with($this->callback( + fn (UserMessage $userMessage): bool => $userMessage->getUser() === $excludedSubscriber + && $userMessage->getStatus() === UserMessageStatus::Excluded + )); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + + public function testInvokeDoesNotOverwriteExistingNonTodoUserMessageWhenMarkingExcluded(): void + { + $handler = $this->createHandler(useListExclude: true); + + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $messageDataLoaderProperty = (new ReflectionClass($handler))->getProperty('messageDataLoader'); + /** @var MessageDataLoader|MockObject $messageDataLoaderMock */ + $messageDataLoaderMock = $messageDataLoaderProperty->getValue($handler); + $messageDataLoaderMock->method('__invoke')->willReturn([ + 'excludelist' => [55 => 1], + ]); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $excludedSubscriber = $this->createMock(Subscriber::class); + $excludedSubscriber->method('getEmail')->willReturn('already-sent@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getExcludedSubscribers') + ->with([55]) + ->willReturn([$excludedSubscriber]); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->willReturn([]); + + $existingUserMessage = $this->createMock(UserMessage::class); + $existingUserMessage->method('getStatus')->willReturn(UserMessageStatus::Sent); + + $this->userMessageRepository->expects($this->once()) + ->method('findByUserAndMessage') + ->with($excludedSubscriber, $campaign) + ->willReturn($existingUserMessage); + + $this->userMessageRepository->expects($this->never()) + ->method('save'); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + public function testInvokeWithInvalidSubscriberEmail(): void { $campaign = $this->createCampaignMock(); @@ -156,8 +363,8 @@ public function testInvokeWithInvalidSubscriberEmail(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -203,8 +410,8 @@ public function testInvokeWithValidSubscriberEmail(): void $campaign->method('getId')->willReturn(1); $data = new CampaignProcessorMessage(1); - $this->messageRepository->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -271,8 +478,8 @@ public function testInvokeWithMailerException(): void $campaign->method('getId')->willReturn(123); $data = new CampaignProcessorMessage(123); - $this->messageRepository->method('findByIdAndStatus') - ->with(123, MessageStatus::Submitted) + $this->messageRepository->method('tryClaimForProcessing') + ->with(123) ->willReturn($campaign); $this->precacheService->expects($this->once()) @@ -348,8 +555,8 @@ public function testInvokeWithMultipleSubscribers(): void $data = new CampaignProcessorMessage(1); $this->messageRepository - ->method('findByIdAndStatus') - ->with(1, MessageStatus::Submitted) + ->method('tryClaimForProcessing') + ->with(1) ->willReturn($campaign); $this->precacheService @@ -464,6 +671,49 @@ function () use (&$buildCampaignEmailCalls): array { $this->assertCount(2, $buildCampaignEmailCalls); } + public function testInvokeSkipsDomainThrottledSubscriberWithoutCreatingUserMessage(): void + { + $campaign = $this->createCampaignMock(); + $metadata = $this->createMock(MessageMetadata::class); + $campaign->method('getMetadata')->willReturn($metadata); + $campaign->method('getId')->willReturn(1); + $data = new CampaignProcessorMessage(1); + + $this->messageRepository->method('tryClaimForProcessing') + ->with(1) + ->willReturn($campaign); + + $this->precacheService->expects($this->once()) + ->method('precacheMessage') + ->with($campaign, $this->anything()) + ->willReturn(true); + + $throttledSubscriber = $this->createMock(Subscriber::class); + $throttledSubscriber->method('getEmail')->willReturn('throttled@example.com'); + + $this->subscriberProvider->expects($this->once()) + ->method('getSubscribersForMessageOrLists') + ->willReturn([$throttledSubscriber]); + + $this->domainRateLimiter = $this->createMock(DomainRateLimiter::class); + $this->domainRateLimiter->method('attemptSend') + ->willReturn(new DomainThrottleResult(allowed: false, domain: 'example.com', blockedAttempts: 1)); + $handler = $this->createHandler(); + + $this->userMessageRepository->expects($this->never()) + ->method('save'); + + $this->requeueHandler->expects($this->once()) + ->method('handle') + ->with($campaign) + ->willReturn(true); + + $metadata->expects($this->atLeastOnce()) + ->method('setStatus'); + + $handler($data); + } + /** * Creates a mock for the Message class with content */ diff --git a/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php new file mode 100644 index 00000000..82af3ef5 --- /dev/null +++ b/tests/Unit/Domain/Messaging/Service/DomainRateLimiterTest.php @@ -0,0 +1,139 @@ +repository = $this->createMock(DomainThrottleStateRepository::class); + $this->logger = $this->createMock(LoggerInterface::class); + } + + private function createLimiter( + bool $enabled = true, + int $domainBatchSize = 1, + int $domainBatchPeriod = 120, + bool $autoThrottle = false, + ): DomainRateLimiter { + return new DomainRateLimiter( + repository: $this->repository, + logger: $this->logger, + enabled: $enabled, + domainBatchSize: $domainBatchSize, + domainBatchPeriod: $domainBatchPeriod, + autoThrottle: $autoThrottle, + ); + } + + public function testAllowsSendsWhenDisabled(): void + { + $this->repository->expects($this->never())->method('tryReserveSlot'); + + $limiter = $this->createLimiter(enabled: false); + + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); + } + + public function testAllowsSendsWhenBatchSizeOrPeriodIsNotPositive(): void + { + $this->repository->expects($this->never())->method('tryReserveSlot'); + + $limiter = $this->createLimiter(domainBatchSize: 0); + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); + + $limiter = $this->createLimiter(domainBatchPeriod: 0); + $this->assertTrue($limiter->attemptSend('a@example.com')->allowed); + } + + public function testAllowsSendsWhenAddressHasNoAtSign(): void + { + $this->repository->expects($this->never())->method('tryReserveSlot'); + + $limiter = $this->createLimiter(); + + $this->assertTrue($limiter->attemptSend('not-an-email')->allowed); + } + + public function testDelegatesReservationToRepositoryUsingLowercasedDomain(): void + { + $this->repository->expects($this->once()) + ->method('tryReserveSlot') + ->with('example.com', $this->isType('int'), 1) + ->willReturn(new DomainThrottleReservation(allowed: true)); + + $limiter = $this->createLimiter(); + $result = $limiter->attemptSend('first@Example.COM'); + + $this->assertTrue($result->allowed); + $this->assertSame('example.com', $result->domain); + } + + public function testReturnsBlockedResultWithAttemptsWhenQuotaReached(): void + { + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 3)); + + $this->logger->expects($this->once()) + ->method('info') + ->with('Send blocked by domain throttle', $this->anything()); + + $limiter = $this->createLimiter(); + $result = $limiter->attemptSend('third@example.com'); + + $this->assertFalse($result->allowed); + $this->assertSame(3, $result->blockedAttempts); + $this->assertFalse($result->backoffApplied); + } + + public function testDoesNotBackoffWhenAutoThrottleDisabled(): void + { + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 999)); + $this->repository->expects($this->never())->method('resetBlockedCount'); + + $limiter = $this->createLimiter(autoThrottle: false); + $result = $limiter->attemptSend('third@example.com'); + + $this->assertFalse($result->backoffApplied); + } + + public function testDoesNotBackoffBelowAttemptThreshold(): void + { + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 5)); + $this->repository->expects($this->never())->method('resetBlockedCount'); + + $limiter = $this->createLimiter(autoThrottle: true); + $result = $limiter->attemptSend('third@example.com'); + + $this->assertFalse($result->backoffApplied); + } + + public function testAppliesBackoffAndResetsBlockedCountOnceThresholdExceeded(): void + { + $this->repository->method('tryReserveSlot') + ->willReturn(new DomainThrottleReservation(allowed: false, blockedAttempts: 26)); + $this->repository->expects($this->once())->method('resetBlockedCount'); + + // Small batch period/size keeps the resulting sleep() short (~1s) so the test stays fast. + $limiter = $this->createLimiter(domainBatchSize: 1, domainBatchPeriod: 4, autoThrottle: true); + $result = $limiter->attemptSend('third@example.com'); + + $this->assertFalse($result->allowed); + $this->assertTrue($result->backoffApplied); + $this->assertGreaterThanOrEqual(1, $result->backoffSeconds); + } +} diff --git a/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php b/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php index 495f496e..ddef5190 100644 --- a/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php +++ b/tests/Unit/Domain/Messaging/Service/Handler/RequeueHandlerTest.php @@ -51,12 +51,40 @@ private function createMessage( return new Message($format, $schedule, $metadata, $content, $options, owner: null, template: null); } - public function testReturnsFalseWhenIntervalIsZeroOrNegative(): void + public function testFallsBackToOneMinuteIntervalWhenNoneConfigured(): void { + // requeueInterval controls how long to wait before resuming, not whether to resume: + // a campaign that stopped early must always be retried (mirrors phplist3's + // unconditional "don't mark sent while anything failed/was throttled" guard), so a + // missing/zero interval must not disable requeuing entirely. $handler = new RequeueHandler($this->logger, new Translator('en')); $message = $this->createMessage(0, null, null); - $this->output->expects($this->never())->method('writeln'); + $this->output->expects($this->once())->method('writeln'); + $this->logger->expects($this->once())->method('info'); + + $before = new DateTime(); + $result = $handler->handle($message, $this->output); + $after = new DateTime(); + + $this->assertTrue($result); + $this->assertSame(MessageStatus::Submitted, $message->getMetadata()->getStatus()); + + $embargo = $message->getSchedule()->getEmbargo(); + $this->assertInstanceOf(DateTime::class, $embargo); + + $minExpected = (clone $before)->add(new DateInterval('PT1M')); + $maxExpected = (clone $after)->add(new DateInterval('PT1M')); + $this->assertGreaterThanOrEqual($minExpected->getTimestamp(), $embargo->getTimestamp()); + $this->assertLessThanOrEqual($maxExpected->getTimestamp(), $embargo->getTimestamp()); + } + + public function testStillReturnsFalseWhenNoIntervalConfiguredButRequeueUntilAlreadyPassed(): void + { + $handler = new RequeueHandler($this->logger, new Translator('en')); + $past = (new DateTime())->sub(new DateInterval('PT5M')); + $message = $this->createMessage(0, $past, null); + $this->logger->expects($this->never())->method('info'); $result = $handler->handle($message, $this->output); diff --git a/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php b/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php index 0021ae87..6e7bcfe3 100644 --- a/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php +++ b/tests/Unit/Domain/Messaging/Service/Manager/MessageManagerTest.php @@ -5,6 +5,7 @@ namespace PhpList\Core\Tests\Unit\Domain\Messaging\Service\Manager; use DateTime; +use DateTimeImmutable; use InvalidArgumentException; use PhpList\Core\Domain\Identity\Model\Administrator; use PhpList\Core\Domain\Messaging\Model\ListMessage; @@ -241,4 +242,21 @@ public function testUpdateStatusSetsSubmittedWhenRequiredFieldsAndListArePresent $this->assertSame($message, $updated); $this->assertSame(Message\MessageStatus::Submitted, $message->getMetadata()->getStatus()); } + + public function testGetStuckCampaignsDelegatesToRepository(): void + { + $messageRepository = $this->createMock(MessageRepository::class); + $messageBuilder = $this->createMock(MessageBuilder::class); + $manager = new MessageManager($messageRepository, $messageBuilder); + + $staleBefore = new DateTimeImmutable('-30 minutes'); + $stuckMessage = $this->createMock(Message::class); + + $messageRepository->expects($this->once()) + ->method('getStuckInProcessing') + ->with($staleBefore) + ->willReturn([$stuckMessage]); + + $this->assertSame([$stuckMessage], $manager->getStuckCampaigns($staleBefore)); + } } diff --git a/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php b/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php index a68576a1..5ba8c29d 100644 --- a/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php +++ b/tests/Unit/Domain/Subscription/Service/Provider/SubscriberProviderTest.php @@ -39,7 +39,7 @@ public function testGetSubscribersForMessageWithNoListsReturnsEmptyArray(): void $this->subscriberRepository ->expects($this->never()) - ->method('getSubscribersBySubscribedListId'); + ->method('getSendableSubscribersBySubscribedListId'); $result = $this->subscriberProvider->getSubscribersForMessageOrLists( $this->createMock(CampaignProcessorMessageInterface::class), @@ -60,7 +60,7 @@ public function testGetSubscribersForMessageWithOneListButNoSubscribersReturnsEm $this->subscriberRepository ->expects($this->once()) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->with(456) ->willReturn([]); @@ -87,7 +87,7 @@ public function testGetSubscribersForMessageWithOneListAndSubscribersReturnsSubs $this->subscriberRepository ->expects($this->once()) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->with(456) ->willReturn([$subscriber1, $subscriber2]); @@ -118,7 +118,7 @@ public function testGetSubscribersForMessageWithMultipleListsReturnsUniqueSubscr $this->subscriberRepository ->expects($this->exactly(2)) - ->method('getSubscribersBySubscribedListId') + ->method('getSendableSubscribersBySubscribedListId') ->willReturnMap([ [456, [$subscriber1, $subscriber2]], [789, [$subscriber2, $subscriber3]], @@ -134,4 +134,39 @@ public function testGetSubscribersForMessageWithMultipleListsReturnsUniqueSubscr $this->assertContains($subscriber2, $result); $this->assertContains($subscriber3, $result); } + + public function testGetSubscribersForMessageExcludesSubscribersOnExcludeLists(): void + { + $message = $this->createMock(Message::class); + $message->method('getId')->willReturn(123); + + $this->subscriberListRepository + ->method('getListIdsByMessage') + ->willReturn([456]); + + $subscriber1 = $this->createMock(Subscriber::class); + $subscriber1->method('getEmail')->willReturn('keep@example.am'); + $subscriber2 = $this->createMock(Subscriber::class); + $subscriber2->method('getEmail')->willReturn('exclude@example.am'); + + $this->subscriberRepository + ->method('getSendableSubscribersBySubscribedListId') + ->with(456) + ->willReturn([$subscriber1, $subscriber2]); + + $this->subscriberRepository + ->expects($this->once()) + ->method('getSubscribersBySubscribedListIds') + ->with([789]) + ->willReturn([$subscriber2]); + + $result = $this->subscriberProvider->getSubscribersForMessageOrLists( + $this->createMock(CampaignProcessorMessageInterface::class), + $message, + [789], + ); + + $this->assertCount(1, $result); + $this->assertSame($subscriber1, $result[0]); + } }