feat(v2): 配置驱动注册三渠道 adapter + crypto 真 adapter 全链 e2e(下单→查单→入账→webhook)

This commit is contained in:
wangjia
2026-07-10 15:17:43 +08:00
parent 79b9111a99
commit c933657350
3 changed files with 168 additions and 4 deletions
+93
View File
@@ -0,0 +1,93 @@
package gateway_test
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"testing"
"time"
"github.com/wangjia/pay/config"
"github.com/wangjia/pay/internal/accounts"
"github.com/wangjia/pay/internal/gateway"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/provider"
"github.com/wangjia/pay/internal/provider/crypto"
"github.com/wangjia/pay/internal/store"
)
// TestE2ECryptoQuerySettles 验证真 crypto adapter(非 fake)跑通 P2 全链:
// 下单(单地址+唯一金额)→ 查单兜底(假 TronGrid only_confirmed 精确匹配)→ 入账 → webhook。
// 不经 provider.BuildRegistry(那是 main 装配期用的),直接用 crypto.New 组 gateway——
// 本用例验证的是「真 adapter 跑通管线」,BuildRegistry 由 registry_build_test.go 覆盖装配决策。
func TestE2ECryptoQuerySettles(t *testing.T) {
const addr = "TWe2eADDRESS00000000000000000000000"
t.Setenv("E2E_ADDRESS", addr)
t.Setenv("E2E_TRONGRID_KEY", "k")
// 假 TronGrid:已确认、金额精确匹配、块时晚于建单的转账(期望金额在下单后从 payload 取)。
var expected int64
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
resp := map[string]any{"success": true, "data": []map[string]any{{
"transaction_id": "tx-e2e",
"to": addr,
"type": "Transfer",
"value": strconv.FormatInt(expected, 10),
"block_timestamp": time.Now().Add(5 * time.Second).UnixMilli(), // 晚于建单(查单在下单后发生)
}}}
_ = json.NewEncoder(w).Encode(resp)
}))
defer ts.Close()
db := model.OpenTestDB(t)
orders := store.NewOrderStore(db)
acctReg := accounts.New([]config.AccountConfig{
{AccountID: "e2e-1", Channel: "crypto", Enabled: true, Region: "global", CredentialEnvPrefix: "e2e"},
})
preg := provider.NewRegistry()
preg.Register(crypto.New(acctReg, crypto.WithBaseURL(ts.URL), crypto.WithHTTPClient(ts.Client())))
// gateway.New 要求 accounts.Picker(非裸 *accounts.Registry);单账户场景用默认
// round_robin 的 Router 包一层即可(同 gateway_test.go newGateway 的装法)。
picker := accounts.NewRouter(acctReg, nil, nil)
spy := &spyEnqueuer{}
g := gateway.New(orders, preg, picker, cryptoResolver{}, spy, "global")
// 下单 → 从 session payload 拿到期望链上金额(base+唯一尾数),喂给假 TronGrid。
res, err := g.CreateOrder(context.Background(), gateway.CreateOrderInput{
SKU: "pro_year", Method: "crypto", BizSystem: "pangolin", BizRef: "u-e2e",
})
if err != nil {
t.Fatalf("create: %v", err)
}
atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
if len(atts) != 1 {
t.Fatalf("want 1 pending attempt, got %d", len(atts))
}
expected = res.Session.Payload["amount_minor"].(int64) // 唯一金额(随机尾数)只有 payload/provider_ref 知道
// 查单兜底 → 命中 → 入账 → webhook
n, err := g.SyncPendingAttempts(context.Background(), 10)
if err != nil || n != 1 {
t.Fatalf("sync settled=%d err=%v want 1", n, err)
}
o, _ := orders.GetOrder(res.OrderNo)
if o.Status != model.OrderPaidV2 {
t.Fatalf("order status = %s want paid", o.Status)
}
if len(spy.calls) != 1 || spy.calls[0]["product_biz_code"] != "pro_year" {
t.Fatalf("webhook = %+v", spy.calls)
}
}
type cryptoResolver struct{}
func (cryptoResolver) Resolve(sku, currency string) (int64, string, string, error) {
if sku == "pro_year" && currency == "USDT" {
return 29990000, "Pro 年付", "pro_year", nil
}
return 0, "", "", gateway.ErrProductNotFound
}
+72
View File
@@ -0,0 +1,72 @@
// Package providerbuild is the *only* place that imports every concrete
// provider adapter package alongside internal/provider — it exists precisely
// so gateway/provider's core stays adapter-neutral (see internal/provider's
// package doc). Only main calls into this package.
//
// It cannot live inside internal/provider itself: each adapter package
// (alipay/crypto/stripe) imports internal/provider for the shared
// CreateRequest/Session/Capabilities types, so internal/provider importing
// them back would be a Go import cycle. This package sits one level up the
// dependency graph instead: providerbuild -> {provider, alipay, crypto, stripe},
// nothing imports providerbuild back.
package providerbuild
import (
"log"
sw "github.com/smartwalle/alipay/v3"
stripeclient "github.com/stripe/stripe-go/v79/client"
"github.com/wangjia/pay/internal/accounts"
"github.com/wangjia/pay/internal/provider"
"github.com/wangjia/pay/internal/provider/alipay"
"github.com/wangjia/pay/internal/provider/crypto"
"github.com/wangjia/pay/internal/provider/stripe"
)
// BuildRegistry 据 enabled 账户装配注册表:有 enabled 账户且凭证齐备的渠道才 Register。
// 缺凭证只跳过该渠道(log,不 fatal)——允许只上线部分渠道;fake adapter 从不在此注册。
func BuildRegistry(accts *accounts.Registry) *provider.Registry {
reg := provider.NewRegistry()
// crypto:整渠道一个 adapter,多地址=多账户(P5 地址池路由前取首个 enabled)。
if len(accts.EnabledFor("crypto", "")) > 0 {
reg.Register(crypto.New(accts))
log.Println("[providers] crypto 已注册")
}
// alipay:首个 enabled 账户的 env 凭证 → *alipay.Client。
if as := accts.EnabledFor("alipay", ""); len(as) > 0 {
a := as[0]
appID := accts.Credential(a.AccountID, "APP_ID")
appPriv := accts.Credential(a.AccountID, "APP_PRIVATE_KEY")
aliPub := accts.Credential(a.AccountID, "ALIPAY_PUBLIC_KEY")
prod := accts.Credential(a.AccountID, "PRODUCTION") == "1"
if appID == "" || appPriv == "" || aliPub == "" {
log.Printf("[providers] alipay 账户 %s 凭证不全,跳过", a.AccountID)
} else if c, err := sw.New(appID, appPriv, prod); err != nil {
log.Printf("[providers] alipay client 构建失败: %v", err)
} else if err := c.LoadAliPayPublicKey(aliPub); err != nil {
log.Printf("[providers] alipay 加载公钥失败: %v", err)
} else {
reg.Register(alipay.New(c))
log.Println("[providers] alipay 已注册")
}
}
// stripe:首个 enabled 账户的 env 凭证 → *client.API。
if ss := accts.EnabledFor("stripe", ""); len(ss) > 0 {
s := ss[0]
key := accts.Credential(s.AccountID, "SECRET_KEY")
wh := accts.Credential(s.AccountID, "WEBHOOK_SECRET")
if key == "" || wh == "" {
log.Printf("[providers] stripe 账户 %s 凭证不全,跳过", s.AccountID)
} else {
sc := stripeclient.New(key, nil)
reg.Register(stripe.New(sc, wh))
log.Println("[providers] stripe 已注册")
}
}
return reg
}
+3 -4
View File
@@ -15,7 +15,7 @@ import (
"github.com/wangjia/pay/internal/channel"
"github.com/wangjia/pay/internal/gateway"
"github.com/wangjia/pay/internal/model"
"github.com/wangjia/pay/internal/provider"
"github.com/wangjia/pay/internal/providerbuild"
"github.com/wangjia/pay/internal/router"
"github.com/wangjia/pay/internal/store"
"github.com/wangjia/pay/internal/webhook"
@@ -37,9 +37,7 @@ func main() {
orderSvc := router.Setup(r, db, reg)
// v2 统一网关装配(P2):provider 注册表 + gateway + webhook notifier。
pReg := provider.NewRegistry()
// P3 起在此 Register 真实渠道:crypto / alipay / stripe …(fake 仅测试用,不注册进生产)。
// v2 统一网关装配(P2 骨架,P3 配置驱动注册接线):provider 注册表 + gateway + webhook notifier。
orderStore := store.NewOrderStore(db)
webhookStore := store.NewWebhookStore(db)
notifier := webhook.NewNotifier(webhookStore, config.C.BizByName, func(no string) (bool, error) {
@@ -52,6 +50,7 @@ func main() {
notifier.Start(60 * time.Second)
productResolver := gateway.NewDBProductResolver(db) // 币种由下单渠道结算能力驱动(设计 §4.1)
acctReg := accounts.New(config.C.Accounts)
pReg := providerbuild.BuildRegistry(acctReg) // 配置驱动:有 enabled 账户才注册对应渠道(P3)
// P5 多账户路由:按 config.routing.<channel> 选策略(缺省 round_robin)。
// limit_aware 用量数据源 P6 对账就绪前用空源(NopUsage,退化为 round_robin)。
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, accounts.NopUsage{})