909f4d0ec2
Some checks failed
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Failing after 8m39s
- 迁移 000259～000265：套餐记录沉默期字段与扫描索引、套餐默认沉默期、延期政策/流水/订单/批次/明细 5 张表、默认政策与系统配置（总开关默认关闭）、企微审批白名单、卡与设备出库时间及出库记录 - 沉默期设置（订单发放、自动购包、钱包续费）、后台购买时调整（超默认值按政策免审或企微审批）、统一判定与激活闸门、排队与孤儿扫描跳过过期套餐、过期套餐禁止退款、每 10 分钟扫描 - 总开关专用接口与打开前 4 项检查；延期政策；延期资格与计费；H5 自助与后台二维码在线支付（商户池路由）、支付回调与查单恢复、线下登记、特殊/批量延期、异常重试与原路退款；沉默期审批场景与终态处理 - 换货沉默期流水与出库时间；卡/设备列表、详情、导出与出库记录接口；延期订单与即将过期套餐列表及导出；财务流水纳入延期支付 - 沉默期与出库时间回填工具 cmd/silence-backfill；前端接口变更说明 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
449 lines
19 KiB
Go
449 lines
19 KiB
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.registerPackageSilenceHandlers()
|
||
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))
|
||
}
|
||
|
||
// registerPackageSilenceHandlers 注册沉默期扫描与延期订单查单恢复任务处理器。
|
||
func (h *Handler) registerPackageSilenceHandlers() {
|
||
services := h.workerResult.Services
|
||
if services == nil || services.SilenceResolver == nil {
|
||
h.logger.Warn("沉默期用例未装配，跳过注册沉默期任务处理器")
|
||
return
|
||
}
|
||
scan := task.NewPackageSilenceScanHandler(services.SilenceResolver, h.logger)
|
||
h.mux.HandleFunc(constants.TaskTypePackageSilenceScan, scan.Handle)
|
||
h.logger.Info("注册沉默期扫描任务处理器", zap.String("task_type", constants.TaskTypePackageSilenceScan))
|
||
if services.SilenceExtensions != nil {
|
||
recoverHandler := task.NewPackageSilenceOrderRecoverHandler(services.SilenceExtensions, h.logger)
|
||
h.mux.HandleFunc(constants.TaskTypePackageSilenceOrderRecover, recoverHandler.Handle)
|
||
h.logger.Info("注册沉默期延期订单恢复任务处理器", zap.String("task_type", constants.TaskTypePackageSilenceOrderRecover))
|
||
}
|
||
}
|
||
|
||
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),
|
||
)
|
||
if services := h.workerResult.Services; services != nil {
|
||
autoPurchaseHandler.SetSilenceSetter(services.SilenceSetter)
|
||
autoPurchaseHandler.SetOutboundRecorder(services.SilenceOutbound)
|
||
}
|
||
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
|
||
}
|