feat: 业务事件与读入口在按运营商账户调度时改为请求检查

- 请求检查器按场景配置（tb_polling_check_scene，缓存 30 秒）把卡的检查类型设为立即到期放入账户优先队列，同卡合并、冷却期内不请求、按场景追加延迟确认；设备按有效绑定卡展开 - 业务事件仍在原事务中写 Outbox（提交安全），carrier 模式下 Outbox 消费者把订单支付、套餐激活、自动续费购包、停复机成功事件转为请求检查，不再起三连查观测序列 - 读入口（C 端资产详情、后台资产实时状态、实名链接、设备操作）经模式接缝改为请求检查；设备信息同步保持原观测；OpenAPI 查询仍不触发 - legacy 模式下全部沿用原观测序列
2026-10-25 18:49:36 +09:00
parent 2fe7c1f8ac
commit 0d966249b8
5 changed files with 333 additions and 3 deletions
--- 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
+	}
+}
--- 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))
--- a/internal/bootstrap/carrier_check.go
+++ 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
+}
--- 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)
--- a/internal/polling/check_requester.go
+++ 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()
+}