From 3f702084d20b3c3a3ea916f3110e99b22bda60b3 Mon Sep 17 00:00:00 2001 From: Chia Date: Thu, 6 Aug 2026 15:58:57 +1200 Subject: feat: complete commercial developer workflows Add tenant-safe usage observability, prepaid billing controls, API key lifecycle management, Embeddings metering, configurable billing alerts, and resilient provider health propagation. Harden Stripe failure handling, migrations, readiness, and the authenticated control-plane UI with end-to-end verification evidence. --- cmd/aigw/main.go | 34 ++++++++++++++++++++++++++++++---- cmd/mockupstream/main.go | 31 +++++++++++++++++++++++++++++++ cmd/stripe-preflight/main.go | 34 ++++++++++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 4 deletions(-) create mode 100644 cmd/stripe-preflight/main.go (limited to 'cmd') diff --git a/cmd/aigw/main.go b/cmd/aigw/main.go index 3ac388a..60956a8 100644 --- a/cmd/aigw/main.go +++ b/cmd/aigw/main.go @@ -9,6 +9,7 @@ import ( "net/http" "os" "os/signal" + "strconv" "syscall" "time" @@ -180,6 +181,27 @@ func run(ctx context.Context, cfg config.Config, logger *slog.Logger) error { billingMeter = billingService } routeHealth := providerhealth.New(providerhealth.Options{}) + var sharedHealth *providerhealth.RedisHistory + if cfg.ProviderHealth.SharedHistoryEnabled { + instanceID := cfg.Server.DeploymentRegion + ":" + strconv.Itoa(os.Getpid()) + ":" + strconv.FormatInt(time.Now().UnixNano(), 36) + sharedHealth = providerhealth.NewRedisHistory(providerhealth.HistoryOptions{ + Enabled: true, RedisURL: cfg.ControlPlane.RedisURL, Stream: cfg.ProviderHealth.SharedHistoryStream, + Instance: instanceID, MaxEvents: cfg.ProviderHealth.SharedHistoryMaxEvents, + TTL: time.Duration(cfg.ProviderHealth.SharedHistoryTTLSeconds) * time.Second, Logger: logger, Metrics: metrics, + }) + if sharedHealth != nil { + routeHealth.SetSink(sharedHealth) + go sharedHealth.Run(ctx, routeHealth) + defer sharedHealth.Close() + } else { + logger.Warn("provider_health_shared_history_disabled", "reason", "redis_not_configured", "fallback", "local") + } + } + go providerhealth.NewProber(providerhealth.ProbeOptions{ + Enabled: cfg.ProviderHealth.ActiveProbesEnabled, Interval: time.Duration(cfg.ProviderHealth.ProbeIntervalSeconds) * time.Second, + Timeout: time.Duration(cfg.ProviderHealth.ProbeTimeoutSeconds) * time.Second, Catalog: modelCatalog, + Tracker: routeHealth, Metrics: metrics, Logger: logger, + }).Run(ctx) inferenceAPI := httpapi.New(httpapi.Options{ Authenticator: authenticator, @@ -204,10 +226,14 @@ func run(ctx context.Context, cfg config.Config, logger *slog.Logger) error { Logger: logger, Prefix: cfg.Admin.BasePath, RegistrationEnabled: cfg.Admin.RegistrationEnabled, SessionTTL: time.Duration(cfg.Admin.SessionTTLHours) * time.Hour, Currency: cfg.Billing.Currency, PublicURL: cfg.Admin.PublicURL, WebAuthn: webAuthn, MailEnabled: cfg.Admin.Mail.Enabled, - InferencePublicURL: cfg.Admin.InferencePublicURL, - DefaultLowBalanceMicros: cfg.Admin.Mail.LowBalanceMicros, - Catalog: modelCatalog, - ProviderHealth: routeHealth, + InferencePublicURL: cfg.Admin.InferencePublicURL, + BillingPreferenceDefaults: controlplane.BillingPreferenceDefaults{ + LowBalanceThresholdMicros: cfg.Admin.Mail.LowBalanceMicros, + SpendAnomalyMultiplier: cfg.Admin.Mail.SpendAnomalyMultiplier, + SpendAnomalyMinMicros: cfg.Admin.Mail.SpendAnomalyMinMicros, + }, + Catalog: modelCatalog, + ProviderHealth: routeHealth, }).Handler() } operationHandler := operations.Handler{Store: store, Manager: manager, Billing: billingService, Metrics: metrics, diff --git a/cmd/mockupstream/main.go b/cmd/mockupstream/main.go index cf695bb..7dd5bbf 100644 --- a/cmd/mockupstream/main.go +++ b/cmd/mockupstream/main.go @@ -14,13 +14,44 @@ func main() { address := flag.String("address", "127.0.0.1:18080", "listen address") flag.Parse() mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) }) + mux.HandleFunc("GET /v1/models", models) mux.HandleFunc("POST /v1/chat/completions", openAI) + mux.HandleFunc("POST /v1/embeddings", embeddings) mux.HandleFunc("POST /v1/messages", anthropic) server := &http.Server{Addr: *address, Handler: mux, ReadHeaderTimeout: 5 * time.Second} log.Printf("mock upstream listening on http://%s", *address) log.Fatal(server.ListenAndServe()) } +func models(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "object": "list", + "data": []any{ + map[string]any{"id": "text-embedding-e2e", "object": "model"}, + map[string]any{"id": "chat-e2e", "object": "model"}, + }, + }) +} + +func embeddings(w http.ResponseWriter, r *http.Request) { + var request struct { + Model string `json:"model"` + Input json.RawMessage `json:"input"` + } + if json.NewDecoder(r.Body).Decode(&request) != nil || request.Model == "" || len(request.Input) == 0 { + http.Error(w, "invalid JSON", http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "object": "list", "model": request.Model, + "data": []any{map[string]any{"object": "embedding", "index": 0, "embedding": []float64{0.125, -0.25, 0.5, 0.75}}}, + "usage": map[string]int{"prompt_tokens": 6, "total_tokens": 6}, + }) +} + func openAI(w http.ResponseWriter, r *http.Request) { var request struct { Model string `json:"model"` diff --git a/cmd/stripe-preflight/main.go b/cmd/stripe-preflight/main.go new file mode 100644 index 0000000..3a0f2d8 --- /dev/null +++ b/cmd/stripe-preflight/main.go @@ -0,0 +1,34 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + "aigw/internal/billing" +) + +func main() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + result, err := billing.CheckStripePermissions(ctx, os.Getenv("AIGW_STRIPE_API_KEY")) + if encodeErr := json.NewEncoder(os.Stdout).Encode(result); encodeErr != nil { + fmt.Fprintln(os.Stderr, "encode Stripe preflight result") + os.Exit(1) + } + if err != nil { + if errors.Is(err, billing.ErrLiveStripeKey) { + fmt.Fprintln(os.Stderr, "Stripe preflight requires an rk_test_ or sk_test_ key; live keys are refused") + } else { + fmt.Fprintln(os.Stderr, "Stripe preflight failed") + } + os.Exit(1) + } + if !result.Ready { + fmt.Fprintln(os.Stderr, "Stripe restricted key is missing one or more required read permissions") + os.Exit(1) + } +} -- cgit v1.2.3