cache.go 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981
  1. package model
  2. import (
  3. "context"
  4. "encoding"
  5. "errors"
  6. "fmt"
  7. "math/rand/v2"
  8. "slices"
  9. "sort"
  10. "sync"
  11. "sync/atomic"
  12. "time"
  13. "github.com/bytedance/sonic"
  14. "github.com/labring/aiproxy/core/common"
  15. "github.com/labring/aiproxy/core/common/config"
  16. "github.com/labring/aiproxy/core/common/conv"
  17. "github.com/labring/aiproxy/core/common/notify"
  18. "github.com/maruel/natural"
  19. "github.com/redis/go-redis/v9"
  20. log "github.com/sirupsen/logrus"
  21. )
  22. const (
  23. SyncFrequency = time.Minute * 3
  24. TokenCacheKey = "token:%s"
  25. GroupCacheKey = "group:%s"
  26. GroupModelTPMKey = "group:%s:model_tpm"
  27. )
  28. var (
  29. _ encoding.BinaryMarshaler = (*redisStringSlice)(nil)
  30. _ redis.Scanner = (*redisStringSlice)(nil)
  31. )
  32. type redisStringSlice []string
  33. func (r *redisStringSlice) ScanRedis(value string) error {
  34. return sonic.Unmarshal(conv.StringToBytes(value), r)
  35. }
  36. func (r redisStringSlice) MarshalBinary() ([]byte, error) {
  37. return sonic.Marshal(r)
  38. }
  39. type redisTime time.Time
  40. var (
  41. _ redis.Scanner = (*redisTime)(nil)
  42. _ encoding.BinaryMarshaler = (*redisTime)(nil)
  43. )
  44. func (t *redisTime) ScanRedis(value string) error {
  45. return (*time.Time)(t).UnmarshalBinary(conv.StringToBytes(value))
  46. }
  47. func (t redisTime) MarshalBinary() ([]byte, error) {
  48. return time.Time(t).MarshalBinary()
  49. }
  50. type TokenCache struct {
  51. ExpiredAt redisTime `json:"expired_at" redis:"e"`
  52. Group string `json:"group" redis:"g"`
  53. Key string `json:"-" redis:"-"`
  54. Name string `json:"name" redis:"n"`
  55. Subnets redisStringSlice `json:"subnets" redis:"s"`
  56. Models redisStringSlice `json:"models" redis:"m"`
  57. ID int `json:"id" redis:"i"`
  58. Status int `json:"status" redis:"st"`
  59. Quota float64 `json:"quota" redis:"q"`
  60. UsedAmount float64 `json:"used_amount" redis:"u"`
  61. availableSets []string
  62. modelsBySet map[string][]string
  63. }
  64. func (t *TokenCache) SetAvailableSets(availableSets []string) {
  65. t.availableSets = availableSets
  66. }
  67. func (t *TokenCache) SetModelsBySet(modelsBySet map[string][]string) {
  68. t.modelsBySet = modelsBySet
  69. }
  70. func (t *TokenCache) ContainsModel(model string) bool {
  71. if len(t.Models) != 0 {
  72. if !slices.Contains(t.Models, model) {
  73. return false
  74. }
  75. }
  76. return containsModel(model, t.availableSets, t.modelsBySet)
  77. }
  78. func containsModel(model string, sets []string, modelsBySet map[string][]string) bool {
  79. for _, set := range sets {
  80. if slices.Contains(modelsBySet[set], model) {
  81. return true
  82. }
  83. }
  84. return false
  85. }
  86. func (t *TokenCache) Range(fn func(model string) bool) {
  87. ranged := make(map[string]struct{})
  88. if len(t.Models) != 0 {
  89. for _, model := range t.Models {
  90. if _, ok := ranged[model]; !ok && containsModel(model, t.availableSets, t.modelsBySet) {
  91. if !fn(model) {
  92. return
  93. }
  94. }
  95. ranged[model] = struct{}{}
  96. }
  97. return
  98. }
  99. for _, set := range t.availableSets {
  100. for _, model := range t.modelsBySet[set] {
  101. if _, ok := ranged[model]; !ok {
  102. if !fn(model) {
  103. return
  104. }
  105. }
  106. ranged[model] = struct{}{}
  107. }
  108. }
  109. }
  110. func (t *Token) ToTokenCache() *TokenCache {
  111. return &TokenCache{
  112. ID: t.ID,
  113. Group: t.GroupID,
  114. Key: t.Key,
  115. Name: t.Name.String(),
  116. Models: t.Models,
  117. Subnets: t.Subnets,
  118. Status: t.Status,
  119. ExpiredAt: redisTime(t.ExpiredAt),
  120. Quota: t.Quota,
  121. UsedAmount: t.UsedAmount,
  122. }
  123. }
  124. func CacheDeleteToken(key string) error {
  125. if !common.RedisEnabled {
  126. return nil
  127. }
  128. return common.RedisDel(fmt.Sprintf(TokenCacheKey, key))
  129. }
  130. //nolint:gosec
  131. func CacheSetToken(token *TokenCache) error {
  132. if !common.RedisEnabled {
  133. return nil
  134. }
  135. key := fmt.Sprintf(TokenCacheKey, token.Key)
  136. pipe := common.RDB.Pipeline()
  137. pipe.HSet(context.Background(), key, token)
  138. expireTime := SyncFrequency + time.Duration(rand.Int64N(60)-30)*time.Second
  139. pipe.Expire(context.Background(), key, expireTime)
  140. _, err := pipe.Exec(context.Background())
  141. return err
  142. }
  143. func CacheGetTokenByKey(key string) (*TokenCache, error) {
  144. if !common.RedisEnabled {
  145. token, err := GetTokenByKey(key)
  146. if err != nil {
  147. return nil, err
  148. }
  149. return token.ToTokenCache(), nil
  150. }
  151. cacheKey := fmt.Sprintf(TokenCacheKey, key)
  152. tokenCache := &TokenCache{}
  153. err := common.RDB.HGetAll(context.Background(), cacheKey).Scan(tokenCache)
  154. if err == nil && tokenCache.ID != 0 {
  155. tokenCache.Key = key
  156. return tokenCache, nil
  157. } else if err != nil && !errors.Is(err, redis.Nil) {
  158. log.Errorf("get token (%s) from redis error: %s", key, err.Error())
  159. }
  160. token, err := GetTokenByKey(key)
  161. if err != nil {
  162. return nil, err
  163. }
  164. tc := token.ToTokenCache()
  165. if err := CacheSetToken(tc); err != nil {
  166. log.Error("redis set token error: " + err.Error())
  167. }
  168. return tc, nil
  169. }
  170. var updateTokenUsedAmountOnlyIncreaseScript = redis.NewScript(`
  171. local used_amount = redis.call("HGet", KEYS[1], "ua")
  172. if used_amount == false then
  173. return redis.status_reply("ok")
  174. end
  175. if ARGV[1] < used_amount then
  176. return redis.status_reply("ok")
  177. end
  178. redis.call("HSet", KEYS[1], "ua", ARGV[1])
  179. return redis.status_reply("ok")
  180. `)
  181. func CacheUpdateTokenUsedAmountOnlyIncrease(key string, amount float64) error {
  182. if !common.RedisEnabled {
  183. return nil
  184. }
  185. return updateTokenUsedAmountOnlyIncreaseScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(TokenCacheKey, key)}, amount).Err()
  186. }
  187. var updateTokenNameScript = redis.NewScript(`
  188. if redis.call("HExists", KEYS[1], "n") then
  189. redis.call("HSet", KEYS[1], "n", ARGV[1])
  190. end
  191. return redis.status_reply("ok")
  192. `)
  193. func CacheUpdateTokenName(key string, name string) error {
  194. if !common.RedisEnabled {
  195. return nil
  196. }
  197. return updateTokenNameScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(TokenCacheKey, key)}, name).Err()
  198. }
  199. var updateTokenStatusScript = redis.NewScript(`
  200. if redis.call("HExists", KEYS[1], "st") then
  201. redis.call("HSet", KEYS[1], "st", ARGV[1])
  202. end
  203. return redis.status_reply("ok")
  204. `)
  205. func CacheUpdateTokenStatus(key string, status int) error {
  206. if !common.RedisEnabled {
  207. return nil
  208. }
  209. return updateTokenStatusScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(TokenCacheKey, key)}, status).Err()
  210. }
  211. type redisMap[K comparable, V any] map[K]V
  212. var (
  213. _ redis.Scanner = (*redisMap[string, any])(nil)
  214. _ encoding.BinaryMarshaler = (*redisMap[string, any])(nil)
  215. )
  216. func (r *redisMap[K, V]) ScanRedis(value string) error {
  217. return sonic.UnmarshalString(value, r)
  218. }
  219. func (r redisMap[K, V]) MarshalBinary() ([]byte, error) {
  220. return sonic.Marshal(r)
  221. }
  222. type (
  223. redisGroupModelConfigMap redisMap[string, GroupModelConfig]
  224. )
  225. type GroupCache struct {
  226. ID string `json:"-" redis:"-"`
  227. Status int `json:"status" redis:"st"`
  228. UsedAmount float64 `json:"used_amount" redis:"ua"`
  229. RPMRatio float64 `json:"rpm_ratio" redis:"rpm_r"`
  230. TPMRatio float64 `json:"tpm_ratio" redis:"tpm_r"`
  231. AvailableSets redisStringSlice `json:"available_sets" redis:"ass"`
  232. ModelConfigs redisGroupModelConfigMap `json:"model_configs" redis:"mc"`
  233. BalanceAlertEnabled bool `json:"balance_alert_enabled" redis:"bae"`
  234. BalanceAlertThreshold float64 `json:"balance_alert_threshold" redis:"bat"`
  235. }
  236. func (g *GroupCache) GetAvailableSets() []string {
  237. if len(g.AvailableSets) == 0 {
  238. return []string{ChannelDefaultSet}
  239. }
  240. return g.AvailableSets
  241. }
  242. func (g *Group) ToGroupCache() *GroupCache {
  243. modelConfigs := make(redisGroupModelConfigMap, len(g.ModelConfigs))
  244. for _, modelConfig := range g.ModelConfigs {
  245. modelConfigs[modelConfig.Model] = modelConfig
  246. }
  247. return &GroupCache{
  248. ID: g.ID,
  249. Status: g.Status,
  250. UsedAmount: g.UsedAmount,
  251. RPMRatio: g.RPMRatio,
  252. TPMRatio: g.TPMRatio,
  253. AvailableSets: g.AvailableSets,
  254. ModelConfigs: modelConfigs,
  255. BalanceAlertEnabled: g.BalanceAlertEnabled,
  256. BalanceAlertThreshold: g.BalanceAlertThreshold,
  257. }
  258. }
  259. func CacheDeleteGroup(id string) error {
  260. if !common.RedisEnabled {
  261. return nil
  262. }
  263. return common.RedisDel(fmt.Sprintf(GroupCacheKey, id))
  264. }
  265. var updateGroupRPMRatioScript = redis.NewScript(`
  266. if redis.call("HExists", KEYS[1], "rpm_r") then
  267. redis.call("HSet", KEYS[1], "rpm_r", ARGV[1])
  268. end
  269. return redis.status_reply("ok")
  270. `)
  271. func CacheUpdateGroupRPMRatio(id string, rpmRatio float64) error {
  272. if !common.RedisEnabled {
  273. return nil
  274. }
  275. return updateGroupRPMRatioScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(GroupCacheKey, id)}, rpmRatio).Err()
  276. }
  277. var updateGroupTPMRatioScript = redis.NewScript(`
  278. if redis.call("HExists", KEYS[1], "tpm_r") then
  279. redis.call("HSet", KEYS[1], "tpm_r", ARGV[1])
  280. end
  281. return redis.status_reply("ok")
  282. `)
  283. func CacheUpdateGroupTPMRatio(id string, tpmRatio float64) error {
  284. if !common.RedisEnabled {
  285. return nil
  286. }
  287. return updateGroupTPMRatioScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(GroupCacheKey, id)}, tpmRatio).Err()
  288. }
  289. var updateGroupStatusScript = redis.NewScript(`
  290. if redis.call("HExists", KEYS[1], "st") then
  291. redis.call("HSet", KEYS[1], "st", ARGV[1])
  292. end
  293. return redis.status_reply("ok")
  294. `)
  295. func CacheUpdateGroupStatus(id string, status int) error {
  296. if !common.RedisEnabled {
  297. return nil
  298. }
  299. return updateGroupStatusScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(GroupCacheKey, id)}, status).Err()
  300. }
  301. //nolint:gosec
  302. func CacheSetGroup(group *GroupCache) error {
  303. if !common.RedisEnabled {
  304. return nil
  305. }
  306. key := fmt.Sprintf(GroupCacheKey, group.ID)
  307. pipe := common.RDB.Pipeline()
  308. pipe.HSet(context.Background(), key, group)
  309. expireTime := SyncFrequency + time.Duration(rand.Int64N(60)-30)*time.Second
  310. pipe.Expire(context.Background(), key, expireTime)
  311. _, err := pipe.Exec(context.Background())
  312. return err
  313. }
  314. func CacheGetGroup(id string) (*GroupCache, error) {
  315. if !common.RedisEnabled {
  316. group, err := GetGroupByID(id, true)
  317. if err != nil {
  318. return nil, err
  319. }
  320. return group.ToGroupCache(), nil
  321. }
  322. cacheKey := fmt.Sprintf(GroupCacheKey, id)
  323. groupCache := &GroupCache{}
  324. err := common.RDB.HGetAll(context.Background(), cacheKey).Scan(groupCache)
  325. if err == nil && groupCache.Status != 0 {
  326. groupCache.ID = id
  327. return groupCache, nil
  328. } else if err != nil && !errors.Is(err, redis.Nil) {
  329. log.Errorf("get group (%s) from redis error: %s", id, err.Error())
  330. }
  331. group, err := GetGroupByID(id, true)
  332. if err != nil {
  333. return nil, err
  334. }
  335. gc := group.ToGroupCache()
  336. if err := CacheSetGroup(gc); err != nil {
  337. log.Error("redis set group error: " + err.Error())
  338. }
  339. return gc, nil
  340. }
  341. var updateGroupUsedAmountOnlyIncreaseScript = redis.NewScript(`
  342. local used_amount = redis.call("HGet", KEYS[1], "ua")
  343. if used_amount == false then
  344. return redis.status_reply("ok")
  345. end
  346. if ARGV[1] < used_amount then
  347. return redis.status_reply("ok")
  348. end
  349. redis.call("HSet", KEYS[1], "ua", ARGV[1])
  350. return redis.status_reply("ok")
  351. `)
  352. func CacheUpdateGroupUsedAmountOnlyIncrease(id string, amount float64) error {
  353. if !common.RedisEnabled {
  354. return nil
  355. }
  356. return updateGroupUsedAmountOnlyIncreaseScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(GroupCacheKey, id)}, amount).Err()
  357. }
  358. type GroupMCPCache struct {
  359. ID string `json:"id" redis:"i"`
  360. GroupID string `json:"group_id" redis:"g"`
  361. Status GroupMCPStatus `json:"status" redis:"s"`
  362. Type GroupMCPType `json:"type" redis:"t"`
  363. ProxyConfig *GroupMCPProxyConfig `json:"proxy_config" redis:"pc"`
  364. OpenAPIConfig *MCPOpenAPIConfig `json:"openapi_config" redis:"oc"`
  365. }
  366. func (g *GroupMCP) ToGroupMCPCache() *GroupMCPCache {
  367. return &GroupMCPCache{
  368. ID: g.ID,
  369. GroupID: g.GroupID,
  370. Status: g.Status,
  371. Type: g.Type,
  372. ProxyConfig: g.ProxyConfig,
  373. OpenAPIConfig: g.OpenAPIConfig,
  374. }
  375. }
  376. const (
  377. GroupMCPCacheKey = "group_mcp:%s:%s" // group_id:mcp_id
  378. )
  379. func CacheDeleteGroupMCP(groupID, mcpID string) error {
  380. if !common.RedisEnabled {
  381. return nil
  382. }
  383. return common.RedisDel(fmt.Sprintf(GroupMCPCacheKey, groupID, mcpID))
  384. }
  385. //nolint:gosec
  386. func CacheSetGroupMCP(groupMCP *GroupMCPCache) error {
  387. if !common.RedisEnabled {
  388. return nil
  389. }
  390. key := fmt.Sprintf(GroupMCPCacheKey, groupMCP.GroupID, groupMCP.ID)
  391. pipe := common.RDB.Pipeline()
  392. pipe.HSet(context.Background(), key, groupMCP)
  393. expireTime := SyncFrequency + time.Duration(rand.Int64N(60)-30)*time.Second
  394. pipe.Expire(context.Background(), key, expireTime)
  395. _, err := pipe.Exec(context.Background())
  396. return err
  397. }
  398. func CacheGetGroupMCP(groupID, mcpID string) (*GroupMCPCache, error) {
  399. if !common.RedisEnabled {
  400. groupMCP, err := GetGroupMCPByID(mcpID, groupID)
  401. if err != nil {
  402. return nil, err
  403. }
  404. return groupMCP.ToGroupMCPCache(), nil
  405. }
  406. cacheKey := fmt.Sprintf(GroupMCPCacheKey, groupID, mcpID)
  407. groupMCPCache := &GroupMCPCache{}
  408. err := common.RDB.HGetAll(context.Background(), cacheKey).Scan(groupMCPCache)
  409. if err == nil && groupMCPCache.ID != "" {
  410. return groupMCPCache, nil
  411. } else if err != nil && !errors.Is(err, redis.Nil) {
  412. log.Errorf("get group mcp (%s:%s) from redis error: %s", groupID, mcpID, err.Error())
  413. }
  414. groupMCP, err := GetGroupMCPByID(mcpID, groupID)
  415. if err != nil {
  416. return nil, err
  417. }
  418. gmc := groupMCP.ToGroupMCPCache()
  419. if err := CacheSetGroupMCP(gmc); err != nil {
  420. log.Error("redis set group mcp error: " + err.Error())
  421. }
  422. return gmc, nil
  423. }
  424. var updateGroupMCPStatusScript = redis.NewScript(`
  425. if redis.call("HExists", KEYS[1], "s") then
  426. redis.call("HSet", KEYS[1], "s", ARGV[1])
  427. end
  428. return redis.status_reply("ok")
  429. `)
  430. func CacheUpdateGroupMCPStatus(groupID, mcpID string, status GroupMCPStatus) error {
  431. if !common.RedisEnabled {
  432. return nil
  433. }
  434. return updateGroupMCPStatusScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(GroupMCPCacheKey, groupID, mcpID)}, status).Err()
  435. }
  436. type PublicMCPCache struct {
  437. ID string `json:"id" redis:"i"`
  438. Status PublicMCPStatus `json:"status" redis:"s"`
  439. Type PublicMCPType `json:"type" redis:"t"`
  440. Price MCPPrice `json:"price" redis:"p"`
  441. ProxyConfig *PublicMCPProxyConfig `json:"proxy_config" redis:"pc"`
  442. OpenAPIConfig *MCPOpenAPIConfig `json:"openapi_config" redis:"oc"`
  443. EmbedConfig *MCPEmbeddingConfig `json:"embed_config" redis:"ec"`
  444. }
  445. func (p *PublicMCP) ToPublicMCPCache() *PublicMCPCache {
  446. return &PublicMCPCache{
  447. ID: p.ID,
  448. Status: p.Status,
  449. Type: p.Type,
  450. Price: p.Price,
  451. ProxyConfig: p.ProxyConfig,
  452. OpenAPIConfig: p.OpenAPIConfig,
  453. EmbedConfig: p.EmbedConfig,
  454. }
  455. }
  456. const (
  457. PublicMCPCacheKey = "public_mcp:%s" // mcp_id
  458. )
  459. func CacheDeletePublicMCP(mcpID string) error {
  460. if !common.RedisEnabled {
  461. return nil
  462. }
  463. return common.RedisDel(fmt.Sprintf(PublicMCPCacheKey, mcpID))
  464. }
  465. //nolint:gosec
  466. func CacheSetPublicMCP(publicMCP *PublicMCPCache) error {
  467. if !common.RedisEnabled {
  468. return nil
  469. }
  470. key := fmt.Sprintf(PublicMCPCacheKey, publicMCP.ID)
  471. pipe := common.RDB.Pipeline()
  472. pipe.HSet(context.Background(), key, publicMCP)
  473. expireTime := SyncFrequency + time.Duration(rand.Int64N(60)-30)*time.Second
  474. pipe.Expire(context.Background(), key, expireTime)
  475. _, err := pipe.Exec(context.Background())
  476. return err
  477. }
  478. func CacheGetPublicMCP(mcpID string) (*PublicMCPCache, error) {
  479. if !common.RedisEnabled {
  480. publicMCP, err := GetPublicMCPByID(mcpID)
  481. if err != nil {
  482. return nil, err
  483. }
  484. return publicMCP.ToPublicMCPCache(), nil
  485. }
  486. cacheKey := fmt.Sprintf(PublicMCPCacheKey, mcpID)
  487. publicMCPCache := &PublicMCPCache{}
  488. err := common.RDB.HGetAll(context.Background(), cacheKey).Scan(publicMCPCache)
  489. if err == nil && publicMCPCache.ID != "" {
  490. return publicMCPCache, nil
  491. } else if err != nil && !errors.Is(err, redis.Nil) {
  492. log.Errorf("get public mcp (%s) from redis error: %s", mcpID, err.Error())
  493. }
  494. publicMCP, err := GetPublicMCPByID(mcpID)
  495. if err != nil {
  496. return nil, err
  497. }
  498. pmc := publicMCP.ToPublicMCPCache()
  499. if err := CacheSetPublicMCP(pmc); err != nil {
  500. log.Error("redis set public mcp error: " + err.Error())
  501. }
  502. return pmc, nil
  503. }
  504. var updatePublicMCPStatusScript = redis.NewScript(`
  505. if redis.call("HExists", KEYS[1], "s") then
  506. redis.call("HSet", KEYS[1], "s", ARGV[1])
  507. end
  508. return redis.status_reply("ok")
  509. `)
  510. func CacheUpdatePublicMCPStatus(mcpID string, status PublicMCPStatus) error {
  511. if !common.RedisEnabled {
  512. return nil
  513. }
  514. return updatePublicMCPStatusScript.Run(context.Background(), common.RDB, []string{fmt.Sprintf(PublicMCPCacheKey, mcpID)}, status).Err()
  515. }
  516. const (
  517. PublicMCPReusingParamCacheKey = "public_mcp_reusing_param:%s:%s" // mcp_id:group_id
  518. )
  519. type PublicMCPReusingParamCache struct {
  520. MCPID string `json:"mcp_id" redis:"m"`
  521. GroupID string `json:"group_id" redis:"g"`
  522. ReusingParams map[string]string `json:"reusing_params" redis:"rp"`
  523. }
  524. func (p *PublicMCPReusingParam) ToPublicMCPReusingParamCache() *PublicMCPReusingParamCache {
  525. return &PublicMCPReusingParamCache{
  526. MCPID: p.MCPID,
  527. GroupID: p.GroupID,
  528. ReusingParams: p.ReusingParams,
  529. }
  530. }
  531. func CacheDeletePublicMCPReusingParam(mcpID, groupID string) error {
  532. if !common.RedisEnabled {
  533. return nil
  534. }
  535. return common.RedisDel(fmt.Sprintf(PublicMCPReusingParamCacheKey, mcpID, groupID))
  536. }
  537. //nolint:gosec
  538. func CacheSetPublicMCPReusingParam(param *PublicMCPReusingParamCache) error {
  539. if !common.RedisEnabled {
  540. return nil
  541. }
  542. key := fmt.Sprintf(PublicMCPReusingParamCacheKey, param.MCPID, param.GroupID)
  543. pipe := common.RDB.Pipeline()
  544. pipe.HSet(context.Background(), key, param)
  545. expireTime := SyncFrequency + time.Duration(rand.Int64N(60)-30)*time.Second
  546. pipe.Expire(context.Background(), key, expireTime)
  547. _, err := pipe.Exec(context.Background())
  548. return err
  549. }
  550. func CacheGetPublicMCPReusingParam(mcpID, groupID string) (*PublicMCPReusingParamCache, error) {
  551. if !common.RedisEnabled {
  552. param, err := GetPublicMCPReusingParam(mcpID, groupID)
  553. if err != nil {
  554. return nil, err
  555. }
  556. return param.ToPublicMCPReusingParamCache(), nil
  557. }
  558. cacheKey := fmt.Sprintf(PublicMCPReusingParamCacheKey, mcpID, groupID)
  559. paramCache := &PublicMCPReusingParamCache{}
  560. err := common.RDB.HGetAll(context.Background(), cacheKey).Scan(paramCache)
  561. if err == nil && paramCache.MCPID != "" {
  562. return paramCache, nil
  563. } else if err != nil && !errors.Is(err, redis.Nil) {
  564. log.Errorf("get public mcp reusing param (%s:%s) from redis error: %s", mcpID, groupID, err.Error())
  565. }
  566. param, err := GetPublicMCPReusingParam(mcpID, groupID)
  567. if err != nil {
  568. return nil, err
  569. }
  570. prc := param.ToPublicMCPReusingParamCache()
  571. if err := CacheSetPublicMCPReusingParam(prc); err != nil {
  572. log.Error("redis set public mcp reusing param error: " + err.Error())
  573. }
  574. return prc, nil
  575. }
  576. //nolint:revive
  577. type ModelConfigCache interface {
  578. GetModelConfig(model string) (*ModelConfig, bool)
  579. }
  580. // read-only cache
  581. //
  582. //nolint:revive
  583. type ModelCaches struct {
  584. ModelConfig ModelConfigCache
  585. // map[set][]model
  586. EnabledModelsBySet map[string][]string
  587. // map[set][]modelconfig
  588. EnabledModelConfigsBySet map[string][]*ModelConfig
  589. // map[model]modelconfig
  590. EnabledModelConfigsMap map[string]*ModelConfig
  591. // map[set]map[model][]channel
  592. EnabledModel2ChannelsBySet map[string]map[string][]*Channel
  593. // map[set]map[model][]channel
  594. DisabledModel2ChannelsBySet map[string]map[string][]*Channel
  595. }
  596. var modelCaches atomic.Pointer[ModelCaches]
  597. func init() {
  598. modelCaches.Store(new(ModelCaches))
  599. }
  600. func LoadModelCaches() *ModelCaches {
  601. return modelCaches.Load()
  602. }
  603. // InitModelConfigAndChannelCache initializes the channel cache from database
  604. func InitModelConfigAndChannelCache() error {
  605. modelConfig, err := initializeModelConfigCache()
  606. if err != nil {
  607. return err
  608. }
  609. // Load enabled channels from database
  610. enabledChannels, err := LoadEnabledChannels()
  611. if err != nil {
  612. return err
  613. }
  614. // Build model to channels map by set
  615. enabledModel2ChannelsBySet := buildModelToChannelsBySetMap(enabledChannels)
  616. // Sort channels by priority within each set
  617. sortChannelsByPriorityBySet(enabledModel2ChannelsBySet)
  618. // Build enabled models and configs by set
  619. enabledModelsBySet, enabledModelConfigsBySet, enabledModelConfigsMap := buildEnabledModelsBySet(enabledModel2ChannelsBySet, modelConfig)
  620. // Load disabled channels
  621. disabledChannels, err := LoadDisabledChannels()
  622. if err != nil {
  623. return err
  624. }
  625. // Build disabled model to channels map by set
  626. disabledModel2ChannelsBySet := buildModelToChannelsBySetMap(disabledChannels)
  627. // Update global cache atomically
  628. modelCaches.Store(&ModelCaches{
  629. ModelConfig: modelConfig,
  630. EnabledModelsBySet: enabledModelsBySet,
  631. EnabledModelConfigsBySet: enabledModelConfigsBySet,
  632. EnabledModelConfigsMap: enabledModelConfigsMap,
  633. EnabledModel2ChannelsBySet: enabledModel2ChannelsBySet,
  634. DisabledModel2ChannelsBySet: disabledModel2ChannelsBySet,
  635. })
  636. return nil
  637. }
  638. func LoadEnabledChannels() ([]*Channel, error) {
  639. var channels []*Channel
  640. err := DB.Where("status = ?", ChannelStatusEnabled).Find(&channels).Error
  641. if err != nil {
  642. return nil, err
  643. }
  644. for _, channel := range channels {
  645. initializeChannelModels(channel)
  646. initializeChannelModelMapping(channel)
  647. }
  648. return channels, nil
  649. }
  650. func LoadDisabledChannels() ([]*Channel, error) {
  651. var channels []*Channel
  652. err := DB.Where("status = ?", ChannelStatusDisabled).Find(&channels).Error
  653. if err != nil {
  654. return nil, err
  655. }
  656. for _, channel := range channels {
  657. initializeChannelModels(channel)
  658. initializeChannelModelMapping(channel)
  659. }
  660. return channels, nil
  661. }
  662. func LoadChannels() ([]*Channel, error) {
  663. var channels []*Channel
  664. err := DB.Find(&channels).Error
  665. if err != nil {
  666. return nil, err
  667. }
  668. for _, channel := range channels {
  669. initializeChannelModels(channel)
  670. initializeChannelModelMapping(channel)
  671. }
  672. return channels, nil
  673. }
  674. func LoadChannelByID(id int) (*Channel, error) {
  675. var channel Channel
  676. err := DB.First(&channel, id).Error
  677. if err != nil {
  678. return nil, err
  679. }
  680. initializeChannelModels(&channel)
  681. initializeChannelModelMapping(&channel)
  682. return &channel, nil
  683. }
  684. var _ ModelConfigCache = (*modelConfigMapCache)(nil)
  685. type modelConfigMapCache struct {
  686. modelConfigMap map[string]*ModelConfig
  687. }
  688. func (m *modelConfigMapCache) GetModelConfig(model string) (*ModelConfig, bool) {
  689. config, ok := m.modelConfigMap[model]
  690. return config, ok
  691. }
  692. var _ ModelConfigCache = (*disabledModelConfigCache)(nil)
  693. type disabledModelConfigCache struct {
  694. modelConfigs ModelConfigCache
  695. }
  696. func (d *disabledModelConfigCache) GetModelConfig(model string) (*ModelConfig, bool) {
  697. if config, ok := d.modelConfigs.GetModelConfig(model); ok {
  698. return config, true
  699. }
  700. return NewDefaultModelConfig(model), true
  701. }
  702. func initializeModelConfigCache() (ModelConfigCache, error) {
  703. modelConfigs, err := GetAllModelConfigs()
  704. if err != nil {
  705. return nil, err
  706. }
  707. newModelConfigMap := make(map[string]*ModelConfig)
  708. for _, modelConfig := range modelConfigs {
  709. newModelConfigMap[modelConfig.Model] = modelConfig
  710. }
  711. configs := &modelConfigMapCache{modelConfigMap: newModelConfigMap}
  712. if config.GetDisableModelConfig() {
  713. return &disabledModelConfigCache{modelConfigs: configs}, nil
  714. }
  715. return configs, nil
  716. }
  717. func initializeChannelModels(channel *Channel) {
  718. if len(channel.Models) == 0 {
  719. channel.Models = config.GetDefaultChannelModels()[int(channel.Type)]
  720. return
  721. }
  722. findedModels, missingModels, err := GetModelConfigWithModels(channel.Models)
  723. if err != nil {
  724. return
  725. }
  726. if len(missingModels) > 0 {
  727. slices.Sort(missingModels)
  728. log.Errorf("model config not found: %v", missingModels)
  729. }
  730. slices.Sort(findedModels)
  731. channel.Models = findedModels
  732. }
  733. func initializeChannelModelMapping(channel *Channel) {
  734. if len(channel.ModelMapping) == 0 {
  735. channel.ModelMapping = config.GetDefaultChannelModelMapping()[int(channel.Type)]
  736. }
  737. }
  738. func buildModelToChannelsBySetMap(channels []*Channel) map[string]map[string][]*Channel {
  739. modelMapBySet := make(map[string]map[string][]*Channel)
  740. for _, channel := range channels {
  741. sets := channel.GetSets()
  742. for _, set := range sets {
  743. if _, ok := modelMapBySet[set]; !ok {
  744. modelMapBySet[set] = make(map[string][]*Channel)
  745. }
  746. for _, model := range channel.Models {
  747. modelMapBySet[set][model] = append(modelMapBySet[set][model], channel)
  748. }
  749. }
  750. }
  751. return modelMapBySet
  752. }
  753. func sortChannelsByPriorityBySet(modelMapBySet map[string]map[string][]*Channel) {
  754. for _, modelMap := range modelMapBySet {
  755. for _, channels := range modelMap {
  756. sort.Slice(channels, func(i, j int) bool {
  757. return channels[i].GetPriority() > channels[j].GetPriority()
  758. })
  759. }
  760. }
  761. }
  762. func buildEnabledModelsBySet(modelMapBySet map[string]map[string][]*Channel, modelConfigCache ModelConfigCache) (
  763. map[string][]string,
  764. map[string][]*ModelConfig,
  765. map[string]*ModelConfig,
  766. ) {
  767. modelsBySet := make(map[string][]string)
  768. modelConfigsBySet := make(map[string][]*ModelConfig)
  769. modelConfigsMap := make(map[string]*ModelConfig)
  770. for set, modelMap := range modelMapBySet {
  771. models := make([]string, 0)
  772. configs := make([]*ModelConfig, 0)
  773. appended := make(map[string]struct{})
  774. for model := range modelMap {
  775. if _, ok := appended[model]; ok {
  776. continue
  777. }
  778. if config, ok := modelConfigCache.GetModelConfig(model); ok {
  779. models = append(models, model)
  780. configs = append(configs, config)
  781. appended[model] = struct{}{}
  782. modelConfigsMap[model] = config
  783. }
  784. }
  785. slices.Sort(models)
  786. slices.SortStableFunc(configs, SortModelConfigsFunc)
  787. modelsBySet[set] = models
  788. modelConfigsBySet[set] = configs
  789. }
  790. return modelsBySet, modelConfigsBySet, modelConfigsMap
  791. }
  792. func SortModelConfigsFunc(i, j *ModelConfig) int {
  793. if i.Owner != j.Owner {
  794. if natural.Less(string(i.Owner), string(j.Owner)) {
  795. return -1
  796. }
  797. return 1
  798. }
  799. if i.Type != j.Type {
  800. if i.Type < j.Type {
  801. return -1
  802. }
  803. return 1
  804. }
  805. if i.Model == j.Model {
  806. return 0
  807. }
  808. if natural.Less(i.Model, j.Model) {
  809. return -1
  810. }
  811. return 1
  812. }
  813. func SyncModelConfigAndChannelCache(ctx context.Context, wg *sync.WaitGroup, frequency time.Duration) {
  814. defer wg.Done()
  815. ticker := time.NewTicker(frequency)
  816. defer ticker.Stop()
  817. for {
  818. select {
  819. case <-ctx.Done():
  820. return
  821. case <-ticker.C:
  822. err := InitModelConfigAndChannelCache()
  823. if err != nil {
  824. notify.ErrorThrottle("syncModelChannel", time.Minute, "failed to sync channels", err.Error())
  825. }
  826. }
  827. }
  828. }