break
98c145fe70
Some checks failed
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Failing after 3m55s
新增收款商户、商户池轮询、微信授权配置独立管理；三类新支付 （C端套餐购买、C端资产钱包充值、代理在线预存款充值）无条件 经商户池选择并冻结路由，无旧综合配置回退。merchant_id 为空 历史支付继续按 payment_config_id 双读。凭证版本化加载与 ID+版本缓存保证轮换一致性。删除商户池新支付创建开关及全部 引用。
242 lines
10 KiB
Go
242 lines
10 KiB
Go
package agentrecharge
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"gorm.io/gorm"
|
||
|
||
merchantpayment "github.com/break/junhong_cmp_fiber/internal/application/merchantpayment"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// RecoverOnlinePaymentService 批量收敛长期缺少支付链接或待支付的代理在线充值。
|
||
type RecoverOnlinePaymentService struct {
|
||
db *gorm.DB
|
||
runtime *merchantpayment.RuntimeLoader
|
||
wechat OnlinePaymentPort
|
||
alipay OnlinePaymentPort
|
||
fuiou OnlinePaymentPort
|
||
confirm *ConfirmOnlinePaymentService
|
||
audit PaymentAuditWriter
|
||
now func() time.Time
|
||
}
|
||
|
||
// NewRecoverOnlinePaymentService 创建代理在线充值支付恢复用例。
|
||
func NewRecoverOnlinePaymentService(db *gorm.DB, runtime *merchantpayment.RuntimeLoader, wechat, alipay, fuiou OnlinePaymentPort, confirm *ConfirmOnlinePaymentService, audit PaymentAuditWriter) *RecoverOnlinePaymentService {
|
||
return &RecoverOnlinePaymentService{db: db, runtime: runtime, wechat: wechat, alipay: alipay, fuiou: fuiou, confirm: confirm, audit: audit, now: time.Now}
|
||
}
|
||
|
||
// ProcessBatch 按固定批次读取本地待处理事实并调用对应渠道收敛状态。
|
||
func (s *RecoverOnlinePaymentService) ProcessBatch(ctx context.Context) (int, error) {
|
||
if s == nil || s.db == nil || s.runtime == nil || s.wechat == nil || s.alipay == nil || s.fuiou == nil || s.confirm == nil || s.audit == nil {
|
||
return 0, errors.New(errors.CodeServiceUnavailable, "代理在线充值支付恢复能力未配置")
|
||
}
|
||
now := s.now().UTC()
|
||
var payments []model.Payment
|
||
if err := s.db.WithContext(ctx).
|
||
Where("order_type = ? AND status = ? AND created_at <= ?", model.PaymentOrderTypeAgentRecharge, model.PaymentRecordStatusPending, now.Add(-constants.AgentRechargeRecoveryMinimumAge)).
|
||
Order("created_at ASC, id ASC").Limit(constants.AgentRechargeRecoveryBatchSize).
|
||
Find(&payments).Error; err != nil {
|
||
return 0, errors.Wrap(errors.CodeDatabaseError, err, "扫描待恢复代理充值支付单失败")
|
||
}
|
||
if len(payments) == 0 {
|
||
return 0, nil
|
||
}
|
||
recharges, configs, err := s.loadRecoveryFacts(ctx, payments)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
processed := 0
|
||
var firstErr error
|
||
for index := range payments {
|
||
payment := &payments[index]
|
||
recharge := recharges[payment.OrderID]
|
||
config, configErr := s.recoveryConfig(ctx, payment, configs)
|
||
if configErr != nil {
|
||
if firstErr == nil {
|
||
firstErr = configErr
|
||
}
|
||
continue
|
||
}
|
||
if recharge == nil || config == nil {
|
||
if firstErr == nil {
|
||
firstErr = errors.New(errors.CodeConflict, "待恢复支付单缺少充值单或创建配置")
|
||
}
|
||
continue
|
||
}
|
||
if err := s.recoverOne(ctx, payment, recharge, config, now); err != nil {
|
||
if firstErr == nil {
|
||
firstErr = err
|
||
}
|
||
continue
|
||
}
|
||
processed++
|
||
}
|
||
return processed, firstErr
|
||
}
|
||
|
||
func (s *RecoverOnlinePaymentService) recoverOne(ctx context.Context, payment *model.Payment, recharge *model.AgentRechargeRecord, config *model.WechatConfig, now time.Time) error {
|
||
adapter := s.adapter(payment.PaymentMethod, config)
|
||
if adapter == nil {
|
||
return errors.New(errors.CodeNoPaymentConfig, "代理充值创建时支付配置不可用")
|
||
}
|
||
if payment.QRContent == "" {
|
||
if !adapter.Available(config) {
|
||
return errors.New(errors.CodeNoPaymentConfig, "代理充值支付配置不可用")
|
||
}
|
||
// 支付宝 WAP 链接由本地签名生成，可以安全重建；微信 H5 下单结果未知时只允许查单。
|
||
if payment.PaymentMethod != constants.RechargeMethodAlipay {
|
||
return s.queryPayment(ctx, adapter, payment, recharge, config)
|
||
}
|
||
expireAt := now.Add(30 * time.Minute)
|
||
if payment.ExpireAt != nil && payment.ExpireAt.After(now) {
|
||
expireAt = *payment.ExpireAt
|
||
}
|
||
result, err := adapter.CreatePaymentURL(ctx, OnlinePaymentRequest{
|
||
PaymentID: payment.ID, PaymentNo: payment.PaymentNo, CorrelationID: payment.PaymentNo,
|
||
Description: "代理主钱包充值", Amount: payment.Amount,
|
||
ExpireAt: expireAt, Config: config,
|
||
})
|
||
if err != nil {
|
||
// 恢复阶段不能仅凭链接生成错误推断未收款，保留本地状态等待下次查单。
|
||
return nil
|
||
}
|
||
if result.QRContent == "" {
|
||
return nil
|
||
}
|
||
update := s.db.WithContext(ctx).Model(&model.Payment{}).
|
||
Where("id = ? AND status = ? AND qr_content = ''", payment.ID, model.PaymentRecordStatusPending).
|
||
Update("qr_content", result.QRContent)
|
||
if update.Error != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, update.Error, "恢复代理充值支付链接失败")
|
||
}
|
||
return nil
|
||
}
|
||
return s.queryPayment(ctx, adapter, payment, recharge, config)
|
||
}
|
||
|
||
func (s *RecoverOnlinePaymentService) queryPayment(ctx context.Context, adapter OnlinePaymentPort, payment *model.Payment, recharge *model.AgentRechargeRecord, config *model.WechatConfig) error {
|
||
queryResult, err := adapter.Query(ctx, OnlinePaymentRequest{
|
||
PaymentID: payment.ID, PaymentNo: payment.PaymentNo, CorrelationID: payment.PaymentNo, Config: config,
|
||
})
|
||
if err != nil {
|
||
return nil
|
||
}
|
||
switch queryResult.State {
|
||
case OnlinePaymentStatePaid:
|
||
if queryResult.PaidAt == nil {
|
||
return errors.New(errors.CodeConflict, "支付渠道成功结果缺少支付时间")
|
||
}
|
||
_, err = s.confirm.Execute(ctx, ConfirmOnlinePaymentCommand{
|
||
PaymentNo: payment.PaymentNo, PaymentMethod: payment.PaymentMethod, ConfigID: config.ID,
|
||
MerchantIdentity: paymentMerchantIdentity(payment.PaymentMethod, config),
|
||
ThirdPartyTradeNo: queryResult.ThirdPartyTradeNo, Amount: queryResult.Amount, PaidAt: *queryResult.PaidAt,
|
||
CorrelationID: payment.PaymentNo,
|
||
})
|
||
return err
|
||
case OnlinePaymentStateClosed:
|
||
return s.closePending(ctx, payment, recharge)
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func (s *RecoverOnlinePaymentService) loadRecoveryFacts(ctx context.Context, payments []model.Payment) (map[uint]*model.AgentRechargeRecord, map[uint]*model.WechatConfig, error) {
|
||
rechargeIDs := make([]uint, 0, len(payments))
|
||
configIDs := make([]uint, 0, len(payments))
|
||
for index := range payments {
|
||
rechargeIDs = append(rechargeIDs, payments[index].OrderID)
|
||
if payments[index].MerchantID == nil && payments[index].PaymentConfigID != nil {
|
||
configIDs = append(configIDs, *payments[index].PaymentConfigID)
|
||
}
|
||
}
|
||
var rechargeRows []model.AgentRechargeRecord
|
||
if err := s.db.WithContext(ctx).Where("id IN ?", rechargeIDs).Find(&rechargeRows).Error; err != nil {
|
||
return nil, nil, errors.Wrap(errors.CodeDatabaseError, err, "批量查询待恢复代理充值单失败")
|
||
}
|
||
var configRows []model.WechatConfig
|
||
if len(configIDs) > 0 {
|
||
// merchant_id 为空仅为留存期内历史支付；独立 Change 删除旧路径前，
|
||
// 必须按其 payment_config_id 读取，包括已软删除的原始配置。
|
||
if err := s.db.WithContext(ctx).Unscoped().Where("id IN ?", configIDs).Find(&configRows).Error; err != nil {
|
||
return nil, nil, errors.Wrap(errors.CodeDatabaseError, err, "批量查询代理充值历史配置失败")
|
||
}
|
||
}
|
||
recharges := make(map[uint]*model.AgentRechargeRecord, len(rechargeRows))
|
||
for index := range rechargeRows {
|
||
recharges[rechargeRows[index].ID] = &rechargeRows[index]
|
||
}
|
||
configs := make(map[uint]*model.WechatConfig, len(configRows))
|
||
for index := range configRows {
|
||
configs[configRows[index].ID] = &configRows[index]
|
||
}
|
||
return recharges, configs, nil
|
||
}
|
||
|
||
// recoveryConfig 对冻结商户支付单按当前凭证版本加载；历史支付单保持 payment_config_id 路径。
|
||
func (s *RecoverOnlinePaymentService) recoveryConfig(ctx context.Context, payment *model.Payment, configs map[uint]*model.WechatConfig) (*model.WechatConfig, error) {
|
||
if payment.MerchantID != nil {
|
||
merchant, err := s.runtime.LoadMerchant(ctx, *payment.MerchantID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 仅微信直连（v3/v2）商户需要全局授权配置中的 AppID；富友商户不依赖该配置。
|
||
var authorization *model.WechatAuthorization
|
||
if merchant.ProviderType == model.ProviderTypeWechat || merchant.ProviderType == model.ProviderTypeWechatV2 {
|
||
authorization, err = s.runtime.LoadAuthorization(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
return merchantpayment.MerchantConfig(merchant, authorization)
|
||
}
|
||
if payment.PaymentConfigID == nil {
|
||
return nil, nil
|
||
}
|
||
return configs[*payment.PaymentConfigID], nil
|
||
}
|
||
|
||
func (s *RecoverOnlinePaymentService) closePending(ctx context.Context, payment *model.Payment, recharge *model.AgentRechargeRecord) error {
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
paymentUpdate := tx.Model(&model.Payment{}).
|
||
Where("id = ? AND status = ?", payment.ID, model.PaymentRecordStatusPending).
|
||
Update("status", model.PaymentRecordStatusFailed)
|
||
if paymentUpdate.Error != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, paymentUpdate.Error, "关闭失效代理充值支付单失败")
|
||
}
|
||
rechargeUpdate := tx.Model(&model.AgentRechargeRecord{}).
|
||
Where("id = ? AND status = ?", recharge.ID, constants.RechargeStatusPending).
|
||
Update("status", constants.RechargeStatusClosed)
|
||
if rechargeUpdate.Error != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, rechargeUpdate.Error, "关闭失效代理充值单失败")
|
||
}
|
||
if paymentUpdate.RowsAffected == 0 {
|
||
return nil
|
||
}
|
||
afterPayment := *payment
|
||
afterPayment.Status = model.PaymentRecordStatusFailed
|
||
return s.audit.WriteAgentRechargePayment(ctx, tx, PaymentAudit{
|
||
ActionCode: constants.AuditActionPaymentFailed, Summary: "支付渠道确认订单已关闭",
|
||
Payment: &afterPayment, Recharge: recharge,
|
||
BeforeData: map[string]any{"status": payment.Status}, AfterData: map[string]any{"status": afterPayment.Status},
|
||
RechargeBeforeData: map[string]any{"status": recharge.Status}, RechargeAfterData: map[string]any{"status": constants.RechargeStatusClosed},
|
||
})
|
||
})
|
||
}
|
||
|
||
func (s *RecoverOnlinePaymentService) adapter(paymentMethod string, config *model.WechatConfig) OnlinePaymentPort {
|
||
if paymentMethod == constants.RechargeMethodWechat {
|
||
if config != nil && config.ProviderType == model.ProviderTypeFuiou {
|
||
return s.fuiou
|
||
}
|
||
return s.wechat
|
||
}
|
||
if paymentMethod == constants.RechargeMethodAlipay {
|
||
return s.alipay
|
||
}
|
||
return nil
|
||
}
|