junhong_cmp_fiber/pkg/queue/handler.go

package queue

import (
	"context"

	"github.com/hibiken/asynq"
	"github.com/redis/go-redis/v9"
	"go.uber.org/zap"
	"gorm.io/gorm"

	operationsReportApp "github.com/break/junhong_cmp_fiber/internal/application/operationsreport"
	packageExpiryApp "github.com/break/junhong_cmp_fiber/internal/application/packageexpiry"
	packageTrafficAlertApp "github.com/break/junhong_cmp_fiber/internal/application/packagetrafficalert"
	"github.com/break/junhong_cmp_fiber/internal/exporter"
	"github.com/break/junhong_cmp_fiber/internal/gateway"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox"
	notification "github.com/break/junhong_cmp_fiber/internal/infrastructure/notification"
	operationsReportInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/operationsreport"
	packageExpiryInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/packageexpiry"
	packageTrafficAlertInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/packagetrafficalert"
	"github.com/break/junhong_cmp_fiber/internal/polling"
	packageExpiryQuery "github.com/break/junhong_cmp_fiber/internal/query/packageexpiry"
	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/internal/task"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/storage"
)

type Handler struct {
	mux                  *asynq.ServeMux
	logger               *zap.Logger
	db                   *gorm.DB
	redis                *redis.Client
	storage              *storage.Service
	gatewayClient        *gateway.Client
	pollingCallback      task.PollingCallback
	workerResult         *WorkerBootstrapResult
	asynqClient          *asynq.Client
	pollingBase          *task.PollingBase
	pollingSteps         PollingStepHandlers
	pollingStopResumeSvc iot_card_svc.StopResumeServiceInterface
}

func NewHandler(
	db *gorm.DB,
	redis *redis.Client,
	storageSvc *storage.Service,
	gatewayClient *gateway.Client,
	pollingCallback task.PollingCallback,
	workerResult *WorkerBootstrapResult,
	asynqClient *asynq.Client,
	logger *zap.Logger,
	pollingBase *task.PollingBase,
	pollingStopResumeSvc iot_card_svc.StopResumeServiceInterface,
) *Handler {
	return &Handler{
		mux:                  asynq.NewServeMux(),
		logger:               logger,
		db:                   db,
		redis:                redis,
		storage:              storageSvc,
		gatewayClient:        gatewayClient,
		pollingCallback:      pollingCallback,
		workerResult:         workerResult,
		asynqClient:          asynqClient,
		pollingBase:          pollingBase,
		pollingStopResumeSvc: pollingStopResumeSvc,
	}
}

func (h *Handler) RegisterHandlers() *asynq.ServeMux {
	emailHandler := task.NewEmailHandler(h.redis, h.logger)

	h.mux.HandleFunc(constants.TaskTypeEmailSend, emailHandler.HandleEmailSend)
	h.logger.Info("注册邮件发送任务处理器", zap.String("task_type", constants.TaskTypeEmailSend))

	h.registerIotCardImportHandler()
	h.registerDeviceImportHandler()
	h.registerOrderPackageInvalidateHandler()
	h.registerAssetPackageBatchOrderHandler()
	h.registerShopBusinessOwnerImportHandler()
	h.registerPhoneAssetUnbindImportHandler()
	h.registerExportHandlers()
	h.registerCommissionStatsHandlers()
	h.registerCommissionCalculationHandler()
	h.registerPollingHandlers()
	h.registerCardObservationSeriesHandler()
	h.registerPackageActivationHandlers()
	h.registerOrderExpireHandler()
	h.registerAlertCheckHandler()
	h.registerDataCleanupHandler()
	h.registerNotificationCleanupHandler()
	h.registerPackageExpiryReminderHandler()
	h.registerPackageTrafficAlertScanHandler()
	h.registerAutoPurchaseHandler()
	h.registerDailyTrafficFlushHandler()
	h.registerOperationsReportSnapshotHandler()

	h.logger.Info("所有任务处理器注册完成")
	return h.mux
}

func (h *Handler) registerCardObservationSeriesHandler() {
	handler := task.NewCardObservationSeriesHandler(h.workerResult.Services.CardObservationSeries, h.logger)
	h.mux.HandleFunc(constants.TaskTypeCardObservationSeries, handler.Handle)
	h.logger.Info("注册卡观测事件序列任务处理器", zap.String("task_type", constants.TaskTypeCardObservationSeries))
}

