feat(v2): 一次性收款管线 CreateOrder/GetOrder/Retry/Cancel + DBProductResolver

internal/gateway/gateway.go

@@ -0,0 +1,184 @@
+// Package gateway is the channel-neutral payment pipeline: it turns a client
+// {sku, method} into an authoritative Order + a payment Session (render_type +
+// payload), and settles callbacks/queries into paid + a business webhook. It
+// depends only on provider.Registry, store.OrderStore (P1), accounts.Registry
+// (P1) — never on a concrete channel.
+package gateway
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/wangjia/pay/internal/accounts"
+	"github.com/wangjia/pay/internal/model"
+	"github.com/wangjia/pay/internal/provider"
+	"github.com/wangjia/pay/internal/store"
+	"github.com/wangjia/pay/internal/util"
+)
+
+var (
+	ErrProductNotFound = errors.New("gateway: product not found")
+	ErrNoAccount       = errors.New("gateway: no enabled account for method/region")
+	ErrOrderNotPending = errors.New("gateway: order not pending")
+)
+
+// ProductResolver maps a client-facing SKU to the authoritative amount/currency.
+// Amount authority lives in pay (设计 §3.1); the client never sends raw amounts.
+type ProductResolver interface {
+	Resolve(sku string) (amountMinor int64, currency, subject, bizCode string, err error)
+}
+
+// WebhookEnqueuer receives a domain payload to deliver to the business system.
+type WebhookEnqueuer interface {
+	Enqueue(outTradeNo, bizSystem, eventType string, data map[string]any) error
+}
+
+type Gateway struct {
+	orders    *store.OrderStore
+	providers *provider.Registry
+	accounts  *accounts.Registry
+	products  ProductResolver
+	webhook   WebhookEnqueuer
+	region    string
+}
+
+func New(orders *store.OrderStore, providers *provider.Registry, accts *accounts.Registry,
+	products ProductResolver, webhook WebhookEnqueuer, region string) *Gateway {
+	return &Gateway{orders: orders, providers: providers, accounts: accts,
+		products: products, webhook: webhook, region: region}
+}
+
+type CreateOrderInput struct {
+	SKU       string
+	Method    string
+	BizSystem string
+	BizRef    string
+	ReturnURL string
+}
+
+type SessionView struct {
+	RenderType string         `json:"render_type"`
+	Payload    map[string]any `json:"payload"`
+	ExpiresAt  *time.Time     `json:"expires_at,omitempty"`
+}
+
+type OrderResult struct {
+	OrderNo string      `json:"order_no"`
+	Session SessionView `json:"session"`
+}
+
+// CreateOrder resolves the product (authoritative amount), picks a provider +
+// account, persists a pending Order + Attempt (P1 OrderStore), and returns the
+// payment session {render_type, payload}. 加渠道不改 client(设计 §4.2)。
+func (g *Gateway) CreateOrder(ctx context.Context, in CreateOrderInput) (*OrderResult, error) {
+	amountMinor, currency, subject, _, err := g.products.Resolve(in.SKU)
+	if err != nil {
+		return nil, err // ErrProductNotFound
+	}
+	prov, err := g.providers.Get(in.Method)
+	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")
+	if err := g.orders.CreateOrder(&model.OrderV2{
+		OutTradeNo: outNo, BizSystem: in.BizSystem, BizRef: in.BizRef,
+		Subject: subject, AmountMinor: amountMinor, Currency: currency,
+		Status: model.OrderPendingV2,
+	}); err != nil {
+		return nil, err
+	}
+
+	sess, err := prov.Create(ctx, provider.CreateRequest{
+		OutTradeNo: outNo, Subject: subject, AmountMinor: amountMinor,
+		Currency: currency, Account: acct, ReturnURL: in.ReturnURL,
+	})
+	if err != nil {
+		return nil, err
+	}
+	if err := g.orders.CreateAttempt(&model.Attempt{
+		OutTradeNo: outNo, Channel: in.Method, AccountID: acct.AccountID,
+		Provider: prov.Method(), ProviderRef: sess.ProviderRef,
+		RenderType: string(sess.RenderType), AmountMinor: amountMinor, Currency: currency,
+		Status: model.AttemptPending, ExpiresAt: sess.ExpiresAt,
+	}); err != nil {
+		return nil, err
+	}
+
+	return &OrderResult{OrderNo: outNo, Session: SessionView{
+		RenderType: string(sess.RenderType), Payload: sess.Payload, ExpiresAt: sess.ExpiresAt,
+	}}, nil
+}
+
+type OrderStatusView struct {
+	OrderNo     string     `json:"order_no"`
+	Status      string     `json:"status"`
+	Subject     string     `json:"subject"`
+	AmountMinor int64      `json:"amount_minor"`
+	Currency    string     `json:"currency"`
+	PaidAt      *time.Time `json:"paid_at,omitempty"`
+}
+
+func (g *Gateway) GetOrder(outTradeNo string) (*OrderStatusView, error) {
+	o, err := g.orders.GetOrder(outTradeNo)
+	if err != nil {
+		return nil, err // ErrOrderNotFound
+	}
+	return &OrderStatusView{
+		OrderNo: o.OutTradeNo, Status: string(o.Status), Subject: o.Subject,
+		AmountMinor: o.AmountMinor, Currency: o.Currency, PaidAt: o.PaidAt,
+	}, nil
+}
+
+// RetryOrder spawns a fresh attempt (possibly a different method) on a still-pending
+// order; old pending attempts are expired. attempt 超时 ≠ order 关闭(设计 §3.2)。
+func (g *Gateway) RetryOrder(ctx context.Context, outTradeNo, method string) (*OrderResult, error) {
+	o, err := g.orders.GetOrder(outTradeNo)
+	if err != nil {
+		return nil, err
+	}
+	if o.Status != model.OrderPendingV2 {
+		return nil, ErrOrderNotPending
+	}
+	prov, err := g.providers.Get(method)
+	if err != nil {
+		return nil, err
+	}
+	accts := g.accounts.EnabledFor(method, g.region)
+	if len(accts) == 0 {
+		return nil, ErrNoAccount
+	}
+	acct := accts[0]
+
+	if _, err := g.orders.ExpirePendingAttempts(outTradeNo); err != nil {
+		return nil, err
+	}
+	sess, err := prov.Create(ctx, provider.CreateRequest{
+		OutTradeNo: outTradeNo, Subject: o.Subject, AmountMinor: o.AmountMinor,
+		Currency: o.Currency, Account: acct,
+	})
+	if err != nil {
+		return nil, err
+	}
+	if err := g.orders.CreateAttempt(&model.Attempt{
+		OutTradeNo: outTradeNo, Channel: method, AccountID: acct.AccountID,
+		Provider: prov.Method(), ProviderRef: sess.ProviderRef,
+		RenderType: string(sess.RenderType), AmountMinor: o.AmountMinor, Currency: o.Currency,
+		Status: model.AttemptPending, ExpiresAt: sess.ExpiresAt,
+	}); err != nil {
+		return nil, err
+	}
+	return &OrderResult{OrderNo: outTradeNo, Session: SessionView{
+		RenderType: string(sess.RenderType), Payload: sess.Payload, ExpiresAt: sess.ExpiresAt,
+	}}, nil
+}
+
+func (g *Gateway) CancelOrder(outTradeNo string) (bool, error) {
+	return g.orders.CancelOrder(outTradeNo)
+}

