junhong_cmp_fiber/internal/task/polling_realname_handler.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
	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)
		}
		if result.RealStatus {
			newStatus = constants.RealNameStatusVerified
		} else {
			newStatus = constants.RealNameStatusNotVerified
		}
	} 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))
	}
}