junhong_cmp_fiber/internal/task/carrier_stop_resume_executor.go

package task

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

	"go.uber.org/zap"

	"github.com/break/junhong_cmp_fiber/internal/gateway"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/gatewaylimit"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// StopResumeCommandService 是停复机指令执行器依赖的停复机服务能力。
type StopResumeCommandService interface {
	DecideCommand(ctx context.Context, cardID uint, scenes []string) (iot_card_svc.CommandDecision, error)
	ExecuteCommand(ctx context.Context, card *model.IotCard, action, reason string) (iot_card_svc.CommandCallResult, error)
	QueryCommandCardStatus(ctx context.Context, card *model.IotCard) (int, bool, error)
	ConfirmCommandState(ctx context.Context, card *model.IotCard, action, reason string) error
	EvaluateDeviceCommands(ctx context.Context, deviceID uint, scenes []string) error
	LegacyDeviceOpLocked(ctx context.Context, deviceID uint) bool
}

// stopResumeTaskTimeout 是一条停复机指令的总执行时长上限（一次判断 + 一次网关调用）。
const stopResumeTaskTimeout = 60 * time.Second

// stopResumeRateLimitedDelay 本账户限流拒绝或等待超时（请求未发出）后的重排延迟，不计失败。
const stopResumeRateLimitedDelay = 5 * time.Second

// StopResumeExecutor 停复机指令执行器：执行前按场景重新判断，只调用一次网关，按错误分类退避与告警，结果未知在队列内确认。
type StopResumeExecutor struct {
	store      *carrierqueue.Store
	settings   *carrierqueue.SettingsProvider
	params     func(carrierID uint) carrierqueue.CarrierParams
	alerts     *carrierqueue.AlertWriter
	service    StopResumeCommandService
	classifier *gatewaylimit.Classifier
	instanceID string
	logger     *zap.Logger
}

// NewStopResumeExecutor 创建停复机指令执行器。
func NewStopResumeExecutor(store *carrierqueue.Store, settings *carrierqueue.SettingsProvider, params func(uint) carrierqueue.CarrierParams,
	alerts *carrierqueue.AlertWriter, service StopResumeCommandService, classifier *gatewaylimit.Classifier, instanceID string, logger *zap.Logger) *StopResumeExecutor {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &StopResumeExecutor{store: store, settings: settings, params: params, alerts: alerts, service: service,
		classifier: classifier, instanceID: instanceID, logger: logger}
}

// srOutcome 是一次执行的收尾决定。
type srOutcome struct {
	stat   string // 计数字段覆盖；statSkip 表示不计数
	done   bool
	nextAt time.Time
	set    map[string]string
	del    []string
}

// statSkip 表示本次执行不计入监控计数（指令不再需要、卡已不存在）。
const statSkip = "-"

func srDone() srOutcome { return srOutcome{done: true} }

func srEnded() srOutcome { return srOutcome{done: true, stat: statSkip} }

func srRetry(at time.Time) srOutcome {
	return srOutcome{nextAt: at, set: map[string]string{}}
}

// Handle 消费一条停复机指令消息；返回后由消费者 FIN。重试与退避全部由 Redis score 表达。
func (e *StopResumeExecutor) Handle(ctx context.Context, body []byte, _ func()) {
	var msg carrierqueue.StopResumeMessage
	if err := json.Unmarshal(body, &msg); err != nil || msg.CardID == 0 || msg.Token == "" {
		e.logger.Warn("停复机指令消息无效，丢弃", zap.ByteString("body", body))
		return
	}
	execValue := msg.Token + "|" + e.instanceID
	got, err := e.store.AcquireStopResumeExec(ctx, msg.CardID, msg.Token, execValue, 2*stopResumeTaskTimeout)
	if err != nil || got != 1 {
		return
	}
	taskCtx, cancel := context.WithTimeout(ctx, stopResumeTaskTimeout)
	outcome := e.execute(taskCtx, msg)
	cancel()

	finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer finishCancel()
	e.countOutcome(finishCtx, msg.CarrierID, outcome)
	result, err := e.store.FinishStopResume(finishCtx, carrierqueue.SRFinish{
		CardID: msg.CardID, CarrierID: msg.CarrierID, Token: msg.Token, Version: msg.Version,
		Done: outcome.done, NextAt: outcome.nextAt, Set: outcome.set, Del: outcome.del,
	})
	switch {
	case err != nil:
		e.logger.Warn("停复机指令收尾失败，等待租约到期后重新领取", zap.Uint("card_id", msg.CardID), zap.Error(err))
	case result == 0:
		e.logger.Warn("停复机指令收尾时处理中标记已不一致", zap.Uint("card_id", msg.CardID), zap.String("metric", "stop_resume.finish_token_mismatch"))
	case result == 2:
		e.logger.Info("停复机指令执行期间已被新指令覆盖，保留新指令", zap.Uint("card_id", msg.CardID), zap.Int64("claimed_version", msg.Version))
	}
}

