pay/docs/superpowers/plans/2026-07-10-pay-v2-p2-pipeline.md

pay v2 · P2 Provider 抽象 + 一次性收款管线 + webhook v2 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

设计文档(全景蓝图): docs/pay-v2-unified-gateway-design.html(重点 §4 Provider 接口 / render_type、§5 webhook v2 event_type、§3 Order/Attempt/Account)。 前置计划: docs/superpowers/plans/2026-07-10-pay-v2-p1-core-model.md(P1 已落地:internal/model/v2.go、internal/store/order.go、internal/money、internal/accounts)。本计划是 P2,直接复用 P1 接口,不重造。

Goal: 在 P1 数据地基上,建立 Provider 渠道抽象 + 注册表、一个 fake provider(测试用)、一次性收款管线(下单→选 provider→建单+尝试→返回 render_type/payload;查单/重试/取消)、统一开通/入账管线(webhook 与 query 都归一化成 PaidEvent → 定位订单 → 幂等 → 金额/币种核对 → 复用 MarkAttemptPaid → 触发对业务方 webhook)、以及 webhook v2(pay→业务方,payload 带 event_type,HMAC 双向签名,重试兜底 outbox)。真实渠道 adapter 在 P3。

Architecture: 干净 v2:Provider 接口把渠道差异封死在 create/verify_callback/query(设计 §4.1),客户端只认 6 个 render_type(设计 §4.2)。管线层 internal/gateway 面向 provider.Registry + store.OrderStore(P1)+ accounts.Registry(P1),不 import 具体渠道。入账用 P1 的 MarkAttemptPaid(条件 UPDATE + RowsAffected 幂等)。对业务方 webhook 走 outbox 表 + 后台重试(unique(out_trade_no,event_type) 幂等入队),沿用 v1 notifyBizByOutTradeNo 的 HMAC 签名头范式(util.HMACSign)。

Tech Stack: Go 1.26.1 · github.com/wangjia/pay · Gin · GORM v1.31 · glebarez/sqlite · internal/util(HMACSign/HMACVerify/RespondError/RespondSuccess/NewOutTradeNo)。

Global Constraints

• 复用 P1,不重造:金额 AmountMinor int64 + Currency(internal/money);订单/尝试模型 model.OrderV2/model.Attempt + V2 后缀状态常量(model.OrderPendingV2/OrderPaidV2/OrderCanceledV2、model.AttemptPending/AttemptPaid/AttemptExpired);数据访问 store.OrderStore(CreateOrder/CreateAttempt/MarkAttemptPaid/CancelOrder/ListOrders);账户 accounts.Registry.EnabledFor + config.AccountConfig。
• 金额权威在 pay:客户端传 sku(product biz_code),金额由 ProductResolver 服务端解析,绝不接受裸传金额(设计 §3.1)。金额一律 int64 最小单位 + 币种码,禁 float。
• 幂等:入账用 P1 MarkAttemptPaid 的条件 UPDATE + RowsAffected(order=pending 原子守卫,"一单 N 渠道只一个成功")。webhook 入队用 INSERT ... ON CONFLICT DO NOTHING(unique(out_trade_no,event_type))。
• 金额/币种核对:入账前校验 PaidEvent.PaidCurrency == attempt.Currency 且 PaidAmountMinor >= attempt.AmountMinor(允许 crypto 多付,拒绝少付/错币种)。
• Provider 中性:internal/gateway 与 internal/provider 不 import 任何具体渠道;真实渠道在 P3 各自实现 provider.Provider 并 Register。
• 签名沿用双向 HMAC:util.HMACSign(secret, system, ts, nonce, rawBody),±5min 窗口 + nonce(设计 §5),与 v1 notifyBizByOutTradeNo / verifyBizSign 一致。
• 每步 go build ./... 通过;测试 go test ./...(全部 :memory: sqlite + httptest,免 docker)。GORM AutoMigrate,模型内嵌 model.Base。
• 每任务严格 bite-sized TDD:写失败测试 → 跑失败 → 实现 → 跑通过 → commit。禁占位。

---

Task 1: Provider 接口 + 核心类型 + 注册表

Files:

• Create: internal/provider/provider.go(接口 + Session/PaidEvent/Capabilities/CreateRequest/CallbackInput + render_type/status 枚举 + Registry)
• Test: internal/provider/provider_test.go

Interfaces:

• Consumes: config.AccountConfig(P1)。

• Produces:

  • type RenderType string + 6 常量 RenderRedirect/RenderQR/RenderCryptoAddress/RenderNativePay/RenderSDKHandoff/RenderIAPReceipt(设计 §4.2)。
  • type PaidStatus string + PaidPending/PaidSucceeded/PaidFailed。
  • type Capabilities struct{ RenderTypes []RenderType; SupportsRefund/SupportsRecurring bool; RecurringKind string; SettleCurrencies/Regions []string }。
  • type CreateRequest struct{ OutTradeNo, Subject string; AmountMinor int64; Currency string; Account config.AccountConfig; ReturnURL string; Metadata map[string]string }。
  • type Session struct{ ProviderRef string; RenderType RenderType; Payload map[string]any; ExpiresAt *time.Time }。
  • type CallbackInput struct{ Raw []byte; Headers, Query map[string]string }。
  • type PaidEvent struct{ ProviderRef string; Status PaidStatus; PaidAmountMinor int64; PaidCurrency string; Raw string; PaidAt *time.Time }(PaidAt=渠道报的支付时间,对账要用;渠道不报则 nil,settle 落 time.Now())。
  • type QueryRequest struct{ ProviderRef, OutTradeNo, AccountID string; AmountMinor int64; Currency string; CreatedAt time.Time; ExpiresAt *time.Time }(查单带尝试完整上下文快照,不是裸 ref——crypto 的"查单"是按地址+金额+时间窗扫链核对)。
  • type Provider interface{ Method() string; Capabilities() Capabilities; Create(ctx,CreateRequest)(*Session,error); VerifyCallback(ctx,CallbackInput)(*PaidEvent,error); Query(ctx,QueryRequest)(*PaidEvent,error) }。
  • 可选扩展接口 RefundingProvider(P4)、RecurringProvider(P8),供类型断言探测。
  • type Registry struct{...} · NewRegistry() · Register(Provider) · Get(method)(Provider,error) · Methods()[]string;ErrUnknownMethod。

• Step 1: 写失败测试

internal/provider/provider_test.go:

package provider_test

import (
	"context"
	"errors"
	"testing"

	"github.com/wangjia/pay/internal/provider"
)

// stubProvider 最小实现,驱动 Provider 接口 + Registry 成型。
type stubProvider struct{ method string }

func (s stubProvider) Method() string                        { return s.method }
func (s stubProvider) Capabilities() provider.Capabilities   { return provider.Capabilities{RenderTypes: []provider.RenderType{provider.RenderQR}} }
func (s stubProvider) Create(context.Context, provider.CreateRequest) (*provider.Session, error) {
	return &provider.Session{ProviderRef: "R-1", RenderType: provider.RenderQR}, nil
}
func (s stubProvider) VerifyCallback(context.Context, provider.CallbackInput) (*provider.PaidEvent, error) {
	return &provider.PaidEvent{ProviderRef: "R-1", Status: provider.PaidSucceeded}, nil
}
func (s stubProvider) Query(context.Context, provider.QueryRequest) (*provider.PaidEvent, error) {
	return &provider.PaidEvent{ProviderRef: "R-1", Status: provider.PaidPending}, nil
}

func TestRegistryRegisterGet(t *testing.T) {
	r := provider.NewRegistry()
	r.Register(stubProvider{method: "alipay"})
	r.Register(stubProvider{method: "crypto"})

	p, err := r.Get("crypto")
	if err != nil || p.Method() != "crypto" {
		t.Fatalf("Get crypto = %v, %v", p, err)
	}
	if _, err := r.Get("nope"); !errors.Is(err, provider.ErrUnknownMethod) {
		t.Fatalf("未知 method 应返回 ErrUnknownMethod, got %v", err)
	}
	if got := r.Methods(); len(got) != 2 || got[0] != "alipay" || got[1] != "crypto" {
		t.Fatalf("Methods 应按字典序返回 [alipay crypto], got %v", got)
	}
}

