Files
ModelRouter/internal/core/core.go

351 lines
9.0 KiB
Go

// Package core wires together the Lua VM, provider registry, scheduler and
// runtime store and exposes management operations (hot reload, adapters,
// sources) for the web UI and gateway.
package core
import (
"crypto/rand"
"encoding/hex"
"fmt"
"os"
"path/filepath"
"sort"
"time"
"llmsproxy/internal/config"
"llmsproxy/internal/lua"
"llmsproxy/internal/provider"
"llmsproxy/internal/scheduler"
)
// Core owns the running configuration and adapters.
type Core struct {
cfg *config.Config
vm *lua.VM
store *config.Store
scheduler *scheduler.Scheduler
registry *provider.Registry
}
// New builds the core from a config file plus runtime overlay.
func New(cfgPath string) (*Core, error) {
cfg, err := config.Load(cfgPath)
if err != nil {
return nil, err
}
return NewFromConfig(cfg)
}
// NewFromConfig builds the core from an already-loaded config.
func NewFromConfig(cfg *config.Config) (*Core, error) {
c := &Core{cfg: cfg}
c.vm = lua.NewVM(cfg.AdapterDir)
if err := c.vm.Start(); err != nil {
return nil, fmt.Errorf("lua vm: %w", err)
}
c.store = config.NewStore(cfg.RuntimeFile)
if err := c.store.Load(); err != nil {
return nil, fmt.Errorf("runtime store: %w", err)
}
c.scheduler = scheduler.New(buildRetries(cfg))
if err := c.seedKeys(); err != nil {
return nil, err
}
if err := c.seedAuto(); err != nil {
return nil, err
}
if err := c.rebuildRegistry(); err != nil {
return nil, err
}
return c, nil
}
// seedAuto migrates the legacy per-source model priority into flat AUTO
// scheduling slots (one slot per model, priority order) the first time no
// explicit auto rules exist.
func (c *Core) seedAuto() error {
if len(c.store.AutoRules()) > 0 {
return nil
}
type item struct {
model string
prio int
}
var flat []item
for _, s := range c.mergedSources() {
for _, m := range s.Models {
if m.Kind == "image" {
continue
}
flat = append(flat, item{m.ID, m.Priority})
}
}
sort.SliceStable(flat, func(i, j int) bool {
if flat[i].prio != flat[j].prio {
return flat[i].prio > flat[j].prio
}
return flat[i].model < flat[j].model
})
entries := make([]config.ModelScope, 0, len(flat))
for _, it := range flat {
entries = append(entries, config.ModelScope{Model: it.model})
}
return c.store.SaveAutoRules(entries)
}
// seedKeys migrates the static config gateway_keys into the runtime store as
// admin keys (once), so later UI-created keys can share the same store.
func (c *Core) seedKeys() error {
existing := map[string]bool{}
for _, k := range c.store.ListKeys() {
existing[k.Key] = true
}
changed := false
for i, raw := range c.cfg.GatewayKeys {
if raw == "" || existing[raw] {
continue
}
name := "admin"
if i > 0 {
name = fmt.Sprintf("admin-%d", i+1)
}
if err := c.store.SaveKey(config.GWKey{
Key: raw,
Role: "admin",
Name: name,
CreatedAt: time.Now().Unix(),
}); err != nil {
return err
}
changed = true
}
if changed {
return c.store.Load()
}
return nil
}
func buildRetries(cfg *config.Config) int {
return len(cfg.Sources) // allow fallback across all sources
}
// VM exposes the Lua adapter runtime.
func (c *Core) VM() *lua.VM { return c.vm }
func (c *Core) Scheduler() *scheduler.Scheduler { return c.scheduler }
func (c *Core) Registry() *provider.Registry { return c.registry }
func (c *Core) DefaultModel() string { return c.cfg.DefaultModel }
func (c *Core) GatewayKeys() []string { return c.cfg.GatewayKeys }
func (c *Core) Listen() string { return c.cfg.Listen }
// ---- gateway key management (web UI) ----
// ListKeys returns all gateway keys (admin view).
func (c *Core) ListKeys() []config.GWKey { return c.store.ListKeys() }
// FindKey looks up a gateway key record by its secret value.
func (c *Core) FindKey(key string) (config.GWKey, bool) { return c.store.KeyByValue(key) }
// CreateKey builds a new random gateway key and persists it.
func (c *Core) CreateKey(name, role string, models []config.ModelScope, note string) (config.GWKey, error) {
key := make([]byte, 16)
if _, err := rand.Read(key); err != nil {
return config.GWKey{}, err
}
rec := config.GWKey{
Key: "sk-gw-" + hex.EncodeToString(key),
Role: role,
Name: name,
Models: models,
Note: note,
CreatedAt: time.Now().Unix(),
}
if rec.Role == "" {
rec.Role = "user"
}
if err := c.store.SaveKey(rec); err != nil {
return config.GWKey{}, err
}
return rec, nil
}
// UpdateKey mutates a key's name/role/model scope and persists it.
func (c *Core) UpdateKey(key, name, role string, models []config.ModelScope, note string) (config.GWKey, error) {
rec, ok := c.store.KeyByValue(key)
if !ok {
return config.GWKey{}, fmt.Errorf("key not found")
}
if name != "" {
rec.Name = name
}
if role == "admin" || role == "user" {
rec.Role = role
}
rec.Models = models
rec.Note = note
if err := c.store.SaveKey(rec); err != nil {
return config.GWKey{}, err
}
return rec, nil
}
// DeleteKey removes a key record; returns false if it did not exist.
func (c *Core) DeleteKey(key string) (bool, error) { return c.store.DeleteKey(key) }
// ---- AUTO scheduling slots (web UI canvas) ----
// AutoRules returns the AUTO scheduling slots in priority order (slot 0 =
// highest priority).
func (c *Core) AutoRules() []config.ModelScope { return c.store.AutoRules() }
// SaveAutoRules persists the AUTO scheduling slots.
func (c *Core) SaveAutoRules(entries []config.ModelScope) error {
clean := make([]config.ModelScope, 0, len(entries))
for _, e := range entries {
if e.Model == "" {
continue
}
clean = append(clean, e)
}
return c.store.SaveAutoRules(clean)
}
// Registry resolves model -> owning provider.
func (c *Core) ProviderForModel(model string) *provider.Provider {
return c.registry.ProviderForModel(model)
}
// Config exposes the underlying configuration (read-only usage).
func (c *Core) Config() *config.Config { return c.cfg }
// mergedSources = base YAML sources + runtime sources (runtime wins by name).
func (c *Core) mergedSources() []config.Source {
byName := map[string]config.Source{}
order := []string{}
for _, s := range c.cfg.Sources {
byName[s.Name] = s
order = append(order, s.Name)
}
for _, s := range c.store.List() {
if _, ok := byName[s.Name]; !ok {
order = append(order, s.Name)
}
byName[s.Name] = s
}
out := make([]config.Source, 0, len(order))
seen := map[string]bool{}
for _, n := range order {
if !seen[n] {
seen[n] = true
out = append(out, byName[n])
}
}
return out
}
func (c *Core) rebuildRegistry() error {
srcs := c.mergedSources()
providers := make([]*provider.Provider, 0, len(srcs))
adapterConcurrency := map[string]int{}
for _, s := range srcs {
providers = append(providers, provider.New(s, c.vm))
if s.Adapter != "" {
adapterConcurrency[s.Adapter] += s.MaxConcurrent
}
}
c.vm.ConfigureConcurrency(adapterConcurrency)
if c.registry == nil {
c.registry = provider.NewRegistry(providers, c.cfg.DefaultModel)
} else {
c.registry.Replace(providers)
}
return nil
}
// Reload re-reads the runtime store and rebuilds sources (adapter reload is not
// strictly needed since adapters are loaded into the VM at startup; uploaded
// adapters are placed in the adapter dir and loaded by the web UI).
func (c *Core) Reload() error {
if err := c.store.Load(); err != nil {
return err
}
return c.rebuildRegistry()
}
// ---- adapter management (web UI) ----
func (c *Core) ListAdapters() []lua.APIAdapter { return c.vm.ListAdapters() }
// UploadAdapter saves a new Lua adapter script to the adapter dir and loads it.
func (c *Core) UploadAdapter(name, code string) error {
if name == "" {
return fmt.Errorf("adapter name required")
}
if err := os.MkdirAll(c.cfg.AdapterDir, 0755); err != nil {
return err
}
path := filepath.Join(c.cfg.AdapterDir, name+".lua")
if err := os.WriteFile(path, []byte(code), 0644); err != nil {
return err
}
if err := c.vm.LoadAdapter(path); err != nil {
return fmt.Errorf("load adapter: %w", err)
}
return nil
}
// RemoveAdapter deletes an adapter script and evicts it from the VM.
func (c *Core) RemoveAdapter(name string) error {
path := filepath.Join(c.cfg.AdapterDir, name+".lua")
_ = os.Remove(path)
c.vm.RemoveAdapter(name)
return nil
}
// ---- source management (web UI) ----
func (c *Core) AddSource(src config.Source) error {
if err := normalizeSource(&src); err != nil {
return err
}
if err := c.store.Upsert(src); err != nil {
return err
}
return c.rebuildRegistry()
}
func (c *Core) RemoveSource(name string) error {
if _, err := c.store.Remove(name); err != nil {
return err
}
return c.rebuildRegistry()
}
func (c *Core) Sources() []config.Source { return c.mergedSources() }
func normalizeSource(s *config.Source) error {
if s.Name == "" || s.BaseURL == "" {
return fmt.Errorf("source requires name and base_url")
}
if len(s.Models) == 0 {
return fmt.Errorf("source requires at least one model")
}
if s.Adapter == "" {
s.Adapter = "openai"
}
if s.MaxConcurrent == 0 {
s.MaxConcurrent = 8
}
return nil
}
// Close releases resources.
func (c *Core) Close() {
if c.vm != nil {
c.vm.Stop()
}
}