FranklinWH-Toolbox/service/services/dataDownloadService.go

package services

import (
	"context"
	"fmt"
	"math"
	"sort"
	"strconv"
	"strings"
	"time"

	"fwh-toolbox-server/cache"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/dto"
	"fwh-toolbox-server/repositories"
	mongoRepo "fwh-toolbox-server/repositories/mongo"
	pgsqlRepo "fwh-toolbox-server/repositories/pgsql"

	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

const dataDownloadMaxDays = 600

func CreateDataDownloadTask(ctx context.Context, task *domain.DataDownloadTask) error {
	if len(task.SNs) == 0 || task.Options.Empty() {
		return fmt.Errorf("at least one SN and one data type are required")
	}
	seen := map[string]struct{}{}
	normalized := make([]string, 0, len(task.SNs))
	for _, raw := range task.SNs {
		sn := strings.TrimSpace(raw)
		if sn == "" {
			continue
		}
		if _, exists := seen[sn]; !exists {
			seen[sn] = struct{}{}
			normalized = append(normalized, sn)
		}
	}
	if len(normalized) == 0 {
		return fmt.Errorf("at least one valid SN is required")
	}
	task.SNs = normalized
	if hasDateBasedOption(task.Options) {
		start, err := time.Parse(time.DateOnly, task.StartDate)
		if err != nil {
			return fmt.Errorf("invalid startDate")
		}
		end, err := time.Parse(time.DateOnly, task.EndDate)
		if err != nil {
			return fmt.Errorf("invalid endDate")
		}
		if end.Before(start) {
			return fmt.Errorf("endDate must not be before startDate")
		}
		if int(end.Sub(start).Hours()/24)+1 > dataDownloadMaxDays {
			return fmt.Errorf("date range must not exceed %d days", dataDownloadMaxDays)
		}
	}
	now := time.Now().UTC()
	task.ID = primitive.NewObjectID()
	task.Status = domain.DataDownloadPending
	task.CreatedAt, task.UpdatedAt = now, now
	return mongoRepo.InsertDataDownloadTask(ctx, task)
}

func hasDateBasedOption(o domain.DataDownloadOptions) bool {
	return o.PowerFlow || len(o.MqttLogs) > 0 || o.EventAlarm || o.AlarmSnapshot || len(o.EmsLogs) > 0 || o.FaultRecord
}

func PlanDataDownloadTask(ctx context.Context, task *domain.DataDownloadTask) ([]domain.DataDownloadItem, error) {
	var start, end time.Time
	var err error
	if hasDateBasedOption(task.Options) {
		start, err = time.Parse(time.DateOnly, task.StartDate)
		if err != nil {
			return nil, err
		}
		end, err = time.Parse(time.DateOnly, task.EndDate)
		if err != nil {
			return nil, err
		}
	}
	now := time.Now().UTC()
	items := []domain.DataDownloadItem{}
	for _, sn := range task.SNs {
		response, err := repositories.QuerySiteInfoList(dto.SiteInfoPara{SN: &sn, PageSize: 20, PageNum: 1})
		if err != nil {
			zap.L().Warn("[DataDownload] 规划：站点信息查询失败",
				zap.String("taskId", task.ID.Hex()), zap.String("sn", sn), zap.String("err", err.Error()))
			item := newDDItem(task.ID, "site_validation", sn, "", "", "", now)
			item.Status, item.Stage, item.FailureReason = domain.DataDownloadFailed, "validation", err.Error()
			items = append(items, item)
			continue
		}
		var site *domain.FwhSiteInfo
		if response != nil && response.Result != nil {
			for i := range *response.Result {
				if (*response.Result)[i].Id == sn {
					site = &(*response.Result)[i]
					break
				}
			}
		}
		if site == nil {
			item := newDDItem(task.ID, "site_validation", sn, "", "", "", now)
			item.Status, item.Stage, item.FailureReason = domain.DataDownloadFailed, "validation", "site does not exist"
			items = append(items, item)
			continue
		}
		zone, loc := siteLocation(site)
		var faultDevices []dataDownloadFaultDevice
		var faultDevicesErr error
		if task.Options.FaultRecord {
			faultDevices, faultDevicesErr = discoverDataDownloadFaultDevices(sn)
			if faultDevicesErr != nil {
				item := newDDItem(task.ID, domain.DataDownloadFaultRecord, sn, "", "device_discovery", zone, now)
				item.Status, item.Stage, item.FailureReason = domain.DataDownloadFailed, "planning", faultDevicesErr.Error()
				items = append(items, item)
			}
		}
		if hasDateBasedOption(task.Options) {
			localToday := now.In(loc).Format(time.DateOnly)
			if task.EndDate > localToday {
				return nil, fmt.Errorf("future date is not allowed for site %s (%s)", sn, zone)
			}
			for date := start; !date.After(end); date = date.AddDate(0, 0, 1) {
				dateText := date.Format(time.DateOnly)
				if task.Options.PowerFlow {
					items = append(items, newDDItem(task.ID, domain.DataDownloadPowerFlow, sn, dateText, "", zone, now))
				}
				for _, subtype := range task.Options.MqttLogs {
					items = append(items, newDDItem(task.ID, domain.DataDownloadMqttLog, sn, dateText, subtype, zone, now))
				}
				if task.Options.EventAlarm {
					items = append(items, newDDItem(task.ID, domain.DataDownloadEventAlarm, sn, dateText, "", zone, now))
				}
				if task.Options.AlarmSnapshot {
					items = append(items, newDDItem(task.ID, domain.DataDownloadAlarmSnapshot, sn, dateText, "", zone, now))
				}
				for _, subtype := range task.Options.EmsLogs {
					items = append(items, newDDItem(task.ID, domain.DataDownloadEmsLog, sn, dateText, subtype, zone, now))
				}
				if task.Options.FaultRecord {
					for _, device := range faultDevices {
						items = append(items, newDDFaultRecordItem(task.ID, sn, device.SN, device.Type, dateText, zone, now))
					}
				}
			}
		}
		if task.Options.UpdateLog {
			items = append(items, newDDItem(task.ID, domain.DataDownloadUpdateLog, sn, "", "", zone, now))
		}
	}
	return items, nil
}

type dataDownloadFaultDevice struct {
	SN   string
	Type int
}

func discoverDataDownloadFaultDevices(siteSN string) ([]dataDownloadFaultDevice, error) {
	devices := map[string]dataDownloadFaultDevice{}
	add := func(sn string, deviceType int) {
		sn = strings.TrimSpace(sn)
		if sn != "" {
			devices[fmt.Sprintf("%d|%s", deviceType, sn)] = dataDownloadFaultDevice{SN: sn, Type: deviceType}
		}
	}
	add(siteSN, 1)

	base, err := repositories.QuerySiteBaseInfoList(dto.SiteBaseInfoPara{SiteList: []string{siteSN}, PageSize: 20, PageNum: 1})
	if err != nil {
		return sortedDataDownloadFaultDevices(devices), fmt.Errorf("discover fault-record devices for %s: %w", siteSN, err)
	}
	if base != nil && base.Result != nil && len(*base.Result) > 0 {
		for _, sn := range strings.Split((*base.Result)[0].FhpSn, ",") {
			add(sn, 0)
		}
	}

	return sortedDataDownloadFaultDevices(devices), nil
}

func sortedDataDownloadFaultDevices(devices map[string]dataDownloadFaultDevice) []dataDownloadFaultDevice {
	result := make([]dataDownloadFaultDevice, 0, len(devices))
	for _, device := range devices {
		result = append(result, device)
	}
	sort.Slice(result, func(i, j int) bool {
		if result[i].Type == result[j].Type {
			return result[i].SN < result[j].SN
		}
		return result[i].Type < result[j].Type
	})
	return result
}

func dataDownloadFaultRecordDeviceSN(record domain.FaultRecordInfo) (string, error) {
	switch record.DeviceType {
	case 0, 2:
		if record.ApowerId != nil && strings.TrimSpace(*record.ApowerId) != "" {
			return strings.TrimSpace(*record.ApowerId), nil
		}
	case 1:
		if strings.TrimSpace(record.GatewayId) != "" {
			return strings.TrimSpace(record.GatewayId), nil
		}
	case 3:
		if record.PowerlinkId != nil && strings.TrimSpace(*record.PowerlinkId) != "" {
			return strings.TrimSpace(*record.PowerlinkId), nil
		}
	default:
		return "", fmt.Errorf("unsupported fault-record device type %d", record.DeviceType)
	}
	return "", fmt.Errorf("fault-record device SN is missing for type %d record %d", record.DeviceType, record.Id)
}

func siteLocation(site *domain.FwhSiteInfo) (string, *time.Location) {
	if site.ZoneInfo != "" {
		if loc, err := time.LoadLocation(site.ZoneInfo); err == nil {
			return site.ZoneInfo, loc
		}
	}
	offset := int(math.Round(float64(site.TimeZone) * 3600))
	name := fmt.Sprintf("UTC%+.2f", site.TimeZone)
	return name, time.FixedZone(name, offset)
}

func dataDownloadLocation(name string) *time.Location {
	if loc, err := time.LoadLocation(name); err == nil {
		return loc
	}
	if strings.HasPrefix(name, "UTC") {
		if hours, err := strconv.ParseFloat(strings.TrimPrefix(name, "UTC"), 64); err == nil {
			return time.FixedZone(name, int(math.Round(hours*3600)))
		}
	}
	return time.UTC
}

func newDDItem(taskID primitive.ObjectID, kind domain.DataDownloadKind, sn, date, subtype, zone string, now time.Time) domain.DataDownloadItem {
	key := strings.Join([]string{string(kind), sn, date, subtype}, "|")
	return domain.DataDownloadItem{ID: primitive.NewObjectID(), TaskID: taskID, ItemKey: key, Kind: kind, SiteSN: sn, Date: date, Subtype: subtype, Timezone: zone, Status: domain.DataDownloadPending, CreatedAt: now, UpdatedAt: now}
}

func newDDFaultRecordItem(taskID primitive.ObjectID, siteSN, deviceSN string, deviceType int, date, zone string, now time.Time) domain.DataDownloadItem {
	item := newDDItem(taskID, domain.DataDownloadFaultRecord, siteSN, date, "", zone, now)
	item.DeviceSN = deviceSN
	item.DeviceType = strconv.Itoa(deviceType)
	item.ItemKey = strings.Join([]string{string(item.Kind), siteSN, deviceSN, item.DeviceType, date}, "|")
	return item
}

func ExecuteStructuredDataDownloadItem(ctx context.Context, item *domain.DataDownloadItem, awaitRequestIDs ...string) error {
	if item.Kind == domain.DataDownloadAlarmSnapshot || item.Kind == domain.DataDownloadEmsLog || item.Kind == domain.DataDownloadFaultRecord {
		return ExecuteBinaryDataDownloadItem(ctx, item, awaitRequestIDs...)
	}
	awaitRequestID := ""
	if len(awaitRequestIDs) > 0 {
		awaitRequestID = awaitRequestIDs[0]
	}
	ready, lease, err := prepareDataDownloadRequest(ctx, item, awaitRequestID)
	if err != nil {
		return err
	}
	if ready != nil {
		return linkDataDownloadItem(ctx, item, ready, ready.Empty)
	}
	defer lease.abort()
	requestKey := lease.requestKey
	documentID := lease.documentID

	loc := dataDownloadLocation(item.Timezone)
	var start, end time.Time
	if item.Date != "" {
		start, err = time.ParseInLocation(time.DateOnly, item.Date, loc)
		if err != nil {
			return err
		}
		end = start.AddDate(0, 0, 1).Add(-time.Nanosecond)
	}
	// 传给上游日志查询的时间必须是「不含时区」的字面时间(按 UTC 表示)。
	// QueryMqttLogListOnTimeRange / QueryEventOrAlarmListOnTimeRange 内部按天拆分时使用
	// time.Truncate(24h)(按 UTC 零点对齐)，若传入带非 UTC 时区(如 America/Chicago)的时间，
	// 游标推进会回退到前一天，导致 periods 切片无限增长并 OOM。这里剥离时区、只保留字面日期。
	// 注意：start/end(本地时区)仍用于后面的 Frozen(D+2)判断，不能改动它们本身。
	startQuery := stripTimeLocation(start)
	endQuery := stripTimeLocation(end)
	fetchedAt := time.Now().UTC()
	rows := []pgsqlRepo.DataDownloadRow{}
	switch item.Kind {
	case domain.DataDownloadPowerFlow:
		response, err := repositories.QuerySitePowerFlow(item.SiteSN, start)
		if err != nil {
			zap.L().Warn("[DataDownload] PowerFlow 上游查询失败", zap.String("sn", item.SiteSN), zap.String("date", item.Date), zap.String("err", err.Error()))
			return err
		}
		if response.Result != nil {
			rows = append(rows, pgsqlRepo.DataDownloadRow{Payload: response.Result})
		}
		zap.L().Debug("[DataDownload] PowerFlow 上游返回", zap.String("sn", item.SiteSN), zap.String("date", item.Date), zap.Bool("hasData", response.Result != nil))
	case domain.DataDownloadMqttLog:
		values, err := repositories.QueryMqttLogListOnTimeRange(item.SiteSN, domain.MqttCmdType(item.Subtype), startQuery, endQuery)
		if err != nil {
			zap.L().Warn("[DataDownload] MqttLog 上游查询失败", zap.String("sn", item.SiteSN), zap.String("subtype", item.Subtype), zap.String("date", item.Date), zap.String("err", err.Error()))
			return err
		}
		for i := range values {
			occurred := parseRemoteTime(values[i].Time)
			rows = append(rows, pgsqlRepo.DataDownloadRow{OccurredAt: occurred, Subtype: item.Subtype, Payload: values[i]})
		}
		zap.L().Debug("[DataDownload] MqttLog 上游返回", zap.String("sn", item.SiteSN), zap.String("subtype", item.Subtype), zap.String("date", item.Date), zap.Int("rows", len(values)))
	case domain.DataDownloadEventAlarm:
		values, err := repositories.QueryEventOrAlarmListOnTimeRange(item.SiteSN, startQuery, endQuery, repositories.EventAndAlarm)
		if err != nil {
			zap.L().Warn("[DataDownload] EventAlarm 上游查询失败", zap.String("sn", item.SiteSN), zap.String("date", item.Date), zap.String("err", err.Error()))
			return err
		}
		for i := range values {
			occurred := parseRemoteTime(values[i].Time)
			rows = append(rows, pgsqlRepo.DataDownloadRow{OccurredAt: occurred, Payload: values[i]})
		}
		zap.L().Debug("[DataDownload] EventAlarm 上游返回", zap.String("sn", item.SiteSN), zap.String("date", item.Date), zap.Int("rows", len(values)))
	case domain.DataDownloadUpdateLog:
		system, err := repositories.QuerySystemUpdateLog(item.SiteSN)
		if err != nil {
			zap.L().Warn("[DataDownload] UpdateLog(system) 上游查询失败", zap.String("sn", item.SiteSN), zap.String("err", err.Error()))
			return err
		}
		for i := range system {
			occurred := parseRemoteTime(system[i].UpgradeTime)
			rows = append(rows, pgsqlRepo.DataDownloadRow{OccurredAt: occurred, DeviceSN: item.SiteSN, DeviceType: "gateway", Payload: system[i]})
			details, detailErr := repositories.QuerySystemUpdateLogDetail(system[i].Id, item.SiteSN)
			if detailErr != nil {
				zap.L().Warn("[DataDownload] UpdateLog(detail) 上游查询失败", zap.String("sn", item.SiteSN), zap.Int64("orderId", system[i].Id), zap.String("err", detailErr.Error()))
				return detailErr
			}
			for j := range details {
				rows = append(rows, pgsqlRepo.DataDownloadRow{DeviceSN: details[j].GatewayId, DeviceType: "system_detail", Payload: details[j]})
			}
		}
		ahub, err := repositories.QueryAHubUpdateLog(item.SiteSN)
		if err != nil {
			zap.L().Warn("[DataDownload] UpdateLog(ahub) 上游查询失败", zap.String("sn", item.SiteSN), zap.String("err", err.Error()))
			return err
		}
		for i := range ahub {
			rows = append(rows, pgsqlRepo.DataDownloadRow{DeviceSN: ahub[i].AHubSn, DeviceType: "ahub", Payload: ahub[i]})
		}
		zap.L().Debug("[DataDownload] UpdateLog 上游返回", zap.String("sn", item.SiteSN), zap.Int("systemOrders", len(system)), zap.Int("ahub", len(ahub)), zap.Int("totalRows", len(rows)))
	default:
		return fmt.Errorf("binary kind %s requires its dedicated collector", item.Kind)
	}
	contentLock, err := cache.Lock(ctx, "data-download:content:"+documentID, 10*time.Minute, 200*time.Millisecond, 30*time.Second)
	if err != nil {
		return err
	}
	zap.L().Debug("[DataDownload] 结构化 fetch 完成，准备提交",
		zap.String("documentId", documentID),
		zap.String("kind", string(item.Kind)),
		zap.String("sn", item.SiteSN),
		zap.String("date", item.Date),
		zap.Int("rowCount", len(rows)),
	)
	defer contentLock.SafeRelease(context.Background())
	if err := lease.validate(ctx); err != nil {
		return err
	}
	// DD-002 方向 A：先失效旧渲染缓存，再提交 PG 权威内容，最后回填 Mongo Document。
	// 断开 RenderSlot 后 currentArtifact 会因 slot.ArtifactID==nil 判定 miss，从而强制
	// 后续读者从 PG 的最新 generation 重新渲染。这样即使在「PG 已提交、Document 尚未回填」
	// 之间崩溃，读取也只会 miss 重渲染（读到 PG 的新代，正确），而不会命中指向旧代的缓存。
	if err := mongoRepo.DetachDataDownloadRenderSlots(ctx, documentID); err != nil {
		return err
	}
	state, err := pgsqlRepo.ReplaceDataDownloadRows(documentID, item.Kind, item.SiteSN, item.Date, item.Timezone, rows, fetchedAt)
	if err != nil {
		return err
	}
	documentState := domain.DataDownloadMutable
	var frozenAt *time.Time
	if item.Kind == domain.DataDownloadUpdateLog {
		documentState = domain.DataDownloadDynamic
	} else if !start.IsZero() && !time.Now().In(loc).Before(start.AddDate(0, 0, 2)) {
		now := time.Now().UTC()
		frozenAt = &now
		documentState = domain.DataDownloadFrozen
	}
	createdAt := fetchedAt
	if existing, err := mongoRepo.GetDataDownloadDocument(ctx, documentID); err == nil {
		createdAt = existing.CreatedAt
	}
	document := &domain.DataDownloadDocument{ID: documentID, RequestKey: requestKey, Kind: item.Kind, SiteSN: item.SiteSN, DeviceSN: item.DeviceSN, Date: item.Date, Subtype: item.Subtype, DeviceType: item.DeviceType, Timezone: item.Timezone, State: documentState, Empty: len(rows) == 0, ContentGeneration: state.ContentGeneration, Checksum: state.Checksum, SourceFetchedAt: fetchedAt, FrozenAt: frozenAt, CreatedAt: createdAt, UpdatedAt: fetchedAt}
	if err := mongoRepo.UpsertDataDownloadDocument(ctx, document); err != nil {
		return err
	}
	zap.L().Debug("[DataDownload] 结构化内容已提交",
		zap.String("documentId", documentID),
		zap.String("kind", string(item.Kind)),
		zap.String("state", string(documentState)),
		zap.Int64("generation", state.ContentGeneration),
		zap.Bool("empty", len(rows) == 0),
	)
	if err := lease.complete(ctx); err != nil {
		return err
	}
	return linkDataDownloadItem(ctx, item, document, len(rows) == 0)
}

// stripTimeLocation 把时间按字面 年/月/日/时/分/秒/纳秒 原样重建为 UTC 时间。
// 用途：传给上游日志查询(QueryMqttLogListOnTimeRange / QueryEventOrAlarmListOnTimeRange)前，
// 剥离本地时区。上游按天拆分时用 time.Truncate(24h)(按 UTC 零点对齐)，若传入带非 UTC 时区
// 的时间，游标推进会回退到前一天导致 periods 无限增长并 OOM。剥离后与 BatchDownload 行为一致，
// 而查询仅把时间 Format 成字符串，字面值不变、查询语义不变。
func stripTimeLocation(t time.Time) time.Time {
	if t.IsZero() {
		return t
	}
	return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), time.UTC)
}

func parseRemoteTime(value string) *time.Time {
	for _, layout := range []string{time.RFC3339Nano, time.DateTime} {
		if parsed, err := time.Parse(layout, value); err == nil {
			return &parsed
		}
	}
	return nil
}