using System.Net;
using System.Text;
using System.Text.Json;
using Confluent.Kafka;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection.Extensions;
using TrackCenter.Application.Helpers;
using TrackCenter.Application.Services.Warning.Signals;
using JsonSerializer = System.Text.Json.JsonSerializer;

namespace TrackCenter.Application.Kafka;

public static class WarningCalcServiceExtension
{
    public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf)
    {
        // Kafka 已按集群隔离，主题/消费组不再拼 Environment 后缀
        // 预警信号批处理后台任务
        services.AddHostedService<WarningCalcBackgroundService>();

        // 预警信号
        RegisterConsumer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaGroup, new[] { GlobalConst.WarningKafkaTopic });
        RegisterProducer<WarningSignalKey, WarningSignal>(services, conf, GlobalConst.WarningKafkaTopic);

        // 预警信号DLX
        RegisterConsumer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaGroup, new[] { GlobalConst.WarningDLXKafkaTopic });
        RegisterProducer<int, WarningSignal>(services, conf, GlobalConst.WarningDLXKafkaTopic);
    }

    private static void RegisterConsumer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic)
    {
        services.AddJsonSerializer<TKey>();
        services.AddJsonSerializer<TValue>();
        services.AddSingleton<IConsumer<TKey, TValue>>(sp =>
        {
            DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
            var consumerConfig = new ConsumerConfig
            {
                ClientId = Dns.GetHostName(),
                GroupId = groupId,
                BootstrapServers = conf.GetValue<string>("Kafka:BootstrapServers"),
                AutoOffsetReset = AutoOffsetReset.Earliest,
                //偏移提交控制
                EnablePartitionEof = false,
                EnableAutoOffsetStore = false,
                EnableAutoCommit = false,
                //链接控制
                SessionTimeoutMs = conf.GetValue<int>("Kafka:Consumer:SessionTimeoutMs"),
                MaxPollIntervalMs = conf.GetValue<int>("Kafka:Consumer:MaxPollIntervalMs"),
                HeartbeatIntervalMs = conf.GetValue<int>("Kafka:Consumer:HeartbeatIntervalMs"),
                //吞吐量
                FetchMinBytes = conf.GetValue<int>("Kafka:Consumer:FetchMinBytes"),
                FetchMaxBytes = conf.GetValue<int>("Kafka:Consumer:FetchMaxBytes"),
                FetchWaitMaxMs = conf.GetValue<int>("Kafka:Consumer:FetchWaitMaxMs"),
                CancellationDelayMaxMs = 100,
                // Debug = "all",
                // LogConnectionClose = false,
                // LogQueue = true
            };
            ApplyOptionalSasl(consumerConfig, conf);

            var bb = new ConsumerBuilder<TKey, TValue>(consumerConfig)
                .SetKeyDeserializer(sp.GetRequiredService<JsonCoreSerializer<TKey>>())
                .SetValueDeserializer(sp.GetRequiredService<JsonCoreSerializer<TValue>>())
                .SetErrorHandler((_, error) =>
                {
                    if (error.IsFatal)
                    {
                        Log.Error($"Kafka致命错误${error}");
                        return;
                    }

                    // tips: 防抖以防止高频上报错误
                    reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () =>
                    {
                        if (error.Code == ErrorCode.Local_Transport)
                            Log.Error($"Kafka链接错误{error}");
                        else if (error.Code == ErrorCode.Local_AllBrokersDown)
                            Log.Error($"Kafka所有Broker都挂了{error}");
                    });
                    reportFuncWrapper.Debounce();
                });


            var c = bb.Build();
            c.Subscribe(subTopic);
            return c;
        });
    }

    private static void RegisterProducer<TKey, TValue>(IServiceCollection services, IConfiguration conf, string? defaultTopic = null)
    {
        services.AddJsonSerializer<TKey>();
        services.AddJsonSerializer<TValue>();
        services.AddSingleton<IProducer<TKey, TValue>>(sp =>
        {
            DebounceHelper.DebounceWrapper? reportFuncWrapper = null;
            var producerConfig = new ProducerConfig
            {
                ClientId = Dns.GetHostName(),
                BootstrapServers = conf.GetValue<string>("Kafka:BootstrapServers"),
                // 确保消息发送可靠
                Acks = Acks.All,
                EnableIdempotence = true,
                MessageSendMaxRetries = conf.GetValue<int>("Kafka:Producer:MessageSendMaxRetries"),
                RequestTimeoutMs = conf.GetValue<int>("Kafka:Producer:RequestTimeoutMs"),
                // 消息体大小
                CompressionType = conf.GetValue<CompressionType>("Kafka:Producer:CompressionType"),
                CompressionLevel = conf.GetValue<int>("Kafka:Producer:CompressionLevel"),
                // 消息发送速率控制(任意一条件先达到就发送)
                LingerMs = conf.GetValue<int>("Kafka:Producer:LingerMs"),
                BatchSize = conf.GetValue<int>("Kafka:Producer:BatchSize"),
                BatchNumMessages = conf.GetValue<int>("Kafka:Producer:BatchNumMessages"),
                // 发送缓冲内存限制
                QueueBufferingMaxMessages = conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxMessages"),
                QueueBufferingMaxKbytes = conf.GetValue<int>("Kafka:Producer:QueueBufferingMaxKbytes"),
                // 分区器 一致性hash/随机
                Partitioner = conf.GetValue<Partitioner>("Kafka:Producer:Partitioner"),
                // Debug = "all",
                // LogConnectionClose = false,
                // LogQueue = true
            };
            ApplyOptionalSasl(producerConfig, conf);

            var bb = new ProducerBuilder<TKey, TValue>(producerConfig)
                .SetKeySerializer(sp.GetRequiredService<JsonCoreSerializer<TKey>>())
                .SetValueSerializer(sp.GetRequiredService<JsonCoreSerializer<TValue>>())
                .SetErrorHandler((_, error) =>
                {
                    if (error.IsFatal)
                    {
                        Log.Error($"Kafka致命错误${error}");
                        return;
                    }

                    // tips: 防抖以防止高频上报错误
                    reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () =>
                    {
                        if (error.Code == ErrorCode.Local_Transport)
                            Log.Error($"Kafka链接错误{error}");
                        else if (error.Code == ErrorCode.Local_AllBrokersDown)
                            Log.Error($"Kafka所有Broker都挂了{error}");
                    });
                    reportFuncWrapper.Debounce();
                });

            return bb.Build();
        });
    }

    /// <summary>
    /// 本地连外口 kafka-ext（SASL）时配置；集群内 PLAINTEXT 不配 SaslUsername 则跳过。
    /// </summary>
    private static void ApplyOptionalSasl(ClientConfig cfg, IConfiguration conf)
    {
        var user = conf.GetValue<string>("Kafka:SaslUsername");
        if (string.IsNullOrWhiteSpace(user))
            return;

        cfg.SecurityProtocol = conf.GetValue("Kafka:SecurityProtocol", SecurityProtocol.SaslPlaintext);
        cfg.SaslMechanism = conf.GetValue("Kafka:SaslMechanism", SaslMechanism.Plain);
        cfg.SaslUsername = user;
        cfg.SaslPassword = conf.GetValue<string>("Kafka:SaslPassword");
    }

    private static void AddJsonSerializer<T>(this IServiceCollection services, Action<JsonSerializerOptions>? setup = null)
    {
        services.TryAddSingleton(_ =>
        {
            var o = new JsonSerializerOptions();
            setup?.Invoke(o);
            return new JsonCoreSerializer<T>(o);
        });
    }

    public static List<ConsumeResult<TKey, TValue>> ConsumeBatch<TKey, TValue>(this IConsumer<TKey, TValue> consumer, int batchSize, TimeSpan timeout)
    {
        using var cts = new CancellationTokenSource(timeout);
        var list = new List<ConsumeResult<TKey, TValue>>();
        while (list.Count < batchSize && !cts.IsCancellationRequested)
        {
            var consumeResult = consumer.Consume(100);
            if (consumeResult != null)
                list.Add(consumeResult);
        }

        return list;
    }
}

internal sealed class JsonCoreSerializer<T> : ISerializer<T>, IDeserializer<T>
{
    private readonly Encoding _encoding;
    private readonly JsonSerializerOptions _options;

    public JsonCoreSerializer(JsonSerializerOptions options)
    {
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _encoding = Encoding.UTF8;
    }

    public byte[] Serialize(T data, SerializationContext context)
    {
        if (data is null)
        {
            return null;
        }

        var json = JsonSerializer.Serialize(data, data.GetType(), _options);

        var result = _encoding.GetBytes(json);

        return result;
    }

    public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
    {
        if (isNull)
        {
            return default;
        }

#if NETSTANDARD2_0
            var json = _encoding.GetString(data.ToArray());
#else
        var json = _encoding.GetString(data);
#endif
        var result = JsonSerializer.Deserialize<T>(json, _options);

        return result;
    }
}