using Confluent.Kafka;
using Microsoft.Extensions.Configuration;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Logging;
using Volo.Abp.DependencyInjection;

namespace JiShe.CollectBus.Kafka.AdminClient
{
    public class AdminClientService : IAdminClientService, IDisposable,ISingletonDependency
    {

        private readonly ILogger<AdminClientService> _logger;

        /// <summary>
        /// Initializes a new instance of the <see cref="AdminClientService"/> class.
        /// </summary>
        /// <param name="configuration">The configuration.</param>
        /// <param name="logger">The logger.</param>
        public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger)
        {
            _logger = logger;
            GetInstance(configuration);
        }

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

        /// <summary>
        /// Gets the instance.
        /// </summary>
        /// <param name="configuration">The configuration.</param>
        /// <returns></returns>
        public IAdminClient GetInstance(IConfiguration configuration)
        {
            ArgumentNullException.ThrowIfNullOrWhiteSpace(configuration["Kafka:EnableAuthorization"]);
            var enableAuthorization = bool.Parse(configuration["Kafka:EnableAuthorization"]!);
            var adminClientConfig = new AdminClientConfig()
            {
                BootstrapServers = configuration["Kafka:BootstrapServers"],
            };
            if (enableAuthorization)
            {
                adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
                adminClientConfig.SaslMechanism = SaslMechanism.Plain;
                adminClientConfig.SaslUsername = configuration["Kafka:SaslUserName"];
                adminClientConfig.SaslPassword = configuration["Kafka:SaslPassword"];
            }
            Instance = new AdminClientBuilder(adminClientConfig).Build();
            return Instance;
        }

        /// <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));
        }

        //// <summary>
        /// 创建Kafka主题
        /// </summary>
        /// <param name="topic">主题名称</param>
        /// <param name="numPartitions">主题分区数量</param>
        /// <param name="replicationFactor">副本数量，不能高于Brokers数量</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();
        }
    }
}