package app

import (
	"errors"
	"et_rpc/pb"
	"fmt"
	"gitea.anxinyun.cn/container/common_utils/configLoad"
	"github.com/panjf2000/ants/v2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
	"log"
	"master/data_source"
	"master/node_manager"
	"net"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
)

type SendStatus struct {
	inProgressCount int32 // 正在处理的消息计数
	limitThreshold  int32 // 限流阈值
	receiving       int32 // 是否接收消息（1表示接收，0表示暂停）
}

// ETMaster 管理 Master 的核心逻辑
type ETMaster struct {
	nodeManager      *node_manager.NodeManager
	grpcServer       *grpc.Server
	masterRPCService *MasterRPCService

	dataSource      *data_source.KafkaDataSource
	aggDataHandlers sync.Map   // 聚合数据处理者
	rawDataHandlers sync.Map   // 原始数据处理者
	aggSendStatus   SendStatus // 聚合数据发送状态
	rawSendStatus   SendStatus // 原始数据发送状态

	errRawChan               chan []string
	errMessagesKafkaProducer *data_source.KafkaProducer // Kafka 生产者，用于发送失败的消息
}

// 创建 ETMaster 实例
func NewETMaster() *ETMaster {
	lb := node_manager.NewLoadBalancer(&node_manager.RoundRobinSelector{})
	nodeManager := node_manager.NewNodeManager(lb)

	grpcServer := grpc.NewServer()
	masterRPCService := NewMasterRPCService(nodeManager)
	pb.RegisterMasterServiceServer(grpcServer, masterRPCService)

	healthServer := health.NewServer()
	grpc_health_v1.RegisterHealthServer(grpcServer, healthServer)
	healthServer.SetServingStatus("MasterService", grpc_health_v1.HealthCheckResponse_SERVING)

	return &ETMaster{
		nodeManager:      nodeManager,
		grpcServer:       grpcServer,
		masterRPCService: masterRPCService,
	}
}

func (mm *ETMaster) StartRPCServer() {
	port := configLoad.LoadConfig().GetUint16("master.port")
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
	if err != nil {
		log.Panicf("启动 Master RPC 服务失败: %v", err)
	}
	defer func() {
		if err := listener.Close(); err != nil {
			log.Printf("关闭监听器失败: %v", err)
		}
	}()
	log.Printf("启动 Master RPC 服务成功，服务端口：%d", port)

	// 启动 gRPC 服务器
	if err := mm.grpcServer.Serve(listener); err != nil {
		log.Panicf("gRPC 服务器服务失败: %v", err)
	}
}

// 初始化 Kafka 数据源
func (mm *ETMaster) InitKafkaDataSource() {
	ds := data_source.NewKafkaDataSource() // 加载 kafka 相关的配置

	// 创建 kafka 生产者实例
	producer, err := data_source.NewKafkaProducer(ds.Brokers)
	if err != nil {
		log.Fatalf("创建 Kafka 生产者失败: %v", err)
	}
	mm.errMessagesKafkaProducer = producer

	// 设置 rawData 的处理者，每个分区一个处理者
	if ds.Master_kafkaConsumer_config.RawData != nil {
		topicCfg := ds.Topics["data_raw"]
		for partId := 0; partId < topicCfg.Partitions; partId++ {
			key := fmt.Sprintf("%s_%d", topicCfg.Topic, partId)
			dataHandler := data_source.NewRawDataHandler(key, topicCfg.Topic, partId)
			mm.rawDataHandlers.Store(key, dataHandler)
		}

		// 发送失败的消息存入 DLP_DATA_RAW 主题）
		dlpKey := "DLP_DATA_RAW"
		mm.rawDataHandlers.Store(dlpKey, data_source.NewRawDataHandler(dlpKey, dlpKey, 0))
	}

	// 设置 aggData 的处理者，每个分区一个处理者
	if ds.Master_kafkaConsumer_config.AggData != nil {
		topicCfg := ds.Topics["data_agg"]
		for partId := 0; partId < topicCfg.Partitions; partId++ {
			key := fmt.Sprintf("%s_%d", topicCfg.Topic, partId)
			dataHandler := data_source.NewAggDataHandler(key, topicCfg.Topic, partId)
			mm.aggDataHandlers.Store(key, dataHandler)
		}

		// 发送失败的消息存入 DLP_DATA_AGG 主题
		dlpKey := "DLP_DATA_AGG"
		mm.rawDataHandlers.Store(dlpKey, data_source.NewRawDataHandler(dlpKey, dlpKey, 0))
	}

	ds.RawDataHandlers = &mm.rawDataHandlers
	ds.AggDataHandlers = &mm.aggDataHandlers
	mm.dataSource = ds
}

