iotlcm/worker-svr/internal/worker/worker.go

package worker

import (
	"context"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/gogf/gf/v2/database/gredis"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gmode"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/batch"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/client/upstream"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
	rabbitmqconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/rabbitmq"
	zsetconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/zset"
	fullsyncarchive "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/archive"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/httpapi"
	fullsyncruntime "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/runtime"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/library/tdengine"
	dailyusageproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/dailyusage"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/imei"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/monthlyusage"
	monthlyhistoryproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/monthlyusagehistory"
	simstatusproducer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/producer/simstatus"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
	batchrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/batch"
	cancelrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/cancel"
	datarepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/data"
	imeirepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/imei"
	inboundrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/inbound"
	lifecyclerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/lifecycle"
	monthlyusagerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/monthlyusage"
	simstatusrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/simstatus"
	trafficusagerepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/trafficusage"
	unsubrepository "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/unsub"
	addpackageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/addpackage"
	allocateservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/allocate"
	balanceservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/balance"
	cancelservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/cancel"
	dailyusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/dailyusage"
	dataservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/data"
	imeiservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/imei"
	inboundservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/inbound"
	lifecycleservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/lifecycle"
	monthlyusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/monthlyusage"
	monthlyhistoryservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/monthlyusagehistory"
	packagecustomerservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/packagecustomer"
	renewchangeservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/renewchange"
	rolesyncservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/rolesync"
	simstatusservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/simstatus"
	trafficcutoffconfigservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/trafficcutoffconfig"
	trafficusageservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/trafficusage"
	unsubservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/unsub"
	warningconfigservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/warningconfig"
	workorderservice "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/workorder"
)

const (
	shutdownTimeout                  = 10 * time.Second
	consumerRestartInitialDelay      = 2 * time.Second
	consumerRestartMaxDelay          = 30 * time.Second
	consumerRestartBackoffResetAfter = time.Minute
)

const monthlyUsageTemplateQPSConfigKey = "trafficUsageFullSync.consumer.templateQPS"

func loadMonthlyUsageTemplateQPS(ctx context.Context) map[string]float64 {
	value, err := g.Cfg().Get(ctx, monthlyUsageTemplateQPSConfigKey)
	if err != nil {
		g.Log().Warningf(ctx, "load monthly usage template QPS config failed: %+v", err)
		return nil
	}
	if value == nil || value.IsNil() {
		return nil
	}
	var configured map[string]float64
	if err = value.Scan(&configured); err != nil {
		g.Log().Warningf(ctx, "parse monthly usage template QPS config failed: %+v", err)
		return nil
	}
	return configured
}

func loadSimStatusTemplateQPS(ctx context.Context) map[string]float64 {
	value, err := g.Cfg().Get(ctx, "simStatusQueueSync.consumer.templateQPS")
	if err != nil || value == nil || value.IsNil() {
		return nil
	}
	var configured map[string]float64
	if err = value.Scan(&configured); err != nil {
		g.Log().Warningf(ctx, "parse sim status template QPS config failed: %+v", err)
		return nil
	}
	return configured
}

func loadImeiTemplateQPS(ctx context.Context) map[string]float64 {
	value, err := g.Cfg().Get(ctx, "imeiFullSync.consumer.templateQPS")
	if err != nil || value == nil || value.IsNil() {
		return nil
	}
	var configured map[string]float64
	if err = value.Scan(&configured); err != nil {
		g.Log().Warningf(ctx, "parse imei template QPS config failed: %+v", err)
		return nil
	}
	return configured
}

func loadSimStatusRunPrefix(ctx context.Context) (string, error) {
	prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "simStatusQueueSync.redisKeyPrefix", "sim-status:").String())
	if prefix == "" {
		return "", gerror.New("sim status redisKeyPrefix must be nonempty")
	}
	return prefix, nil
}

func loadDailyUsageTemplateQPS(ctx context.Context) map[string]float64 {
	value, err := g.Cfg().Get(ctx, "trafficUsageDailySync.consumer.templateQPS")
	if err != nil || value == nil || value.IsNil() {
		return nil
	}
	var configured map[string]float64
	if err = value.Scan(&configured); err != nil {
		g.Log().Warningf(ctx, "parse daily usage template QPS config failed: %+v", err)
		return nil
	}
	return configured
}

func loadDailyUsageRunPrefix(ctx context.Context) (string, error) {
	prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageDailySync.redisKeyPrefix", "daily:").String())
	if prefix == "" {
		return "", gerror.New("daily usage redisKeyPrefix must be nonempty")
	}
	return prefix, nil
}

