using Confluent.Kafka;
using Confluent.Kafka.Admin;
using JiShe.CollectBus.Kafka.Internal;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Volo.Abp.DependencyInjection;

namespace JiShe.CollectBus.Kafka.AdminClient;

public class AdminClientService : IAdminClientService, IDisposable, ISingletonDependency
{
    private readonly ILogger<AdminClientService> _logger;
    private readonly KafkaOptionConfig _kafkaOptionConfig;
    /// <summary>
    /// Initializes a new instance of the <see cref="AdminClientService" /> class.
    /// </summary>
    /// <param name="configuration"></param>
    /// <param name="logger"></param>
    public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig)
    {
        _logger = logger;
        _kafkaOptionConfig = kafkaOptionConfig.Value;
        Instance = GetInstance();
    }

    /// <summary>
    ///     Gets or sets the instance.
    /// </summary>
    /// <value>
    ///     The instance.
    /// </value>
    public IAdminClient Instance { get; set; }

    /// <summary>
    /// 创建Kafka主题
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="numPartitions"></param>
    /// <param name="replicationFactor"></param>
    /// <returns></returns>
    public async Task CreateTopicAsync(string topic, int numPartitions, short replicationFactor)
    {
        try
        {
            if (await CheckTopicAsync(topic)) return;


            await Instance.CreateTopicsAsync(new[]
            {
                new TopicSpecification
                {
                    Name = topic,
                    NumPartitions = numPartitions,
                    ReplicationFactor = replicationFactor
                }
            });
        }
        catch (CreateTopicsException e)
        {
            if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists) throw;
        }
    }

    /// <summary>
    ///     删除Kafka主题
    /// </summary>
    /// <param name="topic"></param>
    /// <returns></returns>
    public async Task DeleteTopicAsync(string topic)
    {
        await Instance.DeleteTopicsAsync(new[] { topic });
    }

    /// <summary>
    ///     获取Kafka主题列表
    /// </summary>
    /// <returns></returns>
    public async Task<List<string>> ListTopicsAsync()
    {
        var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
        return new List<string>(metadata.Topics.Select(t => t.Topic));
    }

    /// <summary>
    ///     判断Kafka主题是否存在
    /// </summary>
    /// <param name="topic"></param>
    /// <returns></returns>
    public async Task<bool> TopicExistsAsync(string topic)
    {
        var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
        return metadata.Topics.Any(t => t.Topic == topic);
    }

    /// <summary>
    ///     检测分区是否存在
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="partitions"></param>
    /// <returns></returns>
    public Dictionary<int, bool> CheckPartitionsExists(string topic, int[] partitions)
    {
        var result = new Dictionary<int, bool>();
        var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
        if (metadata.Topics.Count == 0)
            return partitions.ToDictionary(p => p, p => false);
        var existingPartitions = metadata.Topics[0].Partitions.Select(p => p.PartitionId).ToHashSet();
        return partitions.ToDictionary(p => p, p => existingPartitions.Contains(p));
    }

    /// <summary>
    ///     检测分区是否存在
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="targetPartition"></param>
    /// <returns></returns>
    public bool CheckPartitionsExist(string topic, int targetPartition)
    {
        var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
        if (metadata.Topics.Count == 0)
            return false;
        var partitions = metadata.Topics[0].Partitions;
        return partitions.Any(p => p.PartitionId == targetPartition);
    }

    /// <summary>
    ///     获取主题的分区数量
    /// </summary>
    /// <param name="topic"></param>
    /// <returns></returns>
    public int GetTopicPartitionsNum(string topic)
    {
        var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
        if (metadata.Topics.Count == 0)
            return 0;
        return metadata.Topics[0].Partitions.Count;
    }

    public void Dispose()
    {
        Instance?.Dispose();
    }

    /// <summary>
    ///     Gets the instance.
    /// </summary>
    /// <returns></returns>
    public IAdminClient GetInstance()
    {
        var adminClientConfig = new AdminClientConfig
        {
            BootstrapServers = _kafkaOptionConfig.BootstrapServers
        };
        if (_kafkaOptionConfig.EnableAuthorization)
        {
            adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
            adminClientConfig.SaslMechanism = SaslMechanism.Plain;
            adminClientConfig.SaslUsername = _kafkaOptionConfig.SaslUserName;
            adminClientConfig.SaslPassword = _kafkaOptionConfig.SaslUserName;
        }
        return new AdminClientBuilder(adminClientConfig).Build();
    }

    /// <summary>
    ///     Checks the topic asynchronous.
    /// </summary>
    /// <param name="topic">The topic.</param>
    /// <returns></returns>
    public async Task<bool> CheckTopicAsync(string topic)
    {
        var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
        return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
    }

    /// <summary>
    ///     判断Kafka主题是否存在
    /// </summary>
    /// <param name="topic">主题名称</param>
    /// <param name="numPartitions">副本数量，不能高于Brokers数量</param>
    /// <returns></returns>
    public async Task<bool> CheckTopicAsync(string topic, int numPartitions)
    {
        var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
        if (numPartitions > metadata.Brokers.Count)
            throw new Exception($"{nameof(CheckTopicAsync)} 主题检查时，副本数量大于了节点数量。");

        return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
    }
}