wangjia
87 lines
3.1 KiB
Go
87 lines
3.1 KiB
Go
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))
|
|
}
|