// 等待节点注册
func (mm *ETMaster) WaitNodeRegister() {
	log.Println("==== 等待 Node 注册 ====")
	for mm.masterRPCService.nodeManager.NodesCount() == 0 {
		time.Sleep(time.Second * 10)
	}
}

// AggDataPublishing 发布聚合数据
func (mm *ETMaster) AggDataPublishing() {
	concurrency := configLoad.LoadConfig().GetInt32("performance.master.rpc.concurrency") // 并发请求数 50
	mm.initSendStatus(&mm.aggSendStatus, concurrency)
	go mm.monitorSendStatus(&mm.aggSendStatus, "aggSendStatus")
	mm.startDataPublishing(&mm.aggDataHandlers, "AggData", mm.sendAggData, &mm.aggSendStatus)
}

// RawDataPublishing 发布原始数据
func (mm *ETMaster) RawDataPublishing() {
	concurrency := configLoad.LoadConfig().GetInt32("performance.master.rpc.concurrency") // 并发请求数 50
	mm.initSendStatus(&mm.rawSendStatus, concurrency)
	go mm.monitorSendStatus(&mm.rawSendStatus, "rawSendStatus")
	mm.startDataPublishing(&mm.rawDataHandlers, "RawData", mm.sendRawData, &mm.rawSendStatus)
}

// initSendStatus 初始化发送状态
func (mm *ETMaster) initSendStatus(status *SendStatus, threshold int32) {
	status.limitThreshold = threshold
	atomic.StoreInt32(&status.receiving, 1)
}

// startDataPublishing 启动数据发布
func (mm *ETMaster) startDataPublishing(handlers *sync.Map, handlerType string, sendFunc func(string, []string) error, status *SendStatus) {
	// 创建一个 Goroutine 池，最大并发数为 500
	pool, err := ants.NewPool(500)
	if err != nil {
		log.Fatalf("创建 Goroutine 池失败: %v", err)
	}

	var wg sync.WaitGroup
	index := 0
	handlers.Range(func(key, value any) bool {
		handler := value.(data_source.IMessageHandler)
		dataChannel := handler.GetDataChannel()
		log.Printf("启动[%s-Publishing]协程，Handler%d，dataChannel[%p] 容量：%d", handlerType, index, dataChannel, cap(dataChannel))

		wg.Add(1)
		go func(idx int) {
			defer wg.Done()

			for {
				// 检查是否暂停接收
				if atomic.LoadInt32(&status.receiving) == 0 {
					log.Printf("%sHandler%d: 接收已暂停，等待未完成的消息处理", handlerType, idx)
					time.Sleep(100 * time.Millisecond)
					continue
				}

				select {
				case d, ok := <-dataChannel: // 检查 dataChannel 是否已关闭
					if !ok {
						log.Printf("%sHandler%d: dataChannel 已关闭，退出 Goroutine", handlerType, idx)
						return // 退出 Goroutine
					}

					data := d
					atomic.AddInt32(&status.inProgressCount, 1)
					log.Printf("[%s-Publishing] inProgressCount=%d. Handler%d 预备发送[%d]条数据，dataChannel[%p] 当前长度: %d/%d",
						handlerType, atomic.LoadInt32(&status.inProgressCount), idx, len(data.Messages), dataChannel, len(dataChannel), cap(dataChannel))

					// 使用 ants 提交任务
					poolErr := pool.Submit(func() {
						startTime := time.Now()
						defer atomic.AddInt32(&status.inProgressCount, -1) // 任务完成后减少计数

						if err := sendFunc(data.Id, data.Messages); err != nil {
							log.Printf("%sHandler%d: 发送数据失败: %v. 耗时：%v", handlerType, idx, err, time.Since(startTime))
							// 将失败数据发送到 Kafka（使用 Goroutine 池）
							_ = pool.Submit(func() {
								mm.errMessagesKafkaProducer.SendStringArrayMessage(fmt.Sprintf("DLP_%s", handlerType), data.Id, data.Messages)
							})
						} else {
							log.Printf("[%s-Publishing]协程，Handler%d 成功发送[%d]条数据。耗时：%v，dataChannel[%p] 当前长度: %d/%d",
								handlerType, idx, len(data.Messages), time.Since(startTime), dataChannel, len(dataChannel), cap(dataChannel))
						}
					})

					if poolErr != nil {
						log.Printf("%sHandler%d: 提交任务到 Goroutine 池失败: %v", handlerType, idx, poolErr)
						atomic.AddInt32(&status.inProgressCount, -1) // 提交失败时减少计数
					}

				default:
					// 如果 dataChannel 为空，则等待一段时间
					time.Sleep(10 * time.Millisecond)
				}
			}
		}(index)

		index++
		return true
	})

	wg.Wait()
	defer pool.Release() // 确保在函数结束时释放池
}

