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

package fwhQueue

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

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

// handleBatchDownload 是 BatchDownload 主任务的 Asynq handler
func handleBatchDownload(ctx context.Context, task *asynq.Task) error {
	var payload domain.TaskBatchDownload
	if err := json.Unmarshal(task.Payload(), &payload); err != nil {
		return fmt.Errorf("unmarshal BatchDownload payload: %w", err)
	}

	mainTaskId := payload.Id.Hex()

	// 更新运行状态
	if err := repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressRunning, []string{}); err != nil {
		zap.L().Error("[BatchDownload] 更新状态失败(Running)", zap.String("id", mainTaskId))
		return err
	}

	// 执行主逻辑。
	// 注意：任务终态（complete/failed）由 item 聚合（maybeAggregate）异步派生写入，
	// 这里只处理"任务根本无法开始"的致命错误 → ProgressError。
	if err := executeBatchDownload(ctx, payload); err != nil {
		zap.L().Error("[BatchDownload] 任务执行失败(致命)", zap.String("id", mainTaskId), zap.String("err", err.Error()))
		repositories.UpdateBatchDownloadResult(payload.Id, domain.ProgressError, []string{err.Error()})
		return err
	}

	return nil // 返回 nil 让 Asynq 标记任务为 completed，不触发重试
}

// executeBatchDownload 执行批量下载逻辑。
// 返回 error 仅表示"任务根本无法开始"的致命错误（→ ProgressError）；
// 进入执行阶段后的所有成败都以 item 形式记录，终态由聚合派生。
func executeBatchDownload(ctx context.Context, payload domain.TaskBatchDownload) error {
	mainTaskId := payload.Id.Hex()

	// 解析时间范围（致命错误：早退）
	startTime, err := time.Parse(time.DateOnly, payload.TimePeriod[0])
	if err != nil {
		return fmt.Errorf("startTime 格式错误: %w", err)
	}
	endTime, err := time.Parse(time.DateOnly, payload.TimePeriod[1])
	if err != nil {
		return fmt.Errorf("endTime 格式错误: %w", err)
	}
	endTime = endTime.Add(23*time.Hour + 59*time.Minute + 59*time.Second)
	if startTime.After(endTime) {
		return errors.New("startTime > endTime")
	}

	// 按天拆分时间段
	periods := splitTimePeriods(startTime, endTime)

	// ==================== 阶段0：规划所有 item ====================
	// planItems 枚举所有同步 + 异步工作单元为 pending；构建期查询失败/无数据直接写 failed/skipped。
	items, asyncTasks := planItems(payload, periods, startTime, endTime)

	if err := repositories.InitBatchDownloadItems(payload.Id, items); err != nil {
		return fmt.Errorf("初始化 items 失败: %w", err)
	}

	// 边界：若所有 item 在规划期就已是终态（无异步子任务、无同步任务），主动触发一次聚合。
	doneInit := 0
	for _, it := range items {
		if it.Status != domain.ItemPending {
			doneInit++
		}
	}
	if doneInit == len(items) {
		maybeAggregate(payload.Id, doneInit, len(items))
		return nil
	}

	// ==================== 阶段1：入队异步文件下载子任务 ====================
	for i, cfTask := range asyncTasks {
		taskPayload, _ := json.Marshal(cfTask)
		subTaskID := fmt.Sprintf("callfile-%s-%d", mainTaskId, i)
		t := asynq.NewTask(TaskTypeCallFile, taskPayload,
			asynq.MaxRetry(3),
			asynq.Timeout(35*time.Minute),
			asynq.Queue(QueueDownload),
			asynq.TaskID(subTaskID),
		)
		if _, err := TaskClient.Enqueue(t); err != nil {
			// 入队失败：直接把该 item 记为 failed(build)，doneCount 照常推进，避免永久 pending 挂起。
			zap.L().Error("[BatchDownload] 子任务入队失败",
				zap.String("id", mainTaskId),
				zap.Int("index", i),
				zap.String("key", cfTask.ItemKey),
				zap.String("err", err.Error()),
			)
			finishItem(payload.Id, cfTask.ItemKey, domain.ItemFailed, domain.StageBuild, "子任务入队失败: "+err.Error())
		}
	}

	// ==================== 阶段2：同步处理其他数据任务 ====================
	// 同步任务在主任务内联执行，失败记 item failed 而非中止；每个 item 终态都会驱动 maybeAggregate。
	executeSyncTasks(ctx, payload, periods)

	return nil
}

