<?php
// +----------------------------------------------------------------------
// | onezan-admin 快速开发的管理平台
// +----------------------------------------------------------------------
// | 官方网址：https://www.onezan.cn
// +----------------------------------------------------------------------
// | onezan团队 版权所有 开源版本可自由商用
// +----------------------------------------------------------------------
// | Author: 543885307@qq.com
// +----------------------------------------------------------------------
namespace app\repository\queue;

use app\model\AsyncQueueMessage;
use app\repository\BaseRepository;
use think\facade\Db;


/**
 * 异步队列消息仓储
 * @package app\repository\queue
 */
class AsyncQueueMessageRepository extends BaseRepository
{
    /**
     * 按业务键查找消息
     * @param string $bizKey
     * @return AsyncQueueMessage|null
     */
    public function findByBizKey(string $bizKey): ?AsyncQueueMessage
    {
        $bizKey = trim($bizKey);
        if ($bizKey === '') {
            return null;
        }

        $message = AsyncQueueMessage::where('biz_key', $bizKey)->find();
        return $message instanceof AsyncQueueMessage ? $message : null;
    }


    /**
     * 创建队列消息
     * @param array $data 消息数据
     * @return AsyncQueueMessage
     */
    public function create(array $data): AsyncQueueMessage
    {
        $message = new AsyncQueueMessage();
        $message->save($data);
        return $message;
    }


    /**
     * 预留下一个待处理消息（悲观锁）
     * @param string $queueName
     * @param int $now
     * @param int $reserveSeconds
     * @return AsyncQueueMessage|null
     */
    public function reserveNext(string $queueName, int $now, int $reserveSeconds): ?AsyncQueueMessage
    {
        $expiredReservedAt = $reserveSeconds > 0 ? $now - $reserveSeconds : 0;

        return Db::transaction(function () use ($queueName, $now, $expiredReservedAt) {
            $query = AsyncQueueMessage::where(function ($outer) use ($now, $expiredReservedAt) {
                $outer->where(function ($pending) use ($now) {
                    $pending->where('status', AsyncQueueMessage::STATUS_PENDING)
                        ->where('available_at', '<=', $now);
                })->whereOr(function ($processing) use ($expiredReservedAt) {
                    $processing->where('status', AsyncQueueMessage::STATUS_PROCESSING)
                        ->where('reserved_at', '>', 0)
                        ->where('reserved_at', '<=', $expiredReservedAt);
                });
            });

            if ($queueName !== '') {
                $query->where('queue_name', $queueName);
            }

            $message = $query->order('id', 'asc')->lock(true)->find();
            if (!$message instanceof AsyncQueueMessage) {
                return null;
            }

            $message->status = AsyncQueueMessage::STATUS_PROCESSING;
            $message->reserved_at = $now;
            $message->attempts = (int)($message->attempts ?? 0) + 1;
            $message->last_error = '';
            $message->save();

            return $message;
        });
    }


    /**
     * 标记消息为处理成功
     * @param int $id     消息ID
     * @param int $doneAt 完成时间戳
     * @return void
     */
    public function markSuccess(int $id, int $doneAt): void
    {
        AsyncQueueMessage::where('id', $id)->update([
            'status' => AsyncQueueMessage::STATUS_DONE,
            'done_at' => $doneAt,
            'reserved_at' => 0,
            'last_error' => '',
        ]);
    }


    /**
     * 标记消息为处理失败
     * @param AsyncQueueMessage $message     消息模型
     * @param string            $error       错误信息
     * @param int               $availableAt 下次重试时间戳
     * @return void
     */
    public function markFailure(AsyncQueueMessage $message, string $error, int $availableAt): void
    {
        $attempts = (int)($message->attempts ?? 0);
        $maxAttempts = max(1, (int)($message->max_attempts ?? 1));
        $status = $attempts >= $maxAttempts ? AsyncQueueMessage::STATUS_FAILED : AsyncQueueMessage::STATUS_PENDING;

        AsyncQueueMessage::where('id', (int)$message->id)->update([
            'status' => $status,
            'available_at' => $status === AsyncQueueMessage::STATUS_PENDING ? $availableAt : (int)($message->available_at ?? 0),
            'reserved_at' => 0,
            'last_error' => $this->truncateError($error),
        ]);
    }


    /**
     * 截断错误信息字符串
     * @param string $error 错误信息
     * @return string 截断后的信息
     */
    private function truncateError(string $error): string
    {
        if (function_exists('mb_substr')) {
            return mb_substr($error, 0, 2000);
        }

        return substr($error, 0, 2000);
    }
}