diff --git a/internal/service/device/service.go b/internal/service/device/service.go index dc27647..7715f05 100644 --- a/internal/service/device/service.go +++ b/internal/service/device/service.go @@ -532,6 +532,7 @@ func (s *Service) AllocateDevices(ctx context.Context, req *dto.AllocateDevicesR ) return nil, err } + s.iotCardStore.InvalidateListCountCache(ctx) s.logDeviceOperation( ctx, @@ -729,6 +730,7 @@ func (s *Service) RecallDevices(ctx context.Context, req *dto.RecallDevicesReque ) return nil, err } + s.iotCardStore.InvalidateListCountCache(ctx) s.logDeviceOperation( ctx, diff --git a/internal/service/iot_card/service.go b/internal/service/iot_card/service.go index bbbd174..43cd6b9 100644 --- a/internal/service/iot_card/service.go +++ b/internal/service/iot_card/service.go @@ -537,6 +537,7 @@ func (s *Service) AllocateCards(ctx context.Context, req *dto.AllocateStandalone }) return nil, err } + s.iotCardStore.InvalidateListCountCache(ctx) // 通知轮询调度器状态变化(卡被分配后可能需要重新匹配配置) if s.pollingCallback != nil && len(cardIDs) > 0 { @@ -749,6 +750,7 @@ func (s *Service) RecallCards(ctx context.Context, req *dto.RecallStandaloneCard }) return nil, err } + s.iotCardStore.InvalidateListCountCache(ctx) // 通知轮询调度器状态变化(卡被回收后可能需要重新匹配配置) if s.pollingCallback != nil && len(cardIDs) > 0 { diff --git a/internal/store/postgres/iot_card_store.go b/internal/store/postgres/iot_card_store.go index 1c415fb..58705b2 100644 --- a/internal/store/postgres/iot_card_store.go +++ b/internal/store/postgres/iot_card_store.go @@ -18,7 +18,7 @@ import ( "gorm.io/gorm" ) -const listCountCacheTTL = 30 * time.Minute +const listCountCacheTTL = 30 * time.Second type IotCardStore struct { db *gorm.DB @@ -33,14 +33,22 @@ func NewIotCardStore(db *gorm.DB, redis *redis.Client) *IotCardStore { } func (s *IotCardStore) Create(ctx context.Context, card *model.IotCard) error { - return s.db.WithContext(ctx).Create(card).Error + if err := s.db.WithContext(ctx).Create(card).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } func (s *IotCardStore) CreateBatch(ctx context.Context, cards []*model.IotCard) error { if len(cards) == 0 { return nil } - return s.db.WithContext(ctx).CreateInBatches(cards, 100).Error + if err := s.db.WithContext(ctx).CreateInBatches(cards, 100).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } func (s *IotCardStore) GetByID(ctx context.Context, id uint) (*model.IotCard, error) { @@ -162,7 +170,11 @@ func (s *IotCardStore) ExistsByICCIDBatch(ctx context.Context, iccids []string) } func (s *IotCardStore) Update(ctx context.Context, card *model.IotCard) error { - return s.db.WithContext(ctx).Save(card).Error + if err := s.db.WithContext(ctx).Save(card).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } // UpdateFields 按 ID 部分更新卡字段 @@ -185,9 +197,13 @@ func (s *IotCardStore) UpdateFields(ctx context.Context, cardID uint, fields map } } - return s.db.WithContext(ctx).Model(&model.IotCard{}). + if err := s.db.WithContext(ctx).Model(&model.IotCard{}). Where("id = ?", cardID). - Updates(updates).Error + Updates(updates).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } func shouldSyncActivationStatus(fields map[string]any) bool { @@ -299,7 +315,11 @@ func (s *IotCardStore) UpdateRealnamePolicy(ctx context.Context, cardID uint, re } func (s *IotCardStore) Delete(ctx context.Context, id uint) error { - return s.db.WithContext(ctx).Delete(&model.IotCard{}, id).Error + if err := s.db.WithContext(ctx).Delete(&model.IotCard{}, id).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } func (s *IotCardStore) List(ctx context.Context, opts *store.QueryOptions, filters map[string]any) ([]*model.IotCard, int64, error) { @@ -362,13 +382,13 @@ func (s *IotCardStore) List(ctx context.Context, opts *store.QueryOptions, filte query = query.Where("series_id = ?", seriesID) } - if cachedTotal, ok := s.getCachedCount(ctx, "iot_card", filters); ok { + if cachedTotal, ok, cacheVersion := s.getCachedCount(ctx, "iot_card", filters); ok { total = cachedTotal } else { if err := query.Count(&total).Error; err != nil { return nil, 0, err } - s.cacheCount(ctx, "iot_card", filters, total) + s.cacheCount(ctx, "iot_card", filters, total, cacheVersion) } // 分页处理 @@ -450,13 +470,13 @@ func (s *IotCardStore) listStandaloneTwoPhase(ctx context.Context, opts *store.Q query = middleware.ApplyShopFilter(ctx, query) query = s.applyStandaloneFilters(ctx, query, filters) - if cachedTotal, ok := s.getCachedCount(ctx, "standalone", filters); ok { + if cachedTotal, ok, cacheVersion := s.getCachedCount(ctx, "standalone", filters); ok { total = cachedTotal } else { if err := query.Count(&total).Error; err != nil { return nil, 0, err } - s.cacheCount(ctx, "standalone", filters, total) + s.cacheCount(ctx, "standalone", filters, total, cacheVersion) } offset := (opts.Page - 1) * opts.PageSize @@ -511,13 +531,13 @@ func (s *IotCardStore) listStandaloneDefault(ctx context.Context, opts *store.Qu query = middleware.ApplyShopFilter(ctx, query) query = s.applyStandaloneFilters(ctx, query, filters) - if cachedTotal, ok := s.getCachedCount(ctx, "standalone", filters); ok { + if cachedTotal, ok, cacheVersion := s.getCachedCount(ctx, "standalone", filters); ok { total = cachedTotal } else { if err := query.Count(&total).Error; err != nil { return nil, 0, err } - s.cacheCount(ctx, "standalone", filters, total) + s.cacheCount(ctx, "standalone", filters, total, cacheVersion) } offset := (opts.Page - 1) * opts.PageSize @@ -559,7 +579,7 @@ func (s *IotCardStore) listStandaloneParallel(ctx context.Context, opts *store.Q results := make([]shopResult, len(shopIDs)) var wg sync.WaitGroup - cachedTotal, hasCachedTotal := s.getCachedCount(ctx, "standalone", filters) + cachedTotal, hasCachedTotal, cacheVersion := s.getCachedCount(ctx, "standalone", filters) for i, shopID := range shopIDs { wg.Add(1) @@ -613,7 +633,7 @@ func (s *IotCardStore) listStandaloneParallel(ctx context.Context, opts *store.Q } if !hasCachedTotal && totalCount > 0 { - s.cacheCount(ctx, "standalone", filters, totalCount) + s.cacheCount(ctx, "standalone", filters, totalCount, cacheVersion) } sort.Slice(allCards, func(i, j int) bool { @@ -668,7 +688,7 @@ func (s *IotCardStore) listStandaloneParallelTwoPhase(ctx context.Context, opts results := make([]shopResult, len(shopIDs)) var wg sync.WaitGroup - cachedTotal, hasCachedTotal := s.getCachedCount(ctx, "standalone", filters) + cachedTotal, hasCachedTotal, cacheVersion := s.getCachedCount(ctx, "standalone", filters) // Phase 1: 并行获取每个 shop 的 (id, created_at),走覆盖索引 for i, shopID := range shopIDs { @@ -723,7 +743,7 @@ func (s *IotCardStore) listStandaloneParallelTwoPhase(ctx context.Context, opts } if !hasCachedTotal && totalCount > 0 { - s.cacheCount(ctx, "standalone", filters, totalCount) + s.cacheCount(ctx, "standalone", filters, totalCount, cacheVersion) } sort.Slice(allIDs, func(i, j int) bool { @@ -1006,9 +1026,13 @@ func (s *IotCardStore) BatchUpdateShopIDAndStatus(ctx context.Context, cardIDs [ "shop_id": shopID, "status": status, } - return s.db.WithContext(ctx).Model(&model.IotCard{}). + if err := s.db.WithContext(ctx).Model(&model.IotCard{}). Where("id IN ?", cardIDs). - Updates(updates).Error + Updates(updates).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } func (s *IotCardStore) GetBoundCardIDs(ctx context.Context, cardIDs []uint) ([]uint, error) { @@ -1157,9 +1181,13 @@ func (s *IotCardStore) BatchDelete(ctx context.Context, cardIDs []uint) error { if len(cardIDs) == 0 { return nil } - return s.db.WithContext(ctx). + if err := s.db.WithContext(ctx). Where("id IN ?", cardIDs). - Delete(&model.IotCard{}).Error + Delete(&model.IotCard{}).Error; err != nil { + return err + } + s.invalidateIotCardListCountCache(ctx) + return nil } // ExistsByVirtualNoBatch 批量检查 virtual_no 是否已存在 @@ -1243,24 +1271,73 @@ func (s *IotCardStore) ListForPollingBatch(ctx context.Context, lastID uint, lim // ==================== 列表计数缓存 ==================== -func (s *IotCardStore) getCachedCount(ctx context.Context, table string, filters map[string]any) (int64, bool) { +func (s *IotCardStore) getCachedCount(ctx context.Context, table string, filters map[string]any) (int64, bool, int64) { if s.redis == nil { - return 0, false + return 0, false, 0 } - key := constants.RedisListCountKey(table, middleware.GetUserIDFromContext(ctx), hashFilters(filters)) + version := s.listCountCacheVersion(ctx, table) + key := s.listCountCacheKey(ctx, table, filters, version) val, err := s.redis.Get(ctx, key).Int64() if err != nil { - return 0, false + return 0, false, version } - return val, true + return val, true, version } -func (s *IotCardStore) cacheCount(ctx context.Context, table string, filters map[string]any, total int64) { +func (s *IotCardStore) cacheCount(ctx context.Context, table string, filters map[string]any, total int64, version int64) { if s.redis == nil { return } - key := constants.RedisListCountKey(table, middleware.GetUserIDFromContext(ctx), hashFilters(filters)) - s.redis.Set(ctx, key, total, listCountCacheTTL) + if s.listCountCacheVersion(ctx, table) != version { + return + } + key := s.listCountCacheKey(ctx, table, filters, version) + _ = s.redis.Set(ctx, key, total, listCountCacheTTL).Err() +} + +func (s *IotCardStore) listCountCacheKey(ctx context.Context, table string, filters map[string]any, version int64) string { + return constants.RedisListCountKey( + table, + version, + middleware.GetUserIDFromContext(ctx), + hashDataScope(ctx), + hashFilters(filters), + ) +} + +func (s *IotCardStore) listCountCacheVersion(ctx context.Context, table string) int64 { + version, err := s.redis.Get(ctx, constants.RedisListCountVersionKey(table)).Int64() + if err != nil { + return 0 + } + return version +} + +// InvalidateListCountCache 失效 IoT 卡相关列表计数缓存。 +func (s *IotCardStore) InvalidateListCountCache(ctx context.Context) { + s.invalidateIotCardListCountCache(ctx) +} + +func (s *IotCardStore) invalidateIotCardListCountCache(ctx context.Context) { + s.invalidateListCountCache(ctx, "iot_card", "standalone") +} + +func (s *IotCardStore) invalidateListCountCache(ctx context.Context, tables ...string) { + if s.redis == nil { + return + } + + seen := make(map[string]struct{}, len(tables)) + for _, table := range tables { + if table == "" { + continue + } + if _, ok := seen[table]; ok { + continue + } + seen[table] = struct{}{} + _ = s.redis.Incr(ctx, constants.RedisListCountVersionKey(table)).Err() + } } func hashFilters(filters map[string]any) string { @@ -1280,3 +1357,27 @@ func hashFilters(filters map[string]any) string { } return fmt.Sprintf("%08x", h.Sum32()) } + +func hashDataScope(ctx context.Context) string { + userType := middleware.GetUserTypeFromContext(ctx) + shopID := middleware.GetShopIDFromContext(ctx) + enterpriseID := middleware.GetEnterpriseIDFromContext(ctx) + subordinateIDs := middleware.GetSubordinateShopIDs(ctx) + + h := fnv.New32a() + _, _ = fmt.Fprintf(h, "user_type=%d;shop_id=%d;enterprise_id=%d;", userType, shopID, enterpriseID) + if subordinateIDs == nil { + _, _ = h.Write([]byte("subordinate=nil")) + return fmt.Sprintf("%08x", h.Sum32()) + } + + ids := append([]uint(nil), subordinateIDs...) + sort.Slice(ids, func(i, j int) bool { + return ids[i] < ids[j] + }) + _, _ = fmt.Fprint(h, "subordinate=") + for _, id := range ids { + _, _ = fmt.Fprintf(h, "%d,", id) + } + return fmt.Sprintf("%08x", h.Sum32()) +} diff --git a/internal/task/device_import.go b/internal/task/device_import.go index 15bc10d..efc332d 100644 --- a/internal/task/device_import.go +++ b/internal/task/device_import.go @@ -382,6 +382,9 @@ func (h *DeviceImportHandler) processBatch(ctx context.Context, task *model.Devi result.failCount++ continue } + if len(validCards) > 0 && h.iotCardStore != nil { + h.iotCardStore.InvalidateListCountCache(ctx) + } for _, slotICCID := range row.SlotICCIDs { if card, exists := existingCards[slotICCID.ICCID]; exists && !boundCards[slotICCID.ICCID] && card.ShopID == nil { diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index d287784..dfa70d3 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -339,8 +339,14 @@ func RedisPackageActivationLockKey(carrierType string, carrierID uint) string { // RedisListCountKey 列表查询计数缓存键 // 用途:缓存分页列表的 COUNT(*) 结果,避免每次翻页重复全表计数 // 过期时间:30 秒 -func RedisListCountKey(table string, userID uint, filterHash string) string { - return fmt.Sprintf("list_count:%s:%d:%s", table, userID, filterHash) +func RedisListCountKey(table string, version int64, userID uint, scopeHash string, filterHash string) string { + return fmt.Sprintf("list_count:v2:%s:%d:%d:%s:%s", table, version, userID, scopeHash, filterHash) +} + +// RedisListCountVersionKey 列表查询计数缓存版本键 +// 用途:数据变更后递增版本,让旧 COUNT(*) 缓存立即失效 +func RedisListCountVersionKey(table string) string { + return fmt.Sprintf("list_count_version:%s", table) } // ========================================