func TestSessionAndCaps(t *testing.T) {
	var _ provider.Provider = stubProvider{} // 编译期断言 stub 满足接口
	caps := stubProvider{}.Capabilities()
	if len(caps.RenderTypes) != 1 || caps.RenderTypes[0] != provider.RenderQR {
		t.Fatalf("caps = %+v", caps)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/provider/ -v Expected: 编译失败 —— provider 包不存在。

• Step 3: 写实现

internal/provider/provider.go:

// Package provider abstracts a payment channel behind a uniform interface:
// create → {render_type, payload}, verify_callback / query → normalized PaidEvent.
// Channel-specific quirks (RSA/cert/on-chain confirmations/receipt checks) are
// sealed inside each Provider's VerifyCallback; the pipeline stays channel-neutral.
package provider

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"time"

	"github.com/wangjia/pay/config"
)

// RenderType — 客户端只认的 6 种付款意图形态(设计 §4.2),不含任何 UI。
type RenderType string

const (
	RenderRedirect      RenderType = "redirect"
	RenderQR            RenderType = "qr"
	RenderCryptoAddress RenderType = "crypto_address"
	RenderNativePay     RenderType = "native_pay"
	RenderSDKHandoff    RenderType = "sdk_handoff"
	RenderIAPReceipt    RenderType = "iap_receipt"
)

// PaidStatus — verify_callback / query 归一化后的支付状态。
type PaidStatus string

const (
	PaidPending   PaidStatus = "pending"
	PaidSucceeded PaidStatus = "succeeded"
	PaidFailed    PaidStatus = "failed"
)

// Capabilities — 渠道能力自述(设计 §4.1 capabilities())。
type Capabilities struct {
	RenderTypes       []RenderType
	SupportsRefund    bool
	SupportsRecurring bool
	RecurringKind     string // token_offsession/gateway_scheduled/store_managed/none
	SettleCurrencies  []string
	Regions           []string
}

// CreateRequest — Provider.Create 入参:pay 已定金额/币种/账户,Provider 拿去渠道建单。
type CreateRequest struct {
	OutTradeNo  string
	Subject     string
	AmountMinor int64
	Currency    string
	Account     config.AccountConfig // 选中的收款账户(含 CredentialEnvPrefix)
	ReturnURL   string
	Metadata    map[string]string
}

// Session — Provider.Create 产出:渠道单号 + 付款意图数据(render_type + payload)。
type Session struct {
	ProviderRef string
	RenderType  RenderType
	Payload     map[string]any
	ExpiresAt   *time.Time
}

// CallbackInput — 渠道原始回调(HTTP body/头/query),由各 Provider 自行解析验签。
type CallbackInput struct {
	Raw     []byte
	Headers map[string]string
	Query   map[string]string
}

// PaidEvent — verify_callback / query 的统一产出(设计 §4.1 → {order_ref,status,paid_amount})。
type PaidEvent struct {
	ProviderRef     string
	Status          PaidStatus
	PaidAmountMinor int64
	PaidCurrency    string
	Raw             string
	PaidAt          *time.Time // 渠道报的支付时间;nil 则 settle 用收到时间,对账时两边时间才对得上
}

// QueryRequest — Provider.Query 入参:尝试的完整上下文快照,不是裸 provider_ref。
// crypto 自托管的"查单"= 按地址+期望金额+时间窗扫链核对;裸 ref 会逼渠道 adapter
// 自建 ref→(地址/金额/窗口) 映射表,重复 pay 已持有的数据。管线侧(SyncPendingAttempts)
// 本就拿着整个 attempt,填这个结构零成本。
type QueryRequest struct {
	ProviderRef string
	OutTradeNo  string
	AccountID   string
	AmountMinor int64
	Currency    string
	CreatedAt   time.Time
	ExpiresAt   *time.Time
}

var (
	ErrUnknownMethod = errors.New("provider: unknown method")
	ErrNotSupported  = errors.New("provider: capability not supported")
)

// Provider — 每个支付渠道实现的统一接口(设计 §4.1 PaymentProvider)。
type Provider interface {
	Method() string
	Capabilities() Capabilities
	Create(ctx context.Context, req CreateRequest) (*Session, error)
	VerifyCallback(ctx context.Context, in CallbackInput) (*PaidEvent, error)
	Query(ctx context.Context, req QueryRequest) (*PaidEvent, error)
}

// RefundingProvider — 可选:支持渠道退款的 Provider 额外实现(P4;不支持则 capabilities=false)。
type RefundingProvider interface {
	Provider
	Refund(ctx context.Context, providerRef string, amountMinor int64, reason string) (refundRef string, status PaidStatus, err error)
}

// RecurringProvider — 可选:支持自动续订(P8,设计 §5.1 4 类 kind)。
type RecurringProvider interface {
	Provider
	CreateAgreement(ctx context.Context, req CreateRequest) (agreementRef string, err error)
	Charge(ctx context.Context, agreementRef string, amountMinor int64, currency string) (*PaidEvent, error)
	CancelAgreement(ctx context.Context, agreementRef string) error
}

// Registry — 方法名 → Provider(设计 §2 Provider adapter 注册表)。启动期注册,运行期只读。
type Registry struct{ providers map[string]Provider }

func NewRegistry() *Registry { return &Registry{providers: map[string]Provider{}} }

func (r *Registry) Register(p Provider) { r.providers[p.Method()] = p }

func (r *Registry) Get(method string) (Provider, error) {
	p, ok := r.providers[method]
	if !ok {
		return nil, fmt.Errorf("%w: %s", ErrUnknownMethod, method)
	}
	return p, nil
}

// Methods 返回已注册方法名(字典序,供 GET /methods 下发已启用渠道)。
func (r *Registry) Methods() []string {
	out := make([]string, 0, len(r.providers))
	for m := range r.providers {
		out = append(out, m)
	}
	sort.Strings(out)
	return out
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/provider/ -v Expected: TestRegistryRegisterGet 与 TestSessionAndCaps PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/provider/provider.go internal/provider/provider_test.go
git commit -m "feat(v2): Provider 渠道抽象接口 + 6 render_type + PaidEvent + 注册表"

---

Task 2: fake provider(测试用,验证管线;真渠道在 P3)

Files:

• Create: internal/provider/fake/fake.go(可构建包,非 _test,供管线测试 import)
• Test: internal/provider/fake/fake_test.go

Interfaces:

• Consumes: provider.*(Task 1)。

• Produces:

  • type Provider struct{...} · func New() *Provider(实现 provider.Provider,Method()=="fake",render_type=crypto_address)。
  • Create 确定性产出 provider_ref = "FAKE-"+OutTradeNo,payload 含 address/amount_minor/currency。
  • VerifyCallback 解析 JSON {provider_ref,status,amount_minor,currency} → PaidEvent。
  • Query(ctx, provider.QueryRequest) 按 req.ProviderRef 返回预置结果(默认 pending);SetQueryResult(providerRef, PaidEvent) 测试 seam(模拟查单命中)。

• Step 1: 写失败测试

internal/provider/fake/fake_test.go:

package fake_test

import (
	"context"
	"testing"

	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/provider/fake"
)

func TestFakeCreateAndVerify(t *testing.T) {
	var _ provider.Provider = fake.New() // 编译期断言满足接口

	p := fake.New()
	ctx := context.Background()

	sess, err := p.Create(ctx, provider.CreateRequest{OutTradeNo: "PAY-1", AmountMinor: 29990000, Currency: "USDT"})
	if err != nil {
		t.Fatalf("create: %v", err)
	}
	if sess.ProviderRef != "FAKE-PAY-1" || sess.RenderType != provider.RenderCryptoAddress {
		t.Fatalf("session = %+v", sess)
	}

	ev, err := p.VerifyCallback(ctx, provider.CallbackInput{
		Raw: []byte(`{"provider_ref":"FAKE-PAY-1","status":"succeeded","amount_minor":29990000,"currency":"USDT"}`),
	})
	if err != nil {
		t.Fatalf("verify: %v", err)
	}
	if ev.ProviderRef != "FAKE-PAY-1" || ev.Status != provider.PaidSucceeded || ev.PaidAmountMinor != 29990000 {
		t.Fatalf("event = %+v", ev)
	}
}

func TestFakeQueryPreset(t *testing.T) {
	p := fake.New()
	ctx := context.Background()

	ev, _ := p.Query(ctx, provider.QueryRequest{ProviderRef: "FAKE-PAY-2"})
	if ev.Status != provider.PaidPending {
		t.Fatalf("默认应 pending, got %v", ev.Status)
	}

	p.SetQueryResult("FAKE-PAY-2", provider.PaidEvent{
		ProviderRef: "FAKE-PAY-2", Status: provider.PaidSucceeded, PaidAmountMinor: 100, PaidCurrency: "USDT",
	})
	ev2, _ := p.Query(ctx, provider.QueryRequest{ProviderRef: "FAKE-PAY-2"})
	if ev2.Status != provider.PaidSucceeded || ev2.PaidAmountMinor != 100 {
		t.Fatalf("预置后应 succeeded, got %+v", ev2)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/provider/fake/ -v Expected: 编译失败 —— fake 包不存在。

• Step 3: 写实现

internal/provider/fake/fake.go:

// 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.
package fake

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"

	"github.com/wangjia/pay/internal/provider"
)

type Provider struct {
	mu           sync.Mutex
	queryResults map[string]provider.PaidEvent
}

func New() *Provider { return &Provider{queryResults: map[string]provider.PaidEvent{}} }

func (p *Provider) Method() string { return "fake" }

func (p *Provider) Capabilities() provider.Capabilities {
	return provider.Capabilities{
		RenderTypes:      []provider.RenderType{provider.RenderCryptoAddress},
		SupportsRefund:   false,
		SettleCurrencies: []string{"USDT"},
		Regions:          []string{"global"},
	}
}

func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (*provider.Session, error) {
	exp := time.Now().Add(15 * time.Minute)
	return &provider.Session{
		ProviderRef: "FAKE-" + req.OutTradeNo,
		RenderType:  provider.RenderCryptoAddress,
		Payload: map[string]any{
			"address":      "TFake" + req.Account.AccountID + req.OutTradeNo,
			"amount_minor": req.AmountMinor,
			"currency":     req.Currency,
		},
		ExpiresAt: &exp,
	}, nil
}

// VerifyCallback 解析测试注入的 JSON 回调体成归一化 PaidEvent。
func (p *Provider) VerifyCallback(_ context.Context, in provider.CallbackInput) (*provider.PaidEvent, error) {
	var b struct {
		ProviderRef string `json:"provider_ref"`
		Status      string `json:"status"`
		AmountMinor int64  `json:"amount_minor"`
		Currency    string `json:"currency"`
	}
	if err := json.Unmarshal(in.Raw, &b); err != nil {
		return nil, err
	}
	if b.ProviderRef == "" {
		return nil, errors.New("fake: missing provider_ref")
	}
	return &provider.PaidEvent{
		ProviderRef:     b.ProviderRef,
		Status:          provider.PaidStatus(b.Status),
		PaidAmountMinor: b.AmountMinor,
		PaidCurrency:    b.Currency,
		Raw:             string(in.Raw),
	}, nil
}

func (p *Provider) Query(_ context.Context, req provider.QueryRequest) (*provider.PaidEvent, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if ev, ok := p.queryResults[req.ProviderRef]; ok {
		e := ev
		return &e, nil
	}
	return &provider.PaidEvent{ProviderRef: req.ProviderRef, Status: provider.PaidPending}, nil
}

// SetQueryResult primes Query to report a specific event (test seam).
func (p *Provider) SetQueryResult(providerRef string, ev provider.PaidEvent) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.queryResults[providerRef] = ev
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/provider/... -v Expected: provider 与 provider/fake 两包 PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/provider/fake/
git commit -m "feat(v2): fake provider(确定性,验证收款管线;真渠道 P3)"

---

Task 3: OrderStore 扩展 — 按 provider_ref 定位尝试 / 取单 / 列待处理 / 弃过期尝试

Files:

• Create: internal/store/order_query.go(扩展查询/状态方法 + 哨兵错误)
• Test: internal/store/order_query_test.go

Interfaces:

• Consumes: model.OrderV2/model.Attempt + 状态枚举(P1);model.OpenTestDB。

• Produces(挂 *OrderStore):

  • var ErrOrderNotFound, ErrAttemptNotFound = errors.New(...)
  • GetOrder(outTradeNo string) (*model.OrderV2, error) — 未命中返回 ErrOrderNotFound。
  • AttemptByProviderRef(providerRef string) (*model.Attempt, error) — 由裸 provider_ref 反查出 attempt(含 OutTradeNo+Channel),未命中 ErrAttemptNotFound。
  • ListAttemptsByStatus(status model.AttemptStatus, limit int) ([]model.Attempt, error) — 供查单兜底扫 pending。
  • ExpirePendingAttempts(outTradeNo string) (int64, error) — retry 前把该单 pending 尝试置 expired,返回受影响数。

• Step 1: 写失败测试

internal/store/order_query_test.go:

package store_test

import (
	"errors"
	"testing"
	"time"

	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/store"
)

func TestGetOrderAndAttemptByRef(t *testing.T) {
	s := store.NewOrderStore(model.OpenTestDB(t))
	seedOrder(t, s, "PAY-Q1") // 复用 order_test.go 的 seedOrder(建 order + attempt provider_ref="P-PAY-Q1")

	o, err := s.GetOrder("PAY-Q1")
	if err != nil || o.OutTradeNo != "PAY-Q1" {
		t.Fatalf("GetOrder = %+v, %v", o, err)
	}
	if _, err := s.GetOrder("NOPE"); !errors.Is(err, store.ErrOrderNotFound) {
		t.Fatalf("缺单应 ErrOrderNotFound, got %v", err)
	}

	att, err := s.AttemptByProviderRef("P-PAY-Q1")
	if err != nil || att.OutTradeNo != "PAY-Q1" || att.Channel != "crypto" {
		t.Fatalf("AttemptByProviderRef = %+v, %v", att, err)
	}
	if _, err := s.AttemptByProviderRef("P-UNKNOWN"); !errors.Is(err, store.ErrAttemptNotFound) {
		t.Fatalf("缺尝试应 ErrAttemptNotFound, got %v", err)
	}
}

func TestListPendingAndExpire(t *testing.T) {
	s := store.NewOrderStore(model.OpenTestDB(t))
	seedOrder(t, s, "PAY-Q2")

	pending, err := s.ListAttemptsByStatus(model.AttemptPending, 10)
	if err != nil || len(pending) != 1 {
		t.Fatalf("ListAttemptsByStatus pending = %d, %v", len(pending), err)
	}

	n, err := s.ExpirePendingAttempts("PAY-Q2")
	if err != nil || n != 1 {
		t.Fatalf("ExpirePendingAttempts = %d, %v", n, err)
	}
	pending2, _ := s.ListAttemptsByStatus(model.AttemptPending, 10)
	if len(pending2) != 0 {
		t.Fatalf("弃过期后应无 pending, got %d", len(pending2))
	}
	_ = time.Now
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ -run 'GetOrderAndAttempt|ListPendingAndExpire' -v Expected: 编译失败 —— 方法/哨兵未定义。

• Step 3: 写实现

internal/store/order_query.go:

package store

import (
	"errors"
	"fmt"

	"gorm.io/gorm"

	"github.com/wangjia/pay/internal/model"
)

var (
	ErrOrderNotFound   = errors.New("store: order not found")
	ErrAttemptNotFound = errors.New("store: attempt not found")
)

// GetOrder returns an order by out_trade_no.
func (s *OrderStore) GetOrder(outTradeNo string) (*model.OrderV2, error) {
	var o model.OrderV2
	if err := s.db.Where("out_trade_no = ?", outTradeNo).First(&o).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, ErrOrderNotFound
		}
		return nil, fmt.Errorf("store.GetOrder: %w", err)
	}
	return &o, nil
}

// AttemptByProviderRef resolves an attempt from a bare provider_ref, so settlement
// can recover out_trade_no + channel from a callback/query that only carries the ref.
func (s *OrderStore) AttemptByProviderRef(providerRef string) (*model.Attempt, error) {
	var a model.Attempt
	if err := s.db.Where("provider_ref = ?", providerRef).First(&a).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, ErrAttemptNotFound
		}
		return nil, fmt.Errorf("store.AttemptByProviderRef: %w", err)
	}
	return &a, nil
}

