feat(v2): crypto 孤儿到账发现——OrphanScanner 扫链核对 + orphan_payments 落表告警(对账兜底)

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))
+}

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)
+	}
+}