using Confluent.Kafka;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Consumer;
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;

namespace JiShe.CollectBus.Kafka
{
    public static class KafkaSubcribesExtensions
    {
        /// <summary>
        /// 添加Kafka订阅
        /// </summary>
        /// <param name="app"></param>
        /// <param name="assembly"></param>
        public static void UseKafkaSubscribers(this IApplicationBuilder app, Assembly assembly)
        {
            var subscribeTypes = assembly.GetTypes()
                .Where(t => typeof(IKafkaSubscribe).IsAssignableFrom(t))
                .ToList();

            if (subscribeTypes.Count == 0) return;

            var provider = app.ApplicationServices;
            var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();

            lifetime.ApplicationStarted.Register(() =>
            {
                foreach (var subscribeType in subscribeTypes)
                {
                    var subscribes = provider.GetServices(subscribeType).ToList();
                    subscribes.ForEach(subscribe => { 
                    
                        if(subscribe is IKafkaSubscribe)
                        {
                            BuildKafkaSubscriber(subscribe, provider);
                        }
                    });
                }
            });
        }

        /// <summary>
        /// 构建Kafka订阅
        /// </summary>
        /// <param name="subscribe"></param>
        /// <param name="provider"></param>
        private static void BuildKafkaSubscriber(object subscribe, IServiceProvider provider)
        {
            var methods = subscribe.GetType().GetMethods();
            foreach (var method in methods)
            {
                var attr = method.GetCustomAttribute<KafkaSubscribeAttribute>();
                if (attr == null) continue;

                // 启动后台消费线程
                Task.Run(() => StartConsumerAsync(provider, attr, method, subscribe));
            }
        }

        /// <summary>
        /// 启动后台消费线程
        /// </summary>
        /// <param name="config"></param>
        /// <param name="attr"></param>
        /// <param name="method"></param>
        /// <param name="consumerInstance"></param>
        /// <returns></returns>
        private static async Task StartConsumerAsync(IServiceProvider provider, KafkaSubscribeAttribute attr,MethodInfo method, object subscribe)
        {
            var consumerService = provider.GetRequiredService<IConsumerService<Ignore, string>>();
            await consumerService.SubscribeAsync(attr.Topics, async (key, message) =>
            {
                try
                {
                    if (string.IsNullOrEmpty(message))
                        await Task.CompletedTask ;

                    // 处理消息
                    await ProcessMessageAsync(message, method, subscribe);
                }
                catch (ConsumeException ex)
                {
                    // 处理消费错误
                    throw;
                }
            });
        }

        private static async Task ProcessMessageAsync(string message, MethodInfo method, object subscribe)
        {
            var parameters = method.GetParameters();
            if (parameters.Length != 1) return;

            var paramType = parameters[0].ParameterType;
            var messageObj = paramType == typeof(string)
                ? message
                : JsonConvert.DeserializeObject(message, paramType);

            if (method.ReturnType == typeof(Task))
            {
                await (Task)method.Invoke(subscribe, new[] { messageObj })!;
            }
            else
            {
                method.Invoke(subscribe, new[] { messageObj });
            }
        }

    }
}