// ListAttemptsByStatus lists attempts in a status (for query-sync fallback).
func (s *OrderStore) ListAttemptsByStatus(status model.AttemptStatus, limit int) ([]model.Attempt, error) {
	if limit <= 0 || limit > 200 {
		limit = 100
	}
	var out []model.Attempt
	if err := s.db.Where("status = ?", status).Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
		return nil, fmt.Errorf("store.ListAttemptsByStatus: %w", err)
	}
	return out, nil
}

// ExpirePendingAttempts marks all pending attempts of an order as expired
// (used before a retry spawns a fresh attempt). Order status is untouched.
func (s *OrderStore) ExpirePendingAttempts(outTradeNo string) (int64, error) {
	res := s.db.Model(&model.Attempt{}).
		Where("out_trade_no = ? AND status = ?", outTradeNo, model.AttemptPending).
		Update("status", model.AttemptExpired)
	if res.Error != nil {
		return 0, fmt.Errorf("store.ExpirePendingAttempts: %w", res.Error)
	}
	return res.RowsAffected, nil
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ -v Expected: 新增两测试 + P1 既有 TestMarkPaidIdempotentAndCancel/TestMarkAttemptPaidProviderRefMismatchRollsBack 全 PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/store/order_query.go internal/store/order_query_test.go
git commit -m "feat(v2): OrderStore 扩展 GetOrder/AttemptByProviderRef/ListByStatus/ExpirePending"

---

Task 4: 一次性收款管线 — 下单 / 取单 / 重试 / 取消

Files:

• Create: internal/gateway/gateway.go(Gateway + 接口 ProductResolver/WebhookEnqueuer + CreateOrder/GetOrder/RetryOrder/CancelOrder)
• Create: internal/gateway/product.go(DBProductResolver 默认实现:model.Product 价 string 元 → money int64)
• Test: internal/gateway/gateway_test.go
• Test: internal/gateway/product_test.go

Interfaces:

• Consumes: provider.Registry(T1)、store.OrderStore(P1+T3)、accounts.Registry(P1)、money(P1)、util.NewOutTradeNo。

• Produces:

  • type ProductResolver interface{ Resolve(sku string) (amountMinor int64, currency, subject, bizCode string, err error) }
  • type WebhookEnqueuer interface{ Enqueue(outTradeNo, bizSystem, eventType string, data map[string]any) error }
  • type Gateway struct{...} · func New(orders *store.OrderStore, providers *provider.Registry, accts *accounts.Registry, products ProductResolver, webhook WebhookEnqueuer, region string) *Gateway
  • type CreateOrderInput struct{ SKU, Method, BizSystem, BizRef, ReturnURL string }
  • type SessionView struct{ RenderType string; Payload map[string]any; ExpiresAt *time.Time }
  • type OrderResult struct{ OrderNo string; Session SessionView }
  • CreateOrder(ctx, CreateOrderInput) (*OrderResult, error)
  • type OrderStatusView struct{ OrderNo, Status, Subject, Currency string; AmountMinor int64; PaidAt *time.Time }
  • GetOrder(outTradeNo string) (*OrderStatusView, error)
  • RetryOrder(ctx, outTradeNo, method string) (*OrderResult, error)(order 须 pending;弃旧 pending 尝试 → 新建尝试)
  • CancelOrder(outTradeNo string) (bool, error)
  • 错误:ErrProductNotFound、ErrNoAccount、ErrOrderNotPending。
  • DBProductResolver(默认币种 CNY;按 biz_code 查 active product,money.Parse(price,cur))。

• Step 1: 写失败测试

internal/gateway/gateway_test.go:

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)
	}
	att, err := orders.AttemptByProviderRef("FAKE-" + res.OrderNo)
	if err != nil || att.Channel != "fake" || att.AccountID != "fake-a1" || att.Status != model.AttemptPending {
		t.Fatalf("attempt = %+v, %v", att, err)
	}
}

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_test.go:

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)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/gateway/ -v Expected: 编译失败 —— gateway 包不存在。

• Step 3: 写 gateway 实现

internal/gateway/gateway.go:

// 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)
}

⚠️ retry 会为同一 provider_ref(fake 恒为 FAKE-<outNo>)再次 CreateAttempt,而 Attempt 有 uniqueIndex(channel,provider_ref)(P1)。真实渠道每次 create 返回不同 provider_ref,不冲突;fake provider 的 retry 测试须换 method 或让 fake 每次产不同 ref。上面 TestRetryAndCancel 里 retry 用同 method fake → 同 ref 会撞唯一键。实现 Step 修正:在 fake.Create 的 provider_ref 追加纳秒糊避免自撞:"FAKE-" + req.OutTradeNo(建单)与 retry 冲突。为让测试稳定,fake Create 改为 provider_ref = "FAKE-" + req.OutTradeNo + "-" + strconv.FormatInt(time.Now().UnixNano(),36),并在 Task 2 测试里改断言为 strings.HasPrefix(sess.ProviderRef, "FAKE-PAY-1")。见下方 Step 3a。

• Step 3a: 回改 fake provider 的 provider_ref 唯一化(避免 retry 撞唯一键)

改 internal/provider/fake/fake.go 的 Create:

import (
	// ... 追加
	"strconv"
	"strings"
)

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: ref,
		RenderType:  provider.RenderCryptoAddress,
		Payload: map[string]any{
			"address":      "TFake" + req.Account.AccountID,
			"amount_minor": req.AmountMinor,
			"currency":     req.Currency,
		},
		ExpiresAt: &exp,
	}, nil
}

同步把 Task 2 fake_test.go 的断言改为前缀匹配:

	if !strings.HasPrefix(sess.ProviderRef, "FAKE-PAY-1") || sess.RenderType != provider.RenderCryptoAddress {
		t.Fatalf("session = %+v", sess)
	}

(测试文件加 import "strings"。)并把本 Task gateway_test.go 里 AttemptByProviderRef("FAKE-" + res.OrderNo) 改为按 order 反查尝试:

	atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
	if len(atts) != 1 || atts[0].Channel != "fake" || atts[0].AccountID != "fake-a1" {
		t.Fatalf("attempt = %+v", atts)
	}

说明:此唯一化是 fake 专属(真实渠道天然每单不同 ref);记入 Self-Review。

• Step 4: 写 DBProductResolver 实现

internal/gateway/product.go:

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
}

• Step 5: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/gateway/ ./internal/provider/... -v Expected: gateway 4 测试 + provider/fake 测试全 PASS(fake 断言已改前缀)。

• Step 6: 全量编译 + 测试

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/... Expected: 编译通过,全 ok。

• Step 7: Commit

cd /Users/wangjia/code/pay
git add internal/gateway/gateway.go internal/gateway/product.go internal/gateway/gateway_test.go internal/gateway/product_test.go internal/provider/fake/fake.go internal/provider/fake/fake_test.go
git commit -m "feat(v2): 一次性收款管线 CreateOrder/GetOrder/Retry/Cancel + DBProductResolver"

---