func (mm *ETMaster) sendRawData(thingId string, data []string) error {
	dataLog := fmt.Sprintf("thingId[%s]共[%d]条数据。", thingId, len(data))
	//log.Printf("[RawData-Publishing][sendRawData]1.开始处理。%s", dataLog)

	var nodeConn *node_manager.NodeConnection
	var err error
	retry := 0

	// 尝试获取 NodeConnection
	for retry < 3 {
		startTime := time.Now()
		nodeConn, err = mm.nodeManager.GetNodeConnection()
		duration := time.Since(startTime) // 计算获取连接的耗时
		log.Printf("[sendRawData]1.获取 NodeConnection 耗时: %v", duration)

		if err != nil {
			log.Printf("[sendRawData]1.获取 NodeConnection 失败，错误: %v", err)
			//m.kafkaDS.StopConsumers() // TODO 暂停消费 Kafka 消息
			//log.Println("============ Kafka 消费已暂停...")
			retry++
			time.Sleep(time.Duration(2<<retry) * time.Second) // 指数退避
			continue
		}

		// TODO 成功获取连接，恢复 Kafka 消费并退出循环
		//m.kafkaDS.ResumeConsumers()
		//log.Printf("[sendAggData] 成功获取 NodeConnection: %+v", nodeConn)
		break
	}

	if err != nil || nodeConn == nil {
		log.Printf("[sendRawData]1. 达到最大重试次数，无法获取健康节点连接，错误: %v", err)
		return err
	}

	// 记录调用 Node.ProcessData 的时间
	//defer LogProcessDataTimeCost(nodeConn.NArgs.Addr, "[]aggData", time.Now())
	// RPC 调用 Node.ProcessData，传递 []*pb.AggData
	resultChan := make(chan error, 1)
	log.Printf("[sendRawData]2.开始调用 RPC[Node.HandleRawData] %s", dataLog)
	callStartTime := time.Now()
	callErr := nodeConn.CallHandleIotaData(thingId, data)
	log.Printf("<--[sendRawData]3.RPC调用成功。耗时: %v，%s", time.Since(callStartTime), dataLog)
	resultChan <- callErr

	// 设置超时
	select {
	case callErr := <-resultChan:
		if callErr != nil {
			log.Printf("[sendRawData]4.RPC调用结束，错误: %+v，%s", callErr, dataLog)
			return callErr
		}
		//log.Printf("[sendRawData]4.RPC调用成功")
	case <-time.After(5 * time.Minute): // 设置超时
		log.Printf("[sendRawData]4.请求超过5分钟。%s", dataLog)
		return errors.New("请求超时5m")
	}

	return nil
}

