# pay v2 · P6 对账与后台守护 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. 每个 Task 自包含,给完整 Go 代码 + 测试 + 命令,可交给全新 implementer subagent。 > **设计文档(全景蓝图):** `docs/pay-v2-unified-gateway-design.html`。P6 直接兑现其 §8 结尾铁律「**对账优先于实时**:每 provider 实现 query,跑定时对账 job 防掉单/防重复授权」。 > **前置计划:** > - `docs/superpowers/plans/2026-07-10-pay-v2-p1-core-model.md`(P1:`model.OrderV2`/`Attempt`/`WebhookDelivery`、`store.OrderStore`/`WebhookStore`、`accounts`)。 > - `docs/superpowers/plans/2026-07-10-pay-v2-p2-pipeline.md`(P2:`gateway.Settle`/`SyncPendingAttempts`、webhook outbox、`store.ExpirePendingAttempts`;**本计划直接复用其接口,不重造**)。P2 计划 Self-Review「后续阶段」条目明确把**订单级过期清理**与 **Notifier 硬化(退避/死信/告警)** routed 到 P6。 > - `docs/superpowers/plans/2026-07-10-pay-v2-p3-channel-adapters.md`(P3:crypto 自托管 adapter + 进程内预留表)。P3 Self-Review「已知取舍」把 **crypto 预留表持久化冷启动兜底** 与 **crypto orphan 到账落表** routed 到 P6。 > - `docs/superpowers/plans/2026-07-10-pay-v2-p5-account-routing.md`(P5:`accounts.Router`/`LimitAware`/`UsageSource`)。P5 计划 D5 明确 **LimitAware 真实用量数据源由 P6 对账 job 提供**,届时实现 `accounts.UsageSource` 注入 Router 即可,`LimitAware`/`Router` 不改。 **Goal:** 给 pay v2 装上一套**周期后台守护 + 对账**:统一 `internal/reconcile` 包挂多个幂等、崩溃安全、可重跑的周期任务——① 订单级过期清理(超 TTL 未付 pending 单自动关闭,含零尝试孤儿单);② Notifier 投递硬化(指数退避 + 最大次数 + 死信标记 + 告警钩子);③ crypto 预留表持久化冷启动兜底(用 attempts 表重建 reservation,兜住重启丢内存);④ `LimitAware` 真实用量数据源(attempts 按账户结算币种聚合当日已收);⑤ 对账主体(周期 `SyncPendingAttempts` + 已付订单抽查核对);⑥ crypto 孤儿链上支付发现(到账但不匹配任何 attempt → 落表 + 告警)。全程注入时钟 + 假 provider/httptest,**免 docker、不打真网**。 **Architecture:** 后台 job 统一形态 = `reconcile.Runner`:注册多个 `Task{Name, Interval, Run(ctx)}`,`Start(ctx)` 每任务一 goroutine + 独立 ticker + `recover`(单 tick panic 不拖垮其它),`RunOnce(ctx)` 顺序跑一遍(启动预热 + 单测入口)。每个 Task 是**幂等 sweep**(条件 UPDATE / 快照刷新 / 查单收敛),重跑安全、崩溃后下一 tick 自愈。`reconcile` 处于装配层(可 import `store`/`gateway`/`accounts`/`provider`,类比 main),而 `gateway`/`provider` 核心仍保持中性——不反向依赖 `reconcile`。过期清理走 `store` 新增条件 UPDATE;Notifier 硬化改 `webhook` + `store.WebhookStore` + `model.WebhookDelivery` 加列;crypto 冷启动兜底经**注入 `ReservationLoader`**(crypto 不 import `store`,装配期 main 提供 loader);用量数据源 `reconcile.UsageSource` 结构体满足 `accounts.UsageSource` 接口(Router 注入它、刷新 job 也持它);orphan 经**中性可选接口 `provider.OrphanScanner`**(crypto 实现,`reconcile` type-assert)。 **Tech Stack:** Go 1.26.1 · `github.com/wangjia/pay` · Gin · GORM v1.31 · glebarez/sqlite · 标准库 `time`/`context`/`sync`/`log`。无新增第三方依赖。crypto orphan 复用 P3 已有 TronGrid REST 客户端(`p.http` + `only_confirmed`)。 ## Global Constraints(继承 P2/P3/P5) - **复用不重造**:模型 `model.OrderV2`/`Attempt`/`WebhookDelivery` + `V2` 后缀状态;数据访问 `store.OrderStore`/`WebhookStore`;管线 `gateway.Gateway`(`Settle`/`SyncPendingAttempts`);账户 `accounts.Registry`/`Router`/`LimitAware`/`UsageSource`(接口)/`config.AccountConfig`;Provider 抽象 `provider.Provider`/`QueryRequest`/`PaidEvent`;crypto adapter `internal/provider/crypto`(`allocateAmount`/`Query`/`tailFromRef`/`address`/进程内 `reserved` map)。 - **金额一律 int64 最小单位 + 币种码**,禁 float;时间一律 Go 端算好传 `?`(cutoff/since 由 job 用注入时钟算,不写 SQL 时间函数,保持 sqlite/mysql 可移植)。 - **幂等 + 崩溃安全**:每个 job 是可重跑 sweep;订单过期用 `status = pending` 守卫的条件 UPDATE(与并发 settle 翻转不打架:谁先谁赢,另一方 RowsAffected=0 no-op);死信/退避靠列状态持久化,重启续跑;usage 快照刷新是幂等覆盖;orphan 落表按 tx_id unique 去重。 - **Provider 中性守恒**:`internal/gateway`、`internal/provider`(核心 `provider.go`)不 import `reconcile`;crypto 不 import `store`(经注入 loader / 中性接口传数据)。`reconcile` 是装配层,只被 main 调。 - **凭证 env only**,测试用 `t.Setenv`;所有测试 `:memory:`/临时 sqlite + httptest 假 TronGrid/假业务方,**免 docker、不打真网**。 - 每步 `go build ./...` 通过;测试 `go test ./...` 全绿。每任务严格 bite-sized TDD:写失败测试 → 跑失败 → 实现 → 跑通过 → commit。**禁占位**。 ## 债务台账落点(prior reviews routed 到 P6 的 6 项 + 1 已做项) | # | 债务项 | 落点 | |---|--------|------| | 1 | 订单级过期清理(超 TTL pending 单关闭,含零尝试孤儿 pending 单) | **Task 1** | | 2 | Notifier 重试硬化(退避 / 最大次数 / 死信 / 告警钩子) | **Task 2** | | 3 | crypto 孤儿链上支付(到账不匹配任何 attempt → 记录 + 告警) | **Task 6** | | 4 | crypto 预留表多实例/重启兜底(attempts 表冷启动重建) | **Task 3** | | 5 | LimitAware 真实数据源(attempts/orders 按账户结算币种聚合) | **Task 4** | | 6 | 对账主体(周期 SyncPendingAttempts + 渠道流水 vs 本地订单核对) | **Task 5**(决策:全量对账单下载判定过重 → 降级为「逐 pending attempt query + 已付订单抽查」,见 Task 5 Self-Review) | | 7 | settle-sync 非常态结果日志(764ed55 已做) | 已存在,**不重复**;Task 5 调度 `SyncPendingAttempts` 时天然继承其日志 | --- ### Task 1: reconcile.Runner 骨架 + 订单级过期清理(ExpireStaleOrders) **债务 #1。** 建后台 job 统一形态(`Runner`),并落地第一个 job:超 TTL 未付 pending 订单自动关闭。含 P2-T4 记录的「零尝试孤儿 pending 单」——建单后 `prov.Create`/`CreateAttempt` 失败留下的、调用方拿不到单号无法 retry/cancel 的 pending 单,因其 `status=pending` 同样被扫到。**注意与 attempt 级过期区分**:P2 已有 `store.ExpirePendingAttempts`(把某单的 pending attempt 标 expired,订单状态不动,retry 前调用);本任务是**订单级**——把整张 pending 订单标 `expired`。 **Files:** - Create: `internal/reconcile/runner.go`(Runner 骨架) - Create: `internal/reconcile/runner_test.go` - Create: `internal/reconcile/orders.go`(OrderExpirerTask 工厂) - Create: `internal/reconcile/orders_test.go` - Modify: `internal/store/order_query.go`(新增 `ExpireStaleOrders`) - Modify: `internal/store/order_query_test.go`(若无则新建,断言 `ExpireStaleOrders`) **Interfaces:** - `store.(*OrderStore).ExpireStaleOrders(cutoff time.Time, limit int) (int64, error)` —— 条件 UPDATE:`status=pending AND created_at < cutoff` → `expired`;返回关闭条数。`cutoff` 由调用方(job)用注入时钟算(`now-TTL`),store 不碰时钟。 - `reconcile.Task{Name string; Interval time.Duration; Run func(ctx) error}`;`(*Runner).Add(name, interval, run)` / `RunOnce(ctx)` / `Start(ctx)`。 - [ ] **Step 1: 写失败测试** `internal/store/order_query_test.go` 追加(若文件不存在则新建,`package store` 内部测试以便直接改 `created_at`): ```go package store import ( "testing" "time" "github.com/wangjia/pay/internal/model" ) func TestExpireStaleOrders(t *testing.T) { db := model.OpenTestDB(t) s := 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) } } ``` `internal/reconcile/runner_test.go`: ```go package reconcile_test import ( "context" "errors" "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) } } ``` `internal/reconcile/orders_test.go`: ```go 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) } } ``` - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20` Expected: 编译失败——`ExpireStaleOrders` / `reconcile` 包不存在。 - [ ] **Step 3: 写实现** `internal/store/order_query.go` 追加(`import "time"` 已需要;文件顶已 import errors/fmt/gorm/model,补 `"time"`): ```go // 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 } ``` `internal/reconcile/runner.go`: ```go // Package reconcile 是 pay v2 的后台守护/对账装配层:把多个幂等、崩溃安全、 // 可重跑的周期任务(订单过期清理 / webhook 死信硬化 / crypto 预留冷启动 / // 用量刷新 / 查单对账 / 链上孤儿发现)挂到统一 Runner。它处于 main 之下的装配层, // 可依赖 store/gateway/accounts/provider;gateway/provider 核心不反向依赖它。 package reconcile import ( "context" "log" "time" ) // Task 一个周期任务:名字 + 间隔 + 幂等可重跑的 Run。 type Task struct { Name string Interval time.Duration 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 注册一个周期任务。 func (r *Runner) Add(name string, interval time.Duration, run func(ctx context.Context) error) { r.tasks = append(r.tasks, Task{Name: name, Interval: interval, Run: run}) } // 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 顺序跑一遍全部任务(启动预热 + 单测入口)。 func (r *Runner) RunOnce(ctx context.Context) { for _, t := range r.tasks { 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) } } }() } } ``` `internal/reconcile/orders.go`: ```go 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 } } ``` - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ -v 2>&1 | tail -30` Expected: 全 PASS。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/reconcile/runner.go internal/reconcile/runner_test.go internal/reconcile/orders.go internal/reconcile/orders_test.go internal/store/order_query.go internal/store/order_query_test.go git commit -m "feat(v2): reconcile Runner 骨架 + 订单级过期清理(超 TTL pending 单/零尝试孤儿单自动关闭)" ``` --- ### Task 2: Notifier 投递硬化 —— 指数退避 + 最大次数 + 死信标记 + 告警钩子 **债务 #2。** 现状 `webhook/notifier.go` 的 `Start` 是裸 60s ticker、`MarkFailed` 只 `attempts+1`——**无退避**(业务方挂了每 60s 猛敲)、**无上限**(永远重投)、**无死信/告警**(投不出去的单静默堆积)。本任务:失败走**指数退避**(`next_attempt_at = now + base·2^(attempts-1)`,封顶 `maxBackoff`),达 `maxAttempts` 标 `dead=true` 并触发**告警钩子**;`DeliverPending` 只取「未投递、未死信、且 `next_attempt_at` 到点」的行。不破坏现有幂等(unique 键入队)与投递门禁(订单未付不投、不计失败)。 **Files:** - Modify: `internal/model/webhook_delivery.go`(加 `Dead` + `NextAttemptAt` 列) - Modify: `internal/store/webhook.go`(`ListDeliverable` / `ScheduleRetry` / `MarkDead`) - Modify: `internal/store/webhook_test.go`(断言新查询/调度) - Modify: `internal/webhook/notifier.go`(functional options:时钟/上限/退避/告警;失败分流退避 vs 死信) - Modify: `internal/webhook/notifier_test.go`(注入时钟测退避/死信/告警;修 `TestNotifierRetriesOnFailure` 让第二次投递前推进时钟越过退避) **Interfaces:** - `store.(*WebhookStore).ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error)` - `store.(*WebhookStore).ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error`(attempts+1、last_error、next_attempt_at) - `store.(*WebhookStore).MarkDead(id uint64, errMsg string) error`(attempts+1、dead=true、last_error) - `webhook.NewNotifier(ws, bizConfig, orderPaid, opts ...Option)` + `WithClock` / `WithMaxAttempts` / `WithBaseBackoff` / `WithMaxBackoff` / `WithAlerter`。**保留原 3 参调用式**(options 变参),现有 main/测试不改签名即可编译。 > AutoMigrate 只加不删,新列对存量库安全(`Dead` 默认 false,`NextAttemptAt` NULL = 立即可投)。 - [ ] **Step 1: 写失败测试** `internal/store/webhook_test.go` 追加: ```go 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)) } } ``` `internal/webhook/notifier_test.go` —— 新增退避/死信/告警用例,并**改造** `TestNotifierRetriesOnFailure`(退避后需推进时钟才重试): ```go // 业务方持续 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) } } ``` 并把原 `TestNotifierRetriesOnFailure` 里第二次 `DeliverPending` 前插入时钟推进 —— 该测试改为注入时钟: ```go func TestNotifierRetriesOnFailure(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) 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(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) } } ``` > `time` 需加入 notifier_test.go 的 import。`ws.ListUndelivered` 保留(列 delivered=false 全量,供该断言),Notifier 内部改用 `ListDeliverable`。 - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/webhook/ 2>&1 | tail -20` Expected: 编译失败——`ListDeliverable`/`ScheduleRetry`/`MarkDead`/`WithClock` 等未定义;`Dead`/`NextAttemptAt` 列不存在。 - [ ] **Step 3: 写实现** `internal/model/webhook_delivery.go` —— 结构体尾部加两列: ```go 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=立即可投 ``` > 文件加 `import "time"`。 `internal/store/webhook.go` —— 新增三方法,并保留 `ListUndelivered`/`MarkDelivered`(`MarkFailed` 可删,已无调用方;保留亦无害,这里替换为下面两法): ```go // 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 } ``` > `import "time"` 加入 webhook.go(store 包)。若删 `MarkFailed`,确认无其它调用方(`grep -rn MarkFailed internal/`,当前仅 notifier 用,Step 3 一并改掉)。 `internal/webhook/notifier.go` —— 加 options + 失败分流: ```go // 结构体加字段 type Notifier struct { 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) } type Option func(*Notifier) func WithClock(f func() time.Time) Option { return func(n *Notifier) { n.now = f } } func WithMaxAttempts(m int) Option { return func(n *Notifier) { n.maxAttempts = m } } func WithBaseBackoff(d time.Duration) Option { return func(n *Notifier) { n.baseBackoff = d } } func WithMaxBackoff(d time.Duration) Option { return func(n *Notifier) { n.maxBackoff = d } } func WithAlerter(a func(d *store.WebhookDeliveryRow, reason string)) Option { return func(n *Notifier) { n.alert = a } } 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;移位用 uint 且封顶,防溢出。 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))) } ``` `DeliverPending` 改用 `ListDeliverable(n.now(), limit)`: ```go func (n *Notifier) DeliverPending(limit int) (int, error) { rows, err := n.deliveries.ListDeliverable(n.now(), limit) if err != nil { return 0, err } ok := 0 for i := range rows { if n.deliverOne(&rows[i]) { ok++ } } return ok, nil } ``` `deliverOne` 把原先所有 `n.deliveries.MarkFailed(...)` 调用改成 `n.fail(d, )`;投递门禁(订单未付)分支**不变**(仍 `return false` 不计失败);成功仍 `MarkDelivered`。逐处替换: ```go // biz 未配置 if !found || cfg.CallbackURL == "" { n.fail(d, "biz system not configured") return false } ... // 构建请求失败 / Do 失败 / 非 200 或无 SUCCESS n.fail(d, err.Error()) // 三处原 MarkFailed → fail ... n.fail(d, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120))) ``` `Start` 保持 60s ticker 不变(现在每轮 `DeliverPending` 已退避感知);其 `log` 文案可留。 - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/webhook/ -v 2>&1 | tail -30` Expected: 全 PASS(含改造后的 retry 用例 + 新死信/告警用例)。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/model/webhook_delivery.go internal/store/webhook.go internal/store/webhook_test.go internal/webhook/notifier.go internal/webhook/notifier_test.go git commit -m "feat(v2): webhook 投递硬化——指数退避 + 最大次数死信 + 告警钩子(替裸 60s 猛敲)" ``` --- ### Task 3: crypto 预留表持久化冷启动兜底(attempts 表重建 reservation) **债务 #4。** P3 crypto 的唯一金额预留表是**进程内 `reserved` map**——重启即丢,重启后并发同价新单可能复用一个仍在冷却窗内的金额,让迟到旧款误配新单。canonical(pangolin-pay)用自库 `AmountRecentlyUsed` 查重;pay 移植方案(P3 复审记录):**用 attempts 表冷启动重建**。本任务给 crypto 注入一个 `ReservationLoader`(装配期 main 提供,读 pending crypto attempts),`Warm(ctx)` 时把每个未过冷却窗的 (地址,期望金额) 灌回 `reserved`。crypto 不 import store —— loader 是注入函数,数据经中性结构传入,tail→期望金额的换算仍封在 crypto 内(复用 `tailFromRef`)。 **Files:** - Modify: `internal/provider/crypto/crypto.go`(`ReservationLoader` 类型 + `WithReservationLoader`/`WithNow` Option + `Warm`) - Modify: `internal/provider/crypto/crypto_test.go`(Warm 重建 + 冷却过期跳过) - Create: `internal/reconcile/crypto_warm.go`(装配期 loader 工厂:从 OrderStore 读 pending crypto attempts) - Create: `internal/reconcile/crypto_warm_test.go` **Interfaces:** - `crypto.PendingReservation{AccountID string; AmountMinor int64; ProviderRef string; ReservedAt time.Time}`(`AmountMinor`=attempt 冻结的 base 金额,`ProviderRef`=`CRYPTO--` 供恢复尾数,`ReservedAt`=建单时间=attempt.CreatedAt,冷却窗从此算)。 - `crypto.ReservationLoader func(ctx context.Context) ([]PendingReservation, error)`;`crypto.WithReservationLoader(l)`;`crypto.WithNow(f)`(测试注入时钟)。 - `crypto.(*Provider).Warm(ctx) error`。 - `reconcile.CryptoReservationLoader(orders *store.OrderStore) crypto.ReservationLoader`(读 `ListAttemptsByStatus(AttemptPending,…)` 过滤 `Channel=="crypto"`,映射为 `PendingReservation`)。 - [ ] **Step 1: 写失败测试** `internal/provider/crypto/crypto_test.go` 追加(复用文件里既有的 `t.Setenv("CRY_ADDRESS",…)` + `accounts.New` 模式;`GetReserved` 已由 export_test.go 提供): ```go 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) } } ``` > import 补 `context`/`strconv`/`time`(crypto_test.go 若未 import)。tail 263/777 与 base 29990000 相加即期望金额,键式 `/` 与 `allocateAmount` 一致。 `internal/reconcile/crypto_warm_test.go`: ```go 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) } } ``` - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/provider/crypto/ ./internal/reconcile/ 2>&1 | tail -20` Expected: 编译失败——`PendingReservation`/`WithReservationLoader`/`WithNow`/`Warm`/`CryptoReservationLoader` 未定义。 - [ ] **Step 3: 写实现** `internal/provider/crypto/crypto.go` —— 结构体加 `loader`,补 Option 与 `Warm`: ```go // 结构体加一行: type Provider struct { accts *accounts.Registry baseURL string http *http.Client now func() time.Time loader ReservationLoader // 冷启动预留重建源(装配期注入,nil=不重建) mu sync.Mutex reserved map[string]time.Time } // PendingReservation 冷启动重建一笔预留所需的最小信息(中性结构,crypto 不依赖 store)。 type PendingReservation struct { AccountID string // 收款账户(用于解析地址,链上匹配维度) AmountMinor int64 // attempt 冻结的 base 金额(不含尾数) ProviderRef string // "CRYPTO--",用于恢复尾数 ReservedAt time.Time // 建单时间(= attempt.CreatedAt),冷却窗自此算 } // ReservationLoader 返回当前仍活跃(pending)的 crypto 预留。装配期由 main 用 OrderStore 实现。 type ReservationLoader func(ctx context.Context) ([]PendingReservation, error) 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 } } // 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 } ``` > `New` 里 `now: time.Now` 默认已存在,`WithNow` 允许测试覆盖。`import "context"` crypto.go 已有。 `internal/reconcile/crypto_warm.go`: ```go 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 } } ``` - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/provider/crypto/ ./internal/reconcile/ -v 2>&1 | tail -30` Expected: 全 PASS。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/provider/crypto/crypto.go internal/provider/crypto/crypto_test.go internal/reconcile/crypto_warm.go internal/reconcile/crypto_warm_test.go git commit -m "feat(v2): crypto 预留表冷启动兜底——注入 loader 从 pending attempts 重建 reservation(兜重启丢内存)" ``` --- ### Task 4: LimitAware 真实用量数据源(DBUsageSource + RefreshUsage job) **债务 #5。** P5 的 `accounts.LimitAware` 靠 `UsageSource.TodayUsedMinor(accountID)` 判越限,生产装的是 `NopUsage`(恒 0,退化为 round_robin)。P5 D5 约定:真实数据源由 P6 提供。本任务实现 `reconcile.UsageSource`(满足 `accounts.UsageSource` 接口):持一个「账户→当日已收 minor」快照,由对账 job 周期 `Refresh` 从 attempts 表聚合(`status=paid AND paid_at>=当日起点`,`GROUP BY account_id` 求 `SUM(amount_minor)`)。**量纲**:attempt.AmountMinor 即该账户所属渠道结算币种的 minor(config 注释已在),故按 account_id 直接求和与 `DailyLimit` 同量纲,可比。快照读写加锁,Pick 路径零 DB(读快照);启动预热 Refresh 一次,之后 job 刷。 **Files:** - Modify: `internal/store/order_query.go`(`SumPaidAttemptMinorByAccountSince`) - Modify: `internal/store/order_query_test.go`(断言聚合) - Create: `internal/reconcile/usage.go`(`UsageSource` 结构体 + `Refresh` + `RefreshUsageTask` + `StartOfDay`) - Create: `internal/reconcile/usage_test.go` **Interfaces:** - `store.(*OrderStore).SumPaidAttemptMinorByAccountSince(since time.Time) (map[string]int64, error)` - `reconcile.NewUsageSource(orders *store.OrderStore, now func() time.Time) *UsageSource`;`(*UsageSource).TodayUsedMinor(accountID string) int64`(满足 `accounts.UsageSource`);`(*UsageSource).Refresh(ctx) error`;`reconcile.RefreshUsageTask(u *UsageSource) func(ctx) error`。 - 当日起点用 `now()` 的日期(UTC)算(`StartOfDay`)——与 DailyLimit「按自然日」语义一致;跨区多实例各自 UTC 日窗,单实例部署无歧义。 - [ ] **Step 1: 写失败测试** `internal/store/order_query_test.go` 追加: ```go func TestSumPaidAttemptMinorByAccountSince(t *testing.T) { db := model.OpenTestDB(t) s := 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) } } ``` `internal/reconcile/usage_test.go`: ```go 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") } } ``` - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20` Expected: 编译失败——`SumPaidAttemptMinorByAccountSince` / `reconcile.NewUsageSource` 未定义。 - [ ] **Step 3: 写实现** `internal/store/order_query.go` 追加: ```go // 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 } ``` `internal/reconcile/usage.go`: ```go 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) } } ``` - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ ./internal/accounts/ -v 2>&1 | tail -30` Expected: 全 PASS(含 accounts 既有 limit_aware 测试仍绿)。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/usage.go internal/reconcile/usage_test.go git commit -m "feat(v2): LimitAware 真实用量源——DBUsageSource 快照 + 对账 job 周期聚合当日已收(替 NopUsage)" ``` --- ### Task 5: 对账主体 —— 周期 SyncPendingAttempts + 已付订单抽查 + main 装配 Runner **债务 #6(+ 继承 #7 已做的 settle-sync 日志)。** 对账主体两条腿:① **逐 pending attempt 查单收敛**——直接调度 P2 已有的 `gateway.SyncPendingAttempts`(内部已 query 每笔 pending、settle 命中、并对 not_found/mismatch/failed 打日志 = 债务 #7,764ed55);② **已付订单抽查核对**——对近期已付的 attempt 反向 `Query` 渠道,比对金额/币种,发现漂移(如渠道侧已退款/拒付但本地仍 paid)即记日志告警。**决策**:全量渠道对账单/流水列表下载(alipay `bill.download`/stripe `BalanceTransactions`)判定过重(拉大文件、解析对账单格式、跨日切分),本期降级为「逐 pending attempt query + 已付订单抽查」,已覆盖「防掉单(query 收敛)+ 防重复授权/漂移(抽查)」两大风险,论证见 Self-Review。最后把 Task 1–4 与本任务的 job 全部挂上 `reconcile.Runner`,接进 `main.go`。 **Files:** - Modify: `internal/store/order_query.go`(`ListRecentlyPaidAttempts`) - Modify: `internal/store/order_query_test.go` - Create: `internal/reconcile/sync.go`(`SyncPendingTask` + `PaidSpotCheckTask`) - Create: `internal/reconcile/sync_test.go` - Modify: `config/config.go`(`ReconcileConfig` + 默认值) - Modify: `main.go`(装配 Runner + 各 job + crypto Warm + DBUsageSource 注入 Router) **Interfaces:** - `store.(*OrderStore).ListRecentlyPaidAttempts(since time.Time, limit int) ([]model.Attempt, error)`(`status=paid AND paid_at>=since`)。 - `reconcile.SyncPendingTask(gw *gateway.Gateway, limit int) func(ctx) error`。 - `reconcile.PaidSpotCheckTask(orders *store.OrderStore, providers *provider.Registry, window time.Duration, now func() time.Time) func(ctx) error`。 - `config.ReconcileConfig{Enabled bool; OrderTTLMin, ExpireEverySec, SyncEverySec, UsageEverySec, SpotCheckEverySec, SpotCheckWindowMin int}`。 > 抽查只**日志告警不改状态**:退款/拒付的状态机翻转属 P4,本期仅发现并可见。 - [ ] **Step 1: 写失败测试** `internal/store/order_query_test.go` 追加: ```go func TestListRecentlyPaidAttempts(t *testing.T) { db := model.OpenTestDB(t) s := 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) } } ``` `internal/reconcile/sync_test.go`(用 fake provider + gateway,复用其装配式;这里给独立最小装配): ```go 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{} 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) } _ = time.Second } 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;漂移检测走日志 } } ``` - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20` Expected: 编译失败——`ListRecentlyPaidAttempts` / `SyncPendingTask` / `PaidSpotCheckTask` 未定义。 - [ ] **Step 3: 写实现** `internal/store/order_query.go` 追加: ```go // 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 } ``` `internal/reconcile/sync.go`: ```go 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 } } ``` `config/config.go` —— `Config` 加字段 + 结构体 + 默认值: ```go // Config 结构体加一行: QuerySync QuerySyncConfig `mapstructure:"query_sync"` Reconcile ReconcileConfig `mapstructure:"reconcile"` ``` ```go // ReconcileConfig v2 后台守护/对账周期任务开关与间隔。 type ReconcileConfig struct { Enabled bool `mapstructure:"enabled"` OrderTTLMin int `mapstructure:"order_ttl_min"` // pending 订单存活 TTL(分钟),超则关闭 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) } ``` `Load()` 里补默认(在 query_sync 默认附近): ```go 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) ``` `main.go` —— 装配(替换现有 notifier/Router/gateway 段并追加 Runner)。关键改动: 1. `import "context"` + `"github.com/wangjia/pay/internal/reconcile"` + `"github.com/wangjia/pay/internal/provider/crypto"`(crypto 仅用于 Warm 类型断言,见下 Task 6 亦需); 2. 用 `reconcile.NewUsageSource` 替 `accounts.NopUsage{}` 注入 Router; 3. crypto Warm(若注册了 crypto):给 `providerbuild.BuildRegistry` 传 loader,或建后取 crypto 实例 Warm。**装配式**:crypto 的 loader 需要 orderStore,故 crypto Provider 的构造要能带 `WithReservationLoader`。若 `providerbuild.BuildRegistry` 不便传 loader,改为 Warm 前对注册表里的 crypto 实例调 `Warm`——为此让 crypto 在 build 时就注入 loader。**最小改动**:在 `providerbuild` 暴露把 loader 传进 crypto.New 的入口,或 main 直接从 `pReg.Get("crypto")` 拿到 `*crypto.Provider` 调 `Warm`(loader 用 `reconcile.CryptoReservationLoader(orderStore)`,但 loader 是构造期注入的……)。**决策**:给 `crypto.(*Provider)` 的 loader 支持**构造后注入**——`WithReservationLoader` 已是 Option 构造期注入;为让 main 在 build 之后注入 loader,补一个 setter `SetReservationLoader(l)`。见下 Step 3a。 ```go orderStore := store.NewOrderStore(db) webhookStore := store.NewWebhookStore(db) notifier := webhook.NewNotifier(webhookStore, config.C.BizByName, func(no string) (bool, error) { o, err := orderStore.GetOrder(no) if err != nil { return false, err } return o.Status == model.OrderPaidV2, nil }) // 硬化选项走默认(退避/死信/告警);如需覆盖用 webhook.WithMaxAttempts 等 notifier.Start(60 * time.Second) productResolver := gateway.NewDBProductResolver(db) acctReg := accounts.New(config.C.Accounts) pReg := providerbuild.BuildRegistry(acctReg) usage := reconcile.NewUsageSource(orderStore, time.Now) // P6 真实用量源替 NopUsage acctPicker := accounts.NewRouter(acctReg, config.C.Routing, usage) gw := gateway.New(orderStore, pReg, acctPicker, productResolver, notifier, "cn") router.SetupV2(r, gw) // P6 后台守护 / 对账 if config.C.Reconcile.Enabled { rc := config.C.Reconcile runner := reconcile.NewRunner() runner.Add("order-expire", time.Duration(rc.ExpireEverySec)*time.Second, reconcile.OrderExpirerTask(orderStore, time.Duration(rc.OrderTTLMin)*time.Minute, time.Now)) runner.Add("usage-refresh", time.Duration(rc.UsageEverySec)*time.Second, reconcile.RefreshUsageTask(usage)) 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)) // crypto 孤儿扫描(Task 6)在此追加。 reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, rc) // Task 6 提供;含 Warm + 孤儿扫描注册 ctx := context.Background() runner.RunOnce(ctx) // 启动预热:先跑一遍(usage 快照/过期清理立即生效) runner.Start(ctx) log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/孤儿扫描)") } ``` > `reconcile.AddCryptoJobs` 与 crypto Warm/孤儿逻辑落在 **Task 6**;本任务先让 main 编译通过可**临时**用一个占位……**不**:禁占位。改为 Task 5 只装配前四个 job + `runner.RunOnce/Start`,把 `AddCryptoJobs` 这一行留到 Task 6 再加(Task 6 Step 3 明确「在 main 追加此行」)。故本 Step 3 的 main 片段**删掉 `AddCryptoJobs` 那行**,Task 6 再插回。 - [ ] **Step 3a: crypto loader 构造后注入(供 main Warm 用)** `internal/provider/crypto/crypto.go` 补 setter(Task 6 的 Warm 装配依赖): ```go // SetReservationLoader 构造后注入冷启动预留源(装配期 main 在 BuildRegistry 之后调用: // loader 依赖 OrderStore,而注册表构造不便传 store)。非并发安全,仅启动期单线程调用。 func (p *Provider) SetReservationLoader(l ReservationLoader) { p.loader = l } ``` - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ ./internal/gateway/ -v 2>&1 | tail -30` Expected: 全 PASS;`go build ./...`(main 装配)通过。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/sync.go internal/reconcile/sync_test.go internal/provider/crypto/crypto.go config/config.go main.go git commit -m "feat(v2): 对账主体——周期查单收敛 + 已付订单抽查 + main 装配 reconcile Runner(前 4 job)" ``` --- ### Task 6: crypto 孤儿链上支付发现(OrphanScanner + orphan 记录/告警) **债务 #3。** 到账但**不匹配任何 attempt** 的转账(付款人转错金额、手动转账、迟到到一笔金额早已滚出所有活跃单的旧款)必须被发现并记录,不能静默。canonical(pangolin-pay watcher)把这类记 `orphan_payments` 供人工对账。本任务:中性可选接口 `provider.OrphanScanner`,crypto 实现(扫地址近 window 确认到账,金额不在「近期任一 attempt 的期望金额集」内 → orphan);`reconcile` 收集 crypto 账户的已知期望金额、调 `ScanOrphans`、把 orphan 落 `model.OrphanPayment`(tx_id unique 去重)+ 告警。同时把 crypto Warm 也在此接进 main(`AddCryptoJobs`)。 **Files:** - Modify: `internal/provider/provider.go`(中性 `OrphanScanner` 接口 + `OrphanScanRequest`/`OrphanTransfer`/`KnownAttempt`) - Modify: `internal/provider/crypto/crypto.go`(实现 `ScanOrphans`) - Modify: `internal/provider/crypto/crypto_test.go`(httptest 假 TronGrid:一笔匹配 + 一笔孤儿) - Create: `internal/model/orphan.go`(`OrphanPayment` 表) - Modify: `main.go`(AutoMigrate 加表) - Create: `internal/store/orphan.go`(`OrphanStore.Record` 幂等) - Create: `internal/store/orphan_test.go` - Create: `internal/reconcile/orphan.go`(`AddCryptoJobs`:Warm + `OrphanScanTask`) - Create: `internal/reconcile/orphan_test.go` - Modify: `main.go`(追加 `reconcile.AddCryptoJobs(...)` 那行 + import crypto/context 已在 Task 5) **Interfaces:** - `provider.OrphanScanRequest{AccountID string; Since time.Time; Known []KnownAttempt}`;`provider.KnownAttempt{AmountMinor int64; ProviderRef string}`(期望金额 = base + tail,tail 由渠道自解 ref,pay 不算);`provider.OrphanTransfer{TxID string; AmountMinor int64; Currency string; At time.Time}`;`provider.OrphanScanner interface { ScanOrphans(ctx, OrphanScanRequest) ([]OrphanTransfer, error) }`。 - `store.(*OrderStore)` 复用 `ListAttemptsByChannelSince`(新增:任意状态、channel、created_at>=since)——arphan 的「已知期望金额集」含 pending/paid/expired 全部,凡 pay 合法签发过的金额都不算孤儿。 - `store.(*OrphanStore).Record(o *model.OrphanPayment) (bool, error)`(tx_id 冲突 no-op,返回是否新记)。 - `reconcile.AddCryptoJobs(runner *Runner, providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore, cfg config.ReconcileConfig)`。 - [ ] **Step 1: 写失败测试** `internal/store/order_query.go` 先补 `ListAttemptsByChannelSince`(Task 6 需要),测试放 order_query_test.go: ```go func TestListAttemptsByChannelSince(t *testing.T) { db := model.OpenTestDB(t) s := 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) } } ``` `internal/store/orphan_test.go`: ```go 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) } } ``` `internal/provider/crypto/crypto_test.go` 追加(假 TronGrid 返回两笔:一笔金额匹配 known、一笔不匹配 → 仅后者为 orphan): ```go 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) } } ``` `internal/reconcile/orphan_test.go`(loader→scan→record 全链;用上面的假 TronGrid 同构): ```go 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) } } ``` - [ ] **Step 2: 跑测试确认失败** Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/provider/crypto/ ./internal/reconcile/ 2>&1 | tail -20` Expected: 编译失败——`OrphanScanner`/`OrphanScanRequest`/`OrphanPayment`/`OrphanStore`/`OrphanScanTask`/`ListAttemptsByChannelSince` 未定义。 - [ ] **Step 3: 写实现** `internal/provider/provider.go` 追加(中性类型 + 可选接口,类比 RefundingProvider): ```go // ---- 对账:孤儿到账扫描(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) } ``` `internal/provider/crypto/crypto.go` 实现 `ScanOrphans`(复用 `Query` 里的 TronGrid 请求构造 + tailFromRef): ```go // 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 } ``` `internal/model/orphan.go`: ```go 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"` } ``` `main.go` 的 `autoMigrate` 参数表加 `&model.OrphanPayment{}`。 `internal/store/orphan.go`: ```go 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 } ``` `internal/store/order_query.go` 补 `ListAttemptsByChannelSince`: ```go // 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 } ``` `internal/reconcile/orphan.go`: ```go package reconcile import ( "context" "log" "time" "github.com/wangjia/pay/config" "github.com/wangjia/pay/internal/accounts" "github.com/wangjia/pay/internal/model" "github.com/wangjia/pay/internal/provider" "github.com/wangjia/pay/internal/provider/crypto" "github.com/wangjia/pay/internal/store" ) // OrphanScanTask 对每个 enabled crypto 账户扫链找孤儿到账(不匹配任何近期 attempt), // 落 OrphanStore + 首次记入时告警。渠道须实现 provider.OrphanScanner(crypto 实现)。 func OrphanScanTask(providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore, orphans *store.OrphanStore, window time.Duration, now func() time.Time) func(ctx context.Context) error { return func(ctx context.Context) error { prov, err := providers.Get("crypto") if err != nil { return nil // 未启用 crypto:无事可做 } scanner, ok := prov.(provider.OrphanScanner) if !ok { return nil } since := now().Add(-window) for _, acc := range accts.EnabledFor("crypto", "") { atts, err := orders.ListAttemptsByChannelSince("crypto", since, 200) if err != nil { return err } known := make([]provider.KnownAttempt, 0, len(atts)) for i := range atts { if atts[i].AccountID != acc.AccountID { continue } known = append(known, provider.KnownAttempt{AmountMinor: atts[i].AmountMinor, ProviderRef: atts[i].ProviderRef}) } found, err := scanner.ScanOrphans(ctx, provider.OrphanScanRequest{ AccountID: acc.AccountID, Since: since, Known: known, }) if err != nil { log.Printf("[reconcile] 孤儿扫描 account=%s: %v", acc.AccountID, err) continue // 单账户失败不阻断其它 } for _, o := range found { isNew, rerr := orphans.Record(&model.OrphanPayment{ Channel: "crypto", AccountID: acc.AccountID, TxID: o.TxID, AmountMinor: o.AmountMinor, Currency: o.Currency, DetectedAt: o.At, Note: "到账无主:不匹配任何近期 attempt 期望金额", }) if rerr != nil { log.Printf("[reconcile] 记录孤儿失败 tx=%s: %v", o.TxID, rerr) continue } if isNew { log.Printf("[reconcile][孤儿告警] channel=crypto account=%s tx=%s amount=%d %s", acc.AccountID, o.TxID, o.AmountMinor, o.Currency) } } } return nil } } // AddCryptoJobs 装配 crypto 相关后台任务:① 冷启动 Warm(注入 loader 后立即重建预留); // ② 周期孤儿扫描。若未启用 crypto 渠道则安全跳过。 func AddCryptoJobs(runner *Runner, providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore, orphans *store.OrphanStore, cfg config.ReconcileConfig) { prov, err := providers.Get("crypto") if err != nil { return } if cp, ok := prov.(*crypto.Provider); ok { cp.SetReservationLoader(CryptoReservationLoader(orders)) if werr := cp.Warm(context.Background()); werr != nil { // 起服务前重建预留 log.Printf("[reconcile] crypto 预留冷启动重建: %v", werr) } } runner.Add("crypto-orphan-scan", time.Duration(cfg.OrphanEverySec)*time.Second, OrphanScanTask(providers, accts, orders, orphans, time.Duration(cfg.OrphanWindowMin)*time.Minute, time.Now)) } ``` `main.go` —— 在 Task 5 装配段的注释处补一行(需 `orphanStore := store.NewOrphanStore(db)`): ```go orphanStore := store.NewOrphanStore(db) reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc) ``` (放在 `runner.RunOnce(ctx)` 之前,使 Warm 在起服务前完成、孤儿 job 一并进 runner。) - [ ] **Step 4: 跑测试确认通过** Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./... 2>&1 | tail -30` Expected: 全仓 PASS;`go build ./...` 通过。 - [ ] **Step 5: Commit** ```bash cd /Users/wangjia/code/pay git add internal/provider/provider.go internal/provider/crypto/crypto.go internal/provider/crypto/crypto_test.go internal/model/orphan.go internal/store/orphan.go internal/store/orphan_test.go internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/orphan.go internal/reconcile/orphan_test.go main.go git commit -m "feat(v2): crypto 孤儿到账发现——OrphanScanner 扫链核对 + orphan_payments 落表告警(对账兜底)" ``` --- ## Self-Review **Spec coverage(P6 范围,对照设计 §8「对账优先于实时」+ 三份前置计划 routed 到 P6 的债务):** - **订单级过期清理**(债务 #1,Task 1)= 超 TTL pending 单条件 UPDATE 关闭,含零尝试孤儿单;与 attempt 级 `ExpirePendingAttempts` 语义严格区分 ✓。 - **Notifier 硬化**(债务 #2,Task 2)= 指数退避(`base·2^(n-1)` 封顶)+ 最大次数死信(`dead` 列)+ 告警钩子(`WithAlerter`,默认 log);投递门禁(未付不投不计失败)与 unique 键幂等不破 ✓。 - **crypto 预留冷启动兜底**(债务 #4,Task 3)= 注入 `ReservationLoader` 从 pending crypto attempts 重建 `reserved`,兜重启丢内存;crypto 不 import store(中性结构 + 装配层桥)✓。 - **LimitAware 真实数据源**(债务 #5,Task 4)= `reconcile.UsageSource` 满足 `accounts.UsageSource`,对账 job 周期从 attempts 聚合当日已收(按账户 = 渠道结算币种 minor,与 DailyLimit 同量纲);Router 注入它替 `NopUsage`,`LimitAware`/`Router` 零改动(兑现 P5 D5)✓。 - **对账主体**(债务 #6,Task 5)= 周期 `SyncPendingAttempts`(查单收敛,防掉单;继承其 settle-sync 日志 = 债务 #7 已做,不重复)+ 已付订单抽查反查渠道(防重复授权/漂移)✓。 - **crypto 孤儿发现**(债务 #3,Task 6)= 中性 `OrphanScanner`,crypto 扫链找无主到账,落 `OrphanPayment`(tx_id unique 幂等)+ 首记告警 ✓。 **关键决策:** - **后台 job 统一形态**:`reconcile.Runner` 一任务一 ticker + `recover` + `RunOnce`(启动预热 + 单测入口);每任务幂等 sweep,崩溃后下 tick 自愈。时钟一律注入(`now func() time.Time`),cutoff/since/退避 Go 端算,SQL 不用时间函数(sqlite/mysql 可移植)。 - **对账深度(债务 #6 决策)**:**全量渠道对账单下载判定过重**——alipay `bill.download`(拉 CSV/ZIP、解析对账单格式、按结算日切分、处理手续费行)与 stripe `BalanceTransactions` 分页拉全量,工程量与运维面(存储/解析/对差)远超本期单实例小体量所需。**降级为「逐 pending attempt query(已存在)+ 已付订单抽查反查」**:前者防「掉单」(回调丢失→查单收敛),后者防「重复授权/状态漂移」(本地 paid 但渠道退款/拒付→日志告警)。crypto 因自托管无对账单,其「渠道流水核对」正是 Task 6 的扫链孤儿发现。这套已覆盖设计 §8 两大风险(防掉单 + 防重复授权),真需要全量对账单时再单开任务。 - **orphan「已知金额集」取全状态近窗 attempt**:pending/paid/expired 一律计入期望金额集——凡 pay 合法签发过的金额都不算孤儿(一笔匹配已 paid attempt 金额的到账 = 正常付款,非孤儿;匹配 pending/expired 的 = settle-sync 处理,非孤儿)。孤儿严格 = 「到账金额不在任何近窗 attempt 期望额内」(付错额/手动转/超窗旧款),精确对齐债务 #3 措辞。窗外旧款不扫,避免历史单反复误报。 - **抽查只发现不改状态**:退款/拒付的订单态机翻转属 P4,本期仅日志告警使其可见(不越权改 P4 的状态语义)。 - **Notifier options 向后兼容**:`NewNotifier(...opts)` 变参,现有 3 参调用不改;`ListUndelivered` 保留(测试断言用),Notifier 内部改用退避感知的 `ListDeliverable`。 **Provider 中性守恒:** `internal/gateway`、`internal/provider` 核心不 import `reconcile`;crypto 不 import `store`。孤儿经中性 `provider.OrphanScanner`/`OrphanScanRequest` 传递,`reconcile` type-assert 到具体 `*crypto.Provider` 仅在装配文件 `reconcile/orphan.go`(装配层,类比 main)。`reconcile` 依赖 store/gateway/accounts/provider/crypto/config 属装配层合理。 **幂等 / 崩溃安全审计:** 过期清理 = `status=pending` 守卫条件 UPDATE(并发 settle 翻转互斥,重跑 0 行);退避/死信 = 列状态持久化,重启续跑;usage = 快照幂等覆盖;孤儿 = tx_id unique ON CONFLICT DO NOTHING;Warm = 只加不覆盖更早到期、冷却过期跳过。全部可重跑、崩溃后自愈。 **Placeholder scan:** 无 TODO/占位;每任务给完整 Go 代码 + 测试 + 命令。Task 5 Step 3 明确把 `AddCryptoJobs` 那行留到 Task 6 插入(避免 Task 5 引用未定义符号),非占位——两任务顺序落地后 main 完整。 **已知取舍(记录,不阻塞 P6):** - **单实例假设**:crypto 预留 Warm + orphan 扫描 + usage 快照均**进程内 + 单 pay 实例**语义(P3 已定「同一收款地址单实例服务」)。多实例需共享预留存储 + 扫描去重锁 + 分布式选主,超本期(设计目标单 VPS 单实例)。 - **usage 快照有刷新延迟**:`UsageEverySec`(默认 60s)窗内新付款未计入,`LimitAware` 可能瞬时放行略超 DailyLimit 的单——DailyLimit 是「分摊避免单账户触风控」的软阈,非硬账,秒级漂移可接受;要严格实时可改 Pick 路径直查(本期不做,避免每单 N 次 DB)。 - **抽查/孤儿扫描 TronGrid limit=50**:单次拉最近 50 笔,超高频账户可能漏页——本期小体量足够,需要时加分页/游标。 - **Prometheus/metrics 体系排除**:告警仅 log 钩子(`WithAlerter` / `[孤儿告警]` / `[对账差异]` 日志),接 metrics/Telegram 属后续运维接线,不在 P6。 - **真实渠道对账单下载排除**(见上决策),alipay/stripe 的渠道侧流水核对靠抽查反查 query 覆盖。 **排除项(各自独立):** 退款态机 P4 · 多账户路由策略 P5(已完成)· 订阅/recurring P8 · metrics/告警外发体系(运维接线)。 > 契约同步(pay-contract):webhook `WebhookDelivery` 新增 `dead`/`next_attempt_at` 列(投递侧内部状态,不影响业务方 payload);新增 `orphan_payments` 表(内部对账,无对外接口);config 新增 `reconcile.*` 段。均向后兼容(AutoMigrate 只加不删)。