package channel import ( "errors" "math/rand" "sort" "sync" "go.uber.org/zap" "gorm.io/gorm" "openteam/server/internal/config" "openteam/server/internal/pkg/crypto" "openteam/server/internal/store" ) var ErrNoChannel = errors.New("no available channel for model") // Service manages channels, model bindings and load-balanced selection. type Service struct { db *gorm.DB cfg *config.Config log *zap.Logger mu sync.RWMutex sem map[int64]chan struct{} // per-channel concurrency limiter master string } func NewService(db *gorm.DB, cfg *config.Config, log *zap.Logger) *Service { return &Service{ db: db, cfg: cfg, log: log, sem: map[int64]chan struct{}{}, master: cfg.MasterKey, } } // DecryptKey decrypts a channel's stored upstream API key. func (s *Service) DecryptKey(enc string) (string, error) { return crypto.Decrypt(enc, s.master) } // Acquire takes a concurrency slot for a channel (blocks if saturated). func (s *Service) Acquire(channelID int64) (func(), error) { s.mu.Lock() lim, ok := s.sem[channelID] if !ok { var rec store.Channel if err := s.db.First(&rec, channelID).Error; err != nil { s.mu.Unlock() return nil, err } limit := rec.MaxConcurrency if limit <= 0 { limit = 100 } lim = make(chan struct{}, limit) s.sem[channelID] = lim } s.mu.Unlock() lim <- struct{}{} return func() { <-lim }, nil } // ResolveModel returns the model registry record. func (s *Service) ResolveModel(name string) (*store.Model, error) { var m store.Model if err := s.db.Where("name = ? AND enabled = ?", name, true).First(&m).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, errors.New("model not found: " + name) } return nil, err } return &m, nil } // SelectChannel picks a healthy channel bound to the given model. // Channels with higher priority (lower number) and higher weight win; // cooldown channels are skipped, degraded channels are deprioritized. // Channels in exclude (already tried in this request) are skipped. func (s *Service) SelectChannel(modelID int64, exclude map[int64]bool) (*store.Channel, *store.ChannelModelBinding, error) { var bindings []store.ChannelModelBinding if err := s.db.Preload("Channel").Where("model_id = ?", modelID).Find(&bindings).Error; err != nil { return nil, nil, err } var candidates []struct { ch *store.Channel b *store.ChannelModelBinding score float64 } for i := range bindings { ch := &bindings[i].Channel if !ch.Enabled { continue } if ch.HealthStatus == "cooldown" { continue } if exclude[ch.ID] { continue } score := float64(bindings[i].Weight) if ch.HealthStatus == "degraded" { score *= 0.1 } candidates = append(candidates, struct { ch *store.Channel b *store.ChannelModelBinding score float64 }{ch: ch, b: &bindings[i], score: score}) } if len(candidates) == 0 { return nil, nil, ErrNoChannel } // Sort by priority asc, then weight desc. sort.SliceStable(candidates, func(i, j int) bool { if candidates[i].ch.Priority != candidates[j].ch.Priority { return candidates[i].ch.Priority < candidates[j].ch.Priority } return candidates[i].score > candidates[j].score }) // Weighted random pick among the top priority group. total := 0.0 for _, c := range candidates { total += c.score } if total <= 0 { return candidates[0].ch, candidates[0].b, nil } pick := rand.Float64() * total for _, c := range candidates { pick -= c.score if pick <= 0 { return c.ch, c.b, nil } } return candidates[0].ch, candidates[0].b, nil } // ListModelsForChannel returns the model names a channel serves. func (s *Service) ListModelsForChannel(channelID int64) ([]string, error) { var names []string err := s.db.Table("channel_model_bindings"). Joins("JOIN models ON models.id = channel_model_bindings.model_id"). Where("channel_model_bindings.channel_id = ? AND models.enabled = ?", channelID, true). Order("models.sort ASC, models.id ASC"). Pluck("models.name", &names).Error return names, err } // ListEnabledModels returns all enabled models for GET /v1/models. func (s *Service) ListEnabledModels() ([]store.Model, error) { var models []store.Model err := s.db.Where("enabled = ?", true).Order("sort ASC, id ASC").Find(&models).Error return models, err }