using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NearCardAttendance.Model;
using Newtonsoft.Json;
using Serilog.Context;
using Serilog;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using NearCardAttendance.Common.helper;
using Newtonsoft.Json.Linq;
using NearCardAttendance.Service.MessageQueue.Model;


namespace NearCardAttendance.TcpServer
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        private readonly ServiceConfig _configService;
        private readonly HttpHelper _httpHelper = default!;
        private int _messageCount = 0;

        public Worker(ILogger<Worker> logger, IOptions<ServiceConfig> _optConfigService,HttpHelper httpHelper)
        {
            _logger = logger;
            _configService = _optConfigService.Value;
            _httpHelper = httpHelper;
        }
        /**
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            var kafkaConsumer = CreateKafkaConsumer();
            kafkaConsumer.Subscribe("topics.storage.near_card_attendance");

            // process messages every 50 messages or every minute
            // 每隔1分钟或没50条消息就批量处理一次
            var tasks = new List<Task>();

            #region 每分钟触发一次
            var messageCounter = 0; // 消息计数器
            var timer = new System.Timers.Timer(TimeSpan.FromMinutes(1).TotalMilliseconds); // 定时器，每分钟触发一次
            timer.AutoReset = true;

            // 定时器事件处理程序，用于定时处理消息
            timer.Elapsed += async (sender, e) =>
            {
                // 如果计数器大于 0，则处理消息
                if (messageCounter > 0)
                {
                    await ProcessMessagesAsync(tasks);
                    messageCounter = 0; // 处理完成后重置计数器
                }
            };

            timer.Start(); // 启动定时器
            #endregion

            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    var consumeResult = kafkaConsumer.Consume(stoppingToken);
                    if (consumeResult != null)
                    {
                        //tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer));
                        // 处理消息
                        tasks.Add(ProcessMessageAsync(consumeResult, kafkaConsumer));
                        messageCounter++; // 增加消息计数器

                        // 如果消息计数达到 30 条，则处理消息并重置计数器
                        if (messageCounter >= 30)
                        {
                            await ProcessMessagesAsync(tasks);
                            messageCounter = 0; // 处理完成后重置计数器
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                _logger.LogWarning("Worker exit");
            }
            catch (Exception ex)
            {
                _logger.LogError($"An error occurred: {ex.Message}");
            }

            await Task.WhenAll(tasks);
        }

        // 异步处理消息的方法
        private async Task ProcessMessagesAsync(List<Task> tasks)
        {
            await Task.WhenAll(tasks); // 等待所有任务完成
            tasks.Clear(); // 清空任务列表
        }

        private async Task ProcessMessageAsync(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)
        {
            // 处理消息的逻辑
            if (consumeResult != null)
            {

                try
                {
                    var url = _configService.XinHuaLeYuUrl;
                    var data = new { };
                    var res = await _httpHelper.HttpToPostAsync(url, data);

                }
                catch (Exception ex)
                {
                    //kafkaConsumer.Commit(consumeResult);
                    _logger.LogError($"消费出错:{consumeResult.Message.Value},\n{ex.Message}\n{ex.StackTrace}");
                }
            }
        }
        */

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            var kafkaConsumer = CreateKafkaConsumer();
            kafkaConsumer.Subscribe("topics.storage.near_card_attendance");
            var tasks = new List<Task>();
            List<ConsumeResult<Ignore, string>> consumeBatchResult = new List<ConsumeResult<Ignore, string>>();
            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    var consumeResult = kafkaConsumer.Consume(stoppingToken);
                    if (consumeResult != null)
                    {
                        _messageCount++;
                        consumeBatchResult.Add(consumeResult);
                        //// 30条消息为一批
                        if (_messageCount % 30 == 0)
                        {
                            if (!await ProcessBatchMessageAsync(consumeBatchResult, kafkaConsumer))
                            {   // 返回结果错误暂停5分钟
                                await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
                            }
                        }

                    }
                }
            }
            catch (OperationCanceledException)
            {
                _logger.LogWarning("Worker exit");
            }
            catch (Exception ex)
            {
                _logger.LogError($"An error occurred: {ex.Message}");
            }
        }
        private async Task<bool> ProcessMessageAsync(ConsumeResult<Ignore, string> consumeResult, IConsumer<Ignore, string> kafkaConsumer)
        {
            try
            {
                var url = _configService.XinHuaLeYuUrl;
                var data = new { };
                var res = await _httpHelper.HttpToPostAsync(url, data);
                if (!string.IsNullOrEmpty(res))
                {
                    JObject resObj = (JObject)JsonConvert.DeserializeObject(res!)!;
                    if ((bool)resObj["success"]!)
                    {
                        kafkaConsumer.Commit(consumeResult);
                        return true;
                    }
                }
                
            }
            catch (Exception ex)
            {
                //kafkaConsumer.Commit(consumeResult);
                _logger.LogError($"消费出错:{consumeResult.Message.Value},\n{ex.Message}\n{ex.StackTrace}");
               
            }
            return false;

        }

        private async Task<bool> ProcessBatchMessageAsync(List<ConsumeResult<Ignore, string>> consumeBatchResult, IConsumer<Ignore, string> kafkaConsumer)
        {
            try
            {
                
                var url =$"{_configService.XinHuaLeYuUrl}/user/electronicCardAttendance/receiveTbAttendanceRecordException";
                var list = new List<object>();
                //consumeBatchResult.ForEach(x => {
                //    JObject msg = (JObject)JsonConvert.DeserializeObject(x.Message.Value)!;
                //    list.Add(new
                //    {
                //        attendanceStatus = int.TryParse(msg["data"]!["attendanceStatus"]!.ToString(), out int status) ? status : 0,
                //        attendanceTime = msg["data"]!["attendanceTime"]!.ToString(),
                //        imei = msg["data"]!["imei"]!.ToString()
                //    }) ;
                //});

                consumeBatchResult.ForEach(x => {
                    EventData msg = JsonConvert.DeserializeObject<EventData>(x.Message.Value)!;
                    JObject content = (JObject)JsonConvert.DeserializeObject(msg.Content)!;
                    list.Add(new
                    {
                        attendanceStatus = int.TryParse(content!["attendanceStatus"]!.ToString(), out int status) ? status : 0,
                        attendanceTime = content!["attendanceTime"]!.ToString(),
                        imei = content!["imei"]!.ToString()
                    });
                });


                var data = new {
                    data= list
                };
                var res = await _httpHelper.HttpToPostAsync(url, data);
                if (!string.IsNullOrEmpty(res))
                {
                    JObject resObj = (JObject)JsonConvert.DeserializeObject(res!)!;
                    if ((bool)resObj["success"]!)
                    {
                        consumeBatchResult.ForEach(x =>
                        {
                            kafkaConsumer.Commit(x);
                        });
                        return true;
                    }
                }

            }
            catch (Exception ex)
            {
                _logger.LogError($"处理消息出错 \n{ex.Message}\n{ex.StackTrace}");

            }
            return false;

        }


        /// <summary>
        /// 创建消费者
        /// </summary>
        /// <returns></returns>
        private IConsumer<Ignore, string> CreateKafkaConsumer()
        {
            
            var consumerConfig = new ConsumerConfig
            {
                GroupId = "near_card_attendance",
                BootstrapServers = _configService.KafkaServerAddress,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false, // 关闭自动提交偏移量
                CancellationDelayMaxMs = 1//set CancellationDelayMaxMs
            };

            return new ConsumerBuilder<Ignore, string>(consumerConfig)
                     .SetErrorHandler((_, e) =>
                     {

                         //Console.WriteLine($"消费者创建出错,代码:{e.Code} |原因: {e.Reason}");
                         _logger.LogInformation($"消费者创建出错,代码:{e.Code} |原因: {e.Reason}");
                     })
                     .SetPartitionsAssignedHandler((c, partitions) =>
                     {
                         //// 在这里手动指定要消费的分区
                         //var partitionsToConsume = new List<TopicPartitionOffset>
                         //{
                         //   new TopicPartitionOffset("topics.storage.near_card_attendance", partitionIndex, Offset.Unset)
                         //};

                         ////c.Assign(partitionsToConsume);
                         //Console.WriteLine($"Assigned partitions: {string.Join(", ", partitionsToConsume)}");
                         //return partitionsToConsume;
                     })
                     .SetPartitionsRevokedHandler((c, partitions) =>
                     {

                     })
                     .Build();
        }
    }
}