video_server/app/Repositories/MessageRepository.php

<?php

namespace App\Repositories;

use App\Jobs\MessageDoReadJob;
use App\Model\Mongo\MessageModel;
use Carbon\Carbon;
use MongoDB\BSON\ObjectId;

/**
 * 站内用户消息，非MySQL
 */
class MessageRepository
{
    public function coreSend($params)
    {
        $nowDate = toMongoIsoDate(Carbon::now());
        if (!empty($params['created_at'])) {
            $createdDate = Carbon::parse($params['created_at']);
        } else {
            $createdDate = Carbon::now();
        }
        $createdAt = toMongoIsoDate($createdDate);
        // 没有则创建, 系统消息是不能删除的.
        $setarr = [
            'subject' => data_get($params, 'subject', ''),
            'struct_type' => data_get($params, 'struct_type', 'text'),
            'receive_uid' => data_get($params, 'receive_uid', ''),
            'receive_type' => data_get($params, 'receive_type', 'user'),
            'from_uid' => data_get($params, 'from_uid', ''),
            'from_type' => data_get($params, 'from_type', 'sys'),
            'sys_msg_uid' => data_get($params, 'sys_msg_uid', ''),
            'content' => data_get($params, 'content', ''),
            'created_at' => $createdAt,
            'updated_at' => $nowDate,
            'read_at' => null,  // null代表未阅读
        ];
        make(MessageModel::class)->db()->insertOne($setarr);

        return true;
    }

    public function sendTextFromSys($receiveUid, $content, $subject = '')
    {
        $contentStruct = [
            'text' => $content,
        ];
        $params = [
            'subject' => $subject,
            'content' => $contentStruct,
            'struct_type' => 'text',
            'receive_uid' => $receiveUid,
            'receive_type' => 'user',
            'from_type' => 'sys',
        ];

        return $this->coreSend($params);
    }

    public function sendTextFromUser($receiveUid, $fromUid, $content, $subject = '')
    {
        $contentStruct = [
            'text' => $content,
        ];
        $params = [
            'subject' => $subject,
            'content' => $contentStruct,
            'struct_type' => 'text',
            'receive_uid' => $receiveUid,
            'receive_type' => 'user',
            'from_uid' => $fromUid,
            'from_type' => 'user',
        ];

        return $this->coreSend($params);
    }

    /**
     * 根据上一次时间获取分页系统消息数据
     * @param $receiveUid
     * @param $previousDate
     * @param int $limitNum
     * @return array
     */
    public function previousSysPages($receiveUid, $previousDate, $limitNum = 20, $autoRead = false)
    {
        $createAt = toMongoIsoDate(Carbon::parse($previousDate));
        $filters = [
            'receive_uid' => (string)$receiveUid,
            'from_type' => 'sys',
            'created_at' => [
                '$lt' => $createAt,
            ],
        ];
        $options = [
            'sort' => [
                'created_at' => -1,   // 时间降序
            ],
            'limit' => $limitNum,
        ];
//        $count = make(MessageModel::class)->db()->countDocuments($filters);
        $unreadIds = [];
        $cursor = make(MessageModel::class)->db()->find($filters, $options);
        $data = [];
        if (!empty($cursor)) {
            $data = $cursor->toArray();
            foreach ($data as $k => &$v) {
                $v['msg_id'] = $v['_id']->__toString();
                if ($autoRead === true) {
                    if (is_null($v['read_at'])) {
                        // 未读，自动标记已读
                        $unreadIds[] = $v['msg_id'];
                    }
                }
                $v['created_at'] = mongoIsoDateToCarbonTime($v['created_at'])->toDateTimeString();
                $v['updated_at'] = mongoIsoDateToCarbonTime($v['updated_at'])->toDateTimeString();
                $v['read_at'] = !empty($v['read_at']) ? mongoIsoDateToCarbonTime($v['read_at'])->toDateTimeString() : null;
                unset($v['_id'], $v['from_uid'], $v['from_type']);
            }
//            $data = array_reverse($data);   // 时序反转
        }

        if ($autoRead === true && !empty($unreadIds)) {
            // 异步自动标记已读.
            $jobParams = [
                'ids' => $unreadIds,
            ];
            dispatch_job(new MessageDoReadJob($jobParams));
        }

        return $data;
    }

