FranklinWH-Toolbox/service/jobs/callFwhFile.go

package jobs

import (
	"context"
	"encoding/json"
	"fmt"
	"fwh-toolbox-server/cache"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/logger"
	"fwh-toolbox-server/repositories"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/redis/go-redis/v9"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

type TaskQueueMutex struct {
	running bool
	mu      sync.Mutex
}

type EmsLogType = string

var (
	EmsRunLog     = "EmsRunLog"     // 4-IBG运行日志
	EmsBattLog    = "EmsBattLog"    // 7-电池日志
	EmsMonitorLog = "EmsMonitorLog" // 9-设备系统资源监控日志
	Ems4gLog      = "Ems4gLog"      // 10-4G日志
	EmsAwsLog     = "EmsAwsLog"     // 11-AWS日志
	EmsMainLog    = "EmsMainLog"    // 12-主进程日志
)

// EMS日志类型map
var mapEmsLogType = map[string]int{
	"EmsRunLog":     4,
	"EmsBattLog":    7,
	"EmsMonitorLog": 9,
	"Ems4gLog":      10,
	"EmsAwsLog":     11,
	"EmsMainLog":    12,
}

var (
	callFwhFileCtx   = context.Background()
	streamNamePrefix = "fwhtoolbox:CallFwhFile:"
	groupName        = "task_queue"
	consumerID       = "worker-taskQueue"

	// 防止队列任务函数重复启动
	taskQueueMutex   = map[string]*TaskQueueMutex{}
	taskQueueMutexMu sync.RWMutex // 用于保护 map

	// 最大召唤次数
	maxCount = 3
)

// 召唤文件类型
var (
	FileTypeEmsLog      = "EmsLog"
	FileTypeFaultRecord = "FaultRecord"
)

type CallFileTaskStreamInfo struct {
	Ctx        context.Context
	StreamName string
	GroupName  string
	ID         string
}

type CallFileTaskInfo struct {
	MainTaskId primitive.ObjectID `bson:"mainTaskId" json:"mainTaskId"`
}

func CreateCallFwhFileTask(mainTaskId primitive.ObjectID, taskList []domain.BatchDownloadFileTaskInfo) error {
	// 筛选SN，不同SN可并行执行
	sortTaskList := map[string][]domain.BatchDownloadFileTaskInfo{}
	for _, task := range taskList {
		sortTaskList[task.Sn] = append(sortTaskList[task.Sn], task)
	}

	taskRecordList := map[string]domain.BatchDownloadFileTaskInfo{}
	for sn, payloadList := range sortTaskList {
		for _, payload := range payloadList {
			// 确保消费组存在（忽略已存在错误）
			_ = cache.XGroupCreate(callFwhFileCtx, streamNamePrefix+sn, groupName, "$")
			id, err := enqueueTask(sn, CallFileTaskInfo{
				MainTaskId: mainTaskId,
			}, payload)
			if err != nil {
				zap.L().Error("创建召唤FWH文件任务队列失败")
				return err
			}
			taskRecordList[id] = payload
		}
	}

	if err := repositories.CreateBatchDownloadCallFileTaskList(mainTaskId.Hex(), taskRecordList); err != nil {
		zap.L().Error("插入召唤FWH文件任务队列记录列表失败")
		return err
	}

	// 提取SN list
	snList := make([]string, 0, len(sortTaskList))
	for sn := range sortTaskList {
		snList = append(snList, sn)
	}

	// 启动任务队列
	for _, sn := range snList {
		go startTaskQueueWorker(sn)
		time.Sleep(time.Second) // 任务队列启动延时，避免过多请求同时发出，将任务错开运行
	}

	return nil
}

func callFwhFile(mainTaskId primitive.ObjectID, taskId string, info domain.BatchDownloadFileTaskInfo) {
	var reason string
	streamName := streamNamePrefix + info.Sn
	for count := 0; count < maxCount; count++ {
		// 请求召唤
		switch info.FileType {
		case FileTypeEmsLog: // 召唤EMS日志
			fmt.Printf("callEmsLog SN-%s Date-%s Type-%s\n", info.Sn, info.Date, info.FileSupType)
			if err := callEmsLog(info.Sn, info.Date, info.FileSupType); err != nil {
				ackTask(CallFileTaskStreamInfo{
					Ctx:        callFwhFileCtx,
					StreamName: streamName,
					GroupName:  groupName,
					ID:         taskId,
				}, false, fmt.Errorf("handle task CallEmsLog error - %s", err.Error()))
				return
			}

			ok, err := waitUntilDone(func() (bool, error) {
				// 查询召唤结果
				for count := 0; count < maxCount; count++ {
					fmt.Printf("getEmsLogUrl SN-%s Date-%s Type-%s\n", info.Sn, info.Date, info.FileSupType)
					recordInfo, err := getEmsLogUrl(info.Sn, info.Date, info.FileSupType)
					if err != nil {
						zap.L().Error("getEmsLogUrl读取EMS日志地址失败")
						continue
					}
					if recordInfo.Status == -1 {
						zap.L().Error("EMS日志召唤失败")
						return false, fmt.Errorf("EMS日志召唤失败")
					} else if recordInfo.Status == 1 {
						// 下载文件
						var downloadCount int = 0
						for downloadCount = 0; downloadCount < maxCount; downloadCount++ {
							dataBytes, err := repositories.QueryDownloadEmsLogFile(recordInfo.FileUrl)
							if err != nil {
								zap.L().Error("EMS日志文件下载失败", zap.String("err", err.Error()))
								continue
							}

							// 提取文件后缀
							baseStr := path.Base(recordInfo.FileUrl) // 取最后的文件名部分
							if baseStr == "." || baseStr == "/" {
								baseStr = ""
							}

							// 加载加州时区（Pacific Time）
							loc, err := time.LoadLocation("America/Los_Angeles")
							if err != nil {
								zap.L().Error("Ems日志LoadLocation设置失败", zap.String("callTime", recordInfo.CallTime), zap.String("err", err.Error()))
								return false, err
							}

							// 按指定时区解析时间，生效召唤时间的时间戳
							callTime, err := time.ParseInLocation(time.DateTime, recordInfo.CallTime, loc)
							if err != nil {
								zap.L().Error("Ems日志CallTime解析失败", zap.String("callTime", recordInfo.CallTime), zap.String("err", err.Error()))
								return false, err
							}

							// 保存数据到文件服务器
							objName, err := SaveBatchDownloadEmsLog(info.Sn, dataBytes, baseStr, info.FileSupType, info.Date, callTime)
							if err != nil {
								zap.L().Error("Ems日志文件保存失败", zap.String("devType", strconv.Itoa(info.DevType)), zap.String("sn", info.Sn), zap.String("err", err.Error()))
								return false, fmt.Errorf("Ems日志文件保存失败")
							}
							if err := repositories.InsertBatchDownloadPathEmsLogs(mainTaskId, info.Sn, info.Date, info.FileSupType, objName); err != nil {
								zap.L().Error("BatchDownload - EmsLog文件地址写入失败", zap.String("sn", info.Sn), zap.String("date", info.Date), zap.String("objName", objName), zap.String("err", err.Error()))
								return false, fmt.Errorf("BatchDownload - EmsLog文件地址写入失败")
							}

							break
						}

						// 超过最大尝试次数
						if downloadCount >= maxCount {
							return false, fmt.Errorf("EMS日志文件下载失败，超过最大尝试次数：%d", maxCount)
						}

						return true, nil
					}
					return false, nil
				}
				return false, fmt.Errorf("读取EMS日志地址请求失败")
			}, time.Minute, 30*time.Minute) // 每分钟查看，半小时超时
			if err != nil {
				reason = err.Error() // 保存失败原因
			}
			if ok {
				// 召唤成功
				repositories.UpdateBatchDownloadCallFileResult(mainTaskId.Hex(), taskId, 1, "")
				return
			} else {
				zap.L().Warn("Call FWH file失败, 再次尝试", zap.String("count", strconv.Itoa(count)))
			}
		case FileTypeFaultRecord: // 召唤故障录波
			timeRange := strings.Split(info.Date, ",")
			if len(timeRange) != 2 {
				ackTask(CallFileTaskStreamInfo{
					Ctx:        callFwhFileCtx,
					StreamName: streamName,
					GroupName:  groupName,
					ID:         taskId,
				}, false, fmt.Errorf("handle task CallFaultRecord timeRange error - timeRange: %s", info.Date))
				return
			}

			// 提取下载记录的indexList
			indexList := []int{}
			indexStrList := strings.Split(info.FileSupType, ",")
			for _, indexStr := range indexStrList {
				idx, err := strconv.Atoi(indexStr)
				if err != nil {
					ackTask(CallFileTaskStreamInfo{
						Ctx:        callFwhFileCtx,
						StreamName: streamName,
						GroupName:  groupName,
						ID:         taskId,
					}, false, fmt.Errorf("handle task CallFaultRecord indexStrList error - indexStrList: %s - err: %s", info.FileSupType, err.Error()))
					return
				}
				indexList = append(indexList, idx)
			}

			// 召唤故障录波数据
			err := callFaultRecord(info.DevType, info.Sn, timeRange[0], timeRange[1], indexList)
			if err != nil {
				ackTask(CallFileTaskStreamInfo{
					Ctx:        callFwhFileCtx,
					StreamName: streamName,
					GroupName:  groupName,
					ID:         taskId,
				}, false, fmt.Errorf("handle task CallFaultRecord error - err: %s", err.Error()))
				return
			}

			ok, err := waitUntilDone(func() (bool, error) {
				// 查询召唤结果
				for count := 0; count < maxCount; count++ {
					status, faultId, reasonStr, err := getFaultRecordUrl(info.DevType, info.Sn, timeRange[0], timeRange[1])
					if err != nil {
						zap.L().Error("getFaultRecordUrl读取故障录波地址失败", zap.String("err", err.Error()))
						continue
					}
					if status == -1 {
						zap.L().Error("故障录波召唤失败", zap.String("reason", reasonStr))
						return false, fmt.Errorf("故障录波召唤失败 - %s", reasonStr)
					} else if status == 2 {
						// 查询子记录
						subRecordList, err := repositories.QueryFaultRecordSubList(faultId)
						if err != nil {
							zap.L().Error("故障录波子记录查询失败", zap.String("err", err.Error()))
							return false, fmt.Errorf("故障录波子记录查询失败 - %w", err)
						}

						// 下载所有自己录中的文件
						for _, subRecord := range subRecordList {
							var downloadCount int = 0
							for downloadCount = 0; downloadCount < maxCount; downloadCount++ {
								_, dataBytes, err := repositories.QueryDownloadFaultRecordData(faultId, subRecord.Id)
								if err != nil {
									zap.L().Error("故障录波记录文件下载失败", zap.String("devType", strconv.Itoa(subRecord.DeviceType)), zap.String("sn", subRecord.GatewayId), zap.String("faultId", strconv.FormatInt(subRecord.FaultId, 10)), zap.String("id", strconv.FormatInt(subRecord.Id, 10)), zap.String("err", err.Error()))
									continue
								}

								// 保存数据到文件服务器
								objName, err := SaveBatchDownloadFaultRecord(subRecord.GatewayId, dataBytes, subRecord.FileName+".xls")
								if err != nil {
									zap.L().Error("故障录波记录文件保存失败", zap.String("devType", strconv.Itoa(subRecord.DeviceType)), zap.String("sn", subRecord.GatewayId), zap.String("faultId", strconv.FormatInt(subRecord.FaultId, 10)), zap.String("id", strconv.FormatInt(subRecord.Id, 10)), zap.String("err", err.Error()))
									return false, fmt.Errorf("故障录波记录文件保存失败")
								}
								if err := repositories.InsertBatchDownloadPathFaultRecord(mainTaskId, subRecord.GatewayId, subRecord.FileName, objName); err != nil {
									zap.L().Error("BatchDownload - FaultRecord文件地址写入失败", zap.String("sn", subRecord.GatewayId), zap.String("fileName", subRecord.FileName), zap.String("objName", objName), zap.String("err", err.Error()))
									return false, fmt.Errorf("BatchDownload - FaultRecord文件地址写入失败")
								}

								break
							}

							// 超过最大尝试次数
							if downloadCount >= maxCount {
								return false, fmt.Errorf("故障录波记录文件下载失败，超过最大尝试次数：%d", maxCount)
							}
						}

						return true, nil
					}
					return false, nil
				}
				return false, fmt.Errorf("读取故障录波地址请求失败")
			}, time.Minute, 30*time.Minute) // 每分钟查看，半小时超时
			if err != nil {
				reason = err.Error() // 保存失败原因
			}
			if ok {
				// 召唤成功
				repositories.UpdateBatchDownloadCallFileResult(mainTaskId.Hex(), taskId, 1, "")
				return
			} else {
				zap.L().Warn("Call FWH file失败, 再次尝试", zap.String("count", strconv.Itoa(count)))
			}

		default:
			logger.Std.Println("Unknown call file type:", info.FileType)
		}
	}

	// 召唤失败
	repositories.UpdateBatchDownloadCallFileResult(mainTaskId.Hex(), taskId, -1, reason)
}

/************** 任务轮询 **************/
func waitUntilDone(check func() (bool, error), interval time.Duration, timeout time.Duration) (bool, error) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	timeoutChan := time.After(timeout)

	for {
		select {
		case <-ticker.C:
			ok, err := check()
			if err != nil {
				zap.L().Debug("❌ 轮询任务失败", zap.String("err", err.Error()))
				return false, err
			} else if ok {
				zap.L().Debug("✅ 轮询任务完成")
				return true, nil
			} else {
				zap.L().Debug("⏳ 任务轮询中...")
			}

		case <-timeoutChan:
			zap.L().Debug("❌ 轮询超时")
			return false, fmt.Errorf("召唤结果轮询超时")
		}
	}
}