// callFilePayload 是传递给子任务的 payload 结构
type callFilePayload struct {
	MainTaskId string                           `json:"mainTaskId"`
	ItemKey    string                           `json:"itemKey"` // 对应 BatchDownloadItem.Key，子任务据此上报结果
	TaskInfo   domain.BatchDownloadFileTaskInfo `json:"taskInfo"`
}

// 同步任务的 item fileType 标识（与 executeSyncTasks 内的 key 生成保持一致）
const (
	fileTypePowerFlow     = "PowerFlow"
	fileTypeMqttLog       = "MqttLog"
	fileTypeEventAlarm    = "EventAlarm"
	fileTypeAlarmSnapshot = "AlarmSnapshot"
	fileTypeUpdateLog     = "UpdateLog"
)

// planItems 枚举本次批量下载的所有工作单元：
//   - 返回全部 items（pending / 构建期 failed / skipped）
//   - 返回需要异步入队的文件下载子任务列表（EmsLog / FaultRecord，payload 带 ItemKey）
//
// 同步类任务（PowerFlow/MqttLog/EventAlarm/AlarmSnapshot/UpdateLog）只产出 pending item，
// 由 executeSyncTasks 内联执行并上报。
func planItems(payload domain.TaskBatchDownload, periods []timePeriod, startTime, endTime time.Time) ([]domain.BatchDownloadItem, []callFilePayload) {
	mainTaskId := payload.Id.Hex()
	var items []domain.BatchDownloadItem
	var asyncTasks []callFilePayload

	newItem := func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem {
		return domain.BatchDownloadItem{
			Key:      domain.BuildItemKey(fileType, sn, devType, date, subType),
			Sn:       sn,
			DevType:  devType,
			FileType: fileType,
			SubType:  subType,
			Date:     date,
			Status:   status,
			Stage:    stage,
			Reason:   reason,
		}
	}

	for _, sn := range payload.Sn {
		// ---------- EmsLog（异步） ----------
		for _, period := range periods {
			dateStr := period.Start.Format(time.DateOnly)
			for _, logType := range payload.Options.EmsLogs {
				key := domain.BuildItemKey(jobs.FileTypeEmsLog, sn, -1, dateStr, logType)
				items = append(items, newItem(jobs.FileTypeEmsLog, sn, -1, dateStr, logType, domain.ItemPending, "", ""))
				asyncTasks = append(asyncTasks, callFilePayload{
					MainTaskId: mainTaskId,
					ItemKey:    key,
					TaskInfo: domain.BatchDownloadFileTaskInfo{
						DevType:     -1,
						Sn:          sn,
						Date:        dateStr,
						FileType:    jobs.FileTypeEmsLog,
						FileSupType: logType,
					},
				})
			}
		}

		// ---------- FaultRecord（异步） ----------
		if payload.Options.FaultRecord {
			frItems, frTasks := planFaultRecordItems(mainTaskId, sn, startTime, endTime, newItem)
			items = append(items, frItems...)
			asyncTasks = append(asyncTasks, frTasks...)
		}

		// ---------- 同步类 ----------
		if payload.Options.PowerFlow {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				items = append(items, newItem(fileTypePowerFlow, sn, -1, dateStr, "", domain.ItemPending, "", ""))
			}
		}
		for _, logType := range payload.Options.MqttLogs {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				items = append(items, newItem(fileTypeMqttLog, sn, -1, dateStr, logType, domain.ItemPending, "", ""))
			}
		}
		if payload.Options.EventAlarm {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				items = append(items, newItem(fileTypeEventAlarm, sn, -1, dateStr, "", domain.ItemPending, "", ""))
			}
		}
		if payload.Options.AlarmSnapshot {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				items = append(items, newItem(fileTypeAlarmSnapshot, sn, -1, dateStr, "", domain.ItemPending, "", ""))
			}
		}
		if payload.Options.UpdateLog {
			items = append(items, newItem(fileTypeUpdateLog, sn, -1, "", "", domain.ItemPending, "", ""))
		}
	}

	return items, asyncTasks
}

