package payment import ( "context" "time" "github.com/bytedance/sonic" "gorm.io/gorm" "gorm.io/gorm/clause" agentrecharge "github.com/break/junhong_cmp_fiber/internal/application/agentrecharge" walletapp "github.com/break/junhong_cmp_fiber/internal/application/wallet" "github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" ) // AgentRechargePaymentConsumer 将已确认收款的代理在线充值幂等入账主钱包。 type AgentRechargePaymentConsumer struct { db *gorm.DB posting *walletapp.PostingService } // NewAgentRechargePaymentConsumer 创建代理在线充值入账消费者。 func NewAgentRechargePaymentConsumer(db *gorm.DB, posting *walletapp.PostingService) *AgentRechargePaymentConsumer { return &AgentRechargePaymentConsumer{db: db, posting: posting} } // Consume 校验支付与充值权威事实后,在独立事务中完成唯一入账和充值终态。 func (c *AgentRechargePaymentConsumer) Consume(ctx context.Context, envelope outbox.DeliveryEnvelope) error { if c == nil || c.db == nil || c.posting == nil { return errors.New(errors.CodeInternalError, "代理在线充值入账消费者未配置") } if envelope.EventType != constants.OutboxEventTypeAgentRechargePaymentConfirmed || envelope.PayloadVersion != constants.AgentRechargePaymentConfirmedPayloadVersionV1 { return errors.New(errors.CodeInvalidParam, "代理充值支付确认事件类型或版本不受支持") } var event agentrecharge.PaymentConfirmedEvent if err := sonic.Unmarshal(envelope.Payload, &event); err != nil { return errors.Wrap(errors.CodeInvalidParam, err, "代理充值支付确认事件载荷无法解析") } if event.EventID == "" || event.EventID != envelope.EventID || event.RechargeID == 0 || event.PaymentID == 0 || event.ShopID == 0 || event.WalletID == 0 || event.Amount <= 0 || event.ThirdPartyTradeNo == "" { return errors.New(errors.CodeInvalidParam, "代理充值支付确认事件载荷不完整") } return c.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { recharge, _, err := lockCreditingFacts(ctx, tx, event) if err != nil { return err } if recharge.Status != constants.RechargeStatusPaid && recharge.Status != constants.RechargeStatusCompleted { return errors.New(errors.CodeInvalidStatus, "代理在线充值当前状态不可入账") } if _, err := c.posting.PostInTx(ctx, tx, walletapp.PostingCommand{ ShopID: recharge.ShopID, WalletID: recharge.AgentWalletID, Amount: recharge.Amount, ReferenceType: constants.ReferenceTypeTopup, ReferenceID: recharge.ID, TransactionType: constants.AgentTransactionTypeRecharge, UserID: recharge.UserID, Creator: recharge.UserID, Remark: "代理在线扫码充值", RequestID: envelope.RequestID, CorrelationID: envelope.CorrelationID, }); err != nil { return err } if recharge.Status == constants.RechargeStatusCompleted { return nil } completedAt := time.Now().UTC() update := tx.WithContext(ctx).Model(&model.AgentRechargeRecord{}). Where("id = ? AND status = ?", recharge.ID, constants.RechargeStatusPaid). Updates(map[string]any{"status": constants.RechargeStatusCompleted, "completed_at": completedAt}) if update.Error != nil { return errors.Wrap(errors.CodeDatabaseError, update.Error, "完成代理在线充值单失败") } if update.RowsAffected != 1 { return errors.New(errors.CodeConflict, "代理在线充值状态已变化") } return nil }) } func lockCreditingFacts(ctx context.Context, tx *gorm.DB, event agentrecharge.PaymentConfirmedEvent) (*model.AgentRechargeRecord, *model.Payment, error) { var recharge model.AgentRechargeRecord if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", event.RechargeID).First(&recharge).Error; err != nil { if err == gorm.ErrRecordNotFound { return nil, nil, errors.New(errors.CodeNotFound, "代理在线充值单不存在") } return nil, nil, errors.Wrap(errors.CodeDatabaseError, err, "锁定代理在线充值单失败") } var payment model.Payment if err := tx.WithContext(ctx).Where("id = ?", event.PaymentID).First(&payment).Error; err != nil { if err == gorm.ErrRecordNotFound { return nil, nil, errors.New(errors.CodeConflict, "代理在线充值支付单不存在") } return nil, nil, errors.Wrap(errors.CodeDatabaseError, err, "查询代理在线充值支付单失败") } if recharge.ID != event.RechargeID || recharge.RechargeNo != event.RechargeNo || recharge.ShopID != event.ShopID || recharge.AgentWalletID != event.WalletID || recharge.UserID != event.UserID || recharge.Amount != event.Amount || recharge.PaymentTransactionID == nil || *recharge.PaymentTransactionID != event.ThirdPartyTradeNo || (recharge.PaymentMethod != constants.RechargeMethodWechat && recharge.PaymentMethod != constants.RechargeMethodAlipay) { return nil, nil, errors.New(errors.CodeConflict, "代理充值事件与充值权威事实不一致") } if payment.OrderType != model.PaymentOrderTypeAgentRecharge || payment.OrderID != recharge.ID || payment.PaymentNo != event.PaymentNo || payment.PaymentMethod != event.PaymentMethod || payment.Amount != event.Amount || payment.Status != model.PaymentRecordStatusPaid || payment.ThirdPartyTradeNo != event.ThirdPartyTradeNo { return nil, nil, errors.New(errors.CodeConflict, "代理充值事件与支付权威事实不一致") } return &recharge, &payment, nil } var _ outbox.EventConsumer = (*AgentRechargePaymentConsumer)(nil)