FranklinWH-Toolbox/service/task/queue/faultRecordQueue/executor.go

package faultRecordQueue

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

	"fwh-toolbox-server/domain"
	faultRecordFile "fwh-toolbox-server/function/FaultRecordFile"
	proto "fwh-toolbox-server/function/FaultRecordProtocol"
	minioRepo "fwh-toolbox-server/repositories/minio"
	"fwh-toolbox-server/repositories/mqtt/fwh_mqtt"

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

// BuildRequest 按操作构造 377 的 dataArea
func BuildRequest(op *domain.FaultRecordOperation) (proto.Request, error) {
	switch op.Kind {
	case domain.FaultRecordOpDeviceInfo:
		return proto.NewRealtimeRequest(op.DeviceType, op.DeviceSn, op.Request.CodeList)
	case domain.FaultRecordOpRecordFetch:
		return proto.NewGetRecordRequest(op.DeviceType, op.DeviceSn, op.Request.Indexes())
	case domain.FaultRecordOpParamRead:
		return proto.NewQueryParamRequest(op.DeviceType, op.DeviceSn, op.Request.IdList)
	case domain.FaultRecordOpParamWrite:
		return proto.NewSetParamRequest(op.DeviceType, op.DeviceSn, op.Request.ParaList)
	}
	return proto.Request{}, fmt.Errorf("unknown operation kind %s", op.Kind)
}

func lockKey(op *domain.FaultRecordOperation) string {
	return fmt.Sprintf("fault-record:gateway:%s:%s", op.Env, op.GatewaySn)
}

// 需要记录到操作中的接收报文类型
var recordedCmdTypes = map[uint]bool{
	proto.CmdTypeAck: true, proto.CmdTypeReply: true, proto.CmdTypeReplyAck: true,
	proto.CmdTypeUploadCall: true, proto.CmdTypeUploadAck: true,
	proto.CmdTypeUploadDone: true, proto.CmdTypeUploadDoneA: true,
}

// execution 一次操作的执行过程
type execution struct {
	d        deps
	op       *domain.FaultRecordOperation
	bus      mqttBus
	events   chan fwh_mqtt.Message
	batch    string // 379.filename
	msgCount int    // 已记录的报文条数（本执行器是该操作报文的唯一写入方）
	lastFail *downloadFailure
}

// downloadFailure 最近一条下载失败记录；连续相同的失败只更新这一条并累计次数
type downloadFailure struct {
	index   int
	key     string
	count   int
	firstAt time.Time
}

func (e *execution) s2cTopic() string { return fwh_mqtt.S2CTopic("mng/" + e.op.GatewaySn) }
func (e *execution) c2sTopic() string { return fwh_mqtt.C2STopic("mng/" + e.op.GatewaySn) }

// 文件上传消息 607～610 走 debug topic：607/610 为 s2c，608/609 为 c2s（用户确认，2026-10-19）
func (e *execution) uploadTopics() []string {
	return []string{fwh_mqtt.S2CTopic("debug/" + e.op.GatewaySn), fwh_mqtt.C2STopic("debug/" + e.op.GatewaySn)}
}