/************** 召唤文件任务队列 **************/
func startTaskQueueWorker(sn string) {
	taskQueueMutexMu.Lock()
	taskMutex, ok := taskQueueMutex[sn]
	if !ok || taskMutex == nil {
		taskMutex = &TaskQueueMutex{
			running: false,
			mu:      sync.Mutex{},
		}
		taskQueueMutex[sn] = taskMutex
	}
	taskQueueMutexMu.Unlock()

	taskMutex.mu.Lock()
	if taskMutex.running {
		taskMutex.mu.Unlock()
		zap.L().Info("任务已在运行，跳过执行")
		return
	}
	taskMutex.running = true
	taskMutex.mu.Unlock()

	streamName := streamNamePrefix + sn
	streamsMap := make(map[string]string)
	streamsMap[streamName] = ">"

	for {
		// 处理新任务
		res, err := cache.XReadGroup(callFwhFileCtx, groupName, consumerID, streamsMap, 1) // 等待任务
		if err != nil {
			if err == redis.Nil {
				// 未读取到新消息 - 退出
				logger.Std.Println("未读取到新消息: ", sn)
				taskMutex.mu.Lock()
				taskMutex.running = false
				taskMutex.mu.Unlock()
				cache.Delete(callFwhFileCtx, streamNamePrefix+sn)
				return
			}
			// Redis异常 - 退出
			logger.Std.Println("XReadGroup error:", err)
			taskMutex.mu.Lock()
			taskMutex.running = false
			taskMutex.mu.Unlock()
			return
		}

		for _, stream := range res {
			for _, message := range stream.Messages {
				logger.Std.Println("XReadGroup read message:", message)
				handleTaskQueueMsg(sn, message)
			}
		}
	}
}

