diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index f6da571..968a6db 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -170,7 +170,7 @@ func initServices(s *stores, deps *Dependencies) *services { 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), 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), PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis), PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis), diff --git a/internal/service/recharge/service.go b/internal/service/recharge/service.go index 30103dd..27c5aec 100644 --- a/internal/service/recharge/service.go +++ b/internal/service/recharge/service.go @@ -9,8 +9,11 @@ import ( "github.com/break/junhong_cmp_fiber/internal/model" "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/task" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/queue" + "github.com/hibiken/asynq" "go.uber.org/zap" "gorm.io/gorm" ) @@ -46,6 +49,7 @@ type Service struct { packageSeriesStore *postgres.PackageSeriesStore commissionRecordStore *postgres.CommissionRecordStore wechatConfigService WechatConfigServiceInterface + queueClient *queue.Client // 任务队列客户端,用于触发自动购包任务 logger *zap.Logger } @@ -62,6 +66,7 @@ func New( commissionRecordStore *postgres.CommissionRecordStore, wechatConfigService WechatConfigServiceInterface, logger *zap.Logger, + queueClient *queue.Client, ) *Service { return &Service{ db: db, @@ -74,6 +79,7 @@ func New( packageSeriesStore: packageSeriesStore, commissionRecordStore: commissionRecordStore, wechatConfigService: wechatConfigService, + queueClient: queueClient, logger: logger, } } @@ -382,6 +388,23 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string, 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("充值支付回调处理成功", zap.String("recharge_no", rechargeNo), zap.Int64("amount", recharge.Amount),