package bootstrap import ( pollingPkg "github.com/break/junhong_cmp_fiber/internal/polling" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" assetAutoRenewalApp "github.com/break/junhong_cmp_fiber/internal/application/assetautorenewal" cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation" carrierThresholdApp "github.com/break/junhong_cmp_fiber/internal/application/carrierthreshold" employeecollectionApp "github.com/break/junhong_cmp_fiber/internal/application/employeecollection" "github.com/break/junhong_cmp_fiber/internal/application/merchantpayment" packagesilenceApp "github.com/break/junhong_cmp_fiber/internal/application/packagesilence" walletapp "github.com/break/junhong_cmp_fiber/internal/application/wallet" assetAutoRenewalInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/assetautorenewal" auditInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" cardObservationInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/cardobservation" carrierThresholdInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/carrierthreshold" "github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog" "github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox" paymentInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/payment" prioritypollingInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/prioritypolling" walletinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wallet" assetAutoRenewalQuery "github.com/break/junhong_cmp_fiber/internal/query/assetautorenewal" "github.com/break/junhong_cmp_fiber/internal/service/commission_calculation" "github.com/break/junhong_cmp_fiber/internal/service/commission_stats" deviceSvc "github.com/break/junhong_cmp_fiber/internal/service/device" iotCardSvc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" orderSvc "github.com/break/junhong_cmp_fiber/internal/service/order" packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package" pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling" purchaseValidationSvc "github.com/break/junhong_cmp_fiber/internal/service/purchase_validation" "github.com/break/junhong_cmp_fiber/pkg/queue" "github.com/break/junhong_cmp_fiber/pkg/wechat" ) type workerServices struct { CommissionCalculation *commission_calculation.Service CommissionStats *commission_stats.Service UsageService *packagepkg.UsageService ActivationService *packagepkg.ActivationService ResetService *packagepkg.ResetService AlertService *pollingSvc.AlertService CleanupService *pollingSvc.CleanupService } func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *queue.WorkerServices { auditWriter := auditInfra.NewWriter(auditInfra.NewRegistry(), nil) commissionStatsService := commission_stats.New(stores.ShopSeriesCommissionStats) commissionCalculationService := commission_calculation.New( deps.DB, stores.CommissionRecord, stores.Shop, stores.ShopPackageAllocation, stores.ShopSeriesAllocation, stores.PackageSeries, stores.IotCard, stores.Device, stores.AgentWallet, stores.AgentWalletTransaction, stores.Order, stores.OrderItem, stores.Package, stores.ShopSeriesCommissionStats, commissionStatsService, deps.Logger, ) commissionCalculationService.SetAuditWriter(auditWriter) usageService := packagepkg.NewUsageService( deps.DB, deps.Redis, stores.PackageUsage, stores.PackageUsageDailyRecord, stores.DeviceSimBinding, deps.Logger, ) activationService := packagepkg.NewActivationService( deps.DB, deps.Redis, stores.PackageUsage, stores.Package, stores.PackageUsageDailyRecord, deps.Logger, ) resetService := packagepkg.NewResetService( deps.DB, deps.Redis, stores.PackageUsage, deps.Logger, ) activationService.SetLifecycleAudit(auditWriter) packageSilence := newPackageSilenceBundle(deps.DB, deps.Redis, deps.Logger, auditWriter, activationService) // Worker 内延期用例只承担审批通过后的执行、查单恢复与异常原路退款,不创建新申请。 silencePaymentLog := integrationlog.NewRepository(deps.DB) packageSilence.Extensions.SetPayment(merchantpayment.NewRuntimeLoader(deps.DB, deps.Redis), packagesilenceApp.PaymentPorts{ Wechat: paymentInfra.NewWechatWebAdapter(wechat.NewRedisCache(deps.Redis), silencePaymentLog, deps.Logger), Alipay: paymentInfra.NewAlipayWapAdapter(silencePaymentLog, deps.Logger), Fuiou: paymentInfra.NewFuiouScanAdapter(silencePaymentLog, deps.Logger), }, paymentInfra.NewRefundAdapter(wechat.NewRedisCache(deps.Redis), deps.Logger)) usageService.SetLifecycleAudit(auditWriter) resetService.SetLifecycleAudit(auditWriter) alertService := pollingSvc.NewAlertService( stores.PollingAlertRule, stores.PollingAlertHistory, deps.Redis, deps.Logger, ) cleanupService := pollingSvc.NewCleanupService( stores.DataCleanupConfig, stores.DataCleanupLog, deps.Logger, ) cardObservationOutbox := outbox.NewRepository() observationSeriesEvents := cardObservationInfra.NewSeriesEventWriter(cardObservationOutbox) priorityEvents := prioritypollingInfra.NewPriorityEventWriter(cardObservationOutbox) // 运营商通道流量阈值用例:达量判定嵌入流量观测事务,持锁判定注入停复机入口。 carrierThresholdService := carrierThresholdApp.NewService(deps.DB, cardObservationOutbox).SetLogger(deps.Logger) cardObservationService := cardObservationApp.NewService( deps.DB, cardObservationInfra.NewEventWriter(cardObservationOutbox), cardObservationInfra.NewCacheInvalidator(deps.Redis, deps.Logger), ) iotCardAuditService := iotCardSvc.New( deps.DB, stores.IotCard, stores.Shop, stores.AssetAllocationRecord, stores.ShopPackageAllocation, stores.ShopSeriesAllocation, stores.PackageSeries, deps.GatewayClient, deps.Logger, ) iotCardAuditService.SetAccessAudit(auditWriter) iotCardAuditService.SetPhoneAssetAssociationStore(stores.PhoneAssetAssociation) cardObservationService.SetStateAuditWriter(iotCardAuditService) // 运营商通道流量阈值达量判定嵌入流量观测事务:与卡流量事实同事务写周期锁与停机事件。 cardObservationService.SetChannelThresholdEvaluator(carrierThresholdService) cardObservationService.SetCommandGrace(commandGraceProvider(deps.DB, deps.Redis, deps.Logger)) // 卡流量采集队列:读数先进每卡 Redis 队列,满足写入条件才写日快照;队列配置为受控系统配置。 cardObservationService.SetTrafficQueue(cardObservationInfra.NewTrafficQueue(deps.Redis), cardObservationApp.NewDefaultTrafficQueueSettingsProvider(deps.DB, deps.Redis, deps.Logger), deps.Logger) // 激活前流量结算:与 API 进程一致,注入激活服务与订单服务。 preActivationSettler := cardObservationInfra.NewPreActivationSettler(deps.DB, deps.GatewayClient, cardObservationService, postgres.NewCarrierStore(deps.DB), deps.Logger) activationService.SetPreActivationSettler(preActivationSettler) cardObservationIntegration := integrationlog.NewRepository(deps.DB) cardObservationSeriesCoordinator := cardObservationInfra.NewSeriesCoordinator(deps.Redis) cardObservationSeriesService := cardObservationApp.NewSeriesAttemptService( cardObservationSeriesCoordinator, cardObservationInfra.NewSeriesRunner(deps.DB, deps.GatewayClient, cardObservationService, cardObservationIntegration, auditWriter), cardObservationInfra.NewSeriesAttemptLogger(cardObservationIntegration), ) // 初始化订单服务,供超时取消和批量订购共同复用现有订单规则。 purchaseValidation := purchaseValidationSvc.New(deps.DB, stores.IotCard, stores.Device, stores.Package, stores.ShopPackageAllocation) orderService := orderSvc.New( deps.DB, deps.Redis, stores.Order, stores.OrderItem, stores.AgentWallet, stores.AssetWallet, nil, // paymentStore: 超时取消不需要 purchaseValidation, stores.ShopPackageAllocation, stores.ShopSeriesAllocation, stores.IotCard, stores.Device, stores.PackageSeries, stores.PackageUsage, stores.Package, nil, // wechatConfigService: 超时取消不需要 nil, // wechatPayment: 超时取消不需要 nil, // paymentLoader: 超时取消不需要 deps.QueueClient, deps.Logger, stores.AssetIdentifier, stores.PersonalCustomer, stores.PersonalCustomerPhone, ) // 批量订购线下支付建单需在订单事务内生成员工代收款账单,与 API 进程保持一致。 orderService.SetEmployeeCollectionBillCreation(employeecollectionApp.NewBillCreationService(auditWriter)) orderService.SetPreActivationSettler(preActivationSettler) orderService.SetLifecycleAudit(auditWriter) // 资产轮询开关服务:轮询开关文件导入与订单套餐生效后自动恢复轮询(批量下单在 Worker 内执行)。 assetPollingService := pollingSvc.NewAssetPollingService(deps.DB, stores.Device, stores.DeviceSimBinding, iotCardAuditService, pollingPkg.NewPollingQueueManager(deps.Redis, constants.PollingShardCount, deps.Logger), deps.Logger, auditWriter) orderService.SetPollingRestorer(assetPollingService) orderService.SetPaymentIntegrationLog(integrationlog.NewRepository(deps.DB)) walletOutbox := outbox.NewRepository() walletDebitEvents := walletinfra.NewDebitEventWriter(walletOutbox, auditWriter) orderService.SetAgentWalletReservationService(walletapp.NewReservationService(walletinfra.NewReservationEventWriter(walletOutbox, auditWriter), walletDebitEvents, nil)) orderService.SetAgentWalletDebitService(walletapp.NewDebitService(walletDebitEvents, nil)) orderService.SetObservationSeriesEventWriter(observationSeriesEvents) orderService.SetPriorityEventWriter(priorityEvents) orderService.SetSilenceSetter(packageSilence.Setter) orderService.SetFirstPurchaseRecorder(packageSilence.FirstPurchase) // 创建停复机服务并注入回调:流量耗尽自动停机、套餐激活/重置/支付后自动复机 stopResumeService := iotCardSvc.NewStopResumeService( deps.Redis, stores.IotCard, stores.PackageUsage, stores.DeviceSimBinding, deps.GatewayClient, deps.Logger, ) stopResumeService.SetObservationSeriesEventWriter(deps.DB, observationSeriesEvents) stopResumeService.SetPriorityEventWriter(priorityEvents) stopResumeService.SetUnifiedAudit(auditWriter, integrationlog.NewRepository(deps.DB)) if deps.GatewayLimit != nil { stopResumeService.SetStopResumeRecorder(deps.GatewayLimit.Recorder) deps.GatewayLimit.TerminalPolling.SetAuditWriter(auditWriter) stopResumeService.SetTerminalPollingDisabler(deps.GatewayLimit.TerminalPolling) } // 持通道阈值锁的卡在周期内拒绝一切复机(自动、手动、机卡分离)。 stopResumeService.SetChannelThresholdLockGuard(carrierThresholdService) // 停复机执行端口复用既有停复机服务作为唯一事实源:消费者与两个计划任务共用同一用例实例。 carrierThresholdService.SetCommander(carrierThresholdInfra.NewCardCommander(stopResumeService)) // 资产钱包自动续费:复机执行同样复用既有停复机单一事实源,通知与复机都走公共 Outbox。 assetAutoRenewalService := assetAutoRenewalApp.NewService(assetAutoRenewalApp.Dependencies{ DB: deps.DB, Redis: deps.Redis, Logger: deps.Logger, Outbox: cardObservationOutbox, AuditWriter: auditWriter, PurchaseValidation: purchaseValidation, Candidates: assetAutoRenewalQuery.NewQuery(deps.DB), Resume: assetAutoRenewalInfra.NewCardCommander(stopResumeService), ObservationEvents: observationSeriesEvents, PriorityEvents: priorityEvents, }) assetAutoRenewalService.SetSilenceRecorder(packagesilenceApp.PurchaseRecorder{Setter: packageSilence.Setter, FirstPurchase: packageSilence.FirstPurchase}) activationService.SetObservationSeriesEventWriter(observationSeriesEvents) activationService.SetPriorityEventWriter(priorityEvents) usageService.SetStopResumeCallback(stopResumeService) activationService.SetResumeCallback(stopResumeService) orderService.SetResumeCallback(stopResumeService) resetService.SetResumeCallback(stopResumeService) deviceBatchAllocator := deviceSvc.New( deps.DB, deps.Redis, stores.Device, stores.DeviceSimBinding, stores.IotCard, stores.Shop, stores.AssetAllocationRecord, stores.ShopPackageAllocation, stores.ShopSeriesAllocation, stores.PackageSeries, deps.GatewayClient, stores.AssetIdentifier, nil, nil, ) deviceBatchAllocator.SetPhoneAssetAssociationStore(stores.PhoneAssetAssociation) return &queue.WorkerServices{ PaymentAudit: auditWriter, RechargeAudit: auditWriter, CardObservation: cardObservationService, CardObservationSeries: cardObservationSeriesService, ObservationSeriesEvents: observationSeriesEvents, PriorityEvents: priorityEvents, CommissionCalculation: commissionCalculationService, CommissionStats: commissionStatsService, UsageService: usageService, ActivationService: activationService, ResetService: resetService, AlertService: alertService, CleanupService: cleanupService, StopResumeService: stopResumeService, CarrierThreshold: carrierThresholdService, AssetAutoRenewal: assetAutoRenewalService, OrderExpirer: orderService, AssetPackageOrderCreator: orderService, DeviceBatchAllocator: deviceBatchAllocator, AssetPolling: assetPollingService, IotCardPolling: iotCardAuditService, SilenceSwitch: packageSilence.Switch, SilenceResolver: packageSilence.Resolver, SilenceSetter: packageSilence.Setter, SilenceFirstPurchase: packageSilence.FirstPurchase, SilenceExtensions: packageSilence.Extensions, } }