FranklinWH-Toolbox/service/repositories/mongo/faultRecordRepo.go

package mongo

import (
	"context"
	"errors"
	"strings"
	"time"

	"fwh-toolbox-server/database"
	"fwh-toolbox-server/domain"

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

const CollectionName_FaultRecordOperation = "FaultRecordOperation"

var ErrFaultRecordOpNotFound = errors.New("fault record operation not found")

func faultRecordOps() *mongo.Collection {
	return database.GetCollection(CollectionName_FaultRecordOperation)
}

func InitFaultRecordIndexes(ctx context.Context) error {
	_, err := faultRecordOps().Indexes().CreateMany(ctx, []mongo.IndexModel{
		{Keys: bson.D{{Key: "env", Value: 1}, {Key: "gatewaySn", Value: 1}, {Key: "createdAt", Value: -1}}},
		{Keys: bson.D{{Key: "status", Value: 1}}},
		{Keys: bson.D{{Key: "createdAt", Value: -1}}},
	})
	return err
}

func InsertFaultRecordOp(ctx context.Context, op *domain.FaultRecordOperation) error {
	now := time.Now()
	if op.ID.IsZero() {
		op.ID = primitive.NewObjectID()
	}
	op.CreatedAt, op.UpdatedAt = now, now
	if op.StageTimes == nil {
		op.StageTimes = map[string]time.Time{}
	}
	op.StageTimes[string(op.Status)] = now
	if op.Messages == nil {
		op.Messages = []domain.FaultRecordOpMessage{}
	}
	_, err := faultRecordOps().InsertOne(ctx, op)
	return err
}

func GetFaultRecordOp(ctx context.Context, id primitive.ObjectID) (*domain.FaultRecordOperation, error) {
	var op domain.FaultRecordOperation
	err := faultRecordOps().FindOne(ctx, bson.M{"_id": id}).Decode(&op)
	if errors.Is(err, mongo.ErrNoDocuments) {
		return nil, ErrFaultRecordOpNotFound
	}
	if err != nil {
		return nil, err
	}
	return &op, nil
}

// UpdateFaultRecordOpStatus 更新状态并记录阶段时间；extra 为同时写入的其他字段。
// 只更新尚未终止的操作，返回是否更新成功（已终止时返回 false）。
func UpdateFaultRecordOpStatus(ctx context.Context, id primitive.ObjectID, status domain.FaultRecordOpStatus, extra bson.M) (bool, error) {
	now := time.Now()
	set := bson.M{"status": status, "updatedAt": now, "stageTimes." + string(status): now}
	for k, v := range extra {
		set[k] = v
	}
	res, err := faultRecordOps().UpdateOne(ctx,
		bson.M{"_id": id, "status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}},
		bson.M{"$set": set})
	if err != nil {
		return false, err
	}
	return res.ModifiedCount > 0, nil
}

// SetFaultRecordOpFields 更新字段，不改变状态
func SetFaultRecordOpFields(ctx context.Context, id primitive.ObjectID, fields bson.M) error {
	fields["updatedAt"] = time.Now()
	_, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id}, bson.M{"$set": fields})
	return err
}

func AppendFaultRecordOpMessage(ctx context.Context, id primitive.ObjectID, msg domain.FaultRecordOpMessage) error {
	_, err := faultRecordOps().UpdateOne(ctx, bson.M{"_id": id},
		bson.M{"$push": bson.M{"messages": msg}, "$set": bson.M{"updatedAt": time.Now()}})
	return err
}

// RequestFaultRecordOpCancel 标记取消请求，仅对未终止的操作生效
func RequestFaultRecordOpCancel(ctx context.Context, id primitive.ObjectID) (bool, error) {
	res, err := faultRecordOps().UpdateOne(ctx,
		bson.M{"_id": id, "status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}},
		bson.M{"$set": bson.M{"cancelRequested": true, "updatedAt": time.Now()}})
	if err != nil {
		return false, err
	}
	return res.MatchedCount > 0, nil
}

// ReopenFaultRecordOpForDownloadRetry 将“上传已完成但下载失败／超时”的操作重新置为 uploaded，供重试下载。
// 条件在同一次更新中校验，避免并发重复触发；返回是否成功重开。
func ReopenFaultRecordOpForDownloadRetry(ctx context.Context, id primitive.ObjectID) (bool, error) {
	now := time.Now()
	res, err := faultRecordOps().UpdateOne(ctx,
		bson.M{
			"_id":                 id,
			"kind":                domain.FaultRecordOpRecordFetch,
			"status":              bson.M{"$in": []domain.FaultRecordOpStatus{domain.FaultRecordStatusDownloadFailed, domain.FaultRecordStatusTimeout}},
			"stageTimes.uploaded": bson.M{"$exists": true},
			"fileName":            bson.M{"$nin": []any{"", nil}},
		},
		bson.M{
			"$set":   bson.M{"status": domain.FaultRecordStatusUploaded, "downloadRetryAt": now, "cancelRequested": false, "updatedAt": now},
			"$unset": bson.M{"error": "", "timeoutStage": ""},
			"$inc":   bson.M{"downloadRetryCount": 1},
		})
	if err != nil {
		return false, err
	}
	return res.ModifiedCount > 0, nil
}

// FaultRecordOpFilter 操作历史查询条件
type FaultRecordOpFilter struct {
	Env       string
	GatewaySn string
	Kind      string // 多个用逗号分隔
}

// ListFaultRecordOps 按创建时间倒序分页，不返回报文原文
func ListFaultRecordOps(ctx context.Context, f FaultRecordOpFilter, page, size int64) ([]domain.FaultRecordOperation, int64, error) {
	filter := bson.M{}
	if f.Env != "" {
		filter["env"] = f.Env
	}
	if f.GatewaySn != "" {
		filter["gatewaySn"] = f.GatewaySn
	}
	if f.Kind != "" {
		if kinds := strings.Split(f.Kind, ","); len(kinds) > 1 {
			filter["kind"] = bson.M{"$in": kinds}
		} else {
			filter["kind"] = f.Kind
		}
	}
	total, err := faultRecordOps().CountDocuments(ctx, filter)
	if err != nil {
		return nil, 0, err
	}
	opts := options.Find().
		SetSort(bson.D{{Key: "createdAt", Value: -1}}).
		SetSkip((page - 1) * size).
		SetLimit(size).
		SetProjection(bson.M{"messages": 0})
	cur, err := faultRecordOps().Find(ctx, filter, opts)
	if err != nil {
		return nil, 0, err
	}
	var list []domain.FaultRecordOperation
	if err := cur.All(ctx, &list); err != nil {
		return nil, 0, err
	}
	return list, total, nil
}

// ListActiveFaultRecordOps 全部未终止的操作，用于启动时恢复
func ListActiveFaultRecordOps(ctx context.Context) ([]domain.FaultRecordOperation, error) {
	cur, err := faultRecordOps().Find(ctx,
		bson.M{"status": bson.M{"$nin": domain.FaultRecordTerminalStatuses()}},
		options.Find().SetProjection(bson.M{"messages": 0}))
	if err != nil {
		return nil, err
	}
	var list []domain.FaultRecordOperation
	err = cur.All(ctx, &list)
	return list, err
}