func (e *StopResumeExecutor) execute(ctx context.Context, msg carrierqueue.StopResumeMessage) (outcome srOutcome) {
	defer func() {
		if r := recover(); r != nil {
			e.logger.Error("停复机指令执行 panic，稍后重试", zap.Uint("card_id", msg.CardID), zap.Any("panic", r))
			outcome = srRetry(time.Now().Add(time.Minute))
		}
	}()
	cmd, err := e.store.ReadCommand(ctx, msg.CardID)
	if err != nil {
		return srRetry(time.Now().Add(time.Minute))
	}
	if !cmd.Exists {
		return srEnded()
	}
	decision, err := e.service.DecideCommand(ctx, msg.CardID, cmd.Scenes)
	if err != nil {
		e.logger.Warn("停复机指令执行前判断失败，稍后重试", zap.Uint("card_id", msg.CardID), zap.Error(err))
		return srRetry(time.Now().Add(time.Minute))
	}
	if decision.Gone {
		return srEnded()
	}
	if decision.Action != cmd.Action {
		// 条件不再满足或已是目标状态：结束指令，不调用上游。
		e.logger.Info("停复机指令执行前重新判断不再需要，结束指令", zap.Uint("card_id", msg.CardID),
			zap.String("action", cmd.Action), zap.String("decided", decision.Action), zap.Strings("scenes", cmd.Scenes))
		e.recovered(cmd, decision.Card)
		return srEnded()
	}
	card := decision.Card
	reason := decision.Reason
	params := e.params(card.CarrierID)
	settings := e.settings.Current(ctx)
	gatewayCtx := gateway.WithRateLimitWait(ctx, 20*time.Second)

	if !cmd.Pending {
		return e.call(gatewayCtx, card, cmd, reason, params, settings)
	}
	if confirmed, next := e.confirm(gatewayCtx, card, cmd, reason, params, settings); confirmed != nil {
		return *confirmed
	} else if next != nil {
		return *next
	}
	// 确认不是目标状态且已满足单卡最小间隔：清除待确认后重新调用。
	result := e.call(gatewayCtx, card, cmd, reason, params, settings)
	if !result.done && result.set[carrierqueue.SRFieldPending] != "1" {
		result.del = append(result.del, carrierqueue.SRFieldPending)
	}
	return result
}

// confirm 处理待确认指令：查一次卡状态。已是目标状态返回结束；查询失败保持待确认按退避重查；
// 不是目标状态时清除待确认，未到单卡最小间隔则推后，否则返回 (nil, nil) 由调用方重新发起停复机。
func (e *StopResumeExecutor) confirm(ctx context.Context, card *model.IotCard, cmd carrierqueue.CommandState, reason string,
	params carrierqueue.CarrierParams, settings carrierqueue.Settings) (*srOutcome, *srOutcome) {
	status, known, err := e.service.QueryCommandCardStatus(ctx, card)
	if gateway.IsRateLimited(err) {
		retry := srRetry(time.Now().Add(stopResumeRateLimitedDelay))
		return nil, &retry
	}
	if err != nil || !known {
		retry := srRetry(time.Now().Add(settings.SRBackoffFor(cmd.Unknown)))
		if err != nil {
			retry.set[carrierqueue.SRFieldError] = truncateRunes(constants.StopResumeFailureUnknown+"|确认查询失败: "+err.Error(), 300)
		}
		return nil, &retry
	}
	if status == targetNetworkStatus(cmd.Action) {
		if confirmErr := e.service.ConfirmCommandState(ctx, card, cmd.Action, reason); confirmErr != nil {
			e.logger.Warn("确认已是目标状态但回写失败，稍后再确认", zap.Uint("card_id", card.ID), zap.Error(confirmErr))
			retry := srRetry(time.Now().Add(time.Minute))
			return nil, &retry
		}
		e.logger.Info("停复机结果未知，确认查询已是目标状态，按成功结束", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action))
		e.recovered(cmd, card)
		done := srDone()
		return &done, nil
	}
	last, _ := e.store.LastStopResumeCall(ctx, card.ID)
	if allowed := last.Add(params.SRMinInterval); !last.IsZero() && allowed.After(time.Now()) {
		retry := srRetry(allowed)
		retry.del = []string{carrierqueue.SRFieldPending}
		return nil, &retry
	}
	return nil, nil
}

