<?php

declare(strict_types=1);

namespace App\Command;

use Hyperf\Command\Command as HyperfCommand;
use Hyperf\Command\Annotation\Command;
use Psr\Container\ContainerInterface;
use App\Platform\OrderConsumer;
use Hyperf\Amqp\ConnectionFactory;
use Hyperf\Amqp\Result;
use PhpAmqpLib\Message\AMQPMessage;
use Hyperf\Di\Annotation\Inject;

#[Command]
class AppMessageQueuePullTmall extends HyperfCommand
{

    #[Inject]
    private OrderConsumer $orderConsumer;

    public function __construct(protected ContainerInterface $container)
    {
        parent::__construct('app:mq-pull:tmall');
    }

    public function configure()
    {
        parent::configure();
        $this->setDescription('Manually pull and consume one message from orders.queue');
    }

    public function handle()
    {
        $connectionFactory = $this->container->get(ConnectionFactory::class);
        $connection = $connectionFactory->getConnection('default_consumer');
        $channel = $connection->getChannel();

        $queueName = 'orders.queue';

        $this->info("Pulling one message from queue: {$queueName}");

        // 使用 basic_get 手动拉取一条消息（非阻塞）
        $message = $channel->basic_get($queueName, false);

        if ($message === null) {
            $this->warn('No messages available in queue');
            return 0;
        }

        try {
            // 解析消息体
            $data = json_decode($message->body, true);

            $this->info('Received message:');
            $this->line(json_encode($data, JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE));

            // 创建消费者实例并处理消息
            $result = $this->orderConsumer->consumeMessage($data, $message);

            // 根据消费结果进行 ACK/NACK
            switch ($result) {
                case Result::ACK:
                    $channel->basic_ack($message->getDeliveryTag());
                    $this->info('Message consumed successfully (ACK)');
                    break;
                case Result::NACK:
                    $channel->basic_nack($message->getDeliveryTag(), false, true);
                    $this->warn('Message consumption failed (NACK - requeue)');
                    break;
                case Result::DROP:
                    $channel->basic_nack($message->getDeliveryTag(), false, false);
                    $this->warn('Message dropped (NACK - no requeue)');
                    break;
                case Result::REQUEUE:
                    $channel->basic_nack($message->getDeliveryTag(), false, true);
                    $this->warn('Message requeued');
                    break;
            }

            return 0;
        } catch (\Throwable $e) {
            $this->error('Error consuming message: ' . $e->getMessage());
            $this->error($e->getTraceAsString());

            // 发生异常时，NACK 并重新入队
            $channel->basic_nack($message->getDeliveryTag(), false, true);
            return 1;
        }
    }
}