junhong_cmp_fiber/internal/task/carrier_card_executor.go

package task

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

	"go.uber.org/zap"

	"github.com/break/junhong_cmp_fiber/internal/gateway"
	"github.com/break/junhong_cmp_fiber/internal/model"
	"github.com/break/junhong_cmp_fiber/internal/polling"
	"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
	"github.com/break/junhong_cmp_fiber/pkg/config"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// CarrierCardExecutor 按运营商账户调度的卡任务执行器：一张卡一个卡任务，按配置步骤顺序串行执行已到期的检查类型。
type CarrierCardExecutor struct {
	base       *PollingBase
	store      *carrierqueue.Store
	settings   *carrierqueue.SettingsProvider
	alerts     *carrierqueue.AlertWriter
	realname   *PollingRealnameHandler
	carddata   *PollingCarddataHandler
	cardStatus *PollingCardStatusHandler
	pkg        *PollingPackageHandler
	cfg        config.CarrierSchedulerConfig
	instanceID string
}

// NewCarrierCardExecutor 创建卡任务执行器。
func NewCarrierCardExecutor(base *PollingBase, store *carrierqueue.Store, settings *carrierqueue.SettingsProvider, alerts *carrierqueue.AlertWriter,
	realname *PollingRealnameHandler, carddata *PollingCarddataHandler, cardStatus *PollingCardStatusHandler, pkg *PollingPackageHandler,
	cfg config.CarrierSchedulerConfig, instanceID string) *CarrierCardExecutor {
	return &CarrierCardExecutor{base: base, store: store, settings: settings, alerts: alerts, realname: realname, carddata: carddata,
		cardStatus: cardStatus, pkg: pkg, cfg: cfg, instanceID: instanceID}
}

func (e *CarrierCardExecutor) renewInterval() time.Duration {
	if e.cfg.RenewInterval > 0 {
		return e.cfg.RenewInterval
	}
	return 30 * time.Second
}

func (e *CarrierCardExecutor) lease() time.Duration {
	if e.cfg.Lease > 0 {
		return e.cfg.Lease
	}
	return 5 * time.Minute
}

// Handle 消费一条卡任务消息；返回后由消费者 FIN。重试与退避全部由 Redis score 表达。
func (e *CarrierCardExecutor) Handle(ctx context.Context, body []byte, touch func()) {
	var msg carrierqueue.CardMessage
	if err := json.Unmarshal(body, &msg); err != nil || msg.CardID == 0 || msg.Token == "" {
		e.base.logger.Warn("卡任务消息无效，丢弃", zap.ByteString("body", body))
		return
	}
	execValue := msg.Token + "|" + e.instanceID
	execTTL := e.renewInterval() * 2
	got, err := e.store.AcquireExec(ctx, msg.CardID, msg.Token, execValue, execTTL)
	if err != nil || got != 1 {
		// 0：处理中标记已不是该 token（过期或重复投递）；-1：该 token 已有执行者（NSQ 重投时原消费者仍在执行）。
		return
	}
	settings := e.settings.Current(ctx)
	taskCtx, cancel := context.WithTimeout(ctx, settings.TaskTimeout())
	defer cancel()

	var lost atomic.Bool
	renewDone := make(chan struct{})
	go func() {
		ticker := time.NewTicker(e.renewInterval())
		defer ticker.Stop()
		for {
			select {
			case <-renewDone:
				return
			case <-taskCtx.Done():
				return
			case <-ticker.C:
				ok, renewErr := e.store.Renew(context.WithoutCancel(ctx), msg.CardID, msg.CarrierID, msg.Token, execValue, e.lease(), execTTL)
				if renewErr == nil && !ok {
					// token 已不一致：立即停止后续步骤，不再发起新的网关调用，只写已产生的业务结果。
					lost.Store(true)
					cancel()
					return
				}
				if touch != nil {
					touch()
				}
			}
		}
	}()
	set, del := e.execute(taskCtx, msg, settings, &lost)
	close(renewDone)

	finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer finishCancel()
	finished, finishErr := e.store.Finish(finishCtx, msg.CardID, msg.CarrierID, msg.Token, set, del)
	if finishErr != nil {
		e.base.logger.Warn("卡任务收尾失败，等待租约到期后重新领取", zap.Uint("card_id", msg.CardID), zap.Error(finishErr))
		return
	}
	if !finished {
		e.base.logger.Warn("卡任务收尾时处理中标记已不一致，只写入业务结果", zap.Uint("card_id", msg.CardID),
			zap.String("metric", "polling.carrier.finish_token_mismatch"))
	}
}