func (mm *ETMaster) sendAggData(structId string, data []string) error {
	dataLog := fmt.Sprintf("structId[%s]共[%d]条数据。", structId, len(data))
	//log.Printf("[AggData-Publishing][sendAggData]1.开始处理。%s", dataLog)

	var nodeConn *node_manager.NodeConnection
	var err error
	retry := 0

	for retry < 3 {
		startTime := time.Now()
		nodeConn, err = mm.nodeManager.GetNodeConnection()
		duration := time.Since(startTime) // 计算获取连接的耗时
		log.Printf("[AggData-Publishing][sendAggData]2.获取 NodeConnection 耗时: %v", duration)

		if err != nil {
			log.Printf("[AggData-Publishing][sendAggData]2.1获取 NodeConnection 失败，错误: %v", err)
			//m.kafkaDS.StopConsumers() // TODO 暂停消费 Kafka 消息
			//log.Println("============ Kafka 消费已暂停...")
			retry++
			time.Sleep(time.Duration(2<<retry) * time.Second) // 指数退避
			continue
		}

		// TODO 成功获取连接，恢复 Kafka 消费并退出循环
		//m.kafkaDS.ResumeConsumers()
		//log.Printf("[sendAggData] 成功获取 NodeConnection: %+v", nodeConn)

		break
	}

	if err != nil || nodeConn == nil {
		log.Printf("[AggData-Publishing][sendAggData]2.2 达到最大重试次数，无法获取健康节点连接，错误: %v", err)
		return err
	}

	// 记录调用 Node.ProcessData 的时间
	//defer LogProcessDataTimeCost(nodeConn.NArgs.Addr, "[]aggData", time.Now())
	// RPC 调用 Node.ProcessData，传递 []*pb.AggData
	resultChan := make(chan error, 1)
	log.Printf("[AggData-Publishing][sendAggData]3.开始调用 RPC[Node.HandleAggData] %s", dataLog)
	callStartTime := time.Now()
	callErr := nodeConn.CallHandleAggData(structId, data)
	log.Printf("[AggData-Publishing][sendAggData]4.RPC调用耗时: %v，%s", time.Since(callStartTime), dataLog)
	resultChan <- callErr

	select {
	case callErr := <-resultChan:
		if callErr != nil {
			log.Printf("[AggData-Publishing][sendAggData]4.RPC调用结束，错误: %+v，%s", callErr, dataLog)
			return callErr
		}
		//log.Printf("[sendAggData]4.RPC调用成功")
	case <-time.After(5 * time.Minute): // 设置超时
		log.Printf("[AggData-Publishing][sendAggData]请求超过5分钟。%s", dataLog)
		return errors.New("请求超时5m")
	}

	return nil
}

// monitorSendStatus 监控发送状态
func (mm *ETMaster) monitorSendStatus(status *SendStatus, statusName string) {
	for {
		inProgressCount := atomic.LoadInt32(&status.inProgressCount)
		if inProgressCount > status.limitThreshold {
			atomic.StoreInt32(&status.receiving, 0)
			log.Printf("[%s] 未完成消息数量超过阈值，暂停接收新的消息。%+v\n", statusName, status)
		} else {
			atomic.StoreInt32(&status.receiving, 1)
		}
		time.Sleep(500 * time.Millisecond)
	}
}

// MonitorShutdown 监控退出信号
func (mm *ETMaster) MonitorShutdown() {
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	sig := <-sigChan
	log.Printf("************ 接收到信号: %s，正在关闭服务器...", sig)

	mm.closeDataHandlers(&mm.rawDataHandlers, "DLP_DATA_RAW")
	mm.closeDataHandlers(&mm.aggDataHandlers, "DLP_DATA_AGG")

	mm.errMessagesKafkaProducer.Close()
	mm.grpcServer.GracefulStop()
	log.Println("************ 服务器已成功关闭")
}

// closeDataHandlers 关闭数据处理器
func (mm *ETMaster) closeDataHandlers(handlers *sync.Map, dlpTopic string) {
	handlers.Range(func(key, value any) bool {
		handler := value.(data_source.IMessageHandler)
		ch := handler.GetDataChannel()
		close(ch)

		for data := range ch {
			mm.errMessagesKafkaProducer.SendStringArrayMessage(dlpTopic, data.Id, data.Messages)
		}
		return true
	})
}