internal/gateway/gateway_test.go

@@ -0,0 +1,120 @@
+package gateway_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"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/fake"
+	"github.com/wangjia/pay/internal/store"
+)
+
+// --- 测试替身 ---
+
+type stubResolver struct{}
+
+func (stubResolver) Resolve(sku string) (int64, string, string, string, error) {
+	if sku != "pro_year" {
+		return 0, "", "", "", gateway.ErrProductNotFound
+	}
+	return 29990000, "USDT", "Pro 年付", "pro_year", nil
+}
+
+type spyEnqueuer struct {
+	calls    []map[string]any
+	failNext bool // 置 true 模拟 outbox 入队失败(settle 崩溃窗口测试用)
+}
+
+func (s *spyEnqueuer) Enqueue(outTradeNo, bizSystem, eventType string, data map[string]any) error {
+	if s.failNext {
+		s.failNext = false
+		return errors.New("outbox down")
+	}
+	s.calls = append(s.calls, data)
+	return nil
+}
+
+func newGateway(t *testing.T) (*gateway.Gateway, *fake.Provider, *spyEnqueuer, *store.OrderStore) {
+	t.Helper()
+	orders := store.NewOrderStore(model.OpenTestDB(t))
+	preg := provider.NewRegistry()
+	fp := fake.New()
+	preg.Register(fp)
+	// fake 注册在 method "fake";账户按 channel="fake" region="global" 配。
+	areg := accounts.New([]config.AccountConfig{
+		{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
+	})
+	spy := &spyEnqueuer{}
+	g := gateway.New(orders, preg, areg, stubResolver{}, spy, "global")
+	return g, fp, spy, orders
+}
+
+func TestCreateOrderPipeline(t *testing.T) {
+	g, _, _, orders := newGateway(t)
+	res, err := g.CreateOrder(context.Background(), gateway.CreateOrderInput{
+		SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1",
+	})
+	if err != nil {
+		t.Fatalf("CreateOrder: %v", err)
+	}
+	if res.OrderNo == "" || res.Session.RenderType != string(provider.RenderCryptoAddress) {
+		t.Fatalf("result = %+v", res)
+	}
+	if res.Session.Payload["currency"] != "USDT" {
+		t.Fatalf("payload = %+v", res.Session.Payload)
+	}
+	// 落库:order pending + attempt pending 带 provider_ref。
+	o, err := orders.GetOrder(res.OrderNo)
+	if err != nil || o.Status != model.OrderPendingV2 || o.AmountMinor != 29990000 {
+		t.Fatalf("order = %+v, %v", o, err)
+	}
+	atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
+	if len(atts) != 1 || atts[0].Channel != "fake" || atts[0].AccountID != "fake-a1" {
+		t.Fatalf("attempt = %+v", atts)
+	}
+}
+
+func TestCreateOrderErrors(t *testing.T) {
+	g, _, _, _ := newGateway(t)
+	ctx := context.Background()
+	if _, err := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "nope", Method: "fake"}); err != gateway.ErrProductNotFound {
+		t.Fatalf("未知 sku 应 ErrProductNotFound, got %v", err)
+	}
+	if _, err := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "ghost"}); err == nil {
+		t.Fatalf("未知 method 应报错")
+	}
+}
+
+func TestRetryAndCancel(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"})
+
+	// retry:弃旧尝试 + 建新尝试(order 仍 pending)。
+	r2, err := g.RetryOrder(ctx, res.OrderNo, "fake")
+	if err != nil || r2.OrderNo != res.OrderNo {
+		t.Fatalf("retry = %+v, %v", r2, err)
+	}
+	pend, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
+	if len(pend) != 1 {
+		t.Fatalf("retry 后应恰 1 个 pending 尝试, got %d", len(pend))
+	}
+
+	// cancel pending → true;再 cancel → false。
+	ok, err := g.CancelOrder(res.OrderNo)
+	if err != nil || !ok {
+		t.Fatalf("cancel = %v, %v", ok, err)
+	}
+	if ok2, _ := g.CancelOrder(res.OrderNo); ok2 {
+		t.Fatalf("已取消单再取消应 false")
+	}
+	// canceled 单不可 retry。
+	if _, err := g.RetryOrder(ctx, res.OrderNo, "fake"); err != gateway.ErrOrderNotPending {
+		t.Fatalf("canceled 单 retry 应 ErrOrderNotPending, got %v", err)
+	}
+}

