feat(pay): 收款引擎 —— 建单/派生地址 + TronGrid watcher 侦测到账(#34/34A Phase B.3+C)

- store(SQLite,modernc 纯 Go):pay_orders + addr_cursor(HD 派生游标,地址不复用); 建单/查单/ListPending/MarkPaid(幂等,仅 pending→paid)/MarkExpired。 - pay 服务:CreateOrder 每单 NextAddrIndex→从 xpub watch-only 派生唯一收款地址→写 pending 单(TTL 15min)。 - tron:TronGrid 客户端读已确认 TRC20 到账(only_confirmed + USDT 合约,micro-USDT 整数)。 - watcher:Tick 先过期逾期单,再对每个 pending 单查到账、金额≥期望→MarkPaid;幂等(同 tx 只认一次)、网络错误跳过下轮重试;Loop 定时轮询。 - httpapi:POST /order、GET /order/{orderNo}、/healthz;cmd/paywatch 用 env 装配 + 优雅退出。 - 测试:store/service/watcher(mock TronGrid)/httpapi 全绿——建单派生地址正确、到账侦测、欠额不认、幂等、超时过期、404/400。热服务无私钥。 - README:安全模型 + Phase A 离线备钱包步骤 + 运行/API/测试。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-09 02:57:47 +08:00
parent 4393edf1d7
commit 4bb92209ca
13 changed files with 1007 additions and 0 deletions
@@ -0,0 +1,83 @@
+// 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
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/wangjia/pangolin/pay/internal/store"
+	"github.com/wangjia/pangolin/pay/internal/tron"
+)
+
+type Watcher struct {
+	st   *store.Store
+	tron tron.Fetcher
+	log  *slog.Logger
+	now  func() time.Time
+}
+
+func New(st *store.Store, f tron.Fetcher, log *slog.Logger) *Watcher {
+	if log == nil {
+		log = slog.Default()
+	}
+	return &Watcher{st: st, tron: f, 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).
+func (w *Watcher) Tick(ctx context.Context) error {
+	if n, err := w.st.MarkExpired(ctx, w.now()); err != nil {
+		return err
+	} else if n > 0 {
+		w.log.Info("orders expired", "count", n)
+	}
+
+	pending, err := w.st.ListPending(ctx)
+	if err != nil {
+		return err
+	}
+	for _, o := range pending {
+		transfers, err := w.tron.IncomingTransfers(ctx, o.Address)
+		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)
+			continue
+		}
+		for _, t := range transfers {
+			if t.Value < o.ExpectAmount {
+				continue
+			}
+			ok, err := w.st.MarkPaid(ctx, o.OrderNo, t.TxID)
+			if err != nil {
+				w.log.Error("mark paid", "order", o.OrderNo, "err", err)
+				break
+			}
+			if ok {
+				w.log.Info("order paid", "order", o.OrderNo, "tx", t.TxID, "value", t.Value, "address", o.Address)
+			}
+			break
+		}
+	}
+	return nil
+}
+
+// Loop runs Tick every interval until ctx is cancelled.
+func (w *Watcher) Loop(ctx context.Context, interval time.Duration) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			if err := w.Tick(ctx); err != nil {
+				w.log.Error("watcher tick", "err", err)
+			}
+		}
+	}
+}
@@ -0,0 +1,89 @@
+package watcher
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/wangjia/pangolin/pay/internal/store"
+	"github.com/wangjia/pangolin/pay/internal/tron"
+)
+
+type mockFetcher struct{ m map[string][]tron.Transfer }
+
+func (f *mockFetcher) IncomingTransfers(_ context.Context, addr string) ([]tron.Transfer, error) {
+	return f.m[addr], nil
+}
+
+func newPending(t *testing.T, st *store.Store, orderNo, addr string, amount int64, expires 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,
+	}
+	if err := st.CreateOrder(context.Background(), o); err != nil {
+		t.Fatalf("seed order: %v", err)
+	}
+}
+
+func TestWatcherMarksPaidOnSufficientTransfer(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, "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)
+	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)
+	}
+}