From c06f6d3a3c6d33767810e60ab8926c5f7a72a8b1 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 14:02:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20gateway=20=E7=BB=8F=20Picker=20?= =?UTF-8?q?=E8=B7=AF=E7=94=B1=E9=80=89=E8=B4=A6=E6=88=B7=20+=20retry=20?= =?UTF-8?q?=E6=8D=A2=E5=8F=B7(ExcludeAccounts);=E6=9B=BF=E6=8D=A2=E9=A6=96?= =?UTF-8?q?=E4=B8=AA=20enabled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/gateway.go | 38 +++++++++++++++--------- internal/gateway/gateway_test.go | 19 +++++++++++- internal/gateway/settle_test.go | 3 +- internal/handler/gateway_test.go | 3 +- internal/store/attempt_accounts.go | 19 ++++++++++++ internal/store/attempt_accounts_test.go | 39 +++++++++++++++++++++++++ main.go | 5 +++- 7 files changed, 109 insertions(+), 17 deletions(-) create mode 100644 internal/store/attempt_accounts.go create mode 100644 internal/store/attempt_accounts_test.go diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 4475e0e..47faaf7 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -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 diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index 8049780..dad3171 100644 --- a/internal/gateway/gateway_test.go +++ b/internal/gateway/gateway_test.go @@ -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) + } +} diff --git a/internal/gateway/settle_test.go b/internal/gateway/settle_test.go index 596a9c7..f040bb9 100644 --- a/internal/gateway/settle_test.go +++ b/internal/gateway/settle_test.go @@ -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"}) diff --git a/internal/handler/gateway_test.go b/internal/handler/gateway_test.go index bced738..55aaf36 100644 --- a/internal/handler/gateway_test.go +++ b/internal/handler/gateway_test.go @@ -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 diff --git a/internal/store/attempt_accounts.go b/internal/store/attempt_accounts.go new file mode 100644 index 0000000..a12b80d --- /dev/null +++ b/internal/store/attempt_accounts.go @@ -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 +} diff --git a/internal/store/attempt_accounts_test.go b/internal/store/attempt_accounts_test.go new file mode 100644 index 0000000..af6d10b --- /dev/null +++ b/internal/store/attempt_accounts_test.go @@ -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) + } +} diff --git a/main.go b/main.go index f1bccc4..9556690 100644 --- a/main.go +++ b/main.go @@ -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. 选策略(缺省 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 {