Рассылка сообщений по каналу WhatsApp происходит в несколько этапов.

1. Для каждого клиента рассылки определяем, не было ли нами замечено ранее, что у клиента нет WhatsApp.
Информацию о том, есть WhatsApp или нет мы можем получить только после отправки сообщения. Если после первой отправки мы определяем, что у клиента нет WhatsApp, то мы фиксируем это в специальной таблице в БД. А тут проверяем наличие записи в этой таблице.

2. Для каждого клиента рассылки определяем, с какого номера ему нужно отправить сообщение.

namespace Bizprofi\Cdp\Helpers;

class WhatsappSenderDetectHelper
{
// ... тут есть другая логика хелпера
    public function getSenderByClient(Contact|Company $client, bool $first = false): ?WhatsappSender
    {
        // Есть ли у клиента whatsapp
        if (! $this->getWhatsappAvailabilityByClient($client)) {
            return null;
        }

        // Если в контакте/компании в специальном свойстве указан номер с которого отправлять
        if ($sender = $this->getSenderByClientUf($client)) {
            return $sender;
        }

        // Если клиенту уже отправлялось сообщение ранее, но он на него не среагировал
        if ($sender = $this->getSenderByLastMessage($client)) {
            return $sender;
        }

        // Отправляем сообщение с первого из доступных номеров для отправки.
        if ($first && ($sender = $this->getFirstAvailableSender($client))) {
            return $sender;
        }

        // Отправляем сообщение с случайного из доступных номеров для отправки.
        if ($sender = $this->getRandomAvailableSender($client)) {
            return $sender;
        }

        return null;
    }

    public function getWhatsappAvailabilityByClient(Contact|Company $client): bool
    {
        $clientInfo = $this->whatsappClientInfos->getByPrimary([
            'ENTITY' => $this->entity,
            'ENTITY_ID' => $client->getId(),
        ]);

        // Если нет записи предполагаем что есть whatsapp
        if (! $clientInfo) {
            return true;
        }

        return $clientInfo->getWhatsappAvailability();
    }

    protected function getSenderByClientUf(Contact|Company $client): ?WhatsappSender
    {
        if (! ($uf = $this->getUserFieldCode())) {
            return null;
        }

        try {
            $phoneNumber = $client->get($uf);
        } catch (\Exception $ex) {
            $phoneNumber = null;
        }

        $senders = WhatsappSenderTable::getAll();
        if (! ($sender = $senders->getByPrimary($phoneNumber))) {
            return null;
        }

        if ($sender->getInSpamAtClient($client)) {
            return null;
        }

        return $sender;
    }

    protected function getUserFieldCode(): ?string
    {
        return match ($this->entity) {
            EntityType::COMPANY => WhatsappSettingsHelper::getCompanyPhoneNumberField(),
            EntityType::CONTACT => WhatsappSettingsHelper::getContactPhoneNumberField(),
        };
    }

    protected function getSenderByLastMessage(Contact|Company $client): ?WhatsappSender
    {
        /**
         * @var EO_WhatsappMessage $lastMessage
         */
        if (! ($lastMessage = $this->clientIdToLastMessage[$client->getId()])) {
            return null;
        }

        $senders = WhatsappSenderTable::getAll();
        if (! ($sender = $senders->getByPrimary($lastMessage->getPhoneNumber()))) {
            return null;
        }

        if ($sender->getInSpamAtClient($client)) {
            return null;
        }

        $secondsFromLastMessage = (new DateTime())->getTimestamp() - $lastMessage->getSendDate()->getTimestamp();
        if ($secondsFromLastMessage <= WhatsappSettingsHelper::getIgnoreMessageSeconds()) {
            return $sender;
        }

        if ($sender->getFreeClientsCount() > 0) {
            return $sender;
        }

        return null;
    }

    protected function getFirstAvailableSender(Contact|Company $client): ?WhatsappSender
    {
        if (!($clientInGroups = $this->getClientSenderGroups($client))) {
            return null;
        }

        foreach (WhatsappSenderTable::getAll()->getAvailableSenders() as $sender) {
            if ($sender->getInSpamAtClient($client)) {
                continue;
            }

            if (count(array_intersect($clientInGroups, $sender->getGroups()->getIdList())) > 0) {
                return $sender;
            }
        }

        return null;
    }