Task 5: 统一开通/入账管线 — PaidEvent 归一 → 定位 → 幂等 → 金额核对 → 标付 → 入队 webhook

Files:

• Create: internal/gateway/settle.go(Settle / HandleCallback / SyncPendingAttempts)
• Test: internal/gateway/settle_test.go

Interfaces:

• Consumes: provider.PaidEvent、store.OrderStore.AttemptByProviderRef/MarkAttemptPaid/GetOrder(P1+T3)、WebhookEnqueuer。

• Produces(挂 *Gateway):

  • type SettleResult string + SettleIgnored/SettleNotFound/SettleAmountMismatch/SettleDuplicate/SettleProcessed/SettleFailed。
  • var ErrAmountMismatch = errors.New(...)
  • Settle(ctx, ev *provider.PaidEvent) (SettleResult, error) — 定位 attempt→order,核对币种/金额,先幂等入队 payment.succeeded、再幂等 MarkAttemptPaid 翻转(崩溃安全顺序,见实现注释;入队失败返回 SettleFailed 且不翻转,渠道重投可恢复)。
  • HandleCallback(ctx, method string, in provider.CallbackInput) (SettleResult, error) — provider.VerifyCallback → Settle。
  • SyncPendingAttempts(ctx, limit int) (int, error) — 扫 pending 尝试 → 各自 provider.Query(传 QueryRequest 上下文快照)→ Settle(查单兜底,设计 §8「对账优先于实时」)。

• Step 1: 写失败测试

internal/gateway/settle_test.go:

package gateway_test

import (
	"context"
	"encoding/json"
	"testing"

	"github.com/wangjia/pay/internal/gateway"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/provider"
)

func attemptRef(t *testing.T, orders interface {
	ListAttemptsByStatus(model.AttemptStatus, int) ([]model.Attempt, error)
}) string {
	t.Helper()
	atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
	if len(atts) == 0 {
		t.Fatalf("无 pending 尝试")
	}
	return atts[0].ProviderRef
}

func TestSettleHappyIdempotentAndWebhook(t *testing.T) {
	g, _, spy, orders := newGateway(t)
	ctx := context.Background()
	res, _ := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
	ref := attemptRef(t, orders)

	ev := &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"}
	got, err := g.Settle(ctx, ev)
	if err != nil || got != gateway.SettleProcessed {
		t.Fatalf("settle#1 = %v, %v", got, err)
	}
	// 订单已 paid
	o, _ := orders.GetOrder(res.OrderNo)
	if o.Status != model.OrderPaidV2 {
		t.Fatalf("order 应 paid, got %v", o.Status)
	}
	// webhook 入队一次,payload 带 event_type
	if len(spy.calls) != 1 || spy.calls[0]["event_type"] != "payment.succeeded" || spy.calls[0]["out_trade_no"] != res.OrderNo {
		t.Fatalf("webhook calls = %+v", spy.calls)
	}

	// 幂等:再 settle → duplicate,不重复入队
	got2, _ := g.Settle(ctx, ev)
	if got2 != gateway.SettleDuplicate || len(spy.calls) != 1 {
		t.Fatalf("settle#2 = %v, calls=%d", got2, len(spy.calls))
	}
}

func TestSettleGuards(t *testing.T) {
	g, _, spy, orders := newGateway(t)
	ctx := context.Background()
	g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
	ref := attemptRef(t, orders)

	// 未 succeeded → ignored
	if got, _ := g.Settle(ctx, &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidPending}); got != gateway.SettleIgnored {
		t.Fatalf("pending 应 ignored, got %v", got)
	}
	// 未知 ref → not_found
	if got, _ := g.Settle(ctx, &provider.PaidEvent{ProviderRef: "GHOST", Status: provider.PaidSucceeded, PaidCurrency: "USDT", PaidAmountMinor: 1}); got != gateway.SettleNotFound {
		t.Fatalf("未知 ref 应 not_found, got %v", got)
	}
	// 少付 → amount_mismatch
	if got, err := g.Settle(ctx, &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidSucceeded, PaidCurrency: "USDT", PaidAmountMinor: 1}); got != gateway.SettleAmountMismatch || err == nil {
		t.Fatalf("少付应 amount_mismatch, got %v %v", got, err)
	}
	// 错币种 → amount_mismatch
	if got, _ := g.Settle(ctx, &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidSucceeded, PaidCurrency: "CNY", PaidAmountMinor: 29990000}); got != gateway.SettleAmountMismatch {
		t.Fatalf("错币种应 amount_mismatch, got %v", got)
	}
	if len(spy.calls) != 0 {
		t.Fatalf("守卫失败路径不应入队 webhook, got %d", len(spy.calls))
	}
}