// call 调用一次停机或复机并按结果分类处理。
func (e *StopResumeExecutor) call(ctx context.Context, card *model.IotCard, cmd carrierqueue.CommandState, reason string,
	params carrierqueue.CarrierParams, settings carrierqueue.Settings) srOutcome {
	previousCall, _ := e.store.LastStopResumeCall(ctx, card.ID)
	now := time.Now()
	if err := e.store.MarkStopResumeCall(ctx, card.ID, now, 2*params.SRMinInterval); err != nil {
		return srRetry(now.Add(time.Minute))
	}
	result, err := e.service.ExecuteCommand(ctx, card, cmd.Action, reason)
	if err == nil {
		e.logger.Info("停复机指令执行成功", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action), zap.String("reason", reason),
			zap.Strings("scenes", cmd.Scenes), zap.String("metric", "stop_resume.success"))
		e.recovered(cmd, card)
		return srDone()
	}
	if gateway.IsRateLimited(err) {
		// 本账户限流：请求未发出，恢复最近调用时间，短延迟重排，不计失败。
		e.restoreLastCall(ctx, card.ID, previousCall, params)
		return srRetry(time.Now().Add(stopResumeRateLimitedDelay))
	}
	classification := e.classify(ctx, cmd.Action, err)
	if result.Unknown || classification.MaybeAccepted {
		return e.unknown(card, cmd, err, params, settings)
	}
	message := truncateRunes(classification.Category+"|"+err.Error(), 300)
	switch classification.Category {
	case constants.GatewayClassAlreadyTarget:
		if confirmErr := e.service.ConfirmCommandState(ctx, card, cmd.Action, reason); confirmErr != nil {
			return e.unknown(card, cmd, confirmErr, params, settings)
		}
		e.recovered(cmd, card)
		return srDone()
	case constants.GatewayClassTerminal:
		// 终态不可操作：不重试、结束指令、写告警；停机方向上游已不可用时按目标状态视为已停。
		if cmd.Action == constants.StopResumeActionStop {
			if confirmErr := e.service.ConfirmCommandState(ctx, card, cmd.Action, reason); confirmErr != nil {
				e.logger.Warn("停机遇上游终态，回写已停失败", zap.Uint("card_id", card.ID), zap.Error(confirmErr))
			}
		}
		e.reportFailure(card, cmd, classification.Category, err.Error(), cmd.FailCount+1, now)
		return srOutcome{done: true, stat: constants.StopResumeStatFailed}
	case constants.GatewayClassRateLimitedUpstream:
		retry := srRetry(now.Add(maxDuration(params.SRMinInterval, settings.SRBackoffFor(1))))
		retry.set[carrierqueue.SRFieldError] = message
		return retry
	}
	failCount := cmd.FailCount + 1
	retry := srRetry(now.Add(settings.SRBackoffFor(failCount)))
	retry.set[carrierqueue.SRFieldFail] = strconv.Itoa(failCount)
	retry.set[carrierqueue.SRFieldError] = message
	firstFailed := cmd.FirstFailed
	if firstFailed.IsZero() {
		firstFailed = now
		retry.set[carrierqueue.SRFieldFirst] = strconv.FormatInt(now.UnixMilli(), 10)
	}
	if failCount >= settings.SRAlertFailCount || settings.ImmediateAlert[classification.Category] {
		e.reportFailure(card, cmd, classification.Category, err.Error(), failCount, firstFailed)
	}
	e.logger.Warn("停复机指令执行失败，按退避重试", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action),
		zap.String("category", classification.Category), zap.Int("fail_count", failCount), zap.Time("next_at", retry.nextAt),
		zap.String("metric", "stop_resume.failed"), zap.Error(err))
	return retry
}