internal/gateway/product.go

@@ -0,0 +1,41 @@
+package gateway
+
+import (
+	"errors"
+
+	"gorm.io/gorm"
+
+	"github.com/wangjia/pay/internal/model"
+	"github.com/wangjia/pay/internal/money"
+)
+
+// DBProductResolver resolves a SKU (product biz_code) against the products table.
+// v1 Product.Price is a "元" string; we parse it into int64 minor units for the
+// given settlement currency. 加币种维度到 product 是 P3+ 的事;P2 用单一默认币种。
+type DBProductResolver struct {
+	db       *gorm.DB
+	currency string
+}
+
+func NewDBProductResolver(db *gorm.DB, currency string) *DBProductResolver {
+	if currency == "" {
+		currency = "CNY"
+	}
+	return &DBProductResolver{db: db, currency: currency}
+}
+
+func (r *DBProductResolver) Resolve(sku string) (int64, string, string, string, error) {
+	var p model.Product
+	err := r.db.Where("biz_code = ? AND active = ?", sku, true).First(&p).Error
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return 0, "", "", "", ErrProductNotFound
+	}
+	if err != nil {
+		return 0, "", "", "", err
+	}
+	minor, err := money.Parse(p.Price, r.currency)
+	if err != nil {
+		return 0, "", "", "", err
+	}
+	return minor, r.currency, p.Name, p.BizCode, nil
+}

