channel_service.go 40 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172
  1. package service
  2. import (
  3. "context"
  4. "fmt"
  5. "log/slog"
  6. "strings"
  7. "sync/atomic"
  8. "time"
  9. "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
  10. infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
  11. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  12. "github.com/tidwall/gjson"
  13. "github.com/tidwall/sjson"
  14. "golang.org/x/sync/singleflight"
  15. )
  16. var (
  17. ErrChannelNotFound = infraerrors.NotFound("CHANNEL_NOT_FOUND", "channel not found")
  18. ErrChannelExists = infraerrors.Conflict("CHANNEL_EXISTS", "channel name already exists")
  19. ErrGroupAlreadyInChannel = infraerrors.Conflict(
  20. "GROUP_ALREADY_IN_CHANNEL",
  21. "one or more groups already belong to another channel",
  22. )
  23. )
  24. // ChannelRepository 渠道数据访问接口
  25. type ChannelRepository interface {
  26. Create(ctx context.Context, channel *Channel) error
  27. GetByID(ctx context.Context, id int64) (*Channel, error)
  28. Update(ctx context.Context, channel *Channel) error
  29. Delete(ctx context.Context, id int64) error
  30. List(ctx context.Context, params pagination.PaginationParams, status, search string) ([]Channel, *pagination.PaginationResult, error)
  31. ListAll(ctx context.Context) ([]Channel, error)
  32. ExistsByName(ctx context.Context, name string) (bool, error)
  33. ExistsByNameExcluding(ctx context.Context, name string, excludeID int64) (bool, error)
  34. // 分组关联
  35. GetGroupIDs(ctx context.Context, channelID int64) ([]int64, error)
  36. SetGroupIDs(ctx context.Context, channelID int64, groupIDs []int64) error
  37. GetChannelIDByGroupID(ctx context.Context, groupID int64) (int64, error)
  38. GetGroupsInOtherChannels(ctx context.Context, channelID int64, groupIDs []int64) ([]int64, error)
  39. // 分组平台查询
  40. GetGroupPlatforms(ctx context.Context, groupIDs []int64) (map[int64]string, error)
  41. // 模型定价
  42. ListModelPricing(ctx context.Context, channelID int64) ([]ChannelModelPricing, error)
  43. CreateModelPricing(ctx context.Context, pricing *ChannelModelPricing) error
  44. UpdateModelPricing(ctx context.Context, pricing *ChannelModelPricing) error
  45. DeleteModelPricing(ctx context.Context, id int64) error
  46. ReplaceModelPricing(ctx context.Context, channelID int64, pricingList []ChannelModelPricing) error
  47. }
  48. // channelModelKey 渠道缓存复合键(显式包含 platform 防止跨平台同名模型冲突)
  49. type channelModelKey struct {
  50. groupID int64
  51. platform string // 平台标识
  52. model string // lowercase
  53. }
  54. // normalizeChannelPricingModelName makes Anthropic's dot and hyphen spelling
  55. // differences equivalent in channel pricing cache keys.
  56. func normalizeChannelPricingModelName(model string) string {
  57. model = strings.ToLower(strings.TrimSpace(model))
  58. if strings.HasPrefix(model, "claude-") {
  59. model = strings.ReplaceAll(model, ".", "-")
  60. }
  61. return model
  62. }
  63. // channelGroupPlatformKey 通配符定价缓存键
  64. type channelGroupPlatformKey struct {
  65. groupID int64
  66. platform string
  67. }
  68. // wildcardPricingEntry 通配符定价条目
  69. type wildcardPricingEntry struct {
  70. prefix string
  71. pricing *ChannelModelPricing
  72. }
  73. // wildcardMappingEntry 通配符映射条目
  74. type wildcardMappingEntry struct {
  75. prefix string
  76. target string
  77. }
  78. // channelCache 渠道缓存快照(扁平化哈希结构,热路径 O(1) 查找)
  79. type channelCache struct {
  80. // 热路径查找
  81. pricingByGroupModel map[channelModelKey]*ChannelModelPricing // (groupID, platform, model) → 定价
  82. wildcardByGroupPlatform map[channelGroupPlatformKey][]*wildcardPricingEntry // (groupID, platform) → 通配符定价(按配置顺序,先匹配先使用)
  83. mappingByGroupModel map[channelModelKey]string // (groupID, platform, model) → 映射目标
  84. wildcardMappingByGP map[channelGroupPlatformKey][]*wildcardMappingEntry // (groupID, platform) → 通配符映射(按配置顺序,先匹配先使用)
  85. channelByGroupID map[int64]*Channel // groupID → 渠道
  86. groupPlatform map[int64]string // groupID → platform
  87. // 冷路径(CRUD 操作)
  88. byID map[int64]*Channel
  89. loadedAt time.Time
  90. }
  91. // ChannelMappingResult 渠道映射查找结果
  92. type ChannelMappingResult struct {
  93. MappedModel string // 映射后的模型名(无映射时等于原始模型名)
  94. ChannelID int64 // 渠道 ID(0 = 无渠道关联)
  95. Mapped bool // 是否发生了映射
  96. BillingModelSource string // 计费模型来源("requested" / "upstream" / "channel_mapped" / "response_model")
  97. }
  98. // BuildModelMappingChain 根据映射结果和上游实际模型构建映射链描述。
  99. // reqModel: 客户端请求的原始模型名。
  100. // upstreamModel: 上游实际使用的模型名(ForwardResult.UpstreamModel)。
  101. // 返回空字符串表示无映射。
  102. func (r ChannelMappingResult) BuildModelMappingChain(reqModel, upstreamModel string) string {
  103. if !r.Mapped {
  104. if upstreamModel != "" && upstreamModel != reqModel {
  105. return reqModel + "→" + upstreamModel
  106. }
  107. return ""
  108. }
  109. if upstreamModel != "" && upstreamModel != r.MappedModel {
  110. return reqModel + "→" + r.MappedModel + "→" + upstreamModel
  111. }
  112. return reqModel + "→" + r.MappedModel
  113. }
  114. // ToUsageFields 将渠道映射结果转为使用记录字段
  115. func (r ChannelMappingResult) ToUsageFields(reqModel, upstreamModel string) ChannelUsageFields {
  116. channelMappedModel := reqModel
  117. if r.Mapped {
  118. channelMappedModel = r.MappedModel
  119. }
  120. return ChannelUsageFields{
  121. ChannelID: r.ChannelID,
  122. OriginalModel: reqModel,
  123. ChannelMappedModel: channelMappedModel,
  124. BillingModelSource: r.BillingModelSource,
  125. ModelMappingChain: r.BuildModelMappingChain(reqModel, upstreamModel),
  126. }
  127. }
  128. const (
  129. channelCacheTTL = 10 * time.Minute
  130. channelErrorTTL = 5 * time.Second // DB 错误时的短缓存
  131. channelCacheDBTimeout = 10 * time.Second
  132. channelCacheNotifyTTL = 3 * time.Second
  133. )
  134. // ChannelCachePubSub broadcasts channel cache invalidations between instances.
  135. type ChannelCachePubSub interface {
  136. NotifyUpdate(ctx context.Context) error
  137. SubscribeUpdates(ctx context.Context, handler func())
  138. }
  139. // ChannelService 渠道管理服务
  140. type ChannelService struct {
  141. repo ChannelRepository
  142. groupRepo GroupRepository
  143. authCacheInvalidator APIKeyAuthCacheInvalidator
  144. pricingService *PricingService // 用于「可用渠道」展示时回落到全局定价;可为 nil(测试场景)
  145. cachePubSub ChannelCachePubSub
  146. cache atomic.Value // *channelCache
  147. cacheSF singleflight.Group
  148. }
  149. // NewChannelService 创建渠道服务实例。
  150. // pricingService 仅供 ListAvailable 在渠道未配置定价时回落到全局 LiteLLM 数据;
  151. // 计费热路径走独立的 ModelPricingResolver,与此参数无关。可传 nil。
  152. func NewChannelService(repo ChannelRepository, groupRepo GroupRepository, authCacheInvalidator APIKeyAuthCacheInvalidator, pricingService *PricingService, cachePubSub ChannelCachePubSub) *ChannelService {
  153. s := &ChannelService{
  154. repo: repo,
  155. groupRepo: groupRepo,
  156. authCacheInvalidator: authCacheInvalidator,
  157. pricingService: pricingService,
  158. cachePubSub: cachePubSub,
  159. }
  160. s.subscribeCacheUpdates(context.Background())
  161. return s
  162. }
  163. // loadCache 加载或返回缓存的渠道数据
  164. func (s *ChannelService) loadCache(ctx context.Context) (*channelCache, error) {
  165. if cached, ok := s.cache.Load().(*channelCache); ok && cached != nil {
  166. if time.Since(cached.loadedAt) < channelCacheTTL {
  167. return cached, nil
  168. }
  169. }
  170. result, err, _ := s.cacheSF.Do("channel_cache", func() (any, error) {
  171. // 双重检查
  172. if cached, ok := s.cache.Load().(*channelCache); ok && cached != nil {
  173. if time.Since(cached.loadedAt) < channelCacheTTL {
  174. return cached, nil
  175. }
  176. }
  177. return s.buildCache(ctx)
  178. })
  179. if err != nil {
  180. return nil, err
  181. }
  182. cache, ok := result.(*channelCache)
  183. if !ok {
  184. return nil, fmt.Errorf("unexpected cache type")
  185. }
  186. return cache, nil
  187. }
  188. // newEmptyChannelCache 创建空的渠道缓存(所有 map 已初始化)
  189. func newEmptyChannelCache() *channelCache {
  190. return &channelCache{
  191. pricingByGroupModel: make(map[channelModelKey]*ChannelModelPricing),
  192. wildcardByGroupPlatform: make(map[channelGroupPlatformKey][]*wildcardPricingEntry),
  193. mappingByGroupModel: make(map[channelModelKey]string),
  194. wildcardMappingByGP: make(map[channelGroupPlatformKey][]*wildcardMappingEntry),
  195. channelByGroupID: make(map[int64]*Channel),
  196. groupPlatform: make(map[int64]string),
  197. byID: make(map[int64]*Channel),
  198. }
  199. }
  200. // expandPricingToCache 将渠道的模型定价展开到缓存(按分组+平台维度)。
  201. // 各平台严格独立:antigravity 分组只匹配 antigravity 定价,不会匹配 anthropic/gemini 的定价。
  202. // 查找时通过 lookupPricingAcrossPlatforms() 在本平台内查找。
  203. func expandPricingToCache(cache *channelCache, ch *Channel, gid int64, platform string) {
  204. for j := range ch.ModelPricing {
  205. pricing := &ch.ModelPricing[j]
  206. if !isPlatformPricingMatch(platform, pricing.Platform) {
  207. continue // 跳过非本平台的定价
  208. }
  209. // 使用定价条目的原始平台作为缓存 key,防止跨平台同名模型冲突
  210. pricingPlatform := pricing.Platform
  211. gpKey := channelGroupPlatformKey{groupID: gid, platform: pricingPlatform}
  212. for _, model := range pricing.Models {
  213. if strings.HasSuffix(model, "*") {
  214. prefix := normalizeChannelPricingModelName(strings.TrimSuffix(model, "*"))
  215. cache.wildcardByGroupPlatform[gpKey] = append(cache.wildcardByGroupPlatform[gpKey], &wildcardPricingEntry{
  216. prefix: prefix,
  217. pricing: pricing,
  218. })
  219. } else {
  220. key := channelModelKey{groupID: gid, platform: pricingPlatform, model: normalizeChannelPricingModelName(model)}
  221. cache.pricingByGroupModel[key] = pricing
  222. }
  223. }
  224. }
  225. }
  226. // expandMappingToCache 将渠道的模型映射展开到缓存(按分组+平台维度)。
  227. // 各平台严格独立:antigravity 分组只匹配 antigravity 映射。
  228. func expandMappingToCache(cache *channelCache, ch *Channel, gid int64, platform string) {
  229. for _, mappingPlatform := range matchingPlatforms(platform) {
  230. platformMapping, ok := ch.ModelMapping[mappingPlatform]
  231. if !ok {
  232. continue
  233. }
  234. // 使用映射条目的原始平台作为缓存 key,防止跨平台同名映射冲突
  235. gpKey := channelGroupPlatformKey{groupID: gid, platform: mappingPlatform}
  236. for src, dst := range platformMapping {
  237. if strings.HasSuffix(src, "*") {
  238. prefix := strings.ToLower(strings.TrimSuffix(src, "*"))
  239. cache.wildcardMappingByGP[gpKey] = append(cache.wildcardMappingByGP[gpKey], &wildcardMappingEntry{
  240. prefix: prefix,
  241. target: dst,
  242. })
  243. } else {
  244. key := channelModelKey{groupID: gid, platform: mappingPlatform, model: strings.ToLower(src)}
  245. cache.mappingByGroupModel[key] = dst
  246. }
  247. }
  248. }
  249. }
  250. // storeErrorCache 存入短 TTL 空缓存,防止 DB 错误后紧密重试。
  251. // 通过回退 loadedAt 使剩余 TTL = channelErrorTTL。
  252. func (s *ChannelService) storeErrorCache() {
  253. errorCache := newEmptyChannelCache()
  254. errorCache.loadedAt = time.Now().Add(-(channelCacheTTL - channelErrorTTL))
  255. s.cache.Store(errorCache)
  256. }
  257. // buildCache 从数据库构建渠道缓存。
  258. // 使用独立 context 避免请求取消导致空值被长期缓存。
  259. func (s *ChannelService) buildCache(ctx context.Context) (*channelCache, error) {
  260. dbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), channelCacheDBTimeout)
  261. defer cancel()
  262. channels, groupPlatforms, err := s.fetchChannelData(dbCtx)
  263. if err != nil {
  264. return nil, err
  265. }
  266. cache := populateChannelCache(channels, groupPlatforms)
  267. s.cache.Store(cache)
  268. return cache, nil
  269. }
  270. // fetchChannelData 从数据库加载渠道列表和分组平台映射。
  271. func (s *ChannelService) fetchChannelData(ctx context.Context) ([]Channel, map[int64]string, error) {
  272. channels, err := s.repo.ListAll(ctx)
  273. if err != nil {
  274. slog.Warn("failed to build channel cache", "error", err)
  275. s.storeErrorCache()
  276. return nil, nil, fmt.Errorf("list all channels: %w", err)
  277. }
  278. var allGroupIDs []int64
  279. for i := range channels {
  280. allGroupIDs = append(allGroupIDs, channels[i].GroupIDs...)
  281. }
  282. groupPlatforms := make(map[int64]string)
  283. if len(allGroupIDs) > 0 {
  284. groupPlatforms, err = s.repo.GetGroupPlatforms(ctx, allGroupIDs)
  285. if err != nil {
  286. slog.Warn("failed to load group platforms for channel cache", "error", err)
  287. s.storeErrorCache()
  288. return nil, nil, fmt.Errorf("get group platforms: %w", err)
  289. }
  290. }
  291. return channels, groupPlatforms, nil
  292. }
  293. // populateChannelCache 将渠道列表和分组平台映射填充到缓存快照中。
  294. // 装填时对每个 Channel 统一归一化 BillingModelSource,让缓存命中的所有下游
  295. // (gateway routing / billing / 未来任何 cache-backed 读路径)都拿到已归一化的实体,
  296. // 避免"每个出口各自记得 normalize"反模式。
  297. func populateChannelCache(channels []Channel, groupPlatforms map[int64]string) *channelCache {
  298. cache := newEmptyChannelCache()
  299. cache.groupPlatform = groupPlatforms
  300. cache.byID = make(map[int64]*Channel, len(channels))
  301. cache.loadedAt = time.Now()
  302. for i := range channels {
  303. channels[i].normalizeBillingModelSource()
  304. ch := &channels[i]
  305. cache.byID[ch.ID] = ch
  306. for _, gid := range ch.GroupIDs {
  307. cache.channelByGroupID[gid] = ch
  308. platform := groupPlatforms[gid]
  309. expandPricingToCache(cache, ch, gid, platform)
  310. expandMappingToCache(cache, ch, gid, platform)
  311. }
  312. }
  313. return cache
  314. }
  315. // invalidateCache 使缓存失效,让下次读取时自然重建
  316. // isPlatformPricingMatch 判断定价条目的平台是否匹配分组平台。
  317. // Concrete platforms stay isolated; composite groups may carry concrete-provider
  318. // pricing rows that are selected by the request's resolved target platform.
  319. func isPlatformPricingMatch(groupPlatform, pricingPlatform string) bool {
  320. if groupPlatform == PlatformComposite {
  321. return isConcreteRequestPlatform(pricingPlatform)
  322. }
  323. return groupPlatform == pricingPlatform
  324. }
  325. // matchingPlatforms 返回分组平台对应的可匹配平台列表。
  326. // Concrete platforms return themselves; composite is a configuration-time
  327. // fallback used before a request target has been resolved.
  328. func matchingPlatforms(groupPlatform string) []string {
  329. if groupPlatform == PlatformComposite {
  330. return []string{PlatformAnthropic, PlatformGemini, PlatformOpenAI, PlatformAntigravity, PlatformGrok, PlatformKimi, PlatformZhipu, PlatformDeepseek, PlatformMiniMax}
  331. }
  332. return []string{groupPlatform}
  333. }
  334. func channelLookupPlatform(ctx context.Context, groupPlatform string) string {
  335. if ctx != nil {
  336. if forcePlatform, ok := ctx.Value(ctxkey.ForcePlatform).(string); ok && strings.TrimSpace(forcePlatform) != "" {
  337. return strings.TrimSpace(forcePlatform)
  338. }
  339. if groupPlatform == PlatformComposite {
  340. if platform, ok := ResolvedTargetPlatformFromContext(ctx); ok {
  341. return platform
  342. }
  343. }
  344. }
  345. return groupPlatform
  346. }
  347. // InvalidateCache 失效并重建渠道缓存。
  348. // 供渠道以外、但会影响渠道缓存内容的变更调用(如分组平台变更)。
  349. func (s *ChannelService) InvalidateCache() {
  350. s.invalidateCache()
  351. }
  352. func (s *ChannelService) invalidateCache() {
  353. s.clearCache()
  354. // 主动重建缓存,确保 CRUD 后立即生效
  355. if _, err := s.buildCache(context.Background()); err != nil {
  356. slog.Warn("failed to rebuild channel cache after invalidation", "error", err)
  357. }
  358. s.notifyCacheUpdate()
  359. }
  360. // clearCache clears only the in-process snapshot. Keeping this separate from
  361. // invalidateCache prevents notifications received from Redis from being
  362. // published again in a loop.
  363. func (s *ChannelService) clearCache() {
  364. s.cache.Store((*channelCache)(nil))
  365. s.cacheSF.Forget("channel_cache")
  366. }
  367. func (s *ChannelService) notifyCacheUpdate() {
  368. if s.cachePubSub == nil {
  369. return
  370. }
  371. ctx, cancel := context.WithTimeout(context.Background(), channelCacheNotifyTTL)
  372. defer cancel()
  373. if err := s.cachePubSub.NotifyUpdate(ctx); err != nil {
  374. slog.Warn("failed to publish channel cache invalidation", "error", err)
  375. }
  376. }
  377. func (s *ChannelService) subscribeCacheUpdates(ctx context.Context) {
  378. if s.cachePubSub == nil {
  379. return
  380. }
  381. s.cachePubSub.SubscribeUpdates(ctx, s.clearCache)
  382. }
  383. // matchWildcard 在通配符定价中查找匹配项(最先匹配到优先)
  384. func (c *channelCache) matchWildcard(groupID int64, platform, modelLower string) *ChannelModelPricing {
  385. gpKey := channelGroupPlatformKey{groupID: groupID, platform: platform}
  386. wildcards := c.wildcardByGroupPlatform[gpKey]
  387. for _, wc := range wildcards {
  388. if strings.HasPrefix(modelLower, wc.prefix) {
  389. return wc.pricing
  390. }
  391. }
  392. return nil
  393. }
  394. // matchWildcardMapping 在通配符映射中查找匹配项(最先匹配到优先)
  395. func (c *channelCache) matchWildcardMapping(groupID int64, platform, modelLower string) string {
  396. gpKey := channelGroupPlatformKey{groupID: groupID, platform: platform}
  397. wildcards := c.wildcardMappingByGP[gpKey]
  398. for _, wc := range wildcards {
  399. if strings.HasPrefix(modelLower, wc.prefix) {
  400. return wc.target
  401. }
  402. }
  403. return ""
  404. }
  405. // lookupPricingAcrossPlatforms 在分组平台内查找模型定价。
  406. // 各平台严格独立,只在本平台内查找(先精确匹配,再通配符)。
  407. func lookupPricingAcrossPlatforms(cache *channelCache, groupID int64, groupPlatform, modelLower string) *ChannelModelPricing {
  408. modelLower = normalizeChannelPricingModelName(modelLower)
  409. for _, p := range matchingPlatforms(groupPlatform) {
  410. key := channelModelKey{groupID: groupID, platform: p, model: modelLower}
  411. if pricing, ok := cache.pricingByGroupModel[key]; ok {
  412. return pricing
  413. }
  414. }
  415. // 精确查找全部失败,依次尝试通配符匹配
  416. for _, p := range matchingPlatforms(groupPlatform) {
  417. if pricing := cache.matchWildcard(groupID, p, modelLower); pricing != nil {
  418. return pricing
  419. }
  420. }
  421. return nil
  422. }
  423. // lookupMappingAcrossPlatforms 在分组平台内查找模型映射。
  424. // 逻辑与 lookupPricingAcrossPlatforms 相同:先精确查找,再通配符。
  425. func lookupMappingAcrossPlatforms(cache *channelCache, groupID int64, groupPlatform, modelLower string) string {
  426. for _, p := range matchingPlatforms(groupPlatform) {
  427. key := channelModelKey{groupID: groupID, platform: p, model: modelLower}
  428. if mapped, ok := cache.mappingByGroupModel[key]; ok {
  429. return mapped
  430. }
  431. }
  432. for _, p := range matchingPlatforms(groupPlatform) {
  433. if mapped := cache.matchWildcardMapping(groupID, p, modelLower); mapped != "" {
  434. return mapped
  435. }
  436. }
  437. return ""
  438. }
  439. // GetChannelForGroup 获取分组关联的渠道(热路径 O(1))
  440. func (s *ChannelService) GetChannelForGroup(ctx context.Context, groupID int64) (*Channel, error) {
  441. cache, err := s.loadCache(ctx)
  442. if err != nil {
  443. return nil, err
  444. }
  445. ch, ok := cache.channelByGroupID[groupID]
  446. if !ok || !ch.IsActive() {
  447. return nil, nil
  448. }
  449. return ch.Clone(), nil
  450. }
  451. // GetGroupPlatform 获取分组的平台标识(从缓存)
  452. func (s *ChannelService) GetGroupPlatform(ctx context.Context, groupID int64) string {
  453. cache, err := s.loadCache(ctx)
  454. if err != nil {
  455. return ""
  456. }
  457. return cache.groupPlatform[groupID]
  458. }
  459. // channelLookup 热路径公共查找结果
  460. type channelLookup struct {
  461. cache *channelCache
  462. channel *Channel
  463. platform string
  464. }
  465. // lookupGroupChannel 加载缓存并查找分组对应的渠道信息(公共热路径前置逻辑)。
  466. // 返回 nil 且 err==nil 表示分组无活跃渠道;err!=nil 表示缓存加载失败。
  467. func (s *ChannelService) lookupGroupChannel(ctx context.Context, groupID int64) (*channelLookup, error) {
  468. cache, err := s.loadCache(ctx)
  469. if err != nil {
  470. return nil, err
  471. }
  472. ch, ok := cache.channelByGroupID[groupID]
  473. if !ok || !ch.IsActive() {
  474. return nil, nil
  475. }
  476. return &channelLookup{
  477. cache: cache,
  478. channel: ch,
  479. platform: channelLookupPlatform(ctx, cache.groupPlatform[groupID]),
  480. }, nil
  481. }
  482. // GetChannelModelPricing 获取指定分组+模型的渠道定价(热路径 O(1))。
  483. // 各平台严格独立,只在本平台内查找定价。
  484. func (s *ChannelService) GetChannelModelPricing(ctx context.Context, groupID int64, model string) *ChannelModelPricing {
  485. lk, err := s.lookupGroupChannel(ctx, groupID)
  486. if err != nil {
  487. slog.Warn("failed to load channel cache", "group_id", groupID, "error", err)
  488. return nil
  489. }
  490. if lk == nil {
  491. return nil
  492. }
  493. modelLower := strings.ToLower(model)
  494. pricing := lookupPricingAcrossPlatforms(lk.cache, groupID, lk.platform, modelLower)
  495. if pricing == nil {
  496. return nil
  497. }
  498. cp := pricing.Clone()
  499. return &cp
  500. }
  501. // ResolveChannelMapping 解析渠道级模型映射(热路径 O(1))
  502. // 返回映射结果,包含映射后的模型名、渠道 ID、计费模型来源。
  503. func (s *ChannelService) ResolveChannelMapping(ctx context.Context, groupID int64, model string) ChannelMappingResult {
  504. lk, err := s.lookupGroupChannel(ctx, groupID)
  505. if err != nil {
  506. slog.Warn("failed to load channel cache for mapping", "group_id", groupID, "error", err)
  507. }
  508. if lk == nil {
  509. return ChannelMappingResult{MappedModel: model}
  510. }
  511. return resolveMapping(lk, groupID, model)
  512. }
  513. // IsModelRestricted 检查模型是否被渠道限制。
  514. // 返回 true 表示模型被限制(不在允许列表中)。
  515. // 如果渠道未启用模型限制或分组无渠道关联,返回 false。
  516. func (s *ChannelService) IsModelRestricted(ctx context.Context, groupID int64, model string) bool {
  517. lk, err := s.lookupGroupChannel(ctx, groupID)
  518. if err != nil {
  519. slog.Warn("failed to load channel cache for model restriction check", "group_id", groupID, "error", err)
  520. }
  521. if lk == nil {
  522. return false
  523. }
  524. return checkRestricted(lk, groupID, model)
  525. }
  526. // ResolveChannelMappingAndRestrict 解析渠道映射。
  527. // 返回映射结果。模型限制检查已移至调度阶段(GatewayService.checkChannelPricingRestriction),
  528. // restricted 始终返回 false,保留签名兼容性。
  529. func (s *ChannelService) ResolveChannelMappingAndRestrict(ctx context.Context, groupID *int64, model string) (ChannelMappingResult, bool) {
  530. if groupID == nil {
  531. return ChannelMappingResult{MappedModel: model}, false
  532. }
  533. lk, _ := s.lookupGroupChannel(ctx, *groupID)
  534. if lk == nil {
  535. return ChannelMappingResult{MappedModel: model}, false
  536. }
  537. return resolveMapping(lk, *groupID, model), false
  538. }
  539. // resolveMapping 基于已查找的渠道信息解析模型映射。
  540. // antigravity 分组依次尝试所有匹配平台,确保跨平台同名映射各自独立。
  541. func resolveMapping(lk *channelLookup, groupID int64, model string) ChannelMappingResult {
  542. // lk.channel 来自已装填的缓存,BillingModelSource 已在 populateChannelCache 阶段归一化,
  543. // 这里无需重复兜底。
  544. result := ChannelMappingResult{
  545. MappedModel: model,
  546. ChannelID: lk.channel.ID,
  547. BillingModelSource: lk.channel.BillingModelSource,
  548. }
  549. modelLower := strings.ToLower(model)
  550. if mapped := lookupMappingAcrossPlatforms(lk.cache, groupID, lk.platform, modelLower); mapped != "" {
  551. result.MappedModel = mapped
  552. result.Mapped = true
  553. }
  554. return result
  555. }
  556. // checkRestricted 基于已查找的渠道信息检查模型是否被限制。
  557. // 只在本平台的定价列表中查找。
  558. func checkRestricted(lk *channelLookup, groupID int64, model string) bool {
  559. if !lk.channel.RestrictModels {
  560. return false
  561. }
  562. modelLower := strings.ToLower(model)
  563. // 使用与查找定价相同的跨平台逻辑
  564. if lookupPricingAcrossPlatforms(lk.cache, groupID, lk.platform, modelLower) != nil {
  565. return false
  566. }
  567. return true
  568. }
  569. // ReplaceModelInBody 替换请求体 JSON 中的 model 字段。
  570. func ReplaceModelInBody(body []byte, newModel string) []byte {
  571. if len(body) == 0 {
  572. return body
  573. }
  574. if current := gjson.GetBytes(body, "model"); current.Exists() && current.String() == newModel {
  575. return body
  576. }
  577. newBody, err := sjson.SetBytes(body, "model", newModel)
  578. if err != nil {
  579. return body
  580. }
  581. return newBody
  582. }
  583. // RemovePreviousResponseIDFromBody 删除请求体中的 previous_response_id,用于会话失配时改用完整 input 重建上下文。
  584. func RemovePreviousResponseIDFromBody(body []byte) []byte {
  585. if len(body) == 0 {
  586. return body
  587. }
  588. if !gjson.GetBytes(body, "previous_response_id").Exists() {
  589. return body
  590. }
  591. newBody, err := sjson.DeleteBytes(body, "previous_response_id")
  592. if err != nil {
  593. return body
  594. }
  595. return newBody
  596. }
  597. // validateChannelConfig 校验渠道的定价和映射配置(冲突检测 + 区间校验 + 计费模式校验)。
  598. // Create 和 Update 共用此函数,避免重复。
  599. func validateChannelConfig(pricing []ChannelModelPricing, mapping map[string]map[string]string) error {
  600. if err := validatePricingEntries(pricing); err != nil {
  601. return err
  602. }
  603. return validateNoConflictingMappings(mapping)
  604. }
  605. // validatePricingEntries 校验定价条目(冲突检测 + 区间校验 + 计费模式校验),
  606. // 同时用于主渠道定价和 account_stats_pricing_rules 的内部定价。
  607. func validatePricingEntries(pricing []ChannelModelPricing) error {
  608. if err := validateNoConflictingModels(pricing); err != nil {
  609. return err
  610. }
  611. if err := validatePricingIntervals(pricing); err != nil {
  612. return err
  613. }
  614. if err := validatePricingBillingMode(pricing); err != nil {
  615. return err
  616. }
  617. return validatePricingTimePricing(pricing)
  618. }
  619. func validatePricingTimePricing(pricing []ChannelModelPricing) error {
  620. for i := range pricing {
  621. config := pricing[i].TimePricing
  622. if config == nil {
  623. continue
  624. }
  625. if len(config.Periods) == 0 {
  626. pricing[i].TimePricing = nil
  627. continue
  628. }
  629. mode := pricing[i].BillingMode
  630. if mode != "" && mode != BillingModeToken {
  631. return infraerrors.BadRequest("TIME_PRICING_UNSUPPORTED_MODE", "time pricing only supports token billing mode")
  632. }
  633. if err := validateChannelTimePricing(config); err != nil {
  634. return infraerrors.BadRequest("INVALID_TIME_PRICING", fmt.Sprintf(
  635. "invalid time pricing for platform '%s' models %v: %v", pricing[i].Platform, pricing[i].Models, err))
  636. }
  637. }
  638. return nil
  639. }
  640. func validateAccountStatsPricingRules(rules []AccountStatsPricingRule) error {
  641. for i := range rules {
  642. for _, pricing := range rules[i].Pricing {
  643. if pricing.TimePricing != nil && len(pricing.TimePricing.Periods) > 0 {
  644. return fmt.Errorf("account stats pricing rule #%d: %w", i+1,
  645. infraerrors.BadRequest("ACCOUNT_STATS_TIME_PRICING_UNSUPPORTED", "account stats pricing does not support time pricing"))
  646. }
  647. }
  648. if err := validatePricingEntries(rules[i].Pricing); err != nil {
  649. return fmt.Errorf("account stats pricing rule #%d: %w", i+1, err)
  650. }
  651. }
  652. return nil
  653. }
  654. // validatePricingBillingMode 校验计费模式配置:按次/图片模式必须配价格或区间,所有价格字段不能为负,区间至少有一个价格字段。
  655. func validatePricingBillingMode(pricing []ChannelModelPricing) error {
  656. for _, p := range pricing {
  657. if err := checkBillingModeRequirements(p); err != nil {
  658. return err
  659. }
  660. if err := checkPricesNotNegative(p); err != nil {
  661. return err
  662. }
  663. if err := checkIntervalsHavePrices(p); err != nil {
  664. return err
  665. }
  666. }
  667. return nil
  668. }
  669. func checkBillingModeRequirements(p ChannelModelPricing) error {
  670. if p.BillingMode == BillingModePerRequest || p.BillingMode == BillingModeImage || p.BillingMode == BillingModeVideo {
  671. if p.PerRequestPrice == nil && len(p.Intervals) == 0 {
  672. return infraerrors.BadRequest(
  673. "BILLING_MODE_MISSING_PRICE",
  674. "per-request price or intervals required for per_request/image billing mode",
  675. )
  676. }
  677. }
  678. return nil
  679. }
  680. func checkPricesNotNegative(p ChannelModelPricing) error {
  681. checks := []struct {
  682. field string
  683. val *float64
  684. }{
  685. {"input_price", p.InputPrice},
  686. {"output_price", p.OutputPrice},
  687. {"cache_write_price", p.CacheWritePrice},
  688. {"cache_write_1h_price", p.CacheWrite1hPrice},
  689. {"cache_read_price", p.CacheReadPrice},
  690. {"image_input_price", p.ImageInputPrice},
  691. {"image_output_price", p.ImageOutputPrice},
  692. {"per_request_price", p.PerRequestPrice},
  693. }
  694. for _, c := range checks {
  695. if c.val != nil && *c.val < 0 {
  696. return infraerrors.BadRequest("NEGATIVE_PRICE", fmt.Sprintf("%s must be >= 0", c.field))
  697. }
  698. }
  699. for _, c := range []struct {
  700. field string
  701. val *float64
  702. }{
  703. {"fast_multiplier", p.FastMultiplier},
  704. {"flex_multiplier", p.FlexMultiplier},
  705. } {
  706. if c.val != nil && *c.val <= 0 {
  707. return infraerrors.BadRequest("INVALID_MULTIPLIER", fmt.Sprintf("%s must be > 0", c.field))
  708. }
  709. }
  710. return nil
  711. }
  712. func checkIntervalsHavePrices(p ChannelModelPricing) error {
  713. for _, iv := range p.Intervals {
  714. if iv.InputPrice == nil && iv.OutputPrice == nil &&
  715. iv.CacheWritePrice == nil && iv.CacheWrite1hPrice == nil && iv.CacheReadPrice == nil &&
  716. iv.PerRequestPrice == nil && iv.InputMultiplier == nil &&
  717. iv.OutputMultiplier == nil && iv.CacheWriteMultiplier == nil &&
  718. iv.CacheReadMultiplier == nil {
  719. return infraerrors.BadRequest(
  720. "INTERVAL_MISSING_PRICE",
  721. fmt.Sprintf("interval [%d, %s] has no price fields set for model %v",
  722. iv.MinTokens, formatMaxTokens(iv.MaxTokens), p.Models),
  723. )
  724. }
  725. }
  726. return nil
  727. }
  728. func formatMaxTokens(max *int) string {
  729. if max == nil {
  730. return "∞"
  731. }
  732. return fmt.Sprintf("%d", *max)
  733. }
  734. // --- CRUD ---
  735. // Create 创建渠道
  736. func (s *ChannelService) Create(ctx context.Context, input *CreateChannelInput) (*Channel, error) {
  737. exists, err := s.repo.ExistsByName(ctx, input.Name)
  738. if err != nil {
  739. return nil, fmt.Errorf("check channel exists: %w", err)
  740. }
  741. if exists {
  742. return nil, ErrChannelExists
  743. }
  744. if err := s.checkGroupConflicts(ctx, 0, input.GroupIDs); err != nil {
  745. return nil, err
  746. }
  747. channel := &Channel{
  748. Name: input.Name,
  749. Description: input.Description,
  750. Status: StatusActive,
  751. BillingModelSource: input.BillingModelSource,
  752. RestrictModels: input.RestrictModels,
  753. GroupIDs: input.GroupIDs,
  754. ModelPricing: input.ModelPricing,
  755. ModelMapping: input.ModelMapping,
  756. Features: input.Features,
  757. FeaturesConfig: input.FeaturesConfig,
  758. ApplyPricingToAccountStats: input.ApplyPricingToAccountStats,
  759. AccountStatsPricingRules: input.AccountStatsPricingRules,
  760. }
  761. channel.normalizeBillingModelSource()
  762. if err := validateChannelConfig(channel.ModelPricing, channel.ModelMapping); err != nil {
  763. return nil, err
  764. }
  765. if err := validateAccountStatsPricingRules(channel.AccountStatsPricingRules); err != nil {
  766. return nil, err
  767. }
  768. if err := s.repo.Create(ctx, channel); err != nil {
  769. return nil, fmt.Errorf("create channel: %w", err)
  770. }
  771. s.invalidateCache()
  772. created, err := s.repo.GetByID(ctx, channel.ID)
  773. if err != nil {
  774. return nil, err
  775. }
  776. created.normalizeBillingModelSource()
  777. return created, nil
  778. }
  779. // GetByID 获取渠道详情。返回前统一把空 BillingModelSource 回填为 ChannelMapped,
  780. // 让所有 handler 无需重复处理历史空值。
  781. func (s *ChannelService) GetByID(ctx context.Context, id int64) (*Channel, error) {
  782. ch, err := s.repo.GetByID(ctx, id)
  783. if err != nil {
  784. return nil, err
  785. }
  786. ch.normalizeBillingModelSource()
  787. return ch, nil
  788. }
  789. // Update 更新渠道
  790. func (s *ChannelService) Update(ctx context.Context, id int64, input *UpdateChannelInput) (*Channel, error) {
  791. channel, err := s.repo.GetByID(ctx, id)
  792. if err != nil {
  793. return nil, fmt.Errorf("get channel: %w", err)
  794. }
  795. if err := s.applyUpdateInput(ctx, channel, input); err != nil {
  796. return nil, err
  797. }
  798. if err := validateChannelConfig(channel.ModelPricing, channel.ModelMapping); err != nil {
  799. return nil, err
  800. }
  801. if err := validateAccountStatsPricingRules(channel.AccountStatsPricingRules); err != nil {
  802. return nil, err
  803. }
  804. oldGroupIDs := s.getOldGroupIDs(ctx, id)
  805. if err := s.repo.Update(ctx, channel); err != nil {
  806. return nil, fmt.Errorf("update channel: %w", err)
  807. }
  808. s.invalidateCache()
  809. s.invalidateAuthCacheForGroups(ctx, oldGroupIDs, channel.GroupIDs)
  810. updated, err := s.repo.GetByID(ctx, id)
  811. if err != nil {
  812. return nil, err
  813. }
  814. updated.normalizeBillingModelSource()
  815. return updated, nil
  816. }
  817. // applyUpdateInput 将更新请求的字段应用到渠道实体上。
  818. func (s *ChannelService) applyUpdateInput(ctx context.Context, channel *Channel, input *UpdateChannelInput) error {
  819. if input.Name != "" && input.Name != channel.Name {
  820. exists, err := s.repo.ExistsByNameExcluding(ctx, input.Name, channel.ID)
  821. if err != nil {
  822. return fmt.Errorf("check channel exists: %w", err)
  823. }
  824. if exists {
  825. return ErrChannelExists
  826. }
  827. channel.Name = input.Name
  828. }
  829. if input.Description != nil {
  830. channel.Description = *input.Description
  831. }
  832. if input.Status != "" {
  833. channel.Status = input.Status
  834. }
  835. if input.RestrictModels != nil {
  836. channel.RestrictModels = *input.RestrictModels
  837. }
  838. if input.Features != nil {
  839. channel.Features = *input.Features
  840. }
  841. if input.GroupIDs != nil {
  842. if err := s.checkGroupConflicts(ctx, channel.ID, *input.GroupIDs); err != nil {
  843. return err
  844. }
  845. channel.GroupIDs = *input.GroupIDs
  846. }
  847. if input.ModelPricing != nil {
  848. channel.ModelPricing = *input.ModelPricing
  849. }
  850. if input.ModelMapping != nil {
  851. channel.ModelMapping = input.ModelMapping
  852. }
  853. if input.BillingModelSource != "" {
  854. channel.BillingModelSource = input.BillingModelSource
  855. }
  856. if input.FeaturesConfig != nil {
  857. channel.FeaturesConfig = input.FeaturesConfig
  858. }
  859. if input.ApplyPricingToAccountStats != nil {
  860. channel.ApplyPricingToAccountStats = *input.ApplyPricingToAccountStats
  861. }
  862. if input.AccountStatsPricingRules != nil {
  863. channel.AccountStatsPricingRules = *input.AccountStatsPricingRules
  864. }
  865. return nil
  866. }
  867. // checkGroupConflicts 检查待关联的分组是否已属于其他渠道。
  868. // channelID 为当前渠道 ID(Create 时传 0)。
  869. func (s *ChannelService) checkGroupConflicts(ctx context.Context, channelID int64, groupIDs []int64) error {
  870. if len(groupIDs) == 0 {
  871. return nil
  872. }
  873. conflicting, err := s.repo.GetGroupsInOtherChannels(ctx, channelID, groupIDs)
  874. if err != nil {
  875. return fmt.Errorf("check group conflicts: %w", err)
  876. }
  877. if len(conflicting) > 0 {
  878. return ErrGroupAlreadyInChannel
  879. }
  880. return nil
  881. }
  882. // getOldGroupIDs 获取渠道更新前的关联分组 ID(用于失效 auth 缓存)。
  883. func (s *ChannelService) getOldGroupIDs(ctx context.Context, channelID int64) []int64 {
  884. if s.authCacheInvalidator == nil {
  885. return nil
  886. }
  887. oldGroupIDs, err := s.repo.GetGroupIDs(ctx, channelID)
  888. if err != nil {
  889. slog.Warn("failed to get old group IDs for cache invalidation", "channel_id", channelID, "error", err)
  890. }
  891. return oldGroupIDs
  892. }
  893. // invalidateAuthCacheForGroups 对新旧分组去重后逐个失效 auth 缓存。
  894. func (s *ChannelService) invalidateAuthCacheForGroups(ctx context.Context, groupIDSets ...[]int64) {
  895. if s.authCacheInvalidator == nil {
  896. return
  897. }
  898. seen := make(map[int64]struct{})
  899. for _, ids := range groupIDSets {
  900. for _, gid := range ids {
  901. if _, ok := seen[gid]; ok {
  902. continue
  903. }
  904. seen[gid] = struct{}{}
  905. s.authCacheInvalidator.InvalidateAuthCacheByGroupID(ctx, gid)
  906. }
  907. }
  908. }
  909. // Delete 删除渠道
  910. func (s *ChannelService) Delete(ctx context.Context, id int64) error {
  911. groupIDs, err := s.repo.GetGroupIDs(ctx, id)
  912. if err != nil {
  913. slog.Warn("failed to get group IDs before delete", "channel_id", id, "error", err)
  914. }
  915. if err := s.repo.Delete(ctx, id); err != nil {
  916. return fmt.Errorf("delete channel: %w", err)
  917. }
  918. s.invalidateCache()
  919. s.invalidateAuthCacheForGroups(ctx, groupIDs)
  920. return nil
  921. }
  922. // List 获取渠道列表
  923. func (s *ChannelService) List(ctx context.Context, params pagination.PaginationParams, status, search string) ([]Channel, *pagination.PaginationResult, error) {
  924. channels, res, err := s.repo.List(ctx, params, status, search)
  925. if err != nil {
  926. return nil, nil, err
  927. }
  928. for i := range channels {
  929. channels[i].normalizeBillingModelSource()
  930. }
  931. return channels, res, nil
  932. }
  933. // modelEntry 表示一个模型模式条目(用于冲突检测)
  934. type modelEntry struct {
  935. pattern string // 原始模式(如 "claude-*" 或 "claude-opus-4")
  936. prefix string // lowercase 前缀(通配符去掉 *,精确名保持原样)
  937. wildcard bool
  938. }
  939. // conflictsBetween 检查两个模型模式是否冲突
  940. func conflictsBetween(a, b modelEntry) bool {
  941. switch {
  942. case !a.wildcard && !b.wildcard:
  943. return a.prefix == b.prefix
  944. case a.wildcard && !b.wildcard:
  945. return strings.HasPrefix(b.prefix, a.prefix)
  946. case !a.wildcard && b.wildcard:
  947. return strings.HasPrefix(a.prefix, b.prefix)
  948. default:
  949. return strings.HasPrefix(a.prefix, b.prefix) ||
  950. strings.HasPrefix(b.prefix, a.prefix)
  951. }
  952. }
  953. // toModelEntry 将模型名转换为 modelEntry(用于模型映射的冲突检测)。
  954. // 归一化必须与 expandMappingToCache 写缓存键的方式一致:映射缓存只做 strings.ToLower。
  955. func toModelEntry(pattern string) modelEntry {
  956. prefix, isWild := splitWildcardSuffix(strings.ToLower(pattern))
  957. return modelEntry{pattern: pattern, prefix: prefix, wildcard: isWild}
  958. }
  959. // toPricingModelEntry 将模型名转换为 modelEntry(用于模型定价的冲突检测)。
  960. //
  961. // 与 toModelEntry 的区别:定价缓存的键走 normalizeChannelPricingModelName
  962. // (额外做 TrimSpace,并把 claude-* 的 "." 换成 "-"),冲突检测必须用同一套归一化,
  963. // 否则两个校验时看着不同、写进缓存后键相同的定价会互相静默覆盖。
  964. func toPricingModelEntry(pattern string) modelEntry {
  965. // 先剥通配符再归一化,与 expandPricingToCache 的处理顺序保持一致
  966. prefix, isWild := splitWildcardSuffix(pattern)
  967. return modelEntry{
  968. pattern: pattern,
  969. prefix: normalizeChannelPricingModelName(prefix),
  970. wildcard: isWild,
  971. }
  972. }
  973. // validateNoConflictingModels 检查定价列表中是否有冲突模型模式(同一平台下)。
  974. // 冲突包括:精确重复、通配符之间的前缀包含、通配符与精确名的前缀匹配。
  975. func validateNoConflictingModels(pricingList []ChannelModelPricing) error {
  976. byPlatform := make(map[string][]modelEntry)
  977. for _, p := range pricingList {
  978. for _, model := range p.Models {
  979. byPlatform[p.Platform] = append(byPlatform[p.Platform], toPricingModelEntry(model))
  980. }
  981. }
  982. for platform, entries := range byPlatform {
  983. if err := detectConflicts(entries, platform, "MODEL_PATTERN_CONFLICT", "model patterns"); err != nil {
  984. return err
  985. }
  986. }
  987. return nil
  988. }
  989. // validateNoConflictingMappings 检查模型映射中是否有冲突的源模式
  990. func validateNoConflictingMappings(mapping map[string]map[string]string) error {
  991. for platform, platformMapping := range mapping {
  992. entries := make([]modelEntry, 0, len(platformMapping))
  993. for src := range platformMapping {
  994. entries = append(entries, toModelEntry(src))
  995. }
  996. if err := detectConflicts(entries, platform, "MAPPING_PATTERN_CONFLICT", "mapping source patterns"); err != nil {
  997. return err
  998. }
  999. }
  1000. return nil
  1001. }
  1002. func validatePricingIntervals(pricingList []ChannelModelPricing) error {
  1003. for _, pricing := range pricingList {
  1004. if err := ValidateIntervals(pricing.Intervals, pricing.BillingMode); err != nil {
  1005. return infraerrors.BadRequest(
  1006. "INVALID_PRICING_INTERVALS",
  1007. fmt.Sprintf("invalid pricing intervals for platform '%s' models %v: %v",
  1008. pricing.Platform, pricing.Models, err),
  1009. )
  1010. }
  1011. }
  1012. return nil
  1013. }
  1014. // detectConflicts 在一组 modelEntry 中检测冲突,返回带有 errCode 和 label 的错误
  1015. func detectConflicts(entries []modelEntry, platform, errCode, label string) error {
  1016. for i := 0; i < len(entries); i++ {
  1017. for j := i + 1; j < len(entries); j++ {
  1018. if conflictsBetween(entries[i], entries[j]) {
  1019. return infraerrors.BadRequest(errCode,
  1020. fmt.Sprintf("%s '%s' and '%s' conflict in platform '%s': overlapping match range "+
  1021. "(model names are matched case-insensitively, so an existing entry already covers all case variants)",
  1022. label, entries[i].pattern, entries[j].pattern, platform))
  1023. }
  1024. }
  1025. }
  1026. return nil
  1027. }
  1028. // --- Input types ---
  1029. // CreateChannelInput 创建渠道输入
  1030. type CreateChannelInput struct {
  1031. Name string
  1032. Description string
  1033. GroupIDs []int64
  1034. ModelPricing []ChannelModelPricing
  1035. ModelMapping map[string]map[string]string // platform → {src→dst}
  1036. BillingModelSource string
  1037. RestrictModels bool
  1038. Features string
  1039. FeaturesConfig map[string]any
  1040. ApplyPricingToAccountStats bool
  1041. AccountStatsPricingRules []AccountStatsPricingRule
  1042. }
  1043. // UpdateChannelInput 更新渠道输入
  1044. type UpdateChannelInput struct {
  1045. Name string
  1046. Description *string
  1047. Status string
  1048. GroupIDs *[]int64
  1049. ModelPricing *[]ChannelModelPricing
  1050. ModelMapping map[string]map[string]string // platform → {src→dst}
  1051. BillingModelSource string
  1052. RestrictModels *bool
  1053. Features *string
  1054. FeaturesConfig map[string]any
  1055. ApplyPricingToAccountStats *bool
  1056. AccountStatsPricingRules *[]AccountStatsPricingRule
  1057. }