    protected function getClientSenderGroups(Contact|Company $client): ?array
    {
        $clientInGroups = [];
        foreach ($this->groupIdToClientIds as $groupId => $clientIds) {
            if (in_array($client->getId(), $clientIds, true)) {
                $clientInGroups[] = $groupId;
            }
        }

        return count($clientInGroups) > 0 ? $clientInGroups: null;
    }

    protected function getRandomAvailableSender(Contact|Company $client): ?WhatsappSender
    {
        if (!($clientInGroups = $this->getClientSenderGroups($client))) {
            return null;
        }

        $allAvailableSenders = WhatsappSenderTable::getAll()->getAvailableSenders();

        $allowedSenderNumbers = [];
        foreach ($allAvailableSenders as $sender) {
            if ($sender->getInSpamAtClient($client)) {
                continue;
            }

            if (count(array_intersect($clientInGroups, $sender->getGroups()->getIdList())) <= 0) {
                continue;
            }

            $allowedSenderNumbers[] = $sender->getPhoneNumber();
        }

        if (count($allowedSenderNumbers) <= 0) {
            return null;
        }

        if (count($allowedSenderNumbers) === 1) {
            return $allAvailableSenders->getByPrimary($allowedSenderNumbers[0]);
        }

        return $allAvailableSenders->getByPrimary(
            $allowedSenderNumbers[array_rand($allowedSenderNumbers)]
        );
    }
// ... там далее другая логика хелпера
}

3. Если номер, с которого отправлять удалось определить, то добавляем запись в таблицу WhatsApp сообщений со статусом "Ожидает отправки". Если не удалось определить, то добавляем со статусом "Не валидно" и комментарием с причиной не валидности.

4. Раз в 15 минут запускается агент, который проверяет таблицу WhatsApp сообщений на наличие писем, которые нужно отправить. Если такие письма есть, то для каждого номера, с которого надо отправить в фоновом режиме запускаем отправку сообщений. В фоновом режиме, чтобы не блокировать работу агентов, поскольку в процессе отправки стоят паузы.

5. Контроллер отправки сообщений с конкретного номера берет из таблицы WhatsApp сообщений по 12 сообщений за итерацию и поочередно отправляет их, после каждой отправки выставляется пауза соответственно установленным ограничениям на номере. Если по завершению итерации в базе еще есть сообщения, которые нужно отправить с этого номера, то контроллер вызывает сам себя в фоновом режиме и завершает работу. И так до тех пор пока не будут отправлены все сообщения. В запуске контроллера предусмотрена защита от повторного запуска, если отправка с номера телефона уже запущена, то повторная — не запустится.

namespace Bizprofi\Cdp\Controllers\Actions\DistributionWhatsapp;

use Bitrix\Main\Application;
use Bitrix\Main\Config\Option;
use Bitrix\Main\Engine\Action;
use Bitrix\Main\Engine\UrlManager;
use Bitrix\Main\Error;
use Bitrix\Main\ORM\Objectify\State;
use Bitrix\Main\Result;
use Bitrix\Main\Type\DateTime;
use Bitrix\Main\Web\HttpClient;
use Bizprofi\Cdp\DataManagers\DistributionTable;
use Bizprofi\Cdp\DataManagers\Whatsapp\EO_WhatsappClientInfo;
use Bizprofi\Cdp\DataManagers\Whatsapp\EO_WhatsappClientInfo_Collection;
use Bizprofi\Cdp\DataManagers\Whatsapp\EO_WhatsappMessage;
use Bizprofi\Cdp\DataManagers\Whatsapp\WhatsappClientInfoTable;
use Bizprofi\Cdp\DataManagers\Whatsapp\WhatsappMessageTable;
use Bizprofi\Cdp\DataManagers\Whatsapp\WhatsappSenderHistoryTable;
use Bizprofi\Cdp\Enums\Whatsapp\WhatsappMessageStatus;
use Bizprofi\Cdp\Enums\Whatsapp\WhatsappSenderHistoryAction;
use Bizprofi\Cdp\Integrations\OlChat\ApiRequests\SendFileApiRequest;
use Bizprofi\Cdp\Integrations\OlChat\ApiRequests\SendTextApiRequest;
use Bizprofi\Cdp\Integrations\OlChat\ApiResponses\AbstractApiResponse;
use Bizprofi\Cdp\Integrations\OlChat\ApiResponses\FailureApiResponse;
use Bizprofi\Cdp\Integrations\OlChat\ApiResponses\SendFileApiResponse;
use Bizprofi\Cdp\Integrations\OlChat\ApiResponses\SendTextApiResponse;
use Bizprofi\Cdp\Models\Whatsapp\WhatsappSender;

