package polling import ( "context" "time" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/model" packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" ) // PackageActivationHandler 套餐激活检查处理器 // 任务 19: 处理主套餐过期、加油包级联失效、待生效主套餐激活 type PackageActivationHandler struct { db *gorm.DB redis *redis.Client queueClient *asynq.Client packageUsageStore *postgres.PackageUsageStore deviceSimBinding *postgres.DeviceSimBindingStore activationService *packagepkg.ActivationService stopResumeCallback packagepkg.StopResumeCallback // 停复机回调,用于套餐过期后主动停机 logger *zap.Logger } // PackageActivationPayload 套餐激活任务载荷 type PackageActivationPayload struct { PackageUsageID uint `json:"package_usage_id"` CarrierType string `json:"carrier_type"` // "iot_card" 或 "device" CarrierID uint `json:"carrier_id"` ActivationType string `json:"activation_type"` // "queue" 或 "realname" Timestamp int64 `json:"timestamp"` } // NewPackageActivationHandler 创建套餐激活检查处理器 func NewPackageActivationHandler( db *gorm.DB, redis *redis.Client, queueClient *asynq.Client, activationService *packagepkg.ActivationService, stopResumeCallback packagepkg.StopResumeCallback, logger *zap.Logger, ) *PackageActivationHandler { return &PackageActivationHandler{ db: db, redis: redis, queueClient: queueClient, packageUsageStore: postgres.NewPackageUsageStore(db, redis), deviceSimBinding: postgres.NewDeviceSimBindingStore(db, redis), activationService: activationService, stopResumeCallback: stopResumeCallback, logger: logger, } } // HandlePackageActivationCheck 任务 19.2-19.5: 处理套餐激活检查 // 每 10 秒调度一次,检查过期主套餐并激活下一个待生效主套餐 func (h *PackageActivationHandler) HandlePackageActivationCheck(ctx context.Context) error { startTime := time.Now() // 任务 19.2: 查询已过期的主套餐(status IN (1,2) AND expires_at <= NOW) expiredPackages, err := h.findExpiredMainPackages(ctx) if err != nil { h.logger.Error("查询过期主套餐失败", zap.Error(err)) return err } if len(expiredPackages) > 0 { h.logger.Info("发现过期主套餐", zap.Int("count", len(expiredPackages)), zap.Duration("check_duration", time.Since(startTime))) // 处理每个过期的主套餐 for _, pkg := range expiredPackages { if err := h.processExpiredPackage(ctx, pkg); err != nil { h.logger.Error("处理过期套餐失败", zap.Uint("package_usage_id", pkg.ID), zap.Error(err)) // 继续处理下一个,不中断 continue } } h.logger.Info("套餐激活检查完成", zap.Int("processed", len(expiredPackages)), zap.Duration("total_duration", time.Since(startTime))) } // 任务 6: 孤儿套餐扫描 orphanCount, err := h.findAndActivateOrphanPackages(ctx) if err != nil { h.logger.Error("孤儿套餐扫描失败", zap.Error(err)) // 不中断主流程 } else if orphanCount > 0 { h.logger.Info("孤儿套餐扫描完成", zap.Int("orphan_count", orphanCount)) } return nil } // findAndActivateOrphanPackages 任务 6.1-6.2: 查找孤儿载体并触发激活 // 孤儿定义:存在 status=0 的主套餐,但不存在 status=1 的主套餐。 func (h *PackageActivationHandler) findAndActivateOrphanPackages(ctx context.Context) (int, error) { // 查询孤儿待生效主套餐(无生效主套餐但有待生效主套餐) var orphanUsages []*model.PackageUsage err := h.db.WithContext(ctx). Where("status = ?", constants.PackageUsageStatusPending). Where("master_usage_id IS NULL"). Where("deleted_at IS NULL"). Order("priority ASC, created_at ASC"). Limit(100). Find(&orphanUsages).Error if err != nil { return 0, errors.Wrap(errors.CodeDatabaseError, err, "查询孤儿套餐失败") } if len(orphanUsages) == 0 { return 0, nil } // 按载体分组,去重 type carrierKey struct { carrierType string carrierID uint } carrierMap := make(map[carrierKey]*model.PackageUsage) // 保留 priority 最低的套餐 for _, usage := range orphanUsages { key := carrierKey{} if usage.IotCardID > 0 { key.carrierType = "iot_card" key.carrierID = usage.IotCardID } else if usage.DeviceID > 0 { key.carrierType = "device" key.carrierID = usage.DeviceID } else { continue } // 检查该载体是否已有生效主套餐 var activeCount int64 var countErr error if key.carrierType == "iot_card" { countErr = h.db.WithContext(ctx). Model(&model.PackageUsage{}). Where("status = ?", constants.PackageUsageStatusActive). 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("master_usage_id IS NULL"). Where("device_id = ?", key.carrierID). Count(&activeCount).Error } if countErr != nil { h.logger.Warn("检查载体生效套餐失败", zap.String("carrier_type", key.carrierType), zap.Uint("carrier_id", key.carrierID), zap.Error(countErr)) continue } // 已有生效套餐,跳过 if activeCount > 0 { continue } // 保留购买顺序最靠前的套餐,队首未满足实名条件时不跳过。 if existing, ok := carrierMap[key]; ok { if usage.Priority < existing.Priority || (usage.Priority == existing.Priority && usage.CreatedAt.Before(existing.CreatedAt)) { carrierMap[key] = usage } } else { carrierMap[key] = usage } } // 为每个孤儿载体提交激活任务 count := 0 for key, usage := range carrierMap { if err := h.enqueueActivationTask(ctx, usage.ID, key.carrierType, key.carrierID, "orphan_recovery"); err != nil { h.logger.Warn("提交孤儿套餐激活任务失败", zap.Uint("package_usage_id", usage.ID), zap.String("carrier_type", key.carrierType), zap.Uint("carrier_id", key.carrierID), zap.Error(err)) continue } count++ } return count, nil } // findExpiredMainPackages 任务 19.2: 查询已过期的主套餐 func (h *PackageActivationHandler) findExpiredMainPackages(ctx context.Context) ([]*model.PackageUsage, error) { var packages []*model.PackageUsage now := time.Now() err := h.db.WithContext(ctx). Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}). Where("expires_at <= ?", now). Where("master_usage_id IS NULL"). Limit(1000). Find(&packages).Error return packages, err } // processExpiredPackage 处理单个过期套餐 // triggerStopAfterExpiry 在事务提交后才调用,避免 goroutine 在 TX 提交前读到脏数据 func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pkg *model.PackageUsage) error { carrierType, carrierID := h.getCarrierInfo(pkg) 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 { return err } expiresAt := time.Now() if pkg.ExpiresAt != nil { expiresAt = *pkg.ExpiresAt } h.logger.Info("主套餐已过期", zap.Uint("package_usage_id", pkg.ID), zap.Time("expires_at", expiresAt)) // 任务 19.4: 加油包级联失效 if err := h.invalidateAddons(ctx, tx, pkg.ID); err != nil { h.logger.Warn("加油包级联失效失败", zap.Uint("master_usage_id", pkg.ID), zap.Error(err)) } // 任务 19.5: 查询并激活下一个待生效主套餐 if carrierType != "" && carrierID > 0 { if err := h.activateNextPackage(ctx, tx, carrierType, carrierID); err != nil { h.logger.Warn("激活下一个待生效套餐失败", zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID), zap.Error(err)) } if err := h.updateCarrierSuspendedStatus(ctx, tx, carrierType, carrierID); err != nil { h.logger.Warn("更新载体停用状态失败", zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID), zap.Error(err)) } } return nil }) if err != nil { return err } // 事务提交后再触发异步停机,确保 CheckAndStopCard 读到最新的套餐状态 if carrierType != "" && carrierID > 0 { h.triggerStopAfterExpiry(ctx, carrierType, carrierID) } return nil } // invalidateAddons 任务 19.4: 加油包级联失效 func (h *PackageActivationHandler) invalidateAddons(ctx context.Context, tx *gorm.DB, masterUsageID uint) error { // 查询主套餐下的所有加油包(status IN (0,1,2) 的加油包) result := tx.Model(&model.PackageUsage{}). Where("master_usage_id = ?", masterUsageID). Where("status IN ?", []int{ constants.PackageUsageStatusPending, constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted, }). Update("status", constants.PackageUsageStatusInvalidated) if result.Error != nil { return result.Error } if result.RowsAffected > 0 { h.logger.Info("加油包已级联失效", zap.Uint("master_usage_id", masterUsageID), zap.Int64("invalidated_count", result.RowsAffected)) } return nil } func (h *PackageActivationHandler) updateCarrierSuspendedStatus(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error { if carrierType == constants.AssetTypeIotCard { return h.updateIotCardSuspendedStatus(ctx, tx, carrierID) } if carrierType == constants.AssetTypeDevice { return h.updateDeviceSuspendedStatus(ctx, tx, carrierID) } return nil } func (h *PackageActivationHandler) updateIotCardSuspendedStatus(ctx context.Context, tx *gorm.DB, cardID uint) error { var activeCount int64 if err := tx.WithContext(ctx). Model(&model.PackageUsage{}). Where("iot_card_id = ?", cardID). Where("status = ?", constants.PackageUsageStatusActive). Count(&activeCount).Error; err != nil { return err } if activeCount > 0 { return nil } return tx.WithContext(ctx). Model(&model.IotCard{}). Where("id = ?", cardID). Update("status", constants.IotCardStatusSuspended).Error } func (h *PackageActivationHandler) updateDeviceSuspendedStatus(ctx context.Context, tx *gorm.DB, deviceID uint) error { var activeCount int64 if err := tx.WithContext(ctx). Model(&model.PackageUsage{}). Where("device_id = ?", deviceID). Where("status = ?", constants.PackageUsageStatusActive). Count(&activeCount).Error; err != nil { return err } if activeCount > 0 { return nil } var cardIDs []uint if err := tx.WithContext(ctx). Model(&model.DeviceSimBinding{}). Where("device_id = ?", deviceID). Where("bind_status = ?", constants.BindStatusBound). Pluck("iot_card_id", &cardIDs).Error; err != nil { return err } if len(cardIDs) > 0 { if err := tx.WithContext(ctx). Model(&model.PackageUsage{}). Where("iot_card_id IN ?", cardIDs). Where("status = ?", constants.PackageUsageStatusActive). Count(&activeCount).Error; err != nil { return err } if activeCount > 0 { return nil } } if err := tx.WithContext(ctx). Model(&model.Device{}). Where("id = ?", deviceID). Update("status", constants.DeviceStatusSuspended).Error; err != nil { return err } if len(cardIDs) > 0 { if err := tx.WithContext(ctx). Model(&model.IotCard{}). Where("id IN ?", cardIDs). Update("status", constants.IotCardStatusSuspended).Error; err != nil { return err } } return nil } // getCarrierInfo 获取载体信息 func (h *PackageActivationHandler) getCarrierInfo(pkg *model.PackageUsage) (string, uint) { if pkg.IotCardID > 0 { return "iot_card", pkg.IotCardID } if pkg.DeviceID > 0 { return "device", pkg.DeviceID } return "", 0 } // activateNextPackage 任务 19.5: 激活下一个待生效主套餐 func (h *PackageActivationHandler) activateNextPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error { // 查询下一个待生效主套餐 // WHERE status=0 AND master_usage_id IS NULL ORDER BY priority ASC LIMIT 1 var nextPkg model.PackageUsage query := tx.Where("status = ?", constants.PackageUsageStatusPending). Where("master_usage_id IS NULL"). // 主套餐 Order("priority ASC"). Limit(1) if carrierType == "iot_card" { query = query.Where("iot_card_id = ?", carrierID) } else if carrierType == "device" { query = query.Where("device_id = ?", carrierID) } if err := query.First(&nextPkg).Error; err != nil { if err == gorm.ErrRecordNotFound { // 没有待生效套餐,正常情况 return nil } return err } // 提交 Asynq 任务进行激活(避免长事务) return h.enqueueActivationTask(ctx, nextPkg.ID, carrierType, carrierID, "queue") } // triggerStopAfterExpiry 套餐过期后异步触发停机检查 // 仅在确认无后续生效套餐时有效;CheckAndStopCard 内部有幂等保护,重复调用安全 func (h *PackageActivationHandler) triggerStopAfterExpiry(ctx context.Context, carrierType string, carrierID uint) { if h.stopResumeCallback == nil { return } if carrierType == "iot_card" { go func() { if err := h.stopResumeCallback.CheckAndStopCard(context.Background(), carrierID); err != nil { h.logger.Error("套餐过期后停机失败", zap.Uint("card_id", carrierID), zap.Error(err)) } }() return } if carrierType == "device" { bindings, err := h.deviceSimBinding.ListByDeviceID(ctx, carrierID) if err != nil { h.logger.Error("查询设备绑定卡失败", zap.Uint("device_id", carrierID), zap.Error(err)) return } for _, b := range bindings { cardID := b.IotCardID go func(cID uint) { if err := h.stopResumeCallback.CheckAndStopCard(context.Background(), cID); err != nil { h.logger.Error("套餐过期后停机失败", zap.Uint("card_id", cID), zap.Error(err)) } }(cardID) } } } // enqueueActivationTask 提交套餐激活任务到 Asynq func (h *PackageActivationHandler) enqueueActivationTask(ctx context.Context, packageUsageID uint, carrierType string, carrierID uint, activationType string) error { payload := PackageActivationPayload{ PackageUsageID: packageUsageID, CarrierType: carrierType, CarrierID: carrierID, ActivationType: activationType, Timestamp: time.Now().Unix(), } payloadBytes, err := sonic.Marshal(payload) if err != nil { return err } task := asynq.NewTask(constants.TaskTypePackageQueueActivation, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(30*time.Second), asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageQueueActivation)), ) _, err = h.queueClient.Enqueue(task) if err != nil { h.logger.Error("提交套餐激活任务失败", zap.Uint("package_usage_id", packageUsageID), zap.Error(err)) return err } h.logger.Info("已提交套餐激活任务", zap.Uint("package_usage_id", packageUsageID), zap.String("activation_type", activationType)) return nil } // HandlePackageQueueActivation 处理套餐排队激活任务(Asynq Handler) // 任务 23: 由 Asynq 调用,执行实际的套餐激活逻辑 func (h *PackageActivationHandler) HandlePackageQueueActivation(ctx context.Context, t *asynq.Task) error { var payload PackageActivationPayload if err := sonic.Unmarshal(t.Payload(), &payload); err != nil { h.logger.Error("解析套餐激活任务载荷失败", zap.Error(err)) return nil // 不重试 } h.logger.Info("开始执行套餐激活", zap.Uint("package_usage_id", payload.PackageUsageID), zap.String("activation_type", payload.ActivationType)) // 查询套餐使用记录 var pkg model.PackageUsage if err := h.db.First(&pkg, payload.PackageUsageID).Error; err != nil { if err == gorm.ErrRecordNotFound { h.logger.Warn("套餐使用记录不存在", zap.Uint("package_usage_id", payload.PackageUsageID)) return nil } return err } // 幂等性检查:如果已经是生效状态,跳过 if pkg.Status == constants.PackageUsageStatusActive { h.logger.Info("套餐已激活,跳过", zap.Uint("package_usage_id", payload.PackageUsageID)) return nil } // 调用 ActivationService 执行激活 if h.activationService != nil { if err := h.activationService.ActivateSpecificPackage(ctx, payload.PackageUsageID); err != nil { h.logger.Error("套餐激活失败", zap.Uint("package_usage_id", payload.PackageUsageID), zap.Error(err)) return err } } else { // ActivationService 未注入,无法安全激活(缺少 expires_at 计算) h.logger.Error("激活服务未注入,无法执行套餐激活", zap.Uint("package_usage_id", payload.PackageUsageID)) return errors.New(errors.CodeInternalError, "激活服务未注入,无法执行套餐激活") } h.logger.Info("套餐激活成功", zap.Uint("package_usage_id", payload.PackageUsageID)) return nil } // HandlePackageFirstActivation 处理首次实名激活任务(Asynq Handler) // 由 polling_realname_handler 在检测到首次实名时触发。 // payload 中仅需 carrier_type + carrier_id,ActivateByRealname 内部自行查找 // pending_realname_activation=true 的套餐,无需外部指定 PackageUsageID。 func (h *PackageActivationHandler) HandlePackageFirstActivation(ctx context.Context, t *asynq.Task) error { var payload PackageActivationPayload if err := sonic.Unmarshal(t.Payload(), &payload); err != nil { h.logger.Error("解析首次实名激活任务载荷失败", zap.Error(err)) return nil } if payload.CarrierType == "" || payload.CarrierID == 0 { h.logger.Error("首次实名激活任务 carrier 信息缺失", zap.String("carrier_type", payload.CarrierType), zap.Uint("carrier_id", payload.CarrierID)) return nil } h.logger.Info("开始执行首次实名激活", zap.String("carrier_type", payload.CarrierType), zap.Uint("carrier_id", payload.CarrierID)) if h.activationService == nil { h.logger.Warn("ActivationService 未注入,跳过首次实名激活") return nil } if err := h.activationService.ActivateByRealname(ctx, payload.CarrierType, payload.CarrierID); err != nil { h.logger.Error("首次实名激活失败", zap.String("carrier_type", payload.CarrierType), zap.Uint("carrier_id", payload.CarrierID), zap.Error(err)) return err } h.logger.Info("首次实名激活成功", zap.String("carrier_type", payload.CarrierType), zap.Uint("carrier_id", payload.CarrierID)) return nil }