// 处理单条消息
func handleTaskQueueMsg(sn string, message redis.XMessage) {
	id := message.ID
	rawPayload := message.Values["payload"].(string)
	rawTaskInfo := message.Values["taskInfo"].(string)
	streamName := streamNamePrefix + sn

	// 反序列化payload
	var payload domain.BatchDownloadFileTaskInfo
	if err := json.Unmarshal([]byte(rawPayload), &payload); err != nil {
		ackTask(CallFileTaskStreamInfo{
			Ctx:        callFwhFileCtx,
			StreamName: streamName,
			GroupName:  groupName,
			ID:         id,
		}, false, fmt.Errorf("unmarshal payload error - %s", err.Error()))
		return
	}

	// 反序列化taskInfo
	var taskInfo CallFileTaskInfo
	if err := json.Unmarshal([]byte(rawTaskInfo), &taskInfo); err != nil {
		ackTask(CallFileTaskStreamInfo{
			Ctx:        callFwhFileCtx,
			StreamName: streamName,
			GroupName:  groupName,
			ID:         id,
		}, false, fmt.Errorf("unmarshal taskInfo error - %s", err.Error()))
		return
	}

	logger.Std.Printf("Processing task %s: %+v\n", id, payload)

	// 异步执行召唤文件任务
	callFwhFile(taskInfo.MainTaskId, id, payload)

	// switch payload.FileType {
	// case FileTypeEmsLog: // 召唤EMS日志
	// 	if err := handleTaskSiteAnalysis(ctx, payload.Payload); err != nil {
	// 		ackTask(CallFileTaskStreamInfo{
	// 			Ctx:        callFwhFileCtx,
	// 			StreamName: streamName,
	// 			GroupName:  groupName,
	// 			ID:         id,
	// 		}, false, fmt.Errorf("handle task SiteAnalysis error - %s", err.Error()))
	// 		return
	// 	}
	// case FileTypeFaultRecord: // 召唤故障录波
	// 	if err := handleTaskShutdownStatistics(ctx, payload.Payload); err != nil {
	// 		ackTask(CallFileTaskStreamInfo{
	// 			Ctx:        callFwhFileCtx,
	// 			StreamName: streamName,
	// 			GroupName:  groupName,
	// 			ID:         id,
	// 		}, false, fmt.Errorf("handle task SiteAnalysis error - %s", err.Error()))
	// 		return
	// 	}
	// default:
	// 	logger.Std.Println("Unknown task type:", payload.FileType)
	// }

	// 任务完成
	ackTask(CallFileTaskStreamInfo{
		Ctx:        callFwhFileCtx,
		StreamName: streamName,
		GroupName:  groupName,
		ID:         id,
	}, true, nil)
}