class SendMessagesAction extends Action
{
    /**
     * BX.ajax.runAction('bizprofi:cdp.Controllers.DistributionWhatsappController.sendMessages', {data: {phoneNumber: '79111111111'}})
     * @param WhatsappSender $whatsappSender
     * @return array
     */
    public function run(WhatsappSender $whatsappSender): array
    {
        // Пишем в историю о начале отправки сообщений
        $result = $this->logStartTime($whatsappSender);
        if (! $result->isSuccess()) {
            $this->addErrors($result->getErrors());
            return [];
        }

        // Получаем апи клиент для отправки сообщений
        if (! ($whatsappSender->getApiClient())) {
            $this->addError(
                new Error('Failure get api client for whatsapp sender')
            );
            return [];
        }

        // Флаг наличия сообщений для отправки
        $thereAreMessages = false;

        // Переберем сообщения на отправку
        /**
         * @var EO_WhatsappMessage $message
         */
        foreach ($this->getNextMessages($whatsappSender) as $message) {
            $thereAreMessages = true;

            // Отправку и изменение сообщения в БД делаем в ттранзакции
            Application::getConnection()->startTransaction();

            // Меняем статус и дату отправки
            $message->setStatus(WhatsappMessageStatus::IN_PROCESS);
            $message->setSendDate(new DateTime());
            $result = $message->save();
            if (! $result->isSuccess()) {
                $this->addErrors($result->getErrors());
                Application::getConnection()->rollbackTransaction();
                continue;
            }

            // Отправляем сообщение
            $result = $this->sendWhatsappMessage($whatsappSender, $message);
            if (! $result->isSuccess()) {
                $this->addErrors($result->getErrors());
            }

            // Сохраняем сообщение еще раз после отправки, так как там поменялся статус
            $saveResult = $message->save();
            if (! $saveResult->isSuccess()) {
                $this->addErrors($saveResult->getErrors());
                Application::getConnection()->rollbackTransaction();
                continue;
            }

            // Закрываем транзакцию
            Application::getConnection()->commitTransaction();

            // Фиксируем в истории отправку сообщения
            $this->logSendMessage($whatsappSender, $message, $result);

            // Ставим паузу до отправки следующего сообщения
            $this->sleepAfterSendMessage($whatsappSender, $message);
        }

        // Повторяем запуск до тех пор пока не закончатся сообщения на отправку
        if ($thereAreMessages) {
            static::runSelfInBackground($whatsappSender);
        }

        return [];
    }

    protected function logStartTime(WhatsappSender $whatsappSender): Result
    {
        $history = WhatsappSenderHistoryTable::createObject();
        $history->setPhoneNumber($whatsappSender->getPhoneNumber());
        $history->setAction(WhatsappSenderHistoryAction::START_SENDING);
        $history->setActionDate(new DateTime());
        $history->setData([]);
        return $history->save();
    }

    protected function logFinishTime(WhatsappSender $whatsappSender): Result
    {
        $history = WhatsappSenderHistoryTable::createObject();
        $history->setPhoneNumber($whatsappSender->getPhoneNumber());
        $history->setAction(WhatsappSenderHistoryAction::FINISH_SENDING);
        $history->setActionDate(new DateTime());
        $history->setData([]);
        return $history->save();
    }

    protected function logSendMessage(WhatsappSender $whatsappSender, EO_WhatsappMessage $message, Result $result): Result
    {
        $history = WhatsappSenderHistoryTable::createObject();
        $history->setPhoneNumber($whatsappSender->getPhoneNumber());
        $history->setAction(WhatsappSenderHistoryAction::SEND);
        $history->setActionDate(new DateTime());
        $history->setData([
            'messageId' => $message->getId(),
            'status' => $result->isSuccess() ? 'success': 'failure',
            'data' => $result->getData(),
            'messages' => $result->getErrorMessages(),
        ]);
        return $history->save();
    }