func loadImeiRunPrefix(ctx context.Context) (string, error) {
	prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "imeiFullSync.redisKeyPrefix", "imei:").String())
	if prefix == "" {
		return "", gerror.New("imei redisKeyPrefix must be nonempty")
	}
	return prefix, nil
}

func loadMonthlyHistoryRunPrefix(ctx context.Context) (string, error) {
	prefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageMonthlyHistorySync.redisKeyPrefix", "monthly-history:").String())
	if prefix == "" {
		return "", gerror.New("monthly usage history redisKeyPrefix must be nonempty")
	}
	return prefix, nil
}

func loadMonthlyHistoryTemplateQPS(ctx context.Context) map[string]float64 {
	value, err := g.Cfg().Get(ctx, "trafficUsageMonthlyHistorySync.consumer.templateQPS")
	if err != nil || value == nil || value.IsNil() {
		return nil
	}
	var configured map[string]float64
	if err = value.Scan(&configured); err != nil {
		g.Log().Warningf(ctx, "parse monthly usage history template QPS config failed: %+v", err)
		return nil
	}
	return configured
}

func superviseZSetConsumer(ctx context.Context, eventPrefix string, run func(context.Context) error) {
	retryDelay := consumerRestartInitialDelay
	for {
		startedAt := time.Now()
		runErr := run(ctx)
		if ctx.Err() != nil {
			return
		}
		if time.Since(startedAt) >= consumerRestartBackoffResetAfter {
			retryDelay = consumerRestartInitialDelay
		}

		errorText := ""
		if runErr != nil {
			errorText = runErr.Error()
		}
		g.Log().Warningf(ctx, "event=%s_consumer status=restart_scheduled retryDelayMs=%d error=%q", eventPrefix, retryDelay.Milliseconds(), errorText)

		timer := time.NewTimer(retryDelay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
		}

		retryDelay *= 2
		if retryDelay > consumerRestartMaxDelay {
			retryDelay = consumerRestartMaxDelay
		}
	}
}

func loadFullSyncArchivePrefixes(ctx context.Context) ([]string, error) {
	trafficPrefix := strings.TrimSpace(g.Cfg().MustGet(ctx, "trafficUsageFullSync.redisKeyPrefix", "flow:").String())
	if trafficPrefix == "" {
		trafficPrefix = "flow:"
	}
	statusPrefix, err := loadSimStatusRunPrefix(ctx)
	if err != nil {
		return nil, err
	}
	dailyPrefix, err := loadDailyUsageRunPrefix(ctx)
	if err != nil {
		return nil, err
	}
	monthlyHistoryPrefix, err := loadMonthlyHistoryRunPrefix(ctx)
	if err != nil {
		return nil, err
	}
	imeiPrefix, err := loadImeiRunPrefix(ctx)
	if err != nil {
		return nil, err
	}
	// Duplicate prefixes would make Catalog and SimpleReplay read the same run
	// hash twice. Deduplicate while preserving the traffic-first order.
	prefixes := make([]string, 0, 5)
	seen := make(map[string]struct{}, 5)
	for _, prefix := range []string{trafficPrefix, statusPrefix, dailyPrefix, monthlyHistoryPrefix, imeiPrefix} {
		if _, exists := seen[prefix]; exists {
			continue
		}
		seen[prefix] = struct{}{}
		prefixes = append(prefixes, prefix)
	}
	return prefixes, nil
}

