diff --git a/cmd/worker/carrier_scheduler.go b/cmd/worker/carrier_scheduler.go index 07af0487..075db499 100644 --- a/cmd/worker/carrier_scheduler.go +++ b/cmd/worker/carrier_scheduler.go @@ -2,13 +2,17 @@ package main import ( "context" + "encoding/json" "os" "strconv" + "strings" "sync" "time" "go.uber.org/zap" + cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation" + "github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox" "github.com/break/junhong_cmp_fiber/internal/infrastructure/nsqbus" "github.com/break/junhong_cmp_fiber/internal/polling" "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" @@ -226,3 +230,41 @@ func (c *carrierSchedulerRuntime) shutdown() { func (c *carrierSchedulerRuntime) InitProgress() polling.InitProgress { return c.initializer.Progress() } + +// carrierAwareSeriesConsumer 是业务事件观测序列 Outbox 消费者的模式接缝:按运营商账户调度时把业务事件改为按场景请求检查 +// (事件在原业务事务中写入,消费即在事务提交之后),旧分片调度时沿用原观测序列。 +type carrierAwareSeriesConsumer struct { + legacy outbox.EventConsumer + requester *polling.CheckRequester +} + +// Consume 消费业务观测序列请求事件。 +func (c *carrierAwareSeriesConsumer) Consume(ctx context.Context, envelope outbox.DeliveryEnvelope) error { + if c.requester == nil || !c.requester.Active(ctx) { + return c.legacy.Consume(ctx, envelope) + } + var event cardObservationApp.SeriesRequestedEvent + if err := json.Unmarshal(envelope.Payload, &event); err != nil { + return nil + } + cardIDs := []uint{event.ResourceID} + if event.ResourceType == constants.CardObservationResourceTypeDevice { + cardIDs = event.ResourceIDs + } + c.requester.RequestCards(ctx, businessEventCheckScene(event), cardIDs, event.ExpectedValue) + return nil +} + +// businessEventCheckScene 把业务观测事件映射为请求检查场景:订单支付、套餐激活、自动续费购包按事件来源区分,停复机成功按观测场景。 +func businessEventCheckScene(event cardObservationApp.SeriesRequestedEvent) string { + switch { + case strings.HasPrefix(event.EventID, "card-observation:order:"): + return constants.PollingCheckSceneOrderPaid + case strings.HasPrefix(event.EventID, "card-observation:auto-purchase:"), strings.HasPrefix(event.EventID, "asset-auto-renewal:"): + return constants.PollingCheckSceneAutoRenewalPurchase + case event.Scene == constants.CardObservationSceneBusinessStop || event.Scene == constants.CardObservationSceneBusinessResume: + return constants.PollingCheckSceneStopResumeSucceeded + default: + return constants.PollingCheckScenePackageActivated + } +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 02c5f5e6..94604a5d 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -726,7 +726,10 @@ func registerCardObservationOutboxConsumer(runtime *workerRuntime, appLogger *za queue.NewCardObservationSeriesScheduler(runtime.outboxQueueClient), cardObservationInfra.NewSeriesAttemptLogger(integrationlog.NewRepository(runtime.db)), ) - seriesConsumer := cardObservationInfra.NewSeriesRequestedConsumer(seriesTrigger) + seriesConsumer := &carrierAwareSeriesConsumer{ + legacy: cardObservationInfra.NewSeriesRequestedConsumer(seriesTrigger), + requester: polling.NewCheckRequester(runtime.carrierScheduling, runtime.db, appLogger), + } if err := runtime.outboxConsumers.Register(constants.OutboxEventTypeCardSeriesRequested, seriesConsumer); err != nil { appLogger.Fatal("注册业务卡观测序列 Outbox 消费者失败", zap.String("event_type", constants.OutboxEventTypeCardSeriesRequested), zap.Error(err)) diff --git a/internal/bootstrap/carrier_check.go b/internal/bootstrap/carrier_check.go new file mode 100644 index 00000000..b679ac9f --- /dev/null +++ b/internal/bootstrap/carrier_check.go @@ -0,0 +1,121 @@ +package bootstrap + +import ( + "context" + "strconv" + "strings" + + cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation" + "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// carrierAwareSeriesDispatcher 是读入口观测分发的模式接缝:按运营商账户调度时把读入口改为按场景请求检查 +// (同卡合并、冷却、受账户限流),不再起"立即、3 分钟、5 分钟"观测序列;旧分片调度时沿用原分发器。 +type carrierAwareSeriesDispatcher struct { + legacy cardObservationApp.BestEffortSeriesDispatcher + requester *polling.CheckRequester +} + +var _ cardObservationApp.BestEffortSeriesDispatcher = (*carrierAwareSeriesDispatcher)(nil) + +// ReadEntryCheckScene 把读入口与设备控制的观测场景映射为请求检查场景;不认识的场景返回空串。 +func ReadEntryCheckScene(observationScene string) string { + switch observationScene { + case constants.CardObservationSceneClientAssetRead: + return constants.PollingCheckSceneClientAssetDetail + case constants.CardObservationSceneAdminAssetRead: + return constants.PollingCheckSceneAdminAssetRealtime + case constants.CardObservationSceneClientRealnameLink, constants.CardObservationSceneAdminRealnameLink: + return constants.PollingCheckSceneRealnameLink + case constants.CardObservationSceneBusinessStop, constants.CardObservationSceneBusinessResume: + return constants.PollingCheckSceneStopResumeSucceeded + case constants.CardObservationScenePackageChanged: + return constants.PollingCheckScenePackageActivated + case constants.CardObservationSceneDeviceSwitchCard, constants.CardObservationSceneDeviceSwitchMode, + constants.CardObservationSceneDeviceReboot, constants.CardObservationSceneDeviceReset, constants.CardObservationSceneDeviceSetWiFi: + return constants.PollingCheckSceneDeviceOperation + default: + return "" + } +} + +func (d *carrierAwareSeriesDispatcher) active(ctx context.Context) bool { + return d.requester != nil && d.requester.Active(ctx) +} + +// Dispatch 读入口单卡(或设备)观测。 +func (d *carrierAwareSeriesDispatcher) Dispatch(ctx context.Context, request cardObservationApp.SeriesRequest) { + if !d.active(ctx) { + d.legacy.Dispatch(ctx, request) + return + } + scene := ReadEntryCheckScene(request.Scene) + if scene == "" { + return + } + id, err := strconv.ParseUint(request.ResourceID, 10, 64) + if err != nil || id == 0 { + return + } + if request.ResourceType == constants.CardObservationResourceTypeDevice { + d.requester.RequestDevice(ctx, scene, uint(id)) + return + } + d.requester.RequestCards(ctx, scene, []uint{uint(id)}, request.ExpectedValue) +} + +// DispatchDeviceCards 设备有效绑定卡观测。 +func (d *carrierAwareSeriesDispatcher) DispatchDeviceCards(ctx context.Context, request cardObservationApp.DeviceCardsSeriesRequest) { + if !d.active(ctx) { + d.legacy.DispatchDeviceCards(ctx, request) + return + } + if scene := ReadEntryCheckScene(request.Request.Scene); scene != "" { + d.requester.RequestDevice(ctx, scene, request.DeviceID) + } +} + +// DispatchDeviceControl 设备控制成功后的观测:按设备操作场景为设备下各卡请求检查。 +func (d *carrierAwareSeriesDispatcher) DispatchDeviceControl(ctx context.Context, request cardObservationApp.DeviceControlSeriesRequest) { + if !d.active(ctx) { + d.legacy.DispatchDeviceControl(ctx, request) + return + } + // 设备信息同步(query_device_info,非运营商上游)保持原观测,只把卡检查改为请求检查。 + deviceInfo := request.Request + deviceInfo.ResourceType = constants.CardObservationResourceTypeDevice + deviceInfo.ResourceID = strconv.FormatUint(uint64(request.DeviceID), 10) + deviceInfo.SyncType = constants.CardObservationSyncTypeDeviceInfo + deviceInfo.ExpectedValue = strings.TrimSpace(request.TargetICCID) + d.legacy.Dispatch(ctx, deviceInfo) + cardIDs := append([]uint(nil), request.BoundCardIDs...) + for _, id := range []uint{request.SourceCardID, request.TargetCardID} { + if id != 0 && !containsUint(cardIDs, id) { + cardIDs = append(cardIDs, id) + } + } + if len(cardIDs) == 0 && request.DeviceID != 0 { + d.requester.RequestDevice(ctx, constants.PollingCheckSceneDeviceOperation, request.DeviceID) + return + } + d.requester.RequestCards(ctx, constants.PollingCheckSceneDeviceOperation, cardIDs, "") +} + +// DispatchRealnameWithCapability 获取实名链接后的实名观测。 +func (d *carrierAwareSeriesDispatcher) DispatchRealnameWithCapability(ctx context.Context, request cardObservationApp.RealnameCapabilitySeriesRequest) { + if !d.active(ctx) { + d.legacy.DispatchRealnameWithCapability(ctx, request) + return + } + d.Dispatch(ctx, request.Request) +} + +func containsUint(values []uint, target uint) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 827d33f5..355f6156 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -222,7 +222,8 @@ func initServices(s *stores, deps *Dependencies) *services { cardObservationInfra.NewSeriesRunner(deps.DB, deps.GatewayClient, cardObservationService, seriesIntegration, auditWriter), cardObservationInfra.NewSeriesAttemptLogger(seriesIntegration), ) - observationSeries := cardObservationInfra.NewBestEffortSeriesDispatcher(seriesTrigger, deps.Logger, s.DeviceSimBinding, s.Carrier) + // 读入口观测分发经模式接缝:按运营商账户调度时改为按场景请求检查,旧分片调度时沿用原分发器。 + observationSeries := &carrierAwareSeriesDispatcher{legacy: cardObservationInfra.NewBestEffortSeriesDispatcher(seriesTrigger, deps.Logger, s.DeviceSimBinding, s.Carrier)} iotCard.SetObservationSeriesDispatcher(observationSeries) // 使用 PollingLifecycleService 替代 APICallback:通过分片队列准确操作(修复 api_callback.go 遗漏 protect 队列的 Bug3) pollingConfigStore := postgres.NewPollingConfigStore(deps.DB) @@ -238,7 +239,9 @@ func initServices(s *stores, deps *Dependencies) *services { // 按运营商账户调度:模式为 carrier 时 API 侧生命周期事件写入账户队列与卡到期信息。 carrierSettings := carrierqueue.NewDefaultSettingsProvider(deps.DB, deps.Redis, deps.Logger) carrierStore := carrierqueue.NewStore(deps.Redis) - pollingLifecycleSvc.SetCarrierScheduling(polling.NewCarrierScheduling(carrierStore, carrierSettings, pollingConfigMgr, deps.Logger)) + carrierScheduling := polling.NewCarrierScheduling(carrierStore, carrierSettings, pollingConfigMgr, deps.Logger) + pollingLifecycleSvc.SetCarrierScheduling(carrierScheduling) + observationSeries.requester = polling.NewCheckRequester(carrierScheduling, deps.DB, deps.Logger) iotCard.SetPollingCallback(pollingLifecycleSvc) // 创建支付配置服务(Order 和 Recharge 依赖) wechatConfig := wechatConfigSvc.New(s.WechatConfig, s.Order, s.RechargeOrder, s.AgentRecharge, s.Payment, auditWriter, deps.Redis, deps.Logger) diff --git a/internal/polling/check_requester.go b/internal/polling/check_requester.go new file mode 100644 index 00000000..a4679ff9 --- /dev/null +++ b/internal/polling/check_requester.go @@ -0,0 +1,161 @@ +package polling + +import ( + "context" + "encoding/json" + "strings" + "sync" + "time" + + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// sceneCacheTTL 是请求检查场景配置的进程内缓存时长。 +const sceneCacheTTL = 30 * time.Second + +// SceneConfig 是一个请求检查场景的生效配置。 +type SceneConfig struct { + Scene string + CheckTypes []string + ConfirmDelays []time.Duration + Cooldown time.Duration +} + +// CheckRequester 按场景配置为卡请求检查(业务事件与读入口):把对应检查类型设为立即到期放入账户优先队列, +// 同卡多次请求合并、冷却期内不请求、按场景追加延迟确认。只在按运营商账户调度时生效。 +type CheckRequester struct { + scheduling *CarrierScheduling + db *gorm.DB + logger *zap.Logger + + mu sync.Mutex + scenes map[string]SceneConfig + loadedAt time.Time +} + +// NewCheckRequester 创建请求检查器。 +func NewCheckRequester(scheduling *CarrierScheduling, db *gorm.DB, logger *zap.Logger) *CheckRequester { + if logger == nil { + logger = zap.NewNop() + } + return &CheckRequester{scheduling: scheduling, db: db, logger: logger} +} + +// Active 判断当前是否按运营商账户调度(请求检查生效)。 +func (r *CheckRequester) Active(ctx context.Context) bool { + return r != nil && r.scheduling.Active(ctx) +} + +// RequestCards 为一组卡按场景请求检查;expected 为延迟确认的期望结果(online / offline,可空)。 +// 失败只记日志,不影响业务。 +func (r *CheckRequester) RequestCards(ctx context.Context, scene string, cardIDs []uint, expected string) { + if r == nil || len(cardIDs) == 0 { + return + } + config, ok := r.sceneConfig(ctx, scene) + if !ok || len(config.CheckTypes) == 0 { + return + } + carriers, err := r.cardCarriers(ctx, cardIDs) + if err != nil { + r.logger.Warn("请求检查:查询卡所属运营商账户失败", zap.String("scene", scene), zap.Error(err)) + return + } + store := r.scheduling.Store() + for _, cardID := range cardIDs { + carrierID, found := carriers[cardID] + if !found { + continue + } + if _, err := store.RequestCheck(ctx, carrierqueue.CheckRequest{ + CardID: cardID, CarrierID: carrierID, CheckTypes: config.CheckTypes, Cooldown: config.Cooldown, + ConfirmDelays: config.ConfirmDelays, Expected: expected, + }); err != nil { + r.logger.Warn("请求检查写入账户队列失败", zap.Uint("card_id", cardID), zap.String("scene", scene), zap.Error(err)) + } + } +} + +// RequestDevice 为设备下全部有效绑定卡按场景请求检查。 +func (r *CheckRequester) RequestDevice(ctx context.Context, scene string, deviceID uint) { + if r == nil || deviceID == 0 { + return + } + var cardIDs []uint + if err := r.db.WithContext(ctx).Model(&model.DeviceSimBinding{}). + Where("device_id = ? AND bind_status = ?", deviceID, constants.BindStatusBound).Pluck("iot_card_id", &cardIDs).Error; err != nil { + r.logger.Warn("请求检查:查询设备绑定卡失败", zap.Uint("device_id", deviceID), zap.Error(err)) + return + } + r.RequestCards(ctx, scene, cardIDs, "") +} + +func (r *CheckRequester) cardCarriers(ctx context.Context, cardIDs []uint) (map[uint]uint, error) { + var cards []model.IotCard + if err := r.db.WithContext(ctx).Select("id", "carrier_id").Where("id IN ?", cardIDs).Find(&cards).Error; err != nil { + return nil, err + } + result := make(map[uint]uint, len(cards)) + for _, card := range cards { + result[card.ID] = card.CarrierID + } + return result, nil +} + +// sceneConfig 读取场景配置(缓存);场景未预置时不请求。 +func (r *CheckRequester) sceneConfig(ctx context.Context, scene string) (SceneConfig, bool) { + r.mu.Lock() + defer r.mu.Unlock() + if r.scenes == nil || time.Since(r.loadedAt) >= sceneCacheTTL { + var rows []model.PollingCheckScene + if err := r.db.WithContext(ctx).Find(&rows).Error; err != nil { + r.logger.Warn("读取请求检查场景失败,沿用上一次配置", zap.Error(err)) + } else { + scenes := make(map[string]SceneConfig, len(rows)) + for _, row := range rows { + scenes[row.Scene] = ParseSceneConfig(row) + } + r.scenes, r.loadedAt = scenes, time.Now() + } + } + config, ok := r.scenes[scene] + return config, ok +} + +// ParseSceneConfig 解析场景配置行。 +func ParseSceneConfig(row model.PollingCheckScene) SceneConfig { + config := SceneConfig{Scene: row.Scene} + var types []string + _ = json.Unmarshal(row.CheckTypes, &types) + for _, checkType := range types { + if constants.CheckTypeCallsGateway(strings.TrimSpace(checkType)) { + config.CheckTypes = append(config.CheckTypes, strings.TrimSpace(checkType)) + } + } + var delays []int + _ = json.Unmarshal(row.ConfirmDelays, &delays) + for _, seconds := range delays { + if seconds > 0 { + config.ConfirmDelays = append(config.ConfirmDelays, time.Duration(seconds)*time.Second) + } + } + if row.Cooldown != nil && *row.Cooldown > 0 { + config.Cooldown = time.Duration(*row.Cooldown) * time.Second + } + return config +} + +// InvalidateScenes 使场景缓存立即失效(后台修改后调用)。 +func (r *CheckRequester) InvalidateScenes() { + if r == nil { + return + } + r.mu.Lock() + r.loadedAt = time.Time{} + r.mu.Unlock() +}