Sakurasan
9f4d631fc4
路由与故障转移（参考 openteam 语义） - channel.Candidates：绑定模型优先（携带 upstream_model 映射）， 未绑定模型回退到权重最低的健康备用渠道；新增 Pick 加权随机与 FilterHealthy 内存健康过滤 - gateway.Dispatch：遍历候选渠道，可重试失败（连接错误/429/5xx）自动故障转移， 4xx 透传；不再使用单一 SelectChannel - 修复 gorm default 标签把渠道 weight=0 静默改写为 1 的问题（去掉 default， 权重 0 语义 = 不参与加权选择，仅作备用承接 unbound 流量） - RecordFailure 连续 2 次进入 degraded 快速熔断，健康检查成功或冷却过期后复位 网关功能补全 - /v1/models 返回 DB 中启用的模型列表（替换 TODO 存根） - 请求级 request_id 生成与用量记录接入：流式 SSE 逐块累计 usage、 非流式从响应提取，按模型定价计算成本后经 usage.Recorder 异步落库 - 流式结束检测：chat 的 [DONE]、messages 的 message_stop、responses 的 response.completed，避免 keep-alive 上游发完不关连接导致读阻塞到超时 - ResponsesRequest.input 兼容字符串与条目数组两种客户端写法 测试 - 修复 convert_test 对新 input 形态的断言 - 网关 e2e（/tmp/test_gateway.py + mock upstream）72/72 全部通过，连续 3 次稳定
91 lines
2.8 KiB
Go
91 lines
2.8 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"opencatd-open/internal/channel"
|
|
"opencatd-open/internal/dao"
|
|
"opencatd-open/internal/store"
|
|
"opencatd-open/internal/pkg/crypto"
|
|
)
|
|
|
|
type ChannelServiceImpl struct {
|
|
channelDAO *dao.ChannelDAO
|
|
channelSvc *channel.Service
|
|
}
|
|
|
|
func NewChannelService(channelDAO *dao.ChannelDAO, channelSvc *channel.Service) *ChannelServiceImpl {
|
|
return &ChannelServiceImpl{
|
|
channelDAO: channelDAO,
|
|
channelSvc: channelSvc,
|
|
}
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) Create(ctx context.Context, ch *store.Channel) error {
|
|
return s.channelDAO.Create(ch)
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) GetByID(ctx context.Context, id uint64) (*store.Channel, error) {
|
|
return s.channelDAO.GetByID(id)
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) List(ctx context.Context, limit, offset int) ([]*store.Channel, int64, error) {
|
|
return s.channelDAO.List(limit, offset)
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) ListEnabled(ctx context.Context) ([]*store.Channel, error) {
|
|
return s.channelDAO.ListEnabled()
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) Update(ctx context.Context, ch *store.Channel) error {
|
|
return s.channelDAO.Update(ch)
|
|
}
|
|
|
|
func (s *ChannelServiceImpl) Delete(ctx context.Context, id uint64) error {
|
|
return s.channelDAO.Delete(id)
|
|
}
|
|
|
|
// GetAPIKey decrypts the channel's API key
|
|
func (s *ChannelServiceImpl) GetAPIKey(ctx context.Context, channelID uint64) (string, error) {
|
|
ch, err := s.channelDAO.GetByID(channelID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return crypto.Decrypt(ch.APIKeyEnc)
|
|
}
|
|
|
|
// SelectForModel selects the best channel for a model
|
|
func (s *ChannelServiceImpl) SelectForModel(ctx context.Context, modelName string) (*store.Channel, error) {
|
|
cands := s.channelSvc.Candidates(modelName)
|
|
picked := s.channelSvc.Pick(cands)
|
|
if picked == nil {
|
|
return nil, fmt.Errorf("no enabled channels for model: %s", modelName)
|
|
}
|
|
return picked.Channel, nil
|
|
}
|
|
|
|
// BindModels binds models to a channel
|
|
func (s *ChannelServiceImpl) BindModels(ctx context.Context, channelID uint64, bindings []store.ChannelModelBinding) error {
|
|
return s.channelDAO.BindModels(channelID, bindings)
|
|
}
|
|
|
|
// GetChannelModels returns models bound to a channel
|
|
func (s *ChannelServiceImpl) GetChannelModels(ctx context.Context, channelID uint64) ([]store.ChannelModelBinding, error) {
|
|
return s.channelDAO.GetChannelModels(channelID)
|
|
}
|
|
|
|
// GetModelChannels returns channels for a model
|
|
func (s *ChannelServiceImpl) GetModelChannels(ctx context.Context, modelName string) ([]*store.Channel, error) {
|
|
return s.channelDAO.GetEnabledChannelsByModel(modelName)
|
|
}
|
|
|
|
// RecordSuccess records a successful request
|
|
func (s *ChannelServiceImpl) RecordSuccess(channelID uint64) {
|
|
s.channelSvc.RecordSuccess(channelID)
|
|
}
|
|
|
|
// RecordFailure records a failed request
|
|
func (s *ChannelServiceImpl) RecordFailure(channelID uint64) {
|
|
s.channelSvc.RecordFailure(channelID)
|
|
}
|