// run 执行或恢复一个操作。业务失败写入操作状态并返回 nil；只有存储故障或 ctx 取消才返回错误，交给 asynq 重试／重新投递。
func (d deps) run(ctx context.Context, id primitive.ObjectID) error {
	op, err := d.store.Get(ctx, id)
	if err != nil {
		return err
	}
	if op.Status.Terminal() {
		return nil
	}
	e := &execution{d: d, op: op, batch: proto.BatchFromUploadFileName(op.FileName), msgCount: len(op.Messages)}

	if op.CancelReq && op.Status == domain.FaultRecordStatusPending {
		return e.finish(ctx, domain.FaultRecordStatusCancelled, "", nil)
	}
	switch op.Status {
	case domain.FaultRecordStatusSent, domain.FaultRecordStatusAccepted:
		// 重发会清空设备导出文件，也无法确认设备当前状态，交由用户重新发起
		return e.finish(ctx, domain.FaultRecordStatusInterrupted, "服务重启时命令已发出但未收到结果，无法确认设备状态，请重新发起", nil)
	}

	release, ok, err := d.lock.Acquire(ctx, lockKey(op), op.ID.Hex(), d.timeouts.lockTTL)
	if err != nil {
		return err
	}
	if !ok {
		return e.finish(ctx, domain.FaultRecordStatusGatewayLocked, "该网关在本工具内有其他操作正在进行，请稍后再试", nil)
	}
	defer release()

	switch op.Status {
	case domain.FaultRecordStatusUploaded:
		if op.DownloadRetryAt != nil {
			// 手动重试下载：最多尝试 manualRetryAttempts 次
			return e.downloadUntilTimeout(ctx, 0, d.timeouts.manualRetryAttempts)
		}
		return e.downloadUntilTimeout(ctx, 0, 0)
	case domain.FaultRecordStatusDownloaded:
		data, err := d.files.Get(ctx, op.MinioObject)
		if err != nil {
			return e.finish(ctx, domain.FaultRecordStatusDownloadFailed, "读取已保存的文件失败："+err.Error(), nil)
		}
		return e.parse(ctx, data)
	}

	if e.bus, err = d.bus(op.Env); err != nil {
		return e.finish(ctx, domain.FaultRecordStatusError, "MQTT 不可用："+err.Error(), nil)
	}
	e.events = make(chan fwh_mqtt.Message, 128)
	handler := func(m fwh_mqtt.Message) {
		select {
		case e.events <- m:
		case <-ctx.Done():
		}
	}
	cancelC2S, err := e.bus.Subscribe(ctx, e.c2sTopic(), handler)
	if err != nil {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		return e.finish(ctx, domain.FaultRecordStatusError, "订阅设备上行 topic 失败："+err.Error(), nil)
	}
	defer cancelC2S()
	// 平台下发的 607/380/610 在 s2c 方向；订阅失败不影响主流程，只是看不到这些消息
	if cancelS2C, err := e.bus.Subscribe(ctx, e.s2cTopic(), handler); err == nil {
		defer cancelS2C()
	} else {
		zap.L().Warn("fault record subscribe s2c failed", zap.String("topic", e.s2cTopic()), zap.Error(err))
	}
	if op.Kind == domain.FaultRecordOpRecordFetch {
		for _, topic := range e.uploadTopics() {
			cancelUpload, err := e.bus.Subscribe(ctx, topic, handler)
			if err != nil {
				// 订阅失败时收不到 607～610，只能在超时后按规则直接下载
				zap.L().Warn("fault record subscribe upload topic failed", zap.String("topic", topic), zap.Error(err))
				continue
			}
			defer cancelUpload()
		}
	}
	// 结束前把已到达和稍后到达的报文（如 609 之后的 610）记录下来；先于取消订阅执行
	defer e.drain(ctx)

	switch op.Status {
	case domain.FaultRecordStatusPending:
		if done, err := e.send(ctx); done || err != nil {
			return err
		}
	case domain.FaultRecordStatusReady, domain.FaultRecordStatusUploading:
		// 恢复：文件名已知，先直接尝试下载，下载不到再继续等待 609
		if data, err := e.download(ctx); err == nil {
			return e.storeAndParse(ctx, data)
		}
	}
	return e.loop(ctx)
}

// send 发送 377；返回 done=true 表示操作已结束
func (e *execution) send(ctx context.Context) (bool, error) {
	req, err := BuildRequest(e.op)
	if err != nil {
		return true, e.finish(ctx, domain.FaultRecordStatusError, "构造请求失败："+err.Error(), nil)
	}
	if err := e.publish(ctx, req); err != nil {
		if ctx.Err() != nil {
			return true, ctx.Err()
		}
		return true, e.finish(ctx, domain.FaultRecordStatusError, "发送失败："+err.Error(), nil)
	}
	return false, e.advance(ctx, domain.FaultRecordStatusSent, nil)
}

func (e *execution) publish(ctx context.Context, req proto.Request) error {
	msg, err := fwh_mqtt.NewEnvelope(proto.CmdTypeRequest, e.op.GatewaySn, req)
	if err != nil {
		return err
	}
	if err := e.bus.Publish(ctx, e.s2cTopic(), msg); err != nil {
		return err
	}
	raw, _ := json.Marshal(msg)
	e.record(ctx, "send", e.s2cTopic(), proto.CmdTypeRequest, raw, time.Now())
	return nil
}

func (e *execution) loop(ctx context.Context) error {
	ticker := time.NewTicker(e.d.timeouts.cancelPoll)
	defer ticker.Stop()
	for {
		timer := time.NewTimer(time.Until(e.deadline()))
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case m := <-e.events:
			timer.Stop()
			if done, err := e.handle(ctx, m); done || err != nil {
				return err
			}
		case <-ticker.C:
			timer.Stop()
			if cur, err := e.d.store.Get(ctx, e.op.ID); err == nil && cur.CancelReq {
				return e.cancel(ctx)
			}
		case <-timer.C:
			return e.onTimeout(ctx)
		}
	}
}