    protected function getNextMessages(WhatsappSender $whatsappSender): \Generator
    {
        $currentHour = (new DateTime())->format('H');

        // Формируем запрос на получение сообщений для отправки
        $rows = WhatsappMessageTable::query()
            ->setSelect(['*', 'DSESES'])
            ->where('PHONE_NUMBER', $whatsappSender->getPhoneNumber())
            ->where('STATUS', WhatsappMessageStatus::WAIT)
            ->whereIn(
                'DSESES.DISTRIBUTION_ID',
                DistributionTable::query()
                    ->addSelect('ID')
                    ->where('DISTRIBUTION_WHATSAPP.SENDING_START_TIME', '<=', $currentHour)
                    ->where('DISTRIBUTION_WHATSAPP.SENDING_END_TIME', '>', $currentHour)
            )
            ->setLimit(12)
            ->exec();

        if ($rows->getSelectedRowsCount() <= 0) {
            $result = $this->logFinishTime($whatsappSender);
            if (! $result->isSuccess()) {
                $this->addErrors($result->getErrors());
            }

            return;
        }

        while ($whatsappMessage = $rows->fetchObject()) {
            yield $whatsappMessage;
        }
    }

    protected function sendWhatsappMessage(WhatsappSender $whatsappSender, EO_WhatsappMessage $message): Result
    {
        $message->setSendDate(new DateTime());

        $sendFileResponse = $this->sendWhatsappFileMessage($whatsappSender, $message);
        if (! $sendFileResponse->isSuccess()) {
            return $sendFileResponse;
        }

        $sendTextResponse = $this->sendWhatsappTextMessage($whatsappSender, $message);
        if (! $sendTextResponse->isSuccess()) {
            return $sendTextResponse;
        }

        return new Result();
    }

    protected function sendWhatsappFileMessage(WhatsappSender $whatsappSender, EO_WhatsappMessage $message): Result
    {
        if ((int) $message->getFileId() <= 0) {
            return new Result();
        }

        /**
         * @var FailureApiResponse|SendTextApiResponse $sendTextResponse
         */
        $response = $whatsappSender->getApiClient()->executeRequest(
            new SendFileApiRequest(
                $message->getClientPhoneNumber(),
                (int) $message->getFileId(),
                'Y'
            )
        );

        return $this->checkWhatsappMessageApiResponse($message, $response);
    }

    protected function checkWhatsappMessageApiResponse(EO_WhatsappMessage $message, AbstractApiResponse $response): Result
    {
        if ($response instanceof FailureApiResponse) {
            return $this->checkWhatsappMessageFailureApiResponse($message, $response);
        }

        if ($response instanceof SendFileApiResponse || $response instanceof SendTextApiResponse) {
            return $this->checkWhatsappMessageSuccessApiResponse($message, $response);
        }

        return new Result();
    }

    protected function checkWhatsappMessageFailureApiResponse(EO_WhatsappMessage $message, FailureApiResponse $response): Result
    {
        $message->setStatus(WhatsappMessageStatus::FAILURE);

        $result = new Result();
        $result->addError(
            new Error($response->message)
        );

        if (! $response->isNumberNotFoundFailure()) {
            return $result;
        }

        $saveClientInfoResult = $this->recordMessageNotWhatsappAvailability($message);
        if (! $saveClientInfoResult->isSuccess()) {
            $result->addErrors(
                $saveClientInfoResult->getErrors()
            );
        }

        return $result;
    }

    protected function recordMessageNotWhatsappAvailability(EO_WhatsappMessage $message): Result
    {
        $clientInfos = $this->getClientInfosByMessage($message);

        foreach ($clientInfos as $clientInfo) {
            $clientInfo->setWhatsappAvailability(false);
        }

        return $clientInfos->save(true);
    }

