Merged PR 55606: Sync all contact subscription state changes via a si…

…ngle queue

## What's being changed

Subscription state is now handled by a single consumer, consuming a single queue. The SUBSCRIBER_STATUS data field is no longer synced via customer sync. Bulk subscriber sync will take place at onboarding for initial data ingestion, thereafter queues will take over, and `subscriber_imported` will stay at 1 (imported) unless there is a subscribers reset or a fresh `data:migrate` happens.

Messages in the `ddg.subscription.queue` can be 'subscribe', unsubscribe' or 'resubscribe' (for previously suppressed contacts).

## Why it's being changed

We were seeing mismatching values for SUBSCRIBER_STATUS between Magento and Dotdigital - but the way we had implemented queues partially (with queues for unsubscribes and resubscribes, cron for subscriptions, SUBSCRIBER_STATUS going via customer sync) meant that there has been the possibility for discrepancies to emerge in list membership as well.

## How to review / test this change

- Turn off cron
- Whilst on develop, queue some unsubscribes and resubscribes
- Switch to this branch, turn on cron
- Ensure the backlogged messages for ddg.newsletter.unsubscribe and ddg.newsletter.resubscribe are processed
- Test subscribe
- Test unsubscribe
- Test resubscribe
- Retest subscribe for a subscribed guest who has placed an order (turn on subscriber sales data)
- Test that an automation for a new subscriber leads to the contact being created in DD with the correct data fields (you might want to turn off cron and run the Automation CLI to isolate this process)
- Confirm that customer sync does not sync SUBSCRIBER_STATUS (it should not be a column in the CSV)

## Notes

