message.php

См. документацию.

<?php
namespace Bitrix\MessageService\Internal\Entity;
 
use Bitrix\Main\Application;
use Bitrix\Main\ORM\Data\DataManager;
use Bitrix\Main\ORM\Fields\ArrayField;
use Bitrix\Main\ORM\Fields\DatetimeField;
use Bitrix\Main\ORM\Fields\EnumField;
use Bitrix\Main\ORM\Fields\ExpressionField;
use Bitrix\Main\ORM\Fields\IntegerField;
use Bitrix\Main\ORM\Fields\StringField;
use Bitrix\Main\ORM\Fields\TextField;
use Bitrix\Main\ORM\Fields\Validators\LengthValidator;
use Bitrix\Main\Type\DateTime;
use Bitrix\MessageService\Internal\Entity\Message\SuccessExec;

class MessageTable extends DataManager
{
    public static function getTableName()
    {
        return 'b_messageservice_message';
    }

    public static function getMap()
    {
        return [
            'ID' =>
                (new IntegerField('ID', []))
                    ->configurePrimary(true)
                    ->configureAutocomplete(true)
            ,
            'TYPE' =>
                (new StringField('TYPE', [
                    'validation' => [__CLASS__, 'validateType']
                ]))
                    ->configureRequired(true)
            ,
            'SENDER_ID' =>
                (new StringField('SENDER_ID', [
                    'validation' => [__CLASS__, 'validateSenderId']
                ]))
                    ->configureRequired(true)
            ,
            'AUTHOR_ID' => (new IntegerField('AUTHOR_ID',
                []
            ))
                ->configureDefaultValue(0),
            'MESSAGE_FROM' =>
                (new StringField('MESSAGE_FROM', [
                    'validation' => [__CLASS__, 'validateMessageFrom']
                ]))
            ,
            'MESSAGE_TO' =>
                (new StringField('MESSAGE_TO', [
                    'validation' => [__CLASS__, 'validateMessageTo']
                ]))
                    ->configureRequired(true)
            ,
            'MESSAGE_HEADERS' =>
                (new ArrayField('MESSAGE_HEADERS', []))
                    ->configureSerializationPhp()
            ,
            'MESSAGE_BODY' =>
                (new TextField('MESSAGE_BODY', []))
                    ->configureRequired(true)
            ,
            'DATE_INSERT' =>
                (new DatetimeField('DATE_INSERT'))
                    ->configureDefaultValue(static fn() => new DateTime())
            ,
            'DATE_EXEC' =>
                (new DatetimeField('DATE_EXEC', []))
            ,
            'NEXT_EXEC' =>
                (new DatetimeField('NEXT_EXEC', []))
            ,
            'SUCCESS_EXEC' =>
                (new EnumField('SUCCESS_EXEC', []))
                    ->configureValues([
                        SuccessExec::NO,
                        SuccessExec::YES,
                        SuccessExec::ERROR,
                        SuccessExec::PROCESSED,
                    ])
                    ->configureDefaultValue(SuccessExec::NO)
            ,
            'EXEC_ERROR' =>
                (new StringField('EXEC_ERROR', [
                    'validation' => [__CLASS__, 'validateExecError']
                ]))
            ,
            'STATUS_ID' =>
                (new IntegerField('STATUS_ID', []))
                    ->configureDefaultValue(0)
            ,
            'EXTERNAL_ID' =>
                (new StringField('EXTERNAL_ID', [
                    'validation' => [__CLASS__, 'validateExternalId']
                ]))
            ,
            'EXTERNAL_STATUS' =>
                (new StringField('EXTERNAL_STATUS', [
                    'validation' => [__CLASS__, 'validateExternalStatus']
                ]))
            ,
            'CLUSTER_GROUP' =>
                (new IntegerField('CLUSTER_GROUP', []))
            ,
        ];
    }
 
    public static function getByExternalId(string $senderId, string $externalId, ?string $from = null)
    {
        return MessageTable::getList([
            'filter' => [
                '=SENDER_ID' => $senderId,
                '=EXTERNAL_ID' => $externalId,
            ],
            'limit' => 1
        ]);
    }