// 资金命脉不变量:outbox 入队失败 → 绝不翻转订单(否则"已付但永不通知")。
// 渠道拿不到 200 会重投,重投时入队+翻转都幂等,自然恢复。
func TestSettleEnqueueFailureKeepsOrderPending(t *testing.T) {
	g, _, spy, orders := newGateway(t)
	ctx := context.Background()
	res, _ := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
	ref := attemptRef(t, orders)
	ev := &provider.PaidEvent{ProviderRef: ref, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"}

	spy.failNext = true
	if got, err := g.Settle(ctx, ev); got != gateway.SettleFailed || err == nil {
		t.Fatalf("入队失败应 SettleFailed+err, got %v, %v", got, err)
	}
	o, _ := orders.GetOrder(res.OrderNo)
	if o.Status != model.OrderPendingV2 {
		t.Fatalf("入队失败后订单必须仍 pending, got %v", o.Status)
	}

	// 渠道重投 → 入队成功 → 翻转
	if got, err := g.Settle(ctx, ev); err != nil || got != gateway.SettleProcessed {
		t.Fatalf("重投应 processed, got %v, %v", got, err)
	}
	if len(spy.calls) != 1 {
		t.Fatalf("恢复后应恰入队 1 次, got %d", len(spy.calls))
	}
}

func TestHandleCallbackAndSync(t *testing.T) {
	g, fp, _, orders := newGateway(t)
	ctx := context.Background()
	g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
	ref := attemptRef(t, orders)

	// 回调路径:fake.VerifyCallback 解析 JSON → Settle
	body, _ := json.Marshal(map[string]any{"provider_ref": ref, "status": "succeeded", "amount_minor": 29990000, "currency": "USDT"})
	got, err := g.HandleCallback(ctx, "fake", provider.CallbackInput{Raw: body})
	if err != nil || got != gateway.SettleProcessed {
		t.Fatalf("HandleCallback = %v, %v", got, err)
	}

	// 查单兜底:另起一单,预置 query 命中 → SyncPendingAttempts 收敛
	res2, _ := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-2"})
	atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
	ref2 := atts[0].ProviderRef
	fp.SetQueryResult(ref2, provider.PaidEvent{ProviderRef: ref2, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"})
	n, err := g.SyncPendingAttempts(ctx, 10)
	if err != nil || n < 1 {
		t.Fatalf("SyncPendingAttempts = %d, %v", n, err)
	}
	o2, _ := orders.GetOrder(res2.OrderNo)
	if o2.Status != model.OrderPaidV2 {
		t.Fatalf("查单兜底后 order 应 paid, got %v", o2.Status)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/gateway/ -run 'Settle|HandleCallback' -v Expected: 编译失败 —— Settle/SettleResult 等未定义。

• Step 3: 写实现

internal/gateway/settle.go:

package gateway

import (
	"context"
	"errors"
	"time"

	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/store"
)

type SettleResult string

const (
	SettleIgnored        SettleResult = "ignored"         // 非成功状态(pending/failed)
	SettleNotFound       SettleResult = "not_found"       // provider_ref 无对应 attempt
	SettleAmountMismatch SettleResult = "amount_mismatch" // 币种不符 / 少付
	SettleDuplicate      SettleResult = "duplicate"       // 订单已非 pending(幂等 no-op)
	SettleProcessed      SettleResult = "processed"       // 本次真正翻转为 paid
	SettleFailed         SettleResult = "failed"          // 暂时性失败(outbox 入队失败等):未翻转,渠道应重投
)

var ErrAmountMismatch = errors.New("gateway: paid amount/currency mismatch")

// Settle normalizes a PaidEvent into activation: locate order via provider_ref →
// attempt → order (设计 §4「归一化 PaidEvent」), reconcile currency+amount, then
// enqueue-before-flip:先幂等入队 payment.succeeded(unique 键,重复 no-op),
// 再经 P1 MarkAttemptPaid 幂等翻转(order=pending 原子守卫)。
//
// 顺序不变量(资金命脉):**订单为 paid ⇒ outbox 行必已存在**。
//   - 先翻转后入队:两步间崩溃 → 已收钱但业务方永不知情,且无任何机制重试 → 客诉才发现。
//   - 先入队后翻转:两步间崩溃 → outbox 里躺着一条"单还没付"的行;Notifier 投递前有
//     "订单已付"门禁(Task 6),不会把未付单通知出去。渠道因拿不到 200 会重投回调
//     (查单兜底同样收敛),重投时入队/翻转都幂等,自愈。
//   - 入队本身失败:返回 SettleFailed 且不翻转,同样交给渠道重投恢复。
func (g *Gateway) Settle(ctx context.Context, ev *provider.PaidEvent) (SettleResult, error) {
	if ev.Status != provider.PaidSucceeded {
		return SettleIgnored, nil // 非成功状态:确认收到即可
	}
	att, err := g.orders.AttemptByProviderRef(ev.ProviderRef)
	if err != nil {
		if errors.Is(err, store.ErrAttemptNotFound) {
			return SettleNotFound, nil
		}
		// 读库瞬时失败是可重试态(SettleFailed),不能与"查无此单"(终态,ack 即弃)
		// 混淆——否则按 result 决定 ack 的调用方会让渠道停止重投,已付单永不入账。
		return SettleFailed, err
	}
	// 金额/币种核对:币种须一致,实付须 ≥ 应收(允许 crypto 多付,拒少付)。
	if ev.PaidCurrency != att.Currency || ev.PaidAmountMinor < att.AmountMinor {
		return SettleAmountMismatch, ErrAmountMismatch
	}
	// paid_at 优先用渠道报的支付时间(对账时与渠道流水对得上),渠道不报才落收到时间。
	paidAt := time.Now()
	if ev.PaidAt != nil {
		paidAt = *ev.PaidAt
	}

	if err := g.enqueuePaymentSucceeded(att, paidAt); err != nil {
		return SettleFailed, err // 未入队绝不翻转;渠道重投时幂等恢复
	}
	flipped, err := g.orders.MarkAttemptPaid(att.OutTradeNo, att.Channel, ev.ProviderRef, paidAt)
	if err != nil {
		return SettleFailed, err
	}
	if !flipped {
		return SettleDuplicate, nil // 已处理过 / 已取消 / 已过期 → 幂等 no-op
	}
	return SettleProcessed, nil
}

// enqueuePaymentSucceeded 组 webhook 领域 payload 并幂等入队。仅订单仍 pending 时入队:
// 订单已 paid 说明翻转已发生,而翻转严格发生在成功入队之后(顺序不变量),行必已存在;
// 订单已 canceled 则不该通知(晚到支付走 P4 退款/P6 对账,不自动开通)。
func (g *Gateway) enqueuePaymentSucceeded(att *model.Attempt, paidAt time.Time) error {
	o, err := g.orders.GetOrder(att.OutTradeNo)
	if err != nil {
		return err
	}
	if o.BizSystem == "" || o.Status != model.OrderPendingV2 {
		return nil // 独立收款无业务方回调;或已翻转/已取消
	}
	data := map[string]any{
		"event_type":   "payment.succeeded",
		"out_trade_no": o.OutTradeNo,
		"biz_system":   o.BizSystem,
		"biz_ref":      o.BizRef,
		"amount_minor": o.AmountMinor,
		"currency":     o.Currency,
		"channel":      att.Channel,
		"paid_at":      paidAt.Format(time.RFC3339),
	}
	return g.webhook.Enqueue(o.OutTradeNo, o.BizSystem, "payment.succeeded", data)
}

// HandleCallback runs a channel's raw callback through its Provider.VerifyCallback
// (验签/解析封死在渠道内) then settles the normalized event.
func (g *Gateway) HandleCallback(ctx context.Context, method string, in provider.CallbackInput) (SettleResult, error) {
	prov, err := g.providers.Get(method)
	if err != nil {
		return SettleNotFound, err
	}
	ev, err := prov.VerifyCallback(ctx, in)
	if err != nil {
		return SettleNotFound, err
	}
	return g.Settle(ctx, ev)
}

// SyncPendingAttempts polls every pending attempt via its Provider.Query and
// settles hits — the query-based backstop for lost webhooks (设计 §8 对账优先).
// Returns how many attempts were newly settled to paid.
func (g *Gateway) SyncPendingAttempts(ctx context.Context, limit int) (int, error) {
	atts, err := g.orders.ListAttemptsByStatus(model.AttemptPending, limit)
	if err != nil {
		return 0, err
	}
	settled := 0
	for i := range atts {
		att := &atts[i]
		prov, err := g.providers.Get(att.Channel)
		if err != nil {
			continue
		}
		// 查单带尝试完整上下文(crypto 需要地址/金额/时间窗才能扫链核对)。
		ev, err := prov.Query(ctx, provider.QueryRequest{
			ProviderRef: att.ProviderRef, OutTradeNo: att.OutTradeNo, AccountID: att.AccountID,
			AmountMinor: att.AmountMinor, Currency: att.Currency,
			CreatedAt: att.CreatedAt, ExpiresAt: att.ExpiresAt,
		})
		if err != nil || ev == nil {
			continue
		}
		if res, _ := g.Settle(ctx, ev); res == SettleProcessed {
			settled++
		}
	}
	return settled, nil
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/gateway/ -v Expected: Settle/HandleCallback/Sync 三测试 + Task 4 四测试全 PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/gateway/settle.go internal/gateway/settle_test.go
git commit -m "feat(v2): 入账管线 Settle/HandleCallback/SyncPending(归一→定位→幂等→核对→标付→入队)"

---

Task 6: webhook v2 — outbox 模型 + 幂等入队 store + Notifier(event_type + HMAC + 重试)

Files:

• Create: internal/model/webhook_delivery.go(outbox 表 + unique(out_trade_no,event_type))
• Modify: internal/model/testdb.go(OpenTestDB 追加 &WebhookDelivery{})
• Create: internal/store/webhook.go(WebhookStore:EnqueueDelivery/ListUndelivered/MarkDelivered/MarkFailed)
• Create: internal/webhook/notifier.go(实现 gateway.WebhookEnqueuer + Deliver/DeliverPending/Start)
• Test: internal/store/webhook_test.go
• Test: internal/webhook/notifier_test.go

Interfaces:

• Produces:

  • model.WebhookDelivery{ Base; OutTradeNo/EventType(uniqueIndex uq_delivery); BizSystem; Payload; Delivered bool; Attempts int; LastError string }。
  • store.WebhookStore · NewWebhookStore(db) · EnqueueDelivery(outTradeNo,bizSystem,eventType,payload string) error(ON CONFLICT DO NOTHING 幂等)· ListUndelivered(limit) ([]WebhookDelivery,error) · MarkDelivered(id uint64) error · MarkFailed(id uint64, errMsg string) error。
  • webhook.Notifier · NewNotifier(ws *store.WebhookStore, bizConfig func(string)(config.BizSystemConfig,bool), orderPaid func(outTradeNo string)(bool,error)) *Notifier。实现 Enqueue(outTradeNo,bizSystem,eventType string, data map[string]any) error(JSON 序列化 → EnqueueDelivery)· DeliverPending(limit int) (int,error) · Start(interval time.Duration)。投递门禁:deliverOne 先经 orderPaid 确认订单已付才发(enqueue-before-flip 顺序不变量的另一半,防把"未付单"通知出去);未付则跳过本轮、不计失败,等翻转后自然放行。

• Consumes: util.HMACSign、config.BizSystemConfig。

• Step 1: 写失败测试

internal/store/webhook_test.go:

package store_test

import (
	"testing"

	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/store"
)

func TestWebhookOutboxEnqueueIdempotent(t *testing.T) {
	ws := store.NewWebhookStore(model.OpenTestDB(t))

	if err := ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", `{"a":1}`); err != nil {
		t.Fatalf("enqueue#1: %v", err)
	}
	// 幂等:同 (out_trade_no,event_type) 再入队不新增行、不报错。
	if err := ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", `{"a":1}`); err != nil {
		t.Fatalf("enqueue#2: %v", err)
	}
	list, _ := ws.ListUndelivered(10)
	if len(list) != 1 {
		t.Fatalf("应恰 1 行待投递, got %d", len(list))
	}

	if err := ws.MarkDelivered(list[0].ID); err != nil {
		t.Fatalf("markDelivered: %v", err)
	}
	if again, _ := ws.ListUndelivered(10); len(again) != 0 {
		t.Fatalf("投递后应 0 待投递, got %d", len(again))
	}
}

func TestWebhookMarkFailed(t *testing.T) {
	ws := store.NewWebhookStore(model.OpenTestDB(t))
	_ = ws.EnqueueDelivery("PAY-2", "jiu", "payment.succeeded", `{}`)
	list, _ := ws.ListUndelivered(10)
	if err := ws.MarkFailed(list[0].ID, "boom"); err != nil {
		t.Fatalf("markFailed: %v", err)
	}
	again, _ := ws.ListUndelivered(10)
	if len(again) != 1 || again[0].Attempts != 1 || again[0].LastError != "boom" {
		t.Fatalf("失败后应仍待投递且 attempts=1, got %+v", again)
	}
}

internal/webhook/notifier_test.go:

package webhook_test

import (
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/wangjia/pay/config"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/store"
	"github.com/wangjia/pay/internal/util"
	"github.com/wangjia/pay/internal/webhook"
)

func TestNotifierDeliversSignedEvent(t *testing.T) {
	const secret = "shh-secret"
	var gotBody []byte
	var gotHeaders http.Header
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotBody, _ = io.ReadAll(r.Body)
		gotHeaders = r.Header.Clone()
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("SUCCESS"))
	}))
	defer srv.Close()

	ws := store.NewWebhookStore(model.OpenTestDB(t))
	bizCfg := func(system string) (config.BizSystemConfig, bool) {
		if system == "pangolin" {
			return config.BizSystemConfig{CallbackURL: srv.URL, Secret: secret}, true
		}
		return config.BizSystemConfig{}, false
	}
	alwaysPaid := func(string) (bool, error) { return true, nil }
	n := webhook.NewNotifier(ws, bizCfg, alwaysPaid)

	// 经 Enqueuer 接口入队(gateway 就是这么调的)。
	err := n.Enqueue("PAY-1", "pangolin", "payment.succeeded", map[string]any{
		"event_type": "payment.succeeded", "out_trade_no": "PAY-1", "amount_minor": 29990000, "currency": "USDT",
	})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	sent, err := n.DeliverPending(10)
	if err != nil || sent != 1 {
		t.Fatalf("DeliverPending = %d, %v", sent, err)
	}

	// 校验签名头(pay→业务方,双向 HMAC,业务方可同法验签)。
	sys := gotHeaders.Get("X-Pay-System")
	ts := gotHeaders.Get("X-Pay-Timestamp")
	nonce := gotHeaders.Get("X-Pay-Nonce")
	sign := gotHeaders.Get("X-Pay-Sign")
	if gotHeaders.Get("X-Pay-Event") != "payment.succeeded" {
		t.Fatalf("缺 X-Pay-Event 头")
	}
	if !util.HMACVerify(secret, sign, sys, ts, nonce, string(gotBody)) {
		t.Fatalf("签名校验失败")
	}
	// body 带 event_type
	var m map[string]any
	_ = json.Unmarshal(gotBody, &m)
	if m["event_type"] != "payment.succeeded" || m["out_trade_no"] != "PAY-1" {
		t.Fatalf("body = %s", gotBody)
	}
	// 已标投递:再投不重发
	if again, _ := n.DeliverPending(10); again != 0 {
		t.Fatalf("已投递不应重发, got %d", again)
	}
}

func TestNotifierRetriesOnFailure(t *testing.T) {
	var hits int
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hits++
		w.WriteHeader(http.StatusInternalServerError) // 业务方暂时挂
	}))
	defer srv.Close()

	ws := store.NewWebhookStore(model.OpenTestDB(t))
	n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
		return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
	}, func(string) (bool, error) { return true, nil })
	_ = n.Enqueue("PAY-3", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})

	if sent, _ := n.DeliverPending(10); sent != 0 {
		t.Fatalf("失败不应算投递成功, got %d", sent)
	}
	// 仍待投递,可被下一轮重试兜底
	pend, _ := ws.ListUndelivered(10)
	if len(pend) != 1 || pend[0].Attempts != 1 {
		t.Fatalf("失败后应留队重试, got %+v", pend)
	}
	if _, _ = n.DeliverPending(10); hits < 2 {
		t.Fatalf("应重试第二次, hits=%d", hits)
	}
}

