diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 741f057..0427f16 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -262,6 +262,9 @@ func initWorkerRuntime(ctx context.Context, cfg *config.Config, appLogger *zap.L pollingDeviceStore, appLogger, ) + if stopResumeSvc, ok := workerResult.Services.StopResumeService.(*iot_card_svc.StopResumeService); ok { + stopResumeSvc.SetPollingCallback(lifecycleSvc) + } return &workerRuntime{ redisAddr: redisAddr, diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 867df44..91334e7 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -138,7 +138,8 @@ func initServices(s *stores, deps *Dependencies) *services { } pollingConfigMgr.Start(context.Background()) pollingQueueMgr := polling.NewPollingQueueManager(deps.Redis, constants.PollingShardCount, deps.Logger) - iotCard.SetPollingCallback(polling.NewPollingLifecycleService(pollingQueueMgr, pollingConfigMgr, s.IotCard, s.DeviceSimBinding, s.Device, deps.Logger)) + pollingLifecycleSvc := polling.NewPollingLifecycleService(pollingQueueMgr, pollingConfigMgr, s.IotCard, s.DeviceSimBinding, s.Device, deps.Logger) + iotCard.SetPollingCallback(pollingLifecycleSvc) // 注入流量扣减回调,使手动刷新资产时能触发套餐流量扣减 usageService := packageSvc.NewUsageService(deps.DB, deps.Redis, s.PackageUsage, s.PackageUsageDailyRecord, s.DeviceSimBinding, deps.Logger) iotCard.SetDataDeductor(usageService) @@ -167,6 +168,7 @@ func initServices(s *stores, deps *Dependencies) *services { deps.Logger, assetAudit, ) + stopResumeService.SetPollingCallback(pollingLifecycleSvc) iotCard.SetRealnameActivator(packageActivation) iotCard.SetStopResumeService(stopResumeService) iotCard.SetDeviceSimBindingStore(s.DeviceSimBinding) diff --git a/internal/service/iot_card/stop_resume_service.go b/internal/service/iot_card/stop_resume_service.go index 26239b4..4e7f7e1 100644 --- a/internal/service/iot_card/stop_resume_service.go +++ b/internal/service/iot_card/stop_resume_service.go @@ -38,6 +38,7 @@ type StopResumeService struct { gatewayClient *gateway.Client logger *zap.Logger assetAuditService AssetAuditService + pollingCallback PollingCallback maxRetries int retryInterval time.Duration @@ -66,6 +67,12 @@ func NewStopResumeService( } } +// SetPollingCallback 设置轮询生命周期回调。 +// 停复机成功改变卡状态后,需要立即按新状态重排轮询队列,避免继续使用旧 score。 +func (s *StopResumeService) SetPollingCallback(callback PollingCallback) { + s.pollingCallback = callback +} + // EvaluateAndAct 停复机统一入口 // 根据卡的当前网络状态,自动判断并执行停机或复机操作 // 设备/单卡维度通过 card.IsStandalone 推导(false 则为设备维度,停/复机覆盖设备下所有卡) @@ -476,7 +483,7 @@ func (s *StopResumeService) resumeSingleCard(ctx context.Context, cardID uint) e return nil } - s.invalidatePollingCache(ctx, card.ID) + s.reschedulePolling(ctx, card.ID) s.logger.Info("卡已自动复机", zap.Uint("card_id", cardID), @@ -583,7 +590,7 @@ func (s *StopResumeService) stopCardWithRetry(ctx context.Context, card *model.I }) } - s.invalidatePollingCache(ctx, card.ID) + s.reschedulePolling(ctx, card.ID) s.logCardAudit(ctx, assetAuditSvc.BuildLogParams{ Operator: operator, OperationType: operationType, @@ -752,7 +759,7 @@ func (s *StopResumeService) StartMachineSeparatedCard(ctx context.Context, card return wrapErr } - s.invalidatePollingCache(ctx, card.ID) + s.reschedulePolling(ctx, card.ID) s.logCardAudit(ctx, assetAuditSvc.BuildLogParams{ OperationType: constants.AssetAuditOpCardManualStart, OperationDesc: "机卡分离复机", @@ -971,7 +978,7 @@ func (s *StopResumeService) ManualStartCard(ctx context.Context, iccid string) e return wrapErr } - s.invalidatePollingCache(ctx, card.ID) + s.reschedulePolling(ctx, card.ID) s.logCardAudit(ctx, assetAuditSvc.BuildLogParams{ OperationType: constants.AssetAuditOpCardManualStart, @@ -1002,3 +1009,13 @@ func (s *StopResumeService) invalidatePollingCache(ctx context.Context, cardID u s.logger.Warn("删除轮询卡缓存失败", zap.Uint("card_id", cardID)) } } + +// reschedulePolling 按卡最新状态重排轮询队列。 +// 未注入生命周期回调时退化为只清缓存,保持原有行为。 +func (s *StopResumeService) reschedulePolling(ctx context.Context, cardID uint) { + if s.pollingCallback == nil { + s.invalidatePollingCache(ctx, cardID) + return + } + s.pollingCallback.OnCardStatusChanged(ctx, cardID) +}