FranklinWH-Toolbox/service/services/faultRecordServices.go

package services

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/dto"
	faultRecordFile "fwh-toolbox-server/function/FaultRecordFile"
	proto "fwh-toolbox-server/function/FaultRecordProtocol"
	"fwh-toolbox-server/repositories/api/fwh_pro"
	"fwh-toolbox-server/repositories/api/fwh_test"
	minioRepo "fwh-toolbox-server/repositories/minio"
	mongoRepo "fwh-toolbox-server/repositories/mongo"
	"fwh-toolbox-server/repositories/mqtt/fwh_mqtt"
	"fwh-toolbox-server/task/queue/faultRecordQueue"

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

var ErrFaultRecordInvalid = errors.New("invalid request")

func faultRecordInvalid(format string, a ...any) error {
	return fmt.Errorf("%w: %s", ErrFaultRecordInvalid, fmt.Sprintf(format, a...))
}

/************** 设备列表 **************/

// GetFaultRecordSiteDevices 按环境查询站点信息，返回 aGate 及各 aPower（pe）。
// 只调用 Repository，不写站点信息表（测试服数据不能写入）。
func GetFaultRecordSiteDevices(envStr, gatewaySn string) (*dto.FaultRecordSiteDevices, error) {
	env, err := fwh_mqtt.ParseEnv(envStr)
	if err != nil {
		return nil, faultRecordInvalid("%v", err)
	}
	gatewaySn = strings.TrimSpace(gatewaySn)
	if gatewaySn == "" {
		return nil, faultRecordInvalid("gatewaySn is empty")
	}

	para := dto.SiteInfoPara{SN: &gatewaySn, PageSize: 10, PageNum: 1}
	var res *domain.FwhPageResponse[[]domain.FwhSiteInfo]
	if env == fwh_mqtt.EnvTest {
		res, err = fwh_test.QuerySiteInfoList(para)
	} else {
		res, err = fwh_pro.QuerySiteInfoList(para)
	}
	if err != nil {
		return nil, err
	}
	if res.Result == nil || len(*res.Result) == 0 {
		return nil, fmt.Errorf("站点不存在（%s）", gatewaySn)
	}
	var site *domain.FwhSiteInfo
	for i := range *res.Result {
		if (*res.Result)[i].Id == gatewaySn {
			site = &(*res.Result)[i]
		}
	}
	if site == nil {
		return nil, fmt.Errorf("站点不存在（%s）", gatewaySn)
	}

	out := &dto.FaultRecordSiteDevices{
		Env:       string(env),
		GatewaySn: site.Id,
		Online:    site.Status == 1,
		Devices:   []dto.FaultRecordDevice{{Type: proto.DeviceTypeAGate, Sn: site.Id}},
	}
	if site.Name != nil {
		out.SiteName = *site.Name
	}
	if site.OnLineTime != nil {
		out.OnlineTime = *site.OnLineTime
	}
	for _, sn := range strings.Split(site.FhpSn, ",") {
		if sn = strings.TrimSpace(sn); sn != "" {
			out.Devices = append(out.Devices, dto.FaultRecordDevice{Type: proto.DeviceTypePE, Sn: sn}) // aPower SN 即 pe SN
		}
	}
	return out, nil
}

/************** 创建／取消 **************/