func (e *execution) stageTime(s domain.FaultRecordOpStatus) time.Time {
	if t, ok := e.op.StageTimes[string(s)]; ok {
		return t
	}
	return time.Now()
}

func (e *execution) deadline() time.Time {
	t := e.d.timeouts
	switch e.op.Status {
	case domain.FaultRecordStatusSent:
		return e.stageTime(domain.FaultRecordStatusSent).Add(t.ack)
	case domain.FaultRecordStatusAccepted:
		if e.op.Kind == domain.FaultRecordOpRecordFetch {
			return e.stageTime(domain.FaultRecordStatusAccepted).Add(t.replyRecord + time.Duration(e.extraRecords())*t.replyPerRecord)
		}
		return e.stageTime(domain.FaultRecordStatusAccepted).Add(t.reply)
	default: // ready / uploading / uploaded
		start := e.stageTime(domain.FaultRecordStatusReady)
		if e.op.DownloadRetryAt != nil && e.op.Status == domain.FaultRecordStatusUploaded {
			start = *e.op.DownloadRetryAt // 手动重试下载：从重试时刻起重新计算总超时
		}
		return start.Add(t.upload + time.Duration(e.extraRecords())*t.uploadPerRecord)
	}
}

// extraRecords 第一条之外的记录数，用于按记录数延长等待
func (e *execution) extraRecords() int {
	if n := len(e.op.Request.Indexes()); n > 1 {
		return n - 1
	}
	return 0
}

func (e *execution) onTimeout(ctx context.Context) error {
	switch e.op.Status {
	case domain.FaultRecordStatusSent:
		return e.finishTimeout(ctx, domain.FaultRecordStatusAccepted, "设备未应答指令（378）")
	case domain.FaultRecordStatusAccepted:
		return e.finishTimeout(ctx, domain.FaultRecordStatusReady, "设备未上传结果（379）")
	default:
		// 609 可能漏收，最后再尝试一次下载
		if data, err := e.download(ctx); err == nil {
			return e.storeAndParse(ctx, data)
		}
		return e.finishTimeout(ctx, domain.FaultRecordStatusUploaded, "未收到文件上传结果（609），且文件无法下载")
	}
}