    protected function getClientInfosByMessage(EO_WhatsappMessage $message): EO_WhatsappClientInfo_Collection
    {
        $clientInfos = WhatsappClientInfoTable::createCollection();

        $dseses = $message->getDseses() ?: [];
        foreach ($dseses as $dses) {
            $clientInfo = WhatsappClientInfoTable::getByPrimary([
                'ENTITY' => $dses->getEntity(),
                'ENTITY_ID' => $dses->getEntityId(),
            ])->fetchObject();

            if (! $clientInfo) {
                $clientInfo = WhatsappClientInfoTable::createObject();
                $clientInfo->setEntity($dses->getEntity());
                $clientInfo->setEntityId($dses->getEntityId());
            }

            $clientInfos->add($clientInfo);
        }


        return $clientInfos;
    }

    protected function checkWhatsappMessageSuccessApiResponse(EO_WhatsappMessage $message, SendFileApiResponse|SendTextApiResponse $response): Result
    {
        $message->setWhatsappMessageId($response->messageId);

        if (! $response->isSent()) {
            $message->setStatus(WhatsappMessageStatus::SPAM);

            return $this->recordMessageHowSpam($message);
        }

        return $this->recordMessageWhatsappAvailability($message);
    }

    protected function recordMessageHowSpam(EO_WhatsappMessage $message): Result
    {
        $clientInfos = $this->getClientInfosByMessage($message);

        foreach ($clientInfos as $clientInfo) {
            $senderToSpamDate = $clientInfo->getSenderToSpamDate() ?: [];
            $senderToSpamDate[$message->getPhoneNumber()] = (new DateTime())->getTimestamp();

            $clientInfo->setSenderToSpamDate($senderToSpamDate);
        }

        return $clientInfos->save(true);
    }

    protected function recordMessageWhatsappAvailability(EO_WhatsappMessage $message): Result
    {
        $clientInfos = $this->getClientInfosByMessage($message);

        foreach ($clientInfos as $clientInfo) {
            if ($clientInfo->state !== State::RAW) {
                continue;
            }

            $clientInfo->setWhatsappAvailability(true);
            $clientInfo->setDateWhatsappAvailability(new DateTime());
        }

        return $clientInfos->save(true);
    }

    protected function sendWhatsappTextMessage(WhatsappSender $whatsappSender, EO_WhatsappMessage $message): Result
    {
        if (strlen((string) $message->getMessageBody()) <= 0) {
            return new Result();
        }

        /**
         * @var FailureApiResponse|SendTextApiResponse $sendTextResponse
         */
        $response = $whatsappSender->getApiClient()->executeRequest(
            new SendTextApiRequest(
                $message->getClientPhoneNumber(),
                $message->getMessageBody(),
                'Y'
            )
        );

        return $this->checkWhatsappMessageApiResponse($message, $response);
    }

    // Паузы между сообщениями делаем не статическими. Берем случайное значение между интервалом указанным в настройках модуля
    // и этим же интервалом увеличеным в двое для не не подтвержденных клиентов и в полтора раза для подтвержденных клиентов
    // таким образом имитируем человечность
    protected function sleepAfterSendMessage(WhatsappSender $whatsappSender, EO_WhatsappMessage $message): void
    {
        $factor = 2;
        $interval = $whatsappSender->getMessageIntervalForUnverified();
        if ($message->getIsConfirmed()) {
            $factor = 1.5;
            $interval = $whatsappSender->getMessageIntervalForVerified();
        }

        sleep(rand($interval, $interval * $factor));
    }

    public static function runSelfInBackground(WhatsappSender $whatsappSender): void
    {
        $httpClient = new HttpClient(['waitResponse' => false]);
        $httpClient->get(
            UrlManager::getInstance()->create(
                'bizprofi:cdp.Controllers.DistributionWhatsappController.sendMessages',
                [
                    'phoneNumber' => $whatsappSender->getPhoneNumber(),
                ],
                true
            )->withScheme((int) Option::get('bizprofi.cdp', 'use_https') === 1 ? 'https': 'http')
        );
    }
}

6. Раз в час запускается агент, который определяет номера телефонов, для которых синхронизация статусов не выполнялась более часа. Для таких номеров запускается синхронизация статусов сообщений в фоновом режиме. В фоновом режиме, чтобы не задерживать работу агентов.

7. Контроллер синхронизации статусов для конкретного номера выбирает 12 сообщений, статус которых — "Отправлено", и проверяет не изменился ли их статус, если статус изменился, то меняем его в рассылке.