    public static function updateStatusId(int $id, int $newStatusId): bool
    {
        $connection = Application::getConnection();
        $tableName = static::getTableName();
 
        $connection->query("
            UPDATE
                {$tableName}
            SET
                STATUS_ID = {$newStatusId}
            WHERE
                ID = $id
                AND STATUS_ID != {$newStatusId}
        ");
 
        return $connection->getAffectedRowsCount() === 1;
    }
 
    public static function updateMessageStatuses($id, $newInternalStatusId, $newExternalStatus): bool
    {
        $connection = Application::getConnection();
        $tableName = static::getTableName();
 
        $newExternalStatus = $connection->getSqlHelper()->forSql($newExternalStatus);
 
        $connection->query("
            UPDATE
                {$tableName}
            SET
                STATUS_ID = {$newInternalStatusId},
                EXTERNAL_STATUS = '{$newExternalStatus}'
            WHERE
                ID = {$id}
                AND STATUS_ID < {$newInternalStatusId}
        ");
 
        return $connection->getAffectedRowsCount() === 1;
    }
 
    public static function getDailyCount($senderId, $fromId): int
    {
        $today = (new DateTime)->setTime(0, 0, 0);
 
        return self::getCount([
            '=SUCCESS_EXEC' => 'Y',
            '>=DATE_EXEC' => $today,
            '=SENDER_ID' => $senderId,
            '=MESSAGE_FROM' => $fromId,
        ]);
    }
 
    public static function getAllDailyCount(): array
    {
        $today = (new DateTime)->setTime(0, 0, 0);
 
        $result = self::getList([
            'runtime' => [new ExpressionField('CNT', 'COUNT(*)')],
            'select' => [
                'SENDER_ID', 'MESSAGE_FROM', 'CNT'
            ],
            'filter' => [
                '=SUCCESS_EXEC' => 'Y',
                '>=DATE_EXEC' => $today,
            ],
            'group' => ['SENDER_ID', 'MESSAGE_FROM'],
        ]);
 
        $counts = [];
        while ($row = $result->fetch())
        {
            $id = $row['SENDER_ID'] .':'. $row['MESSAGE_FROM'];
            $counts[$id] = (int)$row['CNT'];
        }
 
        return $counts;
    }
 
    public static function returnDeferredToQueue($senderId, $fromId): bool
    {
        $connection = Application::getConnection();
        $helper = $connection->getSqlHelper();
 
        $senderId = $helper->forSql((string)$senderId);
        $fromId = $helper->forSql((string)$fromId);
 
        $connection->queryExecute("
            UPDATE b_messageservice_message
            SET NEXT_EXEC = NULL
            WHERE
                SUCCESS_EXEC = 'N'
                AND NEXT_EXEC IS NOT NULL
                AND SENDER_ID = '{$senderId}'
                AND MESSAGE_FROM = '{$fromId}'
        ");
 
        return true;
    }

    public static function validateType(): array
    {
        return [
            new LengthValidator(null, 30),
        ];
    }

    public static function validateSenderId(): array
    {
        return [
            new LengthValidator(null, 50),
        ];
    }

    public static function validateMessageFrom(): array
    {
        return [
            new LengthValidator(null, 260),
        ];
    }

    public static function validateMessageTo(): array
    {
        return [
            new LengthValidator(null, 50),
        ];
    }

    public static function validateExecError(): array
    {
        return [
            new LengthValidator(null, 255),
        ];
    }

    public static function validateExternalId(): array
    {
        return [
            new LengthValidator(null, 128),
        ];
    }

    public static function validateExternalStatus(): array
    {
        return [
            new LengthValidator(null, 128),
        ];
    }
 
    public static function onAfterAdd(\Bitrix\Main\ORM\Event $event)
    {
        self::clearQueueCache();
    }
 
    public static function onAfterUpdate(\Bitrix\Main\ORM\Event $event)
    {
        self::clearQueueCache();
    }
 
    public static function onAfterDelete(\Bitrix\Main\ORM\Event $event)
    {
        self::clearQueueCache();
    }
 
    protected static function clearQueueCache(): void
    {
        $cache = \Bitrix\Main\Data\Cache::createInstance();
        $cache->clean(\Bitrix\MessageService\Queue::CACHE_HAS_MESSAGES_ID, \Bitrix\MessageService\Queue::CACHE_HAS_MESSAGES_DIR);
    }
}