// unknown 结果未知：置待确认，确认延迟后先判断再查状态；连续未知达阈值告警，指令不丢弃。
func (e *StopResumeExecutor) unknown(card *model.IotCard, cmd carrierqueue.CommandState, err error, params carrierqueue.CarrierParams,
	settings carrierqueue.Settings) srOutcome {
	unknown := cmd.Unknown + 1
	retry := srRetry(time.Now().Add(params.SRConfirmDelay))
	retry.set[carrierqueue.SRFieldPending] = "1"
	retry.set[carrierqueue.SRFieldUnknown] = strconv.Itoa(unknown)
	retry.set[carrierqueue.SRFieldError] = truncateRunes(constants.StopResumeFailureUnknown+"|"+err.Error(), 300)
	if unknown >= settings.SRUnknownAlert {
		e.reportFailure(card, cmd, constants.StopResumeFailureUnknown, err.Error(), unknown, cmd.WrittenAt)
	}
	e.logger.Warn("停复机结果未知，确认延迟后查状态", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action),
		zap.Int("unknown", unknown), zap.String("metric", "stop_resume.unknown"), zap.Error(err))
	return retry
}

func (e *StopResumeExecutor) classify(ctx context.Context, action string, err error) gatewaylimit.Classification {
	gwErr, ok := gateway.AsGatewayError(err)
	if !ok || gwErr.Category != constants.GatewayErrorUpstream {
		return gatewaylimit.Classification{Category: constants.GatewayClassTransient}
	}
	op := constants.GatewayOpStop
	if action == constants.StopResumeActionStart {
		op = constants.GatewayOpStart
	}
	return e.classifier.Classify(ctx, op, gwErr.ProviderMessage)
}

func (e *StopResumeExecutor) restoreLastCall(ctx context.Context, cardID uint, previous time.Time, params carrierqueue.CarrierParams) {
	if previous.IsZero() {
		e.store.Client().Del(ctx, constants.RedisCardStopResumeLastCallKey(cardID))
		return
	}
	_ = e.store.MarkStopResumeCall(ctx, cardID, previous, 2*params.SRMinInterval)
}

func (e *StopResumeExecutor) reportFailure(card *model.IotCard, cmd carrierqueue.CommandState, category, message string, failCount int, firstFailed time.Time) {
	if firstFailed.IsZero() {
		firstFailed = time.Now()
	}
	e.alerts.ReportFailure(carrierqueue.FailureEvent{CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID,
		CheckType: constants.CheckTypeStopResume, Category: category, Message: cmd.Action + ": " + message,
		FailCount: failCount, FirstFailed: firstFailed, At: time.Now()})
}

func (e *StopResumeExecutor) recovered(cmd carrierqueue.CommandState, card *model.IotCard) {
	if card != nil && (cmd.FailCount > 0 || cmd.Unknown > 0) {
		e.alerts.ReportRecovered(card.ID, constants.CheckTypeStopResume)
	}
}

func targetNetworkStatus(action string) int {
	if action == constants.StopResumeActionStop {
		return constants.NetworkStatusOffline
	}
	return constants.NetworkStatusOnline
}

func maxDuration(a, b time.Duration) time.Duration {
	if a > b {
		return a
	}
	return b
}

// countOutcome 累加执行计数：结束为成功，置待确认为结果未知，带失败次数为失败，其余推后为重试。
func (e *StopResumeExecutor) countOutcome(ctx context.Context, carrierID uint, outcome srOutcome) {
	if outcome.stat == statSkip {
		return
	}
	field := constants.StopResumeStatRetry
	switch {
	case outcome.stat != "":
		field = outcome.stat
	case outcome.done:
		field = constants.StopResumeStatSuccess
	case outcome.set[carrierqueue.SRFieldPending] == "1":
		field = constants.StopResumeStatUnknown
	case outcome.set[carrierqueue.SRFieldFail] != "":
		field = constants.StopResumeStatFailed
	}
	e.store.IncrStopResumeStat(ctx, carrierID, field)
}