// 投递门禁:订单未付(settle 崩在"入队后、翻转前"的窗口)绝不把 payment.succeeded
// 发给业务方;也不计失败次数,等订单翻转后自然放行。
func TestNotifierGateSkipsUnpaidOrder(t *testing.T) {
	var hits int
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hits++
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("SUCCESS"))
	}))
	defer srv.Close()

	ws := store.NewWebhookStore(model.OpenTestDB(t))
	paid := false
	n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
		return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
	}, func(string) (bool, error) { return paid, nil })
	_ = n.Enqueue("PAY-4", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})

	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 0 {
		t.Fatalf("未付单不应投递, sent=%d hits=%d", sent, hits)
	}
	pend, _ := ws.ListUndelivered(10)
	if len(pend) != 1 || pend[0].Attempts != 0 {
		t.Fatalf("门禁跳过不应计失败, got %+v", pend)
	}

	paid = true // 订单翻转后放行
	if sent, _ := n.DeliverPending(10); sent != 1 || hits != 1 {
		t.Fatalf("翻转后应投递, sent=%d hits=%d", sent, hits)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/webhook/ -run Webhook -v; go test ./internal/webhook/ -v Expected: 编译失败 —— WebhookDelivery/WebhookStore/webhook 包未定义。

• Step 3: 写 outbox 模型 + 迁移

internal/model/webhook_delivery.go:

package model

// WebhookDelivery 是 pay→业务方 webhook 的 outbox(v2)。unique(out_trade_no,event_type)
// 保证同一订单同一事件只入队一次(幂等);后台 Notifier 扫 Delivered=false 重试兜底。
type WebhookDelivery struct {
	Base
	OutTradeNo string `gorm:"size:64;not null;uniqueIndex:uq_delivery" json:"out_trade_no"`
	EventType  string `gorm:"size:32;not null;uniqueIndex:uq_delivery" json:"event_type"`
	BizSystem  string `gorm:"index;size:32" json:"biz_system"`
	Payload    string `gorm:"type:text" json:"payload"` // 已序列化的领域 JSON(含 event_type)
	Delivered  bool   `gorm:"index;default:false" json:"delivered"`
	Attempts   int    `json:"attempts"`
	LastError  string `gorm:"size:255" json:"last_error,omitempty"`
}

改 internal/model/testdb.go 的 AutoMigrate 调用,追加 &WebhookDelivery{}:

	if err := db.AutoMigrate(&OrderV2{}, &Attempt{}, &Account{}, &Refund{}, &WebhookDelivery{}); err != nil {

• Step 4: 写 WebhookStore

internal/store/webhook.go:

package store

import (
	"fmt"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/wangjia/pay/internal/model"
)

type WebhookStore struct{ db *gorm.DB }

func NewWebhookStore(db *gorm.DB) *WebhookStore { return &WebhookStore{db: db} }

// EnqueueDelivery inserts an outbox row; a duplicate (out_trade_no,event_type)
// is a no-op (idempotent enqueue) via ON CONFLICT DO NOTHING.
func (s *WebhookStore) EnqueueDelivery(outTradeNo, bizSystem, eventType, payload string) error {
	row := model.WebhookDelivery{
		OutTradeNo: outTradeNo, BizSystem: bizSystem, EventType: eventType, Payload: payload,
	}
	err := s.db.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "out_trade_no"}, {Name: "event_type"}},
		DoNothing: true,
	}).Create(&row).Error
	if err != nil {
		return fmt.Errorf("store.EnqueueDelivery: %w", err)
	}
	return nil
}

func (s *WebhookStore) ListUndelivered(limit int) ([]model.WebhookDelivery, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	var out []model.WebhookDelivery
	if err := s.db.Where("delivered = ?", false).Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
		return nil, fmt.Errorf("store.ListUndelivered: %w", err)
	}
	return out, nil
}

func (s *WebhookStore) MarkDelivered(id uint64) error {
	if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
		Update("delivered", true).Error; err != nil {
		return fmt.Errorf("store.MarkDelivered: %w", err)
	}
	return nil
}

// MarkFailed increments attempts and records the last error, leaving the row
// undelivered for the next retry sweep.
func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error {
	if len(errMsg) > 255 {
		errMsg = errMsg[:255]
	}
	if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
		Updates(map[string]any{
			"attempts":   gorm.Expr("attempts + 1"),
			"last_error": errMsg,
		}).Error; err != nil {
		return fmt.Errorf("store.MarkFailed: %w", err)
	}
	return nil
}

• Step 5: 写 Notifier

internal/webhook/notifier.go:

// Package webhook delivers pay→business webhooks (v2, with event_type) from an
// outbox, signed with the shared HMAC scheme (双向验签,与 v1 notifyBiz 一致),
// with a background retry sweep as the backstop for lost/failed deliveries.
package webhook

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/wangjia/pay/config"
	"github.com/wangjia/pay/internal/store"
	"github.com/wangjia/pay/internal/util"
)

// BizConfigFunc resolves a business system's callback URL + HMAC secret.
type BizConfigFunc func(system string) (config.BizSystemConfig, bool)

// OrderPaidFunc reports whether an order is settled (paid). Delivery gate:
// settle 是"先入队后翻转",崩溃窗口里 outbox 可能存在"未付单"的行——投递前必须
// 门禁,否则会把 payment.succeeded 发给业务方、白给权益。
type OrderPaidFunc func(outTradeNo string) (bool, error)

type Notifier struct {
	deliveries *store.WebhookStore
	bizConfig  BizConfigFunc
	orderPaid  OrderPaidFunc
	client     *http.Client
}

func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc) *Notifier {
	return &Notifier{deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid,
		client: &http.Client{Timeout: 10 * time.Second}}
}

// Enqueue implements gateway.WebhookEnqueuer: serialize the domain payload and
// idempotently persist it to the outbox (delivery happens async).
func (n *Notifier) Enqueue(outTradeNo, bizSystem, eventType string, data map[string]any) error {
	body, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("webhook.Enqueue marshal: %w", err)
	}
	return n.deliveries.EnqueueDelivery(outTradeNo, bizSystem, eventType, string(body))
}

// DeliverPending flushes undelivered rows; returns how many succeeded this pass.
func (n *Notifier) DeliverPending(limit int) (int, error) {
	rows, err := n.deliveries.ListUndelivered(limit)
	if err != nil {
		return 0, err
	}
	ok := 0
	for i := range rows {
		if n.deliverOne(&rows[i]) {
			ok++
		}
	}
	return ok, nil
}

func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
	// 门禁:订单未付不投递(不计失败,等 settle 翻转后自然放行)。
	paid, err := n.orderPaid(d.OutTradeNo)
	if err != nil || !paid {
		return false
	}
	cfg, found := n.bizConfig(d.BizSystem)
	if !found || cfg.CallbackURL == "" {
		_ = n.deliveries.MarkFailed(d.ID, "biz system not configured")
		return false
	}
	ts := strconv.FormatInt(time.Now().Unix(), 10)
	nonce := uuid.NewString()
	sign := util.HMACSign(cfg.Secret, d.BizSystem, ts, nonce, d.Payload)

	req, err := http.NewRequest(http.MethodPost, cfg.CallbackURL, bytes.NewReader([]byte(d.Payload)))
	if err != nil {
		_ = n.deliveries.MarkFailed(d.ID, err.Error())
		return false
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-Pay-System", d.BizSystem)
	req.Header.Set("X-Pay-Event", d.EventType)
	req.Header.Set("X-Pay-Timestamp", ts)
	req.Header.Set("X-Pay-Nonce", nonce)
	req.Header.Set("X-Pay-Sign", sign)

	resp, err := n.client.Do(req)
	if err != nil {
		_ = n.deliveries.MarkFailed(d.ID, err.Error())
		return false
	}
	rb, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	resp.Body.Close()
	// 约定:业务方返回 HTTP 200 且响应含 SUCCESS 视为受理(与 v1 一致)。
	if resp.StatusCode == http.StatusOK && strings.Contains(strings.ToUpper(string(rb)), "SUCCESS") {
		_ = n.deliveries.MarkDelivered(d.ID)
		return true
	}
	_ = n.deliveries.MarkFailed(d.ID, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120)))
	return false
}

func truncate(s string, n int) string {
	if len(s) > n {
		return s[:n]
	}
	return s
}

// Start runs a background retry sweep (backstop for lost/failed webhooks).
func (n *Notifier) Start(interval time.Duration) {
	go func() {
		defer func() {
			if r := recover(); r != nil {
				log.Printf("[webhook] retry sweep panic recovered: %v", r)
			}
		}()
		t := time.NewTicker(interval)
		defer t.Stop()
		for range t.C {
			if _, err := n.DeliverPending(50); err != nil {
				log.Printf("[webhook] DeliverPending: %v", err)
			}
		}
	}()
}

deliverOne 形参类型是 *store.WebhookDeliveryRow —— 为不让 webhook 包直接依赖 model,在 store 暴露别名 type WebhookDeliveryRow = model.WebhookDelivery。在 internal/store/webhook.go 顶部加:

// WebhookDeliveryRow is the row type returned by WebhookStore (alias to the model).
type WebhookDeliveryRow = model.WebhookDelivery

并把 ListUndelivered 返回类型改为 []WebhookDeliveryRow(与 []model.WebhookDelivery 同一类型,测试里 store.WebhookStore 用法不变)。

• Step 6: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/webhook/ ./internal/model/ -v Expected: outbox 幂等 / markFailed / Notifier 签名投递 / 重试 四测试 + 既有全 PASS。

• Step 7: Commit

cd /Users/wangjia/code/pay
git add internal/model/webhook_delivery.go internal/model/testdb.go internal/store/webhook.go internal/store/webhook_test.go internal/webhook/notifier.go internal/webhook/notifier_test.go
git commit -m "feat(v2): webhook v2 outbox + Notifier(event_type + HMAC 双向 + 重试兜底)"

---

Task 7: HTTP 接线 — /api/v2 路由 + handler + 端到端联通 + main 装配

Files:

• Create: internal/handler/gateway.go(GatewayHandler:CreateOrder/GetStatus/Retry/Cancel/Callback)
• Modify: internal/router/router.go(加 SetupV2,挂 /v1 组)
• Modify: main.go(装配 provider 注册表 + gateway + notifier;autoMigrate 加 &model.WebhookDelivery{})
• Test: internal/handler/gateway_test.go(gin httptest 端到端)

Interfaces:

