junhong_cmp_fiber/cmd/worker/carrier_scheduler.go

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/gatewaylimit"
	"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/model"
	"github.com/break/junhong_cmp_fiber/internal/polling"
	"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
	"github.com/break/junhong_cmp_fiber/internal/task"
	"github.com/break/junhong_cmp_fiber/pkg/config"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/queue"
)

// carrierInitLockKey 新结构初始化锁：多实例时只由一个实例初始化。
const carrierInitLockKey = "polling:carrier:init_lock"

// carrierSchedulerRuntime 是按运营商账户调度在 Worker 内的运行组件：模式监听、初始化、分发器与 NSQ 消费者。
type carrierSchedulerRuntime struct {
	cfg         config.CarrierSchedulerConfig
	runtime     *workerRuntime
	initializer *polling.CarrierInitializer
	dispatcher  *carrierqueue.Dispatcher
	consumers   *nsqbus.ConsumerManager
	producer    *nsqbus.Producer
	alerts      *carrierqueue.AlertWriter
	executor    *task.CarrierCardExecutor
	// 停复机指令与设备判断的专用消费者：只看停复机出口开关，不受 scheduler_mode 限制。
	srConsumers  *nsqbus.ConsumerManager
	devConsumers *nsqbus.ConsumerManager
	srActive     bool
	// ingest 是轮询观测入库消费者与积压监控（add-polling-async-ingest）；未配置 nsqd 时为 nil。
	ingest     *pollingIngestRuntime
	singleton  bool
	instanceID string
	logger     *zap.Logger

	mu             sync.Mutex
	lastMode       string
	reinitQueued   bool
	legacyRestarts func()
}

// newCarrierSchedulerParts 在 Worker 运行时初始化阶段创建调度参数、账户队列存储与模式接缝，
// 并注入旧调度相关组件（生命周期、轮询基类），使其在启动前即可按模式写入正确结构。
func newCarrierSchedulerParts(runtime *workerRuntime, appLogger *zap.Logger) {
	settings := carrierqueue.NewDefaultSettingsProvider(runtime.db, runtime.redisClient, appLogger)
	store := carrierqueue.NewStore(runtime.redisClient)
	scheduling := polling.NewCarrierScheduling(store, settings, runtime.pollingConfigMgr, appLogger)
	runtime.carrierSettings = settings
	runtime.carrierStore = store
	runtime.carrierScheduling = scheduling
	runtime.lifecycleSvc.SetCarrierScheduling(scheduling)
	runtime.pollingBase.SetSchedulerModeReader(func(ctx context.Context) string { return settings.Current(ctx).Mode })
	if stopResumeSvc, ok := runtime.workerResult.Services.StopResumeService.(*iot_card_svc.StopResumeService); ok {
		stopResumeSvc.SetCommandQueue(carrierqueue.NewCommandQueue(store, settings))
	}
	if runtime.gatewayLimit != nil && runtime.gatewayLimit.Classifier != nil {
		classifier := runtime.gatewayLimit.Classifier
		runtime.pollingBase.SetErrorClassifier(func(ctx context.Context, op, message string) string {
			return classifier.Classify(ctx, op, message).Category
		})
	}
}

// startCarrierScheduler 启动模式监听；分发器只在 leader/all 运行（多实例安全），消费者在所有处理任务的实例运行。
func startCarrierScheduler(ctx context.Context, cfg *config.Config, runtime *workerRuntime, steps queue.PollingStepHandlers,
	singleton bool, legacyRestart func(), appLogger *zap.Logger) *carrierSchedulerRuntime {
	carrierCfg := cfg.Polling.Carrier
	instanceID := cfg.Worker.InstanceName
	if instanceID == "" {
		instanceID, _ = os.Hostname()
	}
	instanceID += "-" + strconv.Itoa(os.Getpid())
	alerts := carrierqueue.NewAlertWriter(runtime.db, runtime.redisClient, runtime.carrierSettings, appLogger)
	alerts.Start(ctx)
	executor := task.NewCarrierCardExecutor(runtime.pollingBase, runtime.carrierStore, runtime.carrierSettings, alerts,
		steps.Realname, steps.Carddata, steps.CardStatus, steps.Package, carrierCfg, instanceID)
	c := &carrierSchedulerRuntime{
		cfg: carrierCfg, runtime: runtime, singleton: singleton, instanceID: instanceID, logger: appLogger,
		alerts: alerts, executor: executor, legacyRestarts: legacyRestart,
		initializer: polling.NewCarrierInitializer(runtime.carrierScheduling, runtime.pollingIotCardStore, appLogger),
	}
	if carrierCfg.NSQDAddress != "" {
		producer, err := nsqbus.NewProducer(carrierCfg.NSQDAddress, appLogger)
		if err != nil {
			appLogger.Error("创建 NSQ 生产者失败，按运营商账户调度不可用", zap.Error(err))
		} else {
			c.producer = producer
		}
		c.consumers = nsqbus.NewConsumerManager(nsqbus.ConsumerConfig{
			NSQDAddress: carrierCfg.NSQDAddress, NSQLookupdAddresses: nsqbus.SplitAddresses(carrierCfg.NSQLookupdAddresses),
			MsgTimeout: carrierCfg.MsgTimeout, MaxAttempts: carrierCfg.MaxAttempts,
		}, constants.NSQChannelWorker, executor.Handle, appLogger)
	}
	c.ingest = startPollingIngest(ctx, cfg, runtime, c.producer, singleton, c.grace(), appLogger)
	c.setupStopResume(carrierCfg, alerts)
	var budgets carrierqueue.BudgetProvider
	if runtime.gatewayLimit != nil && runtime.gatewayLimit.Limiter != nil {
		budgets = runtime.gatewayLimit.Limiter
	}
	if c.producer != nil {
		c.dispatcher = carrierqueue.NewDispatcher(runtime.db, runtime.carrierStore, runtime.carrierSettings, c.producer, budgets,
			carrierCfg, instanceID, appLogger)
		if singleton {
			go c.dispatcher.Run(ctx)
		}
	}
	go c.watch(ctx)
	return c
}