    /**
     * 标记一个消息已读.
     * @param string $msgId
     * @param mixed $receiveUid
     * @return bool
     */
    public function doRead($msgId, $receiveUid = null)
    {
        $mongoId = new ObjectId($msgId);
        $readAt = toMongoIsoDate(Carbon::now());

        $filters = [
            '_id' => $mongoId,
        ];
        if (!empty($receiveUid)) {
            $filters['receive_uid'] = $receiveUid;
        }
        $count = make(MessageModel::class)->db()->countDocuments($filters);
        if ($count) {
            make(MessageModel::class)->db()->updateOne(
                $filters,
                [
                    '$set' => [
                        'read_at' => $readAt
                    ]
                ]
            );

            return true;
        } else {
            return false;
        }
    }

    /**
     * 静默删除多个.
     * @param $msgIds
     * @param null $receiveUid
     * @return bool
     */
    public function slientDelete($msgIds, $receiveUid = null)
    {
        if (empty($msgIds)) {
            return false;
        }
        if (is_string($msgIds)) {
            $msgIds = [$msgIds];
        }
        $msgIds = array_filter($msgIds);
        $msgIds = array_unique($msgIds);
        if (empty($msgIds)) {
            return false;
        }
        $arrMongoIds = [];
        foreach ($msgIds as $msgId) {
            $mongoId = new ObjectId($msgId);
            $arrMongoIds[] = $mongoId;
        }
        $filters = [
            '_id' => [
                '$in' => $arrMongoIds,
            ],
        ];
        if (!empty($receiveUid)) {
            $filters['receive_uid'] = $receiveUid;
        }

        $lines = make(MessageModel::class)->db()->deleteMany(
            $filters
        )->getDeletedCount();
        if ($lines > 0) {
            return true;
        } else {
            return false;
        }
    }

    /*
     * 获取第一条系统消息，包含统计.
     */
    public function firstSys($receiveUid, $msgType = 'sys')
    {
        $filters = [
            'receive_uid' => (string)$receiveUid,
            'from_type' => $msgType,
        ];
        $countAll = make(MessageModel::class)->db()->countDocuments($filters);

        $filters = [
            'receive_uid' => (string)$receiveUid,
            'from_type' => $msgType,
            'read_at' =>  null,
        ];
        $countUnread = make(MessageModel::class)->db()->countDocuments($filters);

        $filters = [
            'receive_uid' => (string)$receiveUid,
            'from_type' => $msgType,
        ];
        $options = [
            'sort' => [
                'created_at' => -1,   // 时间降序
            ],
            'limit' => 1,
        ];
        $firstMsg = make(MessageModel::class)->db()->findOne($filters, $options);
        if (!empty($firstMsg->_id)) {
            $firstMsg['msg_id'] = $firstMsg['_id']->__toString();
            $firstMsg['created_at'] = mongoIsoDateToCarbonTime($firstMsg['created_at'])->toDateTimeString();
            $firstMsg['updated_at'] = mongoIsoDateToCarbonTime($firstMsg['updated_at'])->toDateTimeString();
            $firstMsg['read_at'] = !empty($firstMsg['read_at']) ? mongoIsoDateToCarbonTime($firstMsg['read_at'])->toDateTimeString() : null;
            unset($firstMsg['_id'], $firstMsg['from_type']);
        } else {
            $firstMsg = null;
        }

        $result = [
            'first_msg' => $firstMsg,
            'count_all' => $countAll,
            'count_unread' => $countUnread,
        ];

        return $result;
    }
}