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

use app\model\AsyncQueueMessage;
use app\repository\queue\AsyncQueueMessageRepository;
use app\service\domain\queue\QueueConfigService;
use Core\service\queue\QueueRegistry;
use Core\service\queue\contract\QueueDriverInterface;
use think\App;


/**
 * Database Queue Driver
 * @package app\infrastructure\queue
 */
class DatabaseQueueDriver implements QueueDriverInterface
{
    protected App $app;

    public function __construct(App $app)
    {
        $this->app = $app;
    }


    /**
     * 返回驱动名称
     * @return string
     */
    public function name(): string
    {
        return 'database';
    }


    /**
     * 发送消息到队列
     * @param string $topic
     * @param array $payload
     * @param array $options
     * @return array
     */
    public function dispatch(string $topic, array $payload, array $options = []): array
    {
        $queueName = trim((string)($options['queue'] ?? $this->config()->defaultQueue()));
        if ($queueName === '') {
            $queueName = $this->config()->defaultQueue();
        }

        $bizKey = trim((string)($options['biz_key'] ?? ''));
        if ($bizKey !== '') {
            $exists = $this->repository()->findByBizKey($bizKey);
            if ($exists instanceof AsyncQueueMessage) {
                return [
                    'mode' => 'database',
                    'topic' => $topic,
                    'queue' => $queueName,
                    'message_id' => (int)$exists->id,
                    'duplicate' => true,
                ];
            }
        }

        $delay = max(0, (int)($options['delay'] ?? 0));
        $maxAttempts = max(1, (int)($options['max_attempts'] ?? $this->config()->maxAttempts()));
        $message = $this->repository()->create([
            'topic' => $topic,
            'queue_name' => $queueName,
            'biz_key' => $bizKey,
            'payload' => json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
            'status' => AsyncQueueMessage::STATUS_PENDING,
            'attempts' => 0,
            'max_attempts' => $maxAttempts,
            'available_at' => time() + $delay,
            'reserved_at' => 0,
            'done_at' => 0,
            'last_error' => '',
        ]);

        return [
            'mode' => 'database',
            'topic' => $topic,
            'queue' => $queueName,
            'message_id' => (int)$message->id,
            'duplicate' => false,
        ];
    }


    /**
     * 消费队列消息
     * @param string $queueName 队列名称
     * @return array
     */
    public function consume(string $queueName = ''): array
    {
        $message = $this->repository()->reserveNext($queueName, time(), $this->config()->reserveSeconds());
        if (!$message instanceof AsyncQueueMessage) {
            return [
                'handled' => 0,
                'mode' => 'database',
                'queue' => $queueName,
            ];
        }

        try {
            $payload = json_decode((string)($message->payload ?? ''), true);
            if (!is_array($payload)) {
                throw new \RuntimeException('Invalid message payload format');
            }

            $this->registry()->handle((string)$message->topic, $payload);
            $this->repository()->markSuccess((int)$message->id, time());

            return [
                'handled' => 1,
                'mode' => 'database',
                'queue' => (string)($message->queue_name ?? ''),
                'topic' => (string)($message->topic ?? ''),
                'message_id' => (int)$message->id,
                'status' => 'done',
            ];
        } catch (\Throwable $e) {
            $this->repository()->markFailure($message, $e->getMessage(), time() + $this->config()->retryDelaySeconds());

            return [
                'handled' => 1,
                'mode' => 'database',
                'queue' => (string)($message->queue_name ?? ''),
                'topic' => (string)($message->topic ?? ''),
                'message_id' => (int)$message->id,
                'status' => 'failed',
                'error' => $e->getMessage(),
            ];
        }
    }

    /**
     * 获取队列配置服务实例
     * @return QueueConfigService
     */
    private function config(): QueueConfigService
    {
        return $this->app->make(QueueConfigService::class);
    }

    /**
     * 获取异步队列消息仓储实例
     * @return AsyncQueueMessageRepository
     */
    private function repository(): AsyncQueueMessageRepository
    {
        return $this->app->make(AsyncQueueMessageRepository::class);
    }

    /**
     * 获取队列注册中心
     * @return QueueRegistry
     */
    private function registry(): QueueRegistry
    {
        return $this->app->make(QueueRegistry::class);
    }
}
