diff options
Diffstat (limited to '')
| -rw-r--r-- | internal/adminapi/api.go | 470 |
1 files changed, 456 insertions, 14 deletions
diff --git a/internal/adminapi/api.go b/internal/adminapi/api.go index 9f460e3..ff87ac9 100644 --- a/internal/adminapi/api.go +++ b/internal/adminapi/api.go @@ -13,13 +13,18 @@ import ( "log/slog" "net" "net/http" + "sort" + "strconv" "strings" "time" "aigw/internal/adminui" "aigw/internal/apierror" "aigw/internal/billing" + "aigw/internal/catalog" "aigw/internal/controlplane" + "aigw/internal/domain" + "aigw/internal/providerhealth" "github.com/go-webauthn/webauthn/webauthn" "github.com/jackc/pgx/v5/pgconn" @@ -38,6 +43,10 @@ type API struct { publicURL string webauthn *webauthn.WebAuthn mailEnabled bool + inferencePublicURL string + defaultLowBalance int64 + catalog *catalog.Catalog + health *providerhealth.Tracker } type actorKey struct{} @@ -58,18 +67,22 @@ func (w *auditWriter) Write(body []byte) (int, error) { } type Options struct { - Store *controlplane.Store - Manager *controlplane.Manager - Billing *billing.Service - Token string - Logger *slog.Logger - Prefix string - RegistrationEnabled bool - SessionTTL time.Duration - Currency string - PublicURL string - WebAuthn *webauthn.WebAuthn - MailEnabled bool + Store *controlplane.Store + Manager *controlplane.Manager + Billing *billing.Service + Token string + Logger *slog.Logger + Prefix string + RegistrationEnabled bool + SessionTTL time.Duration + Currency string + PublicURL string + WebAuthn *webauthn.WebAuthn + MailEnabled bool + InferencePublicURL string + DefaultLowBalanceMicros int64 + Catalog *catalog.Catalog + ProviderHealth *providerhealth.Tracker } func New(options Options) *API { @@ -89,7 +102,8 @@ func New(options Options) *API { return &API{store: options.Store, manager: options.Manager, billing: options.Billing, token: []byte(options.Token), logger: options.Logger, prefix: prefix, registrationEnabled: options.RegistrationEnabled, sessionTTL: options.SessionTTL, currency: options.Currency, publicURL: strings.TrimRight(options.PublicURL, "/") + "/", - webauthn: options.WebAuthn, mailEnabled: options.MailEnabled} + webauthn: options.WebAuthn, mailEnabled: options.MailEnabled, inferencePublicURL: strings.TrimRight(options.InferencePublicURL, "/"), + defaultLowBalance: options.DefaultLowBalanceMicros, catalog: options.Catalog, health: options.ProviderHealth} } func (a *API) Handler() http.Handler { @@ -103,6 +117,8 @@ func (a *API) Handler() http.Handler { http.Redirect(w, r, target, http.StatusTemporaryRedirect) }) mux.Handle(a.prefix+"/", http.StripPrefix(a.prefix, adminui.Handler())) + mux.HandleFunc("GET "+apiPrefix+"/public/models", a.public(a.publicModels)) + mux.HandleFunc("GET "+apiPrefix+"/public/models/{id...}", a.public(a.publicModel)) mux.HandleFunc("GET "+apiPrefix+"/auth/config", a.public(a.authConfig)) mux.HandleFunc("GET "+apiPrefix+"/auth/session", a.public(a.authSession)) mux.HandleFunc("POST "+apiPrefix+"/auth/register", a.public(a.register)) @@ -131,6 +147,11 @@ func (a *API) Handler() http.Handler { mux.HandleFunc("POST "+apiPrefix+"/auth/passkeys/{id}/delete", a.withAuth("overview.read", a.deletePasskey)) mux.HandleFunc("GET "+apiPrefix+"/overview", a.withAuth("overview.read", a.overview)) + mux.HandleFunc("GET "+apiPrefix+"/developer/config", a.withAuth("overview.read", a.developerConfig)) + mux.HandleFunc("GET "+apiPrefix+"/developer/models", a.withAuth("overview.read", a.developerModels)) + mux.HandleFunc("GET "+apiPrefix+"/developer/preferences", a.withAuth("preferences.read", a.developerPreferences)) + mux.HandleFunc("PUT "+apiPrefix+"/developer/preferences", a.withAuth("developer.preferences.write", a.updateDeveloperPreferences)) + mux.HandleFunc("PUT "+apiPrefix+"/developer/preferences/billing", a.withAuth("billing.preferences.write", a.updateBillingPreferences)) mux.HandleFunc("GET "+apiPrefix+"/tenants", a.withAuth("tenants.read", a.listTenants)) mux.HandleFunc("POST "+apiPrefix+"/tenants", a.withAuth("tenants.write", a.createTenant)) mux.HandleFunc("GET "+apiPrefix+"/projects", a.withAuth("projects.read", a.listProjects)) @@ -148,12 +169,17 @@ func (a *API) Handler() http.Handler { mux.HandleFunc("POST "+apiPrefix+"/reload", a.withAuth("platform.write", a.reload)) if a.billing != nil { mux.HandleFunc("GET "+apiPrefix+"/billing/accounts", a.withAuth("billing.read", a.listBillingAccounts)) + mux.HandleFunc("GET "+apiPrefix+"/billing/profile", a.withAuth("billing.read", a.getBillingProfile)) + mux.HandleFunc("PUT "+apiPrefix+"/billing/profile", a.withAuth("billing.topup", a.updateBillingProfile)) mux.HandleFunc("GET "+apiPrefix+"/billing/ledger", a.withAuth("billing.read", a.listBillingLedger)) mux.HandleFunc("GET "+apiPrefix+"/billing/orders", a.withAuth("billing.read", a.listTopUpOrders)) mux.HandleFunc("GET "+apiPrefix+"/billing/orders/{id}", a.withAuth("billing.read", a.getTopUpOrder)) mux.HandleFunc("POST "+apiPrefix+"/billing/adjustments", a.withAuth("billing.adjust", a.adjustBalance)) mux.HandleFunc("POST "+apiPrefix+"/billing/checkout-sessions", a.withAuth("billing.topup", a.createCheckoutSession)) mux.HandleFunc("POST "+apiPrefix+"/billing/portal-sessions", a.withAuth("billing.topup", a.createPortalSession)) + mux.HandleFunc("GET "+apiPrefix+"/billing/auto-topup", a.withAuth("billing.read", a.getAutoTopUp)) + mux.HandleFunc("PUT "+apiPrefix+"/billing/auto-topup", a.withAuth("billing.topup", a.updateAutoTopUp)) + mux.HandleFunc("POST "+apiPrefix+"/billing/auto-topup/setup-sessions", a.withAuth("billing.topup", a.createAutoTopUpSetupSession)) mux.HandleFunc("POST "+apiPrefix+"/billing/orders/{id}/retry", a.withAuth("billing.topup", a.retryCheckoutSession)) mux.HandleFunc("POST "+apiPrefix+"/billing/orders/{id}/resolve-missing", a.withAuth("billing.adjust", a.resolveMissingTopUp)) mux.HandleFunc("POST "+apiPrefix+"/billing/orders/{id}/reverse-missing-credit", a.withAuth("billing.adjust", a.reverseMissingTopUpCredit)) @@ -166,6 +192,8 @@ func (a *API) Handler() http.Handler { } mux.HandleFunc("GET "+apiPrefix+"/usage", a.withAuth("usage.read", a.listUsage)) mux.HandleFunc("GET "+apiPrefix+"/usage/summary", a.withAuth("usage.read", a.usageSummary)) + mux.HandleFunc("GET "+apiPrefix+"/usage/daily", a.withAuth("usage.read", a.usageDaily)) + mux.HandleFunc("GET "+apiPrefix+"/usage/analytics", a.withAuth("usage.read", a.usageAnalytics)) mux.HandleFunc("GET "+apiPrefix+"/limits", a.withAuth("limits.read", a.listLimits)) mux.HandleFunc("POST "+apiPrefix+"/limits/{project_id}", a.withAuth("limits.write", a.setLimit)) mux.HandleFunc("GET "+apiPrefix+"/users", a.withAuth("users.read", a.listUsers)) @@ -932,6 +960,228 @@ func (a *API) overview(w http.ResponseWriter, r *http.Request) { writeJSON(w, result) } +func (a *API) developerConfig(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, map[string]any{ + "base_url": a.inferencePublicURL, + "endpoints": map[string]string{ + "chat_completions": "/v1/chat/completions", + "responses": "/v1/responses", + "messages": "/anthropic/v1/messages", + "models": "/v1/models", + }, + }) +} + +func (a *API) developerModels(w http.ResponseWriter, r *http.Request) { + result, err := a.store.ListDeveloperModels(r.Context(), a.actor(r).TenantID) + if err != nil { + a.databaseError(w, r, err) + return + } + if err := a.addDeveloperModelHealth(r.Context(), result); err != nil { + a.databaseError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) publicModels(w http.ResponseWriter, r *http.Request) { + result, err := a.store.ListPublicModels(r.Context()) + if err != nil { + a.databaseError(w, r, err) + return + } + a.addPublicModelHealth(result) + w.Header().Set("Cache-Control", "public, max-age=30, stale-while-revalidate=120") + writeJSON(w, map[string]any{ + "data": result, + "inference_base_url": a.inferencePublicURL, + "registration_enabled": a.registrationEnabled, + }) +} + +func (a *API) publicModel(w http.ResponseWriter, r *http.Request) { + wanted := strings.Trim(strings.TrimSpace(r.PathValue("id")), "/") + result, err := a.store.ListPublicModels(r.Context()) + if err != nil { + a.databaseError(w, r, err) + return + } + a.addPublicModelHealth(result) + for _, item := range result { + if item.PublicID == wanted { + w.Header().Set("Cache-Control", "public, max-age=30, stale-while-revalidate=120") + writeJSON(w, item) + return + } + } + apierror.Write(w, apierror.Error{Status: http.StatusNotFound, Type: "model_not_found", Message: "Model not found"}, requestID(r)) +} + +func (a *API) addPublicModelHealth(models []controlplane.PublicModel) { + if a.catalog == nil { + return + } + statusByRoute := make(map[providerhealth.RouteKey]providerhealth.Status) + if a.health != nil { + for _, item := range a.health.Snapshot() { + statusByRoute[providerhealth.RouteKey{ModelID: item.ModelID, ProviderID: item.ProviderID, WireAPI: item.WireAPI}] = item + } + } + for index := range models { + model, err := a.catalog.Model(models[index].PublicID) + if err != nil { + continue + } + seen := make(map[string]struct{}) + available := 0 + for _, route := range model.Routes { + wireAPI := route.Provider.EffectiveWireAPI() + dedupe := route.Provider.ID + "\x00" + wireAPI + if _, exists := seen[dedupe]; exists { + continue + } + seen[dedupe] = struct{}{} + status, measured := statusByRoute[providerhealth.RouteKey{ModelID: model.ID, ProviderID: route.Provider.ID, WireAPI: wireAPI}] + if !measured || status.State != "open" { + available++ + } + } + models[index].ProviderCount = len(seen) + models[index].AvailableProviderCount = available + switch { + case len(seen) == 0 || available == 0: + models[index].HealthStatus = "unavailable" + case available < len(seen): + models[index].HealthStatus = "degraded" + default: + models[index].HealthStatus = "available" + } + } +} + +func (a *API) addDeveloperModelHealth(ctx context.Context, models []controlplane.DeveloperModel) error { + if a.catalog == nil { + return nil + } + providers, err := a.store.ListProviders(ctx) + if err != nil { + return err + } + providersByID := make(map[string]controlplane.Provider, len(providers)) + for _, item := range providers { + providersByID[item.ID] = item + } + statusByRoute := make(map[providerhealth.RouteKey]providerhealth.Status) + if a.health != nil { + for _, item := range a.health.Snapshot() { + statusByRoute[providerhealth.RouteKey{ModelID: item.ModelID, ProviderID: item.ProviderID, WireAPI: item.WireAPI}] = item + } + } + for index := range models { + model, err := a.catalog.Model(models[index].PublicID) + if err != nil { + continue + } + seen := make(map[string]struct{}) + items := make([]controlplane.DeveloperProviderHealth, 0, len(model.Routes)) + available := 0 + for _, route := range model.Routes { + key := providerhealth.RouteKey{ModelID: model.ID, ProviderID: route.Provider.ID, WireAPI: route.Provider.EffectiveWireAPI()} + dedupe := key.ProviderID + "\x00" + key.WireAPI + if _, exists := seen[dedupe]; exists { + continue + } + seen[dedupe] = struct{}{} + status, measured := statusByRoute[key] + state := "unknown" + if measured { + state = status.State + } + if state != "open" { + available++ + } + provider := providersByID[route.Provider.ID] + items = append(items, controlplane.DeveloperProviderHealth{Slug: route.Provider.EffectiveSlug(), Name: provider.Name, + Protocol: string(route.Provider.Protocol), WireAPI: key.WireAPI, State: state, Attempts: status.Attempts, + RecentSamples: status.RecentSamples, AvailabilityPercent: status.AvailabilityPercent, + HeaderLatencyEWMA: status.HeaderLatencyEWMA, ConsecutiveFailures: status.ConsecutiveFailures, + LastObservedAt: status.LastObservedAt, CircuitOpenUntil: status.CircuitOpenUntil}) + } + sort.Slice(items, func(i, j int) bool { + iOpen := items[i].State == "open" + jOpen := items[j].State == "open" + if iOpen != jOpen { + return !iOpen + } + if items[i].Name != items[j].Name { + return items[i].Name < items[j].Name + } + return items[i].WireAPI < items[j].WireAPI + }) + models[index].Providers = items + models[index].ProviderCount = len(items) + models[index].AvailableProviderCount = available + switch { + case len(items) == 0 || available == 0: + models[index].HealthStatus = "unavailable" + case available < len(items): + models[index].HealthStatus = "degraded" + default: + models[index].HealthStatus = "online" + } + } + return nil +} + +func (a *API) developerPreferences(w http.ResponseWriter, r *http.Request) { + result, err := a.store.GetTenantPreferences(r.Context(), a.preferenceTenantID(r, ""), a.defaultLowBalance) + if err != nil { + a.databaseError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) updateDeveloperPreferences(w http.ResponseWriter, r *http.Request) { + var input controlplane.SetDeveloperPreferencesInput + if !decodeBody(w, r, &input) { + return + } + input.TenantID = a.preferenceTenantID(r, input.TenantID) + result, err := a.store.SetDeveloperPreferences(r.Context(), input) + if err != nil { + a.mutationError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) updateBillingPreferences(w http.ResponseWriter, r *http.Request) { + var input controlplane.SetBillingPreferencesInput + if !decodeBody(w, r, &input) { + return + } + input.TenantID = a.preferenceTenantID(r, input.TenantID) + result, err := a.store.SetBillingPreferences(r.Context(), input, a.defaultLowBalance) + if err != nil { + a.mutationError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) preferenceTenantID(r *http.Request, requested string) string { + actor := a.actor(r) + if actor.TenantID != "" { + return actor.TenantID + } + if requested = strings.TrimSpace(requested); requested != "" { + return requested + } + return strings.TrimSpace(r.URL.Query().Get("tenant_id")) +} + func (a *API) listBillingAccounts(w http.ResponseWriter, r *http.Request) { result, err := a.billing.ListAccounts(r.Context(), a.actor(r).TenantID) if err != nil { @@ -941,6 +1191,38 @@ func (a *API) listBillingAccounts(w http.ResponseWriter, r *http.Request) { writeJSON(w, result) } +func (a *API) getBillingProfile(w http.ResponseWriter, r *http.Request) { + tenantID := a.preferenceTenantID(r, "") + if tenantID == "" { + a.scopeError(w, r) + return + } + result, err := a.billing.GetBillingProfile(r.Context(), tenantID) + if err != nil { + a.billingError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) updateBillingProfile(w http.ResponseWriter, r *http.Request) { + var input billing.UpdateBillingProfileInput + if !decodeBody(w, r, &input) { + return + } + input.TenantID = a.preferenceTenantID(r, input.TenantID) + if input.TenantID == "" { + a.scopeError(w, r) + return + } + result, err := a.billing.UpdateBillingProfile(r.Context(), input) + if err != nil { + a.billingError(w, r, err) + return + } + writeJSON(w, result) +} + func (a *API) listBillingLedger(w http.ResponseWriter, r *http.Request) { tenantID := a.actor(r).TenantID if tenantID == "" { @@ -1022,6 +1304,46 @@ func (a *API) createPortalSession(w http.ResponseWriter, r *http.Request) { writeStatusJSON(w, http.StatusCreated, result) } +func (a *API) getAutoTopUp(w http.ResponseWriter, r *http.Request) { + tenantID := a.preferenceTenantID(r, "") + result, err := a.billing.GetAutoTopUpSettings(r.Context(), tenantID) + if err != nil { + a.billingError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) updateAutoTopUp(w http.ResponseWriter, r *http.Request) { + var input billing.UpdateAutoTopUpInput + if !decodeBody(w, r, &input) { + return + } + input.TenantID = a.preferenceTenantID(r, input.TenantID) + result, err := a.billing.UpdateAutoTopUp(r.Context(), input) + if err != nil { + a.billingError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) createAutoTopUpSetupSession(w http.ResponseWriter, r *http.Request) { + var input billing.AutoTopUpSetupInput + if !decodeBody(w, r, &input) { + return + } + actor := a.actor(r) + input.TenantID = a.preferenceTenantID(r, input.TenantID) + input.CustomerEmail = actor.Email + result, err := a.billing.CreateAutoTopUpSetupSession(r.Context(), input) + if err != nil { + a.billingError(w, r, err) + return + } + writeStatusJSON(w, http.StatusCreated, result) +} + func (a *API) retryCheckoutSession(w http.ResponseWriter, r *http.Request) { actor := a.actor(r) if actor.TenantID == "" { @@ -1383,7 +1705,11 @@ func (a *API) reload(w http.ResponseWriter, r *http.Request) { } func (a *API) listUsage(w http.ResponseWriter, r *http.Request) { - query := controlplane.UsageQuery{TenantID: a.actor(r).TenantID, ProjectID: r.URL.Query().Get("project_id"), Model: r.URL.Query().Get("model"), Limit: 200} + query, err := a.usageQuery(r) + if err != nil { + a.mutationError(w, r, err) + return + } result, err := a.store.ListUsage(r.Context(), query) if err != nil { a.databaseError(w, r, err) @@ -1392,6 +1718,101 @@ func (a *API) listUsage(w http.ResponseWriter, r *http.Request) { writeJSON(w, result) } +func (a *API) usageDaily(w http.ResponseWriter, r *http.Request) { + query, err := a.usageQuery(r) + if err != nil { + a.mutationError(w, r, err) + return + } + result, err := a.store.UsageDaily(r.Context(), query) + if err != nil { + a.databaseError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) usageAnalytics(w http.ResponseWriter, r *http.Request) { + query, err := a.usageQuery(r) + if err != nil { + a.mutationError(w, r, err) + return + } + result, err := a.store.UsageAnalytics(r.Context(), query) + if err != nil { + a.databaseError(w, r, err) + return + } + writeJSON(w, result) +} + +func (a *API) usageQuery(r *http.Request) (controlplane.UsageQuery, error) { + actor := a.actor(r) + tenantID := actor.TenantID + if tenantID == "" { + tenantID = strings.TrimSpace(r.URL.Query().Get("tenant_id")) + } + query := controlplane.UsageQuery{ + TenantID: tenantID, ProjectID: strings.TrimSpace(r.URL.Query().Get("project_id")), + KeyID: strings.TrimSpace(r.URL.Query().Get("key_id")), Model: strings.TrimSpace(r.URL.Query().Get("model")), + Provider: strings.ToLower(strings.TrimSpace(r.URL.Query().Get("provider"))), Protocol: strings.TrimSpace(r.URL.Query().Get("protocol")), + ErrorType: strings.TrimSpace(r.URL.Query().Get("error_type")), RequestID: strings.TrimSpace(r.URL.Query().Get("request_id")), + Status: strings.TrimSpace(r.URL.Query().Get("status")), Limit: 200, + } + if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" { + limit, err := strconv.Atoi(raw) + if err != nil || limit < 1 || limit > 1000 { + return controlplane.UsageQuery{}, errors.New("usage limit must be between 1 and 1000") + } + query.Limit = limit + } + if query.Status != "" && query.Status != "success" && query.Status != "error" { + return controlplane.UsageQuery{}, errors.New("usage status must be success or error") + } + if query.Protocol != "" && query.Protocol != string(domain.ProtocolOpenAI) && query.Protocol != string(domain.ProtocolOpenAIResponses) && query.Protocol != string(domain.ProtocolAnthropic) { + return controlplane.UsageQuery{}, errors.New("usage protocol is invalid") + } + if len(query.Provider) > 64 || len(query.ErrorType) > 128 { + return controlplane.UsageQuery{}, errors.New("usage provider or error type is too long") + } + if raw := strings.TrimSpace(r.URL.Query().Get("stream")); raw != "" { + value, err := strconv.ParseBool(raw) + if err != nil { + return controlplane.UsageQuery{}, errors.New("usage stream must be true or false") + } + query.Stream = &value + } + var err error + if query.From, err = parseUsageTime(r.URL.Query().Get("from"), false); err != nil { + return controlplane.UsageQuery{}, err + } + if query.To, err = parseUsageTime(r.URL.Query().Get("to"), true); err != nil { + return controlplane.UsageQuery{}, err + } + if !query.From.IsZero() && !query.To.IsZero() && !query.To.After(query.From) { + return controlplane.UsageQuery{}, errors.New("usage to must be after from") + } + return query, nil +} + +func parseUsageTime(raw string, endOfDay bool) (time.Time, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return time.Time{}, nil + } + if parsed, err := time.Parse(time.RFC3339, raw); err == nil { + return parsed.UTC(), nil + } + parsed, err := time.Parse("2006-01-02", raw) + if err != nil { + return time.Time{}, errors.New("usage dates must be RFC3339 or YYYY-MM-DD") + } + if endOfDay { + parsed = parsed.AddDate(0, 0, 1) + } + return parsed.UTC(), nil +} + func (a *API) usageSummary(w http.ResponseWriter, r *http.Request) { result, err := a.store.UsageSummary(r.Context(), a.actor(r).TenantID, r.URL.Query().Get("project_id")) if err != nil { @@ -1567,6 +1988,27 @@ func (a *API) billingError(w http.ResponseWriter, r *http.Request, err error) { status = http.StatusConflict typeName = "topup_order_not_resolvable" message = err.Error() + case errors.Is(err, billing.ErrBillingAccountNotFound): + status = http.StatusNotFound + typeName = "billing_account_not_found" + message = "Billing account was not found" + case errors.Is(err, billing.ErrPaymentMethodRequired): + status = http.StatusConflict + typeName = "payment_method_required" + message = "Save a payment method before enabling automatic top-up" + case errors.Is(err, billing.ErrAutoTopUpNeedsAttention): + status = http.StatusConflict + typeName = "payment_method_attention_required" + message = "Replace or re-authorize the saved payment method before enabling automatic top-up" + case errors.Is(err, billing.ErrInvalidBillingProfile): + status = http.StatusBadRequest + typeName = "invalid_billing_profile" + message = strings.TrimPrefix(err.Error(), billing.ErrInvalidBillingProfile.Error()+": ") + case errors.Is(err, billing.ErrBillingProfileSync): + status = http.StatusBadGateway + typeName = "billing_profile_sync_failed" + message = "Billing details were saved, but Stripe synchronization failed" + a.logger.Error("billing_profile_sync_failed", "error", err) default: a.logger.Error("admin_billing_error", "error", err) } |