Files
pangolin/pay/internal/store/store.go
T
wangjia 4bb92209ca 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

173 lines
4.8 KiB
Go

// Package store persists pay orders + the HD address-derivation cursor in
// SQLite (pure-Go modernc driver, no CGO — same choice as the control plane).
package store
import (
"context"
"database/sql"
"errors"
"time"
_ "modernc.org/sqlite"
)
type Status string
const (
StatusPending Status = "pending"
StatusPaid Status = "paid"
StatusExpired Status = "expired"
)
// Order is one payment request. Amounts are in micro-USDT (1e-6), matching the
// raw integer value of a TRC20 USDT transfer (USDT has 6 decimals).
type Order struct {
OrderNo string
SKU string
ExpectAmount int64
AddrIndex uint32
Address string
Status Status
TxID string
CreatedAt time.Time
ExpiresAt time.Time
}
var ErrNotFound = errors.New("store: order not found")
type Store struct{ db *sql.DB }
func Open(dsn string) (*Store, error) {
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1) // SQLite: serialize writers, avoid "database is locked"
s := &Store{db: db}
if err := s.migrate(); err != nil {
_ = db.Close()
return nil, err
}
return s, nil
}
func (s *Store) Close() error { return s.db.Close() }
func (s *Store) migrate() error {
stmts := []string{
`CREATE TABLE IF NOT EXISTS pay_orders(
order_no TEXT PRIMARY KEY,
sku TEXT NOT NULL,
expect_amount INTEGER NOT NULL,
addr_index INTEGER NOT NULL,
address TEXT NOT NULL,
status TEXT NOT NULL,
tx_id TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL
)`,
`CREATE INDEX IF NOT EXISTS idx_orders_status ON pay_orders(status)`,
`CREATE TABLE IF NOT EXISTS addr_cursor(id INTEGER PRIMARY KEY CHECK(id=1), next_index INTEGER NOT NULL)`,
`INSERT OR IGNORE INTO addr_cursor(id, next_index) VALUES(1, 0)`,
}
for _, q := range stmts {
if _, err := s.db.Exec(q); err != nil {
return err
}
}
return nil
}
// NextAddrIndex atomically returns the current HD index and advances the cursor.
// Addresses are never reused (avoids an old payment landing on a recycled slot).
func (s *Store) NextAddrIndex(ctx context.Context) (uint32, error) {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback() }()
var idx uint32
if err := tx.QueryRowContext(ctx, `SELECT next_index FROM addr_cursor WHERE id=1`).Scan(&idx); err != nil {
return 0, err
}
if _, err := tx.ExecContext(ctx, `UPDATE addr_cursor SET next_index=? WHERE id=1`, idx+1); err != nil {
return 0, err
}
if err := tx.Commit(); err != nil {
return 0, err
}
return idx, nil
}
func (s *Store) CreateOrder(ctx context.Context, o *Order) error {
_, err := s.db.ExecContext(ctx,
`INSERT INTO pay_orders(order_no,sku,expect_amount,addr_index,address,status,created_at,expires_at)
VALUES(?,?,?,?,?,?,?,?)`,
o.OrderNo, o.SKU, o.ExpectAmount, o.AddrIndex, o.Address, o.Status, o.CreatedAt.Unix(), o.ExpiresAt.Unix())
return err
}
const cols = `order_no,sku,expect_amount,addr_index,address,status,tx_id,created_at,expires_at`
func scanOrder(sc interface{ Scan(...any) error }) (*Order, error) {
o := &Order{}
var created, expires int64
if err := sc.Scan(&o.OrderNo, &o.SKU, &o.ExpectAmount, &o.AddrIndex, &o.Address, &o.Status, &o.TxID, &created, &expires); err != nil {
return nil, err
}
o.CreatedAt = time.Unix(created, 0)
o.ExpiresAt = time.Unix(expires, 0)
return o, nil
}
func (s *Store) GetOrder(ctx context.Context, orderNo string) (*Order, error) {
row := s.db.QueryRowContext(ctx, `SELECT `+cols+` FROM pay_orders WHERE order_no=?`, orderNo)
o, err := scanOrder(row)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNotFound
}
return o, err
}
func (s *Store) ListPending(ctx context.Context) ([]*Order, error) {
rows, err := s.db.QueryContext(ctx, `SELECT `+cols+` FROM pay_orders WHERE status=?`, StatusPending)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
var out []*Order
for rows.Next() {
o, err := scanOrder(rows)
if err != nil {
return nil, err
}
out = append(out, o)
}
return out, rows.Err()
}
// MarkPaid transitions pending->paid, idempotently (only affects a still-pending
// row). Returns true if this call was the one that flipped it.
func (s *Store) MarkPaid(ctx context.Context, orderNo, txID string) (bool, error) {
res, err := s.db.ExecContext(ctx,
`UPDATE pay_orders SET status=?, tx_id=? WHERE order_no=? AND status=?`,
StatusPaid, txID, orderNo, StatusPending)
if err != nil {
return false, err
}
n, _ := res.RowsAffected()
return n > 0, nil
}
// MarkExpired flips pending->expired for orders past their deadline.
func (s *Store) MarkExpired(ctx context.Context, now time.Time) (int64, error) {
res, err := s.db.ExecContext(ctx,
`UPDATE pay_orders SET status=? WHERE status=? AND expires_at < ?`,
StatusExpired, StatusPending, now.Unix())
if err != nil {
return 0, err
}
n, _ := res.RowsAffected()
return n, nil
}