feat(01-03): 充值回调触发自动购包任务入队(CRITICAL-04)

- recharge.Service struct 新增 queueClient *queue.Client 字段
- recharge.New() 新增 queueClient 参数
- HandlePaymentCallback 事务成功后,LinkedPackageIDs 非空时入队 TaskTypeAutoPurchaseAfterRecharge
- bootstrap/services.go 的 recharge.New() 传入 deps.QueueClient
This commit is contained in:
2026-03-27 23:29:05 +08:00
parent 0df992c98a
commit 9a4d87a0d4
2 changed files with 24 additions and 1 deletions

View File

@@ -170,7 +170,7 @@ func initServices(s *stores, deps *Dependencies) *services {
PurchaseValidation: purchaseValidation, PurchaseValidation: purchaseValidation,
Order: orderSvc.New(deps.DB, deps.Redis, s.Order, s.OrderItem, s.AgentWallet, s.AssetWallet, purchaseValidation, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.IotCard, s.Device, s.PackageSeries, s.PackageUsage, s.Package, wechatConfig, deps.WechatPayment, deps.QueueClient, deps.Logger), Order: orderSvc.New(deps.DB, deps.Redis, s.Order, s.OrderItem, s.AgentWallet, s.AssetWallet, purchaseValidation, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.IotCard, s.Device, s.PackageSeries, s.PackageUsage, s.Package, wechatConfig, deps.WechatPayment, deps.QueueClient, deps.Logger),
Exchange: exchangeSvc.New(deps.DB, s.ExchangeOrder, s.IotCard, s.Device, s.AssetWallet, s.AssetWalletTransaction, s.PackageUsage, s.PackageUsageDailyRecord, s.ResourceTag, s.PersonalCustomerDevice, deps.Logger), Exchange: exchangeSvc.New(deps.DB, s.ExchangeOrder, s.IotCard, s.Device, s.AssetWallet, s.AssetWalletTransaction, s.PackageUsage, s.PackageUsageDailyRecord, s.ResourceTag, s.PersonalCustomerDevice, deps.Logger),
Recharge: rechargeSvc.New(deps.DB, s.AssetRecharge, s.AssetWallet, s.AssetWalletTransaction, s.IotCard, s.Device, s.ShopSeriesAllocation, s.PackageSeries, s.CommissionRecord, wechatConfig, deps.Logger), Recharge: rechargeSvc.New(deps.DB, s.AssetRecharge, s.AssetWallet, s.AssetWalletTransaction, s.IotCard, s.Device, s.ShopSeriesAllocation, s.PackageSeries, s.CommissionRecord, wechatConfig, deps.Logger, deps.QueueClient),
PollingConfig: pollingSvc.NewConfigService(s.PollingConfig), PollingConfig: pollingSvc.NewConfigService(s.PollingConfig),
PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis), PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis),
PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis), PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis),

View File

@@ -9,8 +9,11 @@ import (
"github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/model/dto" "github.com/break/junhong_cmp_fiber/internal/model/dto"
"github.com/break/junhong_cmp_fiber/internal/store/postgres" "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/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/queue"
"github.com/hibiken/asynq"
"go.uber.org/zap" "go.uber.org/zap"
"gorm.io/gorm" "gorm.io/gorm"
) )
@@ -46,6 +49,7 @@ type Service struct {
packageSeriesStore *postgres.PackageSeriesStore packageSeriesStore *postgres.PackageSeriesStore
commissionRecordStore *postgres.CommissionRecordStore commissionRecordStore *postgres.CommissionRecordStore
wechatConfigService WechatConfigServiceInterface wechatConfigService WechatConfigServiceInterface
queueClient *queue.Client // 任务队列客户端,用于触发自动购包任务
logger *zap.Logger logger *zap.Logger
} }
@@ -62,6 +66,7 @@ func New(
commissionRecordStore *postgres.CommissionRecordStore, commissionRecordStore *postgres.CommissionRecordStore,
wechatConfigService WechatConfigServiceInterface, wechatConfigService WechatConfigServiceInterface,
logger *zap.Logger, logger *zap.Logger,
queueClient *queue.Client,
) *Service { ) *Service {
return &Service{ return &Service{
db: db, db: db,
@@ -74,6 +79,7 @@ func New(
packageSeriesStore: packageSeriesStore, packageSeriesStore: packageSeriesStore,
commissionRecordStore: commissionRecordStore, commissionRecordStore: commissionRecordStore,
wechatConfigService: wechatConfigService, wechatConfigService: wechatConfigService,
queueClient: queueClient,
logger: logger, logger: logger,
} }
} }
@@ -382,6 +388,23 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
return err return err
} }
// 充值成功后,如有关联套餐则触发自动购包(强充绑套餐功能)
linkedIDs := recharge.LinkedPackageIDs
if len(linkedIDs) > 0 && string(linkedIDs) != "[]" && string(linkedIDs) != "null" {
if s.queueClient != nil {
taskPayload := task.AutoPurchasePayload{RechargeRecordID: recharge.ID}
if enqueueErr := s.queueClient.EnqueueTask(ctx, constants.TaskTypeAutoPurchaseAfterRecharge, taskPayload,
asynq.Queue(constants.QueueDefault),
asynq.MaxRetry(3),
); enqueueErr != nil {
s.logger.Error("自动购包任务入队失败",
zap.Uint("recharge_id", recharge.ID),
zap.Error(enqueueErr))
// 不影响充值成功,仅记录日志
}
}
}
s.logger.Info("充值支付回调处理成功", s.logger.Info("充值支付回调处理成功",
zap.String("recharge_no", rechargeNo), zap.String("recharge_no", rechargeNo),
zap.Int64("amount", recharge.Amount), zap.Int64("amount", recharge.Amount),