// execute 串行执行到期检查步骤，返回收尾要写入与删除的到期字段。
func (e *CarrierCardExecutor) execute(ctx context.Context, msg carrierqueue.CardMessage, settings carrierqueue.Settings, lost *atomic.Bool) (map[string]string, []string) {
	set, del := map[string]string{}, []string{}
	card, err := e.base.getCardWithCache(ctx, msg.CardID)
	if err != nil {
		if isNotFound(err) {
			e.removeCard(ctx, msg.CardID, msg.CarrierID)
		}
		return set, del
	}
	if !e.base.scopeAllowsCached(ctx, card) {
		e.removeCard(ctx, msg.CardID, msg.CarrierID)
		return set, del
	}
	due, err := e.store.ReadDue(ctx, msg.CardID)
	if err != nil {
		return set, del
	}
	now := time.Now()
	flowEvaluated := false
	executed := map[string]StepOutcome{}
	for _, checkType := range settings.StepOrder {
		item, ok := due.Types[checkType]
		if !ok || !item.HasDue || item.Due.After(now) {
			continue
		}
		if lost.Load() || ctx.Err() != nil {
			break
		}
		// 异步入库时流量评估发生在入库段，由其写入的评估标记表示套餐检查间隔内已触发过停复机评估。
		if checkType == constants.CheckTypePackage && settings.SkipPackageAfterFlow &&
			(flowEvaluated || e.base.flowEvaluatedRecently(ctx, msg.CardID)) {
			executed[checkType] = stepOK()
			continue
		}
		outcome := e.runStep(ctx, msg, checkType, settings.StepTimeout(checkType))
		if checkType == constants.CheckTypeCarddata && outcome.TriggeredEvaluation {
			flowEvaluated = true
		}
		executed[checkType] = outcome
		if outcome.Result == StepCardGone || outcome.Result == StepStopPolling {
			e.removeCard(ctx, msg.CardID, msg.CarrierID)
			return set, del
		}
	}
	if len(executed) == 0 {
		return set, del
	}
	// 收尾时按最新卡数据重新匹配间隔（步骤中卡状态、套餐可能已变化）。
	e.base.invalidateCardCache(ctx, msg.CardID)
	if fresh, freshErr := e.base.getCardWithCache(ctx, msg.CardID); freshErr == nil {
		card = fresh
	}
	intervals := e.base.configMgr.MergedTaskIntervals(card)
	now = time.Now()
	for checkType, outcome := range executed {
		e.writeBack(ctx, card, checkType, outcome, due.Types[checkType], intervals, settings, now, set, &del)
	}
	return set, del
}

// runStep 以单步超时、等待模式限流与 panic 恢复执行一个检查步骤。
func (e *CarrierCardExecutor) runStep(ctx context.Context, msg carrierqueue.CardMessage, checkType string, timeout time.Duration) (outcome StepOutcome) {
	stepCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	stepCtx = gateway.WithRateLimitWait(stepCtx, timeout)
	defer func() {
		if r := recover(); r != nil {
			e.base.logger.Error("卡任务步骤 panic，按失败处理", zap.Uint("card_id", msg.CardID), zap.String("check_type", checkType), zap.Any("panic", r))
			outcome = StepOutcome{Result: StepFailed, Category: "panic", Message: fmt.Sprint(r)}
		}
	}()
	switch checkType {
	case constants.CheckTypeCardStatus:
		return e.cardStatus.Step(stepCtx, msg.CardID)
	case constants.CheckTypeCarddata:
		return e.carddata.Step(stepCtx, msg.CardID)
	case constants.CheckTypeRealname:
		return e.realname.Step(stepCtx, msg.CardID)
	case constants.CheckTypePackage:
		stepCtx = withPollingWorkerAuditContext(stepCtx, constants.TaskTypePollingPackage, "套餐状态轮询任务", msg.Token)
		return e.pkg.Step(stepCtx, msg.CardID)
	default:
		return stepSkip()
	}
}