- subscriber_imported should only be set to 0 in the case of a subscribers reset (never at any other time)
- In a db queue implementation, the SubscriptionConsumer will consume messages sequentially
- In an AMQP queue implementation, multiple consumers may be recruited to consume available messages in parallel, but it is expected that the queue is still consumed sequentially, albeit in a round-robin style. Some brokers can be configured to strictly process one message at a time (see [RabbitMQ docs](https://www.rabbitmq.com/docs/consumers#single-active-consumer)).

Related work items: #235800, #260620

Model/Queue/Data/ResubscribeData.php

@@ -4,6 +4,10 @@
 
 namespace Dotdigitalgroup\Email\Model\Queue\Data;
 
+/**
+ * @deprecated Use SubscriptionData as the model for all subscription state queue messages.
+ * @see SubscriptionData
+ */
 class ResubscribeData
 {
     /**

Model/Queue/Data/SubscriptionData.php

@@ -0,0 +1,120 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Dotdigitalgroup\Email\Model\Queue\Data;
+
+class SubscriptionData
+{
+    /**
+     * @var string|int
+     */
+    private $id;
+
+    /**
+     * @var string
+     */
+    private $email;
+
+    /**
+     * @var int
+     */
+    private $websiteId;
+
+    /**
+     * @var string
+     */
+    private $type;
+
+    /**
+     * Set id.
+     *
+     * This is the row id from email_contact.
+     *
+     * @param string|int $id
+     *
+     * @return void
+     */
+    public function setId($id)
+    {
+        $this->id = $id;
+    }
+
+    /**
+     * Set email.
+     *
+     * @param string $email
+     *
+     * @return void
+     */
+    public function setEmail(string $email)
+    {
+        $this->email = $email;
+    }
+
+    /**
+     * Set website id.
+     *
+     * @param string|int $websiteId
+     *
+     * @return void
+     */
+    public function setWebsiteId($websiteId)
+    {
+        (int) $this->websiteId = $websiteId;
+    }
+
+    /**
+     * Set type.
+     *
+     * @param string $type
+     *
+     * @return void
+     */
+    public function setType(string $type)
+    {
+        $this->type = $type;
+    }
+
+    /**
+     * Get id.
+     *
+     * @return string
+     */
+    public function getId()
+    {
+        return $this->id;
+    }
+
+    /**
+     * Get email.
+     *
+     * @return string
+     */
+    public function getEmail(): string
+    {
+        return $this->email;
+    }
+
+    /**
+     * Get website id.
+     *
+     * Type cast is NOT redundant.
+     *
+     * @return int
+     */
+    public function getWebsiteId(): int
+    {
+        return (int) $this->websiteId;
+    }
+
+    /**
+     * Get type.
+     *
+     * @return string
+     */
+    public function getType(): string
+    {
+        return $this->type;
+    }
+}

Model/Queue/Data/UnsubscriberData.php

@@ -4,6 +4,10 @@
 
 namespace Dotdigitalgroup\Email\Model\Queue\Data;
 
+/**
+ * @deprecated Use SubscriptionData as the model for all subscription state queue messages.
+ * @see SubscriptionData
+ */
 class UnsubscriberData
 {
     /**

Model/Queue/Newsletter/ResubscribeConsumer.php

@@ -10,6 +10,10 @@
 use Dotdigitalgroup\Email\Model\ResourceModel\Contact as ContactResource;
 use Magento\Newsletter\Model\Subscriber;
 
+/**
+ * @deprecated Subscriptions now use a single consumer.
+ * @see SubscriptionConsumer
+ */
 class ResubscribeConsumer
 {
     /**

Model/Queue/Newsletter/SubscriptionConsumer.php

@@ -0,0 +1,199 @@
+<?php
+
+namespace Dotdigitalgroup\Email\Model\Queue\Newsletter;
+
+use Dotdigitalgroup\Email\Helper\Data;
+use Dotdigitalgroup\Email\Logger\Logger;
+use Dotdigitalgroup\Email\Model\Apiconnector\Client;
+use Dotdigitalgroup\Email\Model\Connector\ContactData;
+use Dotdigitalgroup\Email\Model\ContactFactory;
+use Dotdigitalgroup\Email\Model\ResourceModel\Contact as ContactResource;
+use Dotdigitalgroup\Email\Model\Queue\Data\SubscriptionData;
+use Dotdigitalgroup\Email\Model\Sync\Subscriber\SingleSubscriberSyncer;
+use Magento\Framework\Exception\LocalizedException;
+use Magento\Newsletter\Model\Subscriber;
+
+class SubscriptionConsumer
+{
+    /**
+     * @var Data
+     */
+    private $helper;
+
+    /**
+     * @var Logger
+     */
+    private $logger;
+
+    /**
+     * @var ContactData
+     */
+    private $contactData;
+
+    /**
+     * @var ContactFactory
+     */
+    private $contactFactory;
+
+    /**
+     * @var ContactResource
+     */
+    private $contactResource;
+
+    /**
+     * @var SingleSubscriberSyncer
+     */
+    private $singleSubscriberSyncer;
+
+    /**
+     * @param Data $helper
+     * @param Logger $logger
+     * @param ContactData $contactData
+     * @param ContactFactory $contactFactory
+     * @param ContactResource $contactResource
+     * @param SingleSubscriberSyncer $singleSubscriberSyncer
+     */
+    public function __construct(
+        Data $helper,
+        Logger $logger,
+        ContactData $contactData,
+        ContactFactory $contactFactory,
+        ContactResource $contactResource,
+        SingleSubscriberSyncer $singleSubscriberSyncer
+    ) {
+        $this->helper = $helper;
+        $this->logger = $logger;
+        $this->contactData = $contactData;
+        $this->contactFactory = $contactFactory;
+        $this->contactResource = $contactResource;
+        $this->singleSubscriberSyncer = $singleSubscriberSyncer;
+    }
+
+    /**
+     * Process consumer.
+     *
+     * @param SubscriptionData $data
+     *
+     * @return void
+     * @throws LocalizedException
+     */
+    public function process(SubscriptionData $data)
+    {
+        if (!$data->getType()) {
+            throw new LocalizedException(__('Unknown subscription type'));
+        }
+
+        $client = $this->helper->getWebsiteApiClient($data->getWebsiteId());
+        $listId = (int) $this->helper->getSubscriberAddressBook($data->getWebsiteId());
+
+        switch ($data->getType()) {
+            case 'subscribe':
+                $this->subscribe($data);
+                break;
+            case 'unsubscribe':
+                $this->unsubscribe($data, $client, $listId);
+                break;
+            case 'resubscribe':
+                $this->resubscribe($data, $client, $listId);
+                break;
+        }
+    }
+
+    /**
+     * Subscribe.
+     *
+     * @param SubscriptionData $subscribeData
+     *
+     * @return void
+     */
+    private function subscribe(SubscriptionData $subscribeData)
+    {
+        $contact = $this->contactFactory->create();
+        $this->contactResource->load($contact, $subscribeData->getId());
+
+        try {
+            $this->singleSubscriberSyncer->pushContactToSubscriberAddressBook($contact);
+            $this->logger->info('Newsletter subscribe success', ['email' => $subscribeData->getEmail()]);
+        } catch (\Exception $e) {
+            $this->logger->error(
+                'Newsletter subscribe error',
+                [
+                'identifier' => $subscribeData->getEmail(),
+                'exception' => $e,
+                ]
+            );
+        }
+    }
+
+    /**
+     * Unsubscribe.
+     *
+     * @param SubscriptionData $unsubscribeData
+     * @param Client $client
+     * @param int $listId
+     *
+     * @return void
+     */
+    private function unsubscribe(SubscriptionData $unsubscribeData, Client $client, int $listId)
+    {
+        $data[] = [
+            'Key' => 'SUBSCRIBER_STATUS',
+            'Value' => $this->contactData->getSubscriberStatusString(
+                Subscriber::STATUS_UNSUBSCRIBED
+            )
+        ];
+
+        try {
+            $result = $client->updateContactDatafieldsByEmail($unsubscribeData->getEmail(), $data);
+
+            if (isset($result->id)) {
+                $contactId = $result->id;
+                $client->deleteAddressBookContact(
+                    $listId,
+                    $contactId
+                );
+            } else {
+                $this->contactResource->setContactSuppressedForContactIds([$unsubscribeData->getId()]);
+            }
+            $this->logger->info('Newsletter unsubscribe success', ['email' => $unsubscribeData->getEmail()]);
+        } catch (\Exception $e) {
+            $this->logger->error(
+                'Newsletter unsubscribe error',
+                [
+                'identifier' => $unsubscribeData->getEmail(),
+                'exception' => $e,
+                ]
+            );
+        }
+    }
+
+    /**
+     * Resubscribe.
+     *
+     * @param SubscriptionData $resubscribeData
+     * @param Client $client
+     * @param int $listId
+     *
+     * @return void
+     */
+    private function resubscribe(SubscriptionData $resubscribeData, Client $client, int $listId)
+    {
+        try {
+            ($listId) ?
+                $client->postAddressBookContactResubscribe(
+                    $listId,
+                    $resubscribeData->getEmail()
+                ) :
+                $client->resubscribeContactByEmail($resubscribeData->getEmail());
+            $this->logger->info('Newsletter resubscribe success', ['email' => $resubscribeData->getEmail()]);
+        } catch (\Exception $e) {
+            $this->logger->error(
+                'Newsletter resubscribe error',
+                [
+                'identifier' => $resubscribeData->getEmail(),
+                'exception' => $e,
+                ]
+            );
+        }
+    }
+}

Model/Queue/Newsletter/UnsubscriberConsumer.php

@@ -11,6 +11,10 @@
 use Dotdigitalgroup\Email\Model\ResourceModel\Contact as ContactResource;
 use Magento\Newsletter\Model\Subscriber;
 
+/**
+ * @deprecated Subscriptions now use a single consumer.
+ * @see SubscriptionConsumer
+ */
 class UnsubscriberConsumer
 {
     /**