From 4ac4e6ca9f4da05ec54b3b7ff121e4db8125fb70 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 17:32:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20LimitAware=20=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E7=94=A8=E9=87=8F=E6=BA=90=E2=80=94=E2=80=94DBUsageSource=20?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=20+=20=E5=AF=B9=E8=B4=A6=20job=20=E5=91=A8?= =?UTF-8?q?=E6=9C=9F=E8=81=9A=E5=90=88=E5=BD=93=E6=97=A5=E5=B7=B2=E6=94=B6?= =?UTF-8?q?(=E6=9B=BF=20NopUsage)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/reconcile/usage.go | 55 ++++++++++++++++++++++++++++++ internal/reconcile/usage_test.go | 37 ++++++++++++++++++++ internal/store/order_query.go | 21 ++++++++++++ internal/store/order_query_test.go | 29 ++++++++++++++++ 4 files changed, 142 insertions(+) create mode 100644 internal/reconcile/usage.go create mode 100644 internal/reconcile/usage_test.go diff --git a/internal/reconcile/usage.go b/internal/reconcile/usage.go new file mode 100644 index 0000000..93fbab5 --- /dev/null +++ b/internal/reconcile/usage.go @@ -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) } +} diff --git a/internal/reconcile/usage_test.go b/internal/reconcile/usage_test.go new file mode 100644 index 0000000..f15fb2a --- /dev/null +++ b/internal/reconcile/usage_test.go @@ -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") + } +} diff --git a/internal/store/order_query.go b/internal/store/order_query.go index c306569..53884d0 100644 --- a/internal/store/order_query.go +++ b/internal/store/order_query.go @@ -126,3 +126,24 @@ func (s *OrderStore) ExpireStaleOrders(cutoff time.Time, limit int) (int64, erro } 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 +} diff --git a/internal/store/order_query_test.go b/internal/store/order_query_test.go index 40e509d..820714d 100644 --- a/internal/store/order_query_test.go +++ b/internal/store/order_query_test.go @@ -94,3 +94,32 @@ func TestExpireStaleOrders(t *testing.T) { 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) + } +}