tiered_routing.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216
  1. package service
  2. import (
  3. "context"
  4. "encoding/json"
  5. "strings"
  6. "sync"
  7. "time"
  8. "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
  9. )
  10. // SettingKeyTieredRoutingSettings 分级路由规则(JSON,见 TieredRoutingSettings)。
  11. const SettingKeyTieredRoutingSettings = "tiered_routing_settings"
  12. // 账号池与用户等级:premium 用户优先 premium 池账号,无可用时回退全池。
  13. const (
  14. TierStandard = "standard"
  15. TierPremium = "premium"
  16. )
  17. // AccountPoolExtraKey 账号 extra 中的账号池标记键。
  18. const AccountPoolExtraKey = "pool"
  19. // TieredRoutingSettings 分级路由规则。
  20. type TieredRoutingSettings struct {
  21. Enabled bool `json:"enabled"`
  22. PremiumUserIDs []int64 `json:"premium_user_ids"`
  23. MinBalanceForPremium float64 `json:"min_balance_for_premium"`
  24. }
  25. // DefaultTieredRoutingSettings 默认关闭,需管理员显式启用并配置规则。
  26. func DefaultTieredRoutingSettings() TieredRoutingSettings {
  27. return TieredRoutingSettings{Enabled: false}
  28. }
  29. // TieredRoutingService 分级路由服务:规则快照缓存 + 用户等级判定。
  30. type TieredRoutingService struct {
  31. settingRepo SettingRepository
  32. mu sync.RWMutex
  33. snapshot *TieredRoutingSettings
  34. loadedAt time.Time
  35. }
  36. // tieredRoutingSnapshotTTL 规则快照 TTL。
  37. const tieredRoutingSnapshotTTL = 60 * time.Second
  38. // NewTieredRoutingService 创建分级路由服务。
  39. func NewTieredRoutingService(settingRepo SettingRepository) *TieredRoutingService {
  40. return &TieredRoutingService{settingRepo: settingRepo}
  41. }
  42. func (s *TieredRoutingService) currentSettings() TieredRoutingSettings {
  43. def := DefaultTieredRoutingSettings()
  44. if s == nil || s.settingRepo == nil {
  45. return def
  46. }
  47. s.mu.RLock()
  48. if s.snapshot != nil && time.Since(s.loadedAt) < tieredRoutingSnapshotTTL {
  49. out := *s.snapshot
  50. s.mu.RUnlock()
  51. return out
  52. }
  53. s.mu.RUnlock()
  54. s.mu.Lock()
  55. defer s.mu.Unlock()
  56. if s.snapshot != nil && time.Since(s.loadedAt) < tieredRoutingSnapshotTTL {
  57. return *s.snapshot
  58. }
  59. cfg := def
  60. ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
  61. defer cancel()
  62. if raw, err := s.settingRepo.GetValue(ctx, SettingKeyTieredRoutingSettings); err == nil && strings.TrimSpace(raw) != "" {
  63. var parsed TieredRoutingSettings
  64. if json.Unmarshal([]byte(raw), &parsed) == nil {
  65. cfg = normalizeTieredRoutingSettings(parsed)
  66. }
  67. }
  68. s.snapshot = &cfg
  69. s.loadedAt = time.Now()
  70. return cfg
  71. }
  72. func normalizeTieredRoutingSettings(in TieredRoutingSettings) TieredRoutingSettings {
  73. ids := make([]int64, 0, len(in.PremiumUserIDs))
  74. seen := make(map[int64]bool, len(in.PremiumUserIDs))
  75. for _, id := range in.PremiumUserIDs {
  76. if id > 0 && !seen[id] {
  77. seen[id] = true
  78. ids = append(ids, id)
  79. }
  80. }
  81. in.PremiumUserIDs = ids
  82. if in.MinBalanceForPremium < 0 {
  83. in.MinBalanceForPremium = 0
  84. }
  85. return in
  86. }
  87. // Invalidate 清空规则快照(写操作后调用)。
  88. func (s *TieredRoutingService) Invalidate() {
  89. if s == nil {
  90. return
  91. }
  92. s.mu.Lock()
  93. s.snapshot = nil
  94. s.loadedAt = time.Time{}
  95. s.mu.Unlock()
  96. }
  97. // GetSettings 读取当前规则。
  98. func (s *TieredRoutingService) GetSettings(_ context.Context) TieredRoutingSettings {
  99. if s == nil {
  100. return DefaultTieredRoutingSettings()
  101. }
  102. return s.currentSettings()
  103. }
  104. // UpdateSettings 保存规则。
  105. func (s *TieredRoutingService) UpdateSettings(ctx context.Context, in TieredRoutingSettings) (TieredRoutingSettings, error) {
  106. if s == nil || s.settingRepo == nil {
  107. return DefaultTieredRoutingSettings(), nil
  108. }
  109. norm := normalizeTieredRoutingSettings(in)
  110. raw, err := json.Marshal(norm)
  111. if err != nil {
  112. return norm, err
  113. }
  114. if err := s.settingRepo.Set(ctx, SettingKeyTieredRoutingSettings, string(raw)); err != nil {
  115. return norm, err
  116. }
  117. s.Invalidate()
  118. return norm, nil
  119. }
  120. // ResolveUserTier 判定用户等级:显式名单或余额达标即 premium,否则 standard。
  121. // 未启用时恒返回 standard(关闭即无行为变化)。
  122. func (s *TieredRoutingService) ResolveUserTier(user *User) string {
  123. if s == nil || user == nil {
  124. return TierStandard
  125. }
  126. cfg := s.currentSettings()
  127. if !cfg.Enabled {
  128. return TierStandard
  129. }
  130. return resolveTierByRules(user, cfg)
  131. }
  132. func resolveTierByRules(user *User, cfg TieredRoutingSettings) string {
  133. if user == nil {
  134. return TierStandard
  135. }
  136. for _, id := range cfg.PremiumUserIDs {
  137. if user.ID == id {
  138. return TierPremium
  139. }
  140. }
  141. if cfg.MinBalanceForPremium > 0 && user.Balance >= cfg.MinBalanceForPremium {
  142. return TierPremium
  143. }
  144. return TierStandard
  145. }
  146. // ResolvePool 供中间件调用:解析用户等级对应的目标账号池。
  147. func (s *TieredRoutingService) ResolvePool(user *User) string {
  148. return s.ResolveUserTier(user)
  149. }
  150. // AccountPool 读取账号所属池:extra.pool 显式标记 premium 才进 premium 池。
  151. func AccountPool(account *Account) string {
  152. if account == nil || len(account.Extra) == 0 {
  153. return TierStandard
  154. }
  155. if v, ok := account.Extra[AccountPoolExtraKey]; ok {
  156. if str, ok := v.(string); ok && strings.EqualFold(strings.TrimSpace(str), TierPremium) {
  157. return TierPremium
  158. }
  159. }
  160. return TierStandard
  161. }
  162. // WithTierPool 把目标账号池写入 ctx(认证中间件在鉴权成功后调用一次)。
  163. func WithTierPool(ctx context.Context, pool string) context.Context {
  164. return context.WithValue(ctx, ctxkey.TierPool, pool)
  165. }
  166. // TierPoolFromContext 读取目标账号池,缺省 standard(不过滤)。
  167. func TierPoolFromContext(ctx context.Context) string {
  168. if ctx == nil {
  169. return TierStandard
  170. }
  171. if pool, ok := ctx.Value(ctxkey.TierPool).(string); ok && pool == TierPremium {
  172. return TierPremium
  173. }
  174. return TierStandard
  175. }
  176. // FilterTierPoolAccounts 按目标池过滤候选账号:premium 用户优先 premium 池;
  177. // 池内无可用时回退全量(永不 fail-closed)。
  178. func FilterTierPoolAccounts(accounts []Account, wantPool string) []Account {
  179. if wantPool != TierPremium || len(accounts) == 0 {
  180. return accounts
  181. }
  182. matched := make([]Account, 0, len(accounts))
  183. for _, acc := range accounts {
  184. if AccountPool(&acc) == TierPremium {
  185. matched = append(matched, acc)
  186. }
  187. }
  188. if len(matched) == 0 {
  189. return accounts
  190. }
  191. return matched
  192. }