Cavemanon/piratepoet
8
Fork 0
Code Pull Requests Actions Releases 16 Activity

Use Doctrine in InactiveSubscribersController

Browse Source 
[MAILPOET-3774]
This commit is contained in:
Jan Lysý
2021-09-22 15:53:54 +02:00
committed by Veljko V
parent 93e038371d
commit ddb025c0ec
1 changed files with 104 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

180
lib/Subscribers/InactiveSubscribersController.php
View File

@@ -1,28 +1,33 @@
<?php <?php declare(strict_types = 1);
namespace MailPoet\Subscribers; namespace MailPoet\Subscribers;
use MailPoet\Config\MP2Migrator; use MailPoet\Config\MP2Migrator;
use MailPoet\Models\ScheduledTask; use MailPoet\Entities\ScheduledTaskEntity;
use MailPoet\Models\ScheduledTaskSubscriber; use MailPoet\Entities\ScheduledTaskSubscriberEntity;
use MailPoet\Models\SendingQueue; use MailPoet\Entities\SendingQueueEntity;
use MailPoet\Models\StatisticsOpens; use MailPoet\Entities\SubscriberEntity;
use MailPoet\Models\Subscriber;
use MailPoet\Settings\SettingsRepository; use MailPoet\Settings\SettingsRepository;
use MailPoetVendor\Carbon\Carbon; use MailPoetVendor\Carbon\Carbon;
use MailPoetVendor\Idiorm\ORM; use MailPoetVendor\Doctrine\DBAL\Connection;
use MailPoetVendor\Doctrine\ORM\EntityManager;
class InactiveSubscribersController { class InactiveSubscribersController {
private $inactivesTaskIdsTableCreated = false; private $inactiveTaskIdsTableCreated = false;
/** @var SettingsRepository */ /** @var SettingsRepository */
private $settingsRepository; private $settingsRepository;
/** @var EntityManager */
private $entityManager;
public function __construct( public function __construct(
EntityManager $entityManager,
SettingsRepository $settingsRepository SettingsRepository $settingsRepository
) { ) {
$this->settingsRepository = $settingsRepository; $this->settingsRepository = $settingsRepository;
$this->entityManager = $entityManager;
} }
public function markInactiveSubscribers(int $daysToInactive, int $batchSize, ?int $startId = null) { public function markInactiveSubscribers(int $daysToInactive, int $batchSize, ?int $startId = null) {
@@ -35,15 +40,15 @@ class InactiveSubscribersController {
return $this->activateSubscribers($thresholdDate, $batchSize); return $this->activateSubscribers($thresholdDate, $batchSize);
} }
/** public function reactivateInactiveSubscribers(): void {
* @return void $subscribersTable = $this->entityManager->getClassMetadata(SubscriberEntity::class)->getTableName();
*/ $reactivateAllInactiveQuery = "
public function reactivateInactiveSubscribers() { UPDATE {$subscribersTable} SET status = :statusSubscribed WHERE status = :statusInactive
$reactivateAllInactiveQuery = sprintf( ";
"UPDATE %s SET status = '%s' WHERE status = '%s';", $this->entityManager->getConnection()->executeQuery($reactivateAllInactiveQuery, [
Subscriber::$_table, Subscriber::STATUS_SUBSCRIBED, Subscriber::STATUS_INACTIVE 'statusSubscribed' => SubscriberEntity::STATUS_SUBSCRIBED,
); 'statusInactive' => SubscriberEntity::STATUS_INACTIVE,
ORM::rawExecute($reactivateAllInactiveQuery); ]);
} }
private function getThresholdDate(int $daysToInactive): Carbon { private function getThresholdDate(int $daysToInactive): Carbon {
@@ -55,11 +60,11 @@ class InactiveSubscribersController {
* @return int|bool * @return int|bool
*/ */
private function deactivateSubscribers(Carbon $thresholdDate, int $batchSize, ?int $startId = null) { private function deactivateSubscribers(Carbon $thresholdDate, int $batchSize, ?int $startId = null) {
$subscribersTable = Subscriber::$_table; $subscribersTable = $this->entityManager->getClassMetadata(SubscriberEntity::class)->getTableName();
$scheduledTasksTable = ScheduledTask::$_table; $scheduledTasksTable = $this->entityManager->getClassMetadata(ScheduledTaskEntity::class)->getTableName();
$scheduledTaskSubcribresTable = ScheduledTaskSubscriber::$_table; $scheduledTaskSubscribersTable = $this->entityManager->getClassMetadata(ScheduledTaskSubscriberEntity::class)->getTableName();
$statisticsOpensTable = StatisticsOpens::$_table; $sendingQueuesTable = $this->entityManager->getClassMetadata(SendingQueueEntity::class)->getTableName();
$sendingQueuesTable = SendingQueue::$_table; $connection = $this->entityManager->getConnection();
$thresholdDateIso = $thresholdDate->toDateTimeString(); $thresholdDateIso = $thresholdDate->toDateTimeString();
$dayAgo = new Carbon(); $dayAgo = new Carbon();
@@ -74,46 +79,59 @@ class InactiveSubscribersController {
// We take into account only emails which have at least one opening tracked // We take into account only emails which have at least one opening tracked
// to ensure that tracking was enabled for the particular email // to ensure that tracking was enabled for the particular email
if (!$this->inactivesTaskIdsTableCreated) { $inactiveTaskIdsTable = 'inactive_task_ids';
$inactivesTaskIdsTable = sprintf(" if (!$this->inactiveTaskIdsTableCreated) {
CREATE TEMPORARY TABLE IF NOT EXISTS inactives_task_ids $inactiveTaskIdsTableSql = "
(INDEX task_id_ids (id)) CREATE TEMPORARY TABLE IF NOT EXISTS {$inactiveTaskIdsTable}
SELECT DISTINCT task_id as id FROM $sendingQueuesTable as sq (INDEX task_id_ids (id))
JOIN $scheduledTasksTable as st ON sq.task_id = st.id SELECT DISTINCT task_id as id FROM {$sendingQueuesTable} as sq
WHERE st.processed_at > '%s' JOIN {$scheduledTasksTable} as st ON sq.task_id = st.id
AND st.processed_at < '%s'", WHERE st.processed_at > :thresholdDate
$thresholdDateIso, $dayAgoIso AND st.processed_at < :dayAgo
); ";
ORM::rawExecute($inactivesTaskIdsTable); $connection->executeQuery($inactiveTaskIdsTableSql, [
$this->inactivesTaskIdsTableCreated = true; 'thresholdDate' => $thresholdDateIso,
'dayAgo' => $dayAgoIso,
]);
$this->inactiveTaskIdsTableCreated = true;
} }
// Select subscribers who received a recent tracked email but didn't open it // Select subscribers who received a recent tracked email but didn't open it
$startId = (int)$startId; $startId = (int)$startId;
$endId = $startId + $batchSize; $endId = $startId + $batchSize;
$inactiveSubscriberIdsTmpTable = 'inactive_subscriber_ids'; $inactiveSubscriberIdsTmpTable = 'inactive_subscriber_ids';
ORM::rawExecute(" $connection->executeQuery("
CREATE TEMPORARY TABLE IF NOT EXISTS $inactiveSubscriberIdsTmpTable CREATE TEMPORARY TABLE IF NOT EXISTS {$inactiveSubscriberIdsTmpTable}
(UNIQUE subscriber_id (id)) (UNIQUE subscriber_id (id))
SELECT DISTINCT s.id FROM $subscribersTable as s SELECT DISTINCT s.id FROM {$subscribersTable} as s
JOIN $scheduledTaskSubcribresTable as sts USE INDEX (subscriber_id) ON s.id = sts.subscriber_id JOIN {$scheduledTaskSubscribersTable} as sts USE INDEX (subscriber_id) ON s.id = sts.subscriber_id
JOIN inactives_task_ids task_ids ON task_ids.id = sts.task_id JOIN {$inactiveTaskIdsTable} task_ids ON task_ids.id = sts.task_id
WHERE s.last_subscribed_at < ? AND s.status = ? AND s.id >= ? AND s.id < ?", WHERE s.last_subscribed_at < :thresholdDate
[$thresholdDateIso, Subscriber::STATUS_SUBSCRIBED, $startId, $endId] AND s.status = :status
); AND s.id >= :startId
AND s.id < :endId
",
[
'thresholdDate' => $thresholdDateIso,
'status' => SubscriberEntity::STATUS_SUBSCRIBED,
'startId' => $startId,
'endId' => $endId,
]);
$idsToDeactivate = ORM::forTable($inactiveSubscriberIdsTmpTable)->rawQuery(" $result = $connection->executeQuery("
SELECT isi.id FROM $inactiveSubscriberIdsTmpTable isi SELECT isi.id FROM {$inactiveSubscriberIdsTmpTable} isi
LEFT OUTER JOIN $subscribersTable as s ON isi.id = s.id AND GREATEST( LEFT OUTER JOIN {$subscribersTable} as s ON isi.id = s.id AND GREATEST(
COALESCE(s.last_engagement_at, 0), COALESCE(s.last_engagement_at, 0),
COALESCE(s.last_subscribed_at, 0), COALESCE(s.last_subscribed_at, 0),
COALESCE(s.created_at, 0) COALESCE(s.created_at, 0)
) > ? ) > :thresholdDate
WHERE s.id IS NULL", WHERE s.id IS NULL
[$thresholdDateIso] ", [
)->findArray(); 'thresholdDate' => $thresholdDateIso,
]);
$idsToDeactivate = $result->fetchAllAssociative();
ORM::rawExecute("DROP TABLE $inactiveSubscriberIdsTmpTable"); $connection->executeQuery("DROP TABLE {$inactiveSubscriberIdsTmpTable}");
$idsToDeactivate = array_map( $idsToDeactivate = array_map(
function ($id) { function ($id) {
@@ -124,39 +142,50 @@ class InactiveSubscribersController {
if (!count($idsToDeactivate)) { if (!count($idsToDeactivate)) {
return 0; return 0;
} }
ORM::rawExecute(sprintf( $connection->executeQuery("UPDATE {$subscribersTable} SET status = :statusInactive WHERE id IN (:idsToDeactivate)", [
"UPDATE %s SET status='" . Subscriber::STATUS_INACTIVE . "' WHERE id IN (%s);", 'statusInactive' => SubscriberEntity::STATUS_INACTIVE,
$subscribersTable, 'idsToDeactivate' => $idsToDeactivate,
implode(',', $idsToDeactivate) ], ['idsToDeactivate' => Connection::PARAM_INT_ARRAY]);
));
return count($idsToDeactivate); return count($idsToDeactivate);
} }
private function activateSubscribers(Carbon $thresholdDate, int $batchSize): int { private function activateSubscribers(Carbon $thresholdDate, int $batchSize): int {
$subscribersTable = Subscriber::$_table; $subscribersTable = $this->entityManager->getClassMetadata(SubscriberEntity::class)->getTableName();
$statsOpensTable = StatisticsOpens::$_table; $connection = $this->entityManager->getConnection();
$mp2MigrationDate = $this->getMP2MigrationDate(); $mp2MigrationDate = $this->getMP2MigrationDate();
if ($mp2MigrationDate && $mp2MigrationDate > $thresholdDate) { if ($mp2MigrationDate && $mp2MigrationDate > $thresholdDate) {
// If MP2 migration occurred during detection interval re-activate all subscribers created before migration // If MP2 migration occurred during detection interval re-activate all subscribers created before migration
$idsToActivate = ORM::forTable($subscribersTable)->select("$subscribersTable.id") $idsToActivate = $connection->executeQuery("
->whereLt("$subscribersTable.created_at", $mp2MigrationDate) SELECT id
->where("$subscribersTable.status", Subscriber::STATUS_INACTIVE) FROM {$subscribersTable}
->limit($batchSize) WHERE created_at < :migrationDate
->findArray(); AND status = :statusInactive
LIMIT :batchSize
", [
'migrationDate' => $mp2MigrationDate,
'statusInactive' => SubscriberEntity::STATUS_INACTIVE,
'batchSize' => $batchSize,
], ['batchSize' => \PDO::PARAM_INT])->fetchAllAssociative();
} else { } else {
$idsToActivate = ORM::forTable($subscribersTable)->select("$subscribersTable.id") $idsToActivate = $connection->executeQuery("
->leftOuterJoin($subscribersTable, "$subscribersTable.id = s2.id AND GREATEST( SELECT s.id
FROM {$subscribersTable} s
LEFT OUTER JOIN {$subscribersTable} s2 ON s.id = s2.id AND GREATEST(
COALESCE(s2.last_engagement_at, 0), COALESCE(s2.last_engagement_at, 0),
COALESCE(s2.last_subscribed_at, 0), COALESCE(s2.last_subscribed_at, 0),
COALESCE(s2.created_at, 0) COALESCE(s2.created_at, 0)
) > '$thresholdDate'", 's2') ) > :thresholdDate
->whereLt("$subscribersTable.last_subscribed_at", $thresholdDate) WHERE s.last_subscribed_at < :thresholdDate
->where("$subscribersTable.status", Subscriber::STATUS_INACTIVE) AND s.status = :statusInactive
->whereRaw("s2.id IS NOT NULL") AND s2.id IS NOT NULL
->limit($batchSize) GROUP BY s.id
->groupByExpr("$subscribersTable.id") LIMIT :batchSize
->findArray(); ", [
'thresholdDate' => $thresholdDate,
'statusInactive' => SubscriberEntity::STATUS_INACTIVE,
'batchSize' => $batchSize,
], ['batchSize' => \PDO::PARAM_INT])->fetchAllAssociative();
} }
$idsToActivate = array_map( $idsToActivate = array_map(
@@ -167,11 +196,10 @@ class InactiveSubscribersController {
if (!count($idsToActivate)) { if (!count($idsToActivate)) {
return 0; return 0;
} }
ORM::rawExecute(sprintf( $connection->executeQuery("UPDATE {$subscribersTable} SET status = :statusSubscribed WHERE id IN (:idsToActivate)", [
"UPDATE %s SET status='" . Subscriber::STATUS_SUBSCRIBED . "' WHERE id IN (%s);", 'statusSubscribed' => SubscriberEntity::STATUS_SUBSCRIBED,
$subscribersTable, 'idsToActivate' => $idsToActivate,
implode(',', $idsToActivate) ], ['idsToActivate' => Connection::PARAM_INT_ARRAY]);
));
return count($idsToActivate); return count($idsToActivate);
} }