merge: P6 对账守护并入(七job装配/退款自愈/孤儿发现/Notifier硬化/真实用量源)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u # Conflicts: # main.go
This commit is contained in:
wangjia
@@ -14,6 +14,7 @@ type Config struct {
|
||||
AlipaySandbox AlipaySandboxConfig `mapstructure:"alipay_sandbox"`
|
||||
Wechat WechatConfig `mapstructure:"wechat"`
|
||||
QuerySync QuerySyncConfig `mapstructure:"query_sync"`
|
||||
Reconcile ReconcileConfig `mapstructure:"reconcile"`
|
||||
// Biz 通用：任意业务系统（jiu / dudu / …）在 config 的 biz.<name> 下声明即可接入，无需改代码。
|
||||
Biz map[string]BizSystemConfig `mapstructure:"biz"`
|
||||
// Accounts 多账户配置注册表（v2）：凭证不写死配置，只存 env 前缀，真值运行时从环境变量取。
|
||||
@@ -98,6 +99,27 @@ type QuerySyncConfig struct {
|
||||
MaxAgeMin int `mapstructure:"max_age_min"` // 只查创建时间在该分钟数内的待支付单
|
||||
}
|
||||
|
||||
// ReconcileConfig v2 后台守护/对账周期任务开关与间隔。
|
||||
type ReconcileConfig struct {
|
||||
Enabled bool `mapstructure:"enabled"`
|
||||
OrderTTLMin int `mapstructure:"order_ttl_min"` // pending 订单存活 TTL(分钟),超则关闭;须 ≫ crypto 支付窗 orderTTL(15min,internal/provider/crypto/crypto.go)——否则在途 crypto 支付可能被订单级过期误杀
|
||||
ExpireEverySec int `mapstructure:"expire_every_sec"` // 过期清理间隔
|
||||
SyncEverySec int `mapstructure:"sync_every_sec"` // 查单对账间隔
|
||||
UsageEverySec int `mapstructure:"usage_every_sec"` // 用量快照刷新间隔
|
||||
SpotCheckEverySec int `mapstructure:"spot_check_every_sec"` // 已付抽查间隔
|
||||
SpotCheckWindowMin int `mapstructure:"spot_check_window_min"` // 抽查回溯窗(分钟)
|
||||
OrphanEverySec int `mapstructure:"orphan_every_sec"` // crypto 孤儿扫描间隔(Task 6)
|
||||
OrphanWindowMin int `mapstructure:"orphan_window_min"` // 孤儿扫描回溯窗(Task 6)
|
||||
// —— 以下两项为 P4 T3 opus review 追加义务(退款修复扫描 + 卡滞退款告警),
|
||||
// 本期归到本任务(对账主体)一起装配,brief 原表未列。
|
||||
RefundApplyEverySec int `mapstructure:"refund_apply_every_sec"` // 退款修复扫描 + 卡滞告警 共用间隔
|
||||
RefundStuckWarnMin int `mapstructure:"refund_stuck_warn_min"` // 退款卡滞 processing/manual_pending 告警阈值(分钟)
|
||||
// RefundApplyLookbackMin 退款修复扫描的候选回溯窗(分钟):只扫 completed_at/updated_at
|
||||
// 落在 [now-lookback, now] 内的候选,防 LIMIT 尾部饥饿(见 store.ListDistinctOutTradeNosByStatusSince
|
||||
// / ListOrdersByStatusSince)。默认 48 小时——自愈针对近期崩溃窗口,历史一致性由已收敛状态保证。
|
||||
RefundApplyLookbackMin int `mapstructure:"refund_apply_lookback_min"`
|
||||
}
|
||||
|
||||
var C Config
|
||||
|
||||
func Load() {
|
||||
@@ -130,6 +152,18 @@ func Load() {
|
||||
viper.SetDefault("query_sync.enabled", true)
|
||||
viper.SetDefault("query_sync.interval_sec", 30)
|
||||
viper.SetDefault("query_sync.max_age_min", 30)
|
||||
viper.SetDefault("reconcile.enabled", true)
|
||||
viper.SetDefault("reconcile.order_ttl_min", 60)
|
||||
viper.SetDefault("reconcile.expire_every_sec", 300)
|
||||
viper.SetDefault("reconcile.sync_every_sec", 30)
|
||||
viper.SetDefault("reconcile.usage_every_sec", 60)
|
||||
viper.SetDefault("reconcile.spot_check_every_sec", 300)
|
||||
viper.SetDefault("reconcile.spot_check_window_min", 180)
|
||||
viper.SetDefault("reconcile.orphan_every_sec", 300)
|
||||
viper.SetDefault("reconcile.orphan_window_min", 180)
|
||||
viper.SetDefault("reconcile.refund_apply_every_sec", 300)
|
||||
viper.SetDefault("reconcile.refund_stuck_warn_min", 30)
|
||||
viper.SetDefault("reconcile.refund_apply_lookback_min", 48*60)
|
||||
|
||||
if err := viper.ReadInConfig(); err != nil {
|
||||
log.Println("[config] 未找到 config.yaml，使用默认值 + 环境变量")
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
package model
|
||||
|
||||
import "time"
|
||||
|
||||
// WebhookDelivery 是 pay→业务方 webhook 的 outbox(v2)。unique(out_trade_no,event_type,refund_id)
|
||||
// 保证同一订单同一事件(同一退款单)只入队一次(幂等);后台 Notifier 扫 Delivered=false 重试兜底。
|
||||
type WebhookDelivery struct {
|
||||
@@ -12,4 +14,7 @@ type WebhookDelivery struct {
|
||||
Delivered bool `gorm:"index;default:false" json:"delivered"`
|
||||
Attempts int `json:"attempts"`
|
||||
LastError string `gorm:"size:255" json:"last_error,omitempty"`
|
||||
|
||||
Dead bool `gorm:"index;default:false" json:"dead"` // 达最大次数放弃投递(死信),需人工/对账介入
|
||||
NextAttemptAt *time.Time `gorm:"index" json:"next_attempt_at,omitempty"` // 指数退避的下次可投时刻;nil=立即可投
|
||||
}
|
||||
|
||||
@@ -49,15 +49,33 @@ type Provider struct {
|
||||
baseURL string
|
||||
http *http.Client
|
||||
now func() time.Time
|
||||
loader ReservationLoader // 冷启动预留重建源(装配期注入,nil=不重建)
|
||||
|
||||
mu sync.Mutex
|
||||
reserved map[string]time.Time // "<address>/<amount>" → 预留到期(链上匹配维度,对齐 Query 的 to==addr)(canonical AmountRecentlyUsed 的进程内等价)
|
||||
}
|
||||
|
||||
// PendingReservation 冷启动重建一笔预留所需的最小信息(中性结构,crypto 不依赖 store)。
|
||||
type PendingReservation struct {
|
||||
AccountID string // 收款账户(用于解析地址,链上匹配维度)
|
||||
AmountMinor int64 // attempt 冻结的 base 金额(不含尾数)
|
||||
ProviderRef string // "CRYPTO-<OutTradeNo>-<tail>",用于恢复尾数
|
||||
ReservedAt time.Time // 建单时间(= attempt.CreatedAt),冷却窗自此算
|
||||
}
|
||||
|
||||
// ReservationLoader 返回当前仍活跃(pending)的 crypto 预留。装配期由 main 用 OrderStore 实现。
|
||||
type ReservationLoader func(ctx context.Context) ([]PendingReservation, error)
|
||||
|
||||
type Option func(*Provider)
|
||||
|
||||
func WithBaseURL(u string) Option { return func(p *Provider) { p.baseURL = u } }
|
||||
func WithHTTPClient(c *http.Client) Option { return func(p *Provider) { p.http = c } }
|
||||
func WithBaseURL(u string) Option { return func(p *Provider) { p.baseURL = u } }
|
||||
func WithHTTPClient(c *http.Client) Option { return func(p *Provider) { p.http = c } }
|
||||
func WithReservationLoader(l ReservationLoader) Option { return func(p *Provider) { p.loader = l } }
|
||||
func WithNow(f func() time.Time) Option { return func(p *Provider) { p.now = f } }
|
||||
|
||||
// SetReservationLoader 构造后注入冷启动预留源(装配期 main 在 BuildRegistry 之后调用:
|
||||
// loader 依赖 OrderStore,而注册表构造不便传 store)。非并发安全,仅启动期单线程调用。
|
||||
func (p *Provider) SetReservationLoader(l ReservationLoader) { p.loader = l }
|
||||
|
||||
func New(accts *accounts.Registry, opts ...Option) *Provider {
|
||||
p := &Provider{
|
||||
@@ -151,6 +169,40 @@ func tailFromRef(ref string) (int64, error) {
|
||||
return strconv.ParseInt(ref[i+1:], 10, 64)
|
||||
}
|
||||
|
||||
// Warm 冷启动兜底:把仍在冷却窗内的活跃预留灌回内存表,兜住重启丢 map 导致的金额复用误配。
|
||||
// 幂等:只加不覆盖更早到期时间;冷却已过的跳过。装配期在起服务前调一次即可。
|
||||
func (p *Provider) Warm(ctx context.Context) error {
|
||||
if p.loader == nil {
|
||||
return nil
|
||||
}
|
||||
items, err := p.loader(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
now := p.now()
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for _, it := range items {
|
||||
addr, err := p.address(it.AccountID) // 地址是链上匹配维度真相源
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
tail, err := tailFromRef(it.ProviderRef)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
until := it.ReservedAt.Add(amountCooldown)
|
||||
if !until.After(now) {
|
||||
continue // 冷却已过,金额可安全复用,无需恢复
|
||||
}
|
||||
key := addr + "/" + strconv.FormatInt(it.AmountMinor+tail, 10)
|
||||
if cur, ok := p.reserved[key]; !ok || until.After(cur) {
|
||||
p.reserved[key] = until
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (*provider.Session, error) {
|
||||
if req.Currency != "USDT" {
|
||||
return nil, fmt.Errorf("crypto: 仅支持 USDT, got %s", req.Currency)
|
||||
@@ -188,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 {
|
||||
|
||||
@@ -206,3 +206,66 @@ func TestReservationKeyByAddress(t *testing.T) {
|
||||
t.Fatalf("共享地址的两个账户不应分配相同金额: %d", amt1)
|
||||
}
|
||||
}
|
||||
|
||||
// 冷启动兜底:装配期注入的 ReservationLoader 在 Warm 时把仍在冷却窗内的 pending
|
||||
// 预留灌回内存表;超冷却窗的旧预留(迟到旧款已不可能匹配)不必恢复。
|
||||
func TestWarmRebuildsReservationsFromLoader(t *testing.T) {
|
||||
const addr = "TWarmTestAddr000000000000000000000"
|
||||
t.Setenv("CRY_ADDRESS", addr)
|
||||
reg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"},
|
||||
})
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
// 两条 pending:一条在冷却窗内(应恢复),一条建单于 40min 前(> 30min 冷却窗,应跳过)。
|
||||
loader := func(context.Context) ([]crypto.PendingReservation, error) {
|
||||
return []crypto.PendingReservation{
|
||||
{AccountID: "cry-1", AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-A-263", ReservedAt: now.Add(-5 * time.Minute)},
|
||||
{AccountID: "cry-1", AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-B-777", ReservedAt: now.Add(-40 * time.Minute)},
|
||||
}, nil
|
||||
}
|
||||
p := crypto.New(reg, crypto.WithReservationLoader(loader), crypto.WithNow(func() time.Time { return now }))
|
||||
if err := p.Warm(context.Background()); err != nil {
|
||||
t.Fatalf("warm: %v", err)
|
||||
}
|
||||
res := p.GetReserved()
|
||||
inWindow := addr + "/" + strconv.FormatInt(29990000+263, 10)
|
||||
expired := addr + "/" + strconv.FormatInt(29990000+777, 10)
|
||||
if _, ok := res[inWindow]; !ok {
|
||||
t.Fatalf("冷却窗内的预留应恢复, got %v", res)
|
||||
}
|
||||
if _, ok := res[expired]; ok {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -133,6 +133,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 }
|
||||
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/provider/crypto"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// CryptoReservationLoader 装配 crypto 冷启动预留源:读全部 pending attempt,
|
||||
// 过滤 channel=crypto,映射为 crypto.PendingReservation。ReservedAt 取 attempt.CreatedAt
|
||||
// (建单时刻,冷却窗自此算)。crypto 不 import store,故此桥在装配层。
|
||||
func CryptoReservationLoader(orders *store.OrderStore) crypto.ReservationLoader {
|
||||
return func(ctx context.Context) ([]crypto.PendingReservation, error) {
|
||||
atts, err := orders.ListAttemptsByStatus(model.AttemptPending, 200)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]crypto.PendingReservation, 0, len(atts))
|
||||
for i := range atts {
|
||||
a := &atts[i]
|
||||
if a.Channel != "crypto" {
|
||||
continue
|
||||
}
|
||||
out = append(out, crypto.PendingReservation{
|
||||
AccountID: a.AccountID, AmountMinor: a.AmountMinor,
|
||||
ProviderRef: a.ProviderRef, ReservedAt: a.CreatedAt,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
func TestCryptoReservationLoaderFiltersPendingCrypto(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
// 两条 attempt:crypto pending(要)、alipay pending(不要)。
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "O1", Channel: "crypto", ProviderRef: "CRYPTO-O1-12",
|
||||
AmountMinor: 100, Currency: "USDT", Status: model.AttemptPending})
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "O2", Channel: "alipay", ProviderRef: "AL-O2",
|
||||
AmountMinor: 200, Currency: "CNY", Status: model.AttemptPending})
|
||||
|
||||
loader := reconcile.CryptoReservationLoader(s)
|
||||
items, err := loader(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("loader: %v", err)
|
||||
}
|
||||
if len(items) != 1 || items[0].ProviderRef != "CRYPTO-O1-12" || items[0].AmountMinor != 100 {
|
||||
t.Fatalf("只应含 crypto pending, got %+v", items)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// OrderExpirerTask 返回「关闭超 TTL 未付 pending 订单」的周期任务体。
|
||||
// cutoff = now()-ttl,now 注入(测试确定性);幂等条件 UPDATE(见 store.ExpireStaleOrders)。
|
||||
func OrderExpirerTask(orders *store.OrderStore, ttl time.Duration, now func() time.Time) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
n, err := orders.ExpireStaleOrders(now().Add(-ttl), 500)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 0 {
|
||||
log.Printf("[reconcile] 过期关闭 %d 个超时未付订单(TTL=%s)", n, ttl)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
func TestOrderExpirerTaskClosesStalePending(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
o := &model.OrderV2{OutTradeNo: "OLD", AmountMinor: 100, Currency: "USDT", Status: model.OrderPendingV2}
|
||||
_ = s.CreateOrder(o)
|
||||
_ = db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "OLD").
|
||||
Update("created_at", now.Add(-2*time.Hour)).Error
|
||||
|
||||
task := reconcile.OrderExpirerTask(s, time.Hour, func() time.Time { return now })
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
got, _ := s.GetOrder("OLD")
|
||||
if got.Status != model.OrderExpiredV2 {
|
||||
t.Fatalf("超时 pending 单应 expired, got %v", got.Status)
|
||||
}
|
||||
}
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// RefundApplyTask 是「RefundApply 修复扫描」:P4 T3 opus review 追加的义务(P4 退款
|
||||
// 代码本体在主 checkout,未落到本 worktree;此处只对本 worktree 已有的
|
||||
// store.RefundStore/OrderStore 接口(P4 T2,在 base 里)建自愈扫描,设计为能在合并
|
||||
// P4 后继续工作)。
|
||||
//
|
||||
// 动机:退款成功的崩溃窗口 —— RefundStore.MarkRefundStatus 把某笔退款翻成
|
||||
// succeeded 后,调用方在再调 OrderStore.ApplyRefundToOrder 前进程崩溃/网络抖动,
|
||||
// 订单状态卡在 paid(或旧的 partially_refunded),与「已实际退款成功」的事实脱节。
|
||||
// 本任务周期重算每个候选订单的 succeeded 退款之和,按既有 ApplyRefundToOrder 的
|
||||
// 条件 UPDATE 语义重新 apply 一次 —— 状态已一致时 UPDATE 影响 0 行,天然幂等。
|
||||
//
|
||||
// 候选订单 = distinct(有 succeeded 退款的订单) ∪ (当前处于 refunding/partially_refunded
|
||||
// 态的订单):前者直接命中"退款成功但订单未跟上"的崩溃窗口;后者兜住"已在退款流程
|
||||
// 中、但后续又有退款 succeeded 未被重算"的情形。
|
||||
//
|
||||
// 两路查询都带 lookback 回溯窗(since = now-lookback),防「LIMIT 尾部饥饿」:succeeded
|
||||
// 退款历史只增不减、partially_refunded 订单长期驻留,若不设窗口按 ASC 取前 limit 条,
|
||||
// 超过 limit 后最新的崩溃窗口(本扫描要自愈的目标)会被挤出候选、永远扫不到。lookback
|
||||
// 只覆盖近期崩溃窗口——历史一致性由已收敛状态保证,不需要扫全表。
|
||||
func RefundApplyTask(orders *store.OrderStore, refunds *store.RefundStore, lookback time.Duration, now func() time.Time, limit int) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
since := now().Add(-lookback)
|
||||
succeededNos, err := refunds.ListDistinctOutTradeNosByStatusSince(model.RefundSucceeded, since, limit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
refundingOrders, err := orders.ListOrdersByStatusSince(
|
||||
[]model.OrderStatusV2{model.OrderRefundingV2, model.OrderPartRefundedV2}, since, limit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
seen := make(map[string]bool, len(succeededNos)+len(refundingOrders))
|
||||
candidates := make([]string, 0, len(succeededNos)+len(refundingOrders))
|
||||
for _, no := range succeededNos {
|
||||
if !seen[no] {
|
||||
seen[no] = true
|
||||
candidates = append(candidates, no)
|
||||
}
|
||||
}
|
||||
for i := range refundingOrders {
|
||||
no := refundingOrders[i].OutTradeNo
|
||||
if !seen[no] {
|
||||
seen[no] = true
|
||||
candidates = append(candidates, no)
|
||||
}
|
||||
}
|
||||
|
||||
for _, no := range candidates {
|
||||
if err := reapplyRefundState(orders, refunds, no); err != nil {
|
||||
log.Printf("[reconcile] 退款修复扫描 out_trade_no=%s: %v", no, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// reapplyRefundState 对单个订单重算 succeeded 退款之和并按需重新 apply 状态转移。
|
||||
// 只有目标态与当前态不同才真正调用 ApplyRefundToOrder(避免每轮扫描都打"翻转"日志噪声);
|
||||
// 没有 succeeded 退款(sum==0)的订单跳过 —— 它不属于本扫描要修的窗口。
|
||||
func reapplyRefundState(orders *store.OrderStore, refunds *store.RefundStore, outTradeNo string) error {
|
||||
succ, err := refunds.RefundSum(outTradeNo, model.RefundSucceeded)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if succ <= 0 {
|
||||
return nil
|
||||
}
|
||||
o, err := orders.GetOrder(outTradeNo)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fully := succ >= o.AmountMinor
|
||||
next := model.OrderPartRefundedV2
|
||||
if fully {
|
||||
next = model.OrderRefundedV2
|
||||
}
|
||||
if o.Status == next {
|
||||
return nil // 已一致,无需自愈
|
||||
}
|
||||
|
||||
flipped, err := orders.ApplyRefundToOrder(outTradeNo, fully)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if flipped {
|
||||
log.Printf("[reconcile][退款自愈] out_trade_no=%s 本地曾卡于 %s,succeeded 退款 %d/%d → 重新 apply 为 %s",
|
||||
outTradeNo, o.Status, succ, o.AmountMinor, next)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RefundStuckAlertTask 是「卡滞 processing/manual_pending 退款告警」义务:同 sweep
|
||||
// 家族的姊妹任务,只读观测 —— 只打 WARN,绝不改状态(状态机翻转是
|
||||
// RefundApplyTask/业务方的事)。渠道退款查询 API 面(主动向渠道问退款进度)留待后续;
|
||||
// 这里先用「本地卡滞时长」兜底可见性。
|
||||
func RefundStuckAlertTask(refunds *store.RefundStore, threshold time.Duration, now func() time.Time) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
cutoff := now().Add(-threshold)
|
||||
stuck, err := refunds.ListStuckRefunds(
|
||||
[]model.RefundStatus{model.RefundProcessing, model.RefundManualPending}, cutoff, 200)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range stuck {
|
||||
r := &stuck[i]
|
||||
log.Printf("[reconcile][WARN][退款卡滞] refund_id=%s out_trade_no=%s status=%s 已卡滞 %s(阈值 %s)——"+
|
||||
"仅观测告警不改状态;渠道退款查询 API 面留待后续",
|
||||
r.RefundID, r.OutTradeNo, r.Status, now().Sub(r.UpdatedAt).Round(time.Minute), threshold)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// TestRefundApplyTaskSelfHealsStuckPaidOrder 钉住 P4 T3 review 的崩溃窗口:
|
||||
// MarkRefundStatus 把退款翻成 succeeded 后、调用方在调 ApplyRefundToOrder 前崩溃,
|
||||
// 订单卡在 paid。RefundApplyTask 应重算 succeeded 之和并重新 apply,自愈成
|
||||
// refunded(全额)。
|
||||
func TestRefundApplyTaskSelfHealsStuckPaidOrder(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-1", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
completedAt := time.Now()
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-1", OutTradeNo: "STUCK-1", AttemptProviderRef: "STUCK-1",
|
||||
AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
|
||||
o, err := orders.GetOrder("STUCK-1")
|
||||
if err != nil || o.Status != model.OrderRefundedV2 {
|
||||
t.Fatalf("应自愈为 refunded, got %+v, err=%v", o, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundApplyTaskPartialStaysPartial 部分退款(succeeded 之和 < 订单金额)应自愈为
|
||||
// partially_refunded,而非误判 fully。
|
||||
func TestRefundApplyTaskPartialStaysPartial(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-2", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
completedAt := time.Now()
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-2", OutTradeNo: "STUCK-2", AttemptProviderRef: "STUCK-2",
|
||||
AmountMinor: 4000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
o, err := orders.GetOrder("STUCK-2")
|
||||
if err != nil || o.Status != model.OrderPartRefundedV2 {
|
||||
t.Fatalf("应自愈为 partially_refunded, got %+v, err=%v", o, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundApplyTaskIdempotentNoOpOnRerun 幂等:自愈一次后重跑不应报错、不应再次
|
||||
// "翻转"(状态已一致,ApplyRefundToOrder 不应被重复触发出错误的副作用)。
|
||||
func TestRefundApplyTaskIdempotentNoOpOnRerun(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-3", AmountMinor: 5000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
completedAt := time.Now()
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-3", OutTradeNo: "STUCK-3", AttemptProviderRef: "STUCK-3",
|
||||
AmountMinor: 5000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200)
|
||||
for i := 0; i < 3; i++ {
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("run %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
o, err := orders.GetOrder("STUCK-3")
|
||||
if err != nil || o.Status != model.OrderRefundedV2 {
|
||||
t.Fatalf("重跑后仍应 refunded, got %+v, err=%v", o, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundApplyTaskConsistentOrderUntouched 已一致(无 succeeded 退款,或订单已是
|
||||
// 该退款对应的终态)的订单不应被误触发。
|
||||
func TestRefundApplyTaskConsistentOrderUntouched(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "OK-1", AmountMinor: 1000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 有一笔 processing(未 succeeded)退款,不该触发自愈。
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-ok-1", OutTradeNo: "OK-1", AttemptProviderRef: "OK-1",
|
||||
AmountMinor: 500, Currency: "CNY", Status: model.RefundProcessing, InitiatedBy: "business"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
o, err := orders.GetOrder("OK-1")
|
||||
if err != nil || o.Status != model.OrderPaidV2 {
|
||||
t.Fatalf("无 succeeded 退款不应被翻转, got %+v, err=%v", o, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundApplyTaskLookbackWindowExcludesStaleSucceeded 钉住 LIMIT 尾部饥饿修复:
|
||||
// succeeded 退款侧的候选查询带回溯窗,completed_at 超窗的行不再入候选(窗外订单卡在
|
||||
// paid 不被自愈),只有窗内的才被扫到并自愈。
|
||||
func TestRefundApplyTaskLookbackWindowExcludesStaleSucceeded(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
lookback := 48 * time.Hour
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "WIN-OLD", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
staleCompleted := now.Add(-lookback - time.Hour) // 窗外
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-win-old", OutTradeNo: "WIN-OLD", AttemptProviderRef: "WIN-OLD",
|
||||
AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleCompleted}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "WIN-NEW", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
freshCompleted := now.Add(-time.Hour) // 窗内
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-win-new", OutTradeNo: "WIN-NEW", AttemptProviderRef: "WIN-NEW",
|
||||
AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &freshCompleted}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, lookback, func() time.Time { return now }, 200)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
|
||||
oldOrder, err := orders.GetOrder("WIN-OLD")
|
||||
if err != nil || oldOrder.Status != model.OrderPaidV2 {
|
||||
t.Fatalf("窗外行不应入候选、不应被自愈, got %+v, err=%v", oldOrder, err)
|
||||
}
|
||||
newOrder, err := orders.GetOrder("WIN-NEW")
|
||||
if err != nil || newOrder.Status != model.OrderRefundedV2 {
|
||||
t.Fatalf("窗内行应被自愈为 refunded, got %+v, err=%v", newOrder, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundApplyTaskLookbackWindowExcludesStaleRefundingOrder 同上,针对
|
||||
// refunding/partially_refunded 订单侧的候选查询:updated_at 超窗的订单不再入候选
|
||||
// (即便它挂着可自愈的 succeeded 退款,只要两路查询都被窗口挡在外面就不会被扫到)。
|
||||
func TestRefundApplyTaskLookbackWindowExcludesStaleRefundingOrder(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
refunds := store.NewRefundStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
lookback := 48 * time.Hour
|
||||
staleAt := now.Add(-lookback - time.Hour) // 窗外
|
||||
freshAt := now.Add(-time.Hour) // 窗内
|
||||
|
||||
// 两笔订单的 succeeded 退款 completed_at 都设窗外,隔离掉「succeeded 侧候选」这条路径
|
||||
// ——只让 refunding/partially_refunded 订单侧的时间窗单独决定是否入候选。
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "ORD-OLD", AmountMinor: 10000, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-ord-old", OutTradeNo: "ORD-OLD", AttemptProviderRef: "ORD-OLD",
|
||||
AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "ORD-OLD").Update("updated_at", staleAt).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "ORD-NEW", AmountMinor: 10000, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-ord-new", OutTradeNo: "ORD-NEW", AttemptProviderRef: "ORD-NEW",
|
||||
AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "ORD-NEW").Update("updated_at", freshAt).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
task := reconcile.RefundApplyTask(orders, refunds, lookback, func() time.Time { return now }, 200)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
|
||||
oldOrder, err := orders.GetOrder("ORD-OLD")
|
||||
if err != nil || oldOrder.Status != model.OrderRefundingV2 {
|
||||
t.Fatalf("窗外订单不应入候选、不应被自愈, got %+v, err=%v", oldOrder, err)
|
||||
}
|
||||
newOrder, err := orders.GetOrder("ORD-NEW")
|
||||
if err != nil || newOrder.Status != model.OrderRefundedV2 {
|
||||
t.Fatalf("窗内订单应被自愈为 refunded, got %+v, err=%v", newOrder, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundStuckAlertTaskLogsWarnWithoutChangingState 卡滞 processing/manual_pending
|
||||
// 超阈值只应打 WARN 日志,不改任何状态(观测型)。
|
||||
func TestRefundStuckAlertTaskLogsWarnWithoutChangingState(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
refunds := store.NewRefundStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stall-1", OutTradeNo: "STALL-1",
|
||||
AmountMinor: 100, Currency: "CNY", Status: model.RefundProcessing}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.Refund{}).Where("refund_id = ?", "rf-stall-1").
|
||||
Update("updated_at", now.Add(-45*time.Minute)).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
orig := log.Writer()
|
||||
log.SetOutput(&buf)
|
||||
defer log.SetOutput(orig)
|
||||
|
||||
task := reconcile.RefundStuckAlertTask(refunds, 30*time.Minute, func() time.Time { return now })
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
|
||||
if !strings.Contains(buf.String(), "rf-stall-1") || !strings.Contains(buf.String(), "WARN") {
|
||||
t.Fatalf("应打 WARN 日志含 refund_id, got: %s", buf.String())
|
||||
}
|
||||
|
||||
r, err := refunds.GetRefund("rf-stall-1")
|
||||
if err != nil || r.Status != model.RefundProcessing {
|
||||
t.Fatalf("告警不应改状态, got %+v, err=%v", r, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRefundStuckAlertTaskSkipsFreshAndTerminal 未超阈值 / 已终态的退款不应被告警。
|
||||
func TestRefundStuckAlertTaskSkipsFreshAndTerminal(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
refunds := store.NewRefundStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-fresh", OutTradeNo: "FRESH-1",
|
||||
AmountMinor: 100, Currency: "CNY", Status: model.RefundProcessing}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-done", OutTradeNo: "DONE-1",
|
||||
AmountMinor: 100, Currency: "CNY", Status: model.RefundSucceeded}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.Refund{}).Where("refund_id = ?", "rf-done").
|
||||
Update("updated_at", now.Add(-2*time.Hour)).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
orig := log.Writer()
|
||||
log.SetOutput(&buf)
|
||||
defer log.SetOutput(orig)
|
||||
|
||||
task := reconcile.RefundStuckAlertTask(refunds, 30*time.Minute, func() time.Time { return now })
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
if strings.Contains(buf.String(), "rf-fresh") || strings.Contains(buf.String(), "rf-done") {
|
||||
t.Fatalf("未超阈值/已终态不应告警, got: %s", buf.String())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
// Package reconcile 是 pay v2 的后台守护/对账装配层:把多个幂等、崩溃安全、
|
||||
// 可重跑的周期任务(订单过期清理 / webhook 死信硬化 / crypto 预留冷启动 /
|
||||
// 用量刷新 / 查单对账 / 链上孤儿发现)挂到统一 Runner。它处于 main 之下的装配层,
|
||||
// 可依赖 store/gateway/accounts/provider;gateway/provider 核心不反向依赖它。
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Task 一个周期任务:名字 + 间隔 + 幂等可重跑的 Run。Local 标记该任务是否纯本地
|
||||
// (只碰 DB,无出网 HTTP)——启动预热(RunOnceLocal)只跑 Local 任务,网络型任务交给
|
||||
// 各自 ticker 首跳,避免上游慢拖住服务启动(main.go r.Run(addr) 之前的同步阶段)。
|
||||
type Task struct {
|
||||
Name string
|
||||
Interval time.Duration
|
||||
Local bool
|
||||
Run func(ctx context.Context) error
|
||||
}
|
||||
|
||||
// Runner 持有一组周期任务,逐个跑(RunOnce)或各自 ticker 常驻(Start)。
|
||||
type Runner struct {
|
||||
tasks []Task
|
||||
logf func(format string, args ...any)
|
||||
}
|
||||
|
||||
func NewRunner() *Runner { return &Runner{logf: log.Printf} }
|
||||
|
||||
// Add 注册一个周期任务(默认视为网络型/非 Local,启动预热 RunOnceLocal 不跑它,
|
||||
// 交给 Start 里各自 ticker 的首跳执行)。
|
||||
func (r *Runner) Add(name string, interval time.Duration, run func(ctx context.Context) error) {
|
||||
r.add(Task{Name: name, Interval: interval, Local: false, Run: run})
|
||||
}
|
||||
|
||||
// AddLocal 注册一个纯本地周期任务(只碰 DB,无出网 HTTP)——会被启动预热
|
||||
// RunOnceLocal 同步执行一次,让 usage 快照/过期清理/退款自愈等立即生效。
|
||||
func (r *Runner) AddLocal(name string, interval time.Duration, run func(ctx context.Context) error) {
|
||||
r.add(Task{Name: name, Interval: interval, Local: true, Run: run})
|
||||
}
|
||||
|
||||
// add 是 Add/AddLocal 的共同落地:Interval<=0 是 operator 配置错误(如
|
||||
// expire_every_sec: 0)——time.NewTicker 对 <=0 的间隔会 panic,且 Start 里每 tick
|
||||
// 的 recover 覆盖不到 NewTicker 本身(它在 goroutine 里、ticker 创建那一行就炸,
|
||||
// 无 defer 保护)。显式优于静默改值:跳过注册 + WARN 日志,而不是偷偷 clamp 成默认值
|
||||
// 掩盖配置错误。
|
||||
func (r *Runner) add(t Task) {
|
||||
if t.Interval <= 0 {
|
||||
r.logf("[reconcile] WARN 任务 %s interval<=0(%v),跳过注册(检查配置)", t.Name, t.Interval)
|
||||
return
|
||||
}
|
||||
r.tasks = append(r.tasks, t)
|
||||
}
|
||||
|
||||
// exec 跑单个任务一次:panic recover + error 记录,绝不外抛(单任务失败不拖垮其它)。
|
||||
func (r *Runner) exec(ctx context.Context, t Task) {
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
r.logf("[reconcile] 任务 %s panic 已恢复: %v", t.Name, rec)
|
||||
}
|
||||
}()
|
||||
if err := t.Run(ctx); err != nil {
|
||||
r.logf("[reconcile] 任务 %s: %v", t.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// RunOnce 顺序跑一遍全部任务(单测入口;main.go 启动预热改用 RunOnceLocal,
|
||||
// 网络型任务不再阻塞启动,见下)。
|
||||
func (r *Runner) RunOnce(ctx context.Context) {
|
||||
for _, t := range r.tasks {
|
||||
r.exec(ctx, t)
|
||||
}
|
||||
}
|
||||
|
||||
// RunOnceLocal 只顺序跑一遍 Local 任务(启动预热用):order-expire/usage-refresh/
|
||||
// refund-apply-sweep/refund-stuck-alert 等纯 DB 任务立即生效;sync-pending/
|
||||
// paid-spotcheck/crypto-orphan-scan 等有出网 HTTP(10-15s 超时)的任务跳过,交给
|
||||
// Start 里各自 ticker 的首跳执行,避免上游慢拖住 r.Run(addr) 前的服务启动。
|
||||
func (r *Runner) RunOnceLocal(ctx context.Context) {
|
||||
for _, t := range r.tasks {
|
||||
if !t.Local {
|
||||
continue
|
||||
}
|
||||
r.exec(ctx, t)
|
||||
}
|
||||
}
|
||||
|
||||
// Start 每任务一 goroutine + 独立 ticker 常驻;ctx 取消即退出。每 tick 崩溃安全。
|
||||
func (r *Runner) Start(ctx context.Context) {
|
||||
for _, t := range r.tasks {
|
||||
t := t
|
||||
go func() {
|
||||
tk := time.NewTicker(t.Interval)
|
||||
defer tk.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tk.C:
|
||||
r.exec(ctx, t)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
)
|
||||
|
||||
func TestRunnerRunOnceExecutesAllAndRecoversPanic(t *testing.T) {
|
||||
r := reconcile.NewRunner()
|
||||
var a, b int
|
||||
r.Add("inc-a", time.Minute, func(context.Context) error { a++; return nil })
|
||||
r.Add("boom", time.Minute, func(context.Context) error { panic("kaboom") }) // 不得拖垮后续
|
||||
r.Add("inc-b", time.Minute, func(context.Context) error { b++; return errors.New("soft") })
|
||||
|
||||
r.RunOnce(context.Background()) // panic 被 recover,error 被记录,均不中断
|
||||
if a != 1 || b != 1 {
|
||||
t.Fatalf("a=%d b=%d, want 1/1(panic 任务不应阻断其它)", a, b)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunnerZeroIntervalSkipsRegistrationAndStartDoesNotPanic 覆盖 Important #1:
|
||||
// operator 配 interval<=0(如 expire_every_sec: 0)不得让 Start 里 time.NewTicker
|
||||
// panic 崩进程 —— Add/AddLocal 应在注册期就 skip 该任务(不进 r.tasks),Start 对它
|
||||
// 不会创建 ticker,自然不 panic;RunOnce/RunOnceLocal 也因任务未注册而不执行它。
|
||||
func TestRunnerZeroIntervalSkipsRegistrationAndStartDoesNotPanic(t *testing.T) {
|
||||
r := reconcile.NewRunner()
|
||||
var zeroRan, negRan, okRan int32
|
||||
r.Add("zero-interval", 0, func(context.Context) error {
|
||||
atomic.AddInt32(&zeroRan, 1)
|
||||
return nil
|
||||
})
|
||||
r.AddLocal("neg-interval", -time.Second, func(context.Context) error {
|
||||
atomic.AddInt32(&negRan, 1)
|
||||
return nil
|
||||
})
|
||||
r.Add("ok-interval", 5*time.Millisecond, func(context.Context) error {
|
||||
atomic.AddInt32(&okRan, 1)
|
||||
return nil
|
||||
})
|
||||
|
||||
r.RunOnce(context.Background())
|
||||
if zeroRan != 0 || negRan != 0 {
|
||||
t.Fatalf("interval<=0 的任务不应被注册/执行: zeroRan=%d negRan=%d", zeroRan, negRan)
|
||||
}
|
||||
if okRan != 1 {
|
||||
t.Fatalf("正常任务应正常执行一次, got okRan=%d", okRan)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
r.Start(ctx) // 若 zero-interval/neg-interval 仍被注册,NewTicker(<=0) 会 panic 崩掉这个 goroutine
|
||||
}()
|
||||
time.Sleep(30 * time.Millisecond) // 留时间给 ok-interval 的 ticker 至少 tick 一次
|
||||
cancel()
|
||||
<-done
|
||||
|
||||
if atomic.LoadInt32(&zeroRan) != 0 || atomic.LoadInt32(&negRan) != 0 {
|
||||
t.Fatalf("Start 之后 interval<=0 的任务仍不应执行: zeroRan=%d negRan=%d", zeroRan, negRan)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunnerRunOnceLocalOnlyRunsLocalTasks 覆盖 Important #2:启动预热
|
||||
// RunOnceLocal 只应执行 AddLocal 注册的任务,Add(默认网络型)注册的任务不跑,
|
||||
// 交给 Start 里各自 ticker 的首跳。
|
||||
func TestRunnerRunOnceLocalOnlyRunsLocalTasks(t *testing.T) {
|
||||
r := reconcile.NewRunner()
|
||||
var ran []string
|
||||
r.AddLocal("local-a", time.Minute, func(context.Context) error {
|
||||
ran = append(ran, "local-a")
|
||||
return nil
|
||||
})
|
||||
r.Add("network-b", time.Minute, func(context.Context) error {
|
||||
ran = append(ran, "network-b")
|
||||
return nil
|
||||
})
|
||||
r.AddLocal("local-c", time.Minute, func(context.Context) error {
|
||||
ran = append(ran, "local-c")
|
||||
return nil
|
||||
})
|
||||
|
||||
r.RunOnceLocal(context.Background())
|
||||
|
||||
if len(ran) != 2 {
|
||||
t.Fatalf("RunOnceLocal 应只跑 2 个 Local 任务, got %v", ran)
|
||||
}
|
||||
for _, name := range ran {
|
||||
if name == "network-b" {
|
||||
t.Fatalf("RunOnceLocal 不应执行网络型任务, got %v", ran)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
"github.com/wangjia/pay/internal/provider"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// SyncPendingTask 调度 P2 gateway.SyncPendingAttempts:逐 pending attempt 查单收敛(防掉单)。
|
||||
// 其内部已对 not_found/amount_mismatch/failed 打日志(settle-sync,764ed55),此处不重复。
|
||||
func SyncPendingTask(gw *gateway.Gateway, limit int) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
n, err := gw.SyncPendingAttempts(ctx, limit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 0 {
|
||||
log.Printf("[reconcile] 查单对账收敛 %d 笔待支付 → paid", n)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// PaidSpotCheckTask 已付订单抽查:对近 window 内已付 attempt 反查渠道,金额/币种漂移即告警
|
||||
// (如渠道侧已退款/拒付而本地仍 paid)。只发现不改状态——状态机翻转属 P4。
|
||||
// crypto 之类 query-only 渠道:paid 后再查若命中同额即一致;查不到(链上历史滚出窗口)不报错跳过。
|
||||
func PaidSpotCheckTask(orders *store.OrderStore, providers *provider.Registry, window time.Duration, now func() time.Time) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
atts, err := orders.ListRecentlyPaidAttempts(now().Add(-window), 100)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i := range atts {
|
||||
a := &atts[i]
|
||||
prov, err := providers.Get(a.Channel)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
created := a.CreatedAt
|
||||
ev, err := prov.Query(ctx, provider.QueryRequest{
|
||||
ProviderRef: a.ProviderRef, OutTradeNo: a.OutTradeNo, AccountID: a.AccountID,
|
||||
AmountMinor: a.AmountMinor, Currency: a.Currency, CreatedAt: created, ExpiresAt: a.ExpiresAt,
|
||||
})
|
||||
if err != nil || ev == nil {
|
||||
continue // 查不到/瞬时错:抽查尽力而为,不阻断
|
||||
}
|
||||
// 本地 paid,渠道却报非成功,或金额/币种对不上 → 对账差异,必须可见。
|
||||
if ev.Status != provider.PaidSucceeded || ev.PaidCurrency != a.Currency || ev.PaidAmountMinor < a.AmountMinor {
|
||||
log.Printf("[reconcile][对账差异] attempt=%s channel=%s 本地 paid 但渠道 status=%s amount=%d/%s(本地 %d/%s)",
|
||||
a.ProviderRef, a.Channel, ev.Status, ev.PaidAmountMinor, ev.PaidCurrency, a.AmountMinor, a.Currency)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/provider"
|
||||
"github.com/wangjia/pay/internal/provider/fake"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
type stubResolver struct{}
|
||||
|
||||
func (stubResolver) Resolve(sku, currency string) (int64, string, string, error) {
|
||||
return 29990000, "Pro", "pro_year", nil
|
||||
}
|
||||
|
||||
type nopEnq struct{}
|
||||
|
||||
// Enqueue 签名依当前仓库 gateway.WebhookEnqueuer(4 个 string + map;brief 草稿只写 3 个 —
|
||||
// P4 T2 之后加了 refundID 参数,此处适配现状,见 task-5-report.md 记录的漂移)。
|
||||
func (nopEnq) Enqueue(_, _, _, _ string, _ map[string]any) error { return nil }
|
||||
|
||||
func TestSyncPendingTaskSettlesViaQuery(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
areg := accounts.New([]config.AccountConfig{{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true}})
|
||||
gw := gateway.New(orders, preg, accounts.NewRouter(areg, nil, nil), stubResolver{}, nopEnq{}, "global")
|
||||
|
||||
res, _ := gw.CreateOrder(context.Background(), gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
|
||||
atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
|
||||
fp.SetQueryResult(atts[0].ProviderRef, provider.PaidEvent{
|
||||
ProviderRef: atts[0].ProviderRef, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"})
|
||||
|
||||
task := reconcile.SyncPendingTask(gw, 50)
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("task: %v", err)
|
||||
}
|
||||
o, _ := orders.GetOrder(res.OrderNo)
|
||||
if o.Status != model.OrderPaidV2 {
|
||||
t.Fatalf("查单对账后应 paid, got %v", o.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPaidSpotCheckTaskRunsCleanOnConsistent(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
orders := store.NewOrderStore(db)
|
||||
preg := provider.NewRegistry()
|
||||
fp := fake.New()
|
||||
preg.Register(fp)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
paid := now.Add(-10 * time.Minute)
|
||||
_ = orders.CreateAttempt(&model.Attempt{OutTradeNo: "O1", Channel: "fake", ProviderRef: "R-O1",
|
||||
AmountMinor: 100, Currency: "USDT", Status: model.AttemptPaid, PaidAt: &paid})
|
||||
// 渠道侧查单仍报 succeeded 同额 → 一致,无告警。
|
||||
fp.SetQueryResult("R-O1", provider.PaidEvent{ProviderRef: "R-O1", Status: provider.PaidSucceeded, PaidAmountMinor: 100, PaidCurrency: "USDT"})
|
||||
|
||||
task := reconcile.PaidSpotCheckTask(orders, preg, time.Hour, func() time.Time { return now })
|
||||
if err := task(context.Background()); err != nil {
|
||||
t.Fatalf("spotcheck: %v", err) // 只求不报错、不 panic;漂移检测走日志
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// UsageSource 满足 accounts.UsageSource:持「账户→当日已收 minor」快照,对账 job 周期 Refresh。
|
||||
// Pick 路径读快照(零 DB);Refresh 从 attempts 聚合。跨天由 now() 的自然日窗自动滚动。
|
||||
type UsageSource struct {
|
||||
orders *store.OrderStore
|
||||
now func() time.Time
|
||||
mu sync.RWMutex
|
||||
snap map[string]int64
|
||||
}
|
||||
|
||||
func NewUsageSource(orders *store.OrderStore, now func() time.Time) *UsageSource {
|
||||
if now == nil {
|
||||
now = time.Now
|
||||
}
|
||||
return &UsageSource{orders: orders, now: now, snap: map[string]int64{}}
|
||||
}
|
||||
|
||||
// TodayUsedMinor 读快照(accounts.UsageSource 接口);未刷新/未知账户返回 0(不误拒)。
|
||||
func (u *UsageSource) TodayUsedMinor(accountID string) int64 {
|
||||
u.mu.RLock()
|
||||
defer u.mu.RUnlock()
|
||||
return u.snap[accountID]
|
||||
}
|
||||
|
||||
// Refresh 从 attempts 重算当日快照(幂等覆盖)。当日起点用 now() 的 UTC 日期。
|
||||
func (u *UsageSource) Refresh(ctx context.Context) error {
|
||||
m, err := u.orders.SumPaidAttemptMinorByAccountSince(StartOfDay(u.now()))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
u.mu.Lock()
|
||||
u.snap = m
|
||||
u.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartOfDay 返回 t 所在 UTC 自然日 00:00(DailyLimit 按自然日结算)。
|
||||
func StartOfDay(t time.Time) time.Time {
|
||||
y, mo, d := t.UTC().Date()
|
||||
return time.Date(y, mo, d, 0, 0, 0, 0, time.UTC)
|
||||
}
|
||||
|
||||
// RefreshUsageTask 把 Refresh 包成周期任务体。
|
||||
func RefreshUsageTask(u *UsageSource) func(ctx context.Context) error {
|
||||
return func(ctx context.Context) error { return u.Refresh(ctx) }
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
func TestUsageSourceRefreshAndInterface(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
paid := now.Add(-time.Hour)
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "A", Channel: "alipay", AccountID: "acct-1",
|
||||
ProviderRef: "R-A", AmountMinor: 12000, Currency: "CNY", Status: model.AttemptPaid, PaidAt: &paid})
|
||||
|
||||
u := reconcile.NewUsageSource(s, func() time.Time { return now })
|
||||
var _ accounts.UsageSource = u // 编译期断言满足接口
|
||||
|
||||
if u.TodayUsedMinor("acct-1") != 0 {
|
||||
t.Fatalf("刷新前应 0")
|
||||
}
|
||||
if err := u.Refresh(context.Background()); err != nil {
|
||||
t.Fatalf("refresh: %v", err)
|
||||
}
|
||||
if u.TodayUsedMinor("acct-1") != 12000 {
|
||||
t.Fatalf("刷新后 acct-1 应 12000, got %d", u.TodayUsedMinor("acct-1"))
|
||||
}
|
||||
if u.TodayUsedMinor("unknown") != 0 {
|
||||
t.Fatalf("未知账户应 0")
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package store
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
@@ -94,3 +95,112 @@ func (s *OrderStore) ApplyRefundToOrder(outTradeNo string, fully bool) (bool, er
|
||||
}
|
||||
return res.RowsAffected > 0, nil
|
||||
}
|
||||
|
||||
// ExpireStaleOrders closes pending orders whose created_at predates cutoff
|
||||
// (TTL 到期未付),条件 UPDATE 只翻 status=pending 的行——与并发 settle 翻 paid
|
||||
// 互斥(谁先谁赢,另一方 RowsAffected=0),故幂等且崩溃安全。含"零尝试孤儿单"
|
||||
// (建单后 CreateAttempt 失败、无 attempt 的 pending 单):它 status 仍是 pending,
|
||||
// 同样被扫到关闭(P2-T4 复审记录的缺口)。cutoff 由调用方用注入时钟算,store 不碰时钟。
|
||||
//
|
||||
// 注:与 ExpirePendingAttempts 语义不同——那个是 attempt 级(标 attempt expired,
|
||||
// 订单不动,retry 前用);这个是 order 级(标整张订单 expired)。
|
||||
func (s *OrderStore) ExpireStaleOrders(cutoff time.Time, limit int) (int64, error) {
|
||||
if limit <= 0 || limit > 1000 {
|
||||
limit = 500
|
||||
}
|
||||
// 先选主键再批量 UPDATE:回避 "UPDATE ... ORDER BY LIMIT" 的方言差异(sqlite/mysql)。
|
||||
var ids []uint64
|
||||
if err := s.db.Model(&model.OrderV2{}).
|
||||
Where("status = ? AND created_at < ?", model.OrderPendingV2, cutoff).
|
||||
Order("id ASC").Limit(limit).Pluck("id", &ids).Error; err != nil {
|
||||
return 0, fmt.Errorf("store.ExpireStaleOrders select: %w", err)
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
res := s.db.Model(&model.OrderV2{}).
|
||||
Where("id IN ? AND status = ?", ids, model.OrderPendingV2). // status 守卫兜住 select→update 间的并发翻转
|
||||
Update("status", model.OrderExpiredV2)
|
||||
if res.Error != nil {
|
||||
return 0, fmt.Errorf("store.ExpireStaleOrders update: %w", res.Error)
|
||||
}
|
||||
return res.RowsAffected, nil
|
||||
}
|
||||
|
||||
// SumPaidAttemptMinorByAccountSince 聚合各账户自 since 起的已付金额(minor),供 LimitAware
|
||||
// 判当日用量。量纲:attempt.AmountMinor 即账户所属渠道结算币种 minor(与 DailyLimit 同量纲)。
|
||||
func (s *OrderStore) SumPaidAttemptMinorByAccountSince(since time.Time) (map[string]int64, error) {
|
||||
type row struct {
|
||||
AccountID string
|
||||
Total int64
|
||||
}
|
||||
var rows []row
|
||||
if err := s.db.Model(&model.Attempt{}).
|
||||
Select("account_id, SUM(amount_minor) AS total").
|
||||
Where("status = ? AND account_id <> '' AND paid_at >= ?", model.AttemptPaid, since).
|
||||
Group("account_id").Scan(&rows).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.SumPaidAttemptMinorByAccountSince: %w", err)
|
||||
}
|
||||
out := make(map[string]int64, len(rows))
|
||||
for _, r := range rows {
|
||||
out[r.AccountID] = r.Total
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListRecentlyPaidAttempts 列近期(paid_at>=since)已付 attempt,供对账抽查反查渠道核对。
|
||||
func (s *OrderStore) ListRecentlyPaidAttempts(since time.Time, limit int) ([]model.Attempt, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 100
|
||||
}
|
||||
var out []model.Attempt
|
||||
if err := s.db.Where("status = ? AND paid_at >= ?", model.AttemptPaid, since).
|
||||
Order("id DESC").Limit(limit).Find(&out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListRecentlyPaidAttempts: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListOrdersByStatus 按状态集合列订单,供退款修复扫描(Task 5 P4 义务)定位「当前处于
|
||||
// 退款相关态」的候选订单,与 RefundStore.ListDistinctOutTradeNosByStatus(succeeded)取并集。
|
||||
func (s *OrderStore) ListOrdersByStatus(statuses []model.OrderStatusV2, limit int) ([]model.OrderV2, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 200
|
||||
}
|
||||
var out []model.OrderV2
|
||||
if err := s.db.Where("status IN ?", statuses).
|
||||
Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListOrdersByStatus: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListOrdersByStatusSince 同 ListOrdersByStatus,但加 updated_at>=since 回溯窗——
|
||||
// refunding/partially_refunded 是长期驻留态(订单进入后可能停留很久),不设窗口时
|
||||
// 会占满 limit 名额,把「近期才卡滞、需要自愈」的订单挤出候选(与
|
||||
// RefundStore.ListDistinctOutTradeNosByStatusSince 同一治法,同一 lookback 语义)。
|
||||
func (s *OrderStore) ListOrdersByStatusSince(statuses []model.OrderStatusV2, since time.Time, limit int) ([]model.OrderV2, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 200
|
||||
}
|
||||
var out []model.OrderV2
|
||||
if err := s.db.Where("status IN ? AND updated_at >= ?", statuses, since).
|
||||
Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListOrdersByStatusSince: %w", err)
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -49,3 +49,186 @@ func TestListPendingAndExpire(t *testing.T) {
|
||||
}
|
||||
_ = time.Now
|
||||
}
|
||||
|
||||
func TestExpireStaleOrders(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
// 三张单:旧 pending(应过期)、新 pending(未到点)、旧 paid(不动)。
|
||||
mk := func(no string, st model.OrderStatusV2, createdAgo time.Duration) {
|
||||
o := &model.OrderV2{OutTradeNo: no, AmountMinor: 100, Currency: "USDT", Status: st}
|
||||
if err := s.CreateOrder(o); err != nil {
|
||||
t.Fatalf("create %s: %v", no, err)
|
||||
}
|
||||
// 直接回填 created_at 制造账龄(GORM 默认写 now)。
|
||||
if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", no).
|
||||
Update("created_at", now.Add(-createdAgo)).Error; err != nil {
|
||||
t.Fatalf("backdate %s: %v", no, err)
|
||||
}
|
||||
}
|
||||
mk("STALE", model.OrderPendingV2, 2*time.Hour) // 旧 pending
|
||||
mk("FRESH", model.OrderPendingV2, 5*time.Minute) // 新 pending
|
||||
mk("PAID", model.OrderPaidV2, 2*time.Hour) // 旧 paid
|
||||
|
||||
cutoff := now.Add(-time.Hour) // TTL=1h
|
||||
n, err := s.ExpireStaleOrders(cutoff, 500)
|
||||
if err != nil {
|
||||
t.Fatalf("expire: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("应关闭 1 张(仅 STALE), got %d", n)
|
||||
}
|
||||
assertStatus := func(no string, want model.OrderStatusV2) {
|
||||
o, _ := s.GetOrder(no)
|
||||
if o.Status != want {
|
||||
t.Fatalf("%s status = %v, want %v", no, o.Status, want)
|
||||
}
|
||||
}
|
||||
assertStatus("STALE", model.OrderExpiredV2)
|
||||
assertStatus("FRESH", model.OrderPendingV2)
|
||||
assertStatus("PAID", model.OrderPaidV2)
|
||||
|
||||
// 幂等:再跑一次不再关闭(STALE 已非 pending)。
|
||||
if n2, _ := s.ExpireStaleOrders(cutoff, 500); n2 != 0 {
|
||||
t.Fatalf("重跑应 0, got %d", n2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSumPaidAttemptMinorByAccountSince(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
dayStart := time.Date(2026, 7, 10, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
mk := func(no, acct string, minor int64, st model.AttemptStatus, paidAgo time.Duration) {
|
||||
paid := now.Add(-paidAgo)
|
||||
a := &model.Attempt{OutTradeNo: no, Channel: "alipay", AccountID: acct, ProviderRef: "R-" + no,
|
||||
AmountMinor: minor, Currency: "CNY", Status: st, PaidAt: &paid}
|
||||
if err := s.CreateAttempt(a); err != nil {
|
||||
t.Fatalf("attempt %s: %v", no, err)
|
||||
}
|
||||
}
|
||||
mk("A", "acct-1", 10000, model.AttemptPaid, 1*time.Hour) // 今日,计入
|
||||
mk("B", "acct-1", 5000, model.AttemptPaid, 2*time.Hour) // 今日,计入 → acct-1=15000
|
||||
mk("C", "acct-2", 7000, model.AttemptPaid, 30*time.Minute) // 今日 acct-2=7000
|
||||
mk("D", "acct-1", 9999, model.AttemptPending, 10*time.Minute) // 未付,不计
|
||||
mk("E", "acct-1", 8888, model.AttemptPaid, 20*time.Hour) // 昨天(paid_at < dayStart),不计
|
||||
|
||||
got, err := s.SumPaidAttemptMinorByAccountSince(dayStart)
|
||||
if err != nil {
|
||||
t.Fatalf("sum: %v", err)
|
||||
}
|
||||
if got["acct-1"] != 15000 || got["acct-2"] != 7000 {
|
||||
t.Fatalf("聚合 = %+v, want acct-1=15000 acct-2=7000", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListRecentlyPaidAttempts(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 string, st model.AttemptStatus, paidAgo time.Duration) {
|
||||
paid := now.Add(-paidAgo)
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: no, Channel: "fake", ProviderRef: "R-" + no,
|
||||
AmountMinor: 100, Currency: "USDT", Status: st, PaidAt: &paid})
|
||||
}
|
||||
mk("RECENT", model.AttemptPaid, 10*time.Minute) // 近期已付 → 命中
|
||||
mk("OLD", model.AttemptPaid, 5*time.Hour) // 太旧 → 不命中
|
||||
mk("PEND", model.AttemptPending, 1*time.Minute) // 未付 → 不命中
|
||||
|
||||
got, err := s.ListRecentlyPaidAttempts(now.Add(-time.Hour), 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].OutTradeNo != "RECENT" {
|
||||
t.Fatalf("只应含 RECENT, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListOrdersByStatus 供 P6+P4 义务的「退款修复扫描」定位候选订单:按状态集合
|
||||
// 列订单(如 refunding/partially_refunded),与 refund 表的 succeeded 记录取并集
|
||||
// 作为重算候选。
|
||||
func TestListOrdersByStatus(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
mk := func(no string, st model.OrderStatusV2) {
|
||||
_ = s.CreateOrder(&model.OrderV2{OutTradeNo: no, AmountMinor: 100, Currency: "CNY", Status: st})
|
||||
}
|
||||
mk("O-PAID", model.OrderPaidV2)
|
||||
mk("O-REFUNDING", model.OrderRefundingV2)
|
||||
mk("O-PART", model.OrderPartRefundedV2)
|
||||
mk("O-DONE", model.OrderRefundedV2)
|
||||
mk("O-PENDING", model.OrderPendingV2)
|
||||
|
||||
got, err := s.ListOrdersByStatus([]model.OrderStatusV2{model.OrderRefundingV2, model.OrderPartRefundedV2}, 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("应命中 2 张(REFUNDING/PART), got %d: %+v", len(got), got)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, o := range got {
|
||||
seen[o.OutTradeNo] = true
|
||||
}
|
||||
if !seen["O-REFUNDING"] || !seen["O-PART"] {
|
||||
t.Fatalf("命中集合不对: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListOrdersByStatusSinceExcludesStale 钉住回溯窗:updated_at 早于 since 的订单
|
||||
// 不应入候选(防 refunding/partially_refunded 长期驻留占满 LIMIT 名额——见 RefundApplyTask)。
|
||||
func TestListOrdersByStatusSinceExcludesStale(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
now := time.Now()
|
||||
stale := now.Add(-72 * time.Hour)
|
||||
fresh := now.Add(-time.Hour)
|
||||
|
||||
if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "O-STALE", AmountMinor: 100, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "O-STALE").Update("updated_at", stale).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "O-FRESH", AmountMinor: 100, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "O-FRESH").Update("updated_at", fresh).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
got, err := s.ListOrdersByStatusSince([]model.OrderStatusV2{model.OrderRefundingV2}, now.Add(-48*time.Hour), 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0].OutTradeNo != "O-FRESH" {
|
||||
t.Fatalf("只应命中窗内的 O-FRESH, got %+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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -128,6 +128,59 @@ func (s *RefundStore) MarkRefundStatus(refundID string, from, to model.RefundSta
|
||||
return res.RowsAffected > 0, nil
|
||||
}
|
||||
|
||||
// ListDistinctOutTradeNosByStatus 列有某状态退款的 distinct out_trade_no,供退款修复
|
||||
// 扫描(RefundApplyTask)定位「有 succeeded 退款」的候选订单——self-heal「退款成功但
|
||||
// 订单卡 paid」的崩溃窗口(P4 T3 review 义务,见 task-5-brief 外的两条追加义务)。
|
||||
func (s *RefundStore) ListDistinctOutTradeNosByStatus(status model.RefundStatus, limit int) ([]string, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 200
|
||||
}
|
||||
var out []string
|
||||
if err := s.db.Model(&model.Refund{}).Where("status = ?", status).
|
||||
Group("out_trade_no").Order("out_trade_no ASC").Limit(limit).
|
||||
Pluck("out_trade_no", &out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListDistinctOutTradeNosByStatus: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListDistinctOutTradeNosByStatusSince 同 ListDistinctOutTradeNosByStatus,但加
|
||||
// completed_at>=since 回溯窗——防「LIMIT 尾部饥饿」:succeeded 退款历史只增不减,
|
||||
// 若不设窗口,按 out_trade_no ASC 取前 limit 条,超过 limit 后最新出现的"退款已成但
|
||||
// 订单卡 paid"崩溃窗口(本扫描要自愈的目标)会被排在候选之外、永远扫不到。completed_at
|
||||
// 是 MarkRefundStatus 翻成 succeeded/failed 时打的时间戳(即"进入该终态"的时刻),succeeded
|
||||
// 是终态、之后不会再被改动,用它做窗口锚点比 updated_at 更精确。调用方(RefundApplyTask)
|
||||
// 用 now-lookback 算 since,lookback 只覆盖近期崩溃窗口——历史一致性由已收敛状态保证,
|
||||
// 不需要扫全表。
|
||||
func (s *RefundStore) ListDistinctOutTradeNosByStatusSince(status model.RefundStatus, since time.Time, limit int) ([]string, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 200
|
||||
}
|
||||
var out []string
|
||||
if err := s.db.Model(&model.Refund{}).Where("status = ? AND completed_at >= ?", status, since).
|
||||
Group("out_trade_no").Order("out_trade_no ASC").Limit(limit).
|
||||
Pluck("out_trade_no", &out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListDistinctOutTradeNosByStatusSince: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListStuckRefunds 列 status 落在给定集合、且 updated_at 早于 before 的退款行,供
|
||||
// 「卡滞 processing/manual_pending 退款告警」只读观测扫描用(不改状态)。updated_at
|
||||
// 用作「进入当前状态」的近似时刻——本表除 MarkRefundStatus/CreateRefundGuarded 外
|
||||
// 不写,近似成立。
|
||||
func (s *RefundStore) ListStuckRefunds(statuses []model.RefundStatus, before time.Time, limit int) ([]model.Refund, error) {
|
||||
if limit <= 0 || limit > 500 {
|
||||
limit = 200
|
||||
}
|
||||
var out []model.Refund
|
||||
if err := s.db.Where("status IN ? AND updated_at < ?", statuses, before).
|
||||
Order("updated_at ASC").Limit(limit).Find(&out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListStuckRefunds: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListManualPending lists refunds awaiting manual (crypto) settlement.
|
||||
func (s *RefundStore) ListManualPending(limit int) ([]model.Refund, error) {
|
||||
if limit <= 0 || limit > 200 {
|
||||
|
||||
@@ -100,6 +100,91 @@ func TestApplyRefundToOrderFully(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestListDistinctOutTradeNosByStatus 供退款修复扫描定位「有 succeeded 退款」的候选
|
||||
// 订单(自愈依据):同订单多笔 succeeded 退款只应出现一次(distinct)。
|
||||
func TestListDistinctOutTradeNosByStatus(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
rs := NewRefundStore(db)
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "s1", OutTradeNo: "PAY-S1", AmountMinor: 100, Currency: "CNY", Status: model.RefundSucceeded})
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "s2", OutTradeNo: "PAY-S1", AmountMinor: 200, Currency: "CNY", Status: model.RefundSucceeded}) // 同单第二笔
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "s3", OutTradeNo: "PAY-S2", AmountMinor: 100, Currency: "CNY", Status: model.RefundSucceeded})
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "p1", OutTradeNo: "PAY-S3", AmountMinor: 100, Currency: "CNY", Status: model.RefundProcessing}) // 非 succeeded,不应命中
|
||||
|
||||
got, err := rs.ListDistinctOutTradeNosByStatus(model.RefundSucceeded, 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("应 distinct 出 2 个 out_trade_no, got %d: %+v", len(got), got)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, no := range got {
|
||||
seen[no] = true
|
||||
}
|
||||
if !seen["PAY-S1"] || !seen["PAY-S2"] {
|
||||
t.Fatalf("命中集合不对: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListDistinctOutTradeNosByStatusSinceExcludesStale 钉住回溯窗:completed_at
|
||||
// 早于 since 的 succeeded 退款不应入候选(防 LIMIT 尾部饥饿——见 RefundApplyTask)。
|
||||
func TestListDistinctOutTradeNosByStatusSinceExcludesStale(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
rs := NewRefundStore(db)
|
||||
now := time.Now()
|
||||
stale := now.Add(-72 * time.Hour)
|
||||
fresh := now.Add(-time.Hour)
|
||||
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "s-stale", OutTradeNo: "PAY-STALE", AmountMinor: 100, Currency: "CNY",
|
||||
Status: model.RefundSucceeded, CompletedAt: &stale})
|
||||
_ = rs.CreateRefund(&model.Refund{RefundID: "s-fresh", OutTradeNo: "PAY-FRESH", AmountMinor: 100, Currency: "CNY",
|
||||
Status: model.RefundSucceeded, CompletedAt: &fresh})
|
||||
|
||||
got, err := rs.ListDistinctOutTradeNosByStatusSince(model.RefundSucceeded, now.Add(-48*time.Hour), 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 1 || got[0] != "PAY-FRESH" {
|
||||
t.Fatalf("只应命中窗内的 PAY-FRESH, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListStuckRefunds 供「卡滞 processing/manual_pending 退款告警」:只挑 updated_at
|
||||
// 早于阈值的 processing/manual_pending 行,requested/succeeded/failed 不命中。
|
||||
func TestListStuckRefunds(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
rs := NewRefundStore(db)
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
mk := func(id, no string, st model.RefundStatus, updatedAgo time.Duration) {
|
||||
if err := rs.CreateRefund(&model.Refund{RefundID: id, OutTradeNo: no, AmountMinor: 100, Currency: "CNY", Status: st}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.Refund{}).Where("refund_id = ?", id).
|
||||
Update("updated_at", now.Add(-updatedAgo)).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
mk("stuck-proc", "PAY-T1", model.RefundProcessing, 45*time.Minute) // 超阈值 → 命中
|
||||
mk("fresh-proc", "PAY-T2", model.RefundProcessing, 5*time.Minute) // 未超 → 不命中
|
||||
mk("stuck-manual", "PAY-T3", model.RefundManualPending, 2*time.Hour) // 超阈值 → 命中
|
||||
mk("done", "PAY-T4", model.RefundSucceeded, 2*time.Hour) // 已终态 → 不命中
|
||||
|
||||
got, err := rs.ListStuckRefunds([]model.RefundStatus{model.RefundProcessing, model.RefundManualPending}, now.Add(-30*time.Minute), 50)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("应命中 2 笔卡滞, got %d: %+v", len(got), got)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, r := range got {
|
||||
seen[r.RefundID] = true
|
||||
}
|
||||
if !seen["stuck-proc"] || !seen["stuck-manual"] {
|
||||
t.Fatalf("命中集合不对: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListManualPending(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
rs := NewRefundStore(db)
|
||||
|
||||
@@ -2,6 +2,7 @@ package store
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -67,7 +68,8 @@ func truncateUTF8(s string, n int) string {
|
||||
}
|
||||
|
||||
// MarkFailed increments attempts and records the last error, leaving the row
|
||||
// undelivered for the next retry sweep.
|
||||
// undelivered for the next retry sweep. Superseded in production by
|
||||
// ScheduleRetry/MarkDead (退避感知);kept for existing callers/tests.
|
||||
func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error {
|
||||
errMsg = truncateUTF8(errMsg, 255)
|
||||
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
||||
@@ -79,3 +81,45 @@ func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListDeliverable 取「未投递、未死信、且退避到点(next_attempt_at NULL 或 <= now)」的行。
|
||||
func (s *WebhookStore) ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error) {
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 50
|
||||
}
|
||||
var out []WebhookDeliveryRow
|
||||
if err := s.db.
|
||||
Where("delivered = ? AND dead = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)", false, false, now).
|
||||
Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.ListDeliverable: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ScheduleRetry 记一次失败并排下次重试:attempts+1、last_error、next_attempt_at=nextAt。
|
||||
func (s *WebhookStore) ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error {
|
||||
errMsg = truncateUTF8(errMsg, 255)
|
||||
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
||||
Updates(map[string]any{
|
||||
"attempts": gorm.Expr("attempts + 1"),
|
||||
"last_error": errMsg,
|
||||
"next_attempt_at": nextAt,
|
||||
}).Error; err != nil {
|
||||
return fmt.Errorf("store.ScheduleRetry: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkDead 达最大次数后放弃:attempts+1、dead=true、last_error。行留库供人工/对账排查。
|
||||
func (s *WebhookStore) MarkDead(id uint64, errMsg string) error {
|
||||
errMsg = truncateUTF8(errMsg, 255)
|
||||
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
||||
Updates(map[string]any{
|
||||
"attempts": gorm.Expr("attempts + 1"),
|
||||
"dead": true,
|
||||
"last_error": errMsg,
|
||||
}).Error; err != nil {
|
||||
return fmt.Errorf("store.MarkDead: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package store_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
@@ -104,3 +105,35 @@ func TestEnqueueDeliveryRefundIDUnique(t *testing.T) {
|
||||
t.Fatalf("undelivered rows = %d want 3: %+v", len(rows), rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookScheduleRetryAndDead(t *testing.T) {
|
||||
ws := store.NewWebhookStore(model.OpenTestDB(t))
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
_ = ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", "", `{"x":1}`)
|
||||
|
||||
// 刚入队:next_attempt_at NULL → 立即可投。
|
||||
rows, _ := ws.ListDeliverable(now, 10)
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("新单应可投, got %d", len(rows))
|
||||
}
|
||||
id := rows[0].ID
|
||||
|
||||
// 排下一次重试到 now+30s:此刻不可投,过点可投。
|
||||
if err := ws.ScheduleRetry(id, "http 500", now.Add(30*time.Second)); err != nil {
|
||||
t.Fatalf("schedule: %v", err)
|
||||
}
|
||||
if r, _ := ws.ListDeliverable(now, 10); len(r) != 0 {
|
||||
t.Fatalf("退避窗内不应可投, got %d", len(r))
|
||||
}
|
||||
if r, _ := ws.ListDeliverable(now.Add(31*time.Second), 10); len(r) != 1 || r[0].Attempts != 1 {
|
||||
t.Fatalf("过退避点应可投且 attempts=1, got %+v", r)
|
||||
}
|
||||
|
||||
// 标死信:不再出现在可投集。
|
||||
if err := ws.MarkDead(id, "gave up"); err != nil {
|
||||
t.Fatalf("markdead: %v", err)
|
||||
}
|
||||
if r, _ := ws.ListDeliverable(now.Add(time.Hour), 10); len(r) != 0 {
|
||||
t.Fatalf("死信不应可投, got %d", len(r))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -31,15 +31,85 @@ type BizConfigFunc func(system string) (config.BizSystemConfig, bool)
|
||||
type OrderPaidFunc func(outTradeNo string) (bool, error)
|
||||
|
||||
type Notifier struct {
|
||||
deliveries *store.WebhookStore
|
||||
bizConfig BizConfigFunc
|
||||
orderPaid OrderPaidFunc
|
||||
client *http.Client
|
||||
deliveries *store.WebhookStore
|
||||
bizConfig BizConfigFunc
|
||||
orderPaid OrderPaidFunc
|
||||
client *http.Client
|
||||
now func() time.Time
|
||||
maxAttempts int
|
||||
baseBackoff time.Duration
|
||||
maxBackoff time.Duration
|
||||
alert func(d *store.WebhookDeliveryRow, reason string)
|
||||
}
|
||||
|
||||
func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc) *Notifier {
|
||||
return &Notifier{deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid,
|
||||
client: &http.Client{Timeout: 10 * time.Second}}
|
||||
// Option customizes a Notifier's retry/backoff/alerting behavior (functional options).
|
||||
type Option func(*Notifier)
|
||||
|
||||
// WithClock overrides the time source (tests inject a fake clock to drive backoff windows).
|
||||
func WithClock(f func() time.Time) Option { return func(n *Notifier) { n.now = f } }
|
||||
|
||||
// WithMaxAttempts sets how many failed attempts before a delivery is marked dead.
|
||||
func WithMaxAttempts(m int) Option { return func(n *Notifier) { n.maxAttempts = m } }
|
||||
|
||||
// WithBaseBackoff sets the base duration for exponential backoff (attempt 1).
|
||||
func WithBaseBackoff(d time.Duration) Option { return func(n *Notifier) { n.baseBackoff = d } }
|
||||
|
||||
// WithMaxBackoff caps the exponential backoff duration.
|
||||
func WithMaxBackoff(d time.Duration) Option { return func(n *Notifier) { n.maxBackoff = d } }
|
||||
|
||||
// WithAlerter overrides the dead-letter alert hook (default logs).
|
||||
func WithAlerter(a func(d *store.WebhookDeliveryRow, reason string)) Option {
|
||||
return func(n *Notifier) { n.alert = a }
|
||||
}
|
||||
|
||||
// NewNotifier builds a Notifier. The original 3-arg call form keeps compiling
|
||||
// (opts is variadic); pass Option values to customize clock/backoff/alerting.
|
||||
func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc, opts ...Option) *Notifier {
|
||||
n := &Notifier{
|
||||
deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid,
|
||||
client: &http.Client{Timeout: 10 * time.Second},
|
||||
now: time.Now,
|
||||
maxAttempts: 12, // ~ 覆盖数小时退避后放弃(见 backoffFor 封顶)
|
||||
baseBackoff: 30 * time.Second, // 首次失败退避基
|
||||
maxBackoff: time.Hour, // 单次退避封顶
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(n)
|
||||
}
|
||||
if n.alert == nil {
|
||||
n.alert = func(d *store.WebhookDeliveryRow, reason string) {
|
||||
log.Printf("[webhook][死信] out_trade_no=%s biz=%s event=%s attempts=%d 放弃投递: %s",
|
||||
d.OutTradeNo, d.BizSystem, d.EventType, d.Attempts, reason)
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// backoffFor 计算第 attempts 次失败后的退避:base·2^(attempts-1),封顶 maxBackoff。
|
||||
// attempts≥1;移位用循环倍增且封顶,防溢出。
|
||||
func (n *Notifier) backoffFor(attempts int) time.Duration {
|
||||
d := n.baseBackoff
|
||||
for i := 1; i < attempts; i++ {
|
||||
d *= 2
|
||||
if d >= n.maxBackoff {
|
||||
return n.maxBackoff
|
||||
}
|
||||
}
|
||||
if d > n.maxBackoff {
|
||||
return n.maxBackoff
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// fail 统一失败分流:未达上限→退避重排;达上限→死信+告警。
|
||||
func (n *Notifier) fail(d *store.WebhookDeliveryRow, msg string) {
|
||||
attempts := d.Attempts + 1 // 本次即将记的失败次数
|
||||
if attempts >= n.maxAttempts {
|
||||
_ = n.deliveries.MarkDead(d.ID, msg)
|
||||
n.alert(d, msg)
|
||||
return
|
||||
}
|
||||
_ = n.deliveries.ScheduleRetry(d.ID, msg, n.now().Add(n.backoffFor(attempts)))
|
||||
}
|
||||
|
||||
// Enqueue implements gateway.WebhookEnqueuer: serialize the domain payload and
|
||||
@@ -53,9 +123,10 @@ func (n *Notifier) Enqueue(outTradeNo, bizSystem, eventType, refundID string, da
|
||||
return n.deliveries.EnqueueDelivery(outTradeNo, bizSystem, eventType, refundID, string(body))
|
||||
}
|
||||
|
||||
// DeliverPending flushes undelivered rows; returns how many succeeded this pass.
|
||||
// DeliverPending flushes deliverable rows (未投递、未死信、退避到点); returns how
|
||||
// many succeeded this pass.
|
||||
func (n *Notifier) DeliverPending(limit int) (int, error) {
|
||||
rows, err := n.deliveries.ListUndelivered(limit)
|
||||
rows, err := n.deliveries.ListDeliverable(n.now(), limit)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
@@ -80,7 +151,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
|
||||
}
|
||||
cfg, found := n.bizConfig(d.BizSystem)
|
||||
if !found || cfg.CallbackURL == "" {
|
||||
_ = n.deliveries.MarkFailed(d.ID, "biz system not configured")
|
||||
n.fail(d, "biz system not configured")
|
||||
return false
|
||||
}
|
||||
ts := strconv.FormatInt(time.Now().Unix(), 10)
|
||||
@@ -89,7 +160,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, cfg.CallbackURL, bytes.NewReader([]byte(d.Payload)))
|
||||
if err != nil {
|
||||
_ = n.deliveries.MarkFailed(d.ID, err.Error())
|
||||
n.fail(d, err.Error())
|
||||
return false
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
@@ -101,7 +172,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
|
||||
|
||||
resp, err := n.client.Do(req)
|
||||
if err != nil {
|
||||
_ = n.deliveries.MarkFailed(d.ID, err.Error())
|
||||
n.fail(d, err.Error())
|
||||
return false
|
||||
}
|
||||
rb, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||||
@@ -111,7 +182,7 @@ func (n *Notifier) deliverOne(d *store.WebhookDeliveryRow) bool {
|
||||
_ = n.deliveries.MarkDelivered(d.ID)
|
||||
return true
|
||||
}
|
||||
_ = n.deliveries.MarkFailed(d.ID, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120)))
|
||||
n.fail(d, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120)))
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
@@ -72,6 +73,60 @@ func TestNotifierDeliversSignedEvent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 业务方持续 500:每次失败按指数退避重排;时钟推进后才重投;达上限标死信 + 告警。
|
||||
func TestNotifierBackoffThenDeadWithAlert(t *testing.T) {
|
||||
var hits int
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hits++
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
nowFn := func() time.Time { return clk }
|
||||
var alerted []string
|
||||
ws := store.NewWebhookStore(model.OpenTestDB(t))
|
||||
n := webhook.NewNotifier(ws,
|
||||
func(string) (config.BizSystemConfig, bool) {
|
||||
return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
|
||||
},
|
||||
func(string) (bool, error) { return true, nil },
|
||||
webhook.WithClock(nowFn),
|
||||
webhook.WithBaseBackoff(time.Second),
|
||||
webhook.WithMaxBackoff(4*time.Second),
|
||||
webhook.WithMaxAttempts(3),
|
||||
webhook.WithAlerter(func(d *store.WebhookDeliveryRow, reason string) { alerted = append(alerted, d.OutTradeNo) }),
|
||||
)
|
||||
_ = n.Enqueue("PAY-D", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"})
|
||||
|
||||
// 尝试 1:失败 → attempts=1,退避到 +1s。
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 {
|
||||
t.Fatalf("try1 sent=%d hits=%d", sent, hits)
|
||||
}
|
||||
// 退避窗内不投。
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 {
|
||||
t.Fatalf("退避窗内不应再敲, hits=%d", hits)
|
||||
}
|
||||
// 推进越过退避;尝试 2 失败 → attempts=2,退避到 +2s。
|
||||
clk = clk.Add(2 * time.Second)
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 2 {
|
||||
t.Fatalf("try2 hits=%d", hits)
|
||||
}
|
||||
// 推进;尝试 3 失败 → attempts 达 maxAttempts(3)→ 死信 + 告警。
|
||||
clk = clk.Add(4 * time.Second)
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 {
|
||||
t.Fatalf("try3 hits=%d", hits)
|
||||
}
|
||||
if len(alerted) != 1 || alerted[0] != "PAY-D" {
|
||||
t.Fatalf("死信应触发告警一次, got %+v", alerted)
|
||||
}
|
||||
// 已死信:无论时钟怎么走都不再投。
|
||||
clk = clk.Add(time.Hour)
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 {
|
||||
t.Fatalf("死信后不应再投, hits=%d", hits)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifierRetriesOnFailure(t *testing.T) {
|
||||
var hits int
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -80,20 +135,23 @@ func TestNotifierRetriesOnFailure(t *testing.T) {
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
ws := store.NewWebhookStore(model.OpenTestDB(t))
|
||||
n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
|
||||
return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
|
||||
}, func(string) (bool, error) { return true, nil })
|
||||
}, func(string) (bool, error) { return true, nil },
|
||||
webhook.WithClock(func() time.Time { return clk }),
|
||||
webhook.WithBaseBackoff(time.Second))
|
||||
_ = n.Enqueue("PAY-3", "pangolin", "payment.succeeded", "", map[string]any{"event_type": "payment.succeeded"})
|
||||
|
||||
if sent, _ := n.DeliverPending(10); sent != 0 {
|
||||
t.Fatalf("失败不应算投递成功, got %d", sent)
|
||||
}
|
||||
// 仍待投递,可被下一轮重试兜底
|
||||
pend, _ := ws.ListUndelivered(10)
|
||||
if len(pend) != 1 || pend[0].Attempts != 1 {
|
||||
t.Fatalf("失败后应留队重试, got %+v", pend)
|
||||
}
|
||||
clk = clk.Add(2 * time.Second) // 越过退避窗
|
||||
if _, _ = n.DeliverPending(10); hits < 2 {
|
||||
t.Fatalf("应重试第二次, hits=%d", hits)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"strings"
|
||||
@@ -17,6 +18,7 @@ import (
|
||||
"github.com/wangjia/pay/internal/gateway"
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/providerbuild"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/router"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
"github.com/wangjia/pay/internal/webhook"
|
||||
@@ -54,11 +56,41 @@ func main() {
|
||||
acctReg := accounts.New(config.C.Accounts)
|
||||
pReg := providerbuild.BuildRegistry(acctReg) // 配置驱动:有 enabled 账户才注册对应渠道(P3)
|
||||
// P5 多账户路由:按 config.routing.<channel> 选策略(缺省 round_robin)。
|
||||
// limit_aware 用量数据源 P6 对账就绪前用空源(NopUsage,退化为 round_robin)。
|
||||
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, accounts.NopUsage{})
|
||||
// limit_aware 用量数据源:P6 对账 Runner 周期 Refresh 的真实用量源(替 NopUsage)。
|
||||
usage := reconcile.NewUsageSource(orderStore, time.Now)
|
||||
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, usage)
|
||||
gw := gateway.New(orderStore, refundStore, pReg, acctPicker, productResolver, notifier, "cn")
|
||||
router.SetupV2(r, gw)
|
||||
|
||||
// P6 后台守护 / 对账:订单过期清理 + 用量刷新 + 查单对账收敛 + 已付抽查 +
|
||||
// 退款修复扫描/卡滞告警(P4 T3 review 追加义务,归到本任务一起装配)+
|
||||
// crypto 预留冷启动 Warm + 孤儿扫描(Task 6,AddCryptoJobs)。
|
||||
if config.C.Reconcile.Enabled {
|
||||
rc := config.C.Reconcile
|
||||
runner := reconcile.NewRunner()
|
||||
// 纯本地(DB-only)任务用 AddLocal:会被启动预热 RunOnceLocal 同步跑一遍。
|
||||
runner.AddLocal("order-expire", time.Duration(rc.ExpireEverySec)*time.Second,
|
||||
reconcile.OrderExpirerTask(orderStore, time.Duration(rc.OrderTTLMin)*time.Minute, time.Now))
|
||||
runner.AddLocal("usage-refresh", time.Duration(rc.UsageEverySec)*time.Second,
|
||||
reconcile.RefreshUsageTask(usage))
|
||||
runner.AddLocal("refund-apply-sweep", time.Duration(rc.RefundApplyEverySec)*time.Second,
|
||||
reconcile.RefundApplyTask(orderStore, refundStore, time.Duration(rc.RefundApplyLookbackMin)*time.Minute, time.Now, 200))
|
||||
runner.AddLocal("refund-stuck-alert", time.Duration(rc.RefundApplyEverySec)*time.Second,
|
||||
reconcile.RefundStuckAlertTask(refundStore, time.Duration(rc.RefundStuckWarnMin)*time.Minute, time.Now))
|
||||
// 网络型(出网 HTTP,10-15s 超时)任务用 Add:启动预热不跑,交各自 ticker 首跳。
|
||||
runner.Add("sync-pending", time.Duration(rc.SyncEverySec)*time.Second,
|
||||
reconcile.SyncPendingTask(gw, 100))
|
||||
runner.Add("paid-spotcheck", time.Duration(rc.SpotCheckEverySec)*time.Second,
|
||||
reconcile.PaidSpotCheckTask(orderStore, pReg, time.Duration(rc.SpotCheckWindowMin)*time.Minute, time.Now))
|
||||
orphanStore := store.NewOrphanStore(db)
|
||||
reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc) // crypto 预留冷启动 Warm(直跑,不受影响)+ 孤儿扫描(网络型,交 ticker 首跳)
|
||||
|
||||
ctx := context.Background()
|
||||
runner.RunOnceLocal(ctx) // 启动预热:只跑本地任务(usage 快照/过期清理/退款自愈立即生效);网络型任务(sync-pending/paid-spotcheck/crypto-orphan-scan)交各自 ticker 首跳,避免上游慢拖住 r.Run(addr) 前的启动
|
||||
runner.Start(ctx)
|
||||
log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/退款修复扫描/退款卡滞告警/crypto 孤儿扫描)")
|
||||
}
|
||||
|
||||
if config.C.QuerySync.Enabled {
|
||||
orderSvc.StartQuerySync(
|
||||
time.Duration(config.C.QuerySync.IntervalSec)*time.Second,
|
||||
@@ -121,7 +153,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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user