// CreateFaultRecordOperation 校验并创建操作，提交异步执行
func CreateFaultRecordOperation(ctx context.Context, req dto.FaultRecordCreateOpReq, operator string) (string, error) {
	env, err := fwh_mqtt.ParseEnv(req.Env)
	if err != nil {
		return "", faultRecordInvalid("%v", err)
	}
	if _, err := fwh_mqtt.GetClient(env); err != nil {
		return "", fmt.Errorf("%s 环境 MQTT 不可用：%w", env, err)
	}
	gatewaySn := strings.TrimSpace(req.GatewaySn)
	if gatewaySn == "" {
		return "", faultRecordInvalid("gatewaySn is empty")
	}
	deviceType, err := proto.NormalizeDeviceType(req.DeviceType)
	if err != nil {
		return "", faultRecordInvalid("%v", err)
	}
	deviceSn := strings.TrimSpace(req.DeviceSn)
	if deviceType == proto.DeviceTypeAGate && deviceSn == "" {
		deviceSn = gatewaySn
	}

	kind := domain.FaultRecordOpKind(req.Kind)
	op := &domain.FaultRecordOperation{
		Env:        string(env),
		GatewaySn:  gatewaySn,
		DeviceType: deviceType,
		DeviceSn:   deviceSn,
		Kind:       kind,
		Cmd:        kind.Cmd(),
		Operator:   operator,
		Status:     domain.FaultRecordStatusPending,
		Request: domain.FaultRecordOpRequest{
			CodeList:      req.CodeList,
			RecordIndexes: recordIndexesOf(req),
			IdList:        req.IdList,
			ParaList:      req.ParaList,
		},
	}
	if op.Cmd < 0 {
		return "", faultRecordInvalid("unknown kind %s", req.Kind)
	}
	// 与执行器使用同一构造函数校验参数，尽早返回错误
	if _, err := faultRecordQueue.BuildRequest(op); err != nil {
		return "", faultRecordInvalid("%v", err)
	}

	if err := mongoRepo.InsertFaultRecordOp(ctx, op); err != nil {
		return "", err
	}
	if err := faultRecordQueue.Enqueue(op.ID); err != nil {
		_, _ = mongoRepo.UpdateFaultRecordOpStatus(ctx, op.ID, domain.FaultRecordStatusError, map[string]any{"error": "提交任务失败：" + err.Error()})
		return "", err
	}
	return op.ID.Hex(), nil
}

// CancelFaultRecordOperation 请求取消；执行器在 1 秒内响应
func CancelFaultRecordOperation(ctx context.Context, idStr string) error {
	id, err := primitive.ObjectIDFromHex(idStr)
	if err != nil {
		return faultRecordInvalid("invalid id")
	}
	ok, err := mongoRepo.RequestFaultRecordOpCancel(ctx, id)
	if err != nil {
		return err
	}
	if !ok {
		return faultRecordInvalid("操作已结束，无法取消")
	}
	return nil
}

// RetryFaultRecordDownload 上传已完成但下载失败时，手动重新下载（按 15 秒间隔重试至总超时）
func RetryFaultRecordDownload(ctx context.Context, idStr, operator string) error {
	op, err := getFaultRecordOp(ctx, idStr)
	if err != nil {
		return err
	}
	if !op.CanRetryDownload() {
		return faultRecordInvalid("当前状态不能重试下载（需上传已完成且下载失败或超时）")
	}
	ok, err := mongoRepo.ReopenFaultRecordOpForDownloadRetry(ctx, op.ID)
	if err != nil {
		return err
	}
	if !ok {
		return faultRecordInvalid("操作状态已变化，请刷新后再试")
	}
	// 在原始报文中记录本次手动重试（先于提交任务，保证执行器读到的报文条数包含它）
	attempt := op.DownloadRetryCount + 1
	raw, _ := json.Marshal(map[string]any{
		"action":     "retry_download",
		"operator":   operator,
		"attempt":    attempt,
		"prevStatus": op.Status,
		"prevError":  op.Error,
	})
	if err := mongoRepo.AppendFaultRecordOpMessage(ctx, op.ID, domain.FaultRecordOpMessage{
		Direction: "action", Topic: "手动重试下载", At: time.Now(), Raw: string(raw),
	}); err != nil {
		zap.L().Warn("fault record append retry action failed", zap.String("id", op.ID.Hex()), zap.Error(err))
	}
	if err := faultRecordQueue.EnqueueDownloadRetry(op.ID, attempt); err != nil {
		_, _ = mongoRepo.UpdateFaultRecordOpStatus(ctx, op.ID, domain.FaultRecordStatusDownloadFailed, map[string]any{"error": "提交重试下载失败：" + err.Error()})
		return err
	}
	return nil
}

/************** 查询 **************/

func ListFaultRecordOperations(ctx context.Context, f mongoRepo.FaultRecordOpFilter, page, size int64) (*dto.ResPage[dto.FaultRecordOpSummary], error) {
	list, total, err := mongoRepo.ListFaultRecordOps(ctx, f, page, size)
	if err != nil {
		return nil, err
	}
	out := &dto.ResPage[dto.FaultRecordOpSummary]{
		Data:     make([]dto.FaultRecordOpSummary, 0, len(list)),
		PageNum:  uint(page),
		PageSize: uint(size),
		Total:    uint(total),
	}
	for i := range list {
		out.Data = append(out.Data, toFaultRecordSummary(&list[i]))
	}
	return out, nil
}

