Files
junhong_cmp_fiber/pkg/queue/handler.go
break 5e78809b93
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 15m33s
feat(运营报表): AUG26-015 设备激活与套餐续费日报快照、查询趋势与受控导出
- 新增成对迁移 000231 与三张快照表:日级头行、设备粒度激活行、到期事件粒度续费行,以快照日期为唯一键
- 新增每日 03:30(Asia/Shanghai)日报快照生成任务与幂等整日替换,失败重试沿用同一目标日
- 新增六条受控入口:两张报表的汇总、日/月趋势与受控导出,配套查询层只读快照事实
- 新增 operations_activation 与 operations_renewal 两个导出场景,创建期冻结筛选与可见店铺范围、派发期冻结表头、执行期只按冻结值复核资格
- 采购数量口径按系统内未删除设备数实施并在 PRD 标注,附实测差额依据
- 同步证据链 requirement-evidence.json 与入口能力矩阵、ARCHITECTURE 与验证记录
- 归档 change add-operations-reports 并新建主 Spec openspec/specs/operations-report/spec.md
2026-09-18 09:42:28 +08:00

412 lines
17 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package queue
import (
"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
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)
protectHandler := task.NewPollingProtectHandler(
h.db, h.workerResult.Services.ObservationSeriesEvents,
h.pollingBase, h.gatewayClient, h.workerResult.Stores.IotCard,
h.workerResult.Stores.DeviceSimBinding, h.pollingStopResumeSvc)
cardStatusHandler := task.NewPollingCardStatusHandler(
h.pollingBase, h.gatewayClient, h.workerResult.Services.CardObservation, integrationRepository)
h.mux.HandleFunc(constants.TaskTypePollingRealname, realnameHandler.Handle)
h.mux.HandleFunc(constants.TaskTypePollingCarddata, carddataHandler.Handle)
h.mux.HandleFunc(constants.TaskTypePollingPackage, packageHandler.Handle)
h.mux.HandleFunc(constants.TaskTypePollingProtect, protectHandler.Handle)
h.mux.HandleFunc(constants.TaskTypePollingCardStatus, cardStatusHandler.Handle)
h.logger.Info("已注册轮询任务处理器realname/carddata/package/protect/card_status")
}
func (h *Handler) registerPackageActivationHandlers() {
packageActivationHandler := polling.NewPackageActivationHandler(
h.db,
h.redis,
h.asynqClient,
h.workerResult.Services.ActivationService,
nil,
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
}