wangjia/pay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v2): gateway 经 Picker 路由选账户 + retry 换号(ExcludeAccounts);替换首个 enabled

Browse Source
This commit is contained in:
wangjia
2026-07-10 14:02:33 +08:00
parent 381288e707
commit c06f6d3a3c
7 changed files with 109 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/gateway/gateway.go
+25 -13
View File
@@ -37,15 +37,15 @@ type WebhookEnqueuer interface {
type Gateway struct {
orders *store.OrderStore
providers *provider.Registry
accounts *accounts.Registry
picker accounts.Picker
products ProductResolver
webhook WebhookEnqueuer
region string
}
func New(orders *store.OrderStore, providers *provider.Registry, accts *accounts.Registry,
func New(orders *store.OrderStore, providers *provider.Registry, picker accounts.Picker,
products ProductResolver, webhook WebhookEnqueuer, region string) *Gateway {
return &Gateway{orders: orders, providers: providers, accounts: accts,
return &Gateway{orders: orders, providers: providers, picker: picker,
products: products, webhook: webhook, region: region}
}
@@ -80,13 +80,17 @@ func (g *Gateway) CreateOrder(ctx context.Context, in CreateOrderInput) (*OrderR
if err != nil {
return nil, err // ErrUnknownMethod
}
accts := g.accounts.EnabledFor(in.Method, g.region)
if len(accts) == 0 {
return nil, ErrNoAccount
}
acct := accts[0] // 路由策略(round_robin/weighted/…)在 P5;P2 取首个 enabled。
outNo := util.NewOutTradeNo("pay")
acct, err := g.picker.Pick(in.Method, g.region, accounts.PickHint{
OutTradeNo: outNo, AmountMinor: amountMinor,
})
if err != nil {
if errors.Is(err, accounts.ErrNoAccount) {
return nil, ErrNoAccount
}
return nil, err
}
if err := g.orders.CreateOrder(&model.OrderV2{
OutTradeNo: outNo, BizSystem: in.BizSystem, BizRef: in.BizRef,
Subject: subject, AmountMinor: amountMinor, Currency: currency,
@@ -150,11 +154,19 @@ func (g *Gateway) RetryOrder(ctx context.Context, outTradeNo, method string) (*O
if err != nil {
return nil, err
}
accts := g.accounts.EnabledFor(method, g.region)
if len(accts) == 0 {
return nil, ErrNoAccount
tried, err := g.orders.AttemptAccountIDs(outTradeNo, method)
if err != nil {
return nil, err
}
acct, err := g.picker.Pick(method, g.region, accounts.PickHint{
OutTradeNo: outTradeNo, AmountMinor: o.AmountMinor, ExcludeAccounts: tried,
})
if err != nil {
if errors.Is(err, accounts.ErrNoAccount) {
return nil, ErrNoAccount
}
return nil, err
}
acct := accts[0]
if _, err := g.orders.ExpirePendingAttempts(outTradeNo); err != nil {
return nil, err
internal/gateway/gateway_test.go
+18 -1
View File
@@ -48,9 +48,11 @@ func newGateway(t *testing.T) (*gateway.Gateway, *fake.Provider, *spyEnqueuer, *
// fake 注册在 method "fake";账户按 channel="fake" region="global" 配。
areg := accounts.New([]config.AccountConfig{
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
{AccountID: "fake-a2", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
})
picker := accounts.NewRouter(areg, nil, nil) // 默认 round_robin
spy := &spyEnqueuer{}
g := gateway.New(orders, preg, areg, stubResolver{}, spy, "global")
g := gateway.New(orders, preg, picker, stubResolver{}, spy, "global")
return g, fp, spy, orders
}
@@ -118,3 +120,18 @@ func TestRetryAndCancel(t *testing.T) {
t.Fatalf("canceled 单 retry 应 ErrOrderNotPending, got %v", err)
}
}
func TestRetrySwitchesAccount(t *testing.T) {
g, _, _, orders := newGateway(t)
ctx := context.Background()
res, _ := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
// 首单落 fake-a1(round_robin 计数 0);retry 排除 a1 → 必落 fake-a2。
if _, err := g.RetryOrder(ctx, res.OrderNo, "fake"); err != nil {
t.Fatalf("retry: %v", err)
}
pend, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
if len(pend) != 1 || pend[0].AccountID != "fake-a2" {
t.Fatalf("retry 应换到 fake-a2, got %+v", pend)
}
}
internal/gateway/settle_test.go
+2 -1
View File
@@ -120,8 +120,9 @@ func TestSettleTransientReadErrorIsFailed(t *testing.T) {
areg := accounts.New([]config.AccountConfig{
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
})
picker := accounts.NewRouter(areg, nil, nil)
spy := &spyEnqueuer{}
g := gateway.New(orders, preg, areg, stubResolver{}, spy, "global")
g := gateway.New(orders, preg, picker, stubResolver{}, spy, "global")
ctx := context.Background()
g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
internal/handler/gateway_test.go
+2 -1
View File
@@ -44,7 +44,8 @@ func buildEngineWithStore(t *testing.T) (*gin.Engine, *store.OrderStore) {
areg := accounts.New([]config.AccountConfig{
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
})
g := gateway.New(orders, preg, areg, oneResolver{}, nopEnqueuer{}, "global")
picker := accounts.NewRouter(areg, nil, nil)
g := gateway.New(orders, preg, picker, oneResolver{}, nopEnqueuer{}, "global")
r := gin.New()
router.SetupV2(r, g)
return r, orders
internal/store/attempt_accounts.go
+19
View File
@@ -0,0 +1,19 @@
package store
import (
"fmt"
"github.com/wangjia/pay/internal/model"
)
// AttemptAccountIDs returns the distinct account_ids already tried for an order on a
// channel (used by retry to switch to a not-yet-tried account via PickHint.ExcludeAccounts).
func (s *OrderStore) AttemptAccountIDs(outTradeNo, channel string) ([]string, error) {
var ids []string
if err := s.db.Model(&model.Attempt{}).
Where("out_trade_no = ? AND channel = ? AND account_id <> ''", outTradeNo, channel).
Distinct().Pluck("account_id", &ids).Error; err != nil {
return nil, fmt.Errorf("store.AttemptAccountIDs: %w", err)
}
return ids, nil
}
internal/store/attempt_accounts_test.go
+39
View File
@@ -0,0 +1,39 @@
package store_test
import (
"testing"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/store"
)
func TestAttemptAccountIDs(t *testing.T) {
s := store.NewOrderStore(model.OpenTestDB(t))
if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "PAY-AC1", Subject: "x", AmountMinor: 100, Currency: "USDT", Status: model.OrderPendingV2}); err != nil {
t.Fatalf("order: %v", err)
}
for _, tc := range []struct{ ref, acct string }{{"R1", "a1"}, {"R2", "a2"}, {"R3", "a1"}} {
if err := s.CreateAttempt(&model.Attempt{
OutTradeNo: "PAY-AC1", Channel: "fake", AccountID: tc.acct, Provider: "fake",
ProviderRef: tc.ref, RenderType: "crypto_address", AmountMinor: 100, Currency: "USDT",
Status: model.AttemptPending,
}); err != nil {
t.Fatalf("attempt %s: %v", tc.ref, err)
}
}
ids, err := s.AttemptAccountIDs("PAY-AC1", "fake")
if err != nil {
t.Fatalf("AttemptAccountIDs: %v", err)
}
set := map[string]bool{}
for _, id := range ids {
set[id] = true
}
if len(set) != 2 || !set["a1"] || !set["a2"] {
t.Fatalf("应含 distinct a1,a2, got %v", ids)
}
// 其它渠道 / 其它单不串。
if got, _ := s.AttemptAccountIDs("PAY-AC1", "alipay"); len(got) != 0 {
t.Fatalf("别的渠道应空, got %v", got)
}
}
main.go
+4 -1
View File
@@ -52,7 +52,10 @@ func main() {
notifier.Start(60 * time.Second)
productResolver := gateway.NewDBProductResolver(db, "CNY") // 币种按部署区配(cn=CNY / global=USDT)
acctReg := accounts.New(config.C.Accounts)
gw := gateway.New(orderStore, pReg, acctReg, productResolver, notifier, "cn")
// P5 多账户路由:按 config.routing.<channel> 选策略(缺省 round_robin)。
// limit_aware 用量数据源 P6 对账就绪前用空源(NopUsage,退化为 round_robin)。
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, accounts.NopUsage{})
gw := gateway.New(orderStore, pReg, acctPicker, productResolver, notifier, "cn")
router.SetupV2(r, gw)
if config.C.QuerySync.Enabled {