// planFaultRecordItems 为单个 gateway SN 规划故障录波 item（aGate + 其下 aPower）。
// 构建期查询失败 → failed(build)；无满足时间范围的数据 → skipped；有数据 → pending + 异步 payload。
func planFaultRecordItems(mainTaskId string, sn string, startTime, endTime time.Time,
	newItem func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem,
) ([]domain.BatchDownloadItem, []callFilePayload) {
	var items []domain.BatchDownloadItem
	var tasks []callFilePayload

	// aGate 可召唤列表
	frItems, frTasks := planFaultRecordItemsForDev(mainTaskId, 1, sn, startTime, endTime, newItem)
	items = append(items, frItems...)
	tasks = append(tasks, frTasks...)

	// aPower 子设备
	dateRange := startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime)
	siteBaseInfoList, err := repositories.QuerySiteBaseInfoList(dto.SiteBaseInfoPara{
		SiteList: []string{sn},
		PageSize: 10,
		PageNum:  1,
	})
	if err != nil {
		zap.L().Error("[BatchDownload] 查询站点基本信息失败", zap.String("sn", sn), zap.String("err", err.Error()))
		// 站点信息查询失败 → 无法枚举 aPower，记一条 failed(build) 归属该 SN
		items = append(items, newItem(jobs.FileTypeFaultRecord, sn, 0, dateRange, "", domain.ItemFailed, domain.StageBuild, "查询站点基本信息失败: "+err.Error()))
		return items, tasks
	}

	if siteBaseInfoList != nil && siteBaseInfoList.Result != nil && len(*siteBaseInfoList.Result) > 0 {
		aPowerSnList := strings.Split((*siteBaseInfoList.Result)[0].FhpSn, ",")
		for _, aPowerSn := range aPowerSnList {
			aPowerSn = strings.TrimSpace(aPowerSn)
			if aPowerSn == "" {
				continue
			}
			apItems, apTasks := planFaultRecordItemsForDev(mainTaskId, 0, aPowerSn, startTime, endTime, newItem)
			items = append(items, apItems...)
			tasks = append(tasks, apTasks...)
		}
	}

	return items, tasks
}

// planFaultRecordItemsForDev 为指定设备规划故障录波 item。
func planFaultRecordItemsForDev(mainTaskId string, devType int, sn string, startTime, endTime time.Time,
	newItem func(fileType, sn string, devType int, date, subType string, status domain.ItemStatus, stage, reason string) domain.BatchDownloadItem,
) ([]domain.BatchDownloadItem, []callFilePayload) {
	dateRange := startTime.Format(time.DateTime) + "," + endTime.Format(time.DateTime)

	callInfoList, err := repositories.QueryFaultRecordCallableList(devType, sn)
	if err != nil {
		zap.L().Error("[BatchDownload] 查询故障录波召唤选项失败", zap.Int("devType", devType), zap.String("sn", sn), zap.String("err", err.Error()))
		// 构建期查询失败 → failed(build)，错误得以上报
		item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, "", domain.ItemFailed, domain.StageBuild, "查询故障录波召唤选项失败: "+err.Error())
		return []domain.BatchDownloadItem{item}, nil
	}

	// 筛选日期范围内的记录
	var indexList string
	for _, info := range callInfoList {
		failureTime, err := parseFaultRecordTime(info.FailureTime)
		if err != nil {
			continue
		}
		inRange := (failureTime.Equal(startTime) || failureTime.After(startTime)) &&
			(failureTime.Equal(endTime) || failureTime.Before(endTime))
		if inRange {
			if indexList == "" {
				indexList = strconv.Itoa(info.Idx)
			} else {
				indexList = indexList + "," + strconv.Itoa(info.Idx)
			}
		}
	}

	// 无符合日期范围的记录 → skipped（非错误）
	if indexList == "" {
		item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, "", domain.ItemSkipped, "", "无满足时间范围的录波数据")
		return []domain.BatchDownloadItem{item}, nil
	}

	// 有数据 → pending + 异步 payload
	key := domain.BuildItemKey(jobs.FileTypeFaultRecord, sn, devType, dateRange, indexList)
	item := newItem(jobs.FileTypeFaultRecord, sn, devType, dateRange, indexList, domain.ItemPending, "", "")
	task := callFilePayload{
		MainTaskId: mainTaskId,
		ItemKey:    key,
		TaskInfo: domain.BatchDownloadFileTaskInfo{
			DevType:     devType,
			Sn:          sn,
			Date:        dateRange,
			FileType:    jobs.FileTypeFaultRecord,
			FileSupType: indexList,
		},
	}
	return []domain.BatchDownloadItem{item}, []callFilePayload{task}
}