func ackTask(streamInfo CallFileTaskStreamInfo, success bool, err error) {
	// 删除任务
	if err := cache.XAckDel(streamInfo.Ctx, streamInfo.StreamName, streamInfo.GroupName, streamInfo.ID); err != nil {
		logger.Std.Println("XAck error:", err)
	}

	if success {
		// TODO 确认任务完成

	} else {
		// TODO 任务失败
		logger.Std.Println("Task queue execution failed, error:", err)
	}

	if err != nil {
		zap.L().Error("taskQueue ackTask error.", zap.String("err", err.Error()), zap.String("StreamName", streamInfo.StreamName), zap.String("GroupName", streamInfo.GroupName), zap.String("ID", streamInfo.ID))
	}
}

func enqueueTask(sn string, taskInfo CallFileTaskInfo, payload domain.BatchDownloadFileTaskInfo) (string, error) {
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return "", err
	}

	taskInfoBytes, err := json.Marshal(taskInfo)
	if err != nil {
		return "", err
	}

	return cache.XAdd(callFwhFileCtx, streamNamePrefix+sn, map[string]interface{}{"payload": string(payloadBytes), "taskInfo": string(taskInfoBytes)})
}

/************* 文件召唤&查询&下载接口 *************/
// 召唤EMS日志
func callEmsLog(sn string, date string, logType string) error {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		zap.L().Error("logType不存在", zap.String("logType", logType))
		return fmt.Errorf("logType不存在，logType=%s", logType)
	}
	err := repositories.QueryCallEmsLog(sn, date, logTypeNum)
	if err != nil {
		zap.L().Error("QueryCallEmsLog请求失败", zap.String("err", err.Error()))
		return err
	}

	return nil
}

