using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using TakeoutSaaS.Module.Messaging.Abstractions;
using TakeoutSaaS.Module.Messaging.Options;
using TakeoutSaaS.Module.Messaging.Serialization;

namespace TakeoutSaaS.Module.Messaging.Services;

/// <summary>
/// RabbitMQ 消费者实现。
/// </summary>
public sealed class RabbitMqMessageSubscriber(RabbitMqConnectionFactory connectionFactory, IOptionsMonitor<RabbitMqOptions> optionsMonitor, JsonMessageSerializer serializer, ILogger<RabbitMqMessageSubscriber> logger)
    : IMessageSubscriber
{
    private IConnection? _connection;
    private IChannel? _channel;
    private bool _disposed;

    /// <inheritdoc />
    public async Task SubscribeAsync<T>(string queue, string routingKey, Func<T, CancellationToken, Task<bool>> handler, CancellationToken cancellationToken = default)
    {
        // 1. 确保通道可用
        await EnsureChannelAsync(cancellationToken);
        var options = optionsMonitor.CurrentValue;

        var channel = _channel ?? throw new InvalidOperationException("RabbitMQ channel is not available.");

        // 2. 声明交换机、队列及绑定
        await channel.ExchangeDeclareAsync(options.Exchange, options.ExchangeType, durable: true, autoDelete: false, cancellationToken: cancellationToken);
        await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false, cancellationToken: cancellationToken);
        await channel.QueueBindAsync(queue, options.Exchange, routingKey, cancellationToken: cancellationToken);
        await channel.BasicQosAsync(0, options.PrefetchCount, global: false, cancellationToken: cancellationToken);

        // 3. 设置消费者回调
        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            var message = serializer.Deserialize<T>(ea.Body.ToArray());
            if (message == null)
            {
                await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken);
                return;
            }

            var success = false;
            try
            {
                success = await handler(message, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "处理消息失败：{RoutingKey}", ea.RoutingKey);
            }

            if (success)
            {
                await channel.BasicAckAsync(ea.DeliveryTag, multiple: false, cancellationToken);
            }
            else
            {
                await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false, cancellationToken);
            }
        };

        // 4. 开始消费
        await channel.BasicConsumeAsync(queue, autoAck: false, consumer, cancellationToken);
    }

    private async Task EnsureChannelAsync(CancellationToken cancellationToken)
    {
        if (_channel != null && _channel.IsOpen)
        {
            return;
        }

        _connection ??= await connectionFactory.CreateConnectionAsync(cancellationToken);
        _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
    }

    /// <inheritdoc />
    public async ValueTask DisposeAsync()
    {
        if (_disposed)
        {
            return;
        }

        _disposed = true;
        await Task.Run(() =>
        {
            _channel?.Dispose();
            _connection?.Dispose();
        }).ConfigureAwait(false);
    }
}
