From 76f657c986fd9296deb5fe0e7b8ae82885b3991b Mon Sep 17 00:00:00 2001 From: break Date: Tue, 30 Jun 2026 15:54:55 +0900 Subject: [PATCH] =?UTF-8?q?1.=20=E5=A5=97=E9=A4=90=E8=BF=87=E6=9C=9F?= =?UTF-8?q?=E5=89=8D=E5=86=8D=E4=B8=BB=E5=8A=A8=E5=90=8C=E6=AD=A5=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=B5=81=E9=87=8F=202.=20=E7=94=9F=E6=95=88=E5=A5=97?= =?UTF-8?q?=E9=A4=90=E9=80=BB=E8=BE=91=E9=94=99=E8=AF=AF=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/worker/main.go | 13 ++++ .../polling/package_activation_handler.go | 65 +++++++++++++++++-- internal/service/iot_card/service.go | 13 ++++ .../service/package/activation_service.go | 14 ++-- 4 files changed, 94 insertions(+), 11 deletions(-) diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 0427f16..6134e5a 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -346,6 +346,19 @@ func startPollingScheduler( appLogger, ) + // 注入套餐失效前流量同步器：最小化 iot_card.Service（只需 gateway + store + redis + 流量扣减） + trafficSyncer := iot_card_svc.New( + runtime.db, + runtime.pollingIotCardStore, + nil, nil, nil, nil, nil, + runtime.gatewayClient, + appLogger, + nil, + ) + trafficSyncer.SetRedisClient(runtime.redisClient) + trafficSyncer.SetDataDeductor(runtime.workerResult.Services.UsageService) + activationHandler.SetTrafficSyncer(trafficSyncer) + pollingScheduler := polling.NewScheduler( runtime.redisClient, runtime.asynqClient, diff --git a/internal/polling/package_activation_handler.go b/internal/polling/package_activation_handler.go index 9e058c1..a4642ad 100644 --- a/internal/polling/package_activation_handler.go +++ b/internal/polling/package_activation_handler.go @@ -17,6 +17,12 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/errors" ) +// TrafficSyncer 套餐失效前流量同步接口 +// 在主套餐标记为已过期之前，从 Gateway 拉取最新流量写入 DB +type TrafficSyncer interface { + RefreshCardDataByID(ctx context.Context, cardID uint) error +} + // PackageActivationHandler 套餐激活检查处理器 // 任务 19: 处理主套餐过期、加油包级联失效、待生效主套餐激活 type PackageActivationHandler struct { @@ -27,6 +33,7 @@ type PackageActivationHandler struct { deviceSimBinding *postgres.DeviceSimBindingStore activationService *packagepkg.ActivationService stopResumeCallback packagepkg.StopResumeCallback // 停复机回调，用于套餐过期后主动停机 + trafficSyncer TrafficSyncer // 套餐失效前流量同步，可选 logger *zap.Logger } @@ -60,6 +67,48 @@ func NewPackageActivationHandler( } } +// SetTrafficSyncer 注入套餐失效前流量同步器（在 Start 前调用） +func (h *PackageActivationHandler) SetTrafficSyncer(syncer TrafficSyncer) { + h.trafficSyncer = syncer +} + +// syncTrafficBeforeExpiry 套餐失效前从 Gateway 同步一次最新流量 +// iot_card 直接同步；device 遍历所有绑定卡同步 +// 失败只记录警告，不阻断失效流程 +func (h *PackageActivationHandler) syncTrafficBeforeExpiry(ctx context.Context, carrierType string, carrierID uint) { + if h.trafficSyncer == nil { + return + } + + if carrierType == constants.AssetTypeIotCard { + if err := h.trafficSyncer.RefreshCardDataByID(ctx, carrierID); err != nil { + h.logger.Warn("套餐失效前流量同步失败", + zap.String("carrier_type", carrierType), + zap.Uint("carrier_id", carrierID), + zap.Error(err)) + } + return + } + + if carrierType == constants.AssetTypeDevice { + bindings, err := h.deviceSimBinding.ListByDeviceID(ctx, carrierID) + if err != nil { + h.logger.Warn("套餐失效前查询设备绑定卡失败", + zap.Uint("device_id", carrierID), + zap.Error(err)) + return + } + for _, b := range bindings { + if err := h.trafficSyncer.RefreshCardDataByID(ctx, b.IotCardID); err != nil { + h.logger.Warn("套餐失效前流量同步失败（设备绑定卡）", + zap.Uint("device_id", carrierID), + zap.Uint("card_id", b.IotCardID), + zap.Error(err)) + } + } + } +} + // HandlePackageActivationCheck 任务 19.2-19.5: 处理套餐激活检查 // 每 10 秒调度一次，检查过期主套餐并激活下一个待生效主套餐 func (h *PackageActivationHandler) HandlePackageActivationCheck(ctx context.Context) error { @@ -145,20 +194,21 @@ func (h *PackageActivationHandler) findAndActivateOrphanPackages(ctx context.Con continue } - // 检查该载体是否已有生效主套餐 + // 检查该载体是否已有占位主套餐（生效中或已用完均视为占位，不允许激活待生效套餐） var activeCount int64 var countErr error + occupiedStatuses := []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted} if key.carrierType == "iot_card" { countErr = h.db.WithContext(ctx). Model(&model.PackageUsage{}). - Where("status = ?", constants.PackageUsageStatusActive). + Where("status IN ?", occupiedStatuses). Where("master_usage_id IS NULL"). Where("iot_card_id = ?", key.carrierID). Count(&activeCount).Error } else { countErr = h.db.WithContext(ctx). Model(&model.PackageUsage{}). - Where("status = ?", constants.PackageUsageStatusActive). + Where("status IN ?", occupiedStatuses). Where("master_usage_id IS NULL"). Where("device_id = ?", key.carrierID). Count(&activeCount).Error @@ -171,7 +221,7 @@ func (h *PackageActivationHandler) findAndActivateOrphanPackages(ctx context.Con continue } - // 已有生效套餐，跳过 + // 已有生效或已用完（占位）套餐，跳过 if activeCount > 0 { continue } @@ -219,10 +269,15 @@ func (h *PackageActivationHandler) findExpiredMainPackages(ctx context.Context) } // processExpiredPackage 处理单个过期套餐 -// triggerStopAfterExpiry 在事务提交后才调用，避免 goroutine 在 TX 提交前读到脏数据 +// 流程：先同步最新流量 → 事务内标记过期/失效/激活下一个 → 事务提交后触发停机 func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pkg *model.PackageUsage) error { carrierType, carrierID := h.getCarrierInfo(pkg) + // 失效前先从 Gateway 同步一次最新流量，确保 data_usage_mb 精确 + if carrierType != "" && carrierID > 0 { + h.syncTrafficBeforeExpiry(ctx, carrierType, carrierID) + } + err := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // 任务 19.3: 更新过期主套餐状态为 Expired (status=3) if err := tx.Model(pkg).Update("status", constants.PackageUsageStatusExpired).Error; err != nil { diff --git a/internal/service/iot_card/service.go b/internal/service/iot_card/service.go index 1027548..b3316a9 100644 --- a/internal/service/iot_card/service.go +++ b/internal/service/iot_card/service.go @@ -1494,6 +1494,19 @@ func (s *Service) buildCardNotFoundFailedItems(iccids []string) []dto.CardSeries return items } +// RefreshCardDataByID 通过卡 ID 从 Gateway 同步卡数据 +// 内部查询 ICCID 后委托给 RefreshCardDataFromGateway，供套餐失效前流量同步使用 +func (s *Service) RefreshCardDataByID(ctx context.Context, cardID uint) error { + card, err := s.iotCardStore.GetByID(ctx, cardID) + if err != nil { + if err == gorm.ErrRecordNotFound { + return nil + } + return err + } + return s.RefreshCardDataFromGateway(ctx, card.ICCID) +} + // RefreshCardDataFromGateway 从 Gateway 完整同步卡数据 // 调用网关查询网络状态、实名状态、本月流量，并写回数据库 // 流量采用增量计算（与轮询逻辑一致）：increment = 当前网关读数 - 上次网关读数 diff --git a/internal/service/package/activation_service.go b/internal/service/package/activation_service.go index 841bfaf..65fdd3c 100644 --- a/internal/service/package/activation_service.go +++ b/internal/service/package/activation_service.go @@ -92,19 +92,20 @@ func (s *ActivationService) ActivateByRealname(ctx context.Context, carrierType // 检查是否是主套餐 if usage.MasterUsageID == nil { - // 主套餐：需要检查是否有已激活的主套餐 + // 主套餐：需要检查是否有已激活或已用完（仍在有效期）的主套餐 var activeMain model.PackageUsage - err := tx.Where("status = ?", constants.PackageUsageStatusActive). + err := tx.Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}). Where("master_usage_id IS NULL"). Where(carrierType+"_id = ?", carrierID). Order("priority ASC"). First(&activeMain).Error if err == nil { - // 已有激活的主套餐，保持排队状态 - s.logger.Warn("已有激活主套餐，跳过激活", + // 已有生效中或已用完（占位）的主套餐，保持排队状态 + s.logger.Warn("已有占位主套餐，跳过激活", zap.Uint("usage_id", usage.ID), - zap.Uint("active_main_id", activeMain.ID)) + zap.Uint("active_main_id", activeMain.ID), + zap.Int("active_main_status", activeMain.Status)) continue } if err != gorm.ErrRecordNotFound { @@ -500,9 +501,10 @@ func (s *ActivationService) getNextPendingMainPackage(ctx context.Context, tx *g func (s *ActivationService) hasActiveMainPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) (bool, error) { var count int64 + // 生效中（1）和已用完（2）都属于"占位"状态：已用完的套餐仍在有效期内，不允许提前激活下一个主套餐 if err := tx.WithContext(ctx). Model(&model.PackageUsage{}). - Where("status = ?", constants.PackageUsageStatusActive). + Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}). Where("master_usage_id IS NULL"). Where(carrierType+"_id = ?", carrierID). Count(&count).Error; err != nil {