csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
76d0f38b9eec531d3b38cb8b1f5e2acad2e35191
junhong_cmp_fiber/internal/bootstrap/worker_services.go
邹进 9c3c258f16
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m27s
Details
feat: 卡流量读数先过采集队列再入账，扣减按采集时刻确定套餐 
- 每卡 Redis 采集队列（默认 N=3、跨天清空、回退容差 1 MB），攒满且不下降才写最后一个点
- 测试环境全量切换日快照计量，下线灰度名单、切换水位与卡表基线扣费；全部卡置待建起点
- 读数区分没给值与 0；有套餐的卡读数 0 也写快照与日用量 0 行；重置日比较基准为 0 照常防旧值
- 新增采集记录（按月分区保留 90 天）、人工冲正流水与后台接口、采集告警
- 扣减按采集时刻有效套餐查找，只对生效中套餐做状态迁移，查不到套餐按成功跳过
- 换货新卡（新设备下的卡）打待建起点；采集间隔上限只校验启用中的轮询配置

Change: add-card-traffic-queue-and-observed-deduction

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-29 12:18:47 +09:00

273 lines
14 KiB
Go
Raw Blame History

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,
}
}
Reference in New Issue View Git Blame Copy Permalink