04c3f4f31a
Task 5(债务 #6):两条腿收敛对账——① SyncPendingTask 调度 P2 gateway.SyncPendingAttempts 逐 pending attempt 查单收敛(防掉单);② PaidSpotCheckTask 对近期已付 attempt 反查渠道 核对金额/币种,漂移(渠道侧已退款/拒付而本地仍 paid)只告警不改状态。 追加两条 P4 T3 opus review 义务(该 review 产出时本 worktree 已分叉,P4 退款主体在主 checkout;这里对本 worktree 已有的 store.RefundStore/OrderStore 接口——P4 T2,在 base 里——建自愈扫描,设计为可在合并 P4 后继续工作): - RefundApplyTask:退款修复扫描,重算 succeeded 退款之和,自愈「退款成功但订单卡 paid」 的崩溃窗口(MarkRefundStatus 翻 succeeded 后、ApplyRefundToOrder 调用前崩溃)。候选订单 =有 succeeded 退款的订单 ∪ 当前处于 refunding/partially_refunded 态的订单;走既有 ApplyRefundToOrder 条件 UPDATE,目标态与当前态一致时跳过,天然幂等。 - RefundStuckAlertTask:卡滞 processing/manual_pending 退款超阈值(默认 30min)打 WARN, 只观测不改状态;渠道退款查询 API 面留待后续。 main 装配前 4 job(order-expire/usage-refresh/sync-pending/paid-spotcheck)+ 上述两个退款 job 挂上 reconcile.Runner;acctPicker 的 limit_aware 用量源改用 reconcile.NewUsageSource 替 NopUsage。crypto 冷启动 Warm/孤儿扫描(AddCryptoJobs)留给 Task 6 追加。 config.go 新增 ReconcileConfig(含 Task 6 预留的 orphan_* 字段)+ 默认值;crypto.go 补 SetReservationLoader 构造后注入 setter(Task 6 依赖)。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u
177 lines
6.7 KiB
Go
177 lines
6.7 KiB
Go
package store
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"github.com/wangjia/pay/internal/model"
|
|
)
|
|
|
|
var (
|
|
ErrOrderNotFound = errors.New("store: order not found")
|
|
ErrAttemptNotFound = errors.New("store: attempt not found")
|
|
)
|
|
|
|
// GetOrder returns an order by out_trade_no.
|
|
func (s *OrderStore) GetOrder(outTradeNo string) (*model.OrderV2, error) {
|
|
var o model.OrderV2
|
|
if err := s.db.Where("out_trade_no = ?", outTradeNo).First(&o).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, ErrOrderNotFound
|
|
}
|
|
return nil, fmt.Errorf("store.GetOrder: %w", err)
|
|
}
|
|
return &o, nil
|
|
}
|
|
|
|
// AttemptByProviderRef resolves an attempt from a bare provider_ref, so settlement
|
|
// can recover out_trade_no + channel from a callback/query that only carries the ref.
|
|
func (s *OrderStore) AttemptByProviderRef(providerRef string) (*model.Attempt, error) {
|
|
var a model.Attempt
|
|
if err := s.db.Where("provider_ref = ?", providerRef).First(&a).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, ErrAttemptNotFound
|
|
}
|
|
return nil, fmt.Errorf("store.AttemptByProviderRef: %w", err)
|
|
}
|
|
return &a, nil
|
|
}
|
|
|
|
// ListAttemptsByStatus lists attempts in a status (for query-sync fallback).
|
|
func (s *OrderStore) ListAttemptsByStatus(status model.AttemptStatus, limit int) ([]model.Attempt, error) {
|
|
if limit <= 0 || limit > 200 {
|
|
limit = 100
|
|
}
|
|
var out []model.Attempt
|
|
if err := s.db.Where("status = ?", status).Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
|
return nil, fmt.Errorf("store.ListAttemptsByStatus: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ExpirePendingAttempts marks all pending attempts of an order as expired
|
|
// (used before a retry spawns a fresh attempt). Order status is untouched.
|
|
func (s *OrderStore) ExpirePendingAttempts(outTradeNo string) (int64, error) {
|
|
res := s.db.Model(&model.Attempt{}).
|
|
Where("out_trade_no = ? AND status = ?", outTradeNo, model.AttemptPending).
|
|
Update("status", model.AttemptExpired)
|
|
if res.Error != nil {
|
|
return 0, fmt.Errorf("store.ExpirePendingAttempts: %w", res.Error)
|
|
}
|
|
return res.RowsAffected, nil
|
|
}
|
|
|
|
// PaidAttempt returns the settled (paid) attempt of an order — the payment a
|
|
// refund reverses (channel + provider_ref for the original transaction).
|
|
func (s *OrderStore) PaidAttempt(outTradeNo string) (*model.Attempt, error) {
|
|
var a model.Attempt
|
|
err := s.db.Where("out_trade_no = ? AND status = ?", outTradeNo, model.AttemptPaid).First(&a).Error
|
|
if err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, ErrAttemptNotFound
|
|
}
|
|
return nil, fmt.Errorf("store.PaidAttempt: %w", err)
|
|
}
|
|
return &a, nil
|
|
}
|
|
|
|
// ApplyRefundToOrder advances a settled order's status per cumulative refunds:
|
|
// fully refunded → refunded, else → partially_refunded. Guard: only from a
|
|
// post-paid, non-fully-refunded state (paid / partially_refunded / refunding).
|
|
func (s *OrderStore) ApplyRefundToOrder(outTradeNo string, fully bool) (bool, error) {
|
|
next := model.OrderPartRefundedV2
|
|
if fully {
|
|
next = model.OrderRefundedV2
|
|
}
|
|
res := s.db.Model(&model.OrderV2{}).
|
|
Where("out_trade_no = ? AND status IN ?", outTradeNo,
|
|
[]model.OrderStatusV2{model.OrderPaidV2, model.OrderPartRefundedV2, model.OrderRefundingV2}).
|
|
Update("status", next)
|
|
if res.Error != nil {
|
|
return false, fmt.Errorf("store.ApplyRefundToOrder: %w", res.Error)
|
|
}
|
|
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
|
|
}
|