package task import ( "context" "time" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/gateway" iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" ) // PollingRealnameHandler 实名检查任务处理器 // 职责:调 Gateway 查实名状态 → 写 DB → 触发首次实名激活任务 // 不做:停复机决策(实名状态 0→1 时通过 EvaluateAndAct 触发 not_realname 复机) type PollingRealnameHandler struct { base *PollingBase gateway *gateway.Client iotCardStore *postgres.IotCardStore asynqClient *asynq.Client stopResumeSvc iot_card_svc.StopResumeServiceInterface } // NewPollingRealnameHandler 创建实名检查任务处理器 func NewPollingRealnameHandler( base *PollingBase, gw *gateway.Client, iotCardStore *postgres.IotCardStore, asynqClient *asynq.Client, stopResumeSvc iot_card_svc.StopResumeServiceInterface, ) *PollingRealnameHandler { return &PollingRealnameHandler{ base: base, gateway: gw, iotCardStore: iotCardStore, asynqClient: asynqClient, stopResumeSvc: stopResumeSvc, } } // Handle 处理实名检查任务 func (h *PollingRealnameHandler) Handle(ctx context.Context, t *asynq.Task) error { startTime := time.Now() cardID, ok := parseTaskPayload(t.Payload(), h.base.logger) if !ok { return nil } if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingRealname) { h.base.logger.Debug("并发已满,重新入队", zap.Uint("card_id", cardID)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingRealname) card, err := h.base.getCardWithCache(ctx, cardID) if err != nil { if isNotFound(err) { return nil } h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(err)) h.base.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } if card.CardCategory == constants.CardCategoryIndustry { return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } var newStatus int var realStatusBool bool if h.gateway != nil { result, gwErr := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID}) if gwErr != nil { h.base.logger.Warn("查询实名状态失败", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.Error(gwErr)) h.base.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } realStatusBool = result.RealStatus if result.RealStatus { newStatus = constants.RealNameStatusVerified } else { newStatus = constants.RealNameStatusNotVerified } if h.base.verboseLog { h.base.logger.Info("实名状态轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.Bool("real_status", realStatusBool), zap.Int("new_status", newStatus), zap.Bool("changed", newStatus != card.RealNameStatus)) } } else { newStatus = card.RealNameStatus } statusChanged := newStatus != card.RealNameStatus isFirstRealname := statusChanged && card.RealNameStatus != constants.RealNameStatusVerified && newStatus == constants.RealNameStatusVerified now := time.Now() fields := map[string]any{"last_real_name_check_at": now} if statusChanged { fields["real_name_status"] = newStatus } if isFirstRealname { fields["first_realname_at"] = now } if err := h.iotCardStore.UpdateFields(ctx, cardID, fields); err != nil { h.base.logger.Error("更新卡实名信息失败", zap.Uint("card_id", cardID), zap.Error(err)) } if statusChanged { h.base.updateCardCache(ctx, cardID, map[string]any{"real_name_status": newStatus}) h.base.logger.Info("实名状态已变化", zap.Uint("card_id", cardID), zap.Int("old_status", card.RealNameStatus), zap.Int("new_status", newStatus)) if isFirstRealname { // 0→1:首次实名,触发套餐激活并立即复机评估(不等待下一个 carddata/package 周期) h.triggerFirstRealnameActivation(ctx, cardID) if h.stopResumeSvc != nil { freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID) if loadErr == nil { if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, freshCard); evalErr != nil { h.base.logger.Warn("实名完成后触发复机评估失败", zap.Uint("card_id", cardID), zap.Error(evalErr)) } } } } else if newStatus == constants.RealNameStatusNotVerified { // 1→0:实名逆转,立即触发停机评估(不等待下一个 carddata/package 周期,否则最长延迟 1 小时) h.base.logger.Warn("检测到实名逆转,立即触发停机评估", zap.Uint("card_id", cardID)) if h.stopResumeSvc != nil { freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID) if loadErr == nil { if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, freshCard); evalErr != nil { h.base.logger.Warn("实名逆转后触发停机评估失败", zap.Uint("card_id", cardID), zap.Error(evalErr)) } } } } } h.base.updateStats(ctx, constants.TaskTypePollingRealname, true, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } // triggerFirstRealnameActivation 首次实名后触发套餐激活任务 func (h *PollingRealnameHandler) triggerFirstRealnameActivation(ctx context.Context, cardID uint) { if h.asynqClient == nil { return } payload := map[string]any{ "carrier_type": "iot_card", "carrier_id": cardID, "activation_type": "realname", "timestamp": time.Now().Unix(), } payloadBytes, err := sonic.Marshal(payload) if err != nil { h.base.logger.Warn("序列化首次实名激活任务载荷失败", zap.Uint("card_id", cardID), zap.Error(err)) return } task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(30*time.Second), asynq.Queue(constants.QueueDefault), ) if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil { h.base.logger.Warn("提交首次实名激活任务失败", zap.Uint("card_id", cardID), zap.Error(err)) } }