// writeBack 按步骤结果计算该检查类型下次到期、失败计数与延迟确认，并触发告警或恢复。
func (e *CarrierCardExecutor) writeBack(ctx context.Context, card *model.IotCard, checkType string, outcome StepOutcome, previous carrierqueue.TypeDue,
	intervals map[string]polling.TaskTypeInterval, settings carrierqueue.Settings, now time.Time, set map[string]string, del *[]string) {
	info, ok := intervals[constants.TaskTypeFromCheckType(checkType)]
	if !ok || info.Interval <= 0 {
		// 该类型已不在配置中（间隔为 0 或未配置）：移除到期，不再执行。
		*del = append(*del, carrierqueue.FieldDue(checkType), carrierqueue.FieldFail(checkType), carrierqueue.FieldFirstFail(checkType),
			carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType))
		return
	}
	interval := time.Duration(info.Interval) * time.Second
	normalNext := e.base.configMgr.NextCheckAt(ctx, card, constants.TaskTypeFromCheckType(checkType), info, now)
	switch outcome.Result {
	case StepSucceeded:
		next := normalNext
		set[carrierqueue.FieldOK(checkType)] = strconv.FormatInt(now.UnixMilli(), 10)
		if previous.FailCount > 0 {
			e.alerts.ReportRecovered(card.ID, checkType)
		}
		*del = append(*del, carrierqueue.FieldFail(checkType), carrierqueue.FieldFirstFail(checkType), carrierqueue.FieldError(checkType))
		// 延迟确认：结果已满足预期时取消，否则按序列取下一次确认 = min(正常间隔, 当前 + 确认延迟)。
		if len(previous.Confirm) > 0 {
			if previous.Expected != "" && outcome.Observed == previous.Expected {
				*del = append(*del, carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType))
			} else {
				if confirmAt := now.Add(previous.Confirm[0]); confirmAt.Before(next) {
					next = confirmAt
				}
				if rest := previous.Confirm[1:]; len(rest) > 0 {
					set[carrierqueue.FieldConfirm(checkType)] = joinSeconds(rest)
				} else {
					*del = append(*del, carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType))
				}
			}
		}
		set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(next.UnixMilli(), 10)
	case StepFailed:
		set[carrierqueue.FieldError(checkType)] = truncateRunes(outcome.Category+"|"+outcome.Message, 300)
		if !outcome.CountsAsFailure() {
			// 上游频率限制、已是目标状态：不计入连续失败，按首个退避间隔重试。
			set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(settings.BackoffFor(checkType, 1, interval)).UnixMilli(), 10)
			return
		}
		failCount := previous.FailCount + 1
		firstFailed := previous.FirstFailed
		if firstFailed.IsZero() {
			firstFailed = now
			set[carrierqueue.FieldFirstFail(checkType)] = strconv.FormatInt(now.UnixMilli(), 10)
		}
		set[carrierqueue.FieldFail(checkType)] = strconv.Itoa(failCount)
		set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(settings.BackoffFor(checkType, failCount, interval)).UnixMilli(), 10)
		if failCount >= settings.AlertThreshold(checkType) || now.Sub(firstFailed) >= settings.AlertFailDuration || settings.ImmediateAlert[outcome.Category] {
			e.alerts.ReportFailure(carrierqueue.FailureEvent{CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID, CheckType: checkType,
				Category: outcome.Category, Message: outcome.Message, FailCount: failCount, FirstFailed: firstFailed, At: now})
		}
	case StepRateLimited:
		// 限流等待超时：请求未发出，按短延迟重排，不计失败。
		delay := 5 * time.Second
		if e.base.rateLimitedDelay != nil {
			if configured := e.base.rateLimitedDelay(); configured > 0 {
				delay = configured
			}
		}
		set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(delay).UnixMilli(), 10)
	case StepDeferred:
		delay := outcome.DeferDelay
		if delay <= 0 {
			delay = 30 * time.Second
		}
		set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(delay).UnixMilli(), 10)
	default:
		// 跳过（如所属账户未开实名管控）视同本轮已检查：记最近成功时间，使请求检查的冷却同样生效。
		if outcome.Result == StepSkipped {
			set[carrierqueue.FieldOK(checkType)] = strconv.FormatInt(now.UnixMilli(), 10)
		}
		set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(normalNext.UnixMilli(), 10)
	}
}

func (e *CarrierCardExecutor) removeCard(ctx context.Context, cardID, carrierID uint) {
	removeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
	defer cancel()
	if err := e.store.Remove(removeCtx, cardID, carrierID); err != nil {
		e.base.logger.Warn("移出运营商账户轮询队列失败", zap.Uint("card_id", cardID), zap.Error(err))
	}
}

func joinSeconds(delays []time.Duration) string {
	parts := make([]string, 0, len(delays))
	for _, delay := range delays {
		parts = append(parts, strconv.FormatInt(int64(delay/time.Second), 10))
	}
	return strings.Join(parts, ",")
}

func truncateRunes(s string, max int) string {
	runes := []rune(s)
	if len(runes) <= max {
		return s
	}
	return string(runes[:max])
}