huang
0ec16d4afa
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m34s
Gateway 返回 data=null 时 sonic 解析为零值结构体（ICCID="" RealStatus=false）， 导致已实名卡被误判为未实名并触发停机。 通过检查响应中 ICCID 是否回填来判断数据有效性： - ICCID 为空：视为上游接口故障，重新入队，不更新实名状态 - ICCID 有值：响应有效，正常处理（含合法的未实名降级）
195 lines
6.9 KiB
Go
195 lines
6.9 KiB
Go
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)
|
||
}
|
||
// 上游接口故障时 Gateway 可能返回 data=null，
|
||
// sonic 将其解析为零值结构体（ICCID="" RealStatus=false），
|
||
// 需通过 ICCID 是否回填来判断响应是否有效。
|
||
if result.ICCID == "" {
|
||
h.base.logger.Warn("实名查询响应异常：ICCID 为空，疑似上游接口故障，跳过本次同步",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("iccid", card.ICCID))
|
||
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))
|
||
}
|
||
}
|