break
cbf909b878
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m6s
190 lines
7.1 KiB
Go
190 lines
7.1 KiB
Go
package agentrecharge
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"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
|
|
wechat OnlinePaymentPort
|
|
alipay OnlinePaymentPort
|
|
confirm *ConfirmOnlinePaymentService
|
|
now func() time.Time
|
|
}
|
|
|
|
// NewRecoverOnlinePaymentService 创建代理在线充值支付恢复用例。
|
|
func NewRecoverOnlinePaymentService(db *gorm.DB, wechat, alipay OnlinePaymentPort, confirm *ConfirmOnlinePaymentService) *RecoverOnlinePaymentService {
|
|
return &RecoverOnlinePaymentService{db: db, wechat: wechat, alipay: alipay, confirm: confirm, now: time.Now}
|
|
}
|
|
|
|
// ProcessBatch 按固定批次读取本地待处理事实并调用对应渠道收敛状态。
|
|
func (s *RecoverOnlinePaymentService) ProcessBatch(ctx context.Context) (int, error) {
|
|
if s == nil || s.db == nil || s.wechat == nil || s.alipay == nil || s.confirm == 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 := recoveryConfig(payment, configs)
|
|
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)
|
|
if adapter == nil {
|
|
return errors.New(errors.CodeNoPaymentConfig, "代理充值创建时支付配置不可用")
|
|
}
|
|
if payment.QRContent == "" {
|
|
if !adapter.Available(config) {
|
|
return errors.New(errors.CodeNoPaymentConfig, "代理充值预下单配置不可用")
|
|
}
|
|
expireAt := now.Add(30 * time.Minute)
|
|
if payment.ExpireAt != nil && payment.ExpireAt.After(now) {
|
|
expireAt = *payment.ExpireAt
|
|
}
|
|
result, err := adapter.PreCreate(ctx, OnlinePaymentRequest{
|
|
PaymentNo: 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
|
|
}
|
|
queryResult, err := adapter.Query(ctx, payment.PaymentNo, 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.ID, recharge.ID)
|
|
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].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 {
|
|
if err := s.db.WithContext(ctx).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
|
|
}
|
|
|
|
func recoveryConfig(payment *model.Payment, configs map[uint]*model.WechatConfig) *model.WechatConfig {
|
|
if payment.PaymentConfigID == nil {
|
|
return nil
|
|
}
|
|
return configs[*payment.PaymentConfigID]
|
|
}
|
|
|
|
func (s *RecoverOnlinePaymentService) closePending(ctx context.Context, paymentID, rechargeID uint) error {
|
|
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
paymentUpdate := tx.Model(&model.Payment{}).
|
|
Where("id = ? AND status = ?", paymentID, 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 = ?", rechargeID, constants.RechargeStatusPending).
|
|
Update("status", constants.RechargeStatusClosed)
|
|
if rechargeUpdate.Error != nil {
|
|
return errors.Wrap(errors.CodeDatabaseError, rechargeUpdate.Error, "关闭失效代理充值单失败")
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (s *RecoverOnlinePaymentService) adapter(paymentMethod string) OnlinePaymentPort {
|
|
if paymentMethod == constants.RechargeMethodWechat {
|
|
return s.wechat
|
|
}
|
|
if paymentMethod == constants.RechargeMethodAlipay {
|
|
return s.alipay
|
|
}
|
|
return nil
|
|
}
|