package task import ( "context" "strings" "time" "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" ) // PollingCardStatusHandler 卡开停机状态轮询任务处理器 // 职责:调 Gateway 查卡状态(正常/停机/准备)→ 映射为 network_status → 写 DB → 触发停复机评估 // 不做:直接停复机决策,委托给 StopResumeService.EvaluateAndAct type PollingCardStatusHandler struct { base *PollingBase gateway *gateway.Client iotCardStore *postgres.IotCardStore stopResumeSvc iot_card_svc.StopResumeServiceInterface } // NewPollingCardStatusHandler 创建卡状态轮询任务处理器 func NewPollingCardStatusHandler( base *PollingBase, gw *gateway.Client, iotCardStore *postgres.IotCardStore, stopResumeSvc iot_card_svc.StopResumeServiceInterface, ) *PollingCardStatusHandler { return &PollingCardStatusHandler{ base: base, gateway: gw, iotCardStore: iotCardStore, stopResumeSvc: stopResumeSvc, } } // Handle 处理卡状态轮询任务 func (h *PollingCardStatusHandler) 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.TaskTypePollingCardStatus) { h.base.logger.Debug("并发已满,重新入队", zap.Uint("card_id", cardID)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCardStatus) 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.TaskTypePollingCardStatus, false, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } newNetworkStatus := card.NetworkStatus gatewayCardStatus := "" gatewayExtend := card.GatewayExtend gatewayCardIMEI := "" statusQueried := false if h.gateway != nil { result, gwErr := h.gateway.QueryCardStatus(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.TaskTypePollingCardStatus, false, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } gatewayCardStatus = result.CardStatus gatewayExtend = strings.TrimSpace(result.Extend) gatewayCardIMEI = strings.TrimSpace(result.IMEI) statusQueried = true parsedStatus, ok := gateway.ParseCardNetworkStatus(result.CardStatus, gatewayExtend) if !ok { h.base.logger.Warn("卡状态轮询:未知 Gateway 卡状态", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.String("card_status", result.CardStatus), zap.String("extend", gatewayExtend)) } else { newNetworkStatus = parsedStatus } if h.base.verboseLog { h.base.logger.Info("卡状态轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.String("card_status", result.CardStatus), zap.String("extend", gatewayExtend), zap.Int("new_network_status", newNetworkStatus), zap.Bool("changed", newNetworkStatus != card.NetworkStatus)) } } statusChanged := newNetworkStatus != card.NetworkStatus now := time.Now() // 无论状态是否变化,都更新最后一次检查时间 fields := map[string]any{"last_card_status_check_at": now} if statusQueried { fields["gateway_extend"] = gatewayExtend fields["last_sync_time"] = now if gatewayCardIMEI != "" { fields["gateway_card_imei"] = gatewayCardIMEI } } if statusChanged { fields["network_status"] = newNetworkStatus // 网关侧停机事件:补写 stop_reason,便于状态追踪和区分手动停机 if newNetworkStatus == constants.NetworkStatusOffline && card.StopReason == "" && gateway.IsGatewayCardStopped(gatewayCardStatus) { fields["stop_reason"] = constants.StopReasonCarrierStopped } } if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, fields); updateErr != nil { h.base.logger.Warn("更新卡状态信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr)) h.base.invalidateCardCache(ctx, cardID) h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, false, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } if statusQueried || statusChanged { cacheUpdates := map[string]any{} if statusQueried { cacheUpdates["gateway_extend"] = gatewayExtend } if statusChanged { cacheUpdates["network_status"] = newNetworkStatus } h.base.updateCardCache(ctx, cardID, cacheUpdates) } if statusChanged { h.base.logger.Info("卡状态已变化", zap.Uint("card_id", cardID), zap.Int("old_status", card.NetworkStatus), zap.Int("new_status", newNetworkStatus)) 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.TaskTypePollingCardStatus, true, time.Since(startTime)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) }