• Produces:

  • type GatewayHandler struct{...} · NewGatewayHandler(g *gateway.Gateway) *GatewayHandler。
  • router.SetupV2(r *gin.Engine, g *gateway.Gateway) 挂:POST /api/v2/orders、GET /api/v2/orders/:order_no、POST /api/v2/orders/:order_no/retry、POST /api/v2/orders/:order_no/cancel、POST /api/v2/callback/:method。

• Consumes: gateway.Gateway(T4/T5)、util.RespondError/RespondSuccess。业务方下单签名沿用 v1 verifyBizSign 范式(biz_system 非空即校验)。

• Step 1: 写失败测试(端到端:下单→查→回调→已付)

internal/handler/gateway_test.go:

package handler_test

import (
	"bytes"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"

	"github.com/gin-gonic/gin"

	"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/router"
	"github.com/wangjia/pay/internal/store"
)

type nopEnqueuer struct{}

func (nopEnqueuer) Enqueue(string, string, string, map[string]any) error { return nil }

type oneResolver struct{}

func (oneResolver) Resolve(sku string) (int64, string, string, string, error) {
	return 29990000, "USDT", "Pro 年付", "pro_year", nil
}

func buildEngine(t *testing.T) *gin.Engine {
	t.Helper()
	gin.SetMode(gin.TestMode)
	orders := store.NewOrderStore(model.OpenTestDB(t))
	preg := provider.NewRegistry()
	preg.Register(fake.New())
	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")
	r := gin.New()
	router.SetupV2(r, g)
	return r
}

func do(t *testing.T, r *gin.Engine, method, path string, body any) (*httptest.ResponseRecorder, map[string]any) {
	t.Helper()
	var buf bytes.Buffer
	if body != nil {
		_ = json.NewEncoder(&buf).Encode(body)
	}
	req := httptest.NewRequest(method, path, &buf)
	req.Header.Set("Content-Type", "application/json")
	w := httptest.NewRecorder()
	r.ServeHTTP(w, req)
	var out map[string]any
	_ = json.Unmarshal(w.Body.Bytes(), &out)
	return w, out
}

func TestV2OrderLifecycle(t *testing.T) {
	r := buildEngine(t)

	// 下单(独立收款,无 biz_system → 无需签名)
	w, out := do(t, r, http.MethodPost, "/api/v2/orders", map[string]any{"sku": "pro_year", "method": "fake"})
	if w.Code != http.StatusOK {
		t.Fatalf("create code=%d body=%v", w.Code, out)
	}
	data := out["data"].(map[string]any)
	orderNo := data["order_no"].(string)
	sess := data["session"].(map[string]any)
	if sess["render_type"] != "crypto_address" {
		t.Fatalf("session = %v", sess)
	}

	// 查单:pending
	_, out2 := do(t, r, http.MethodGet, "/api/v2/orders/"+orderNo, nil)
	if out2["data"].(map[string]any)["status"] != "pending" {
		t.Fatalf("status = %v", out2["data"])
	}

	// 取 provider_ref:直接构造 fake 回调体(provider_ref 由 payload 无法拿,需查尝试)
	// 这里用 callback 端点 + fake JSON:先取尝试 ref。测试通过再次下单太绕,
	// 改为:回调体里 provider_ref 用 order_no 反查——fake ref 前缀 FAKE-<orderNo>。
	// 直接命中:构造 verify 输入需真实 ref,故经 /api/v2/callback 前先查库拿 ref。
	// 简化:暴露一个内部查询——本测试用 status 已足够验证下单/查单闭环;
	// 回调闭环在 gateway settle_test 已覆盖。此处验证 callback 路由存在且 404 语义:
	wc, _ := do(t, r, http.MethodPost, "/api/v2/callback/fake", map[string]any{
		"provider_ref": "GHOST", "status": "succeeded", "amount_minor": 1, "currency": "USDT",
	})
	if wc.Code != http.StatusOK { // not_found 也回 200(渠道无需重投未知单)
		t.Fatalf("callback code=%d", wc.Code)
	}

	// 取消
	wCancel, outCancel := do(t, r, http.MethodPost, "/api/v2/orders/"+orderNo+"/cancel", nil)
	if wCancel.Code != http.StatusOK || outCancel["data"].(map[string]any)["canceled"] != true {
		t.Fatalf("cancel = %d %v", wCancel.Code, outCancel)
	}
}

说明:回调"成功入账"闭环已在 settle_test.go(Task 5)用真实 provider_ref 覆盖;handler 测试聚焦路由 + 请求/响应形状 + 下单/查/取消闭环,回调端点只验路由存在与 200 语义(未知单回 200 不重投)。

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/handler/ -run TestV2OrderLifecycle -v Expected: 编译失败 —— router.SetupV2 / GatewayHandler 未定义。

• Step 3: 写 handler

internal/handler/gateway.go:

package handler

import (
	"encoding/json"
	"errors"
	"io"
	"log"
	"net/http"

	"github.com/gin-gonic/gin"

	"github.com/wangjia/pay/internal/gateway"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/store"
	"github.com/wangjia/pay/internal/util"
)

type GatewayHandler struct {
	g *gateway.Gateway
}

func NewGatewayHandler(g *gateway.Gateway) *GatewayHandler { return &GatewayHandler{g: g} }

type createV2Request struct {
	SKU       string `json:"sku"`
	Method    string `json:"method"`
	BizSystem string `json:"biz_system,omitempty"`
	BizRef    string `json:"biz_ref,omitempty"`
	ReturnURL string `json:"return_url,omitempty"`
}

// CreateOrder POST /api/v2/orders —— 下单,返回 {order_no, session:{render_type, payload}}。
// biz_system 非空 → 校验 HMAC 签名(复用 v1 verifyBizSign)。
func (h *GatewayHandler) CreateOrder(c *gin.Context) {
	raw, err := io.ReadAll(http.MaxBytesReader(c.Writer, c.Request.Body, maxOrderBodyBytes))
	if err != nil {
		util.RespondError(c, http.StatusBadRequest, "bad_request", "请求体过大或读取失败")
		return
	}
	var req createV2Request
	if err := json.Unmarshal(raw, &req); err != nil {
		util.RespondError(c, http.StatusBadRequest, "bad_request", "参数格式错误")
		return
	}
	if req.SKU == "" || req.Method == "" {
		util.RespondError(c, http.StatusBadRequest, "bad_request", "缺少 sku 或 method")
		return
	}
	if req.BizSystem != "" {
		if err := verifyBizSign(c, req.BizSystem, raw); err != nil {
			util.RespondError(c, http.StatusUnauthorized, "unauthorized", err.Error())
			return
		}
	}
	res, err := h.g.CreateOrder(c.Request.Context(), gateway.CreateOrderInput{
		SKU: req.SKU, Method: req.Method, BizSystem: req.BizSystem, BizRef: req.BizRef, ReturnURL: req.ReturnURL,
	})
	if err != nil {
		h.writeCreateErr(c, "下单", req.Method, err)
		return
	}
	util.RespondSuccess(c, res)
}

// GetStatus GET /api/v2/orders/:order_no
func (h *GatewayHandler) GetStatus(c *gin.Context) {
	v, err := h.g.GetOrder(c.Param("order_no"))
	if err != nil {
		util.RespondError(c, http.StatusNotFound, "order_not_found", "订单不存在")
		return
	}
	util.RespondSuccess(c, v)
}

type retryRequest struct {
	Method string `json:"method"`
}

// Retry POST /api/v2/orders/:order_no/retry
func (h *GatewayHandler) Retry(c *gin.Context) {
	var req retryRequest
	_ = c.ShouldBindJSON(&req)
	if req.Method == "" {
		util.RespondError(c, http.StatusBadRequest, "bad_request", "缺少 method")
		return
	}
	res, err := h.g.RetryOrder(c.Request.Context(), c.Param("order_no"), req.Method)
	if err != nil {
		if errors.Is(err, gateway.ErrOrderNotPending) {
			util.RespondError(c, http.StatusConflict, "order_not_pending", "订单非待支付态,不可重试")
			return
		}
		if errors.Is(err, store.ErrOrderNotFound) {
			util.RespondError(c, http.StatusNotFound, "order_not_found", "订单不存在")
			return
		}
		h.writeCreateErr(c, "重试", req.Method, err)
		return
	}
	util.RespondSuccess(c, res)
}

// Cancel POST /api/v2/orders/:order_no/cancel
func (h *GatewayHandler) Cancel(c *gin.Context) {
	ok, err := h.g.CancelOrder(c.Param("order_no"))
	if err != nil {
		util.RespondError(c, http.StatusInternalServerError, "cancel_failed", "取消失败")
		return
	}
	util.RespondSuccess(c, gin.H{"canceled": ok})
}

// Callback POST /api/v2/callback/:method —— 渠道异步回调;经 provider.VerifyCallback → Settle。
// 已受理(含未知单/幂等/金额不符,都不需要渠道重投)一律回 200。
func (h *GatewayHandler) Callback(c *gin.Context) {
	method := c.Param("method")
	raw, err := io.ReadAll(http.MaxBytesReader(c.Writer, c.Request.Body, maxOrderBodyBytes))
	if err != nil {
		util.RespondError(c, http.StatusBadRequest, "bad_request", "请求体过大")
		return
	}
	headers := map[string]string{}
	for k := range c.Request.Header {
		headers[k] = c.GetHeader(k)
	}
	res, err := h.g.HandleCallback(c.Request.Context(), method, provider.CallbackInput{
		Raw: raw, Headers: headers,
	})
	// 按 SettleResult(而非只看 err)分终态/可重试:amount_mismatch 是终态——
	// 重投同一回调永远解决不了,必须回 200 让渠道停投(差错单归 P6 对账/人工);
	// SettleFailed/验签失败才是可重试态,回 400 让渠道重投。
	switch {
	case err == nil:
		c.JSON(http.StatusOK, gin.H{"result": string(res)})
	case res == gateway.SettleAmountMismatch:
		log.Printf("[v2 callback] method=%s 金额/币种不符(终态,已受理停投): %v", method, err)
		c.JSON(http.StatusOK, gin.H{"result": string(res)})
	default:
		log.Printf("[v2 callback] method=%s result=%s err=%v", method, res, err)
		util.RespondError(c, http.StatusBadRequest, "callback_failed", "回调处理失败")
	}
}

