feat(v2): 入账管线 Settle/HandleCallback/SyncPending(归一→定位→幂等→核对→标付→入队)
This commit is contained in:
wangjia
@@ -0,0 +1,138 @@
|
||||
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
|
||||
}
|
||||
return SettleNotFound, 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
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user