diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go new file mode 100644 index 0000000..0002fe4 --- /dev/null +++ b/internal/task/polling_cardstatus_handler.go @@ -0,0 +1,125 @@ +package task + +import ( + "context" + "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) + } + + var newNetworkStatus int + 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) + } + // "停机" → 0(停机),"准备"/"正常" → 1(开机) + if result.CardStatus == "停机" { + newNetworkStatus = constants.NetworkStatusOffline + } else { + newNetworkStatus = constants.NetworkStatusOnline + } + 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.Int("new_network_status", newNetworkStatus), + zap.Bool("changed", newNetworkStatus != card.NetworkStatus)) + } + } else { + newNetworkStatus = card.NetworkStatus + } + + statusChanged := newNetworkStatus != card.NetworkStatus + now := time.Now() + + // 无论状态是否变化,都更新最后一次检查时间 + fields := map[string]any{"last_card_status_check_at": now} + if statusChanged { + fields["network_status"] = newNetworkStatus + } + if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, fields); updateErr != nil { + h.base.logger.Warn("更新卡状态信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr)) + } + + if statusChanged { + h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": newNetworkStatus}) + 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) +} diff --git a/internal/task/polling_utils.go b/internal/task/polling_utils.go index 5d9d119..1b1d27c 100644 --- a/internal/task/polling_utils.go +++ b/internal/task/polling_utils.go @@ -31,6 +31,10 @@ func getIntervalByTaskType(cfg *model.PollingConfig, taskType string) int { if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { return *cfg.ProtectCheckInterval } + case constants.TaskTypePollingCardStatus: + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + return *cfg.CardStatusCheckInterval + } } return 0 }