feat(v2): LimitAware 真实用量源——DBUsageSource 快照 + 对账 job 周期聚合当日已收(替 NopUsage)
This commit is contained in:
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -126,3 +126,24 @@ func (s *OrderStore) ExpireStaleOrders(cutoff time.Time, limit int) (int64, erro
|
|||||||
}
|
}
|
||||||
return res.RowsAffected, nil
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -94,3 +94,32 @@ func TestExpireStaleOrders(t *testing.T) {
|
|||||||
t.Fatalf("重跑应 0, got %d", n2)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user