// handle 处理一条收到的消息；返回 done=true 表示操作已结束
func (e *execution) handle(ctx context.Context, m fwh_mqtt.Message) (bool, error) {
	cmdType, equipNo, da, err := proto.SplitEnvelope(m.Raw)
	if err != nil || (equipNo != "" && equipNo != e.op.GatewaySn) || !recordedCmdTypes[cmdType] {
		return false, nil
	}
	e.record(ctx, "recv", m.Topic, cmdType, m.Raw, m.ReceivedAt)

	switch cmdType {
	case proto.CmdTypeAck:
		ack, err := proto.ParseAck(da)
		if err != nil || ack.Cmd != e.op.Cmd || e.op.Status != domain.FaultRecordStatusSent {
			return false, nil
		}
		if !ack.Ok() {
			return true, e.finish(ctx, domain.FaultRecordStatusRejected, fmt.Sprintf("设备拒绝指令（378.result=%d）", ack.Result), nil)
		}
		return false, e.advance(ctx, domain.FaultRecordStatusAccepted, nil)

	case proto.CmdTypeReply:
		r, err := proto.ParseReply(da)
		if err != nil || r.Cmd != e.op.Cmd {
			return false, nil
		}
		if e.op.Status != domain.FaultRecordStatusSent && e.op.Status != domain.FaultRecordStatusAccepted {
			return false, nil
		}
		return e.handleReply(ctx, r)

	case proto.CmdTypeUploadCall:
		call, err := proto.ParseUploadCall(da)
		if err != nil || proto.BatchFromUploadFileName(call.FileName) != e.batch {
			return false, nil
		}
		switch e.op.Status {
		case domain.FaultRecordStatusReady:
			return false, e.advance(ctx, domain.FaultRecordStatusUploading, bson.M{"faultRecordId": call.FaultRecordID})
		case domain.FaultRecordStatusUploading, domain.FaultRecordStatusUploaded:
			// 608 可能先于 607 到达并已推进状态，此时只补记平台记录 ID
			return false, e.d.store.SetFields(ctx, e.op.ID, bson.M{"faultRecordId": call.FaultRecordID})
		}
		return false, nil

	case proto.CmdTypeUploadAck:
		// 608 在设备上行方向；订阅不到 607 时，靠它确认上传已开始并取得平台记录 ID
		ack, err := proto.ParseUploadAck(da)
		if err != nil || ack.Filename != e.batch || e.op.Status != domain.FaultRecordStatusReady {
			return false, nil
		}
		// 修复后的固件 608 不再带 id，此时不覆盖 607 给出的平台记录 ID
		extra := bson.M{}
		if ack.Id != 0 {
			extra["faultRecordId"] = ack.Id
		}
		return false, e.advance(ctx, domain.FaultRecordStatusUploading, extra)

	case proto.CmdTypeUploadDone:
		done, err := proto.ParseUploadDone(da)
		if err != nil || proto.BatchFromUploadFileName(done.FileName) != e.batch {
			return false, nil
		}
		if e.op.Status != domain.FaultRecordStatusReady && e.op.Status != domain.FaultRecordStatusUploading {
			return false, nil
		}
		if !done.Ok() {
			return true, e.finish(ctx, domain.FaultRecordStatusUploadFailed, fmt.Sprintf("设备上传文件失败（609.result=%d）", done.Result), nil)
		}
		if equipNo == "" {
			equipNo = e.op.GatewaySn
		}
		// 609 成功只推进状态，等平台应答 610 后再下载
		// 609.fileName 有的固件带 .tar.bz2、有的不带；统一补齐后缀，避免下载地址缺少后缀
		return false, e.advance(ctx, domain.FaultRecordStatusUploaded, bson.M{"fileName": proto.UploadFileName(done.FileName), "uploadEquipNo": equipNo})

	case proto.CmdTypeUploadDoneA:
		// 610 没有文件名，按“本操作收到 609 之后的下一条 610”对应；result 含义未确认，只记录不判断
		if e.op.Status != domain.FaultRecordStatusUploaded {
			return false, nil
		}
		return true, e.downloadUntilTimeout(ctx, e.d.timeouts.afterUploadAck, 0)
	}
	return false, nil
}

func (e *execution) handleReply(ctx context.Context, r proto.Reply) (bool, error) {
	if r.Busy() {
		return true, e.finish(ctx, domain.FaultRecordStatusBusy, "设备正在处理其他命令（可能来自 FWH 平台），请稍后再试", bson.M{"reply": r})
	}
	if r.IsErrorReply() {
		return true, e.finish(ctx, domain.FaultRecordStatusRejected, "设备拒绝命令："+proto.DescribeReplyResult(r.Result), bson.M{"reply": r})
	}
	info, ok := r.FindInfo(e.op.DeviceType, e.op.DeviceSn)
	if !ok {
		return true, e.finish(ctx, domain.FaultRecordStatusRejected, "设备回复中没有目标设备", bson.M{"reply": r})
	}

	if e.op.Kind != domain.FaultRecordOpRecordFetch {
		status := domain.FaultRecordStatusDone
		if !r.Ok() {
			status = domain.FaultRecordStatusItemFailed
		}
		return true, e.finish(ctx, status, "", bson.M{"reply": r})
	}

	allOk := r.Ok() && len(info.IndexInfo) > 0
	for _, it := range info.IndexInfo {
		if it.ResultOne != proto.ResultOk {
			allOk = false
		}
	}
	if !allOk {
		return true, e.finish(ctx, domain.FaultRecordStatusItemFailed, "设备读取录波记录失败", bson.M{"reply": r})
	}
	if r.Filename == "" {
		return true, e.finish(ctx, domain.FaultRecordStatusRejected, "设备回复中没有文件名", bson.M{"reply": r})
	}
	e.batch = r.Filename
	return false, e.advance(ctx, domain.FaultRecordStatusReady, bson.M{"reply": r, "fileName": proto.UploadFileName(r.Filename)})
}

// uploadFileName 上传文件名：由批次名统一拼接 .tar.bz2
func (e *execution) uploadFileName() string {
	return proto.UploadFileName(e.batch)
}

func (e *execution) uploadEquipNo() string {
	if e.op.UploadEquipNo != "" {
		return e.op.UploadEquipNo
	}
	return e.op.GatewaySn
}

