FranklinWH-Toolbox/service/task/queue/fwhQueue/fwhTaskCallFile.go

package fwhQueue

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

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

const maxCallRetry = 3

// handleCallFile 是文件下载子任务的 Asynq handler
// Asynq 本身会做重试（MaxRetry=3），所以 handler 内部不再做外层重试循环
// 返回 error 会触发 Asynq 重试，返回 nil 表示成功
func handleCallFile(ctx context.Context, task *asynq.Task) error {
	var payload callFilePayload
	if err := json.Unmarshal(task.Payload(), &payload); err != nil {
		// payload 解析失败不重试
		reportCallFileResult(ctx, payload.MainTaskId, fmt.Sprintf("unmarshal callFile payload: %s", err.Error()))
		return nil
	}

	info := payload.TaskInfo
	mainTaskId := payload.MainTaskId

	var err error
	switch info.FileType {
	case jobs.FileTypeEmsLog:
		err = handleEmsLogDownload(ctx, mainTaskId, info)
	case jobs.FileTypeFaultRecord:
		err = handleFaultRecordDownload(ctx, mainTaskId, info)
	default:
		err = fmt.Errorf("unknown file type: %s", info.FileType)
	}

	if err != nil {
		// 判断是否还有重试机会
		retryCount, _ := asynq.GetRetryCount(ctx)
		maxRetry, _ := asynq.GetMaxRetry(ctx)
		if retryCount >= maxRetry-1 {
			// 最后一次重试仍失败，记录失败原因并完成计数
			reportCallFileResult(ctx, mainTaskId, err.Error())
			return nil // 返回 nil 不再重试
		}
		return err // 返回 error 触发 Asynq 重试
	}

	// 成功：递减计数器
	reportCallFileResult(ctx, mainTaskId, "")
	return nil
}

// reportCallFileResult 报告子任务完成结果
// reason 为空表示成功，非空表示失败
func reportCallFileResult(ctx context.Context, mainTaskId string, reason string) {
	if mainTaskId == "" {
		return
	}

	counterKey := callFileCounterPrefix + mainTaskId
	failuresKey := callFileFailuresPrefix + mainTaskId

	// 递减计数器
	cache.Decr(ctx, counterKey)

	// 如果失败，记录失败原因
	if reason != "" {
		cache.LPush(ctx, failuresKey, reason)
		// 设置 failures key 过期时间
		cache.Expire(ctx, failuresKey, 2*time.Hour)
	}
}

// ==================== EmsLog 下载 ====================

func handleEmsLogDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) error {
	// 1. 召唤 EmsLog
	if err := callEmsLog(info.Sn, info.Date, info.FileSupType); err != nil {
		return fmt.Errorf("[EmsLog] 召唤失败(sn=%s, date=%s, type=%s): %w", info.Sn, info.Date, info.FileSupType, err)
	}

	// 2. 轮询等待召唤结果
	recordInfo, err := pollEmsLogResult(ctx, info.Sn, info.Date, info.FileSupType)
	if err != nil {
		return err
	}

	// 3. 下载并保存文件
	mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
	return downloadAndSaveEmsLog(ctx, mainObjId, info, recordInfo)
}

func pollEmsLogResult(ctx context.Context, sn, date, fileSupType string) (*domain.EmsLogRecordInfo, error) {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	timeout := time.After(30 * time.Minute)

	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-timeout:
			return nil, fmt.Errorf("[EmsLog] 轮询超时(sn=%s, date=%s)", sn, date)
		case <-ticker.C:
			for attempt := 0; attempt < maxCallRetry; attempt++ {
				recordInfo, err := getEmsLogUrl(sn, date, fileSupType)
				if err != nil {
					continue
				}
				if recordInfo.Status == -1 {
					return nil, fmt.Errorf("[EmsLog] 召唤失败(sn=%s)", sn)
				}
				if recordInfo.Status == 1 {
					return recordInfo, nil
				}
				// status == 0: 还在处理中
				break
			}
		}
	}
}

