diff --git a/config/config.go b/config/config.go index a1b02ed..c06430e 100644 --- a/config/config.go +++ b/config/config.go @@ -18,6 +18,9 @@ type Config struct { Biz map[string]BizSystemConfig `mapstructure:"biz"` // Accounts 多账户配置注册表(v2):凭证不写死配置,只存 env 前缀,真值运行时从环境变量取。 Accounts []AccountConfig `mapstructure:"accounts"` + // Routing 路由策略:channel → 策略名(round_robin/weighted/limit_aware)。 + // 缺省/未知一律当 round_robin(见 P5 计划 D3)。密钥无关,可写 config.yaml。 + Routing map[string]string `mapstructure:"routing"` } // AccountConfig 单个收款账户的配置(v2 多账户路由用)。凭证严禁写进 config.yaml, diff --git a/internal/accounts/limit_aware.go b/internal/accounts/limit_aware.go new file mode 100644 index 0000000..854888d --- /dev/null +++ b/internal/accounts/limit_aware.go @@ -0,0 +1,67 @@ +package accounts + +import ( + "sync" + + "github.com/wangjia/pay/config" +) + +// UsageSource 提供账户今日已收金额(minor),供 limit_aware 判定是否越 DailyLimit。 +// P5 内存实现(MemUsage)+ 生产空源(NopUsage);真实用量数据源由 P6 对账 job 提供, +// 届时实现本接口注入 Router 即可,LimitAware/Router 不改(见计划 D5)。 +type UsageSource interface { + TodayUsedMinor(accountID string) int64 +} + +// NopUsage 恒返回 0:P6 前生产装配用,limit_aware 退化为基础策略(不误拒)。 +type NopUsage struct{} + +func (NopUsage) TodayUsedMinor(string) int64 { return 0 } + +// MemUsage 进程内用量计数(测试确定性 / 简单场景)。 +type MemUsage struct { + mu sync.Mutex + used map[string]int64 +} + +func NewMemUsage() *MemUsage { return &MemUsage{used: map[string]int64{}} } + +func (m *MemUsage) Add(accountID string, minor int64) { + m.mu.Lock() + m.used[accountID] += minor + m.mu.Unlock() +} + +func (m *MemUsage) TodayUsedMinor(accountID string) int64 { + m.mu.Lock() + defer m.mu.Unlock() + return m.used[accountID] +} + +// LimitAware 过滤掉「DailyLimit>0 且 已用+本单 > DailyLimit」的账户,再把合格集交给 +// base 策略选(默认 round_robin)。全部越限 → ErrNoAccount(分摊避免单账户触风控冻结)。 +type LimitAware struct { + base Strategy + usage UsageSource +} + +func NewLimitAware(base Strategy, usage UsageSource) *LimitAware { + if usage == nil { + usage = NopUsage{} + } + return &LimitAware{base: base, usage: usage} +} + +func (l *LimitAware) Pick(key string, candidates []config.AccountConfig, hint PickHint) (config.AccountConfig, error) { + eligible := make([]config.AccountConfig, 0, len(candidates)) + for _, c := range candidates { + if c.DailyLimit > 0 && l.usage.TodayUsedMinor(c.AccountID)+hint.AmountMinor > c.DailyLimit { + continue + } + eligible = append(eligible, c) + } + if len(eligible) == 0 { + return config.AccountConfig{}, ErrNoAccount + } + return l.base.Pick(key, eligible, hint) +} diff --git a/internal/accounts/limit_aware_test.go b/internal/accounts/limit_aware_test.go new file mode 100644 index 0000000..eb1c395 --- /dev/null +++ b/internal/accounts/limit_aware_test.go @@ -0,0 +1,61 @@ +package accounts_test + +import ( + "errors" + "testing" + + "github.com/wangjia/pay/config" + "github.com/wangjia/pay/internal/accounts" +) + +func TestLimitAwareFiltersOverLimit(t *testing.T) { + usage := accounts.NewMemUsage() + usage.Add("a1", 9000) // a1 今日已用 9000 + la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage) + cs := []config.AccountConfig{ + {AccountID: "a1", DailyLimit: 10000}, // 9000+2000=11000 > 10000 → 排除 + {AccountID: "a2", DailyLimit: 10000}, // 0+2000 <= 10000 → 合格 + } + for i := 0; i < 3; i++ { + got, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 2000}) + if err != nil || got.AccountID != "a2" { + t.Fatalf("越限的 a1 应被过滤、只剩 a2, got %v err %v", got.AccountID, err) + } + } +} + +func TestLimitAwareZeroLimitIsUnlimited(t *testing.T) { + usage := accounts.NewMemUsage() + usage.Add("a1", 1_000_000) + la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage) + cs := []config.AccountConfig{{AccountID: "a1", DailyLimit: 0}} // 0=不限 + got, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 999}) + if err != nil || got.AccountID != "a1" { + t.Fatalf("DailyLimit=0 应不限, got %v err %v", got.AccountID, err) + } +} + +func TestLimitAwareAllOverLimit(t *testing.T) { + usage := accounts.NewMemUsage() + usage.Add("a1", 10000) + la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage) + cs := []config.AccountConfig{{AccountID: "a1", DailyLimit: 10000}} + if _, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 1}); !errors.Is(err, accounts.ErrNoAccount) { + t.Fatalf("全部越限应 ErrNoAccount, got %v", err) + } +} + +func TestNopUsageDegradesToBase(t *testing.T) { + la := accounts.NewLimitAware(accounts.NewRoundRobin(), accounts.NopUsage{}) + cs := []config.AccountConfig{ + {AccountID: "a1", DailyLimit: 100}, {AccountID: "a2", DailyLimit: 100}, + } + // 空用量源 → 谁都不越限 → 行为 = round_robin。 + // 注:AmountMinor 须 <= DailyLimit,否则即便 used=0 单笔也会越限(0+amount>limit), + // 那样测的是「单笔超限」而非「NopUsage 退化」,与用例名/注释矛盾——故取 10 而非 999。 + g1, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 10}) + g2, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 10}) + if g1.AccountID != "a1" || g2.AccountID != "a2" { + t.Fatalf("NopUsage 下应退化为轮询 a1,a2, got %s,%s", g1.AccountID, g2.AccountID) + } +} diff --git a/internal/accounts/router.go b/internal/accounts/router.go new file mode 100644 index 0000000..618db12 --- /dev/null +++ b/internal/accounts/router.go @@ -0,0 +1,66 @@ +package accounts + +import "github.com/wangjia/pay/config" + +// Router 是 gateway 面向的 Picker 实现:取候选(EnabledFor)→ 按 hint 排除 → 派发策略。 +// 策略实例进程内单例(round_robin 计数器需长存);未配置/未知策略默认 round_robin。 +type Router struct { + reg *Registry + strategies map[string]string // channel -> 策略名 + byName map[string]Strategy + def Strategy +} + +// NewRouter 装配路由。routing:channel→策略名(缺省/未知→round_robin);usage=nil→NopUsage。 +// limit_aware 复合 round_robin 作为合格集内的均摊基础策略。 +func NewRouter(reg *Registry, routing map[string]string, usage UsageSource) *Router { + rr := NewRoundRobin() + byName := map[string]Strategy{ + StrategyRoundRobin: rr, + StrategyWeighted: NewWeighted(), + StrategyLimitAware: NewLimitAware(rr, usage), + } + cp := map[string]string{} + for k, v := range routing { + cp[k] = v + } + return &Router{reg: reg, strategies: cp, byName: byName, def: rr} +} + +func (r *Router) strategyFor(channel string) Strategy { + if name, ok := r.strategies[channel]; ok { + if s, ok := r.byName[name]; ok { + return s + } + } + return r.def // 缺省 / 未知策略名 → round_robin(见计划 D3) +} + +// Pick 实现 accounts.Picker。 +func (r *Router) Pick(channel, region string, hint PickHint) (config.AccountConfig, error) { + cands := r.reg.EnabledFor(channel, region) + if len(cands) == 0 { + return config.AccountConfig{}, ErrNoAccount + } + if len(hint.ExcludeAccounts) > 0 { + if filtered := excludeAccounts(cands, hint.ExcludeAccounts); len(filtered) > 0 { + cands = filtered // 过滤到空则保留全集:单账户渠道 retry 仍可用(见计划 D4) + } + } + return r.strategyFor(channel).Pick(channel+"|"+region, cands, hint) +} + +func excludeAccounts(cands []config.AccountConfig, exclude []string) []config.AccountConfig { + skip := make(map[string]struct{}, len(exclude)) + for _, id := range exclude { + skip[id] = struct{}{} + } + out := make([]config.AccountConfig, 0, len(cands)) + for _, c := range cands { + if _, ok := skip[c.AccountID]; ok { + continue + } + out = append(out, c) + } + return out +} diff --git a/internal/accounts/router_test.go b/internal/accounts/router_test.go new file mode 100644 index 0000000..5a22a33 --- /dev/null +++ b/internal/accounts/router_test.go @@ -0,0 +1,55 @@ +package accounts_test + +import ( + "errors" + "testing" + + "github.com/wangjia/pay/config" + "github.com/wangjia/pay/internal/accounts" +) + +func newReg() *accounts.Registry { + return accounts.New([]config.AccountConfig{ + {AccountID: "f1", Channel: "fake", Region: "global", Enabled: true, Weight: 1}, + {AccountID: "f2", Channel: "fake", Region: "global", Enabled: true, Weight: 1}, + {AccountID: "f3", Channel: "fake", Region: "global", Enabled: false, Weight: 1}, // 禁用不参与 + {AccountID: "a1", Channel: "alipay", Region: "cn", Enabled: true, Weight: 1}, + }) +} + +func TestRouterDefaultRoundRobin(t *testing.T) { + r := accounts.NewRouter(newReg(), nil, nil) // 无 routing → 默认 round_robin + seq := []string{"f1", "f2", "f1"} + for i, w := range seq { + got, err := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-1"}) + if err != nil || got.AccountID != w { + t.Fatalf("call %d got %v err %v want %s", i, got.AccountID, err, w) + } + } +} + +func TestRouterExcludeSwitchesAccount(t *testing.T) { + r := accounts.NewRouter(newReg(), map[string]string{"fake": accounts.StrategyWeighted}, nil) + base, _ := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-RETRY"}) + // retry:排除首选账户 → 必换到另一个。 + got, err := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-RETRY", ExcludeAccounts: []string{base.AccountID}}) + if err != nil || got.AccountID == base.AccountID { + t.Fatalf("排除 %s 后应换账户, got %v err %v", base.AccountID, got.AccountID, err) + } +} + +func TestRouterExcludeEmptyFallsBackToAll(t *testing.T) { + r := accounts.NewRouter(newReg(), nil, nil) + // alipay/cn 只有 a1;排除 a1 后候选空 → 回退全集,仍返回 a1(单账户 retry 不失败)。 + got, err := r.Pick("alipay", "cn", accounts.PickHint{OutTradeNo: "PAY-2", ExcludeAccounts: []string{"a1"}}) + if err != nil || got.AccountID != "a1" { + t.Fatalf("排除到空应回退全集, got %v err %v", got.AccountID, err) + } +} + +func TestRouterNoAccount(t *testing.T) { + r := accounts.NewRouter(newReg(), nil, nil) + if _, err := r.Pick("wechat", "cn", accounts.PickHint{}); !errors.Is(err, accounts.ErrNoAccount) { + t.Fatalf("无账户渠道应 ErrNoAccount, got %v", err) + } +}