// downloadUntilTimeout 等待 firstWait 后开始下载，失败后每隔 downloadInterval 再试，直到总超时；
// 等待期间响应取消请求
// maxAttempts 为 0 时不限次数（一直重试至总超时）
func (e *execution) downloadUntilTimeout(ctx context.Context, firstWait time.Duration, maxAttempts int) error {
	deadline := e.deadline()
	wait := firstWait
	attempts := 0
	for {
		if cancelled, err := e.sleepCancellable(ctx, wait); err != nil {
			return err
		} else if cancelled {
			return e.cancel(ctx)
		}
		data, err := e.download(ctx)
		attempts++
		if err == nil {
			return e.storeAndParse(ctx, data)
		}
		if maxAttempts > 0 && attempts >= maxAttempts {
			return e.finish(ctx, domain.FaultRecordStatusDownloadFailed,
				fmt.Sprintf("重试下载失败，已尝试 %d 次：%s", attempts, err.Error()), nil)
		}
		remaining := time.Until(deadline)
		if remaining <= 0 {
			return e.finish(ctx, domain.FaultRecordStatusDownloadFailed,
				fmt.Sprintf("下载文件失败，已尝试 %d 次直至总超时：%s", attempts, err.Error()), nil)
		}
		wait = min(e.d.timeouts.downloadInterval, remaining)
	}
}

// sleepCancellable 等待 d；期间检查取消请求。返回 cancelled=true 表示用户请求取消
func (e *execution) sleepCancellable(ctx context.Context, d time.Duration) (bool, error) {
	if d <= 0 {
		return false, nil
	}
	timer := time.NewTimer(d)
	defer timer.Stop()
	ticker := time.NewTicker(e.d.timeouts.cancelPoll)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return false, ctx.Err()
		case <-timer.C:
			return false, nil
		case <-ticker.C:
			if cur, err := e.d.store.Get(ctx, e.op.ID); err == nil && cur.CancelReq {
				return true, nil
			}
		}
	}
}

func (e *execution) storeAndParse(ctx context.Context, data []byte) error {
	key, err := minioRepo.FaultRecordObjectKey(e.op.Env, e.op.GatewaySn, e.uploadFileName())
	if err != nil {
		return e.finish(ctx, domain.FaultRecordStatusDownloadFailed, err.Error(), nil)
	}
	if err := e.d.files.Put(ctx, key, data); err != nil {
		return e.finish(ctx, domain.FaultRecordStatusDownloadFailed, "保存文件失败："+err.Error(), nil)
	}
	if err := e.advance(ctx, domain.FaultRecordStatusDownloaded, bson.M{"minioObject": key, "fileSize": len(data)}); err != nil {
		return err
	}
	return e.parse(ctx, data)
}

func (e *execution) parse(ctx context.Context, data []byte) error {
	records, err := faultRecordFile.ParseArchive(data, e.op.DeviceType)
	if err != nil {
		return e.finish(ctx, domain.FaultRecordStatusParseFailed, "文件已保存，但解析失败："+err.Error(), nil)
	}
	// 核对压缩包中是否包含全部请求的记录
	got := map[int]bool{}
	for _, r := range records {
		got[r.RecordIndex] = true
	}
	var missing []string
	for _, i := range e.op.Request.Indexes() {
		if !got[i] {
			missing = append(missing, strconv.Itoa(i))
		}
	}
	msg := ""
	if len(missing) > 0 {
		msg = "压缩包中缺少记录序号：" + strings.Join(missing, ",")
	}
	return e.finish(ctx, domain.FaultRecordStatusDone, msg, nil)
}