// executeSyncTasks 同步执行非文件下载类任务（PowerFlow、MqttLogs、EventAlarm 等）。
// 每个工作单元成功/失败都通过 finishItem 上报，失败不中止整个任务。
func executeSyncTasks(ctx context.Context, payload domain.TaskBatchDownload, periods []timePeriod) {
	for _, sn := range payload.Sn {
		// PowerFlow
		if payload.Options.PowerFlow {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				key := domain.BuildItemKey(fileTypePowerFlow, sn, -1, dateStr, "")
				if err := syncPowerFlow(payload.Id, sn, dateStr, period); err != nil {
					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
				} else {
					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
				}
			}
		}

		// MqttLogs
		for _, logType := range payload.Options.MqttLogs {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				key := domain.BuildItemKey(fileTypeMqttLog, sn, -1, dateStr, logType)
				if err := syncMqttLog(payload.Id, sn, logType, dateStr, period); err != nil {
					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
				} else {
					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
				}
			}
		}

		// EventAlarm
		if payload.Options.EventAlarm {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				key := domain.BuildItemKey(fileTypeEventAlarm, sn, -1, dateStr, "")
				if err := syncEventAlarm(payload.Id, sn, dateStr, period); err != nil {
					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
				} else {
					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
				}
			}
		}

		// AlarmSnapshot
		if payload.Options.AlarmSnapshot {
			for _, period := range periods {
				dateStr := period.Start.Format(time.DateOnly)
				key := domain.BuildItemKey(fileTypeAlarmSnapshot, sn, -1, dateStr, "")
				if err := syncAlarmSnapshot(ctx, payload.Id, sn, dateStr); err != nil {
					finishItem(payload.Id, key, domain.ItemFailed, err.stage, err.reason)
				} else {
					finishItem(payload.Id, key, domain.ItemSuccess, "", "")
				}
			}
		}

		// UpdateLog
		if payload.Options.UpdateLog {
			key := domain.BuildItemKey(fileTypeUpdateLog, sn, -1, "", "")
			if err := executeSyncUpdateLog(sn, payload); err != nil {
				finishItem(payload.Id, key, domain.ItemFailed, domain.StageQuery, err.Error())
			} else {
				finishItem(payload.Id, key, domain.ItemSuccess, "", "")
			}
		}
	}
}

// syncItemError 携带失败阶段与原因，便于 finishItem 记录 stage
type syncItemError struct {
	stage  string
	reason string
}

func (e *syncItemError) Error() string { return e.reason }

func newSyncErr(stage string, err error) *syncItemError {
	return &syncItemError{stage: stage, reason: err.Error()}
}

// syncPowerFlow 处理单个 PowerFlow 工作单元
func syncPowerFlow(id primitive.ObjectID, sn, dateStr string, period timePeriod) *syncItemError {
	res, err := repositories.QuerySitePowerFlow(sn, period.Start)
	if err != nil {
		return newSyncErr(domain.StageQuery, err)
	}
	objName, err := jobs.SaveBatchDownloadPowerFlow(sn, dateStr, *res.Result)
	if err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	if err := repositories.InsertBatchDownloadPathPowerFlow(id, sn, dateStr, objName); err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	return nil
}

