新增收款商户、商户池轮询、微信授权配置独立管理;三类新支付 (C端套餐购买、C端资产钱包充值、代理在线预存款充值)无条件 经商户池选择并冻结路由,无旧综合配置回退。merchant_id 为空 历史支付继续按 payment_config_id 双读。凭证版本化加载与 ID+版本缓存保证轮换一致性。删除商户池新支付创建开关及全部 引用。
This commit is contained in:
721
internal/application/merchantpayment/management.go
Normal file
721
internal/application/merchantpayment/management.go
Normal file
@@ -0,0 +1,721 @@
|
||||
// Package merchantpayment provides merchant pool payment routing use cases.
|
||||
package merchantpayment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
|
||||
systemconfigapp "github.com/break/junhong_cmp_fiber/internal/application/systemconfig"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||||
"github.com/bytedance/sonic"
|
||||
)
|
||||
|
||||
// ManagementService 负责商户、商户池与授权配置写入。
|
||||
type ManagementService struct {
|
||||
db *gorm.DB
|
||||
audit systemconfigapp.AuditWriter
|
||||
}
|
||||
|
||||
// NewManagementService 创建商户配置用例。
|
||||
func NewManagementService(db *gorm.DB, audit systemconfigapp.AuditWriter) *ManagementService {
|
||||
return &ManagementService{db: db, audit: audit}
|
||||
}
|
||||
|
||||
func requireManager(ctx context.Context) error {
|
||||
kind := middleware.GetUserTypeFromContext(ctx)
|
||||
if kind != constants.UserTypeSuperAdmin && kind != constants.UserTypePlatform {
|
||||
return errors.New(errors.CodeForbidden, "无权限访问支付商户配置")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func normalizePage(page, size int) (int, int) {
|
||||
if page < 1 {
|
||||
page = 1
|
||||
}
|
||||
if size < 1 {
|
||||
size = 20
|
||||
}
|
||||
if size > 100 {
|
||||
size = 100
|
||||
}
|
||||
return page, size
|
||||
}
|
||||
|
||||
func validPaymentMethod(method string) bool {
|
||||
return method == "wechat" || method == "alipay"
|
||||
}
|
||||
|
||||
func validateMerchantConfiguration(paymentMethod, providerType, merchantIdentity string, credentials model.JSONB) error {
|
||||
paymentMethod, providerType, merchantIdentity = strings.TrimSpace(paymentMethod), strings.TrimSpace(providerType), strings.TrimSpace(merchantIdentity)
|
||||
if !validPaymentMethod(paymentMethod) || merchantIdentity == "" || len(credentials) == 0 {
|
||||
return errors.New(errors.CodeInvalidParam, "支付商户配置不完整")
|
||||
}
|
||||
var config model.WechatConfig
|
||||
raw, err := sonic.Marshal(credentials)
|
||||
if err != nil || sonic.Unmarshal(raw, &config) != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "支付商户凭证格式无效")
|
||||
}
|
||||
switch paymentMethod {
|
||||
case "wechat":
|
||||
switch providerType {
|
||||
case model.ProviderTypeWechat:
|
||||
if config.WxMchID != merchantIdentity || strings.TrimSpace(config.WxAPIV3Key) == "" || strings.TrimSpace(config.WxCertContent) == "" || strings.TrimSpace(config.WxKeyContent) == "" || strings.TrimSpace(config.WxSerialNo) == "" || strings.TrimSpace(config.WxNotifyURL) == "" {
|
||||
return errors.New(errors.CodeInvalidParam, "微信直连商户凭证不完整或身份不一致")
|
||||
}
|
||||
case model.ProviderTypeWechatV2:
|
||||
if config.WxMchID != merchantIdentity || strings.TrimSpace(config.WxAPIV2Key) == "" || strings.TrimSpace(config.WxNotifyURL) == "" {
|
||||
return errors.New(errors.CodeInvalidParam, "微信 v2 商户凭证不完整或身份不一致")
|
||||
}
|
||||
case model.ProviderTypeFuiou:
|
||||
if config.FyMchntCd != merchantIdentity || strings.TrimSpace(config.FyInsCd) == "" || strings.TrimSpace(config.FyTermID) == "" || strings.TrimSpace(config.FyPrivateKey) == "" || strings.TrimSpace(config.FyPublicKey) == "" || strings.TrimSpace(config.FyAPIURL) == "" || strings.TrimSpace(config.FyNotifyURL) == "" {
|
||||
return errors.New(errors.CodeInvalidParam, "富友商户凭证不完整或身份不一致")
|
||||
}
|
||||
default:
|
||||
return errors.New(errors.CodeInvalidParam, "微信支付服务商类型无效")
|
||||
}
|
||||
case "alipay":
|
||||
if providerType != "alipay" || config.AliAppID != merchantIdentity || strings.TrimSpace(config.AliPrivateKey) == "" || strings.TrimSpace(config.AliPublicKey) == "" || strings.TrimSpace(config.AliNotifyURL) == "" || strings.TrimSpace(config.AliReturnURL) == "" {
|
||||
return errors.New(errors.CodeInvalidParam, "支付宝商户凭证不完整或身份不一致")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validatePoolRequest(req dto.PaymentMerchantPoolRequest) error {
|
||||
if !validPaymentMethod(strings.TrimSpace(req.PaymentMethod)) {
|
||||
return errors.New(errors.CodeInvalidParam, "支付方式仅支持微信或支付宝")
|
||||
}
|
||||
switch req.Strategy {
|
||||
case model.PaymentMerchantStrategyAmount:
|
||||
if req.ThresholdAmount == nil || *req.ThresholdAmount <= 0 || req.ThresholdCount != nil || req.StatisticCycle == nil || !validStatisticCycle(*req.StatisticCycle) || req.TimePeriodValue != nil || req.TimePeriodUnit != nil || req.TimePeriodStartedAt != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "金额轮询策略参数不完整")
|
||||
}
|
||||
case model.PaymentMerchantStrategyCount:
|
||||
if req.ThresholdCount == nil || *req.ThresholdCount <= 0 || req.ThresholdAmount != nil || req.StatisticCycle == nil || !validStatisticCycle(*req.StatisticCycle) || req.TimePeriodValue != nil || req.TimePeriodUnit != nil || req.TimePeriodStartedAt != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "笔数轮询策略参数不完整")
|
||||
}
|
||||
case model.PaymentMerchantStrategyTime:
|
||||
if req.TimePeriodValue == nil || *req.TimePeriodValue < 1 || req.TimePeriodUnit == nil || !validTimeUnit(*req.TimePeriodUnit) || req.TimePeriodStartedAt == nil || req.ThresholdAmount != nil || req.ThresholdCount != nil || req.StatisticCycle != nil {
|
||||
return errors.New(errors.CodeInvalidParam, "时间轮询策略参数不完整")
|
||||
}
|
||||
default:
|
||||
return errors.New(errors.CodeInvalidParam, "不支持的商户池轮询策略")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validStatisticCycle(value string) bool {
|
||||
return value == "round" || value == "day" || value == "month"
|
||||
}
|
||||
func validTimeUnit(value string) bool { return value == "minute" || value == "hour" || value == "day" }
|
||||
|
||||
func (s *ManagementService) writeAudit(ctx context.Context, tx *gorm.DB, operation, description, key, name string, id uint, identity, before, after map[string]any) error {
|
||||
if s.audit == nil {
|
||||
return errors.New(errors.CodeInvalidStatus, "支付商户管理审计接缝未配置")
|
||||
}
|
||||
resourceID := strconv.FormatUint(uint64(id), 10)
|
||||
return s.audit.WriteConfigChange(ctx, tx, systemconfigapp.ChangeAudit{
|
||||
OperatorID: middleware.GetUserIDFromContext(ctx), OperationType: operation, Description: description,
|
||||
ConfigKey: key, Module: "payment_merchant", ResourceID: &resourceID, DisplayName: name,
|
||||
Identity: identity, BeforeData: before, AfterData: after, Result: constants.AuditResultSuccess,
|
||||
})
|
||||
}
|
||||
|
||||
func merchantAuditIdentity(m *model.PaymentMerchant) map[string]any {
|
||||
return map[string]any{"id": m.ID, "name": m.Name, "payment_method": m.PaymentMethod, "provider_type": m.ProviderType, "merchant_identity": m.MerchantIdentity, "status": m.Status, "credential_version": m.CredentialVersion}
|
||||
}
|
||||
|
||||
func poolAuditIdentity(p *model.PaymentMerchantPool) map[string]any {
|
||||
return map[string]any{"id": p.ID, "name": p.Name, "payment_method": p.PaymentMethod, "strategy": p.Strategy, "status": p.Status, "routing_epoch": p.RoutingEpoch}
|
||||
}
|
||||
|
||||
func authorizationAuditIdentity(a *model.WechatAuthorization) map[string]any {
|
||||
return map[string]any{"id": a.ID, "status": a.Status, "credential_version": a.CredentialVersion, "oa_app_id": a.OaAppID, "miniapp_app_id": a.MiniappAppID}
|
||||
}
|
||||
|
||||
// CreateMerchant 创建独立管理的支付商户。
|
||||
func (s *ManagementService) CreateMerchant(ctx context.Context, req dto.PaymentMerchantRequest) (*dto.PaymentMerchantResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s == nil || s.db == nil || strings.TrimSpace(req.Name) == "" || len(req.Credentials) == 0 {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "商户参数或凭证不完整")
|
||||
}
|
||||
if !validPaymentMethod(strings.TrimSpace(req.PaymentMethod)) {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "支付方式仅支持微信或支付宝")
|
||||
}
|
||||
m := &model.PaymentMerchant{Name: strings.TrimSpace(req.Name), PaymentMethod: strings.TrimSpace(req.PaymentMethod), ProviderType: strings.TrimSpace(req.ProviderType), MerchantIdentity: strings.TrimSpace(req.MerchantIdentity), Credentials: req.Credentials, CredentialVersion: 1, Remark: strings.TrimSpace(req.Remark), BaseModel: model.BaseModel{Creator: middleware.GetUserIDFromContext(ctx), Updater: middleware.GetUserIDFromContext(ctx)}}
|
||||
if req.Enabled {
|
||||
m.Status = model.PaymentMerchantStatusEnabled
|
||||
}
|
||||
if m.MerchantIdentity == "" || m.ProviderType == "" {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "商户身份或服务商类型不能为空")
|
||||
}
|
||||
if err := validateMerchantConfiguration(m.PaymentMethod, m.ProviderType, m.MerchantIdentity, m.Credentials); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(m).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return s.writeAudit(ctx, tx, constants.AuditOperationPaymentConfigCreate, "创建支付商户", "payment_merchant:"+strconv.FormatUint(uint64(m.ID), 10), m.Name, m.ID, merchantAuditIdentity(m), nil, merchantAuditIdentity(m))
|
||||
}); err != nil {
|
||||
if appErr, ok := err.(*errors.AppError); ok {
|
||||
return nil, appErr
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "创建支付商户失败")
|
||||
}
|
||||
return merchantResponse(m), nil
|
||||
}
|
||||
|
||||
// ListMerchants returns the privileged configuration projection.
|
||||
|
||||
// ListPools returns one page of merchant pools and their ordered members without per-pool member queries.
|
||||
func (s *ManagementService) ListPools(ctx context.Context, req dto.PaymentMerchantPoolListRequest) ([]*dto.PaymentMerchantPoolResponse, int64, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
page, size := normalizePage(req.Page, req.PageSize)
|
||||
query := s.db.WithContext(ctx).Model(&model.PaymentMerchantPool{})
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "统计商户池失败")
|
||||
}
|
||||
var pools []model.PaymentMerchantPool
|
||||
if err := query.Order("id DESC").Offset((page - 1) * size).Limit(size).Find(&pools).Error; err != nil {
|
||||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询商户池失败")
|
||||
}
|
||||
poolIDs := make([]uint, 0, len(pools))
|
||||
for index := range pools {
|
||||
poolIDs = append(poolIDs, pools[index].ID)
|
||||
}
|
||||
membersByPool := make(map[uint][]uint, len(pools))
|
||||
if len(poolIDs) > 0 {
|
||||
var members []model.PaymentMerchantPoolMember
|
||||
if err := s.db.WithContext(ctx).Where("pool_id IN ?", poolIDs).Order("pool_id ASC, sort_order ASC").Find(&members).Error; err != nil {
|
||||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询商户池成员失败")
|
||||
}
|
||||
for index := range members {
|
||||
member := &members[index]
|
||||
membersByPool[member.PoolID] = append(membersByPool[member.PoolID], member.MerchantID)
|
||||
}
|
||||
}
|
||||
result := make([]*dto.PaymentMerchantPoolResponse, 0, len(pools))
|
||||
for index := range pools {
|
||||
pool := &pools[index]
|
||||
result = append(result, &dto.PaymentMerchantPoolResponse{ID: pool.ID, Name: pool.Name, PaymentMethod: pool.PaymentMethod, Enabled: pool.Status == model.PaymentMerchantStatusEnabled, Strategy: pool.Strategy, ThresholdAmount: pool.ThresholdAmount, ThresholdCount: pool.ThresholdCount, StatisticCycle: pool.StatisticCycle, TimePeriodValue: pool.TimePeriodValue, TimePeriodUnit: pool.TimePeriodUnit, TimePeriodStartedAt: pool.TimePeriodStartedAt, RoutingEpoch: pool.RoutingEpoch, MemberIDs: membersByPool[pool.ID], Remark: pool.Remark})
|
||||
}
|
||||
return result, total, nil
|
||||
}
|
||||
|
||||
// GetPool 查询一个商户池及其有序成员。
|
||||
func (s *ManagementService) GetPool(ctx context.Context, id uint) (*dto.PaymentMerchantPoolResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pool model.PaymentMerchantPool
|
||||
if err := s.db.WithContext(ctx).First(&pool, id).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "商户池不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询商户池失败")
|
||||
}
|
||||
return poolResponse(ctx, s.db, &pool)
|
||||
}
|
||||
|
||||
// ListMerchants 分页查询特权商户配置。
|
||||
func (s *ManagementService) ListMerchants(ctx context.Context, req dto.PaymentMerchantListRequest) ([]*dto.PaymentMerchantResponse, int64, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
page, size := normalizePage(req.Page, req.PageSize)
|
||||
query := s.db.WithContext(ctx).Model(&model.PaymentMerchant{})
|
||||
if req.PaymentMethod != nil {
|
||||
query = query.Where("payment_method = ?", strings.TrimSpace(*req.PaymentMethod))
|
||||
}
|
||||
if req.Enabled != nil {
|
||||
status := model.PaymentMerchantStatusDisabled
|
||||
if *req.Enabled {
|
||||
status = model.PaymentMerchantStatusEnabled
|
||||
}
|
||||
query = query.Where("status = ?", status)
|
||||
}
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询支付商户失败")
|
||||
}
|
||||
var rows []model.PaymentMerchant
|
||||
if err := query.Order("id DESC").Offset((page - 1) * size).Limit(size).Find(&rows).Error; err != nil {
|
||||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询支付商户失败")
|
||||
}
|
||||
result := make([]*dto.PaymentMerchantResponse, 0, len(rows))
|
||||
for index := range rows {
|
||||
result = append(result, merchantResponse(&rows[index]))
|
||||
}
|
||||
return result, total, nil
|
||||
}
|
||||
|
||||
// GetMerchant 查询一个特权商户配置。
|
||||
func (s *ManagementService) GetMerchant(ctx context.Context, id uint) (*dto.PaymentMerchantResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var m model.PaymentMerchant
|
||||
if err := s.db.WithContext(ctx).First(&m, id).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "支付商户不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询支付商户失败")
|
||||
}
|
||||
return merchantResponse(&m), nil
|
||||
}
|
||||
|
||||
// UpdateMerchant 更新商户凭证和可变配置。
|
||||
func (s *ManagementService) UpdateMerchant(ctx context.Context, id uint, req dto.PaymentMerchantUpdateRequest) (*dto.PaymentMerchantResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if req.PaymentMethod != nil && !validPaymentMethod(strings.TrimSpace(*req.PaymentMethod)) {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "支付方式仅支持微信或支付宝")
|
||||
}
|
||||
var m model.PaymentMerchant
|
||||
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&m, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
before := merchantAuditIdentity(&m)
|
||||
previous := m
|
||||
var refs int64
|
||||
if err := tx.Model(&model.Payment{}).Where("merchant_id = ?", id).Count(&refs).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if refs > 0 && ((req.PaymentMethod != nil && *req.PaymentMethod != m.PaymentMethod) || (req.ProviderType != nil && *req.ProviderType != m.ProviderType) || (req.MerchantIdentity != nil && *req.MerchantIdentity != m.MerchantIdentity)) {
|
||||
return errors.New(errors.CodeConflict, "已被支付单引用,不能修改收款身份")
|
||||
}
|
||||
if req.Name != nil {
|
||||
m.Name = strings.TrimSpace(*req.Name)
|
||||
}
|
||||
if req.PaymentMethod != nil {
|
||||
m.PaymentMethod = strings.TrimSpace(*req.PaymentMethod)
|
||||
}
|
||||
if req.ProviderType != nil {
|
||||
m.ProviderType = strings.TrimSpace(*req.ProviderType)
|
||||
}
|
||||
if req.MerchantIdentity != nil {
|
||||
m.MerchantIdentity = strings.TrimSpace(*req.MerchantIdentity)
|
||||
}
|
||||
if req.Remark != nil {
|
||||
m.Remark = strings.TrimSpace(*req.Remark)
|
||||
}
|
||||
if req.Enabled != nil {
|
||||
m.Status = model.PaymentMerchantStatusDisabled
|
||||
if *req.Enabled {
|
||||
m.Status = model.PaymentMerchantStatusEnabled
|
||||
}
|
||||
}
|
||||
if req.Credentials != nil && !reflect.DeepEqual(m.Credentials, *req.Credentials) {
|
||||
m.Credentials = *req.Credentials
|
||||
}
|
||||
if previous.Name != m.Name || previous.PaymentMethod != m.PaymentMethod || previous.ProviderType != m.ProviderType || previous.MerchantIdentity != m.MerchantIdentity || previous.Status != m.Status || !reflect.DeepEqual(previous.Credentials, m.Credentials) {
|
||||
m.CredentialVersion++
|
||||
}
|
||||
if err := validateMerchantConfiguration(m.PaymentMethod, m.ProviderType, m.MerchantIdentity, m.Credentials); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Save(&m).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return s.writeAudit(ctx, tx, constants.AuditOperationPaymentConfigUpdate, "更新支付商户", "payment_merchant:"+strconv.FormatUint(uint64(m.ID), 10), m.Name, m.ID, merchantAuditIdentity(&m), before, merchantAuditIdentity(&m))
|
||||
}); err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "支付商户不存在")
|
||||
}
|
||||
if appErr, ok := err.(*errors.AppError); ok {
|
||||
return nil, appErr
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "更新支付商户失败")
|
||||
}
|
||||
return merchantResponse(&m), nil
|
||||
}
|
||||
|
||||
// DeleteMerchant 仅在未被引用且二次确认后删除商户。
|
||||
func (s *ManagementService) DeleteMerchant(ctx context.Context, id uint, confirm bool) error {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
if !confirm {
|
||||
return errors.New(errors.CodeInvalidParam, "删除商户必须二次确认")
|
||||
}
|
||||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var refs, members int64
|
||||
var merchant model.PaymentMerchant
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&merchant, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
before := merchantAuditIdentity(&merchant)
|
||||
if err := tx.Model(&model.Payment{}).Where("merchant_id = ?", id).Count(&refs).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if refs > 0 {
|
||||
return errors.New(errors.CodeConflict, "已被支付单引用的商户不能删除")
|
||||
}
|
||||
if err := tx.Model(&model.PaymentMerchantPoolMember{}).Where("merchant_id = ?", id).Count(&members).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if members > 0 {
|
||||
return errors.New(errors.CodeConflict, "商户仍属于商户池")
|
||||
}
|
||||
r := tx.Delete(&model.PaymentMerchant{}, id)
|
||||
if r.Error != nil {
|
||||
return r.Error
|
||||
}
|
||||
if r.RowsAffected == 0 {
|
||||
return errors.New(errors.CodeNotFound, "支付商户不存在")
|
||||
}
|
||||
if err := s.writeAudit(ctx, tx, constants.AuditOperationPaymentConfigDelete, "删除支付商户", "payment_merchant:"+strconv.FormatUint(uint64(merchant.ID), 10), merchant.Name, merchant.ID, before, before, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// SavePool 创建或更新商户池,并原子替换有序成员。
|
||||
func (s *ManagementService) SavePool(ctx context.Context, id uint, req dto.PaymentMerchantPoolRequest) (*dto.PaymentMerchantPoolResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := validatePoolRequest(req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(req.MemberIDs) == 0 {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "商户池至少需要一个商户")
|
||||
}
|
||||
var pool model.PaymentMerchantPool
|
||||
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
creating := id == 0
|
||||
var previousMemberIDs []uint
|
||||
if !creating {
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&pool, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
var previousMembers []model.PaymentMerchantPoolMember
|
||||
if err := tx.Where("pool_id = ?", pool.ID).Order("sort_order ASC").Find(&previousMembers).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
previousMemberIDs = make([]uint, 0, len(previousMembers))
|
||||
for _, member := range previousMembers {
|
||||
previousMemberIDs = append(previousMemberIDs, member.MerchantID)
|
||||
}
|
||||
} else {
|
||||
pool.Creator = middleware.GetUserIDFromContext(ctx)
|
||||
pool.RoutingEpoch = 1
|
||||
}
|
||||
before := poolAuditIdentity(&pool)
|
||||
if err := validatePoolMembers(ctx, tx, req.PaymentMethod, req.MemberIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
if !creating && poolEpochChanged(&pool, &req, previousMemberIDs) {
|
||||
pool.RoutingEpoch++
|
||||
}
|
||||
pool.Name, pool.PaymentMethod, pool.Strategy, pool.Remark = strings.TrimSpace(req.Name), strings.TrimSpace(req.PaymentMethod), strings.TrimSpace(req.Strategy), strings.TrimSpace(req.Remark)
|
||||
pool.ThresholdAmount, pool.ThresholdCount, pool.StatisticCycle, pool.TimePeriodValue, pool.TimePeriodUnit, pool.TimePeriodStartedAt = req.ThresholdAmount, req.ThresholdCount, req.StatisticCycle, req.TimePeriodValue, req.TimePeriodUnit, req.TimePeriodStartedAt
|
||||
if req.Enabled {
|
||||
var others int64
|
||||
if err := tx.Model(&model.PaymentMerchantPool{}).Where("payment_method = ? AND status = ? AND id <> ?", pool.PaymentMethod, model.PaymentMerchantStatusEnabled, pool.ID).Count(&others).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if others > 0 {
|
||||
return errors.New(errors.CodeConflict, "该支付方式已有启用商户池")
|
||||
}
|
||||
}
|
||||
pool.Status = model.PaymentMerchantStatusDisabled
|
||||
if req.Enabled {
|
||||
pool.Status = model.PaymentMerchantStatusEnabled
|
||||
}
|
||||
pool.Updater = middleware.GetUserIDFromContext(ctx)
|
||||
if creating {
|
||||
if err := tx.Create(&pool).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else if err := tx.Save(&pool).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Where("pool_id = ?", pool.ID).Delete(&model.PaymentMerchantPoolMember{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
members := make([]model.PaymentMerchantPoolMember, 0, len(req.MemberIDs))
|
||||
for i, merchantID := range req.MemberIDs {
|
||||
members = append(members, model.PaymentMerchantPoolMember{PoolID: pool.ID, MerchantID: merchantID, SortOrder: int64(i), BaseModel: model.BaseModel{Creator: middleware.GetUserIDFromContext(ctx), Updater: middleware.GetUserIDFromContext(ctx)}})
|
||||
}
|
||||
if err := tx.Create(&members).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
op := constants.AuditOperationPaymentConfigUpdate
|
||||
summary := "更新商户池"
|
||||
if creating {
|
||||
op = constants.AuditOperationPaymentConfigCreate
|
||||
summary = "创建商户池"
|
||||
}
|
||||
return s.writeAudit(ctx, tx, op, summary, "payment_merchant_pool:"+strconv.FormatUint(uint64(pool.ID), 10), pool.Name, pool.ID, poolAuditIdentity(&pool), before, poolAuditIdentity(&pool))
|
||||
})
|
||||
if err != nil {
|
||||
if appErr, ok := err.(*errors.AppError); ok {
|
||||
return nil, appErr
|
||||
}
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "商户池不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "保存商户池失败")
|
||||
}
|
||||
return poolResponse(ctx, s.db, &pool)
|
||||
}
|
||||
|
||||
// SetPoolEnabled enables or disables a pool after rechecking the active-pool and member invariants.
|
||||
func (s *ManagementService) SetPoolEnabled(ctx context.Context, id uint, enabled bool) (*dto.PaymentMerchantPoolResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pool model.PaymentMerchantPool
|
||||
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&pool, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
before := poolAuditIdentity(&pool)
|
||||
if enabled {
|
||||
var others int64
|
||||
if err := tx.Model(&model.PaymentMerchantPool{}).Where("payment_method = ? AND status = ? AND id <> ?", pool.PaymentMethod, model.PaymentMerchantStatusEnabled, pool.ID).Count(&others).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if others > 0 {
|
||||
return errors.New(errors.CodeConflict, "该支付方式已有启用商户池")
|
||||
}
|
||||
var members []model.PaymentMerchantPoolMember
|
||||
if err := tx.Where("pool_id = ?", pool.ID).Order("sort_order ASC").Find(&members).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
ids := make([]uint, 0, len(members))
|
||||
for _, member := range members {
|
||||
ids = append(ids, member.MerchantID)
|
||||
}
|
||||
if err := validatePoolMembers(ctx, tx, pool.PaymentMethod, ids); err != nil {
|
||||
return err
|
||||
}
|
||||
pool.Status = model.PaymentMerchantStatusEnabled
|
||||
} else {
|
||||
pool.Status = model.PaymentMerchantStatusDisabled
|
||||
}
|
||||
pool.Updater = middleware.GetUserIDFromContext(ctx)
|
||||
if err := tx.Save(&pool).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
op := constants.AuditOperationPaymentConfigDeactivate
|
||||
summary := "停用商户池"
|
||||
if enabled {
|
||||
op = constants.AuditOperationPaymentConfigActivate
|
||||
summary = "启用商户池"
|
||||
}
|
||||
return s.writeAudit(ctx, tx, op, summary, "payment_merchant_pool:"+strconv.FormatUint(uint64(pool.ID), 10), pool.Name, pool.ID, poolAuditIdentity(&pool), before, poolAuditIdentity(&pool))
|
||||
})
|
||||
if err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "商户池不存在")
|
||||
}
|
||||
if appErr, ok := err.(*errors.AppError); ok {
|
||||
return nil, appErr
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "更新商户池状态失败")
|
||||
}
|
||||
return poolResponse(ctx, s.db, &pool)
|
||||
}
|
||||
|
||||
func validatePoolMembers(ctx context.Context, tx *gorm.DB, method string, ids []uint) error {
|
||||
seen := map[uint]struct{}{}
|
||||
|
||||
for _, id := range ids {
|
||||
if id == 0 {
|
||||
return errors.New(errors.CodeInvalidParam, "商户ID无效")
|
||||
}
|
||||
if _, ok := seen[id]; ok {
|
||||
return errors.New(errors.CodeInvalidParam, "商户池成员不能重复")
|
||||
}
|
||||
seen[id] = struct{}{}
|
||||
}
|
||||
var merchants []model.PaymentMerchant
|
||||
if err := tx.WithContext(ctx).Where("id IN ? AND payment_method = ? AND status = ?", ids, method, model.PaymentMerchantStatusEnabled).Find(&merchants).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if len(merchants) != len(ids) {
|
||||
return errors.New(errors.CodeConflict, "商户池成员必须存在、启用且支付方式一致")
|
||||
}
|
||||
for index := range merchants {
|
||||
merchant := &merchants[index]
|
||||
if err := validateMerchantConfiguration(merchant.PaymentMethod, merchant.ProviderType, merchant.MerchantIdentity, merchant.Credentials); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func poolEpochChanged(pool *model.PaymentMerchantPool, request *dto.PaymentMerchantPoolRequest, previousMemberIDs []uint) bool {
|
||||
if pool.PaymentMethod != strings.TrimSpace(request.PaymentMethod) ||
|
||||
pool.Strategy != request.Strategy ||
|
||||
!sameString(pool.StatisticCycle, request.StatisticCycle) ||
|
||||
!sameInt64(pool.TimePeriodValue, request.TimePeriodValue) ||
|
||||
!sameString(pool.TimePeriodUnit, request.TimePeriodUnit) ||
|
||||
!sameTime(pool.TimePeriodStartedAt, request.TimePeriodStartedAt) {
|
||||
return true
|
||||
}
|
||||
if sameMemberOrder(previousMemberIDs, request.MemberIDs) {
|
||||
return false
|
||||
}
|
||||
return pool.StatisticCycle == nil || (*pool.StatisticCycle != "day" && *pool.StatisticCycle != "month") || !sameMemberSet(previousMemberIDs, request.MemberIDs)
|
||||
}
|
||||
|
||||
func sameMemberOrder(left, right []uint) bool {
|
||||
if len(left) != len(right) {
|
||||
return false
|
||||
}
|
||||
for i := range left {
|
||||
if left[i] != right[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func sameMemberSet(left, right []uint) bool {
|
||||
if len(left) != len(right) {
|
||||
return false
|
||||
}
|
||||
seen := make(map[uint]struct{}, len(left))
|
||||
for _, id := range left {
|
||||
seen[id] = struct{}{}
|
||||
}
|
||||
for _, id := range right {
|
||||
if _, ok := seen[id]; !ok {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func sameString(left, right *string) bool {
|
||||
if left == nil || right == nil {
|
||||
return left == right
|
||||
}
|
||||
return *left == *right
|
||||
}
|
||||
|
||||
func sameInt64(left, right *int64) bool {
|
||||
if left == nil || right == nil {
|
||||
return left == right
|
||||
}
|
||||
return *left == *right
|
||||
}
|
||||
|
||||
func sameTime(a, b *time.Time) bool {
|
||||
if a == nil || b == nil {
|
||||
return a == b
|
||||
}
|
||||
return a.Equal(*b)
|
||||
}
|
||||
func merchantResponse(m *model.PaymentMerchant) *dto.PaymentMerchantResponse {
|
||||
return &dto.PaymentMerchantResponse{ID: m.ID, Name: m.Name, PaymentMethod: m.PaymentMethod, ProviderType: m.ProviderType, MerchantIdentity: m.MerchantIdentity, Credentials: m.Credentials, CredentialVersion: m.CredentialVersion, Enabled: m.Status == model.PaymentMerchantStatusEnabled, Remark: m.Remark, CreatedAt: m.CreatedAt, UpdatedAt: m.UpdatedAt}
|
||||
}
|
||||
func poolResponse(ctx context.Context, db *gorm.DB, p *model.PaymentMerchantPool) (*dto.PaymentMerchantPoolResponse, error) {
|
||||
var rows []model.PaymentMerchantPoolMember
|
||||
if err := db.WithContext(ctx).Where("pool_id = ?", p.ID).Order("sort_order ASC").Find(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ids := make([]uint, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
ids = append(ids, row.MerchantID)
|
||||
}
|
||||
return &dto.PaymentMerchantPoolResponse{ID: p.ID, Name: p.Name, PaymentMethod: p.PaymentMethod, Enabled: p.Status == model.PaymentMerchantStatusEnabled, Strategy: p.Strategy, ThresholdAmount: p.ThresholdAmount, ThresholdCount: p.ThresholdCount, StatisticCycle: p.StatisticCycle, TimePeriodValue: p.TimePeriodValue, TimePeriodUnit: p.TimePeriodUnit, TimePeriodStartedAt: p.TimePeriodStartedAt, RoutingEpoch: p.RoutingEpoch, MemberIDs: ids, Remark: p.Remark}, nil
|
||||
}
|
||||
|
||||
// GetAuthorization 查询特权全局授权配置。
|
||||
func (s *ManagementService) GetAuthorization(ctx context.Context) (*dto.WechatAuthorizationResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a model.WechatAuthorization
|
||||
if err := s.db.WithContext(ctx).First(&a).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询微信授权配置失败")
|
||||
}
|
||||
return authorizationResponse(&a), nil
|
||||
}
|
||||
|
||||
// SaveAuthorization 创建或更新唯一启用的授权配置。
|
||||
func (s *ManagementService) SaveAuthorization(ctx context.Context, req dto.WechatAuthorizationRequest) (*dto.WechatAuthorizationResponse, error) {
|
||||
if err := requireManager(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if req.Enabled && (strings.TrimSpace(req.OaAppID) == "" || strings.TrimSpace(req.OaAppSecret) == "" || strings.TrimSpace(req.MiniappAppID) == "" || strings.TrimSpace(req.MiniappAppSecret) == "") {
|
||||
return nil, errors.New(errors.CodeInvalidParam, "启用微信授权配置时公众号和小程序凭证必须完整")
|
||||
}
|
||||
var authorization model.WechatAuthorization
|
||||
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&authorization).Error
|
||||
if err != nil && err != gorm.ErrRecordNotFound {
|
||||
return err
|
||||
}
|
||||
previousStatus := authorization.Status
|
||||
creating := err == gorm.ErrRecordNotFound
|
||||
before := authorizationAuditIdentity(&authorization)
|
||||
if creating {
|
||||
authorization.Creator = middleware.GetUserIDFromContext(ctx)
|
||||
authorization.CredentialVersion = 1
|
||||
}
|
||||
changed := authorization.OaAppID != req.OaAppID || authorization.OaAppSecret != req.OaAppSecret || authorization.OaToken != req.OaToken || authorization.OaAesKey != req.OaAesKey || authorization.OaOAuthRedirectURL != req.OaOAuthRedirectURL || authorization.MiniappAppID != req.MiniappAppID || authorization.MiniappAppSecret != req.MiniappAppSecret
|
||||
authorization.OaAppID, authorization.OaAppSecret, authorization.OaToken, authorization.OaAesKey, authorization.OaOAuthRedirectURL, authorization.MiniappAppID, authorization.MiniappAppSecret = req.OaAppID, req.OaAppSecret, req.OaToken, req.OaAesKey, req.OaOAuthRedirectURL, req.MiniappAppID, req.MiniappAppSecret
|
||||
authorization.Status = model.PaymentMerchantStatusDisabled
|
||||
if req.Enabled {
|
||||
authorization.Status = model.PaymentMerchantStatusEnabled
|
||||
}
|
||||
if !creating && (changed || previousStatus != authorization.Status) {
|
||||
authorization.CredentialVersion++
|
||||
}
|
||||
authorization.Updater = middleware.GetUserIDFromContext(ctx)
|
||||
if creating {
|
||||
if err := tx.Create(&authorization).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
if err := tx.Save(&authorization).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
op := constants.AuditOperationPaymentConfigUpdate
|
||||
summary := "更新微信授权配置"
|
||||
if creating {
|
||||
op = constants.AuditOperationPaymentConfigCreate
|
||||
summary = "创建微信授权配置"
|
||||
}
|
||||
return s.writeAudit(ctx, tx, op, summary, "wechat_authorization:"+strconv.FormatUint(uint64(authorization.ID), 10), "微信授权配置", authorization.ID, authorizationAuditIdentity(&authorization), before, authorizationAuditIdentity(&authorization))
|
||||
})
|
||||
if err != nil {
|
||||
if appErr, ok := err.(*errors.AppError); ok {
|
||||
return nil, appErr
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "保存微信授权配置失败")
|
||||
}
|
||||
return authorizationResponse(&authorization), nil
|
||||
}
|
||||
func authorizationResponse(a *model.WechatAuthorization) *dto.WechatAuthorizationResponse {
|
||||
return &dto.WechatAuthorizationResponse{ID: a.ID, OaAppID: a.OaAppID, OaAppSecret: a.OaAppSecret, OaToken: a.OaToken, OaAesKey: a.OaAesKey, OaOAuthRedirectURL: a.OaOAuthRedirectURL, MiniappAppID: a.MiniappAppID, MiniappAppSecret: a.MiniappAppSecret, CredentialVersion: a.CredentialVersion, Enabled: a.Status == model.PaymentMerchantStatusEnabled, UpdatedAt: a.UpdatedAt}
|
||||
}
|
||||
386
internal/application/merchantpayment/routing.go
Normal file
386
internal/application/merchantpayment/routing.go
Normal file
@@ -0,0 +1,386 @@
|
||||
package merchantpayment
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/bytedance/sonic"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||||
)
|
||||
|
||||
// RouteSelection is the non-sensitive route frozen onto a new payment.
|
||||
type RouteSelection struct {
|
||||
Merchant *model.PaymentMerchant
|
||||
Pool *model.PaymentMerchantPool
|
||||
}
|
||||
|
||||
// RuntimeLoader loads current merchant and authorization credentials by version.
|
||||
type RuntimeLoader struct {
|
||||
db *gorm.DB
|
||||
redis *redis.Client
|
||||
}
|
||||
|
||||
// merchantCachePayload is used only for the internal versioned Redis cache and deliberately includes credentials.
|
||||
// It must never be used for DTOs, logs, audits, or payment snapshots.
|
||||
type merchantCachePayload struct {
|
||||
ID uint `json:"id"`
|
||||
Name string `json:"name"`
|
||||
PaymentMethod string `json:"payment_method"`
|
||||
ProviderType string `json:"provider_type"`
|
||||
MerchantIdentity string `json:"merchant_identity"`
|
||||
Credentials model.JSONB `json:"credentials"`
|
||||
CredentialVersion int64 `json:"credential_version"`
|
||||
Status int `json:"status"`
|
||||
Remark string `json:"remark"`
|
||||
}
|
||||
|
||||
func merchantCachePayloadFrom(merchant *model.PaymentMerchant) merchantCachePayload {
|
||||
return merchantCachePayload{ID: merchant.ID, Name: merchant.Name, PaymentMethod: merchant.PaymentMethod, ProviderType: merchant.ProviderType, MerchantIdentity: merchant.MerchantIdentity, Credentials: merchant.Credentials, CredentialVersion: merchant.CredentialVersion, Status: merchant.Status, Remark: merchant.Remark}
|
||||
}
|
||||
|
||||
func (p merchantCachePayload) merchant() *model.PaymentMerchant {
|
||||
return &model.PaymentMerchant{Model: gorm.Model{ID: p.ID}, Name: p.Name, PaymentMethod: p.PaymentMethod, ProviderType: p.ProviderType, MerchantIdentity: p.MerchantIdentity, Credentials: p.Credentials, CredentialVersion: p.CredentialVersion, Status: p.Status, Remark: p.Remark}
|
||||
}
|
||||
|
||||
// authorizationCachePayload is used only for the internal versioned Redis cache and deliberately includes secrets.
|
||||
// It must never be used for DTOs, logs, audits, or payment snapshots.
|
||||
type authorizationCachePayload struct {
|
||||
ID uint `json:"id"`
|
||||
OaAppID string `json:"oa_app_id"`
|
||||
OaAppSecret string `json:"oa_app_secret"`
|
||||
OaToken string `json:"oa_token"`
|
||||
OaAesKey string `json:"oa_aes_key"`
|
||||
OaOAuthRedirectURL string `json:"oa_oauth_redirect_url"`
|
||||
MiniappAppID string `json:"miniapp_app_id"`
|
||||
MiniappAppSecret string `json:"miniapp_app_secret"`
|
||||
CredentialVersion int64 `json:"credential_version"`
|
||||
Status int `json:"status"`
|
||||
}
|
||||
|
||||
func authorizationCachePayloadFrom(authorization *model.WechatAuthorization) authorizationCachePayload {
|
||||
return authorizationCachePayload{ID: authorization.ID, OaAppID: authorization.OaAppID, OaAppSecret: authorization.OaAppSecret, OaToken: authorization.OaToken, OaAesKey: authorization.OaAesKey, OaOAuthRedirectURL: authorization.OaOAuthRedirectURL, MiniappAppID: authorization.MiniappAppID, MiniappAppSecret: authorization.MiniappAppSecret, CredentialVersion: authorization.CredentialVersion, Status: authorization.Status}
|
||||
}
|
||||
|
||||
func (p authorizationCachePayload) authorization() *model.WechatAuthorization {
|
||||
return &model.WechatAuthorization{Model: gorm.Model{ID: p.ID}, OaAppID: p.OaAppID, OaAppSecret: p.OaAppSecret, OaToken: p.OaToken, OaAesKey: p.OaAesKey, OaOAuthRedirectURL: p.OaOAuthRedirectURL, MiniappAppID: p.MiniappAppID, MiniappAppSecret: p.MiniappAppSecret, CredentialVersion: p.CredentialVersion, Status: p.Status}
|
||||
}
|
||||
|
||||
func NewRuntimeLoader(db *gorm.DB, redis *redis.Client) *RuntimeLoader {
|
||||
return &RuntimeLoader{db: db, redis: redis}
|
||||
}
|
||||
|
||||
// LoadMerchant first reads the current version from the primary database, then uses only that version's cache entry.
|
||||
// Disabled merchants remain loadable for frozen historical payments.
|
||||
func (l *RuntimeLoader) LoadMerchant(ctx context.Context, id uint) (*model.PaymentMerchant, error) {
|
||||
if l == nil || l.db == nil {
|
||||
return nil, errors.New(errors.CodeServiceUnavailable, "支付商户加载能力未配置")
|
||||
}
|
||||
return l.loadMerchant(ctx, l.db, id)
|
||||
}
|
||||
|
||||
// loadMerchant 先从当前事务或主库读取版本,再仅命中该版本的缓存。
|
||||
// 版本在凭证事务提交时递增,因此提交前遗留的旧缓存永远不会被新读取命中。
|
||||
func (l *RuntimeLoader) loadMerchant(ctx context.Context, db *gorm.DB, id uint) (*model.PaymentMerchant, error) {
|
||||
var current model.PaymentMerchant
|
||||
if err := db.WithContext(ctx).First(¤t, id).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNotFound, "支付商户不存在")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "读取支付商户失败")
|
||||
}
|
||||
key := fmt.Sprintf("payment:merchant:%d:%d", current.ID, current.CredentialVersion)
|
||||
if l.redis != nil {
|
||||
if text, err := l.redis.Get(ctx, key).Result(); err == nil {
|
||||
var cached merchantCachePayload
|
||||
if sonic.UnmarshalString(text, &cached) == nil && cached.ID == current.ID && cached.CredentialVersion == current.CredentialVersion {
|
||||
return cached.merchant(), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
if l.redis != nil {
|
||||
if text, err := sonic.MarshalString(merchantCachePayloadFrom(¤t)); err == nil {
|
||||
_ = l.redis.Set(ctx, key, text, time.Hour).Err()
|
||||
}
|
||||
}
|
||||
return ¤t, nil
|
||||
}
|
||||
|
||||
// LoadAuthorization first reads the current enabled version and only then resolves its versioned cache entry.
|
||||
func (l *RuntimeLoader) LoadAuthorization(ctx context.Context) (*model.WechatAuthorization, error) {
|
||||
if l == nil || l.db == nil {
|
||||
return nil, errors.New(errors.CodeServiceUnavailable, "微信授权加载能力未配置")
|
||||
}
|
||||
var current model.WechatAuthorization
|
||||
if err := l.db.WithContext(ctx).Where("status = ?", model.PaymentMerchantStatusEnabled).First(¤t).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeWechatConfigUnavailable, "微信授权未配置")
|
||||
}
|
||||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "读取微信授权配置失败")
|
||||
}
|
||||
key := fmt.Sprintf("payment:wechat-authorization:%d:%d", current.ID, current.CredentialVersion)
|
||||
if l.redis != nil {
|
||||
if text, err := l.redis.Get(ctx, key).Result(); err == nil {
|
||||
var cached authorizationCachePayload
|
||||
if sonic.UnmarshalString(text, &cached) == nil && cached.ID == current.ID && cached.CredentialVersion == current.CredentialVersion {
|
||||
return cached.authorization(), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
if l.redis != nil {
|
||||
if text, err := sonic.MarshalString(authorizationCachePayloadFrom(¤t)); err == nil {
|
||||
_ = l.redis.Set(ctx, key, text, time.Hour).Err()
|
||||
}
|
||||
}
|
||||
return ¤t, nil
|
||||
}
|
||||
|
||||
// MerchantConfig adapts the merchant credential payload to existing channel constructors without persisting credentials in a payment snapshot.
|
||||
func MerchantConfig(merchant *model.PaymentMerchant, authorization *model.WechatAuthorization) (*model.WechatConfig, error) {
|
||||
if merchant == nil {
|
||||
return nil, errors.New(errors.CodeNoPaymentConfig, "支付商户不存在")
|
||||
}
|
||||
if authorization == nil {
|
||||
authorization = &model.WechatAuthorization{}
|
||||
}
|
||||
raw, err := sonic.Marshal(merchant.Credentials)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(errors.CodeInvalidParam, err, "支付商户凭证格式无效")
|
||||
}
|
||||
var cfg model.WechatConfig
|
||||
if err := sonic.Unmarshal(raw, &cfg); err != nil {
|
||||
return nil, errors.Wrap(errors.CodeInvalidParam, err, "支付商户凭证格式无效")
|
||||
}
|
||||
cfg.ID = merchant.ID
|
||||
cfg.ProviderType = merchant.ProviderType
|
||||
cfg.IsActive = true
|
||||
if authorization != nil {
|
||||
cfg.OaAppID = authorization.OaAppID
|
||||
cfg.OaAppSecret = authorization.OaAppSecret
|
||||
cfg.OaToken = authorization.OaToken
|
||||
cfg.OaAesKey = authorization.OaAesKey
|
||||
cfg.OaOAuthRedirectURL = authorization.OaOAuthRedirectURL
|
||||
cfg.MiniappAppID = authorization.MiniappAppID
|
||||
cfg.MiniappAppSecret = authorization.MiniappAppSecret
|
||||
}
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
// MerchantConfigWithAuthorization 在需要 AppID 的渠道实例前,按当前版本加载全局微信授权配置。
|
||||
// 授权字段只进入内存中的渠道配置,绝不写入支付快照、普通 DTO、日志、审计或导出。
|
||||
func (l *RuntimeLoader) MerchantConfigWithAuthorization(ctx context.Context, merchant *model.PaymentMerchant) (*model.WechatConfig, error) {
|
||||
authorization, err := l.LoadAuthorization(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return MerchantConfig(merchant, authorization)
|
||||
}
|
||||
|
||||
// SelectForNewPayment atomically reads the active pool and chooses its current eligible member.
|
||||
func (l *RuntimeLoader) SelectForNewPayment(ctx context.Context, paymentMethod string, now time.Time) (*RouteSelection, error) {
|
||||
if l == nil || l.db == nil {
|
||||
return nil, errors.New(errors.CodeServiceUnavailable, "商户池路由能力未配置")
|
||||
}
|
||||
var out *RouteSelection
|
||||
err := l.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
var err error
|
||||
out, err = l.SelectForNewPaymentWithTx(ctx, tx, paymentMethod, now)
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SelectForNewPaymentWithTx chooses an eligible merchant while retaining the caller's business transaction.
|
||||
func (l *RuntimeLoader) SelectForNewPaymentWithTx(ctx context.Context, tx *gorm.DB, paymentMethod string, now time.Time) (*RouteSelection, error) {
|
||||
if l == nil || tx == nil {
|
||||
return nil, errors.New(errors.CodeServiceUnavailable, "商户池路由能力未配置")
|
||||
}
|
||||
var pool model.PaymentMerchantPool
|
||||
if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("payment_method = ? AND status = ?", paymentMethod, model.PaymentMerchantStatusEnabled).First(&pool).Error; err != nil {
|
||||
if err == gorm.ErrRecordNotFound {
|
||||
return nil, errors.New(errors.CodeNoPaymentConfig, "暂无可用商户")
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var members []model.PaymentMerchantPoolMember
|
||||
if err := tx.WithContext(ctx).Where("pool_id = ?", pool.ID).Order("sort_order ASC").Find(&members).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(members) == 0 {
|
||||
return nil, errors.New(errors.CodeNoPaymentConfig, "暂无可用商户")
|
||||
}
|
||||
ids := make([]uint, 0, len(members))
|
||||
for _, member := range members {
|
||||
ids = append(ids, member.MerchantID)
|
||||
}
|
||||
var merchants []model.PaymentMerchant
|
||||
if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).Where("id IN ? AND payment_method = ? AND status = ?", ids, pool.PaymentMethod, model.PaymentMerchantStatusEnabled).Find(&merchants).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
byID := make(map[uint]*model.PaymentMerchant, len(merchants))
|
||||
for i := range merchants {
|
||||
byID[merchants[i].ID] = &merchants[i]
|
||||
}
|
||||
ordered := make([]*model.PaymentMerchant, 0, len(members))
|
||||
for _, member := range members {
|
||||
if merchant := byID[member.MerchantID]; merchant != nil {
|
||||
ordered = append(ordered, merchant)
|
||||
}
|
||||
}
|
||||
chosen, err := chooseMerchant(ctx, tx, &pool, ordered, now)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 新支付在冻结前也按“商户 ID + 当前版本”读取缓存;事务锁保证本次
|
||||
// 选择与凭证版本属于同一提交边界,避免新建支付误用旧版本缓存。
|
||||
chosen, err = l.loadMerchant(ctx, tx, chosen.ID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &RouteSelection{Merchant: chosen, Pool: &pool}, nil
|
||||
}
|
||||
|
||||
func chooseMerchant(ctx context.Context, tx *gorm.DB, pool *model.PaymentMerchantPool, merchants []*model.PaymentMerchant, now time.Time) (*model.PaymentMerchant, error) {
|
||||
if len(merchants) == 0 {
|
||||
return nil, errors.New(errors.CodeNoPaymentConfig, "暂无可用商户")
|
||||
}
|
||||
if pool.Strategy == model.PaymentMerchantStrategyTime {
|
||||
return chooseTimedMerchant(pool, merchants, now)
|
||||
}
|
||||
if pool.Strategy != model.PaymentMerchantStrategyAmount && pool.Strategy != model.PaymentMerchantStrategyCount {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "商户池轮询策略无效")
|
||||
}
|
||||
if pool.StatisticCycle == nil {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "商户池统计周期未配置")
|
||||
}
|
||||
|
||||
query := tx.WithContext(ctx).Where("pool_id = ? AND routing_epoch = ?", pool.ID, pool.RoutingEpoch)
|
||||
if start, limited := routingWindowStart(*pool.StatisticCycle, now); limited {
|
||||
query = query.Where("paid_at >= ?", start)
|
||||
}
|
||||
var rows []model.PaymentMerchantRoutingSuccess
|
||||
if err := query.Find(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
amounts := make(map[uint]int64, len(merchants))
|
||||
counts := make(map[uint]int64, len(merchants))
|
||||
for _, row := range rows {
|
||||
amounts[row.MerchantID] += row.Amount
|
||||
counts[row.MerchantID]++
|
||||
}
|
||||
for _, merchant := range merchants {
|
||||
if pool.Strategy == model.PaymentMerchantStrategyAmount {
|
||||
if pool.ThresholdAmount == nil {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "金额轮询阈值未配置")
|
||||
}
|
||||
if amounts[merchant.ID] < *pool.ThresholdAmount {
|
||||
return merchant, nil
|
||||
}
|
||||
continue
|
||||
}
|
||||
if pool.ThresholdCount == nil {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "笔数轮询阈值未配置")
|
||||
}
|
||||
if counts[merchant.ID] < *pool.ThresholdCount {
|
||||
return merchant, nil
|
||||
}
|
||||
}
|
||||
if *pool.StatisticCycle != "round" {
|
||||
return nil, errors.New(errors.CodeNoPaymentConfig, "当前统计周期内暂无可用商户")
|
||||
}
|
||||
if err := advanceRoutingEpoch(ctx, tx, pool); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return merchants[0], nil
|
||||
}
|
||||
|
||||
func chooseTimedMerchant(pool *model.PaymentMerchantPool, merchants []*model.PaymentMerchant, now time.Time) (*model.PaymentMerchant, error) {
|
||||
if pool.TimePeriodStartedAt == nil || pool.TimePeriodValue == nil || pool.TimePeriodUnit == nil {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "时间轮询配置不完整")
|
||||
}
|
||||
unit := time.Minute
|
||||
switch *pool.TimePeriodUnit {
|
||||
case "hour":
|
||||
unit = time.Hour
|
||||
case "day":
|
||||
unit = 24 * time.Hour
|
||||
case "minute":
|
||||
default:
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "时间轮询单位无效")
|
||||
}
|
||||
period := unit * time.Duration(*pool.TimePeriodValue)
|
||||
if period <= 0 {
|
||||
return nil, errors.New(errors.CodeInvalidStatus, "时间轮询周期无效")
|
||||
}
|
||||
slot := now.Sub(*pool.TimePeriodStartedAt) / period
|
||||
if slot < 0 {
|
||||
slot = 0
|
||||
}
|
||||
return merchants[int(slot%time.Duration(len(merchants)))], nil
|
||||
}
|
||||
|
||||
func routingWindowStart(cycle string, now time.Time) (time.Time, bool) {
|
||||
local := now.In(now.Location())
|
||||
switch cycle {
|
||||
case "day":
|
||||
return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, local.Location()), true
|
||||
case "month":
|
||||
return time.Date(local.Year(), local.Month(), 1, 0, 0, 0, 0, local.Location()), true
|
||||
default:
|
||||
return time.Time{}, false
|
||||
}
|
||||
}
|
||||
|
||||
func advanceRoutingEpoch(ctx context.Context, tx *gorm.DB, pool *model.PaymentMerchantPool) error {
|
||||
next := pool.RoutingEpoch + 1
|
||||
result := tx.WithContext(ctx).Model(&model.PaymentMerchantPool{}).Where("id = ? AND routing_epoch = ?", pool.ID, pool.RoutingEpoch).Update("routing_epoch", next)
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected != 1 {
|
||||
return errors.New(errors.CodeConflict, "商户池统计世代已变化")
|
||||
}
|
||||
pool.RoutingEpoch = next
|
||||
return nil
|
||||
}
|
||||
|
||||
// FreezeRoute writes only non-sensitive route facts onto the payment.
|
||||
func FreezeRoute(payment *model.Payment, route *RouteSelection) {
|
||||
if payment == nil || route == nil || route.Merchant == nil || route.Pool == nil {
|
||||
return
|
||||
}
|
||||
payment.MerchantID = &route.Merchant.ID
|
||||
payment.MerchantPoolID = &route.Pool.ID
|
||||
payment.MerchantIdentity = route.Merchant.MerchantIdentity
|
||||
payment.MerchantNameSnapshot = route.Merchant.Name
|
||||
payment.MerchantPaymentMethodSnapshot = route.Merchant.PaymentMethod
|
||||
payment.MerchantProviderTypeSnapshot = route.Merchant.ProviderType
|
||||
payment.MerchantPoolNameSnapshot = route.Pool.Name
|
||||
payment.RoutingStrategySnapshot = route.Pool.Strategy
|
||||
epoch := route.Pool.RoutingEpoch
|
||||
payment.RoutingEpoch = &epoch
|
||||
}
|
||||
|
||||
// RecordFirstSuccess 在支付成功事务内写入支付不可变的路由事实。
|
||||
func RecordFirstSuccess(ctx context.Context, tx *gorm.DB, payment *model.Payment, paidAt time.Time) error {
|
||||
if payment == nil || payment.MerchantID == nil || payment.MerchantPoolID == nil || payment.RoutingEpoch == nil {
|
||||
return nil
|
||||
}
|
||||
fact := model.PaymentMerchantRoutingSuccess{PaymentID: payment.ID, MerchantID: *payment.MerchantID, PoolID: *payment.MerchantPoolID, RoutingEpoch: *payment.RoutingEpoch, Amount: payment.Amount, PaidAt: paidAt}
|
||||
if err := tx.WithContext(ctx).Create(&fact).Error; err != nil {
|
||||
if strings.Contains(err.Error(), "duplicate key") {
|
||||
return nil
|
||||
}
|
||||
return errors.Wrap(errors.CodeDatabaseError, err, "写入商户池成功统计失败")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user