func GetFaultRecordOperation(ctx context.Context, idStr string) (*dto.FaultRecordOpDetail, error) {
	op, err := getFaultRecordOp(ctx, idStr)
	if err != nil {
		return nil, err
	}
	d := &dto.FaultRecordOpDetail{
		FaultRecordOpSummary: toFaultRecordSummary(op),
		Request:              op.Request,
		Items:                []dto.FaultRecordResultItem{},
		Messages:             make([]dto.FaultRecordOpMessageView, 0, len(op.Messages)),
	}
	for _, m := range op.Messages {
		d.Messages = append(d.Messages, dto.FaultRecordOpMessageView(m))
	}
	if op.Reply != nil {
		d.ReplyResult = op.Reply.Result
		d.ReplyDesc = proto.DescribeReplyResult(op.Reply.Result)
		if info, ok := op.Reply.FindInfo(op.DeviceType, op.DeviceSn); ok {
			d.Items = faultRecordResultItems(op, info)
		}
	}
	return d, nil
}

// GetFaultRecordRecord 读取并解析已保存的录波文件
func GetFaultRecordRecord(ctx context.Context, idStr string) (*dto.FaultRecordRecordView, error) {
	op, data, err := readFaultRecordFile(ctx, idStr)
	if err != nil {
		return nil, err
	}
	records, err := faultRecordFile.ParseArchive(data, op.DeviceType)
	if err != nil {
		return nil, fmt.Errorf("解析录波文件失败：%w", err)
	}
	return &dto.FaultRecordRecordView{OperationID: op.ID.Hex(), Records: records}, nil
}

// ExportFaultRecordFile 导出原始压缩包或 CSV（zip），返回文件名和内容
func ExportFaultRecordFile(ctx context.Context, idStr, format string) (string, []byte, error) {
	op, data, err := readFaultRecordFile(ctx, idStr)
	if err != nil {
		return "", nil, err
	}
	switch format {
	case "", "raw":
		return proto.UploadFileName(op.FileName), data, nil
	case "csv":
		records, err := faultRecordFile.ParseArchive(data, op.DeviceType)
		if err != nil {
			return "", nil, fmt.Errorf("解析录波文件失败：%w", err)
		}
		zipData, err := faultRecordFile.BuildCSVZip(records)
		if err != nil {
			return "", nil, err
		}
		return proto.BatchFromUploadFileName(op.FileName) + "_csv.zip", zipData, nil
	}
	return "", nil, faultRecordInvalid("unknown format %s", format)
}

func recordIndexesOf(req dto.FaultRecordCreateOpReq) []int {
	if len(req.RecordIndexes) > 0 {
		return req.RecordIndexes
	}
	if req.RecordIndex > 0 {
		return []int{req.RecordIndex}
	}
	return nil
}

func getFaultRecordOp(ctx context.Context, idStr string) (*domain.FaultRecordOperation, error) {
	id, err := primitive.ObjectIDFromHex(idStr)
	if err != nil {
		return nil, faultRecordInvalid("invalid id")
	}
	return mongoRepo.GetFaultRecordOp(ctx, id)
}

func readFaultRecordFile(ctx context.Context, idStr string) (*domain.FaultRecordOperation, []byte, error) {
	op, err := getFaultRecordOp(ctx, idStr)
	if err != nil {
		return nil, nil, err
	}
	if op.MinioObject == "" {
		return nil, nil, faultRecordInvalid("该操作没有已保存的录波文件")
	}
	data, err := minioRepo.ReadFaultRecordFile(ctx, op.MinioObject)
	if err != nil {
		return nil, nil, err
	}
	return op, data, nil
}

/************** 视图转换 **************/