func downloadAndSaveEmsLog(ctx context.Context, mainTaskId primitive.ObjectID, info domain.BatchDownloadFileTaskInfo, recordInfo *domain.EmsLogRecordInfo) error {
	var dataBytes []byte
	var err error
	for attempt := 0; attempt < maxCallRetry; attempt++ {
		dataBytes, err = repositories.QueryDownloadEmsLogFile(recordInfo.FileUrl)
		if err == nil {
			break
		}
		zap.L().Warn("[EmsLog] 文件下载重试", zap.String("sn", info.Sn), zap.Int("attempt", attempt))
	}
	if err != nil {
		return fmt.Errorf("[EmsLog] 文件下载失败(sn=%s): %w", info.Sn, err)
	}

	// 提取文件名
	baseStr := path.Base(recordInfo.FileUrl)
	if baseStr == "." || baseStr == "/" {
		baseStr = ""
	}

	// 解析召唤时间
	loc, err := time.LoadLocation("America/Los_Angeles")
	if err != nil {
		return fmt.Errorf("[EmsLog] 时区加载失败: %w", err)
	}
	callTime, err := time.ParseInLocation(time.DateTime, recordInfo.CallTime, loc)
	if err != nil {
		return fmt.Errorf("[EmsLog] CallTime解析失败: %w", err)
	}

	// 保存到 MinIO
	objName, err := jobs.SaveBatchDownloadEmsLog(info.Sn, dataBytes, baseStr, info.FileSupType, info.Date, callTime)
	if err != nil {
		return fmt.Errorf("[EmsLog] 文件保存失败(sn=%s): %w", info.Sn, err)
	}

	// 写入路径记录
	if err := repositories.InsertBatchDownloadPathEmsLogs(mainTaskId, info.Sn, info.Date, info.FileSupType, objName); err != nil {
		return fmt.Errorf("[EmsLog] 路径写入失败(sn=%s): %w", info.Sn, err)
	}

	return nil
}

// ==================== FaultRecord 下载 ====================

func handleFaultRecordDownload(ctx context.Context, mainTaskId string, info domain.BatchDownloadFileTaskInfo) error {
	// 1. 解析参数
	timeRange := strings.Split(info.Date, ",")
	if len(timeRange) != 2 {
		return fmt.Errorf("[FaultRecord] timeRange 格式错误: %s", info.Date)
	}

	// 2. 查询故障录波列表，判断是否存在满足时间范围的数据
	existingRecords, err := repositories.QueryFaultRecordList(info.Sn, timeRange[0], timeRange[1])
	if err != nil {
		zap.L().Error("[FaultRecord] 查询已有录波列表失败", zap.String("sn", info.Sn), zap.String("err", err.Error()))
		return err
	} else {
		// 判断列表中是否存在 earliestFaultTime 在时间范围内的记录
		faultID, err := checkFaultRecordInTimeRange(existingRecords, info.DevType, info.Sn, timeRange[0], timeRange[1])
		if err != nil {
			zap.L().Error("[FaultRecord] 判断时间范围失败", zap.String("sn", info.Sn), zap.String("err", err.Error()))
			return err
		}
		if faultID >= 0 {
			// 故障录波文件存在，直接下载
			// 下载所有子记录文件
			mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
			return downloadAndSaveFaultRecords(ctx, mainObjId, faultID)
		}
		zap.L().Info("[FaultRecord] 无满足时间范围的录波数据，需要召唤",
			zap.String("sn", info.Sn),
			zap.String("startTime", timeRange[0]),
			zap.String("endTime", timeRange[1]),
		)
	}

	indexList, err := parseIndexList(info.FileSupType)
	if err != nil {
		return fmt.Errorf("[FaultRecord] indexList 解析错误: %w", err)
	}

	// 3. 召唤故障录波
	if err := callFaultRecord(info.DevType, info.Sn, timeRange[0], timeRange[1], indexList); err != nil {
		return fmt.Errorf("[FaultRecord] 召唤失败(sn=%s): %w", info.Sn, err)
	}

	// 4. 轮询等待召唤结果
	faultId, err := pollFaultRecordResult(ctx, info.DevType, info.Sn, timeRange[0], timeRange[1])
	if err != nil {
		return err
	}

	// 5. 下载所有子记录文件
	mainObjId, _ := primitive.ObjectIDFromHex(mainTaskId)
	return downloadAndSaveFaultRecords(ctx, mainObjId, faultId)
}

// checkFaultRecordInTimeRange 检查故障录波列表中是否存在 earliestFaultTime 满足时间范围的记录，返回FaultID和错误信息，不在范围内FaultID = -1
func checkFaultRecordInTimeRange(records []domain.FaultRecordInfo, devType int, sn string, startTime string, endTime string) (int64, error) {
	start, err := time.Parse(time.DateTime, startTime)
	if err != nil {
		zap.L().Error("[checkFaultRecordInTimeRange] 解析startTime失败",
			zap.String("startTime", startTime),
			zap.String("err", err.Error()),
		)
		return -1, err // 解析失败，保守处理：继续召唤
	}
	end, err := time.Parse(time.DateTime, endTime)
	if err != nil {
		zap.L().Error("[checkFaultRecordInTimeRange] 解析endTime失败",
			zap.String("endTime", endTime),
			zap.String("err", err.Error()),
		)
		return -1, err
	}

	for _, record := range records {
		// 匹配设备类型和 SN
		if record.DeviceType != devType {
			continue
		}
		var recordSn string
		switch devType {
		case 0: // aPower
			if record.ApowerId != nil {
				recordSn = *record.ApowerId
			}
		case 1: // aGate
			recordSn = record.GatewayId
		case 2: // MPPT
			if record.ApowerId != nil {
				recordSn = *record.ApowerId
			}
		case 3: // aHub
			if record.PowerlinkId != nil {
				recordSn = *record.PowerlinkId
			}
		}
		if recordSn != sn {
			continue
		}

		// 解析 earliestFaultTime 并判断是否在时间范围内
		faultTime, err := time.Parse(time.DateTime, record.EarliestFaultTime)
		if err != nil {
			// 尝试时间戳格式
			continue
		}

		inRange := (faultTime.Equal(start) || faultTime.After(start)) &&
			(faultTime.Equal(end) || faultTime.Before(end))
		if inRange {
			return record.Id, nil
		}
	}

	return -1, nil
}