func (h *GatewayHandler) writeCreateErr(c *gin.Context, action, method string, err error) {
	switch {
	case errors.Is(err, gateway.ErrProductNotFound):
		util.RespondError(c, http.StatusNotFound, "product_not_found", "套餐不存在或已下架")
	case errors.Is(err, provider.ErrUnknownMethod):
		util.RespondError(c, http.StatusBadRequest, "unknown_method", "不支持的支付方式")
	case errors.Is(err, gateway.ErrNoAccount):
		util.RespondError(c, http.StatusServiceUnavailable, "no_account", "该支付方式暂不可用")
	default:
		log.Printf("[v2 order] %s失败 method=%s: %v", action, method, err)
		util.RespondError(c, http.StatusInternalServerError, "create_failed", action+"失败,请稍后重试")
	}
}

• Step 4: 写路由 SetupV2

internal/router/router.go 追加(不动既有 Setup):

import (
	// ... 既有 import 保留,追加:
	"github.com/wangjia/pay/internal/gateway"
)

// SetupV2 装配 pay v2 统一网关路由(/api/v2,与旧版同一 /api 前缀风格)。
// 旧 /api/v1 仅为存量支付宝当面付部署保留(路径 POST /api/v1/orders 与 v2 新契约
// 同名不同形,无法在同一前缀下并存);P3 支付宝 adapter 迁入 v2 后整组删除,
// 最终对外只剩 /api/v2 一套。
func SetupV2(r *gin.Engine, g *gateway.Gateway) {
	h := handler.NewGatewayHandler(g)
	v2 := r.Group("/api/v2")
	{
		v2.POST("/orders", h.CreateOrder)
		v2.GET("/orders/:order_no", h.GetStatus)
		v2.POST("/orders/:order_no/retry", h.Retry)
		v2.POST("/orders/:order_no/cancel", h.Cancel)
		v2.POST("/callback/:method", h.Callback)
	}
}

• Step 5: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go test ./internal/handler/ -run TestV2OrderLifecycle -v Expected: PASS。

• Step 6: main 装配(编译期接线,不新增测试)

main.go:①autoMigrate 追加 &model.WebhookDelivery{};②main() 里装配 v2(在 router.Setup(...) 之后):

// v2 统一网关装配(P2):provider 注册表 + gateway + webhook notifier。
pReg := provider.NewRegistry()
// P3 起在此 Register 真实渠道:crypto / alipay / stripe …(fake 仅测试用,不注册进生产)。
orderStore := store.NewOrderStore(db)
webhookStore := store.NewWebhookStore(db)
notifier := webhook.NewNotifier(webhookStore, config.C.BizByName, func(no string) (bool, error) {
	o, err := orderStore.GetOrder(no) // 投递门禁:订单已付才发(enqueue-before-flip 不变量)
	if err != nil {
		return false, err
	}
	return o.Status == model.OrderPaidV2, nil
})
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")
router.SetupV2(r, gw)

追加 import:internal/provider、internal/store、internal/webhook、internal/gateway、internal/accounts。

config.C.BizByName 签名 func(string)(config.BizSystemConfig,bool) 恰配 webhook.BizConfigFunc,直接传方法值。

• Step 7: 全量编译 + 测试

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./... Expected: 编译通过,全部包 ok(含 handler/gateway/provider/store/webhook/model/money/accounts)。

• Step 8: Commit

cd /Users/wangjia/code/pay
git add internal/handler/gateway.go internal/router/router.go main.go
git commit -m "feat(v2): /api/v2 网关路由 + handler(下单/查/重试/取消/回调)+ main 装配 gateway/notifier"

---

Self-Review

Spec coverage(P2 范围,对照设计 §3/§4/§5):

• Provider 接口 capabilities/create/verify_callback/query + 可选 refund/recurring 留空 + 注册表 map[method]Provider(Task 1)= 设计 §4.1 ✓。
• fake provider 验证管线,真渠道 P3(Task 2)✓。
• 一次性收款管线 POST /api/v2/orders → 选 provider → create → 落 Order+Attempt(复用 P1 OrderStore)→ 返回 {order_no, session:{render_type, payload}};GET /api/v2/orders/{no};retry/cancel(Task 4/7)= 设计 §3.1/3.2/4.2 ✓。
• 统一开通/入账:webhook 与 query 都产出归一化 PaidEvent → 定位(provider_ref→attempt→order)→ 金额/币种核对 → 先幂等入队、再幂等 MarkAttemptPaid 翻转(Task 5)= 设计 §4 ✓。顺序不变量「paid ⇒ outbox 行存在」消灭"已付但永不通知"的崩溃窗口;反向窗口("入队了但未翻转")由 Notifier 投递门禁(订单已付才发,Task 6)+ 渠道重投/查单兜底自愈。取消单晚到支付:入队 skip(状态非 pending)→ 翻转失败 → duplicate,不自动开通,渠道已扣款场景归 P4 退款/P6 对账。
• webhook v2:pay→业务方,payload 带 event_type(先 payment.succeeded),HMAC 双向签名(复用 util.HMACSign),outbox + 重试兜底(Task 6)= 设计 §5 ✓。
• 幂等/金额核对贯穿:入账走 P1 条件 UPDATE(order=pending 守卫);webhook 入队走 unique 键 ON CONFLICT;核对币种一致 + 实付 ≥ 应收(Task 5/6)✓。

复用 P1(不重造): 金额 money.Parse/Format、模型 model.OrderV2/Attempt + V2 后缀状态、store.OrderStore.CreateOrder/CreateAttempt/MarkAttemptPaid/CancelOrder/ListOrders、accounts.Registry.EnabledFor + config.AccountConfig、util.HMACSign/NewOutTradeNo/RespondError/RespondSuccess、model.OpenTestDB 全部直接复用。仅新增:provider 抽象、gateway 管线、store 查询扩展(GetOrder/AttemptByProviderRef/ListByStatus/ExpirePending)、webhook outbox。

Placeholder scan: 无 TODO/占位;每任务给完整 Go 代码 + 测试 + 命令。唯一"回改"是 Task 4 Step 3a 把 fake provider_ref 唯一化(纳秒后缀),原因:Attempt 有 uniqueIndex(channel,provider_ref)(P1),retry 同 method 会撞键;真实渠道每单天然不同 ref,此唯一化是 fake 专属测试保真手段,已在计划内闭合(同步改 Task 2 断言为前缀匹配 + gateway 测试按 ListAttemptsByStatus 取 ref)。

Type consistency: provider.Provider/Session/PaidEvent/CreateRequest/CallbackInput 在 Task 1 定义,Task 2(fake)、Task 4/5(gateway)、Task 7(handler)一致消费。gateway.WebhookEnqueuer.Enqueue(outTradeNo,bizSystem,eventType,data) 由 Task 5 消费、Task 6 webhook.Notifier 实现。store.WebhookDeliveryRow = model.WebhookDelivery(别名)让 webhook 包不直接依赖 model。config.C.BizByName 签名恰配 webhook.BizConfigFunc。金额一律 int64 minor + Currency。

评审修订(2026-07-10 用户评审 D3/D4 后,执行前已折入上文各 Task):

• D3 路由:v2 从裸 /v1 改挂 /api/v2(统一 /api 前缀;POST /api/v1/orders 被旧契约占用无法共存),P3 渠道迁入后删除整组 /api/v1。
• D4-A1 崩溃安全顺序:Settle 由"先翻转后入队"改为"先幂等入队、再幂等翻转"+ Notifier"订单已付"投递门禁 + SettleFailed(入队失败不翻转,渠道重投自愈)。消灭"已付但业务方永不知情"的窗口。
• D4-A2 Query 带上下文:Provider.Query(ctx, providerRef string) → Query(ctx, QueryRequest)(尝试完整快照)。crypto 的查单是"按地址+金额+时间窗扫链",裸 ref 会逼 adapter 自建映射表。
• D4-A3 渠道支付时间:PaidEvent 增 PaidAt *time.Time,settle/payload 优先用渠道报的时间,对账时与渠道流水对得上。
• D4 记录未调整项(有意延后):回调原始报文落表留痕(P2 先靠 PaidEvent.Raw + 服务日志,P3 落 callback_logs 表);Notifier 无退避/无死信上限(P6 加 max attempts + 告警);多账户回调验签需 provider 持账户注册表(P3 装配注入,接口不变);crypto 分笔凑单由 crypto adapter 聚合后才发 succeeded(P3)。

已知 scope 取舍(记录,不阻塞 P2):

• webhook payload 未含 product_biz_code(设计 §5 列了):model.OrderV2 无 biz_code 列(P1 已定),不改 P1 schema;P2 业务方用 biz_ref 映射自身订单足够,product_biz_code 待 P3 给 order 补 biz_code 列或经 product 关联时补齐。
• 路由策略取"首个 enabled 账户"(设计 §3.3 的 round_robin/weighted/limit_aware 在 P5)。
• 币种维度未进 product(DBProductResolver 用单一部署默认币种):多币种 product 在 P3+。
• v2 走 /api/v2(与旧版统一 /api 前缀;POST /api/v1/orders 已被旧契约占用,同前缀无法并存)。旧 /api/v1 仅为存量当面付部署保留,P3 渠道迁入后整组删除,收口为一套(设计 §1 / brain todo #5)。

---

后续阶段(各自独立成计划,落地前逐一细化)

• P3 首批渠道 adapter 实现 provider.Provider:crypto 自托管(包裹 pangolin-pay 单地址/地址池收款 + 链上确认 query)/ 支付宝(迁移 v1 internal/channel/alipay.go 到 provider,复用验签)/ Stripe(sdk_handoff + webhook 验签)。各自 Register 进注册表;补 order 的 biz_code / 多币种。
• P4 退款:POST /api/v2/refunds + provider.RefundingProvider + 三向(业务发起/crypto 人工 sweep/平台通知)+ refund.succeeded/refund.failed 事件 + order 退款态机(refunding/partially_refunded/refunded)。
• P5 多账户路由策略(round_robin/weighted/limit_aware/by_region/crypto 地址池)替换"首个 enabled"。
• P6 对账 job:周期 SyncPendingAttempts + 每 provider query 拉渠道流水 vs 本地订单,防掉单/防重复授权。必含订单级过期清理:超时未付 pending 单自动关闭——同时兜住 Task 4 复审记录的 plan-mandated 缺口(CreateOrder 建单后 prov.Create/CreateAttempt 失败 → 零尝试孤儿 pending 单,调用方拿不到单号无法 retry/cancel)。
• P7 codes 共享库(A 嵌入各产品)。
• P8(later) 订阅/recurring(provider.RecurringProvider 4 类 kind)、拒付 chargeback 事件。

契约同步(pay-contract):webhook event_type 断代 v2.0.0、render_type/多币种向后兼容加字段,随 P2/P3 落地登记。