From a887f91686072bfad86b3cf285ebf04ecf8257b6 Mon Sep 17 00:00:00 2001 From: huang Date: Sun, 26 Apr 2026 12:41:16 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=8D=A1=E8=BD=AE=E8=AE=AD?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/store/postgres/iot_card_store.go | 22 +++++++++++++-------- internal/task/polling_card_cache.go | 8 ++++++++ internal/task/polling_cardstatus_handler.go | 3 +++ internal/task/polling_protect_handler.go | 6 ++++++ internal/task/polling_realname_handler.go | 3 +++ 5 files changed, 34 insertions(+), 8 deletions(-) diff --git a/internal/store/postgres/iot_card_store.go b/internal/store/postgres/iot_card_store.go index 8a1c8e9..e7ee5d1 100644 --- a/internal/store/postgres/iot_card_store.go +++ b/internal/store/postgres/iot_card_store.go @@ -226,10 +226,13 @@ func buildActivationStatusExpr(fields map[string]any) any { return gorm.Expr(` CASE - WHEN COALESCE(?, network_status) = ? - AND (COALESCE(?, card_category) = ? OR COALESCE(?, real_name_status) = ?) - THEN ? - ELSE ? + WHEN COALESCE(CAST(? AS integer), network_status) = CAST(? AS integer) + AND ( + COALESCE(CAST(? AS text), card_category) = CAST(? AS text) + OR COALESCE(CAST(? AS integer), real_name_status) = CAST(? AS integer) + ) + THEN CAST(? AS integer) + ELSE CAST(? AS integer) END`, networkStatus, constants.NetworkStatusOnline, @@ -261,10 +264,13 @@ func buildActivatedAtExpr(fields map[string]any, now time.Time) any { return gorm.Expr(` CASE - WHEN activation_status = ? - AND COALESCE(?, network_status) = ? - AND (COALESCE(?, card_category) = ? OR COALESCE(?, real_name_status) = ?) - THEN COALESCE(activated_at, ?) + WHEN activation_status = CAST(? AS integer) + AND COALESCE(CAST(? AS integer), network_status) = CAST(? AS integer) + AND ( + COALESCE(CAST(? AS text), card_category) = CAST(? AS text) + OR COALESCE(CAST(? AS integer), real_name_status) = CAST(? AS integer) + ) + THEN COALESCE(activated_at, CAST(? AS timestamptz)) ELSE activated_at END`, constants.ActivationStatusInactive, diff --git a/internal/task/polling_card_cache.go b/internal/task/polling_card_cache.go index 08e1dc5..2008dca 100644 --- a/internal/task/polling_card_cache.go +++ b/internal/task/polling_card_cache.go @@ -46,6 +46,14 @@ func (b *PollingBase) updateCardCache(ctx context.Context, cardID uint, updates } } +// invalidateCardCache 删除轮询卡缓存，确保后续从数据库重建 +func (b *PollingBase) invalidateCardCache(ctx context.Context, cardID uint) { + key := constants.RedisPollingCardInfoKey(cardID) + if err := b.redis.Del(ctx, key).Err(); err != nil { + b.logger.Warn("删除卡缓存失败", zap.Uint("card_id", cardID), zap.Error(err)) + } +} + // parseCardFromCache 从 Redis Hash 反序列化卡信息 func parseCardFromCache(cardID uint, result map[string]string) *model.IotCard { card := &model.IotCard{} diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go index 10d93f0..a992984 100644 --- a/internal/task/polling_cardstatus_handler.go +++ b/internal/task/polling_cardstatus_handler.go @@ -104,6 +104,9 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, t *asynq.Task) er } 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 statusChanged { diff --git a/internal/task/polling_protect_handler.go b/internal/task/polling_protect_handler.go index c48de48..7bbd987 100644 --- a/internal/task/polling_protect_handler.go +++ b/internal/task/polling_protect_handler.go @@ -107,6 +107,9 @@ func (h *PollingProtectHandler) Handle(ctx context.Context, t *asynq.Task) error "stop_reason": constants.StopReasonProtectPeriod, }); updateErr != nil { h.base.logger.Warn("保护期停机 DB 更新失败", zap.Uint("card_id", cardID), zap.Error(updateErr)) + h.base.invalidateCardCache(ctx, cardID) + h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime)) + return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect) } h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOffline}) actionTaken = "forced_stop" @@ -130,6 +133,9 @@ func (h *PollingProtectHandler) Handle(ctx context.Context, t *asynq.Task) error "stop_reason": "", }); updateErr != nil { h.base.logger.Warn("保护期复机 DB 更新失败", zap.Uint("card_id", cardID), zap.Error(updateErr)) + h.base.invalidateCardCache(ctx, cardID) + h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime)) + return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect) } h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOnline}) actionTaken = "forced_resume" diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index eb260a0..abc03d3 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -129,6 +129,9 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, t *asynq.Task) erro } if err := h.iotCardStore.UpdateFields(ctx, cardID, fields); err != nil { h.base.logger.Error("更新卡实名信息失败", zap.Uint("card_id", cardID), zap.Error(err)) + h.base.invalidateCardCache(ctx, cardID) + h.base.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime)) + return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } if statusChanged {