// syncMqttLog 处理单个 MqttLog 工作单元
func syncMqttLog(id primitive.ObjectID, sn, logType, dateStr string, period timePeriod) *syncItemError {
	logList, err := repositories.QueryMqttLogListOnTimeRange(sn, domain.MqttCmdType(logType), period.Start, period.End)
	if err != nil {
		return newSyncErr(domain.StageQuery, err)
	}
	objName, err := jobs.SaveBatchDownloadMqttLog(sn, logType, dateStr, logList)
	if err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	if err := repositories.InsertBatchDownloadPathMqttLogs(id, sn, logType, dateStr, objName); err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	return nil
}

// syncEventAlarm 处理单个 EventAlarm 工作单元
func syncEventAlarm(id primitive.ObjectID, sn, dateStr string, period timePeriod) *syncItemError {
	eventAlarmList, err := repositories.QueryEventOrAlarmListOnTimeRange(sn, period.Start, period.End, repositories.EventAndAlarm)
	if err != nil {
		return newSyncErr(domain.StageQuery, err)
	}
	objName, err := jobs.SaveBatchDownloadEventAlarm(sn, dateStr, eventAlarmList)
	if err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	if err := repositories.InsertBatchDownloadPathEventAlarm(id, sn, dateStr, objName); err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	return nil
}

// syncAlarmSnapshot 处理单个 AlarmSnapshot 工作单元
func syncAlarmSnapshot(ctx context.Context, id primitive.ObjectID, sn, dateStr string) *syncItemError {
	alarmSnapshotList, err := jobs.GetAlarmSnapshotData(ctx, sn, dateStr)
	if err != nil {
		return newSyncErr(domain.StageQuery, err)
	}
	objName, err := jobs.SaveBatchDownloadAlarmSnapshot(sn, dateStr, alarmSnapshotList)
	if err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	if err := repositories.InsertBatchDownloadPathAlarmSnapshot(id, sn, dateStr, objName); err != nil {
		return newSyncErr(domain.StageSave, err)
	}
	return nil
}

// ==================== 辅助函数 ====================

type timePeriod struct {
	Start time.Time
	End   time.Time
}

func splitTimePeriods(startTime, endTime time.Time) []timePeriod {
	var periods []timePeriod
	curr := startTime
	for curr.Before(endTime) {
		dayEnd := time.Date(curr.Year(), curr.Month(), curr.Day(), 23, 59, 59, 0, curr.Location())
		if dayEnd.After(endTime) {
			dayEnd = endTime
		}
		periods = append(periods, timePeriod{Start: curr, End: dayEnd})
		curr = time.Date(curr.Year(), curr.Month(), curr.Day()+1, 0, 0, 0, 0, curr.Location())
	}
	return periods
}

func parseFaultRecordTime(failureTime string) (time.Time, error) {
	parts := strings.Split(failureTime, " ")
	if len(parts) < 2 {
		return time.Time{}, fmt.Errorf("invalid failureTime format: %s", failureTime)
	}
	date := parts[0]
	timeStr := parts[1]

	d := strings.Split(date, "-")
	t := strings.Split(timeStr, ":")
	if len(d) < 3 || len(t) < 3 {
		return time.Time{}, fmt.Errorf("invalid failureTime format: %s", failureTime)
	}

	standardTime := fmt.Sprintf("%04s-%02s-%02s %02s:%02s:%02s", d[0], d[1], d[2], t[0], t[1], t[2])
	return time.Parse(time.DateTime, standardTime)
}

