FranklinWH-Toolbox/service/repositories/api/fwh_pro/fwh_mqttLog.go

package fwh_pro

import (
	"errors"
	"fmt"
	"fwh-toolbox-server/domain"
	"net/url"
	"strconv"
	"strings"
	"time"

	"go.uber.org/zap"
)

type QueryMqttLogParam struct {
	SN        *string
	CmdTypes  []domain.MqttCmdType
	StartTime *time.Time
	EndTime   *time.Time
}

// 查询站点信息
func queryMqttLog(param QueryMqttLogParam) (*domain.FwhPageResponse[[]domain.MqttLogItem], error) {
	// 构造 query 参数
	params := url.Values{}
	if param.SN != nil {
		params.Add("searchName", *param.SN)
	}
	if len(param.CmdTypes) > 0 {
		cmdTypesStr := make([]string, len(param.CmdTypes))
		for i, v := range param.CmdTypes {
			cmdTypesStr[i] = string(v) // 转成 string
		}
		params.Add("cmdTypes", strings.Join(cmdTypesStr, ","))
	}
	if param.StartTime != nil {
		params.Add("startTime", param.StartTime.Format(time.DateTime))
	}
	if param.EndTime != nil {
		params.Add("endTime", param.EndTime.Format(time.DateTime))
	}
	if param.StartTime != nil || param.EndTime != nil {
		params.Add("timeType", "1") // 0-通讯时间 1-设备时间时
	}
	params.Add("pageSize", strconv.FormatUint(REQ_PAGE_SIZE, 10))
	params.Add("pageNum", strconv.Itoa(1))
	params.Add("lang", "ZH_CN")

	response, err := FwhHttpGetUnmarshalPage[[]domain.MqttLogItem](getFullEnergyUrl("/hes-gateway/manage/iotMqttCommlogs"), params)
	if err != nil {
		return nil, err
	}
	return response, nil
}

func queryMqttLogListOnePage(sn string, cmdType domain.MqttCmdType, startTime, endTime time.Time) ([]domain.MqttLogItem, error) {
	param := QueryMqttLogParam{
		SN:        &sn,
		CmdTypes:  []domain.MqttCmdType{cmdType},
		StartTime: &startTime,
		EndTime:   &endTime,
	}
	mqttLogRes, err := queryMqttLog(param)
	if err != nil {
		return nil, err
	}

	if mqttLogRes.Code != 200 {
		return nil, fmt.Errorf("[queryMqttLogListOnePage] 查询失败 code: %d, message: %s", mqttLogRes.Code, mqttLogRes.Message)
	}

	// 判断是否超出可处理数据上限 10万条数据
	if mqttLogRes.Total > 10*10000 {
		return nil, fmt.Errorf("超出数据请求上限 - Total: %d (>10万)", mqttLogRes.Total)
	}

	return *mqttLogRes.Result, nil
}

// 从FWH获取MQTT log数据
//
// 限制条件（重要）：startTime/endTime 必须为「不含时区」的字面时间（用 UTC 表示）。
// 本函数按天拆分时间段时用 time.Truncate(24h) 推进游标，Truncate 按 UTC 零点对齐；
// 若传入带非 UTC 时区（如 America/Chicago）的时间，游标会回退到前一天，
// currTime.Before(endTime) 恒成立，periods 切片无限 append，导致内存 OOM。
// 调用方（DataDownload）需先用 stripTimeLocation 剥离时区再传入。
func QueryMqttLogListOnTimeRange(sn string, cmdType domain.MqttCmdType, startTime, endTime time.Time) ([]domain.MqttLogItem, error) {
	// 保证 startTime <= endTime
	if startTime.After(endTime) {
		return nil, errors.New("startTime > endTime")
	}

	// 按天拆分时间段
	var periods []struct{ Start, End time.Time }
	currTime := startTime
	for currTime.Before(endTime) {
		// 当天结束时间
		dayEnd := time.Date(currTime.Year(), currTime.Month(), currTime.Day(), 23, 59, 59, 0, currTime.Location())

		// 如果当天结束大于 endTime，使用 endTime
		if dayEnd.After(endTime) {
			dayEnd = endTime
		}

		periods = append(periods, struct{ Start, End time.Time }{Start: currTime, End: dayEnd})

		// 下一天开始
		currTime = dayEnd.Add(time.Second).Truncate(24 * time.Hour)
		currTime = time.Date(currTime.Year(), currTime.Month(), currTime.Day(), 0, 0, 0, 0, currTime.Location())
	}
	zap.L().Debug("[fwh_pro] MQTT periods 构建完成",
		zap.String("sn", sn), zap.Any("cmdType", cmdType),
		zap.Time("start", startTime), zap.Time("end", endTime),
		zap.Int("periods", len(periods)),
	)

	// 按天查询
	var mqttLogList []domain.MqttLogItem
	for _, period := range periods {
		var mqttLogRes []domain.MqttLogItem = nil
		periodEndTime := period.End
		pages := 0
		// 检查数据是否超过1000(单次查询上限)
		for {
			var err error
			pages++
			zap.L().Debug("[fwh_pro] MQTT 分页请求发起",
				zap.String("sn", sn), zap.Any("cmdType", cmdType),
				zap.Time("periodStart", period.Start), zap.Time("periodEnd", periodEndTime),
				zap.Int("page", pages), zap.Int("accumulated", len(mqttLogList)),
			)
			mqttLogRes, err = queryMqttLogListOnePage(sn, cmdType, period.Start, periodEndTime)
			if err != nil {
				return nil, err
			}

			time.Sleep(time.Second) // 延时1s

			mqttLogList = append(mqttLogList, mqttLogRes...)
			zap.L().Debug("[fwh_pro] MQTT 分页请求返回",
				zap.String("sn", sn), zap.Any("cmdType", cmdType),
				zap.Int("page", pages), zap.Int("pageRows", len(mqttLogRes)),
				zap.Int("accumulated", len(mqttLogList)),
			)

			// 判断是否未查完
			if len(mqttLogRes) < int(REQ_PAGE_SIZE) {
				// 请求完成，退出
				break
			}

			// 跳转时间段，再次请求
			// 提取最后一条数据的设备时间
			next, err := time.Parse(time.DateTime, mqttLogRes[len(mqttLogRes)-1].LogTime)
			if err != nil {
				return nil, err
			}

			// 游标未能前进：说明同一秒内 >= REQ_PAGE_SIZE(1000) 条数据，
			// 单页游标（按最后一条 LogTime）无法翻过这一秒，继续循环会死循环并 OOM。
			// 现实中基本不会发生，此处仅作防护：记录错误并返回，不再无限拉取。
			if !next.Before(periodEndTime) {
				zap.L().Error("[fwh_pro] MQTT 日志分页游标停滞，疑似同一秒内数据超过单页上限",
					zap.String("sn", sn),
					zap.Any("cmdType", cmdType),
					zap.Time("stuckAt", next),
					zap.Int("pageSize", int(REQ_PAGE_SIZE)),
					zap.Int("accumulated", len(mqttLogList)),
				)
				return nil, fmt.Errorf("mqtt log pagination stalled at %s: more than %d rows within one second", next.Format(time.DateTime), REQ_PAGE_SIZE)
			}
			periodEndTime = next
		}
	}

	return mqttLogList, nil
}