From 727ed7489989db8e6ef6928886710d2f78c47a5e Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 17:52:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20crypto=20=E5=AD=A4=E5=84=BF?= =?UTF-8?q?=E5=88=B0=E8=B4=A6=E5=8F=91=E7=8E=B0=E2=80=94=E2=80=94OrphanSca?= =?UTF-8?q?nner=20=E6=89=AB=E9=93=BE=E6=A0=B8=E5=AF=B9=20+=20orphan=5Fpaym?= =?UTF-8?q?ents=20=E8=90=BD=E8=A1=A8=E5=91=8A=E8=AD=A6(=E5=AF=B9=E8=B4=A6?= =?UTF-8?q?=E5=85=9C=E5=BA=95)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/model/orphan.go | 16 +++++ internal/model/testdb.go | 2 +- internal/provider/crypto/crypto.go | 63 ++++++++++++++++++ internal/provider/crypto/crypto_test.go | 31 +++++++++ internal/provider/provider.go | 30 +++++++++ internal/reconcile/orphan.go | 86 +++++++++++++++++++++++++ internal/reconcile/orphan_test.go | 47 ++++++++++++++ internal/store/order_query.go | 14 ++++ internal/store/order_query_test.go | 25 +++++++ internal/store/orphan.go | 26 ++++++++ internal/store/orphan_test.go | 24 +++++++ main.go | 13 ++-- 12 files changed, 370 insertions(+), 7 deletions(-) create mode 100644 internal/model/orphan.go create mode 100644 internal/reconcile/orphan.go create mode 100644 internal/reconcile/orphan_test.go create mode 100644 internal/store/orphan.go create mode 100644 internal/store/orphan_test.go diff --git a/internal/model/orphan.go b/internal/model/orphan.go new file mode 100644 index 0000000..9efc8fd --- /dev/null +++ b/internal/model/orphan.go @@ -0,0 +1,16 @@ +package model + +import "time" + +// OrphanPayment 到账但不匹配任何 attempt 的转账(对账发现,供人工核对/退款)。 +// tx_id 唯一 → 重复扫描幂等 no-op。canonical pangolin-pay orphan_payments 的 v2 对应物。 +type OrphanPayment struct { + Base + Channel string `gorm:"index;size:32;not null"` + AccountID string `gorm:"index;size:64"` + TxID string `gorm:"uniqueIndex;size:128;not null"` + AmountMinor int64 `gorm:"not null"` + Currency string `gorm:"size:16;not null"` + DetectedAt time.Time + Note string `gorm:"size:255"` +} diff --git a/internal/model/testdb.go b/internal/model/testdb.go index 4910b80..5a95e71 100644 --- a/internal/model/testdb.go +++ b/internal/model/testdb.go @@ -32,7 +32,7 @@ func OpenTestDB(t *testing.T) *gorm.DB { t.Fatalf("open test db: %v", err) } if err := db.AutoMigrate(&OrderV2{}, &Attempt{}, &Account{}, &Refund{}, &WebhookDelivery{}, - &Product{}, &ProductPrice{}); err != nil { + &Product{}, &ProductPrice{}, &OrphanPayment{}); err != nil { t.Fatalf("migrate: %v", err) } if err := UpgradeWebhookDeliveryIndex(db); err != nil { diff --git a/internal/provider/crypto/crypto.go b/internal/provider/crypto/crypto.go index 0576132..606b00f 100644 --- a/internal/provider/crypto/crypto.go +++ b/internal/provider/crypto/crypto.go @@ -240,6 +240,69 @@ func (p *Provider) VerifyCallback(_ context.Context, _ provider.CallbackInput) ( return nil, provider.ErrNotSupported } +// ScanOrphans 扫地址近 Since 的确认到账,金额不在"任一 Known 的期望金额集"内 → 孤儿。 +// 期望金额 = known.AmountMinor + tailFromRef(known.ProviderRef);块时须晚于 Since。 +func (p *Provider) ScanOrphans(ctx context.Context, req provider.OrphanScanRequest) ([]provider.OrphanTransfer, error) { + addr, err := p.address(req.AccountID) + if err != nil { + return nil, err + } + expected := make(map[int64]struct{}, len(req.Known)) + for _, k := range req.Known { + tail, terr := tailFromRef(k.ProviderRef) + if terr != nil { + continue // 无尾数的 ref 跳过(不误判为孤儿依据) + } + expected[k.AmountMinor+tail] = struct{}{} + } + + endpoint := fmt.Sprintf("%s/v1/accounts/%s/transactions/trc20?only_confirmed=true&contract_address=%s&limit=50", + p.baseURL, url.PathEscape(addr), url.QueryEscape(USDTContract)) + httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + if k := p.apiKey(req.AccountID); k != "" { + httpReq.Header.Set("TRON-PRO-API-KEY", k) + } + resp, err := p.http.Do(httpReq) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("crypto: TronGrid HTTP %d: %s", resp.StatusCode, body) + } + var tr trc20Resp + if err := json.Unmarshal(body, &tr); err != nil { + return nil, err + } + + sinceUnix := req.Since.Unix() + var out []provider.OrphanTransfer + for _, d := range tr.Data { + if d.To != addr || d.Type != "Transfer" { + continue + } + blockTs := d.BlockMs / 1000 + if blockTs < sinceUnix { + continue // 窗外旧款不扫(避免把历史正常单反复报孤儿) + } + val, perr := strconv.ParseInt(d.Value, 10, 64) + if perr != nil { + continue + } + if _, ok := expected[val]; ok { + continue // 金额有主(匹配某 attempt 期望额)→ 非孤儿 + } + out = append(out, provider.OrphanTransfer{ + TxID: d.TxID, AmountMinor: val, Currency: "USDT", At: time.Unix(blockTs, 0), + }) + } + return out, nil +} + // trc20Resp 对应 TronGrid /v1/accounts/{addr}/transactions/trc20 响应 // (canonical tron/client.go 同构;contract_address 查询参数已在服务端过滤合约)。 type trc20Resp struct { diff --git a/internal/provider/crypto/crypto_test.go b/internal/provider/crypto/crypto_test.go index d6fdcea..91464ab 100644 --- a/internal/provider/crypto/crypto_test.go +++ b/internal/provider/crypto/crypto_test.go @@ -238,3 +238,34 @@ func TestWarmRebuildsReservationsFromLoader(t *testing.T) { t.Fatalf("超冷却窗的预留不应恢复, got %v", res) } } + +// 孤儿到账发现:假 TronGrid 返回两笔确认到账,一笔匹配 known 期望金额(base+tail), +// 一笔无主 → 仅后者报孤儿。 +func TestScanOrphansFlagsUnmatchedTransfer(t *testing.T) { + const addr = "TOrphanScanAddr00000000000000000000" + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + // 假 TronGrid:to=addr 两笔确认到账。29990263 匹配 known(base 29990000 + tail 263);88880000 无主。 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"data":[ + {"transaction_id":"TX-MATCH","to":"` + addr + `","type":"Transfer","value":"29990263","block_timestamp":` + strconv.FormatInt(now.Add(-5*time.Minute).UnixMilli(), 10) + `}, + {"transaction_id":"TX-ORPHAN","to":"` + addr + `","type":"Transfer","value":"88880000","block_timestamp":` + strconv.FormatInt(now.Add(-3*time.Minute).UnixMilli(), 10) + `} + ]}`)) + })) + defer ts.Close() + + t.Setenv("CRY_ADDRESS", addr) + reg := accounts.New([]config.AccountConfig{{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"}}) + p := crypto.New(reg, crypto.WithBaseURL(ts.URL), crypto.WithHTTPClient(ts.Client()), crypto.WithNow(func() time.Time { return now })) + + orphans, err := p.ScanOrphans(context.Background(), provider.OrphanScanRequest{ + AccountID: "cry-1", Since: now.Add(-time.Hour), + Known: []provider.KnownAttempt{{AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-A-263"}}, // 期望 29990263 + }) + if err != nil { + t.Fatalf("scan: %v", err) + } + if len(orphans) != 1 || orphans[0].TxID != "TX-ORPHAN" || orphans[0].AmountMinor != 88880000 { + t.Fatalf("只应报 1 笔孤儿 TX-ORPHAN, got %+v", orphans) + } +} diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 5e27b94..e2d2b0e 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -123,6 +123,36 @@ type RecurringProvider interface { CancelAgreement(ctx context.Context, agreementRef string) error } +// ---- 对账:孤儿到账扫描(P6,可选接口)---- + +// KnownAttempt 是 pay 合法签发过的一笔尝试的对账维度:期望金额 = base(AmountMinor)+ 渠道尾数 +// (尾数封在 provider_ref,由渠道自解,pay 不算)。渠道据此判断一笔到账是否"有主"。 +type KnownAttempt struct { + AmountMinor int64 + ProviderRef string +} + +// OrphanScanRequest 扫描某账户 Since 以来、不匹配任何 Known 的到账。 +type OrphanScanRequest struct { + AccountID string + Since time.Time + Known []KnownAttempt +} + +// OrphanTransfer 一笔"有钱到账但无主"的转账(付错金额/手动转/超窗迟到旧款)。 +type OrphanTransfer struct { + TxID string + AmountMinor int64 + Currency string + At time.Time +} + +// OrphanScanner 自托管渠道(crypto)可选实现:发现到账但不匹配任何 attempt 的转账。 +// 网关侧渠道(alipay/stripe)以对账单核对,不实现此接口。 +type OrphanScanner interface { + ScanOrphans(ctx context.Context, req OrphanScanRequest) ([]OrphanTransfer, error) +} + // Registry — 方法名 → Provider(设计 §2 Provider adapter 注册表)。启动期注册,运行期只读。 type Registry struct{ providers map[string]Provider } diff --git a/internal/reconcile/orphan.go b/internal/reconcile/orphan.go new file mode 100644 index 0000000..74cef08 --- /dev/null +++ b/internal/reconcile/orphan.go @@ -0,0 +1,86 @@ +package reconcile + +import ( + "context" + "log" + "time" + + "github.com/wangjia/pay/config" + "github.com/wangjia/pay/internal/accounts" + "github.com/wangjia/pay/internal/model" + "github.com/wangjia/pay/internal/provider" + "github.com/wangjia/pay/internal/provider/crypto" + "github.com/wangjia/pay/internal/store" +) + +// OrphanScanTask 对每个 enabled crypto 账户扫链找孤儿到账(不匹配任何近期 attempt), +// 落 OrphanStore + 首次记入时告警。渠道须实现 provider.OrphanScanner(crypto 实现)。 +func OrphanScanTask(providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore, + orphans *store.OrphanStore, window time.Duration, now func() time.Time) func(ctx context.Context) error { + return func(ctx context.Context) error { + prov, err := providers.Get("crypto") + if err != nil { + return nil // 未启用 crypto:无事可做 + } + scanner, ok := prov.(provider.OrphanScanner) + if !ok { + return nil + } + since := now().Add(-window) + for _, acc := range accts.EnabledFor("crypto", "") { + atts, err := orders.ListAttemptsByChannelSince("crypto", since, 200) + if err != nil { + return err + } + known := make([]provider.KnownAttempt, 0, len(atts)) + for i := range atts { + if atts[i].AccountID != acc.AccountID { + continue + } + known = append(known, provider.KnownAttempt{AmountMinor: atts[i].AmountMinor, ProviderRef: atts[i].ProviderRef}) + } + found, err := scanner.ScanOrphans(ctx, provider.OrphanScanRequest{ + AccountID: acc.AccountID, Since: since, Known: known, + }) + if err != nil { + log.Printf("[reconcile] 孤儿扫描 account=%s: %v", acc.AccountID, err) + continue // 单账户失败不阻断其它 + } + for _, o := range found { + isNew, rerr := orphans.Record(&model.OrphanPayment{ + Channel: "crypto", AccountID: acc.AccountID, TxID: o.TxID, + AmountMinor: o.AmountMinor, Currency: o.Currency, DetectedAt: o.At, + Note: "到账无主:不匹配任何近期 attempt 期望金额", + }) + if rerr != nil { + log.Printf("[reconcile] 记录孤儿失败 tx=%s: %v", o.TxID, rerr) + continue + } + if isNew { + log.Printf("[reconcile][孤儿告警] channel=crypto account=%s tx=%s amount=%d %s", + acc.AccountID, o.TxID, o.AmountMinor, o.Currency) + } + } + } + return nil + } +} + +// AddCryptoJobs 装配 crypto 相关后台任务:① 冷启动 Warm(注入 loader 后立即重建预留); +// ② 周期孤儿扫描。若未启用 crypto 渠道则安全跳过。 +func AddCryptoJobs(runner *Runner, providers *provider.Registry, accts *accounts.Registry, + orders *store.OrderStore, orphans *store.OrphanStore, cfg config.ReconcileConfig) { + prov, err := providers.Get("crypto") + if err != nil { + return + } + if cp, ok := prov.(*crypto.Provider); ok { + cp.SetReservationLoader(CryptoReservationLoader(orders)) + if werr := cp.Warm(context.Background()); werr != nil { // 起服务前重建预留 + log.Printf("[reconcile] crypto 预留冷启动重建: %v", werr) + } + } + runner.Add("crypto-orphan-scan", time.Duration(cfg.OrphanEverySec)*time.Second, + OrphanScanTask(providers, accts, orders, orphans, + time.Duration(cfg.OrphanWindowMin)*time.Minute, time.Now)) +} diff --git a/internal/reconcile/orphan_test.go b/internal/reconcile/orphan_test.go new file mode 100644 index 0000000..c17b01a --- /dev/null +++ b/internal/reconcile/orphan_test.go @@ -0,0 +1,47 @@ +package reconcile_test + +import ( + "context" + "net/http" + "net/http/httptest" + "strconv" + "testing" + "time" + + "github.com/wangjia/pay/config" + "github.com/wangjia/pay/internal/accounts" + "github.com/wangjia/pay/internal/model" + "github.com/wangjia/pay/internal/provider" + "github.com/wangjia/pay/internal/provider/crypto" + "github.com/wangjia/pay/internal/reconcile" + "github.com/wangjia/pay/internal/store" +) + +func TestOrphanScanTaskRecordsUnmatched(t *testing.T) { + const addr = "TOrphanJobAddr0000000000000000000000" + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"data":[ + {"transaction_id":"TX-ORPHAN","to":"` + addr + `","type":"Transfer","value":"88880000","block_timestamp":` + strconv.FormatInt(now.Add(-3*time.Minute).UnixMilli(), 10) + `} + ]}`)) + })) + defer ts.Close() + + t.Setenv("CRY_ADDRESS", addr) + db := model.OpenTestDB(t) + orders := store.NewOrderStore(db) + orphanStore := store.NewOrphanStore(db) + acctReg := accounts.New([]config.AccountConfig{{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"}}) + preg := provider.NewRegistry() + preg.Register(crypto.New(acctReg, crypto.WithBaseURL(ts.URL), crypto.WithHTTPClient(ts.Client()), crypto.WithNow(func() time.Time { return now }))) + + task := reconcile.OrphanScanTask(preg, acctReg, orders, orphanStore, time.Hour, func() time.Time { return now }) + if err := task(context.Background()); err != nil { + t.Fatalf("task: %v", err) + } + var cnt int64 + db.Model(&model.OrphanPayment{}).Where("tx_id = ?", "TX-ORPHAN").Count(&cnt) + if cnt != 1 { + t.Fatalf("应落 1 条孤儿, got %d", cnt) + } +} diff --git a/internal/store/order_query.go b/internal/store/order_query.go index 1f07bae..1c08002 100644 --- a/internal/store/order_query.go +++ b/internal/store/order_query.go @@ -174,3 +174,17 @@ func (s *OrderStore) ListOrdersByStatus(statuses []model.OrderStatusV2, limit in } return out, nil } + +// ListAttemptsByChannelSince 列某渠道 created_at>=since 的 attempt(任意状态), +// 供 orphan 扫描构造"已知期望金额集"(凡 pay 合法签发过的金额都不算孤儿)。 +func (s *OrderStore) ListAttemptsByChannelSince(channel string, since time.Time, limit int) ([]model.Attempt, error) { + if limit <= 0 || limit > 500 { + limit = 200 + } + var out []model.Attempt + if err := s.db.Where("channel = ? AND created_at >= ?", channel, since). + Order("id DESC").Limit(limit).Find(&out).Error; err != nil { + return nil, fmt.Errorf("store.ListAttemptsByChannelSince: %w", err) + } + return out, nil +} diff --git a/internal/store/order_query_test.go b/internal/store/order_query_test.go index 968aa50..254ecc3 100644 --- a/internal/store/order_query_test.go +++ b/internal/store/order_query_test.go @@ -176,3 +176,28 @@ func TestListOrdersByStatus(t *testing.T) { t.Fatalf("命中集合不对: %+v", got) } } + +// TestListAttemptsByChannelSince 供 orphan 扫描构造"已知期望金额集":同渠道、 +// created_at>=since、任意状态的 attempt(pending/paid/expired 都算——凡 pay +// 合法签发过的金额都不算孤儿)。 +func TestListAttemptsByChannelSince(t *testing.T) { + db := model.OpenTestDB(t) + s := store.NewOrderStore(db) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + mk := func(no, ch string, ago time.Duration) { + _ = s.CreateAttempt(&model.Attempt{OutTradeNo: no, Channel: ch, ProviderRef: "R-" + no, + AmountMinor: 100, Currency: "USDT", Status: model.AttemptPending}) + _ = db.Model(&model.Attempt{}).Where("out_trade_no = ?", no).Update("created_at", now.Add(-ago)).Error + } + mk("C1", "crypto", 10*time.Minute) + mk("C2", "crypto", 5*time.Hour) // 太旧 + mk("A1", "alipay", 1*time.Minute) + + got, err := s.ListAttemptsByChannelSince("crypto", now.Add(-time.Hour), 100) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 || got[0].OutTradeNo != "C1" { + t.Fatalf("只应含近期 crypto, got %+v", got) + } +} diff --git a/internal/store/orphan.go b/internal/store/orphan.go new file mode 100644 index 0000000..72a56ab --- /dev/null +++ b/internal/store/orphan.go @@ -0,0 +1,26 @@ +package store + +import ( + "fmt" + + "gorm.io/gorm" + "gorm.io/gorm/clause" + + "github.com/wangjia/pay/internal/model" +) + +type OrphanStore struct{ db *gorm.DB } + +func NewOrphanStore(db *gorm.DB) *OrphanStore { return &OrphanStore{db: db} } + +// Record 幂等落一条孤儿(tx_id 冲突 no-op)。返回是否新记(供告警只喊一次)。 +func (s *OrphanStore) Record(o *model.OrphanPayment) (bool, error) { + res := s.db.Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "tx_id"}}, + DoNothing: true, + }).Create(o) + if res.Error != nil { + return false, fmt.Errorf("store.OrphanStore.Record: %w", res.Error) + } + return res.RowsAffected > 0, nil +} diff --git a/internal/store/orphan_test.go b/internal/store/orphan_test.go new file mode 100644 index 0000000..a2b4cbd --- /dev/null +++ b/internal/store/orphan_test.go @@ -0,0 +1,24 @@ +package store_test + +import ( + "testing" + "time" + + "github.com/wangjia/pay/internal/model" + "github.com/wangjia/pay/internal/store" +) + +func TestOrphanStoreRecordIdempotent(t *testing.T) { + os := store.NewOrphanStore(model.OpenTestDB(t)) + o := &model.OrphanPayment{Channel: "crypto", AccountID: "cry-1", TxID: "TX-1", + AmountMinor: 12345, Currency: "USDT", DetectedAt: time.Now()} + first, err := os.Record(o) + if err != nil || !first { + t.Fatalf("首次应记入, first=%v err=%v", first, err) + } + again, err := os.Record(&model.OrphanPayment{Channel: "crypto", AccountID: "cry-1", TxID: "TX-1", + AmountMinor: 12345, Currency: "USDT", DetectedAt: time.Now()}) + if err != nil || again { + t.Fatalf("同 tx_id 应幂等 no-op, again=%v err=%v", again, err) + } +} diff --git a/main.go b/main.go index 2cd8cdc..89864bd 100644 --- a/main.go +++ b/main.go @@ -62,8 +62,8 @@ func main() { router.SetupV2(r, gw) // P6 后台守护 / 对账:订单过期清理 + 用量刷新 + 查单对账收敛 + 已付抽查 + - // 退款修复扫描/卡滞告警(P4 T3 review 追加义务,归到本任务一起装配)。 - // crypto 预留冷启动 Warm + 孤儿扫描留给 Task 6 追加(AddCryptoJobs)。 + // 退款修复扫描/卡滞告警(P4 T3 review 追加义务,归到本任务一起装配)+ + // crypto 预留冷启动 Warm + 孤儿扫描(Task 6,AddCryptoJobs)。 if config.C.Reconcile.Enabled { rc := config.C.Reconcile refundStore := store.NewRefundStore(db) @@ -80,12 +80,13 @@ func main() { reconcile.RefundApplyTask(orderStore, refundStore, 200)) runner.Add("refund-stuck-alert", time.Duration(rc.RefundApplyEverySec)*time.Second, reconcile.RefundStuckAlertTask(refundStore, time.Duration(rc.RefundStuckWarnMin)*time.Minute, time.Now)) - // crypto 孤儿扫描(Task 6)在此追加:reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, rc) + orphanStore := store.NewOrphanStore(db) + reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc) // crypto 预留冷启动 Warm + 孤儿扫描(Task 6) ctx := context.Background() - runner.RunOnce(ctx) // 启动预热:先跑一遍(usage 快照/过期清理/退款自愈立即生效) + runner.RunOnce(ctx) // 启动预热:先跑一遍(usage 快照/过期清理/退款自愈/crypto Warm 立即生效) runner.Start(ctx) - log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/退款修复扫描/退款卡滞告警)") + log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/退款修复扫描/退款卡滞告警/crypto 孤儿扫描)") } if config.C.QuerySync.Enabled { @@ -150,7 +151,7 @@ func autoMigrate(db *gorm.DB) { &model.Order{}, &model.NotifyLog{}, &model.BizNotifyLog{}, - &model.OrderV2{}, &model.Attempt{}, &model.Account{}, &model.Refund{}, &model.WebhookDelivery{}, // v2 + &model.OrderV2{}, &model.Attempt{}, &model.Account{}, &model.Refund{}, &model.WebhookDelivery{}, &model.OrphanPayment{}, // v2 ); err != nil { log.Fatalf("自动迁移失败: %v", err) }