// 获取日志下载地址
func getEmsLogUrl(sn string, date string, logType string) (*domain.EmsLogRecordInfo, error) {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		zap.L().Error("logType不存在", zap.String("logType", logType))
		return nil, fmt.Errorf("logType不存在，logType=%s", logType)
	}
	recordInfo, err := repositories.QueryEmsLogCallRecord(sn, date, logTypeNum)
	if err != nil {
		zap.L().Error("GetEmsLogUrl请求失败", zap.String("err", err.Error()))
		return nil, err
	}

	return recordInfo, nil
}

// 召唤故障录波
func callFaultRecord(devType int, sn string, startTime string, endTime string, indexList []int) error {
	// 查询故障录波列表
	err := repositories.QueryCellFaultRecord(devType, sn, indexList)
	if err != nil {
		return err
	}

	return nil
}

// 获取日志下载地址
// return (status, faultId, faultReason, error)
// status: -1-失败 0-召唤中 2-成功
func getFaultRecordUrl(devType int, sn string, startTime string, endTime string) (int, int64, string, error) {
	// 查询召唤记录
	recordList, err := repositories.QueryFaultRecordCallRecordList()
	if err != nil {
		return 0, 0, "", err
	}

	for _, record := range recordList {
		// // 解析时间
		// t, err := time.Parse(time.DateTime, record.CellTime)
		// if err != nil {
		// 	return 0, nil, err
		// }

		// // 加 8 小时
		// newTime := t.Add(8 * time.Hour)

		// // 转回字符串
		// callTime := newTime.Format(time.DateTime)

		// // 判断记录是否当天召唤
		// currTime := time.Now().Format(time.DateOnly)
		// if !strings.HasPrefix(callTime, currTime) {
		// 	// 之后的非当天召唤的数据
		// 	break
		// }

		// 匹配符合条件的第一个数据
		if record.DeviceType == devType {
			var devSn string = ""
			switch devType {
			case 0: // aPower
				devSn = *record.ApowerId
			case 1: // aGate
				devSn = record.GatewayId
			case 2: // MPPT
				devSn = *record.ApowerId
			case 3: // aHub
				devSn = *record.PowerlinkId
			default:
				continue
			}

			if devSn == sn {
				// 匹配到的站点
				if *record.Status == -1 {
					return *record.Status, record.Id, *record.FaultReason, nil
				}
				return *record.Status, record.Id, "", nil
			}
		}
	}

	return 0, 0, "", fmt.Errorf("未找到召唤匹配记录(devType=%d,sn=%s,startTime=%s,endTime=%s)", devType, sn, startTime, endTime)
}