func pollFaultRecordResult(ctx context.Context, devType int, sn, startTime, endTime string) (int64, error) {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	timeout := time.After(30 * time.Minute)

	for {
		select {
		case <-ctx.Done():
			return 0, ctx.Err()
		case <-timeout:
			return 0, fmt.Errorf("[FaultRecord] 轮询超时(sn=%s)", sn)
		case <-ticker.C:
			for attempt := 0; attempt < maxCallRetry; attempt++ {
				status, faultId, reason, err := getFaultRecordUrl(devType, sn, startTime, endTime)
				if err != nil {
					continue
				}
				if status == -1 {
					return 0, fmt.Errorf("[FaultRecord] 召唤失败(sn=%s): %s", sn, reason)
				}
				if status == 2 {
					return faultId, nil
				}
				// status == 0: 还在处理中
				break
			}
		}
	}
}

func downloadAndSaveFaultRecords(ctx context.Context, mainTaskId primitive.ObjectID, faultId int64) error {
	subRecordList, err := repositories.QueryFaultRecordSubList(faultId)
	if err != nil {
		return fmt.Errorf("[FaultRecord] 子记录查询失败: %w", err)
	}

	for _, subRecord := range subRecordList {
		var dataBytes []byte
		var downloadErr error
		for attempt := 0; attempt < maxCallRetry; attempt++ {
			_, dataBytes, downloadErr = repositories.QueryDownloadFaultRecordData(faultId, subRecord.Id)
			if downloadErr == nil {
				break
			}
			zap.L().Warn("[FaultRecord] 文件下载重试",
				zap.String("sn", subRecord.GatewayId),
				zap.Int64("subId", subRecord.Id),
				zap.Int("attempt", attempt),
			)
		}
		if downloadErr != nil {
			return fmt.Errorf("[FaultRecord] 文件下载失败(sn=%s, id=%d): %w", subRecord.GatewayId, subRecord.Id, downloadErr)
		}

		// 保存到 MinIO
		objName, err := jobs.SaveBatchDownloadFaultRecord(subRecord.GatewayId, dataBytes, subRecord.FileName+".xls")
		if err != nil {
			return fmt.Errorf("[FaultRecord] 文件保存失败(sn=%s): %w", subRecord.GatewayId, err)
		}

		// 写入路径记录
		if err := repositories.InsertBatchDownloadPathFaultRecord(mainTaskId, subRecord.GatewayId, subRecord.FileName, objName); err != nil {
			return fmt.Errorf("[FaultRecord] 路径写入失败(sn=%s): %w", subRecord.GatewayId, err)
		}
	}

	return nil
}

// ==================== 辅助函数（复用原有逻辑） ====================

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

func callEmsLog(sn string, date string, logType string) error {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		return fmt.Errorf("logType不存在: %s", logType)
	}
	return repositories.QueryCallEmsLog(sn, date, logTypeNum)
}

func getEmsLogUrl(sn string, date string, logType string) (*domain.EmsLogRecordInfo, error) {
	logTypeNum, ok := mapEmsLogType[logType]
	if !ok {
		return nil, fmt.Errorf("logType不存在: %s", logType)
	}
	return repositories.QueryEmsLogCallRecord(sn, date, logTypeNum)
}

func callFaultRecord(devType int, sn string, startTime string, endTime string, indexList []int) error {
	return repositories.QueryCellFaultRecord(devType, sn, indexList)
}

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 {
		if record.DeviceType == devType {
			var devSn string
			switch devType {
			case 0:
				devSn = *record.ApowerId
			case 1:
				devSn = record.GatewayId
			case 2:
				devSn = *record.ApowerId
			case 3:
				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)", devType, sn)
}

func parseIndexList(fileSupType string) ([]int, error) {
	indexStrList := strings.Split(fileSupType, ",")
	indexList := make([]int, 0, len(indexStrList))
	for _, s := range indexStrList {
		idx, err := strconv.Atoi(s)
		if err != nil {
			return nil, fmt.Errorf("无法解析 index '%s': %w", s, err)
		}
		indexList = append(indexList, idx)
	}
	return indexList, nil
}