engine.go

// Package optaris is a pure-forwarding LLM API routing engine, embedded as a library in the
// consumer's gateway.
//
// Three disciplines:
//   - **Zero storage**: only holds an in-memory config snapshot, loaded via New / LoadConfig and
//     read back via Snapshot; persistence (sqlite / redis / file / none) is entirely the consumer's
//     job.
//   - **Zero auth / zero tenancy**: knows nothing about API Keys, nor about "users". Which group
//     each request routes to is injected by the consumer via RouteContext; "who is consuming" lives
//     entirely on the consumer side, correlated back through the optional RequestID to OnEvent.
//   - **Pure observability**: request-lifecycle stages surface synchronously through OnEvent
//     (including raw capture); the engine itself persists nothing.
//
// Data plane: one http.Handler per inbound format (see the *Handler methods); the consumer mounts
// them on its own mux, wrapping them with its own auth middleware.
package optaris

import (
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"github.com/getoptaris/optaris-core/internal/affinity"
	"github.com/getoptaris/optaris-core/internal/cache"
	"github.com/getoptaris/optaris-core/internal/stats"
	"github.com/getoptaris/optaris-core/model"
	"github.com/getoptaris/optaris-core/settings"
)

const (
	affinityMaxEntries = 100_000 // session-affinity ledger cap (lazy expiry + cap as a backstop, no background GC)
	statsBuckets       = 60      // number of buckets in the stats sliding window (bucket width = StatsWindow / 60)
)

// Engine is the engine instance, holding the config snapshot and runtime stats. Once constructed,
// mount its Handlers to serve requests; LoadConfig can hot-reload the config at any time without
// affecting in-flight requests.
type Engine struct {
	cache    *cache.Cache
	settings *settingsHolder
	window   *stats.Window
	cooldown *stats.Cooldown
	affinity affinity.Store
	now      func() time.Time
	hooks    Hooks

	mu   sync.RWMutex // guards subs registration and snapshot reads
	subs []func(Event)
}

// New constructs the engine from an initial config. The stats sliding window's bucket width is
// **fixed once** here from initial.Settings.StatsWindow (a later LoadConfig changing settings does
// not rebuild the window, to avoid clearing accumulated samples).
func New(initial Config, opts ...Option) *Engine {
	o := options{now: time.Now}
	for _, opt := range opts {
		opt(&o)
	}

	e := &Engine{
		cache:    cache.New(),
		settings: newSettingsHolder(initial.Settings),
		now:      o.now,
		hooks:    o.hooks,
	}
	e.window = stats.NewWindow(time.Duration(initial.Settings.StatsWindow), statsBuckets, e.now)
	e.cooldown = stats.NewCooldown(e.now)
	if o.affinity != nil {
		e.affinity = o.affinity
	} else {
		e.affinity = affinity.NewMemory(affinityMaxEntries, e.now)
	}
	e.cache.Replace(initial.Channels, initial.Groups)
	return e
}

// LoadConfig atomically replaces the current snapshot with a brand-new config. Subsequent requests
// take effect immediately, in-flight requests are unaffected. The stats window's bucket width is
// not rebuilt along with it (see New).
func (e *Engine) LoadConfig(c Config) {
	e.cache.Replace(c.Channels, c.Groups)
	e.settings.Replace(c.Settings)
}

// Snapshot reads back the current config (for the consumer to persist). Channel / group order is
// not guaranteed.
func (e *Engine) Snapshot() Config {
	snap := e.cache.Current()
	chPtrs := snap.Channels()
	channels := make([]model.Channel, 0, len(chPtrs))
	for _, ch := range chPtrs {
		channels = append(channels, *ch)
	}
	grPtrs := snap.Groups()
	groups := make([]model.Group, 0, len(grPtrs))
	for _, g := range grPtrs {
		groups = append(groups, *g)
	}
	return Config{Channels: channels, Groups: groups, Settings: e.settings.Current()}
}

// GroupModels returns the de-duplicated, sorted set of model ids the given group can serve: the
// union of Models over the group's enabled member channels. It mirrors the enabled-only filter and
// exact-match model semantics of the request router (see internal/router.Candidates), but
// deliberately ignores runtime cooldown and the client allowlist — it answers "what models does
// this group declare", which is what a client-facing model listing (e.g. GET /v1/models) reports.
// Returns an empty (non-nil) slice when the group is missing or empty.
func (e *Engine) GroupModels(groupID string) []string {
	snap := e.cache.Current()
	g, ok := snap.Group(groupID)
	if !ok {
		return []string{}
	}
	seen := make(map[string]struct{})
	for _, chID := range g.ChannelIDs {
		ch, ok := snap.Channel(chID)
		if !ok || !ch.Enabled {
			continue
		}
		for _, m := range ch.Models {
			seen[m] = struct{}{}
		}
	}
	out := make([]string, 0, len(seen))
	for m := range seen {
		out = append(out, m)
	}
	sort.Strings(out)
	return out
}

// OnEvent registers a lifecycle subscriber. The callback is invoked **synchronously** within the
// request goroutine and the engine recovers panics; subscribers should stay **non-blocking** (to
// persist, enqueue and return immediately). Multiple may be registered, invoked in registration
// order.
func (e *Engine) OnEvent(fn func(Event)) {
	if fn == nil {
		return
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	e.subs = append(e.subs, fn)
}

// settingsHolder holds an atomic snapshot of the global settings. Settings is a value type, stored
// by address via atomic.Pointer: reading Current() takes a lock-free value copy, writing Replace
// atomically swaps the pointer.
type settingsHolder struct {
	p atomic.Pointer[settings.Settings]
}

func newSettingsHolder(s settings.Settings) *settingsHolder {
	h := &settingsHolder{}
	h.p.Store(&s)
	return h
}

// Current returns a value copy of the current settings (the hot path reads it once at the request
// entry and reuses it for the whole lifecycle).
func (h *settingsHolder) Current() settings.Settings { return *h.p.Load() }

// Replace atomically replaces the current settings.
func (h *settingsHolder) Replace(s settings.Settings) { h.p.Store(&s) }