func (h *Handler) registerIotCardImportHandler() {
	iotCardImportHandler := task.NewIotCardImportHandler(
		h.db,
		h.redis,
		h.workerResult.Stores.IotCardImportTask,
		h.workerResult.Stores.IotCard,
		h.workerResult.Stores.AssetWallet,
		h.storage,
		h.pollingCallback,
		audit.NewWriter(audit.NewRegistry(), nil),
		h.logger,
	)

	h.mux.HandleFunc(constants.TaskTypeIotCardImport, iotCardImportHandler.HandleIotCardImport)
	h.logger.Info("注册 IoT 卡导入任务处理器", zap.String("task_type", constants.TaskTypeIotCardImport))
}

func (h *Handler) registerOrderPackageInvalidateHandler() {
	orderPkgHandler := task.NewOrderPackageInvalidateHandler(
		h.workerResult.Stores.OrderPackageInvalidateTask,
		h.workerResult.Stores.Order,
		h.workerResult.Stores.PackageUsage,
		h.storage,
		h.logger,
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	h.mux.HandleFunc(constants.TaskTypeOrderPackageInvalidate, orderPkgHandler.Handle)
	h.logger.Info("注册订单套餐批量失效任务处理器", zap.String("task_type", constants.TaskTypeOrderPackageInvalidate))
}

func (h *Handler) registerAssetPackageBatchOrderHandler() {
	handler := task.NewAssetPackageBatchOrderHandler(
		h.workerResult.Stores.AssetPackageBatchOrderTask,
		h.workerResult.Stores.Shop,
		h.workerResult.Services.AssetPackageOrderCreator,
		h.storage,
		h.logger,
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	h.mux.HandleFunc(constants.TaskTypeAssetPackageBatchOrder, handler.Handle)
	h.logger.Info("注册资产套餐批量订购任务处理器", zap.String("task_type", constants.TaskTypeAssetPackageBatchOrder))
}

func (h *Handler) registerShopBusinessOwnerImportHandler() {
	handler := task.NewShopBusinessOwnerImportHandler(
		h.db,
		h.workerResult.Stores.ShopBusinessOwnerImportTask,
		h.storage,
		h.logger,
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	h.mux.HandleFunc(constants.TaskTypeShopBusinessOwnerImport, handler.Handle)
	h.logger.Info("注册店铺负责人导入任务处理器", zap.String("task_type", constants.TaskTypeShopBusinessOwnerImport))
}

func (h *Handler) registerPhoneAssetUnbindImportHandler() {
	handler := task.NewPhoneAssetUnbindImportHandler(
		h.db,
		h.workerResult.Stores.PhoneAssetUnbindImportTask,
		h.workerResult.Stores.PhoneAssetAssociation,
		h.workerResult.Stores.AssetIdentifier,
		h.workerResult.Stores.IotCard,
		h.workerResult.Stores.Device,
		h.storage,
		h.logger,
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	h.mux.HandleFunc(constants.TaskTypePhoneAssetUnbindImport, handler.Handle)
	h.logger.Info("注册手机号资产解绑导入任务处理器", zap.String("task_type", constants.TaskTypePhoneAssetUnbindImport))
}

func (h *Handler) registerDeviceImportHandler() {
	deviceImportHandler := task.NewDeviceImportHandler(
		h.db,
		h.redis,
		h.workerResult.Stores.DeviceImportTask,
		h.workerResult.Stores.Device,
		h.workerResult.Stores.DeviceSimBinding,
		h.workerResult.Stores.IotCard,
		h.workerResult.Stores.AssetWallet,
		h.workerResult.Stores.AssetIdentifier,
		h.storage,
		audit.NewWriter(audit.NewRegistry(), nil),
		h.logger,
		h.workerResult.Services.DeviceBatchAllocator,
	)

	h.mux.HandleFunc(constants.TaskTypeDeviceImport, deviceImportHandler.HandleDeviceImport)
	h.logger.Info("注册设备导入任务处理器", zap.String("task_type", constants.TaskTypeDeviceImport))
}

func (h *Handler) registerExportHandlers() {
	sceneRegistry := exporter.NewDefaultRegistry(h.db)

	dispatchHandler := task.NewExportDispatchHandler(
		h.db,
		h.redis,
		h.workerResult.Stores.ExportTask,
		h.workerResult.Stores.ExportShardTask,
		h.asynqClient,
		sceneRegistry,
		h.logger,
	)
	shardHandler := task.NewExportShardHandler(
		h.redis,
		h.workerResult.Stores.ExportTask,
		h.workerResult.Stores.ExportShardTask,
		h.asynqClient,
		h.storage,
		sceneRegistry,
		h.logger,
	)
	finalizeHandler := task.NewExportFinalizeHandler(
		h.redis,
		h.workerResult.Stores.ExportTask,
		h.workerResult.Stores.ExportShardTask,
		h.storage,
		sceneRegistry,
		h.logger,
	)

	h.mux.HandleFunc(constants.TaskTypeExportDispatch, dispatchHandler.HandleExportDispatch)
	h.logger.Info("注册导出 dispatch 任务处理器", zap.String("task_type", constants.TaskTypeExportDispatch))

	h.mux.HandleFunc(constants.TaskTypeExportShard, shardHandler.HandleExportShard)
	h.logger.Info("注册导出 shard 任务处理器", zap.String("task_type", constants.TaskTypeExportShard))

	h.mux.HandleFunc(constants.TaskTypeExportFinalize, finalizeHandler.HandleExportFinalize)
	h.logger.Info("注册导出 finalize 任务处理器", zap.String("task_type", constants.TaskTypeExportFinalize))
}

func (h *Handler) registerCommissionStatsHandlers() {
	updateHandler := task.NewCommissionStatsUpdateHandler(
		h.redis,
		h.workerResult.Stores.ShopSeriesCommissionStats,
		h.workerResult.Stores.ShopPackageAllocation,
		h.logger,
	)
	syncHandler := task.NewCommissionStatsSyncHandler(
		h.db,
		h.redis,
		h.workerResult.Stores.ShopSeriesCommissionStats,
		h.logger,
	)
	archiveHandler := task.NewCommissionStatsArchiveHandler(
		h.db,
		h.redis,
		h.workerResult.Stores.ShopSeriesCommissionStats,
		h.logger,
	)

	h.mux.HandleFunc(constants.TaskTypeCommissionStatsUpdate, updateHandler.HandleCommissionStatsUpdate)
	h.logger.Info("注册佣金统计更新任务处理器", zap.String("task_type", constants.TaskTypeCommissionStatsUpdate))

	h.mux.HandleFunc(constants.TaskTypeCommissionStatsSync, syncHandler.HandleCommissionStatsSync)
	h.logger.Info("注册佣金统计同步任务处理器", zap.String("task_type", constants.TaskTypeCommissionStatsSync))

	h.mux.HandleFunc(constants.TaskTypeCommissionStatsArchive, archiveHandler.HandleCommissionStatsArchive)
	h.logger.Info("注册佣金统计归档任务处理器", zap.String("task_type", constants.TaskTypeCommissionStatsArchive))
}

func (h *Handler) registerCommissionCalculationHandler() {
	commissionCalculationHandler := task.NewCommissionCalculationHandler(
		h.db,
		h.workerResult.Services.CommissionCalculation,
		h.logger,
	)
	h.mux.HandleFunc(constants.TaskTypeCommission, commissionCalculationHandler.HandleCommissionCalculation)
	h.logger.Info("注册佣金计算任务处理器", zap.String("task_type", constants.TaskTypeCommission))
}

func (h *Handler) registerPollingHandlers() {
	integrationRepository := integrationlog.NewRepository(h.db)
	realnameHandler := task.NewPollingRealnameHandler(
		h.pollingBase, h.gatewayClient, h.workerResult.Services.CardObservation, integrationRepository)
	carrierStore := postgres.NewCarrierStore(h.db)
	carddataHandler := task.NewPollingCarddataHandler(
		h.pollingBase, h.gatewayClient, carrierStore, h.workerResult.Services.CardObservation, integrationRepository)
	packageHandler := task.NewPollingPackageHandler(
		h.pollingBase, h.workerResult.Stores.IotCard,
		h.pollingStopResumeSvc)
	cardStatusHandler := task.NewPollingCardStatusHandler(
		h.pollingBase, h.gatewayClient, h.workerResult.Services.CardObservation, integrationRepository)

	h.pollingSteps = PollingStepHandlers{Realname: realnameHandler, Carddata: carddataHandler, Package: packageHandler, CardStatus: cardStatusHandler}
	h.mux.HandleFunc(constants.TaskTypePollingRealname, realnameHandler.Handle)
	h.mux.HandleFunc(constants.TaskTypePollingCarddata, carddataHandler.Handle)
	h.mux.HandleFunc(constants.TaskTypePollingPackage, packageHandler.Handle)
	// 设备保护期已下线：保留空处理函数消化发布前已入队的 polling:protect 任务，下一次发布再移除。
	h.mux.HandleFunc(constants.TaskTypePollingProtect, func(context.Context, *asynq.Task) error { return nil })
	h.mux.HandleFunc(constants.TaskTypePollingCardStatus, cardStatusHandler.Handle)
	h.logger.Info("已注册轮询任务处理器（realname/carddata/package/card_status）")
}

func (h *Handler) registerPackageActivationHandlers() {
	packageActivationHandler := polling.NewPackageActivationHandler(
		h.db,
		h.redis,
		h.asynqClient,
		h.workerResult.Services.ActivationService,
		// 激活成功后须重新评估停复机（客户付款或实名后及时复机）；与轮询调度中的激活处理器注入同一服务。
		h.workerResult.Services.StopResumeService,
		h.logger,
	)

	h.mux.HandleFunc(constants.TaskTypePackageFirstActivation, packageActivationHandler.HandlePackageFirstActivation)
	h.logger.Info("注册首次实名激活任务处理器", zap.String("task_type", constants.TaskTypePackageFirstActivation))

	h.mux.HandleFunc(constants.TaskTypePackageQueueActivation, packageActivationHandler.HandlePackageQueueActivation)
	h.logger.Info("注册排队激活任务处理器", zap.String("task_type", constants.TaskTypePackageQueueActivation))
}

func (h *Handler) registerOrderExpireHandler() {
	orderExpireHandler := task.NewOrderExpireHandler(h.workerResult.Services.OrderExpirer, h.logger)
	h.mux.HandleFunc(constants.TaskTypeOrderExpire, orderExpireHandler.HandleOrderExpire)
	h.logger.Info("注册订单超时取消任务处理器", zap.String("task_type", constants.TaskTypeOrderExpire))
}

func (h *Handler) registerAlertCheckHandler() {
	alertCheckHandler := task.NewAlertCheckHandler(h.workerResult.Services.AlertService, h.logger)
	h.mux.HandleFunc(constants.TaskTypeAlertCheck, alertCheckHandler.HandleAlertCheck)
	h.logger.Info("注册告警检查任务处理器", zap.String("task_type", constants.TaskTypeAlertCheck))
}

func (h *Handler) registerDataCleanupHandler() {
	dataCleanupHandler := task.NewDataCleanupHandler(h.workerResult.Services.CleanupService, h.logger)
	h.mux.HandleFunc(constants.TaskTypeDataCleanup, dataCleanupHandler.HandleDataCleanup)
	h.logger.Info("注册数据清理任务处理器", zap.String("task_type", constants.TaskTypeDataCleanup))
}

func (h *Handler) registerNotificationCleanupHandler() {
	cleanupService := notification.NewCleanupService(h.db, h.logger, audit.NewWriter(audit.NewRegistry(), nil))
	cleanupHandler := task.NewNotificationCleanupHandler(cleanupService, h.logger)
	h.mux.HandleFunc(constants.TaskTypeNotificationCleanup, cleanupHandler.Handle)
	h.logger.Info("注册站内通知保留清理任务处理器", zap.String("task_type", constants.TaskTypeNotificationCleanup))
}

func (h *Handler) registerPackageExpiryReminderHandler() {
	query := packageExpiryQuery.NewQuery(h.db)
	publisher := packageExpiryInfra.NewReminderPublisher(h.db, outbox.NewRepository())
	service := packageExpiryApp.NewReminderService(query, publisher)
	handler := task.NewPackageExpiryReminderHandler(service, h.logger)
	h.mux.HandleFunc(constants.TaskTypePackageExpiryReminder, handler.Handle)
	h.logger.Info("注册每日套餐临期提醒扫描任务处理器", zap.String("task_type", constants.TaskTypePackageExpiryReminder))
}

func (h *Handler) registerPackageTrafficAlertScanHandler() {
	scanner := packageTrafficAlertInfra.NewScanner(h.db)
	writer := packageTrafficAlertInfra.NewAlertWriter(
		h.db,
		postgres.NewPackageTrafficAlertStore(h.db),
		outbox.NewRepository(),
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	service := packageTrafficAlertApp.NewScanService(scanner, writer, h.logger)
	handler := task.NewPackageTrafficAlertScanHandler(service, h.logger)
	h.mux.HandleFunc(constants.TaskTypePackageTrafficAlertScan, handler.Handle)
	h.logger.Info("注册每日套餐真流量达量预警扫描任务处理器", zap.String("task_type", constants.TaskTypePackageTrafficAlertScan))
}

func (h *Handler) registerAutoPurchaseHandler() {
	autoPurchaseHandler := task.NewAutoPurchaseHandler(
		h.db,
		h.workerResult.Stores.Order,
		nil, // RechargeOrderStore：在 NewAutoPurchaseHandler 内按需初始化
		nil, // PaymentStore：在 NewAutoPurchaseHandler 内按需初始化
		h.workerResult.Stores.AssetWallet,
		nil, // AssetWalletTransactionStore：在 NewAutoPurchaseHandler 内按需初始化
		h.workerResult.Stores.PackageUsage,
		h.redis,
		h.asynqClient,
		h.logger,
		h.workerResult.Services.ObservationSeriesEvents,
		h.workerResult.Services.PriorityEvents,
		audit.NewWriter(audit.NewRegistry(), nil),
	)
	h.mux.HandleFunc(constants.TaskTypeAutoPurchaseAfterRecharge, autoPurchaseHandler.ProcessTask)
	h.logger.Info("注册自动购包任务处理器", zap.String("task_type", constants.TaskTypeAutoPurchaseAfterRecharge))
}

func (h *Handler) registerDailyTrafficFlushHandler() {
	cardDailyUsageStore := postgres.NewCardDailyUsageStore(h.db)
	dailyTrafficFlushHandler := task.NewDailyTrafficFlushHandler(h.redis, cardDailyUsageStore, h.logger)
	h.mux.HandleFunc(constants.TaskTypeDailyTrafficFlush, dailyTrafficFlushHandler.HandleDailyTrafficFlush)
	h.logger.Info("注册每日流量落盘任务处理器", zap.String("task_type", constants.TaskTypeDailyTrafficFlush))
}

// registerOperationsReportSnapshotHandler 注册运营报表日报快照生成任务处理器。
// 生成用例依赖只读数据源与整日替换写入适配，二者都在此按需装配。
func (h *Handler) registerOperationsReportSnapshotHandler() {
	source := operationsReportInfra.NewSource(h.db)
	store := operationsReportInfra.NewSnapshotStore(h.db)
	generator := operationsReportApp.NewGenerator(source, store, h.logger)
	handler := task.NewOperationsReportSnapshotHandler(generator, h.logger)
	h.mux.HandleFunc(constants.TaskTypeOperationsReportSnapshot, handler.Handle)
	h.logger.Info("注册每日运营报表日报快照生成任务处理器", zap.String("task_type", constants.TaskTypeOperationsReportSnapshot))
}

// GetMux 获取 ServeMux（用于启动 Worker 服务器）
func (h *Handler) GetMux() *asynq.ServeMux {
	return h.mux
}

// PollingStepHandlers 是四个轮询检查步骤处理器，旧分片调度注册为 Asynq handler，按运营商账户调度的卡任务执行器复用其 Step。
type PollingStepHandlers struct {
	Realname   *task.PollingRealnameHandler
	Carddata   *task.PollingCarddataHandler
	Package    *task.PollingPackageHandler
	CardStatus *task.PollingCardStatusHandler
}

// PollingSteps 返回轮询检查步骤处理器（RegisterHandlers 之后可用）。
func (h *Handler) PollingSteps() PollingStepHandlers {
	return h.pollingSteps
}