// RequestReinit 轮询配置变更时请求按新配置重新初始化新结构（仅 carrier 模式生效）。
func (c *carrierSchedulerRuntime) RequestReinit() {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.reinitQueued = true
	c.mu.Unlock()
}

func (c *carrierSchedulerRuntime) watch(ctx context.Context) {
	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	for {
		c.step(ctx)
		select {
		case <-ctx.Done():
			c.shutdown()
			return
		case <-ticker.C:
		}
	}
}

func (c *carrierSchedulerRuntime) step(ctx context.Context) {
	defer func() {
		if r := recover(); r != nil {
			c.logger.Error("按运营商账户调度模式监听 panic，已恢复", zap.Any("panic", r))
		}
	}()
	c.syncStopResume(ctx)
	mode := c.runtime.carrierSettings.Current(ctx).Mode
	c.mu.Lock()
	previous := c.lastMode
	c.lastMode = mode
	reinit := c.reinitQueued
	c.reinitQueued = false
	c.mu.Unlock()
	if mode == constants.PollingSchedulerModeCarrier {
		if previous != mode {
			c.logger.Info("卡轮询调度模式：按运营商账户调度", zap.String("previous", previous))
		}
		if c.singleton {
			c.ensureInitialized(ctx, reinit)
		}
		c.syncConsumers(ctx)
		return
	}
	if previous == constants.PollingSchedulerModeCarrier {
		c.logger.Info("卡轮询调度模式回到旧分片调度：停止领取与消费，重建旧分片队列")
		if c.consumers != nil {
			c.consumers.StopAll(c.grace())
			c.consumers.Reset()
		}
		if c.singleton {
			c.runtime.redisClient.Del(ctx, constants.RedisCarrierSchedulerInitStateKey)
			if c.legacyRestarts != nil {
				c.legacyRestarts()
			}
		}
	}
}

// ensureInitialized 切到 carrier 后初始化新结构；初始化完成前分发器不领取。多实例时用初始化锁协调。
func (c *carrierSchedulerRuntime) ensureInitialized(ctx context.Context, reinit bool) {
	redisClient := c.runtime.redisClient
	state, _ := redisClient.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result()
	if state == carrierqueue.InitStateDone && !reinit {
		return
	}
	ok, err := redisClient.SetNX(ctx, carrierInitLockKey, c.instanceID, 30*time.Minute).Result()
	if err != nil || !ok {
		return
	}
	defer redisClient.Del(context.WithoutCancel(ctx), carrierInitLockKey)
	if state != carrierqueue.InitStateDone {
		redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateRunning, 0)
	}
	started := time.Now()
	written, initErr := c.initializer.Run(ctx)
	if initErr != nil {
		c.logger.Error("按运营商账户调度初始化失败，下一轮重试", zap.Error(initErr))
		return
	}
	redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateDone, 0)
	c.logger.Info("按运营商账户调度初始化完成，开始领取", zap.Int64("cards", written), zap.Duration("duration", time.Since(started)),
		zap.String("metric", "polling.carrier.initialized"))
}

// syncConsumers 按当前运营商账户订阅卡任务 topic，max_in_flight 取账户调度参数。
func (c *carrierSchedulerRuntime) syncConsumers(ctx context.Context) {
	if c.consumers == nil || c.dispatcher == nil {
		return
	}
	desired := map[string]int{}
	for _, carrierID := range c.dispatcher.CarrierIDs(ctx) {
		desired[constants.NSQCardTopic(carrierID)] = c.dispatcher.ParamsFor(carrierID).CardMaxInFlight
	}
	c.consumers.Sync(desired)
}