// download 下载上传的文件，并把每次下载作为一条 http 记录写入原始报文
func (e *execution) download(ctx context.Context) ([]byte, error) {
	equipNo := e.uploadEquipNo()
	// 每次下载都按规则重新拼接文件名（批次名 + .tar.bz2），不依赖已保存的文件名是否带后缀
	fileName := e.uploadFileName()
	entry := map[string]any{
		"env":  e.op.Env,
		"path": "/hes-upload/v1/file/download/logger/" + equipNo + "/" + fileName,
	}
	if e.d.downloadURL != nil {
		entry["url"] = e.d.downloadURL(e.op.Env, equipNo, fileName)
	}
	start := time.Now()
	data, err := e.d.download(e.op.Env, equipNo, fileName)
	now := time.Now()
	entry["durationMs"] = now.Sub(start).Milliseconds()
	topic := "GET " + entry["path"].(string)
	if err == nil {
		entry["result"] = "ok"
		entry["size"] = len(data)
		raw, _ := json.Marshal(entry)
		e.record(ctx, "http", topic, 0, raw, now)
		e.lastFail = nil
		return data, nil
	}

	entry["result"] = "failed"
	entry["error"] = err.Error()
	key := topic + "|" + err.Error()
	// 与上一条记录是同一种失败：只保留最后一条，累计次数
	if f := e.lastFail; f != nil && f.key == key && f.index == e.msgCount-1 {
		f.count++
		entry["count"] = f.count
		entry["firstAt"] = f.firstAt
		raw, _ := json.Marshal(entry)
		e.replaceMessage(ctx, f.index, domain.FaultRecordOpMessage{Direction: "http", Topic: topic, At: now, Raw: string(raw)})
		return nil, err
	}
	entry["count"] = 1
	entry["firstAt"] = now
	raw, _ := json.Marshal(entry)
	e.record(ctx, "http", topic, 0, raw, now)
	e.lastFail = &downloadFailure{index: e.msgCount - 1, key: key, count: 1, firstAt: now}
	return nil, err
}

// drain 记录结束时缓冲区中的报文，并在短时间内继续记录随后到达的报文
func (e *execution) drain(ctx context.Context) {
	if e.events == nil || ctx.Err() != nil {
		return
	}
	timer := time.NewTimer(e.d.timeouts.drain)
	defer timer.Stop()
	for {
		select {
		case m := <-e.events:
			if cmdType, equipNo, _, err := proto.SplitEnvelope(m.Raw); err == nil && recordedCmdTypes[cmdType] && (equipNo == "" || equipNo == e.op.GatewaySn) {
				e.record(ctx, "recv", m.Topic, cmdType, m.Raw, m.ReceivedAt)
			}
		case <-timer.C:
			return
		case <-ctx.Done():
			return
		}
	}
}

// cancel 设备正在处理本操作的命令时发送 cmd=4，其余阶段只停止等待
func (e *execution) cancel(ctx context.Context) error {
	extra := bson.M{}
	msg := ""
	if e.op.Status == domain.FaultRecordStatusSent || e.op.Status == domain.FaultRecordStatusAccepted {
		if err := e.publish(ctx, proto.NewExitRequest()); err != nil {
			msg = "发送退出命令失败：" + err.Error()
		} else {
			extra["exitSent"] = true
		}
	}
	return e.finish(ctx, domain.FaultRecordStatusCancelled, msg, extra)
}

func (e *execution) record(ctx context.Context, dir, topic string, cmdType uint, raw []byte, at time.Time) {
	err := e.d.store.AppendMessage(ctx, e.op.ID, domain.FaultRecordOpMessage{
		Direction: dir, Topic: topic, CmdType: cmdType, At: at, Raw: string(raw),
	})
	if err != nil {
		zap.L().Warn("fault record append message failed", zap.String("id", e.op.ID.Hex()), zap.Error(err))
		return
	}
	e.msgCount++
}

func (e *execution) replaceMessage(ctx context.Context, index int, msg domain.FaultRecordOpMessage) {
	if err := e.d.store.SetFields(ctx, e.op.ID, bson.M{fmt.Sprintf("messages.%d", index): msg}); err != nil {
		zap.L().Warn("fault record replace message failed", zap.String("id", e.op.ID.Hex()), zap.Error(err))
	}
}

func (e *execution) advance(ctx context.Context, status domain.FaultRecordOpStatus, extra bson.M) error {
	if _, err := e.d.store.UpdateStatus(ctx, e.op.ID, status, extra); err != nil {
		return err
	}
	e.op.Status = status
	if e.op.StageTimes == nil {
		e.op.StageTimes = map[string]time.Time{}
	}
	e.op.StageTimes[string(status)] = time.Now()
	if v, ok := extra["fileName"].(string); ok {
		e.op.FileName = v
	}
	if v, ok := extra["uploadEquipNo"].(string); ok {
		e.op.UploadEquipNo = v
	}
	return nil
}

func (e *execution) finish(ctx context.Context, status domain.FaultRecordOpStatus, msg string, extra bson.M) error {
	if extra == nil {
		extra = bson.M{}
	}
	if msg != "" {
		extra["error"] = msg
	}
	return e.advance(ctx, status, extra)
}

func (e *execution) finishTimeout(ctx context.Context, waiting domain.FaultRecordOpStatus, msg string) error {
	return e.finish(ctx, domain.FaultRecordStatusTimeout, msg, bson.M{"timeoutStage": waiting})
}