video_server/app/Jobs/MessageDoReadJob.php

<?php

namespace App\Jobs;

use App\Model\Mongo\MessageModel;
use Carbon\Carbon;
use Hyperf\AsyncQueue\Job;
use MongoDB\BSON\ObjectId;

class MessageDoReadJob extends Job
{
    public $params = [];

    public function __construct($params)
    {
        // 这里最好是普通数据，不要使用携带 IO 的对象，比如 PDO 对象
        $this->params = $params;
    }

    public function handle()
    {
        $ids = $this->params['ids'];
        if (!empty($ids) && is_array($ids)) {
            $readAt = toMongoIsoDate(Carbon::now());
            foreach ($ids as $id) {
                $mongoId = new ObjectId($id);
                $filters = [
                    '_id' => $mongoId,
                ];
                make(MessageModel::class)->db()->updateOne(
                    $filters,
                    [
                        '$set' => [
                            'read_at' => $readAt
                        ]
                    ]
                );
            }
        }
    }
}