// executeSyncUpdateLog 同步执行升级日志任务（从原代码提取）
func executeSyncUpdateLog(sn string, payload domain.TaskBatchDownload) error {
	updateLogList := []domain.BatchDownloadSystemUpdateLogInfo{}

	systemUpdateLogList, err := repositories.QuerySystemUpdateLog(sn)
	if err != nil {
		return err
	}

	for _, logList := range systemUpdateLogList {
		var updateLogInfo domain.BatchDownloadSystemUpdateLogInfo
		updateLogInfo.OrderNo = logList.OrderNo
		updateLogInfo.GateSn = sn
		updateLogInfo.FileName = logList.Name

		upgradeTime, err := time.ParseInLocation(time.DateTime, logList.UpgradeTime, time.UTC)
		if err != nil {
			zap.L().Warn("[BatchDownload] 升级时间解析失败", zap.String("time", logList.UpgradeTime))
			continue
		}
		beijing, _ := time.LoadLocation("Asia/Shanghai")
		updateLogInfo.UpdateTime = upgradeTime.In(beijing).Format(time.DateTime)

		if strings.HasPrefix(logList.OrderNo, "CH") {
			updateLogInfo.TriggerType = "Auto"
		} else if strings.HasPrefix(logList.OrderNo, "DH") {
			updateLogInfo.TriggerType = "Manual"
		} else if strings.HasPrefix(logList.OrderNo, "SH") {
			updateLogInfo.TriggerType = "Span Auto"
		} else {
			updateLogInfo.TriggerType = "unknown"
		}

		switch logList.Result {
		case 1:
			updateLogInfo.Result = "Updating"
		case 2:
			updateLogInfo.Result = "Success"
		case 3:
			updateLogInfo.Result = "Fault"
		default:
			updateLogInfo.Result = fmt.Sprintf("unknown(%d)", logList.Result)
		}

		systemUpdateLogListDetail, err := repositories.QuerySystemUpdateLogDetail(logList.Id, sn)
		if err != nil {
			return err
		}

		for _, logDetail := range systemUpdateLogListDetail {
			if logDetail.GatewayId == sn {
				for _, ibgVer := range logDetail.IbgVersionNew {
					if ibgVer.Children != nil {
						for _, childVerInfo := range ibgVer.Children {
							updateLogInfo.GateVerList = append(updateLogInfo.GateVerList, domain.BatchDownloadUpdateLogVersionInfo{
								Name: childVerInfo.Type, OldVer: childVerInfo.OldVersion, NewVer: childVerInfo.Version,
							})
						}
					} else {
						updateLogInfo.GateVerList = append(updateLogInfo.GateVerList, domain.BatchDownloadUpdateLogVersionInfo{
							Name: ibgVer.Type, OldVer: ibgVer.OldVersion, NewVer: ibgVer.Version,
						})
					}
				}
				for _, subInfo := range logDetail.SonNewDevices {
					var subDevInfo domain.BatchDownloadUpdateLogSubDevInfo
					subDevInfo.DevSn = subInfo.ApowerSn
					for _, firstItem := range subInfo.FirstList {
						for _, secondItem := range firstItem.Secondlist {
							for _, thirdItem := range secondItem.Thirdlist {
								subDevInfo.DevVerList = append(subDevInfo.DevVerList, domain.BatchDownloadUpdateLogVersionInfo{
									Name: thirdItem.SubType, OldVer: thirdItem.OldVersion, NewVer: thirdItem.Version,
								})
							}
						}
					}
					updateLogInfo.SubDevList = append(updateLogInfo.SubDevList, subDevInfo)
				}
				break
			}
		}
		updateLogList = append(updateLogList, updateLogInfo)
	}

	// aHub 升级日志
	aHubUpdateLogList, err := repositories.QueryAHubUpdateLog(sn)
	if err != nil {
		return err
	}

	var aHubUpdateLogData []domain.BatchDownloadAHubUpdateLogInfo
	for _, aHubVerInfo := range aHubUpdateLogList {
		var info domain.BatchDownloadAHubUpdateLogInfo
		info.OrderNo = aHubVerInfo.OrderNo
		info.GateSn = sn
		info.AHubSn = aHubVerInfo.AHubSn

		if strings.HasPrefix(aHubVerInfo.OrderNo, "CH") {
			info.TriggerType = "Auto"
		} else if strings.HasPrefix(aHubVerInfo.OrderNo, "DH") {
			info.TriggerType = "Manual"
		} else if strings.HasPrefix(aHubVerInfo.OrderNo, "SH") {
			info.TriggerType = "Span Auto"
		} else {
			info.TriggerType = "unknown"
		}

		switch aHubVerInfo.UpgradeResult {
		case 0:
			info.Result = "Success"
		case -1:
			info.Result = "Fault"
		default:
			info.Result = fmt.Sprintf("unknown(%d)", aHubVerInfo.UpgradeResult)
		}

		for _, detailInfo := range aHubVerInfo.DetailList {
			info.DetailList = append(info.DetailList, domain.BatchDownloadUpdateLogAHubVersionInfo{
				FileName:   detailInfo.UpgradeFileName,
				FileOrder:  detailInfo.UpgradeFileOrder,
				Result:     detailInfo.UpgradeResult,
				ResultDesc: detailInfo.UpgradeResultDesc,
				OldVer:     detailInfo.OldVersion,
				NewVer:     detailInfo.NewVersion,
			})
		}
		aHubUpdateLogData = append(aHubUpdateLogData, info)
	}

	objName, err := jobs.SaveBatchDownloadUpdateLog(sn, updateLogList, aHubUpdateLogData)
	if err != nil {
		return err
	}
	return repositories.InsertBatchDownloadPathUpdateLog(payload.Id, sn, objName)
}