func toFaultRecordSummary(op *domain.FaultRecordOperation) dto.FaultRecordOpSummary {
	s := dto.FaultRecordOpSummary{
		ID:               op.ID.Hex(),
		Env:              op.Env,
		GatewaySn:        op.GatewaySn,
		DeviceType:       op.DeviceType,
		DeviceSn:         op.DeviceSn,
		Kind:             string(op.Kind),
		Summary:          faultRecordRequestSummary(op),
		Operator:         op.Operator,
		Status:           string(op.Status),
		TimeoutStage:     string(op.TimeoutStage),
		Error:            op.Error,
		StageTimes:       op.StageTimes,
		FileName:         op.FileName,
		FaultRecordID:    op.FaultRecordID,
		CanCancel:        !op.Status.Terminal() && !op.CancelReq,
		CanRetryDownload: op.CanRetryDownload(),
		HasFile:          op.MinioObject != "",
		CreatedAt:        op.CreatedAt,
		UpdatedAt:        op.UpdatedAt,
	}
	return s
}

func joinInts(list []int) string {
	parts := make([]string, len(list))
	for i, v := range list {
		parts[i] = fmt.Sprint(v)
	}
	return strings.Join(parts, ",")
}

func faultRecordRequestSummary(op *domain.FaultRecordOperation) string {
	r := op.Request
	switch op.Kind {
	case domain.FaultRecordOpDeviceInfo:
		return "code=" + joinInts(r.CodeList)
	case domain.FaultRecordOpRecordFetch:
		return "记录序号 " + joinInts(r.Indexes())
	case domain.FaultRecordOpParamRead:
		return "功能码 " + joinInts(r.IdList)
	case domain.FaultRecordOpParamWrite:
		parts := make([]string, len(r.ParaList))
		for i, p := range r.ParaList {
			parts[i] = fmt.Sprintf("%d=%s", p.Id, p.SetVal)
		}
		return strings.Join(parts, "; ")
	}
	return ""
}

func faultRecordResultItems(op *domain.FaultRecordOperation, info *proto.ReplyInfo) []dto.FaultRecordResultItem {
	items := []dto.FaultRecordResultItem{}
	result := func(it *dto.FaultRecordResultItem, resultOne string) {
		it.ResultOne = resultOne
		it.ResultDesc = proto.DescribeResultOne(resultOne)
		it.Ok = resultOne == proto.ResultOk
	}
	switch op.Kind {
	case domain.FaultRecordOpDeviceInfo:
		for _, h := range info.HexList {
			it := dto.FaultRecordResultItem{Code: h.Code, Name: proto.RealtimeCodeNames[h.Code], Raw: h.HexData}
			result(&it, h.ResultOne)
			if it.Ok && h.HexData != "" {
				if v, err := faultRecordFile.ParseRealtime(h.Code, h.HexData); err != nil {
					it.DecodeErr = err.Error()
				} else {
					it.Decoded = v
				}
			}
			items = append(items, it)
		}
	case domain.FaultRecordOpRecordFetch:
		// 每条记录展开为 4（故障数据）、5（波形数据）两项，按请求顺序对应（待实测确认）
		indexes := op.Request.Indexes()
		for i, x := range info.IndexInfo {
			name := map[int]string{4: "故障数据", 5: "波形数据"}[x.Index]
			if len(indexes) > 1 && len(info.IndexInfo) == 2*len(indexes) {
				name = fmt.Sprintf("记录 %d %s", indexes[i/2], name)
			}
			it := dto.FaultRecordResultItem{Code: x.Index, Name: name}
			result(&it, x.ResultOne)
			items = append(items, it)
		}
	case domain.FaultRecordOpParamRead:
		for _, x := range info.IdInfo {
			it := dto.FaultRecordResultItem{Id: x.Id, Name: proto.FunctionCodeNames[x.Id], Raw: x.IdData}
			result(&it, x.ResultOne)
			if it.Ok {
				it.Decoded = proto.DecodeIdData(x.Id, x.IdData)
			}
			items = append(items, it)
		}
	case domain.FaultRecordOpParamWrite:
		// 同一功能码可重复（如多个 id=6），按请求顺序对应（待实测确认）
		for i, x := range info.SetResultList {
			it := dto.FaultRecordResultItem{Id: x.Id, Name: proto.FunctionCodeNames[x.Id]}
			if i < len(op.Request.ParaList) && op.Request.ParaList[i].Id == x.Id {
				it.SetVal = op.Request.ParaList[i].SetVal
			}
			result(&it, x.ResultOne)
			items = append(items, it)
		}
	}
	return items
}