func (c *carrierSchedulerRuntime) grace() time.Duration {
	if c.cfg.ShutdownGrace > 0 {
		return c.cfg.ShutdownGrace
	}
	return 30 * time.Second
}

// setupStopResume 创建停复机指令执行器、设备判断执行器与两个专用消费者。
func (c *carrierSchedulerRuntime) setupStopResume(carrierCfg config.CarrierSchedulerConfig, alerts *carrierqueue.AlertWriter) {
	service, ok := c.runtime.workerResult.Services.StopResumeService.(*iot_card_svc.StopResumeService)
	if !ok || carrierCfg.NSQDAddress == "" {
		return
	}
	var classifier *gatewaylimit.Classifier
	if c.runtime.gatewayLimit != nil {
		classifier = c.runtime.gatewayLimit.Classifier
	}
	params := func(carrierID uint) carrierqueue.CarrierParams {
		if c.dispatcher != nil {
			return c.dispatcher.ParamsFor(carrierID)
		}
		return carrierqueue.MergeParams(model.CarrierPollingParam{}, false, carrierCfg)
	}
	srExecutor := task.NewStopResumeExecutor(c.runtime.carrierStore, c.runtime.carrierSettings, params, alerts, service, classifier,
		c.instanceID, c.logger)
	devExecutor := task.NewDeviceEvalExecutor(c.runtime.carrierStore, service, c.instanceID, c.logger)
	consumerCfg := nsqbus.ConsumerConfig{
		NSQDAddress: carrierCfg.NSQDAddress, NSQLookupdAddresses: nsqbus.SplitAddresses(carrierCfg.NSQLookupdAddresses),
		MsgTimeout: carrierCfg.MsgTimeout, MaxAttempts: carrierCfg.MaxAttempts,
	}
	c.srConsumers = nsqbus.NewConsumerManager(consumerCfg, constants.NSQChannelWorker, srExecutor.Handle, c.logger)
	c.devConsumers = nsqbus.NewConsumerManager(consumerCfg, constants.NSQChannelWorker, devExecutor.Handle, c.logger)
}

// syncStopResume 停复机出口开关开启时订阅各账户停复机 topic 与设备判断 topic；关闭时停止订阅（队列中的指令保留）。
func (c *carrierSchedulerRuntime) syncStopResume(ctx context.Context) {
	if c.srConsumers == nil || c.dispatcher == nil {
		return
	}
	enabled := c.runtime.carrierSettings.Current(ctx).StopResumeQueueEnabled
	if !enabled {
		if c.srActive {
			c.logger.Info("停复机出口开关已关闭：停止停复机指令消费，未执行的指令保留")
			c.srConsumers.StopAll(c.grace())
			c.srConsumers.Reset()
			c.devConsumers.StopAll(c.grace())
			c.devConsumers.Reset()
			c.srActive = false
		}
		return
	}
	if !c.srActive {
		c.logger.Info("停复机出口开关已开启：自动停复机经停复机指令队列执行")
		c.srActive = true
	}
	desired := map[string]int{}
	for _, carrierID := range c.dispatcher.CarrierIDs(ctx) {
		desired[constants.NSQStopResumeTopic(carrierID)] = c.dispatcher.ParamsFor(carrierID).SRMaxInFlight
	}
	c.srConsumers.Sync(desired)
	c.devConsumers.Sync(map[string]int{constants.NSQDeviceEvalTopic: deviceEvalMaxInFlight})
}

// deviceEvalMaxInFlight 是每进程设备判断消费者并发。
const deviceEvalMaxInFlight = 10

// shutdown 优雅退出：停止订阅，宽限期内等待执行中的卡任务完成；未完成的由租约到期后重新领取。
func (c *carrierSchedulerRuntime) shutdown() {
	// 三类消费者并行停止，共用同一宽限期。
	var wg sync.WaitGroup
	for _, manager := range []*nsqbus.ConsumerManager{c.consumers, c.srConsumers, c.devConsumers} {
		if manager == nil {
			continue
		}
		wg.Add(1)
		go func(m *nsqbus.ConsumerManager) {
			defer wg.Done()
			m.StopAll(c.grace())
		}(manager)
	}
	wg.Wait()
	// 卡任务消费者停止后不再产生入库消息，再停入库与采集记录消费者，最后停生产者。
	c.ingest.shutdown()
	if c.producer != nil {
		c.producer.Stop()
	}
}

// InitProgress 返回新结构初始化进度。
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
	}
}