feat(pay): 收款模型改为单地址+唯一金额(#34/34A Phase A-C)

从"每单唯一 HD 地址"改为"单个固定收款地址 + 每单唯一金额",归集成本 O(订单数)→O(1)。 - store: pay_orders 加 user_ref/expect_amount(唯一金额)/matched_tx_id;新 orphan_payments 表; ActiveOrderByUser(同用户单订单)、AmountRecentlyUsed(迟到窗口内金额不复用)、TxHandled(幂等)、 RecordOrphan。去掉每单派生游标。 - pay: CreateOrder(userRef,sku,priceMicro)——同用户单订单校验 + 分配唯一金额(base+随机微尾数[1,9999]、 cooldown 内不复用),address 恒为收款地址。 - tron: Transfer 加 BlockTs(区块时间秒),取 block_timestamp。 - watcher: 单地址取到账,按"金额==expect && block_ts>建单"匹配 → paid;不匹配的到账 → orphan;幂等。 - httpapi: POST /order 加 user_ref,同用户重复 → 409;main 收款地址=PAY_RECEIVE_ADDRESS 或 xpub index0。 - 测试:唯一金额/同地址、同用户单订单、精确匹配、付错成孤儿、迟到不误配新单、付款早于建单不匹配、超时、幂等、409,全绿。README 更新为单地址模型+API(user_ref/精确金额/orphan)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-09 15:15:38 +08:00
parent 543a54c606
commit c5949a595a
11 changed files with 535 additions and 240 deletions
@@ -1,6 +1,7 @@
-// Package watcher polls TronGrid for incoming USDT and marks paid orders.
-// It only reads the chain and flips order state — it never holds keys or moves
-// funds (sweeping is a separate offline step).
+// Package watcher polls TronGrid for incoming USDT to the single receiving
+// address and matches each confirmed payment to a pending order by exact amount
+// + block time. It only reads the chain and flips order state — it never holds
+// keys or moves funds (sweeping is a separate offline step).
 package watcher

 import (
@@ -13,23 +14,28 @@ import (
 )

 type Watcher struct {
-	st   *store.Store
-	tron tron.Fetcher
-	log  *slog.Logger
-	now  func() time.Time
+	st      *store.Store
+	tron    tron.Fetcher
+	address string
+	log     *slog.Logger
+	now     func() time.Time
 }

-func New(st *store.Store, f tron.Fetcher, log *slog.Logger) *Watcher {
+func New(st *store.Store, f tron.Fetcher, address string, log *slog.Logger) *Watcher {
 	if log == nil {
 		log = slog.Default()
 	}
-	return &Watcher{st: st, tron: f, log: log, now: time.Now}
+	return &Watcher{st: st, tron: f, address: address, log: log, now: time.Now}
 }

-// Tick: (1) expire overdue pending orders; (2) for each still-pending order,
-// look for a confirmed incoming transfer >= the expected amount on its unique
-// address and mark it paid. Idempotent — a transfer seen twice flips the order
-// at most once (MarkPaid only affects a still-pending row).
+// Tick:
+//  1. expire overdue pending orders;
+//  2. fetch confirmed incoming USDT transfers to the single receiving address;
+//  3. match each transfer to a pending order by **exact amount** and **block time
+//     after the order was created**; a confirmed transfer that matches no active
+//     order (wrong amount / late after reuse) is recorded as an orphan.
+//
+// Idempotent: a tx already matched to an order or recorded as orphan is skipped.
 func (w *Watcher) Tick(ctx context.Context) error {
 	if n, err := w.st.MarkExpired(ctx, w.now()); err != nil {
 		return err
@@ -37,31 +43,54 @@ func (w *Watcher) Tick(ctx context.Context) error {
 		w.log.Info("orders expired", "count", n)
 	}

+	transfers, err := w.tron.IncomingTransfers(ctx, w.address)
+	if err != nil {
+		w.log.Warn("fetch transfers failed", "err", err)
+		return nil // transient (rate limit / network); retried next tick
+	}
+	if len(transfers) == 0 {
+		return nil
+	}
+
 	pending, err := w.st.ListPending(ctx)
 	if err != nil {
 		return err
 	}
+	// Index pending orders by their unique expect amount.
+	byAmount := make(map[int64]*store.Order, len(pending))
 	for _, o := range pending {
-		transfers, err := w.tron.IncomingTransfers(ctx, o.Address)
+		byAmount[o.ExpectAmount] = o
+	}
+
+	for _, t := range transfers {
+		handled, err := w.st.TxHandled(ctx, t.TxID)
 		if err != nil {
-			// Transient (rate limit / network): log and move on; retried next tick.
-			w.log.Warn("fetch transfers failed", "order", o.OrderNo, "err", err)
+			w.log.Error("tx handled check", "tx", t.TxID, "err", err)
 			continue
 		}
-		for _, t := range transfers {
-			if t.Value < o.ExpectAmount {
-				continue
-			}
+		if handled {
+			continue // already matched or already an orphan
+		}
+
+		if o := byAmount[t.Value]; o != nil && t.BlockTs > o.CreatedAt.Unix() {
 			ok, err := w.st.MarkPaid(ctx, o.OrderNo, t.TxID)
 			if err != nil {
 				w.log.Error("mark paid", "order", o.OrderNo, "err", err)
-				break
+				continue
 			}
 			if ok {
-				w.log.Info("order paid", "order", o.OrderNo, "tx", t.TxID, "value", t.Value, "address", o.Address)
+				w.log.Info("order paid", "order", o.OrderNo, "tx", t.TxID, "value", t.Value)
+				delete(byAmount, t.Value) // a second transfer of the same amount can't reuse this order
 			}
-			break
+			continue
 		}
+
+		// Confirmed payment matching no active order -> orphan (needs reconciliation).
+		if err := w.st.RecordOrphan(ctx, t.TxID, w.address, t.Value, t.BlockTs, w.now()); err != nil {
+			w.log.Error("record orphan", "tx", t.TxID, "err", err)
+			continue
+		}
+		w.log.Warn("orphan payment", "tx", t.TxID, "value", t.Value, "block_ts", t.BlockTs)
 	}
 	return nil
 }
@@ -9,81 +9,159 @@ import (
 	"github.com/wangjia/pangolin/pay/internal/tron"
 )

-type mockFetcher struct{ m map[string][]tron.Transfer }
+const recvAddr = "TRecv00000000000000000000000000000A"

-func (f *mockFetcher) IncomingTransfers(_ context.Context, addr string) ([]tron.Transfer, error) {
-	return f.m[addr], nil
+type mockFetcher struct{ transfers []tron.Transfer }
+
+func (f *mockFetcher) IncomingTransfers(_ context.Context, _ string) ([]tron.Transfer, error) {
+	return f.transfers, nil
 }

-func newPending(t *testing.T, st *store.Store, orderNo, addr string, amount int64, expires time.Time) {
+func memStore(t *testing.T) *store.Store {
+	t.Helper()
+	st, err := store.Open(":memory:")
+	if err != nil {
+		t.Fatalf("store: %v", err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+	return st
+}
+
+func seed(t *testing.T, st *store.Store, no string, amount int64, created time.Time) {
 	t.Helper()
 	o := &store.Order{
-		OrderNo: orderNo, SKU: "pro", ExpectAmount: amount, Address: addr,
-		Status: store.StatusPending, CreatedAt: time.Unix(1_700_000_000, 0), ExpiresAt: expires,
+		OrderNo: no, UserRef: "u", SKU: "pro", ExpectAmount: amount, Address: recvAddr,
+		Status: store.StatusPending, CreatedAt: created, ExpiresAt: created.Add(time.Hour),
 	}
 	if err := st.CreateOrder(context.Background(), o); err != nil {
-		t.Fatalf("seed order: %v", err)
+		t.Fatalf("seed: %v", err)
 	}
 }

-func TestWatcherMarksPaidOnSufficientTransfer(t *testing.T) {
-	st, _ := store.Open(":memory:")
-	t.Cleanup(func() { _ = st.Close() })
+func TestWatcherMatchesByAmountAndTime(t *testing.T) {
+	st := memStore(t)
 	ctx := context.Background()
 	now := time.Unix(1_700_000_100, 0)
+	created := now.Add(-5 * time.Minute)
+	seed(t, st, "PAY1", 5_000017, created)
+	seed(t, st, "PAY2", 5_000018, created)

-	newPending(t, st, "PAY1", "TADDR1", 5_000000, now.Add(time.Hour))
-	fetch := &mockFetcher{m: map[string][]tron.Transfer{}}
-	w := New(st, fetch, nil)
-	w.now = func() time.Time { return now }
-
-	// No transfer yet -> stays pending.
-	if err := w.Tick(ctx); err != nil {
-		t.Fatalf("tick1: %v", err)
-	}
-	if o, _ := st.GetOrder(ctx, "PAY1"); o.Status != store.StatusPending {
-		t.Fatalf("want pending, got %s", o.Status)
-	}
-
-	// Underpayment -> still pending.
-	fetch.m["TADDR1"] = []tron.Transfer{{TxID: "tx-under", To: "TADDR1", Value: 4_000000}}
-	_ = w.Tick(ctx)
-	if o, _ := st.GetOrder(ctx, "PAY1"); o.Status != store.StatusPending {
-		t.Fatalf("underpay should stay pending, got %s", o.Status)
-	}
-
-	// Sufficient payment -> paid, tx recorded.
-	fetch.m["TADDR1"] = []tron.Transfer{{TxID: "tx-ok", To: "TADDR1", Value: 5_000000}}
-	_ = w.Tick(ctx)
-	o, _ := st.GetOrder(ctx, "PAY1")
-	if o.Status != store.StatusPaid || o.TxID != "tx-ok" {
-		t.Fatalf("want paid/tx-ok, got %s/%s", o.Status, o.TxID)
-	}
-
-	// Idempotent: another tick with same transfer doesn't error or flip anything.
-	if err := w.Tick(ctx); err != nil {
-		t.Fatalf("idempotent tick: %v", err)
-	}
-	o, _ = st.GetOrder(ctx, "PAY1")
-	if o.Status != store.StatusPaid || o.TxID != "tx-ok" {
-		t.Fatalf("idempotency broken: %s/%s", o.Status, o.TxID)
-	}
-}
-
-func TestWatcherExpiresOverdue(t *testing.T) {
-	st, _ := store.Open(":memory:")
-	t.Cleanup(func() { _ = st.Close() })
-	ctx := context.Background()
-	now := time.Unix(1_700_000_100, 0)
-
-	newPending(t, st, "OLD", "TADDR2", 1_000000, now.Add(-time.Minute)) // already overdue
-	w := New(st, &mockFetcher{m: map[string][]tron.Transfer{}}, nil)
+	fetch := &mockFetcher{transfers: []tron.Transfer{
+		{TxID: "tx1", To: recvAddr, Value: 5_000017, BlockTs: created.Add(time.Minute).Unix()},
+	}}
+	w := New(st, fetch, recvAddr, nil)
 	w.now = func() time.Time { return now }

 	if err := w.Tick(ctx); err != nil {
 		t.Fatalf("tick: %v", err)
 	}
-	if o, _ := st.GetOrder(ctx, "OLD"); o.Status != store.StatusExpired {
-		t.Fatalf("want expired, got %s", o.Status)
+	o1, _ := st.GetOrder(ctx, "PAY1")
+	if o1.Status != store.StatusPaid || o1.TxID != "tx1" {
+		t.Fatalf("PAY1 %s/%s", o1.Status, o1.TxID)
+	}
+	o2, _ := st.GetOrder(ctx, "PAY2")
+	if o2.Status != store.StatusPending {
+		t.Fatalf("PAY2 should stay pending, got %s", o2.Status)
+	}
+
+	if err := w.Tick(ctx); err != nil { // idempotent
+		t.Fatalf("tick2: %v", err)
+	}
+	o1, _ = st.GetOrder(ctx, "PAY1")
+	if o1.Status != store.StatusPaid || o1.TxID != "tx1" {
+		t.Fatal("idempotency broken")
+	}
+}
+
+func TestWatcherWrongAmountIsOrphan(t *testing.T) {
+	st := memStore(t)
+	ctx := context.Background()
+	now := time.Unix(1_700_000_100, 0)
+	created := now.Add(-5 * time.Minute)
+	seed(t, st, "PAY1", 5_000017, created)
+
+	fetch := &mockFetcher{transfers: []tron.Transfer{
+		{TxID: "tx-wrong", To: recvAddr, Value: 5_000000, BlockTs: created.Add(time.Minute).Unix()},
+	}}
+	w := New(st, fetch, recvAddr, nil)
+	w.now = func() time.Time { return now }
+	_ = w.Tick(ctx)
+
+	o, _ := st.GetOrder(ctx, "PAY1")
+	if o.Status != store.StatusPending {
+		t.Fatalf("PAY1 should stay pending, got %s", o.Status)
+	}
+	if h, _ := st.TxHandled(ctx, "tx-wrong"); !h {
+		t.Fatal("wrong-amount payment should be recorded as orphan")
+	}
+}
+
+func TestWatcherLatePaymentDoesNotMatchNewOrder(t *testing.T) {
+	// Order1 (amount 5_000017) expired; a NEW order (amount 5_000018) is now active
+	// on the SAME address. A late payment of the OLD amount must NOT match the new
+	// order (different amount) -> orphan.
+	st := memStore(t)
+	ctx := context.Background()
+	now := time.Unix(1_700_000_500, 0)
+	seed(t, st, "PAY2", 5_000018, now.Add(-time.Minute))
+
+	fetch := &mockFetcher{transfers: []tron.Transfer{
+		{TxID: "tx-late", To: recvAddr, Value: 5_000017, BlockTs: now.Unix()},
+	}}
+	w := New(st, fetch, recvAddr, nil)
+	w.now = func() time.Time { return now }
+	_ = w.Tick(ctx)
+
+	o2, _ := st.GetOrder(ctx, "PAY2")
+	if o2.Status != store.StatusPending {
+		t.Fatalf("PAY2 must not be matched by a wrong-amount late payment, got %s", o2.Status)
+	}
+	if h, _ := st.TxHandled(ctx, "tx-late"); !h {
+		t.Fatal("late payment should be orphan")
+	}
+}
+
+func TestWatcherIgnoresPaymentBeforeOrder(t *testing.T) {
+	// A payment whose block time is BEFORE the order was created must not match
+	// (guards address reuse: prior balance / old tx).
+	st := memStore(t)
+	ctx := context.Background()
+	now := time.Unix(1_700_000_500, 0)
+	created := now.Add(-2 * time.Minute)
+	seed(t, st, "PAY1", 5_000017, created)
+
+	fetch := &mockFetcher{transfers: []tron.Transfer{
+		{TxID: "tx-old", To: recvAddr, Value: 5_000017, BlockTs: created.Add(-time.Minute).Unix()},
+	}}
+	w := New(st, fetch, recvAddr, nil)
+	w.now = func() time.Time { return now }
+	_ = w.Tick(ctx)
+
+	o, _ := st.GetOrder(ctx, "PAY1")
+	if o.Status != store.StatusPending {
+		t.Fatalf("payment before order must not match, got %s", o.Status)
+	}
+	if h, _ := st.TxHandled(ctx, "tx-old"); !h {
+		t.Fatal("pre-order payment should be orphan")
+	}
+}
+
+func TestWatcherExpires(t *testing.T) {
+	st := memStore(t)
+	ctx := context.Background()
+	now := time.Unix(1_700_000_500, 0)
+	o := &store.Order{
+		OrderNo: "OLD", UserRef: "u", SKU: "pro", ExpectAmount: 1, Address: recvAddr,
+		Status: store.StatusPending, CreatedAt: now.Add(-time.Hour), ExpiresAt: now.Add(-time.Minute),
+	}
+	_ = st.CreateOrder(ctx, o)
+
+	w := New(st, &mockFetcher{}, recvAddr, nil)
+	w.now = func() time.Time { return now }
+	_ = w.Tick(ctx)
+
+	got, _ := st.GetOrder(ctx, "OLD")
+	if got.Status != store.StatusExpired {
+		t.Fatalf("want expired, got %s", got.Status)
 	}
 }