huang
2a7ac3f86e
Some checks failed
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Has been cancelled
- OnCardStatusChanged/Enabled/Disabled 添加 InvalidateCardCache， 解决 Refresh API 更新 DB 后 polling 缓存仍为旧值的 bug - AssetResolveResponse 的 enable_polling/network_status 去掉 omitempty，解决 false/0 时字段从响应中消失的问题 - Scheduler/Initializer/PackageHandler 增加 INFO 级别日志， 可通过日志判断轮询是否工作、处理了哪些卡
83 lines
2.8 KiB
Go
83 lines
2.8 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/hibiken/asynq"
|
||
"go.uber.org/zap"
|
||
|
||
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"
|
||
)
|
||
|
||
// PollingPackageHandler 套餐检查任务处理器
|
||
// 职责：触发 EvaluateAndAct 检查套餐状态并判断停复机
|
||
// 不做：直接停复机决策，委托给 StopResumeService.EvaluateAndAct
|
||
// 注：套餐状态由 PackageActivationHandler/DataResetHandler/UsageService 维护，
|
||
// 本 Handler 无需查 Gateway，只负责周期性触发再评估。
|
||
type PollingPackageHandler struct {
|
||
base *PollingBase
|
||
iotCardStore *postgres.IotCardStore
|
||
stopResumeSvc iot_card_svc.StopResumeServiceInterface
|
||
}
|
||
|
||
// NewPollingPackageHandler 创建套餐检查任务处理器
|
||
func NewPollingPackageHandler(
|
||
base *PollingBase,
|
||
iotCardStore *postgres.IotCardStore,
|
||
stopResumeSvc iot_card_svc.StopResumeServiceInterface,
|
||
) *PollingPackageHandler {
|
||
return &PollingPackageHandler{
|
||
base: base,
|
||
iotCardStore: iotCardStore,
|
||
stopResumeSvc: stopResumeSvc,
|
||
}
|
||
}
|
||
|
||
// Handle 处理套餐检查任务
|
||
func (h *PollingPackageHandler) 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.TaskTypePollingPackage) {
|
||
h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID))
|
||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
|
||
}
|
||
defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingPackage)
|
||
|
||
if _, cacheErr := h.base.getCardWithCache(ctx, cardID); cacheErr != nil {
|
||
if isNotFound(cacheErr) {
|
||
return nil
|
||
}
|
||
h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(cacheErr))
|
||
h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
|
||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
if h.stopResumeSvc != nil {
|
||
freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID)
|
||
if loadErr == nil {
|
||
h.base.logger.Info("套餐检查：执行停复机评估",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Int("network_status", freshCard.NetworkStatus),
|
||
zap.String("stop_reason", freshCard.StopReason))
|
||
if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, freshCard); evalErr != nil {
|
||
h.base.logger.Warn("套餐检查后停复机评估失败",
|
||
zap.Uint("card_id", cardID), zap.Error(evalErr))
|
||
}
|
||
} else {
|
||
h.base.logger.Warn("套餐检查：加载卡信息失败",
|
||
zap.Uint("card_id", cardID), zap.Error(loadErr))
|
||
}
|
||
}
|
||
|
||
h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
|
||
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
|
||
}
|