// ==================== 聚合与完成判定（B2 + 策略C） ====================

// deriveProgressAndSummary 从 items 聚合派生任务终态与摘要 failReasons（策略C：摘要，明细在 items）
func deriveProgressAndSummary(items []domain.BatchDownloadItem) (domain.Progress, []string) {
	failed := 0
	var samples []string
	for _, it := range items {
		if it.Status == domain.ItemFailed {
			failed++
			if len(samples) < 5 { // 摘要仅保留前若干条样本
				samples = append(samples, fmt.Sprintf("[%s] sn=%s date=%s stage=%s: %s",
					it.FileType, it.Sn, it.Date, it.Stage, it.Reason))
			}
		}
	}

	if failed == 0 {
		return domain.ProgressComplete, []string{}
	}

	summary := append([]string{fmt.Sprintf("共 %d 项失败（详见明细 items）", failed)}, samples...)
	return domain.ProgressFailed, summary
}

// maybeAggregate 在 done==total（本任务全部 item 已终态）时聚合派生终态并写库。
// 由 FinishBatchDownloadItem 的原子性保证只有一个调用会命中 done==total，聚合仅触发一次。
// done/total 为 -1 表示幂等跳过（重复回调），不处理。
func maybeAggregate(id primitive.ObjectID, done, total int) {
	if done < 0 || total < 0 {
		return
	}
	if done != total {
		return
	}

	items, err := repositories.GetBatchDownloadItems(id)
	if err != nil {
		zap.L().Error("[BatchDownload] 聚合读取 items 失败", zap.String("id", id.Hex()), zap.String("err", err.Error()))
		return
	}

	progress, summary := deriveProgressAndSummary(items)
	if err := repositories.UpdateBatchDownloadResult(id, progress, summary); err != nil {
		zap.L().Error("[BatchDownload] 聚合写入结果失败", zap.String("id", id.Hex()), zap.String("err", err.Error()))
		return
	}
	zap.L().Info("[BatchDownload] 任务聚合完成", zap.String("id", id.Hex()), zap.String("progress", string(progress)))
}

// finishItem 是对 repositories.FinishBatchDownloadItem + maybeAggregate 的封装，
// 主任务同步任务与子任务共用，保证每个 item 转终态后驱动一次完成检查。
func finishItem(id primitive.ObjectID, key string, status domain.ItemStatus, stage, reason string) {
	done, total, err := repositories.FinishBatchDownloadItem(id, key, status, stage, reason)
	if err != nil {
		zap.L().Error("[BatchDownload] 更新 item 结果失败",
			zap.String("id", id.Hex()), zap.String("key", key), zap.String("err", err.Error()))
		return
	}
	maybeAggregate(id, done, total)
}