NewsletterReplayMetadata.php
2 months ago
ScheduledTaskSubscribersListingRepository.php
2 years ago
ScheduledTaskSubscribersRepository.php
1 month ago
ScheduledTasksRepository.php
1 month ago
SendingQueuesRepository.php
1 week ago
TimeZoneCampaignScheduler.php
1 week ago
index.php
3 years ago
ScheduledTaskSubscribersRepository.php
343 lines
| 1 | <?php declare(strict_types = 1); |
| 2 | |
| 3 | namespace MailPoet\Newsletter\Sending; |
| 4 | |
| 5 | if (!defined('ABSPATH')) exit; |
| 6 | |
| 7 | |
| 8 | use MailPoet\Doctrine\Repository; |
| 9 | use MailPoet\Entities\ScheduledTaskEntity; |
| 10 | use MailPoet\Entities\ScheduledTaskSubscriberEntity; |
| 11 | use MailPoet\Entities\SubscriberEntity; |
| 12 | use MailPoet\InvalidStateException; |
| 13 | use MailPoetVendor\Carbon\Carbon; |
| 14 | use MailPoetVendor\Doctrine\DBAL\ArrayParameterType; |
| 15 | use MailPoetVendor\Doctrine\DBAL\ParameterType; |
| 16 | use MailPoetVendor\Doctrine\ORM\QueryBuilder; |
| 17 | |
| 18 | /** |
| 19 | * @extends Repository<ScheduledTaskSubscriberEntity> |
| 20 | */ |
| 21 | class ScheduledTaskSubscribersRepository extends Repository { |
| 22 | protected function getEntityClassName() { |
| 23 | return ScheduledTaskSubscriberEntity::class; |
| 24 | } |
| 25 | |
| 26 | public function isSubscriberProcessed(ScheduledTaskEntity $task, SubscriberEntity $subscriber): bool { |
| 27 | $scheduledTaskSubscriber = $this |
| 28 | ->doctrineRepository |
| 29 | ->createQueryBuilder('sts') |
| 30 | ->andWhere('sts.processed = 1') |
| 31 | ->andWhere('sts.task = :task') |
| 32 | ->andWhere('sts.subscriber = :subscriber') |
| 33 | ->setParameter('subscriber', $subscriber) |
| 34 | ->setParameter('task', $task) |
| 35 | ->getQuery() |
| 36 | ->getOneOrNullResult(); |
| 37 | return !empty($scheduledTaskSubscriber); |
| 38 | } |
| 39 | |
| 40 | public function createOrUpdate(array $data): ?ScheduledTaskSubscriberEntity { |
| 41 | if (!isset($data['task_id'], $data['subscriber_id'])) { |
| 42 | return null; |
| 43 | } |
| 44 | |
| 45 | $taskSubscriber = $this->findOneBy(['task' => $data['task_id'], 'subscriber' => $data['subscriber_id']]); |
| 46 | if (!$taskSubscriber) { |
| 47 | $task = $this->entityManager->getReference(ScheduledTaskEntity::class, (int)$data['task_id']); |
| 48 | $subscriber = $this->entityManager->getReference(SubscriberEntity::class, (int)$data['subscriber_id']); |
| 49 | if (!$task || !$subscriber) throw new InvalidStateException('Task or subscriber not found'); |
| 50 | |
| 51 | $taskSubscriber = new ScheduledTaskSubscriberEntity($task, $subscriber); |
| 52 | $this->persist($taskSubscriber); |
| 53 | } |
| 54 | |
| 55 | $processed = $data['processed'] ?? ScheduledTaskSubscriberEntity::STATUS_UNPROCESSED; |
| 56 | $failed = $data['failed'] ?? ScheduledTaskSubscriberEntity::FAIL_STATUS_OK; |
| 57 | |
| 58 | $taskSubscriber->setProcessed($processed); |
| 59 | $taskSubscriber->setFailed($failed); |
| 60 | $this->flush(); |
| 61 | return $taskSubscriber; |
| 62 | } |
| 63 | |
| 64 | public function countSubscriberIdsBatchForTask(int $taskId, int $lastProcessedSubscriberId): int { |
| 65 | $queryBuilder = $this->getBaseSubscribersIdsBatchForTaskQuery($taskId, $lastProcessedSubscriberId); |
| 66 | $countSubscribers = $queryBuilder |
| 67 | ->select('count(sts.subscriber)') |
| 68 | ->getQuery() |
| 69 | ->getSingleScalarResult(); |
| 70 | |
| 71 | return intval($countSubscribers); |
| 72 | } |
| 73 | |
| 74 | public function getSubscriberIdsBatchForTask(int $taskId, int $lastProcessedSubscriberId, int $limit): array { |
| 75 | $queryBuilder = $this->getBaseSubscribersIdsBatchForTaskQuery($taskId, $lastProcessedSubscriberId); |
| 76 | $subscribersIds = $queryBuilder |
| 77 | ->select('IDENTITY(sts.subscriber) AS subscriber_id') |
| 78 | ->orderBy('sts.subscriber', 'asc') |
| 79 | ->setMaxResults($limit) |
| 80 | ->getQuery() |
| 81 | ->getSingleColumnResult(); |
| 82 | |
| 83 | return $subscribersIds; |
| 84 | } |
| 85 | |
| 86 | /** |
| 87 | * @param int[] $subscriberIds |
| 88 | */ |
| 89 | public function updateProcessedSubscribers(ScheduledTaskEntity $task, array $subscriberIds): void { |
| 90 | if ($subscriberIds) { |
| 91 | $this->entityManager->createQueryBuilder() |
| 92 | ->update(ScheduledTaskSubscriberEntity::class, 'sts') |
| 93 | ->set('sts.processed', ScheduledTaskSubscriberEntity::STATUS_PROCESSED) |
| 94 | ->where('sts.subscriber IN (:subscriberIds)') |
| 95 | ->andWhere('sts.task = :task') |
| 96 | ->setParameter('subscriberIds', $subscriberIds, ArrayParameterType::INTEGER) |
| 97 | ->setParameter('task', $task) |
| 98 | ->getQuery() |
| 99 | ->execute(); |
| 100 | |
| 101 | // update was done via DQL, make sure the entities are also refreshed in the entity manager |
| 102 | $this->refreshAll(function (ScheduledTaskSubscriberEntity $entity) use ($task, $subscriberIds) { |
| 103 | return $entity->getTask() === $task && in_array($entity->getSubscriberId(), $subscriberIds, true); |
| 104 | }); |
| 105 | } |
| 106 | |
| 107 | $this->checkCompleted($task); |
| 108 | } |
| 109 | |
| 110 | /** @param int[] $subscriberIds */ |
| 111 | public function addSubscribersByIds(ScheduledTaskEntity $task, array $subscriberIds): int { |
| 112 | $subscriberIds = array_values(array_unique(array_filter(array_map('intval', $subscriberIds)))); |
| 113 | if ($subscriberIds === []) { |
| 114 | return 0; |
| 115 | } |
| 116 | |
| 117 | $scheduledTaskSubscribersTable = $this->entityManager->getClassMetadata(ScheduledTaskSubscriberEntity::class)->getTableName(); |
| 118 | $subscribersTable = $this->entityManager->getClassMetadata(SubscriberEntity::class)->getTableName(); |
| 119 | |
| 120 | $result = $this->entityManager->getConnection()->executeQuery( |
| 121 | "INSERT IGNORE INTO $scheduledTaskSubscribersTable |
| 122 | (task_id, subscriber_id, processed) |
| 123 | SELECT DISTINCT ? as task_id, subscribers.`id` as subscriber_id, ? as processed |
| 124 | FROM $subscribersTable subscribers |
| 125 | WHERE subscribers.`deleted_at` IS NULL |
| 126 | AND subscribers.`status` = ? |
| 127 | AND subscribers.`id` IN (?)", |
| 128 | [ |
| 129 | $task->getId(), |
| 130 | ScheduledTaskSubscriberEntity::STATUS_UNPROCESSED, |
| 131 | SubscriberEntity::STATUS_SUBSCRIBED, |
| 132 | $subscriberIds, |
| 133 | ], |
| 134 | [ |
| 135 | ParameterType::INTEGER, |
| 136 | ParameterType::INTEGER, |
| 137 | ParameterType::STRING, |
| 138 | ArrayParameterType::INTEGER, |
| 139 | ] |
| 140 | ); |
| 141 | |
| 142 | return (int)$result->rowCount(); |
| 143 | } |
| 144 | |
| 145 | /** @param int[] $ids */ |
| 146 | public function deleteByTaskIds(array $ids): void { |
| 147 | $this->entityManager->createQueryBuilder() |
| 148 | ->delete(ScheduledTaskSubscriberEntity::class, 'sts') |
| 149 | ->where('sts.task IN (:taskIds)') |
| 150 | ->setParameter('taskIds', $ids) |
| 151 | ->getQuery() |
| 152 | ->execute(); |
| 153 | |
| 154 | // delete was done via DQL, make sure the entities are also detached from the entity manager |
| 155 | $this->detachAll(function (ScheduledTaskSubscriberEntity $entity) use ($ids) { |
| 156 | $task = $entity->getTask(); |
| 157 | return $task && in_array($task->getId(), $ids, true); |
| 158 | }); |
| 159 | } |
| 160 | |
| 161 | public function deleteByScheduledTask(ScheduledTaskEntity $scheduledTask): void { |
| 162 | $this->entityManager->createQueryBuilder() |
| 163 | ->delete(ScheduledTaskSubscriberEntity::class, 'sts') |
| 164 | ->where('sts.task = :task') |
| 165 | ->setParameter('task', $scheduledTask) |
| 166 | ->getQuery() |
| 167 | ->execute(); |
| 168 | |
| 169 | // delete was done via DQL, make sure the entities are also detached from the entity manager |
| 170 | $this->detachAll(function (ScheduledTaskSubscriberEntity $entity) use ($scheduledTask) { |
| 171 | return $entity->getTask() === $scheduledTask; |
| 172 | }); |
| 173 | } |
| 174 | |
| 175 | public function deleteByScheduledTaskAndSubscriberIds(ScheduledTaskEntity $scheduledTask, array $subscriberIds): void { |
| 176 | $this->entityManager->createQueryBuilder() |
| 177 | ->delete(ScheduledTaskSubscriberEntity::class, 'sts') |
| 178 | ->where('sts.task = :task') |
| 179 | ->andWhere('sts.subscriber IN (:subscriberIds)') |
| 180 | ->setParameter('task', $scheduledTask) |
| 181 | ->setParameter('subscriberIds', $subscriberIds, ArrayParameterType::INTEGER) |
| 182 | ->getQuery() |
| 183 | ->execute(); |
| 184 | |
| 185 | // delete was done via DQL, make sure the entities are also detached from the entity manager |
| 186 | $this->detachAll(function (ScheduledTaskSubscriberEntity $entity) use ($scheduledTask, $subscriberIds) { |
| 187 | return $entity->getTask() === $scheduledTask && in_array($entity->getSubscriberId(), $subscriberIds, true); |
| 188 | }); |
| 189 | |
| 190 | $this->checkCompleted($scheduledTask); |
| 191 | } |
| 192 | |
| 193 | public function setSubscribers(ScheduledTaskEntity $task, array $subscriberIds): void { |
| 194 | $this->deleteByScheduledTask($task); |
| 195 | |
| 196 | foreach ($subscriberIds as $subscriberId) { |
| 197 | $this->createOrUpdate([ |
| 198 | 'task_id' => $task->getId(), |
| 199 | 'subscriber_id' => $subscriberId, |
| 200 | ]); |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | public function saveError(ScheduledTaskEntity $scheduledTask, int $subscriberId, string $errorMessage): void { |
| 205 | $scheduledTaskSubscriber = $this->findOneBy(['task' => $scheduledTask, 'subscriber' => $subscriberId]); |
| 206 | |
| 207 | if ($scheduledTaskSubscriber instanceof ScheduledTaskSubscriberEntity) { |
| 208 | $scheduledTaskSubscriber->setFailed(ScheduledTaskSubscriberEntity::FAIL_STATUS_FAILED); |
| 209 | $scheduledTaskSubscriber->setProcessed(ScheduledTaskSubscriberEntity::STATUS_PROCESSED); |
| 210 | $scheduledTaskSubscriber->setError($errorMessage); |
| 211 | $this->persist($scheduledTaskSubscriber); |
| 212 | $this->flush(); |
| 213 | |
| 214 | $this->checkCompleted($scheduledTask); |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | public function countProcessed(ScheduledTaskEntity $scheduledTaskEntity): int { |
| 219 | return $this->countBy(['task' => $scheduledTaskEntity, 'processed' => ScheduledTaskSubscriberEntity::STATUS_PROCESSED]); |
| 220 | } |
| 221 | |
| 222 | public function countUnprocessed(ScheduledTaskEntity $scheduledTaskEntity): int { |
| 223 | return $this->countBy(['task' => $scheduledTaskEntity, 'processed' => ScheduledTaskSubscriberEntity::STATUS_UNPROCESSED]); |
| 224 | } |
| 225 | |
| 226 | public function purgeOldTaskSubscribers(int $daysToKeep, int $taskBatchSize, int $rowLimit): int { |
| 227 | $stTable = $this->entityManager->getClassMetadata(ScheduledTaskEntity::class)->getTableName(); |
| 228 | $stsTable = $this->entityManager->getClassMetadata(ScheduledTaskSubscriberEntity::class)->getTableName(); |
| 229 | $cutoff = Carbon::now()->subDays($daysToKeep)->toDateTimeString(); |
| 230 | |
| 231 | $taskIds = $this->entityManager->getConnection()->executeQuery( |
| 232 | "SELECT DISTINCT st.`id` |
| 233 | FROM `{$stTable}` st |
| 234 | INNER JOIN `{$stsTable}` sts ON sts.`task_id` = st.`id` |
| 235 | WHERE st.`type` = :type |
| 236 | AND st.`status` = :status |
| 237 | AND st.`processed_at` < :cutoff |
| 238 | AND st.`deleted_at` IS NULL |
| 239 | LIMIT :taskBatchSize", |
| 240 | [ |
| 241 | 'type' => 'sending', |
| 242 | 'status' => ScheduledTaskEntity::STATUS_COMPLETED, |
| 243 | 'cutoff' => $cutoff, |
| 244 | 'taskBatchSize' => $taskBatchSize, |
| 245 | ], |
| 246 | [ |
| 247 | 'type' => ParameterType::STRING, |
| 248 | 'status' => ParameterType::STRING, |
| 249 | 'cutoff' => ParameterType::STRING, |
| 250 | 'taskBatchSize' => ParameterType::INTEGER, |
| 251 | ] |
| 252 | )->fetchFirstColumn(); |
| 253 | |
| 254 | if (!$taskIds) { |
| 255 | return 0; |
| 256 | } |
| 257 | |
| 258 | /** @var int[] $taskIds */ |
| 259 | $taskIdsList = implode(',', array_map('intval', $taskIds)); |
| 260 | |
| 261 | $deleted = $this->entityManager->getConnection()->executeStatement( |
| 262 | "DELETE FROM `{$stsTable}` |
| 263 | WHERE `task_id` IN ({$taskIdsList}) |
| 264 | LIMIT :rowLimit", |
| 265 | [ |
| 266 | 'rowLimit' => $rowLimit, |
| 267 | ], |
| 268 | [ |
| 269 | 'rowLimit' => ParameterType::INTEGER, |
| 270 | ] |
| 271 | ); |
| 272 | |
| 273 | return (int)$deleted; |
| 274 | } |
| 275 | |
| 276 | public function purgeCompletedBounceTaskSubscribers(int $taskBatchSize, int $rowLimit): int { |
| 277 | $stTable = $this->entityManager->getClassMetadata(ScheduledTaskEntity::class)->getTableName(); |
| 278 | $stsTable = $this->entityManager->getClassMetadata(ScheduledTaskSubscriberEntity::class)->getTableName(); |
| 279 | |
| 280 | $taskIds = $this->entityManager->getConnection()->executeQuery( |
| 281 | "SELECT DISTINCT st.`id` |
| 282 | FROM `{$stTable}` st |
| 283 | INNER JOIN `{$stsTable}` sts ON sts.`task_id` = st.`id` |
| 284 | WHERE st.`type` = :type |
| 285 | AND st.`status` = :status |
| 286 | AND st.`deleted_at` IS NULL |
| 287 | LIMIT :taskBatchSize", |
| 288 | [ |
| 289 | 'type' => 'bounce', |
| 290 | 'status' => ScheduledTaskEntity::STATUS_COMPLETED, |
| 291 | 'taskBatchSize' => $taskBatchSize, |
| 292 | ], |
| 293 | [ |
| 294 | 'type' => ParameterType::STRING, |
| 295 | 'status' => ParameterType::STRING, |
| 296 | 'taskBatchSize' => ParameterType::INTEGER, |
| 297 | ] |
| 298 | )->fetchFirstColumn(); |
| 299 | |
| 300 | if (!$taskIds) { |
| 301 | return 0; |
| 302 | } |
| 303 | |
| 304 | /** @var int[] $taskIds */ |
| 305 | $taskIdsList = implode(',', array_map('intval', $taskIds)); |
| 306 | |
| 307 | $deleted = $this->entityManager->getConnection()->executeStatement( |
| 308 | "DELETE FROM `{$stsTable}` |
| 309 | WHERE `task_id` IN ({$taskIdsList}) |
| 310 | LIMIT :rowLimit", |
| 311 | [ |
| 312 | 'rowLimit' => $rowLimit, |
| 313 | ], |
| 314 | [ |
| 315 | 'rowLimit' => ParameterType::INTEGER, |
| 316 | ] |
| 317 | ); |
| 318 | |
| 319 | return (int)$deleted; |
| 320 | } |
| 321 | |
| 322 | private function checkCompleted(ScheduledTaskEntity $task): void { |
| 323 | $count = $this->countUnprocessed($task); |
| 324 | if ($count === 0) { |
| 325 | $task->setStatus(ScheduledTaskEntity::STATUS_COMPLETED); |
| 326 | $task->setProcessedAt(Carbon::now()->millisecond(0)); |
| 327 | $this->entityManager->flush(); |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | private function getBaseSubscribersIdsBatchForTaskQuery(int $taskId, int $lastProcessedSubscriberId): QueryBuilder { |
| 332 | return $this->entityManager |
| 333 | ->createQueryBuilder() |
| 334 | ->from(ScheduledTaskSubscriberEntity::class, 'sts') |
| 335 | ->andWhere('sts.task = :taskId') |
| 336 | ->andWhere('sts.subscriber > :lastProcessedSubscriberId') |
| 337 | ->andWhere('sts.processed = :status') |
| 338 | ->setParameter('taskId', $taskId) |
| 339 | ->setParameter('lastProcessedSubscriberId', $lastProcessedSubscriberId) |
| 340 | ->setParameter('status', ScheduledTaskSubscriberEntity::STATUS_UNPROCESSED); |
| 341 | } |
| 342 | } |
| 343 |