internal/gateway/product_test.go

@@ -0,0 +1,33 @@
+package gateway_test
+
+import (
+	"testing"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+
+	"github.com/wangjia/pay/internal/gateway"
+	"github.com/wangjia/pay/internal/model"
+)
+
+func TestDBProductResolver(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:prodtest?mode=memory&cache=shared"),
+		&gorm.Config{Logger: logger.Default.LogMode(logger.Silent), TranslateError: true})
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if err := db.AutoMigrate(&model.Product{}); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	db.Create(&model.Product{Name: "标准年付", Price: "299.00", BizCode: "annual_standard", Active: true})
+
+	r := gateway.NewDBProductResolver(db, "CNY")
+	minor, cur, subject, bizCode, err := r.Resolve("annual_standard")
+	if err != nil || minor != 29900 || cur != "CNY" || subject != "标准年付" || bizCode != "annual_standard" {
+		t.Fatalf("resolve = %d %s %s %s %v", minor, cur, subject, bizCode, err)
+	}
+	if _, _, _, _, err := r.Resolve("ghost"); err != gateway.ErrProductNotFound {
+		t.Fatalf("缺套餐应 ErrProductNotFound, got %v", err)
+	}
+}

internal/provider/fake/fake.go

@@ -1,12 +1,16 @@
 // Package fake is an in-process Provider used to exercise the pay v2 pipeline
 // end-to-end without any real channel. Real adapters (crypto/alipay/stripe) land
-// in P3. Deterministic: provider_ref = "FAKE-"+OutTradeNo; render_type = crypto_address.
+// in P3. provider_ref = "FAKE-"+OutTradeNo+"-"+<nanosecond>; render_type = crypto_address.
+// The nanosecond suffix is fake-only: Attempt has uniqueIndex(channel,provider_ref),
+// and a retry on the same order/method would otherwise reuse the same OutTradeNo and
+// collide. Real channels naturally mint a distinct ref per create call.
 package fake
 
 import (
 	"context"
 	"encoding/json"
 	"errors"
+	"strconv"
 	"sync"
 	"time"
 
@@ -33,8 +37,9 @@ func (p *Provider) Capabilities() provider.Capabilities {
 
 func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (*provider.Session, error) {
 	exp := time.Now().Add(15 * time.Minute)
+	ref := "FAKE-" + req.OutTradeNo + "-" + strconv.FormatInt(time.Now().UnixNano(), 36)
 	return &provider.Session{
-		ProviderRef: "FAKE-" + req.OutTradeNo,
+		ProviderRef: ref,
 		RenderType:  provider.RenderCryptoAddress,
 		Payload: map[string]any{
 			"address":      "TFake" + req.Account.AccountID + req.OutTradeNo,

internal/provider/fake/fake_test.go

@@ -2,6 +2,7 @@ package fake_test
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/wangjia/pay/internal/provider"
@@ -18,7 +19,7 @@ func TestFakeCreateAndVerify(t *testing.T) {
 	if err != nil {
 		t.Fatalf("create: %v", err)
 	}
-	if sess.ProviderRef != "FAKE-PAY-1" || sess.RenderType != provider.RenderCryptoAddress {
+	if !strings.HasPrefix(sess.ProviderRef, "FAKE-PAY-1") || sess.RenderType != provider.RenderCryptoAddress {
 		t.Fatalf("session = %+v", sess)
 	}