Build AI gateway control plane and admin UI

2 files changed, 593 insertions, 0 deletions

diff --git a/internal/httpapi/api.go b/internal/httpapi/api.go
new file mode 100644
index 0000000..ed7939f
--- /dev/null
+++ b/internal/httpapi/api.go
@@ -0,0 +1,369 @@
+package httpapi
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"strings"
+	"time"
+
+	"aigw/internal/apierror"
+	"aigw/internal/auth"
+	"aigw/internal/catalog"
+	"aigw/internal/domain"
+	"aigw/internal/provider"
+	"aigw/internal/routing"
+	"aigw/internal/telemetry"
+	"aigw/internal/usage"
+)
+
+type requestIDKey struct{}
+
+type API struct {
+	authenticator auth.Authenticator
+	catalog       *catalog.Catalog
+	router        *routing.Router
+	forwarder     *provider.Forwarder
+	usageSink     telemetry.UsageSink
+	metrics       *telemetry.Metrics
+	logger        *slog.Logger
+	maxBodyBytes  int64
+	exposeMetrics bool
+}
+
+type Options struct {
+	Authenticator auth.Authenticator
+	Catalog       *catalog.Catalog
+	Router        *routing.Router
+	Forwarder     *provider.Forwarder
+	UsageSink     telemetry.UsageSink
+	Metrics       *telemetry.Metrics
+	Logger        *slog.Logger
+	MaxBodyBytes  int64
+	ExposeMetrics bool
+}
+
+func New(options Options) *API {
+	return &API{
+		authenticator: options.Authenticator,
+		catalog:       options.Catalog,
+		router:        options.Router,
+		forwarder:     options.Forwarder,
+		usageSink:     options.UsageSink,
+		metrics:       options.Metrics,
+		logger:        options.Logger,
+		maxBodyBytes:  options.MaxBodyBytes,
+		exposeMetrics: options.ExposeMetrics,
+	}
+}
+
+func (a *API) Handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /healthz", a.health)
+	mux.HandleFunc("GET /readyz", a.health)
+	if a.exposeMetrics {
+		mux.Handle("GET /metrics", a.metrics)
+	}
+
+	mux.HandleFunc("GET /v1/models", a.openAIModels)
+	mux.HandleFunc("GET /api/v1/models", a.openAIModels)
+	mux.HandleFunc("POST /v1/chat/completions", a.openAIChat)
+	mux.HandleFunc("POST /api/v1/chat/completions", a.openAIChat)
+
+	mux.HandleFunc("GET /anthropic/v1/models", a.anthropicModels)
+	mux.HandleFunc("GET /api/anthropic/v1/models", a.anthropicModels)
+	mux.HandleFunc("POST /anthropic/v1/messages", a.anthropicMessages)
+	mux.HandleFunc("POST /api/anthropic/v1/messages", a.anthropicMessages)
+
+	return a.withRequestID(a.recoverPanics(mux))
+}
+
+func (a *API) health(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	_, _ = io.WriteString(w, `{"status":"ok"}`+"\n")
+}
+
+func (a *API) openAIChat(w http.ResponseWriter, r *http.Request) {
+	a.serveInference(w, r, domain.ProtocolOpenAI)
+}
+
+func (a *API) anthropicMessages(w http.ResponseWriter, r *http.Request) {
+	a.serveInference(w, r, domain.ProtocolAnthropic)
+}
+
+func (a *API) serveInference(w http.ResponseWriter, r *http.Request, protocol domain.Protocol) {
+	requestID := requestIDFrom(r.Context())
+	startedAt := time.Now().UTC()
+	a.metrics.RequestStarted()
+	success := false
+	defer func() { a.metrics.RequestFinished(success) }()
+
+	principal, authErr := a.authenticator.Authenticate(r)
+	if authErr != nil {
+		apierror.Write(w, apierror.Error{Status: http.StatusForbidden, Type: "access_denied", Message: "Invalid or missing API key"}, requestID)
+		return
+	}
+	if !hasScope(principal, "inference") {
+		apierror.Write(w, apierror.Error{Status: http.StatusForbidden, Type: "access_denied", Message: "API key does not have inference permission"}, requestID)
+		return
+	}
+
+	body, err := readBody(w, r, a.maxBodyBytes)
+	if err != nil {
+		status := http.StatusBadRequest
+		message := "Invalid request body"
+		if errors.As(err, new(*http.MaxBytesError)) {
+			status = http.StatusRequestEntityTooLarge
+			message = "Request body is too large"
+		}
+		apierror.Write(w, apierror.Error{Status: status, Type: "invalid_params", Message: message}, requestID)
+		return
+	}
+
+	var envelope struct {
+		Model  string `json:"model"`
+		Stream bool   `json:"stream"`
+	}
+	if err := json.Unmarshal(body, &envelope); err != nil || strings.TrimSpace(envelope.Model) == "" {
+		apierror.Write(w, apierror.Error{Status: http.StatusBadRequest, Type: "invalid_params", Message: "Parameter model is required and the body must be valid JSON"}, requestID)
+		return
+	}
+
+	routes, err := a.router.Plan(envelope.Model, protocol)
+	if err != nil {
+		if errors.Is(err, routing.ErrNoRoute) {
+			apierror.Write(w, apierror.Error{Status: http.StatusNotFound, Type: "model_not_supported", Message: "Model does not support this API protocol"}, requestID)
+		} else {
+			apierror.Write(w, apierror.Error{Status: http.StatusNotFound, Type: "invalid_model", Message: "Model does not exist"}, requestID)
+		}
+		return
+	}
+
+	result, err := a.forwarder.Forward(r.Context(), protocol, requestID, body, r.Header, routes)
+	if err != nil {
+		errorType := "no_provider_available"
+		status := http.StatusBadGateway
+		if errors.Is(err, context.Canceled) {
+			errorType = "client_disconnected"
+			status = 499
+		} else {
+			apierror.Write(w, apierror.Error{Status: status, Type: errorType, Message: "No upstream provider is currently available"}, requestID)
+		}
+		a.publishUsage(domain.UsageEvent{
+			RequestID: requestID, KeyID: principal.KeyID, TenantID: principal.TenantID, ProjectID: principal.ProjectID,
+			PublicModel: envelope.Model, Protocol: protocol, Stream: envelope.Stream, StatusCode: status,
+			Success: false, ErrorType: errorType, Attempts: result.Attempts, StartedAt: startedAt, DurationMS: time.Since(startedAt).Milliseconds(),
+		})
+		return
+	}
+	defer result.Response.Body.Close()
+
+	if result.Response.StatusCode >= 400 {
+		gatewayError := normalizeProviderError(result.Response)
+		apierror.Write(w, gatewayError, requestID)
+		a.publishUsage(domain.UsageEvent{
+			RequestID: requestID, KeyID: principal.KeyID, TenantID: principal.TenantID, ProjectID: principal.ProjectID,
+			PublicModel: envelope.Model, ProviderID: result.Route.Provider.ID, UpstreamModel: result.Route.UpstreamModel,
+			Protocol: protocol, Stream: envelope.Stream, StatusCode: gatewayError.Status, Success: false,
+			ErrorType: gatewayError.Type, Attempts: result.Attempts, StartedAt: startedAt, DurationMS: time.Since(startedAt).Milliseconds(),
+		})
+		return
+	}
+
+	stream := envelope.Stream || strings.HasPrefix(strings.ToLower(result.Response.Header.Get("Content-Type")), "text/event-stream")
+	copyResponseHeaders(w.Header(), result.Response.Header, stream)
+	w.Header().Set("X-AIGW-Request-ID", requestID)
+	if stream {
+		w.Header().Set("X-Accel-Buffering", "no")
+	}
+	w.WriteHeader(result.Response.StatusCode)
+
+	observer := usage.NewObserver(protocol, stream)
+	copyErr := copyResponse(w, result.Response.Body, observer, stream)
+	usageResult := observer.Usage()
+	success = copyErr == nil
+	errorType := ""
+	if copyErr != nil {
+		errorType = "stream_interrupted"
+	}
+	a.publishUsage(domain.UsageEvent{
+		RequestID: requestID, KeyID: principal.KeyID, TenantID: principal.TenantID, ProjectID: principal.ProjectID,
+		PublicModel: envelope.Model, ProviderID: result.Route.Provider.ID, UpstreamModel: result.Route.UpstreamModel,
+		Protocol: protocol, Stream: stream, StatusCode: result.Response.StatusCode, Success: success,
+		ErrorType: errorType, Attempts: result.Attempts, StartedAt: startedAt, DurationMS: time.Since(startedAt).Milliseconds(), Usage: usageResult,
+	})
+	a.logger.Info("inference_request",
+		"request_id", requestID,
+		"tenant_id", principal.TenantID,
+		"project_id", principal.ProjectID,
+		"model", envelope.Model,
+		"provider", result.Route.Provider.ID,
+		"status", result.Response.StatusCode,
+		"attempts", result.Attempts,
+		"duration_ms", time.Since(startedAt).Milliseconds(),
+	)
+}
+
+func (a *API) openAIModels(w http.ResponseWriter, r *http.Request) {
+	if !a.authorize(w, r) {
+		return
+	}
+	models := a.catalog.Models(domain.ProtocolOpenAI)
+	data := make([]map[string]any, 0, len(models))
+	for _, model := range models {
+		data = append(data, map[string]any{"id": model.ID, "object": "model", "created": 0, "owned_by": model.OwnedBy})
+	}
+	writeJSON(w, map[string]any{"object": "list", "data": data})
+}
+
+func (a *API) anthropicModels(w http.ResponseWriter, r *http.Request) {
+	if !a.authorize(w, r) {
+		return
+	}
+	models := a.catalog.Models(domain.ProtocolAnthropic)
+	data := make([]map[string]any, 0, len(models))
+	for _, model := range models {
+		data = append(data, map[string]any{"id": model.ID, "display_name": model.ID, "created_at": "1970-01-01T00:00:00Z", "type": "model"})
+	}
+	response := map[string]any{"data": data, "has_more": false, "first_id": nil, "last_id": nil}
+	if len(models) > 0 {
+		response["first_id"] = models[0].ID
+		response["last_id"] = models[len(models)-1].ID
+	}
+	writeJSON(w, response)
+}
+
+func (a *API) authorize(w http.ResponseWriter, r *http.Request) bool {
+	principal, err := a.authenticator.Authenticate(r)
+	if err != nil || !hasScope(principal, "inference") {
+		apierror.Write(w, apierror.Error{Status: http.StatusForbidden, Type: "access_denied", Message: "Invalid API key or insufficient permission"}, requestIDFrom(r.Context()))
+		return false
+	}
+	return true
+}
+
+func (a *API) publishUsage(event domain.UsageEvent) {
+	if a.usageSink != nil {
+		a.usageSink.Publish(event)
+	}
+}
+
+func (a *API) withRequestID(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		id := newRequestID()
+		w.Header().Set("X-AIGW-Request-ID", id)
+		next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), requestIDKey{}, id)))
+	})
+}
+
+func (a *API) recoverPanics(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if recovered := recover(); recovered != nil {
+				a.logger.Error("request_panic", "request_id", requestIDFrom(r.Context()), "error", recovered)
+				apierror.Write(w, apierror.Error{Status: http.StatusInternalServerError, Type: "internal_server_error", Message: "Internal server error"}, requestIDFrom(r.Context()))
+			}
+		}()
+		next.ServeHTTP(w, r)
+	})
+}
+
+func newRequestID() string {
+	var value [16]byte
+	if _, err := rand.Read(value[:]); err != nil {
+		return fmt.Sprintf("req_%d", time.Now().UnixNano())
+	}
+	return "req_" + hex.EncodeToString(value[:])
+}
+
+func requestIDFrom(ctx context.Context) string {
+	id, _ := ctx.Value(requestIDKey{}).(string)
+	return id
+}
+
+func hasScope(principal domain.Principal, wanted string) bool {
+	if len(principal.Scopes) == 0 {
+		return true
+	}
+	for _, scope := range principal.Scopes {
+		if scope == wanted || scope == "*" {
+			return true
+		}
+	}
+	return false
+}
+
+func readBody(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, error) {
+	r.Body = http.MaxBytesReader(w, r.Body, limit)
+	defer r.Body.Close()
+	return io.ReadAll(r.Body)
+}
+
+func copyResponseHeaders(target, source http.Header, stream bool) {
+	for _, name := range []string{"Content-Type", "Cache-Control", "Retry-After"} {
+		if value := source.Get(name); value != "" {
+			target.Set(name, value)
+		}
+	}
+	if !stream {
+		if value := source.Get("Content-Length"); value != "" {
+			target.Set("Content-Length", value)
+		}
+	}
+}
+
+func copyResponse(w http.ResponseWriter, body io.Reader, observer io.Writer, stream bool) error {
+	var destination io.Writer = w
+	if stream {
+		destination = &flushingWriter{writer: w, controller: http.NewResponseController(w)}
+	}
+	_, err := io.CopyBuffer(io.MultiWriter(destination, observer), body, make([]byte, 32<<10))
+	return err
+}
+
+type flushingWriter struct {
+	writer     io.Writer
+	controller *http.ResponseController
+}
+
+func (w *flushingWriter) Write(p []byte) (int, error) {
+	n, err := w.writer.Write(p)
+	if err == nil {
+		_ = w.controller.Flush()
+	}
+	return n, err
+}
+
+func normalizeProviderError(response *http.Response) apierror.Error {
+	payload, _ := io.ReadAll(io.LimitReader(response.Body, 64<<10))
+	message := "Upstream provider rejected the request"
+	var common struct {
+		Error struct {
+			Message string `json:"message"`
+		} `json:"error"`
+	}
+	if json.Unmarshal(payload, &common) == nil && common.Error.Message != "" {
+		message = common.Error.Message
+	}
+	switch response.StatusCode {
+	case http.StatusBadRequest:
+		return apierror.Error{Status: http.StatusUnprocessableEntity, Type: "provider_unprocessable_entity_error", Message: message}
+	case http.StatusRequestEntityTooLarge:
+		return apierror.Error{Status: http.StatusRequestEntityTooLarge, Type: "invalid_params", Message: message}
+	case http.StatusTooManyRequests:
+		return apierror.Error{Status: http.StatusTooManyRequests, Type: "rate_limit", Message: message}
+	default:
+		return apierror.Error{Status: http.StatusBadGateway, Type: "provider_error", Message: message}
+	}
+}
+
+func writeJSON(w http.ResponseWriter, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(value)
+}
diff --git a/internal/httpapi/api_test.go b/internal/httpapi/api_test.go
new file mode 100644
index 0000000..66a344b
--- /dev/null
+++ b/internal/httpapi/api_test.go
@@ -0,0 +1,224 @@
+package httpapi
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"aigw/internal/auth"
+	"aigw/internal/catalog"
+	"aigw/internal/config"
+	"aigw/internal/domain"
+	"aigw/internal/provider"
+	"aigw/internal/routing"
+	"aigw/internal/telemetry"
+)
+
+type captureUsageSink struct {
+	events chan domain.UsageEvent
+}
+
+func (s *captureUsageSink) Publish(event domain.UsageEvent) {
+	s.events <- event
+}
+
+func TestOpenAIProxyRewritesModelAndEmitsUsage(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/chat/completions" {
+			t.Errorf("unexpected path: %s", r.URL.Path)
+		}
+		if r.Header.Get("Authorization") != "Bearer upstream-secret" {
+			t.Errorf("upstream authorization leaked or missing: %q", r.Header.Get("Authorization"))
+		}
+		var request map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+			t.Error(err)
+		}
+		if request["model"] != "upstream-model" {
+			t.Errorf("model was not rewritten: %+v", request)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"id":"chat-1","choices":[],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}`)
+	}))
+	defer upstream.Close()
+
+	gateway, sink := newTestGateway(t, []config.ProviderConfig{{
+		ID: "primary", Protocol: domain.ProtocolOpenAI, BaseURL: upstream.URL + "/v1", APIKey: "upstream-secret",
+	}}, []config.RouteConfig{{Provider: "primary", UpstreamModel: "upstream-model", Weight: 1}})
+	defer gateway.Close()
+
+	request, _ := http.NewRequest(http.MethodPost, gateway.URL+"/v1/chat/completions", strings.NewReader(`{"model":"public/model","messages":[{"role":"user","content":"hello"}]}`))
+	request.Header.Set("Authorization", "Bearer client-secret")
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		payload, _ := io.ReadAll(response.Body)
+		t.Fatalf("unexpected status %d: %s", response.StatusCode, payload)
+	}
+	if response.Header.Get("X-AIGW-Request-ID") == "" {
+		t.Fatal("missing gateway request id")
+	}
+
+	select {
+	case event := <-sink.events:
+		if event.PublicModel != "public/model" || event.UpstreamModel != "upstream-model" || event.Usage.TotalTokens != 5 || !event.Success {
+			t.Fatalf("unexpected usage event: %+v", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("usage event was not emitted")
+	}
+}
+
+func TestProxyFailsOverBeforeWritingResponse(t *testing.T) {
+	var primaryCalls atomic.Int64
+	primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		primaryCalls.Add(1)
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer primary.Close()
+	var fallbackCalls atomic.Int64
+	fallback := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		fallbackCalls.Add(1)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`)
+	}))
+	defer fallback.Close()
+
+	gateway, sink := newTestGateway(t,
+		[]config.ProviderConfig{
+			{ID: "primary", Protocol: domain.ProtocolOpenAI, BaseURL: primary.URL + "/v1", APIKey: "one"},
+			{ID: "fallback", Protocol: domain.ProtocolOpenAI, BaseURL: fallback.URL + "/v1", APIKey: "two"},
+		},
+		[]config.RouteConfig{
+			{Provider: "primary", UpstreamModel: "model", Priority: 0, Weight: 1},
+			{Provider: "fallback", UpstreamModel: "model", Priority: 10, Weight: 1},
+		},
+	)
+	defer gateway.Close()
+
+	response := postOpenAI(t, gateway.URL, false)
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK || primaryCalls.Load() != 1 || fallbackCalls.Load() != 1 {
+		t.Fatalf("failover did not complete: status=%d primary=%d fallback=%d", response.StatusCode, primaryCalls.Load(), fallbackCalls.Load())
+	}
+	event := <-sink.events
+	if event.Attempts != 2 || event.ProviderID != "fallback" {
+		t.Fatalf("unexpected failover event: %+v", event)
+	}
+}
+
+func TestSSEIsFlushedBeforeUpstreamCompletes(t *testing.T) {
+	release := make(chan struct{})
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, "data: {\"id\":\"chunk-1\",\"choices\":[]}\n\n")
+		w.(http.Flusher).Flush()
+		<-release
+		_, _ = io.WriteString(w, "data: [DONE]\n\n")
+	}))
+	defer upstream.Close()
+	gateway, _ := newTestGateway(t, []config.ProviderConfig{{
+		ID: "primary", Protocol: domain.ProtocolOpenAI, BaseURL: upstream.URL + "/v1", APIKey: "secret",
+	}}, []config.RouteConfig{{Provider: "primary", UpstreamModel: "model", Weight: 1}})
+	defer gateway.Close()
+
+	response := postOpenAI(t, gateway.URL, true)
+	defer response.Body.Close()
+	reader := bufio.NewReader(response.Body)
+	line, err := reader.ReadString('\n')
+	if err != nil {
+		close(release)
+		t.Fatal(err)
+	}
+	if !strings.Contains(line, "chunk-1") {
+		close(release)
+		t.Fatalf("unexpected first SSE line: %q", line)
+	}
+	close(release)
+}
+
+func TestAnthropicHeadersAndPath(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/messages" || r.Header.Get("x-api-key") != "anthropic-upstream" || r.Header.Get("anthropic-version") != "2023-06-01" {
+			t.Errorf("unexpected anthropic request: path=%s key=%q version=%q", r.URL.Path, r.Header.Get("x-api-key"), r.Header.Get("anthropic-version"))
+		}
+		_, _ = io.WriteString(w, `{"type":"message","content":[],"usage":{"input_tokens":4,"output_tokens":6}}`)
+	}))
+	defer upstream.Close()
+	gateway, sink := newTestGateway(t, []config.ProviderConfig{{
+		ID: "anthropic", Protocol: domain.ProtocolAnthropic, BaseURL: upstream.URL + "/v1", APIKey: "anthropic-upstream",
+	}}, []config.RouteConfig{{Provider: "anthropic", UpstreamModel: "claude-upstream", Weight: 1}})
+	defer gateway.Close()
+
+	request, _ := http.NewRequest(http.MethodPost, gateway.URL+"/anthropic/v1/messages", strings.NewReader(`{"model":"public/model","max_tokens":10,"messages":[{"role":"user","content":"hello"}]}`))
+	request.Header.Set("x-api-key", "client-secret")
+	request.Header.Set("anthropic-version", "2023-06-01")
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status: %d", response.StatusCode)
+	}
+	event := <-sink.events
+	if event.Usage.InputTokens != 4 || event.Usage.OutputTokens != 6 {
+		t.Fatalf("unexpected anthropic usage: %+v", event.Usage)
+	}
+}
+
+func newTestGateway(t *testing.T, providers []config.ProviderConfig, routes []config.RouteConfig) (*httptest.Server, *captureUsageSink) {
+	t.Helper()
+	authenticator, err := auth.NewStatic(`[{"key":"client-secret","key_id":"key-1","tenant_id":"tenant-1","project_id":"project-1","scopes":["inference"]}]`, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg := config.Config{
+		Providers: providers,
+		Models:    []config.ModelConfig{{ID: "public/model", OwnedBy: "test", Routes: routes}},
+		UpstreamHTTP: config.UpstreamHTTPConfig{
+			MaxIdleConnections: 100, MaxIdleConnectionsPerHost: 20,
+			IdleConnectionTimeoutSecs: 10, ResponseHeaderTimeoutSecs: 2,
+		},
+	}
+	metrics := &telemetry.Metrics{}
+	modelCatalog := catalog.New(cfg)
+	sink := &captureUsageSink{events: make(chan domain.UsageEvent, 10)}
+	api := New(Options{
+		Authenticator: authenticator,
+		Catalog:       modelCatalog,
+		Router:        routing.New(modelCatalog),
+		Forwarder:     provider.New(cfg.UpstreamHTTP, metrics),
+		UsageSink:     sink,
+		Metrics:       metrics,
+		Logger:        slog.New(slog.NewTextHandler(io.Discard, nil)),
+		MaxBodyBytes:  1 << 20,
+	})
+	return httptest.NewServer(api.Handler()), sink
+}
+
+func postOpenAI(t *testing.T, gatewayURL string, stream bool) *http.Response {
+	t.Helper()
+	payload := []byte(`{"model":"public/model","messages":[{"role":"user","content":"hello"}],"stream":false}`)
+	if stream {
+		payload = []byte(`{"model":"public/model","messages":[{"role":"user","content":"hello"}],"stream":true}`)
+	}
+	request, _ := http.NewRequest(http.MethodPost, gatewayURL+"/v1/chat/completions", bytes.NewReader(payload))
+	request.Header.Set("Authorization", "Bearer client-secret")
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return response
+}