junhong_cmp_fiber/internal/application/merchantpayment/routing.go

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(&current, 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(&current)); err == nil {
			_ = l.redis.Set(ctx, key, text, time.Hour).Err()
		}
	}
	return &current, 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(&current).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(&current)); err == nil {
			_ = l.redis.Set(ctx, key, text, time.Hour).Err()
		}
	}
	return &current, 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
}