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

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

        public async Task DeleteTopicAsync(string topic)
        {
            await Instance.DeleteTopicsAsync(new[] { topic });
        }

        public async Task<List<string>> ListTopicsAsync()
        {
            var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
            return new List<string>(metadata.Topics.Select(t => t.Topic));
        }

        public async Task<bool> TopicExistsAsync(string topic)
        {
            var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
            return metadata.Topics.Any(t => t.Topic == topic);
        }

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