func Run(ctx context.Context) error {
	workerCtx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
	defer stop()

	if err := initialize(workerCtx); err != nil {
		return err
	}

	var (
		upstreamClient         = upstream.New()
		usageRepository        = trafficusagerepository.New()
		trafficUsage           = trafficusageservice.New(upstreamClient, usageRepository)
		workOrder              = workorderservice.New()
		balance                = balanceservice.New()
		inboundRepo            = inboundrepository.New()
		inboundExecutor        = inboundservice.NewExecutorWithUpstream(inboundRepo, upstreamClient)
		lifecycleRepo          = lifecyclerepository.New()
		lifecycleExecutor      = lifecycleservice.NewExecutor(lifecycleRepo, upstreamClient)
		cancelRepo             = cancelrepository.New()
		cancelExecutor         = cancelservice.NewExecutor(cancelRepo)
		renewChangeRepo        = renewchangeservice.NewRepository()
		addPackageCardExecutor = addpackageservice.NewCardExecutor(renewChangeRepo)
		addPackagePoolExecutor = addpackageservice.NewPoolExecutor(renewChangeRepo)
		renewPrepareExecutor   = renewchangeservice.NewPrepareExecutor(renewChangeRepo)
		prepareExecutor        = addpackageservice.NewPrepareMux(renewChangeRepo, renewPrepareExecutor)
		renewExecutor          = renewchangeservice.NewRenewExecutor(renewChangeRepo)
		changeExecutor         = renewchangeservice.NewChangeExecutor(renewChangeRepo)
		dataRepo               = datarepository.New()
		dataExecutor           = dataservice.NewExecutor(dataRepo, upstreamClient)
		unsubRepo              = unsubrepository.New()
		unsubExecutor          = unsubservice.NewExecutor(unsubRepo)
		restartExecutor        = lifecycleservice.NewRestartExecutor(lifecycleRepo, upstreamClient)
		batchRepository        = batchrepository.New()
		batchDispatcher        = batch.NewDispatcher(lifecycleExecutor)
		batchRuntime           = batch.NewRuntime(
			batchRepository,
			batchDispatcher,
			batch.WithExecutionTimeout(g.Cfg().MustGet(workerCtx, "rabbitmq.batchExecute.executionTimeout", "0s").Duration()),
			batch.WithFinalizationTimeout(g.Cfg().MustGet(workerCtx, "rabbitmq.batchExecute.finalizationTimeout", "10s").Duration()),
		)
		allocateExecutor            = allocateservice.NewExecutor()
		warningConfigExecutor       = warningconfigservice.NewExecutor()
		trafficCutoffConfigExecutor = trafficcutoffconfigservice.NewExecutor()
		packageCustomerSync         = packagecustomerservice.New()
	)
	var (
		monthlyProducer        *monthlyusage.Producer
		monthlyConsumer        *zsetconsumer.Consumer
		statusProducer         *simstatusproducer.Producer
		statusConsumer         *zsetconsumer.Consumer
		dailyProducer          *dailyusageproducer.Producer
		dailyConsumer          *zsetconsumer.Consumer
		monthlyHistoryProducer *monthlyhistoryproducer.Producer
		monthlyHistoryConsumer *zsetconsumer.Consumer
		imeiProducer           *imei.Producer
		imeiConsumer           *zsetconsumer.Consumer
	)
	if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.enabled", false).Bool() {
		prefix := g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.redisKeyPrefix", "flow:").String()
		store, err := zset.New(g.Redis(), prefix)
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage Redis store failed")
		}
		store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
		producerConfig, configErr := loadMonthlyUsageProducerConfig(workerCtx)
		if configErr != nil {
			return configErr
		}
		var supplyTracker *supply.Tracker
		if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.supplyObservation.enabled", false).Bool() {
			supplyTracker = supply.NewTracker(g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.enabled", false).Bool())
			producerConfig.SupplyObservation = supplyTracker
		}
		monthlyProducer, err = monthlyusage.New(&monthlyusage.DatabaseScanner{}, store, producerConfig)
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage producer failed")
		}
		monthlyUsageClient := upstream.NewWithCurrentUsageClient(
			g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.chanClientName", "").String(),
			g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.chanClientTimeout", "90s").Duration(),
		)
		usageHandler, handlerErr := monthlyusageservice.NewHandler(monthlyUsageClient, monthlyusagerepository.NewRepository())
		if handlerErr != nil {
			return gerror.Wrap(handlerErr, "initialize monthly usage handler failed")
		}
		monthlyConsumer, err = zsetconsumer.New(store, usageHandler, zsetconsumer.Config{
			SupplyObservation:  supplyTracker,
			LockKey:            g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.lockKey", "flow:consumer:active").String(),
			LockTTL:            g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.lockTTL", "1m").Duration(),
			PollInterval:       g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.pollInterval", "1s").Duration(),
			LeaseTTL:           g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.leaseTTL", "3m").Duration(),
			TaskTimeout:        g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.taskTimeout", "2m").Duration(),
			ClaimBatch:         g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.claimBatch", 100).Int(),
			MobileConcurrency:  g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.mobileConcurrency", 30).Int(),
			TelecomConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.telecomConcurrency", 20).Int(),
			GenericConcurrency: g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.genericConcurrency", 1).Int(),
			GenericQPS:         g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.genericQPS", 1.5).Float64(),
			TemplateQPS:        loadMonthlyUsageTemplateQPS(workerCtx),
			RoundRetryEnabled:  g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.enabled", true).Bool(),
			MaxRounds:          g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.maxRounds", 5).Int(),
			FailureRetention:   g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.retry.failureRetention", "720h").Duration(),
			ShouldRetryNextRound: func(result zsetconsumer.Result) bool {
				return !result.Success && result.Retryable
			},
		})
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage consumer failed")
		}
		if g.Cfg().MustGet(workerCtx, "trafficUsageFullSync.consumer.enabled", false).Bool() {
			go superviseZSetConsumer(workerCtx, "monthly_usage", monthlyConsumer.Run)
		}
	}
	if g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.enabled", false).Bool() {
		prefix, prefixErr := loadDailyUsageRunPrefix(workerCtx)
		if prefixErr != nil {
			return prefixErr
		}
		store, err := zset.New(g.Redis(), prefix)
		if err != nil {
			return gerror.Wrap(err, "initialize daily usage Redis store failed")
		}
		store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
		dailyScanMode, dailyScanConcurrency, scanErr := loadHistoryUsageScanSettings(workerCtx, "trafficUsageDailySync")
		if scanErr != nil {
			return scanErr
		}
		dailyProducer, err = dailyusageproducer.New(&dailyusageproducer.DatabaseScanner{}, store, dailyusageproducer.Config{
			ScanMode:             dailyScanMode,
			ScanConcurrency:      dailyScanConcurrency,
			PageSize:             g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.pageSize", 100).Int(),
			GlobalHighWaterMark:  int64(g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.globalHighWaterMark", 0).Int64()),
			ChannelHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.channelHighWaterMark", 0).Int64()),
			PollInterval:         g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.pollInterval", "2s").Duration(),
			RunLockKey:           g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.runLockKey", "daily:producer:active").String(),
			RunLockTTL:           g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.producer.runLockTTL", "5m").Duration(),
			RoundRetryEnabled:    g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.enabled", true).Bool(),
			MaxRounds:            g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.maxRounds", 3).Int(),
			FailureRetention:     g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.failureRetention", "720h").Duration(),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize daily usage producer failed")
		}
		dailyHandler, handlerErr := dailyusageservice.NewHandler(upstream.New(), trafficusagerepository.New())
		if handlerErr != nil {
			return gerror.Wrap(handlerErr, "initialize daily usage handler failed")
		}
		dailyConsumer, err = zsetconsumer.New(store, dailyHandler, zsetconsumer.Config{
			RunQueueEnabled:        true,
			RoundRetryEnabled:      g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.enabled", true).Bool(),
			MaxRounds:              g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.maxRounds", 3).Int(),
			FailureRetention:       g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.retry.failureRetention", "720h").Duration(),
			ShouldRetryNextRound:   dailyusageservice.ShouldRetryNextRound,
			BlockedTasksAsFailures: true,
			EventPrefix:            "daily_usage",
			LockKey:                g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.lockKey", "daily:consumer:active").String(),
			LockTTL:                g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.lockTTL", "1m").Duration(),
			PollInterval:           g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.pollInterval", "1s").Duration(),
			LeaseTTL:               g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.leaseTTL", "3m").Duration(),
			TaskTimeout:            g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.taskTimeout", "2m").Duration(),
			ClaimBatch:             g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.claimBatch", 100).Int(),
			MobileConcurrency:      g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.mobileConcurrency", 30).Int(),
			TelecomConcurrency:     g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.telecomConcurrency", 8).Int(),
			GenericConcurrency:     g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.genericConcurrency", 1).Int(),
			GenericQPS:             g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.genericQPS", 1.5).Float64(),
			TemplateQPS:            loadDailyUsageTemplateQPS(workerCtx),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize daily usage consumer failed")
		}
		if g.Cfg().MustGet(workerCtx, "trafficUsageDailySync.consumer.enabled", false).Bool() {
			go superviseZSetConsumer(workerCtx, "daily_usage", dailyConsumer.Run)
		}
	}
	if g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.enabled", false).Bool() {
		prefix, prefixErr := loadMonthlyHistoryRunPrefix(workerCtx)
		if prefixErr != nil {
			return prefixErr
		}
		store, err := zset.New(g.Redis(), prefix)
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage history Redis store failed")
		}
		store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
		monthlyHistoryScanMode, monthlyHistoryScanConcurrency, scanErr := loadHistoryUsageScanSettings(workerCtx, "trafficUsageMonthlyHistorySync")
		if scanErr != nil {
			return scanErr
		}
		monthlyHistoryProducer, err = monthlyhistoryproducer.New(&monthlyhistoryproducer.DatabaseScanner{}, store, monthlyhistoryproducer.Config{
			ScanMode:             monthlyHistoryScanMode,
			ScanConcurrency:      monthlyHistoryScanConcurrency,
			PageSize:             g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.pageSize", 100).Int(),
			GlobalHighWaterMark:  int64(g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.globalHighWaterMark", 0).Int64()),
			ChannelHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.channelHighWaterMark", 0).Int64()),
			PollInterval:         g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.pollInterval", "2s").Duration(),
			RunLockKey:           g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.runLockKey", "monthly-history:producer:active").String(),
			RunLockTTL:           g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.producer.runLockTTL", "5m").Duration(),
			RoundRetryEnabled:    g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.enabled", true).Bool(),
			MaxRounds:            g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.maxRounds", 3).Int(),
			FailureRetention:     g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.failureRetention", "720h").Duration(),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage history producer failed")
		}
		monthlyHistoryHandler, handlerErr := monthlyhistoryservice.NewHandler(upstream.New(), trafficusagerepository.New())
		if handlerErr != nil {
			return gerror.Wrap(handlerErr, "initialize monthly usage history handler failed")
		}
		monthlyHistoryConsumer, err = zsetconsumer.New(store, monthlyHistoryHandler, zsetconsumer.Config{
			RunQueueEnabled:        true,
			RoundRetryEnabled:      g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.enabled", true).Bool(),
			MaxRounds:              g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.maxRounds", 3).Int(),
			FailureRetention:       g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.retry.failureRetention", "720h").Duration(),
			ShouldRetryNextRound:   monthlyhistoryservice.ShouldRetryNextRound,
			BlockedTasksAsFailures: true,
			EventPrefix:            "monthly_usage_history",
			LockKey:                g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.lockKey", "monthly-history:consumer:active").String(),
			LockTTL:                g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.lockTTL", "1m").Duration(),
			PollInterval:           g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.pollInterval", "1s").Duration(),
			LeaseTTL:               g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.leaseTTL", "3m").Duration(),
			TaskTimeout:            g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.taskTimeout", "2m").Duration(),
			ClaimBatch:             g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.claimBatch", 100).Int(),
			MobileConcurrency:      g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.mobileConcurrency", 30).Int(),
			TelecomConcurrency:     g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.telecomConcurrency", 8).Int(),
			GenericConcurrency:     g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.genericConcurrency", 1).Int(),
			GenericQPS:             g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.genericQPS", 1.5).Float64(),
			TemplateQPS:            loadMonthlyHistoryTemplateQPS(workerCtx),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize monthly usage history consumer failed")
		}
		if g.Cfg().MustGet(workerCtx, "trafficUsageMonthlyHistorySync.consumer.enabled", false).Bool() {
			go superviseZSetConsumer(workerCtx, "monthly_usage_history", monthlyHistoryConsumer.Run)
		}
	}
	if g.Cfg().MustGet(workerCtx, "simStatusQueueSync.enabled", false).Bool() {
		prefix, prefixErr := loadSimStatusRunPrefix(workerCtx)
		if prefixErr != nil {
			return prefixErr
		}
		store, err := zset.New(g.Redis(), prefix)
		if err != nil {
			return gerror.Wrap(err, "initialize sim status Redis store failed")
		}
		store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
		statusProducer, err = simstatusproducer.New(&simstatusproducer.DatabaseScanner{}, store, simstatusproducer.Config{
			PageSize:            g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.pageSize", 1000).Int(),
			GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.globalHighWaterMark", 0).Int64()),
			PollInterval:        g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.pollInterval", "1s").Duration(),
			RunLockKey:          g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.runLockKey", "sim-status:producer:active").String(),
			RunLockTTL:          g.Cfg().MustGet(workerCtx, "simStatusQueueSync.producer.runLockTTL", "5m").Duration(),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize sim status producer failed")
		}
		statusHandler, handlerErr := simstatusservice.NewHandler(
			upstream.New(),
			simstatusrepository.NewRepository(),
			g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.networkRetries", 2).Int(),
		)
		if handlerErr != nil {
			return gerror.Wrap(handlerErr, "initialize sim status handler failed")
		}
		statusConsumer, err = zsetconsumer.New(store, statusHandler, zsetconsumer.Config{
			RunQueueEnabled:           true,
			RoundRetryEnabled:         g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.enabled", true).Bool(),
			MaxRounds:                 g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.maxRounds", 3).Int(),
			FailureRetention:          g.Cfg().MustGet(workerCtx, "simStatusQueueSync.retry.failureRetention", "720h").Duration(),
			ShouldRetryNextRound:      simstatusservice.ShouldRetryNextRound,
			BlockedTasksAsFailures:    true,
			EventPrefix:               "sim_status_queue",
			LockKey:                   g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.lockKey", "sim-status:consumer:active").String(),
			LockTTL:                   g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.lockTTL", "1m").Duration(),
			PollInterval:              g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.pollInterval", "1s").Duration(),
			LeaseTTL:                  g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.leaseTTL", "3m").Duration(),
			TaskTimeout:               g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.taskTimeout", "2m").Duration(),
			ClaimBatch:                g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.claimBatch", 100).Int(),
			MobileConcurrency:         g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.mobileConcurrency", 100).Int(),
			TelecomConcurrency:        g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.telecomConcurrency", 8).Int(),
			GenericConcurrency:        g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.genericConcurrency", 1).Int(),
			GenericQPS:                g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.genericQPS", 1.5).Float64(),
			TemplateQPS:               loadSimStatusTemplateQPS(workerCtx),
			FailurePoolAlertThreshold: g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.failurePoolAlertThreshold", 50000).Int64(),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize sim status consumer failed")
		}
		if g.Cfg().MustGet(workerCtx, "simStatusQueueSync.consumer.enabled", false).Bool() {
			go superviseZSetConsumer(workerCtx, "sim_status_queue", statusConsumer.Run)
		}
	}
	if g.Cfg().MustGet(workerCtx, "imeiFullSync.enabled", false).Bool() {
		prefix, prefixErr := loadImeiRunPrefix(workerCtx)
		if prefixErr != nil {
			return prefixErr
		}
		store, err := zset.New(g.Redis(), prefix)
		if err != nil {
			return gerror.Wrap(err, "initialize imei Redis store failed")
		}
		store.EnableArchiveIndex(g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool())
		// g.Redis may be initialized from g.Cfg without registering GetConfig.
		leaseConfig, configured := gredis.GetConfig()
		if !configured {
			value, configErr := g.Cfg().Get(workerCtx, "redis.default")
			if configErr != nil {
				return gerror.Wrap(configErr, "load IMEI lease Redis configuration failed")
			}
			leaseConfig, err = gredis.ConfigFromMap(value.Map())
			if err != nil {
				return gerror.Wrap(err, "parse IMEI lease Redis configuration failed")
			}
		}
		imeiProducer, err = imei.New(&imei.DatabaseScanner{}, store, imei.Config{
			PageSize:            g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.pageSize", 1000).Int(),
			GlobalHighWaterMark: int64(g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.globalHighWaterMark", 0).Int64()),
			PollInterval:        g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.pollInterval", "1s").Duration(),
			RunLockKey:          g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.runLockKey", "imei:producer:active").String(),
			RunLockTTL:          g.Cfg().MustGet(workerCtx, "imeiFullSync.producer.runLockTTL", "5m").Duration(),
			LeaseRedisConfig:    leaseConfig,
		})
		if err != nil {
			return gerror.Wrap(err, "initialize imei producer failed")
		}
		imeiHandler, handlerErr := imeiservice.NewHandler(
			upstream.NewWithImeiClient(
				g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.chanClientName", "").String(),
				g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.chanClientTimeout", "30s").Duration(),
			),
			imeirepository.NewRepository(),
			g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.networkRetries", 2).Int(),
		)
		if handlerErr != nil {
			return gerror.Wrap(handlerErr, "initialize imei handler failed")
		}
		imeiConsumer, err = zsetconsumer.New(store, imeiHandler, zsetconsumer.Config{
			RunQueueEnabled:        true,
			RoundRetryEnabled:      g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.enabled", true).Bool(),
			MaxRounds:              g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.maxRounds", 5).Int(),
			FailureRetention:       g.Cfg().MustGet(workerCtx, "imeiFullSync.retry.failureRetention", "720h").Duration(),
			ShouldRetryNextRound:   imeiservice.ShouldRetryNextRound,
			BlockedTasksAsFailures: true,
			EventPrefix:            "imei_queue",
			LockKey:                g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.lockKey", "imei:consumer:active").String(),
			LockTTL:                g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.lockTTL", "1m").Duration(),
			PollInterval:           g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.pollInterval", "1s").Duration(),
			LeaseTTL:               g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.leaseTTL", "3m").Duration(),
			TaskTimeout:            g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.taskTimeout", "2m").Duration(),
			ClaimBatch:             g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.claimBatch", 100).Int(),
			MobileConcurrency:      g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.mobileConcurrency", 100).Int(),
			TelecomConcurrency:     g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.telecomConcurrency", 8).Int(),
			GenericConcurrency:     g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.genericConcurrency", 1).Int(),
			GenericQPS:             g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.genericQPS", 1.5).Float64(),
			TemplateQPS:            loadImeiTemplateQPS(workerCtx),
			// 失败池是唯一随卡数线性增长的每卡结构（单条约 1KB），
			// 且会保留整个 failureRetention。超阀值只告警不截断。
			FailurePoolAlertThreshold: g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.failurePoolAlertThreshold", 50000).Int64(),
		})
		if err != nil {
			return gerror.Wrap(err, "initialize imei consumer failed")
		}
		if g.Cfg().MustGet(workerCtx, "imeiFullSync.consumer.enabled", false).Bool() {
			go superviseZSetConsumer(workerCtx, "imei_queue", imeiConsumer.Run)
		}
	}
	var rabbitMQConsumers *rabbitmqconsumer.Consumers
	if monthlyProducer != nil {
		rabbitMQConsumers = rabbitmqconsumer.NewWithPackageCustomerSyncAndMonthlyUsage(trafficUsage, workOrder, balance, packageCustomerSync, monthlyProducer, batchRuntime)
	} else {
		rabbitMQConsumers = rabbitmqconsumer.NewWithPackageCustomerSync(trafficUsage, workOrder, balance, packageCustomerSync, batchRuntime)
	}
	if statusProducer != nil {
		rabbitMQConsumers.WithSimStatusQueue(statusProducer, statusProducer)
	}
	if dailyProducer != nil {
		rabbitMQConsumers.WithDailyUsageStart(dailyProducer)
	}
	if monthlyHistoryProducer != nil {
		rabbitMQConsumers.WithMonthlyHistoryStart(monthlyHistoryProducer)
	}
	if imeiProducer != nil {
		rabbitMQConsumers.WithImeiQueue(imeiProducer)
	}
	rabbitMQConsumers.WithDataReviewQuery(dataExecutor)
	var (
		fullSyncHTTP     *httpapi.Server
		fullSyncArchives []*fullsyncarchive.Service
	)
	{ // The authenticated control plane is always available.
		prefixes, err := loadFullSyncArchivePrefixes(workerCtx)
		if err != nil {
			return err
		}
		stores := make([]*zset.Store, 0, len(prefixes))
		for _, prefix := range prefixes {
			store, e := zset.New(g.Redis(), prefix)
			if e != nil {
				return e
			}
			stores = append(stores, store)
		}
		archiveRoot := g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String()
		repositories := make([]*fullsyncruntime.Repository, 0, len(stores))
		for _, store := range stores {
			repo, e := fullsyncruntime.New(store, archiveRoot)
			if e != nil {
				return gerror.Wrap(e, "initialize full-sync runtime repository failed")
			}
			repositories = append(repositories, repo)
		}
		catalog, err := fullsyncruntime.NewCatalog(repositories...)
		if err != nil {
			return gerror.Wrap(err, "initialize full-sync run catalog failed")
		}
		simple, err := fullsyncruntime.NewSimpleReplayWithDefault(g.Cfg().MustGet(workerCtx, "fullSyncOps.replay.maxRounds", fullsyncruntime.DefaultReplayMaxRounds).Int(), stores...)
		if err != nil {
			return err
		}
		// Only Admin exposes replay to users. This listener stays inside the
		// Docker network; the Worker service must not publish port 18081.
		fullSyncHTTP, err = httpapi.NewServer(httpapi.Config{Addr: ":18081", ShutdownTimeout: g.Cfg().MustGet(workerCtx, "fullSyncOps.control.shutdownTimeout", "10s").Duration()}, httpapi.Dependencies{Runs: catalog, List: catalog, Failures: catalog, Replays: repositories[0], Actions: repositories[0], Simple: simple})
		if err != nil {
			return gerror.Wrap(err, "initialize full-sync control server failed")
		}
		if err := fullSyncHTTP.Start(workerCtx); err != nil {
			return gerror.Wrap(err, "start full-sync control server failed")
		}
		simple.Start(workerCtx)
	}
	if g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.enabled", false).Bool() {
		prefixes, err := loadFullSyncArchivePrefixes(workerCtx)
		if err != nil {
			return gerror.Wrap(err, "load full-sync archive Redis prefixes failed")
		}
		for _, prefix := range prefixes {
			store, err := zset.New(g.Redis(), prefix)
			if err != nil {
				return gerror.Wrapf(err, "initialize full-sync archive Redis store failed for prefix %q", prefix)
			}
			store.EnableArchiveIndex(true)
			repo, err := fullsyncruntime.New(store, g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String())
			if err != nil {
				return gerror.Wrapf(err, "initialize full-sync archive repository failed for prefix %q", prefix)
			}
			service := fullsyncarchive.NewService(fullsyncarchive.Config{Root: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.rootDir", "/app/logs/full-sync").String(), Interval: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.pollInterval", "10s").Duration(), InitialBackoff: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.retryMin", "5s").Duration(), MaxBackoff: g.Cfg().MustGet(workerCtx, "fullSyncOps.archive.retryMax", "5m").Duration()}, repo, repo, repo)
			if err := service.Start(workerCtx); err != nil {
				return gerror.Wrapf(err, "start full-sync archive service failed for prefix %q", prefix)
			}
			fullSyncArchives = append(fullSyncArchives, service)
		}
	}
	batchDispatcher.Register(consts.BatchTypeInbound, inboundExecutor)
	batchDispatcher.Register(consts.BatchTypeAllocate, allocateExecutor)
	batchDispatcher.Register(consts.BatchTypeCardAddPackage, addPackageCardExecutor)
	batchDispatcher.Register(consts.BatchTypePoolAddPackage, addPackagePoolExecutor)
	batchDispatcher.Register(consts.BatchTypePrepare, prepareExecutor)
	batchDispatcher.Register(consts.BatchTypeRenew, renewExecutor)
	batchDispatcher.Register(consts.BatchTypeChange, changeExecutor)
	batchDispatcher.Register(consts.BatchTypeDataCut, dataExecutor)
	batchDispatcher.Register(consts.BatchTypeDataStart, dataExecutor)
	batchDispatcher.Register(consts.BatchTypeDataStop, dataExecutor)
	// Resume durable APN sub-batches independently of RabbitMQ delivery and process restarts.
	if reconciler, ok := dataExecutor.(interface{ RunReconciler(context.Context) error }); ok {
		go func() {
			if err := reconciler.RunReconciler(workerCtx); err != nil && workerCtx.Err() == nil {
				g.Log().Errorf(workerCtx, "APN operation reconciler stopped: %v", err)
			}
		}()
	}
	batchDispatcher.Register(consts.BatchTypeSimCardUnsub, unsubExecutor)
	batchDispatcher.Register(consts.BatchTypeSimCardRestart, restartExecutor)
	batchDispatcher.Register(consts.BatchTypeSimCardCancel, cancelExecutor)
	batchDispatcher.Register(consts.BatchTypeSimWarningConfig, warningConfigExecutor)
	batchDispatcher.Register(consts.BatchTypeTrafficCutoffConfig, trafficCutoffConfigExecutor)
	if err := rabbitMQConsumers.Start(workerCtx); err != nil {
		_ = rabbitMQConsumers.Stop(context.Background())
		return gerror.Wrap(err, "start RabbitMQ consumers failed")
	}

	stopRoleSync, err := rolesyncservice.Start(workerCtx)
	if err != nil {
		_ = rabbitMQConsumers.Stop(context.Background())
		return gerror.Wrap(err, "start role sync Worker failed")
	}
	defer stopRoleSync()

	g.Log().Info(workerCtx, "iotlcm worker started")
	<-workerCtx.Done()
	g.Log().Info(ctx, "iotlcm worker stopping")
	stopRoleSync()

	shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
	defer cancel()
	if err := rabbitMQConsumers.Stop(shutdownCtx); err != nil {
		return gerror.Wrap(err, "stop RabbitMQ consumers failed")
	}
	for _, service := range fullSyncArchives {
		if err := service.Stop(); err != nil {
			return gerror.Wrap(err, "stop full-sync archive service failed")
		}
	}
	if fullSyncHTTP != nil {
		if err := fullSyncHTTP.Wait(); err != nil {
			return gerror.Wrap(err, "stop full-sync control server failed")
		}
	}
	if err := tdengine.Close(); err != nil {
		return gerror.Wrap(err, "close TDengine failed")
	}
	return nil
}

func initialize(ctx context.Context) error {
	if err := g.DB().PingMaster(); err != nil {
		return gerror.Wrap(err, "connect database failed")
	}
	if err := gtime.SetTimeZone("Asia/Shanghai"); err != nil {
		return gerror.Wrap(err, "set timezone failed")
	}

	mode := g.Cfg().MustGet(ctx, "system.gfmode", gmode.NOT_SET).String()
	for _, supported := range []string{gmode.DEVELOP, gmode.TESTING, gmode.STAGING, gmode.PRODUCT} {
		if mode == supported {
			gmode.Set(mode)
			break
		}
	}
	return nil
}