Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e280c0875c | ||
|
|
142dcc7932 | ||
|
|
e3dfe8162b | ||
|
|
69b0c107d3 | ||
|
|
e533ec2367 | ||
|
|
bfa17a3aba | ||
|
|
86c374b5c2 |
@@ -5208,6 +5208,12 @@
|
||||
"schema": {
|
||||
"$ref": "#/definitions/httpapi.ErrorEnvelope"
|
||||
}
|
||||
},
|
||||
"503": {
|
||||
"description": "Service Unavailable",
|
||||
"schema": {
|
||||
"$ref": "#/definitions/httpapi.ErrorEnvelope"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6441,6 +6441,10 @@ paths:
|
||||
description: Internal Server Error
|
||||
schema:
|
||||
$ref: '#/definitions/httpapi.ErrorEnvelope'
|
||||
"503":
|
||||
description: Service Unavailable
|
||||
schema:
|
||||
$ref: '#/definitions/httpapi.ErrorEnvelope'
|
||||
summary: 本地登录
|
||||
tags:
|
||||
- auth
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
func TestReadyReturnsPostgresUnavailableWithinTwoSeconds(t *testing.T) {
|
||||
db := newExhaustedPostgresStore(t)
|
||||
server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(io.Discard, nil))}
|
||||
requestContext, cancel := context.WithTimeout(context.Background(), 4*time.Second)
|
||||
defer cancel()
|
||||
request := httptest.NewRequest(http.MethodGet, "/readyz", nil).WithContext(requestContext)
|
||||
recorder := httptest.NewRecorder()
|
||||
|
||||
startedAt := time.Now()
|
||||
server.ready(recorder, request)
|
||||
elapsed := time.Since(startedAt)
|
||||
|
||||
assertUnavailableResponse(t, recorder, "POSTGRES_UNAVAILABLE", "postgres unavailable")
|
||||
if elapsed > 3*time.Second {
|
||||
t.Fatalf("readiness timeout took %s, want no more than 3s", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoginReturnsAuthStoreUnavailableWithinFiveSeconds(t *testing.T) {
|
||||
db := newExhaustedPostgresStore(t)
|
||||
var logs bytes.Buffer
|
||||
server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(&logs, nil))}
|
||||
requestContext, cancel := context.WithTimeout(context.Background(), 7*time.Second)
|
||||
defer cancel()
|
||||
request := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"account":"timeout-test-account","password":"timeout-test-password"}`)).WithContext(requestContext)
|
||||
recorder := httptest.NewRecorder()
|
||||
|
||||
startedAt := time.Now()
|
||||
server.login(recorder, request)
|
||||
elapsed := time.Since(startedAt)
|
||||
|
||||
assertUnavailableResponse(t, recorder, "AUTH_STORE_UNAVAILABLE", "authentication service temporarily unavailable")
|
||||
if elapsed > 6*time.Second {
|
||||
t.Fatalf("login timeout took %s, want no more than 6s", elapsed)
|
||||
}
|
||||
logOutput := logs.String()
|
||||
for _, field := range []string{"postgres_pool_max_connections", "postgres_pool_acquired_connections", "postgres_pool_idle_connections", "postgres_pool_empty_acquire_count", "postgres_pool_canceled_acquire_count"} {
|
||||
if !strings.Contains(logOutput, field) {
|
||||
t.Fatalf("login failure log did not include %q: %s", field, logOutput)
|
||||
}
|
||||
}
|
||||
if strings.Contains(logOutput, "timeout-test-account") || strings.Contains(logOutput, "timeout-test-password") {
|
||||
t.Fatalf("login failure log exposed credentials: %s", logOutput)
|
||||
}
|
||||
}
|
||||
|
||||
func newExhaustedPostgresStore(t *testing.T) *store.Store {
|
||||
t.Helper()
|
||||
databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
|
||||
if databaseURL == "" {
|
||||
t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run PostgreSQL availability timeout tests")
|
||||
}
|
||||
parsed, err := url.Parse(databaseURL)
|
||||
if err != nil {
|
||||
t.Fatalf("parse test database URL: %v", err)
|
||||
}
|
||||
query := parsed.Query()
|
||||
query.Set("pool_max_conns", "1")
|
||||
query.Set("pool_min_conns", "0")
|
||||
parsed.RawQuery = query.Encode()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
db, err := store.Connect(ctx, parsed.String())
|
||||
if err != nil {
|
||||
t.Fatalf("connect timeout test store: %v", err)
|
||||
}
|
||||
t.Cleanup(db.Close)
|
||||
connection, err := db.Pool().Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("exhaust timeout test pool: %v", err)
|
||||
}
|
||||
t.Cleanup(connection.Release)
|
||||
return db
|
||||
}
|
||||
|
||||
func assertUnavailableResponse(t *testing.T, recorder *httptest.ResponseRecorder, expectedCode, expectedMessage string) {
|
||||
t.Helper()
|
||||
if recorder.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want 503; body=%s", recorder.Code, recorder.Body.String())
|
||||
}
|
||||
var envelope ErrorEnvelope
|
||||
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
|
||||
t.Fatalf("decode unavailable response: %v", err)
|
||||
}
|
||||
if envelope.Error.Code != expectedCode {
|
||||
t.Fatalf("error code = %q, want %q; body=%s", envelope.Error.Code, expectedCode, recorder.Body.String())
|
||||
}
|
||||
if envelope.Error.Message != expectedMessage {
|
||||
t.Fatalf("error message = %q, want %q; body=%s", envelope.Error.Message, expectedMessage, recorder.Body.String())
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,14 @@ import (
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
const (
|
||||
postgresReadinessTimeout = 2 * time.Second
|
||||
localLoginStoreTimeout = 5 * time.Second
|
||||
errorCodePostgresDown = "POSTGRES_UNAVAILABLE"
|
||||
errorCodeAuthStoreDown = "AUTH_STORE_UNAVAILABLE"
|
||||
authStoreUnavailableError = "authentication service temporarily unavailable"
|
||||
)
|
||||
|
||||
// health godoc
|
||||
// @Summary 健康检查
|
||||
// @Description 返回服务进程、运行环境和身份模式,供负载均衡或人工排障使用。
|
||||
@@ -42,8 +50,11 @@ func (s *Server) health(w http.ResponseWriter, r *http.Request) {
|
||||
// @Failure 503 {object} ErrorEnvelope
|
||||
// @Router /readyz [get]
|
||||
func (s *Server) ready(w http.ResponseWriter, r *http.Request) {
|
||||
if err := s.store.Ping(r.Context()); err != nil {
|
||||
writeError(w, http.StatusServiceUnavailable, "postgres unavailable")
|
||||
ctx, cancel := context.WithTimeout(r.Context(), postgresReadinessTimeout)
|
||||
defer cancel()
|
||||
if err := s.store.Ping(ctx); err != nil {
|
||||
s.logPostgresUnavailable("postgres readiness check failed")
|
||||
writeError(w, http.StatusServiceUnavailable, "postgres unavailable", errorCodePostgresDown)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||||
@@ -121,6 +132,7 @@ func (s *Server) register(w http.ResponseWriter, r *http.Request) {
|
||||
// @Failure 401 {object} ErrorEnvelope
|
||||
// @Failure 403 {object} ErrorEnvelope
|
||||
// @Failure 500 {object} ErrorEnvelope
|
||||
// @Failure 503 {object} ErrorEnvelope
|
||||
// @Router /api/v1/auth/login [post]
|
||||
func (s *Server) login(w http.ResponseWriter, r *http.Request) {
|
||||
var input store.LocalLoginInput
|
||||
@@ -128,12 +140,19 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusBadRequest, "invalid json body")
|
||||
return
|
||||
}
|
||||
user, err := s.store.AuthenticateLocalUser(r.Context(), input)
|
||||
ctx, cancel := context.WithTimeout(r.Context(), localLoginStoreTimeout)
|
||||
defer cancel()
|
||||
user, err := s.store.AuthenticateLocalUser(ctx, input)
|
||||
if err != nil {
|
||||
if errors.Is(err, store.ErrInvalidCredentials) {
|
||||
writeError(w, http.StatusUnauthorized, "invalid account or password")
|
||||
return
|
||||
}
|
||||
if store.IsPostgresUnavailable(err) {
|
||||
s.logPostgresUnavailable("login authentication store unavailable")
|
||||
writeError(w, http.StatusServiceUnavailable, authStoreUnavailableError, errorCodeAuthStoreDown)
|
||||
return
|
||||
}
|
||||
s.logger.Error("login local user failed", "error", err)
|
||||
writeError(w, http.StatusInternalServerError, "login failed")
|
||||
return
|
||||
@@ -145,6 +164,22 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) {
|
||||
s.writeAuthResponse(w, http.StatusOK, user)
|
||||
}
|
||||
|
||||
func (s *Server) logPostgresUnavailable(message string) {
|
||||
if s.logger == nil || s.store == nil || s.store.Pool() == nil {
|
||||
return
|
||||
}
|
||||
statistics := s.store.Pool().Stat()
|
||||
s.logger.Error(message,
|
||||
"error_category", "postgres_unavailable",
|
||||
"postgres_pool_max_connections", statistics.MaxConns(),
|
||||
"postgres_pool_total_connections", statistics.TotalConns(),
|
||||
"postgres_pool_acquired_connections", statistics.AcquiredConns(),
|
||||
"postgres_pool_idle_connections", statistics.IdleConns(),
|
||||
"postgres_pool_empty_acquire_count", statistics.EmptyAcquireCount(),
|
||||
"postgres_pool_canceled_acquire_count", statistics.CanceledAcquireCount(),
|
||||
)
|
||||
}
|
||||
|
||||
func (s *Server) localIdentityEnabled() bool {
|
||||
mode := strings.ToLower(strings.TrimSpace(s.cfg.IdentityMode))
|
||||
return mode == "" || mode == "standalone" || mode == "hybrid"
|
||||
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
const videoBillingUnitSeconds = 5
|
||||
|
||||
type EstimateResult struct {
|
||||
Items []any `json:"items"`
|
||||
Resolver string `json:"resolver"`
|
||||
@@ -135,7 +137,7 @@ func (s *Service) billings(ctx context.Context, user *auth.User, kind string, bo
|
||||
baseKey = "videoBase"
|
||||
duration, durationSource := billingDurationSeconds(body, response)
|
||||
audioEnabled, audioSource := billingAudioEnabled(body, response)
|
||||
durationUnits := math.Max(1, math.Ceil(duration/5))
|
||||
durationUnits := videoDurationUnits(duration)
|
||||
amount := float64(count) *
|
||||
durationUnits *
|
||||
resourcePrice(config, resource, baseKey, "basePrice") *
|
||||
@@ -144,7 +146,7 @@ func (s *Service) billings(ctx context.Context, user *auth.User, kind string, bo
|
||||
resourceWeight(config, resource, "referenceVideoWeights", boolWeightKey(requestHasReferenceVideo(body))) *
|
||||
resourceWeight(config, resource, "voiceSpecifiedWeights", boolWeightKey(requestHasVoiceID(body, audioEnabled))) *
|
||||
discount
|
||||
return []any{billingLineWithDetails(candidate, resource, unit, count*int(durationUnits), roundPrice(amount), discount, simulated, map[string]any{
|
||||
return []any{billingLineWithDetails(candidate, resource, unit, videoDurationQuantity(duration, count), roundPrice(amount), discount, simulated, map[string]any{
|
||||
"count": count,
|
||||
"audio": audioEnabled,
|
||||
"audioSource": audioSource,
|
||||
@@ -416,6 +418,16 @@ func weightValueAliases(key string, name string) []string {
|
||||
}
|
||||
}
|
||||
|
||||
func videoDurationUnits(durationSeconds float64) float64 {
|
||||
return videoDurationQuantity(durationSeconds, 1)
|
||||
}
|
||||
|
||||
func videoDurationQuantity(durationSeconds float64, count int) float64 {
|
||||
const durationPrecision = 1_000_000_000
|
||||
scaledDuration := math.Round(durationSeconds * durationPrecision)
|
||||
return scaledDuration * float64(count) / (durationPrecision * videoBillingUnitSeconds)
|
||||
}
|
||||
|
||||
func requestOutputCount(body map[string]any) int {
|
||||
for _, key := range []string{"n", "count", "batch_size", "batchSize"} {
|
||||
if value := int(math.Ceil(floatFromAny(body[key]))); value > 0 {
|
||||
@@ -476,11 +488,7 @@ func generatedVideoDurationSeconds(result map[string]any) (float64, bool) {
|
||||
if duration <= 0 {
|
||||
continue
|
||||
}
|
||||
rounded := math.Round(duration)
|
||||
if rounded <= 0 {
|
||||
rounded = 1
|
||||
}
|
||||
return rounded, true
|
||||
return duration, true
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
@@ -38,7 +38,7 @@ func TestImageBillingEstimateUsesCountResolutionAndQuality(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestVideoBillingEstimateUsesFiveSecondUnitsAndDynamicWeights(t *testing.T) {
|
||||
func TestVideoBillingEstimateProratesFiveSecondUnitsAndDynamicWeights(t *testing.T) {
|
||||
service := &Service{}
|
||||
candidate := store.RuntimeModelCandidate{
|
||||
ModelName: "video-model",
|
||||
@@ -67,13 +67,13 @@ func TestVideoBillingEstimateUsesFiveSecondUnitsAndDynamicWeights(t *testing.T)
|
||||
}, candidate, clients.Response{}, true)
|
||||
|
||||
line := firstBillingLine(t, items)
|
||||
if got, want := floatFromAny(line["amount"]), 1620.0; got != want {
|
||||
if got, want := floatFromAny(line["amount"]), 1296.0; got != want {
|
||||
t.Fatalf("video estimated amount = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := floatFromAny(line["durationUnitCount"]), 3.0; got != want {
|
||||
if got, want := floatFromAny(line["durationUnitCount"]), 2.4; got != want {
|
||||
t.Fatalf("video duration units = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := line["quantity"], 3; got != want {
|
||||
if got, want := floatFromAny(line["quantity"]), 2.4; got != want {
|
||||
t.Fatalf("video quantity = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := line["durationSource"], "preprocessed_request"; got != want {
|
||||
@@ -172,13 +172,13 @@ func TestVideoBillingPrefersGeneratedDuration(t *testing.T) {
|
||||
}, false)
|
||||
|
||||
line := firstBillingLine(t, items)
|
||||
if got, want := floatFromAny(line["durationSeconds"]), 7.0; got != want {
|
||||
if got, want := floatFromAny(line["durationSeconds"]), 6.6; got != want {
|
||||
t.Fatalf("video generated duration = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := floatFromAny(line["durationUnitCount"]), 2.0; got != want {
|
||||
if got, want := floatFromAny(line["durationUnitCount"]), 1.32; got != want {
|
||||
t.Fatalf("video generated duration units = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := floatFromAny(line["amount"]), 200.0; got != want {
|
||||
if got, want := floatFromAny(line["amount"]), 132.0; got != want {
|
||||
t.Fatalf("video generated duration amount = %v, want %v", got, want)
|
||||
}
|
||||
if got, want := line["durationSource"], "generated_video"; got != want {
|
||||
|
||||
@@ -184,6 +184,22 @@ func multiplyFixedAmountRatio(amount fixedAmount, numerator int, denominator int
|
||||
return fixedAmountFromBigInt(roundBigIntRatio(product, big.NewInt(int64(denominator))))
|
||||
}
|
||||
|
||||
func multiplyFixedProductRatio(base fixedAmount, integerFactors []int, fixedFactors []fixedAmount, denominator int) (fixedAmount, error) {
|
||||
if denominator == 0 {
|
||||
return 0, fmt.Errorf("division by zero")
|
||||
}
|
||||
product := big.NewInt(int64(base))
|
||||
for _, factor := range integerFactors {
|
||||
product.Mul(product, big.NewInt(int64(factor)))
|
||||
}
|
||||
divisor := big.NewInt(int64(denominator))
|
||||
for _, factor := range fixedFactors {
|
||||
product.Mul(product, big.NewInt(int64(factor)))
|
||||
divisor.Mul(divisor, big.NewInt(fixedScale))
|
||||
}
|
||||
return fixedAmountFromBigInt(roundBigIntRatio(product, divisor))
|
||||
}
|
||||
|
||||
func fixedAmountFromBigInt(value *big.Int) (fixedAmount, error) {
|
||||
if value == nil || !value.IsInt64() {
|
||||
return 0, errFixedAmountOverflow
|
||||
@@ -649,7 +665,11 @@ func (s *Service) billingsWithResolvedPricingV2(
|
||||
baseKey = "videoBase"
|
||||
duration, durationSource := billingDurationSeconds(body, response)
|
||||
audioEnabled, audioSource := billingAudioEnabled(body, response)
|
||||
durationUnits := int(math.Max(1, math.Ceil(duration/5)))
|
||||
durationUnits := videoDurationUnits(duration)
|
||||
durationFixed, durationErr := fixedAmountFromAny(duration)
|
||||
if durationErr != nil {
|
||||
return nil, 0, resolvedPricing{}, pricing.calculationError(resource, durationErr)
|
||||
}
|
||||
price, priceErr := pricing.requiredPrice(resource, baseKey, "basePrice")
|
||||
if priceErr != nil {
|
||||
return nil, 0, resolvedPricing{}, priceErr
|
||||
@@ -670,11 +690,16 @@ func (s *Service) billingsWithResolvedPricingV2(
|
||||
if weightErr != nil {
|
||||
return nil, 0, resolvedPricing{}, weightErr
|
||||
}
|
||||
amount, calculationErr := pricing.calculate(resource, price, []int{count, durationUnits}, resolutionWeight, audioWeight, referenceVideoWeight, voiceWeight, discount)
|
||||
amount, calculationErr := multiplyFixedProductRatio(
|
||||
price,
|
||||
[]int{count},
|
||||
[]fixedAmount{durationFixed, resolutionWeight, audioWeight, referenceVideoWeight, voiceWeight, discount},
|
||||
videoBillingUnitSeconds,
|
||||
)
|
||||
if calculationErr != nil {
|
||||
return nil, 0, resolvedPricing{}, calculationErr
|
||||
return nil, 0, resolvedPricing{}, pricing.calculationError(resource, calculationErr)
|
||||
}
|
||||
item := buildLine(resource, unit, count*durationUnits, amount, map[string]any{
|
||||
item := buildLine(resource, unit, videoDurationQuantity(duration, count), amount, map[string]any{
|
||||
"count": count, "audio": audioEnabled, "audioSource": audioSource,
|
||||
"durationSeconds": duration, "durationSource": durationSource,
|
||||
"durationUnit": "5s", "durationUnitCount": durationUnits,
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
package runner
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"testing"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
@@ -39,6 +41,9 @@ func TestFixedAmountOperationsRejectOverflow(t *testing.T) {
|
||||
if _, err := pricing.calculate("image", maximum, []int{2}); !isPricingUnavailable(err) {
|
||||
t.Fatalf("pricing overflow should be unavailable: %v", err)
|
||||
}
|
||||
if _, err := multiplyFixedProductRatio(maximum, []int{2}, nil, 1); !errors.Is(err, errFixedAmountOverflow) {
|
||||
t.Fatalf("product ratio overflow error=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEstimatedOutputTokensUsesAliasesAndCapabilityFallback(t *testing.T) {
|
||||
@@ -138,6 +143,108 @@ func TestPricingWeightsUseFixedPrecisionAndRejectInvalidValues(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestVideoBillingV2ProratesFiveSecondPriceByActualDuration(t *testing.T) {
|
||||
service := &Service{}
|
||||
candidate := store.RuntimeModelCandidate{ModelName: "video-model"}
|
||||
pricing := resolvedPricing{
|
||||
Config: map[string]any{
|
||||
"video": map[string]any{
|
||||
"basePrice": 100,
|
||||
"dynamicWeight": map[string]any{
|
||||
"audioWeights": map[string]any{"true": 2},
|
||||
},
|
||||
},
|
||||
},
|
||||
Currency: "resource",
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
duration float64
|
||||
wantUnits float64
|
||||
wantAmount float64
|
||||
}{
|
||||
{name: "three seconds uses zero point six units", duration: 3, wantUnits: 0.6, wantAmount: 120},
|
||||
{name: "five seconds uses one unit", duration: 5, wantUnits: 1, wantAmount: 200},
|
||||
{name: "six seconds uses one point two units", duration: 6, wantUnits: 1.2, wantAmount: 240},
|
||||
{name: "fractional seconds retain fixed amount precision", duration: 6.000000001, wantUnits: 1.2000000002, wantAmount: 240.00000004},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
items, total, _, err := service.billingsWithResolvedPricingV2(
|
||||
context.Background(), nil, "videos.generations",
|
||||
map[string]any{"duration": test.duration, "audio": true},
|
||||
candidate, clients.Response{}, true, pricing,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("bill video: %v", err)
|
||||
}
|
||||
line := firstBillingLine(t, items)
|
||||
if got := total.Float64(); math.Abs(got-test.wantAmount) > 1e-9 {
|
||||
t.Fatalf("total amount=%v, want %v", got, test.wantAmount)
|
||||
}
|
||||
if got := floatFromAny(line["amount"]); math.Abs(got-test.wantAmount) > 1e-9 {
|
||||
t.Fatalf("line amount=%v, want %v", got, test.wantAmount)
|
||||
}
|
||||
if got := floatFromAny(line["quantity"]); math.Abs(got-test.wantUnits) > 1e-12 {
|
||||
t.Fatalf("quantity=%v, want %v", got, test.wantUnits)
|
||||
}
|
||||
if got := floatFromAny(line["durationUnitCount"]); math.Abs(got-test.wantUnits) > 1e-12 {
|
||||
t.Fatalf("duration units=%v, want %v", got, test.wantUnits)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestVideoBillingV2RoundsOnlyAfterApplyingDurationCountAndWeights(t *testing.T) {
|
||||
service := &Service{}
|
||||
candidate := store.RuntimeModelCandidate{ModelName: "video-model"}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
body map[string]any
|
||||
pricing resolvedPricing
|
||||
wantAmount string
|
||||
}{
|
||||
{
|
||||
name: "count preserves a sub-nano duration share",
|
||||
body: map[string]any{"duration": 1, "count": 5},
|
||||
pricing: resolvedPricing{Config: map[string]any{
|
||||
"video": map[string]any{"basePrice": "0.000000001"},
|
||||
}},
|
||||
wantAmount: "0.000000001",
|
||||
},
|
||||
{
|
||||
name: "weight does not amplify a rounded duration share",
|
||||
body: map[string]any{"duration": 3, "audio": true},
|
||||
pricing: resolvedPricing{Config: map[string]any{
|
||||
"video": map[string]any{
|
||||
"basePrice": "0.000000001",
|
||||
"dynamicWeight": map[string]any{
|
||||
"audioWeights": map[string]any{"true": 2},
|
||||
},
|
||||
},
|
||||
}},
|
||||
wantAmount: "0.000000001",
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
_, total, _, err := service.billingsWithResolvedPricingV2(
|
||||
context.Background(), nil, "videos.generations", test.body,
|
||||
candidate, clients.Response{}, true, test.pricing,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("bill video: %v", err)
|
||||
}
|
||||
if got := total.String(); got != test.wantAmount {
|
||||
t.Fatalf("total amount=%s, want %s", got, test.wantAmount)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func mustFixedAmount(t *testing.T, value string) fixedAmount {
|
||||
t.Helper()
|
||||
amount, err := parseFixedAmount(value)
|
||||
|
||||
+4
-4
@@ -166,16 +166,16 @@ The response has this shape:
|
||||
"platformModelId": "<platform-model-id>",
|
||||
"resourceType": "video",
|
||||
"unit": "5s_video",
|
||||
"quantity": 3,
|
||||
"quantity": 2.4,
|
||||
"amount": 12.5,
|
||||
"currency": "resource",
|
||||
"discountFactor": 0.8,
|
||||
"simulated": true,
|
||||
"durationSeconds": 12,
|
||||
"durationUnitCount": 3
|
||||
"durationUnitCount": 2.4
|
||||
}
|
||||
],
|
||||
"resolver": "effective-pricing-v1",
|
||||
"resolver": "effective-pricing-v2",
|
||||
"totalAmount": 12.5,
|
||||
"currency": "resource"
|
||||
}
|
||||
@@ -197,7 +197,7 @@ Useful calculation checks:
|
||||
text input = input tokens / 1000 × input price × discount
|
||||
text output = output tokens / 1000 × output price × discount
|
||||
image = count × base price × quality/size/resolution weights × discount
|
||||
video = count × ceil(duration seconds / 5) × base price × applicable weights × discount
|
||||
video = count × (duration seconds / 5) × base price × applicable weights × discount
|
||||
speech = Unicode character count × audio price × discount
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func TestVerifyLocalAPIKeyWorksWithSingleConnectionPool(t *testing.T) {
|
||||
db, verificationStore, created, user := newLocalAPIKeyVerificationFixture(t, 1)
|
||||
ctx := context.Background()
|
||||
|
||||
verifyCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
verified, err := verificationStore.VerifyLocalAPIKey(verifyCtx, created.Secret)
|
||||
if err != nil {
|
||||
t.Fatalf("verify local API key with one connection: %v", err)
|
||||
}
|
||||
if verified.APIKeyID != created.APIKey.ID || verified.GatewayUserID != user.ID {
|
||||
t.Fatalf("verified identity = %+v, want API key %q and user %q", verified, created.APIKey.ID, user.ID)
|
||||
}
|
||||
|
||||
var lastUsedAt *time.Time
|
||||
if err := db.pool.QueryRow(ctx, `SELECT last_used_at FROM gateway_api_keys WHERE id=$1::uuid`, created.APIKey.ID).Scan(&lastUsedAt); err != nil {
|
||||
t.Fatalf("read API key last_used_at: %v", err)
|
||||
}
|
||||
if lastUsedAt == nil {
|
||||
t.Fatal("successful API key verification did not update last_used_at")
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyLocalAPIKeyHandlesEightConcurrentRequestsWithFourConnections(t *testing.T) {
|
||||
_, verificationStore, created, _ := newLocalAPIKeyVerificationFixture(t, 4)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
const requestCount = 8
|
||||
start := make(chan struct{})
|
||||
errorsByRequest := make(chan error, requestCount)
|
||||
var requests sync.WaitGroup
|
||||
requests.Add(requestCount)
|
||||
for range requestCount {
|
||||
go func() {
|
||||
defer requests.Done()
|
||||
<-start
|
||||
_, err := verificationStore.VerifyLocalAPIKey(ctx, created.Secret)
|
||||
errorsByRequest <- err
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
requests.Wait()
|
||||
close(errorsByRequest)
|
||||
|
||||
for err := range errorsByRequest {
|
||||
if err != nil {
|
||||
t.Fatalf("concurrent API key verification failed: %v", err)
|
||||
}
|
||||
}
|
||||
if acquired := verificationStore.pool.Stat().AcquiredConns(); acquired != 0 {
|
||||
t.Fatalf("API key verification left %d connections acquired", acquired)
|
||||
}
|
||||
}
|
||||
|
||||
func newLocalAPIKeyVerificationFixture(t *testing.T, maxConnections int32) (*Store, *Store, CreatedAPIKey, GatewayUser) {
|
||||
t.Helper()
|
||||
db := newIdentityPairingPostgresTestStore(t)
|
||||
ctx := context.Background()
|
||||
user, err := db.RegisterLocalUser(ctx, LocalRegisterInput{
|
||||
Username: "api-key-verification-user",
|
||||
Password: "api-key-verification-password",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("register API key verification user: %v", err)
|
||||
}
|
||||
created, err := db.CreateAPIKey(ctx, CreateAPIKeyInput{Name: "API key verification fixture"}, &auth.User{
|
||||
ID: user.ID,
|
||||
GatewayUserID: user.ID,
|
||||
GatewayTenantID: user.GatewayTenantID,
|
||||
TenantID: user.TenantID,
|
||||
TenantKey: user.TenantKey,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create API key verification fixture: %v", err)
|
||||
}
|
||||
|
||||
config := db.pool.Config()
|
||||
config.MaxConns = maxConnections
|
||||
config.MinConns = 0
|
||||
pool, err := pgxpool.NewWithConfig(ctx, config)
|
||||
if err != nil {
|
||||
t.Fatalf("create verification pool with %d connections: %v", maxConnections, err)
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
return db, &Store{pool: pool}, created, user
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
func TestVerifyLocalAPIKeyClosesCandidateRowsBeforeUpdatingUsage(t *testing.T) {
|
||||
secret := "sk-gw-matching-secret"
|
||||
wrongHash, err := bcrypt.GenerateFromPassword([]byte("sk-gw-different-secret"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatalf("hash non-matching API key: %v", err)
|
||||
}
|
||||
matchingHash, err := bcrypt.GenerateFromPassword([]byte(secret), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatalf("hash matching API key: %v", err)
|
||||
}
|
||||
rows := &fakeLocalAPIKeyRows{candidates: []localAPIKeyCandidate{
|
||||
{apiKeyID: "wrong-key", hash: string(wrongHash), keyPrefix: apiKeyPrefix(secret)},
|
||||
{
|
||||
apiKeyID: "matching-key",
|
||||
hash: string(matchingHash),
|
||||
keyPrefix: apiKeyPrefix(secret),
|
||||
keyName: "Matching key",
|
||||
scopesBytes: []byte(`["chat"]`),
|
||||
userGroupID: "group-id",
|
||||
gatewayUserID: "user-id",
|
||||
username: "api-key-user",
|
||||
rolesBytes: []byte(`["user"]`),
|
||||
gatewayTenantID: "gateway-tenant-id",
|
||||
tenantID: "tenant-id",
|
||||
tenantKey: "tenant-key",
|
||||
},
|
||||
}}
|
||||
database := &fakeLocalAPIKeyDatabase{rows: rows}
|
||||
|
||||
user, err := verifyLocalAPIKey(context.Background(), database, secret)
|
||||
if err != nil {
|
||||
t.Fatalf("verify local API key: %v", err)
|
||||
}
|
||||
if !rows.closed {
|
||||
t.Fatal("candidate rows remained open after API key verification")
|
||||
}
|
||||
if database.updatedAPIKeyID != "matching-key" {
|
||||
t.Fatalf("updated API key = %q, want matching-key", database.updatedAPIKeyID)
|
||||
}
|
||||
if user.APIKeyID != "matching-key" || user.GatewayUserID != "user-id" {
|
||||
t.Fatalf("verified user = %+v", user)
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyLocalAPIKeyReturnsUnauthorizedAfterClosingCandidateRows(t *testing.T) {
|
||||
secret := "sk-gw-unknown-secret"
|
||||
wrongHash, err := bcrypt.GenerateFromPassword([]byte("sk-gw-different-secret"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatalf("hash non-matching API key: %v", err)
|
||||
}
|
||||
rows := &fakeLocalAPIKeyRows{candidates: []localAPIKeyCandidate{{
|
||||
apiKeyID: "wrong-key",
|
||||
hash: string(wrongHash),
|
||||
}}}
|
||||
database := &fakeLocalAPIKeyDatabase{rows: rows}
|
||||
|
||||
_, err = verifyLocalAPIKey(context.Background(), database, secret)
|
||||
if !errors.Is(err, auth.ErrUnauthorized) {
|
||||
t.Fatalf("verify error = %v, want unauthorized", err)
|
||||
}
|
||||
if !rows.closed {
|
||||
t.Fatal("candidate rows remained open after unsuccessful API key verification")
|
||||
}
|
||||
if database.updatedAPIKeyID != "" {
|
||||
t.Fatalf("unexpected API key usage update for %q", database.updatedAPIKeyID)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeLocalAPIKeyDatabase struct {
|
||||
rows *fakeLocalAPIKeyRows
|
||||
updatedAPIKeyID string
|
||||
}
|
||||
|
||||
func (database *fakeLocalAPIKeyDatabase) Query(context.Context, string, ...any) (pgx.Rows, error) {
|
||||
return database.rows, nil
|
||||
}
|
||||
|
||||
func (database *fakeLocalAPIKeyDatabase) Exec(_ context.Context, _ string, arguments ...any) (pgconn.CommandTag, error) {
|
||||
if !database.rows.closed {
|
||||
return pgconn.CommandTag{}, errors.New("API key usage update started before candidate rows closed")
|
||||
}
|
||||
database.updatedAPIKeyID, _ = arguments[0].(string)
|
||||
return pgconn.NewCommandTag("UPDATE 1"), nil
|
||||
}
|
||||
|
||||
type fakeLocalAPIKeyRows struct {
|
||||
candidates []localAPIKeyCandidate
|
||||
current int
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Close() {
|
||||
rows.closed = true
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Err() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) CommandTag() pgconn.CommandTag {
|
||||
return pgconn.CommandTag{}
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) FieldDescriptions() []pgconn.FieldDescription {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Next() bool {
|
||||
if rows.current >= len(rows.candidates) {
|
||||
rows.Close()
|
||||
return false
|
||||
}
|
||||
rows.current++
|
||||
return true
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Scan(destinations ...any) error {
|
||||
candidate := rows.candidates[rows.current-1]
|
||||
values := []any{
|
||||
candidate.apiKeyID,
|
||||
candidate.hash,
|
||||
candidate.keyPrefix,
|
||||
candidate.keyName,
|
||||
candidate.scopesBytes,
|
||||
candidate.userGroupID,
|
||||
candidate.gatewayUserID,
|
||||
candidate.username,
|
||||
candidate.rolesBytes,
|
||||
candidate.gatewayTenantID,
|
||||
candidate.tenantID,
|
||||
candidate.tenantKey,
|
||||
}
|
||||
for index, value := range values {
|
||||
switch destination := destinations[index].(type) {
|
||||
case *string:
|
||||
*destination = value.(string)
|
||||
case *[]byte:
|
||||
*destination = value.([]byte)
|
||||
default:
|
||||
return errors.New("unsupported fake row destination")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Values() ([]any, error) {
|
||||
return nil, errors.New("not implemented")
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) RawValues() [][]byte {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rows *fakeLocalAPIKeyRows) Conn() *pgx.Conn {
|
||||
return nil
|
||||
}
|
||||
@@ -685,11 +685,13 @@ func newIdentityPairingPostgresTestStore(t *testing.T) *Store {
|
||||
migrationDirectory := filepath.Join(filepath.Dir(filename), "..", "..", "migrations")
|
||||
for _, migrationName := range []string{
|
||||
"0001_init.sql",
|
||||
"0017_task_record_enrichment.sql",
|
||||
"0061_oidc_server_sessions.sql",
|
||||
"0065_identity_configuration_revisions.sql",
|
||||
"0066_identity_onboarding_exchanges.sql",
|
||||
"0067_identity_secret_cleanup_queue.sql",
|
||||
"0068_identity_pairing_start_reservation.sql",
|
||||
"0069_billing_correctness_v2.sql",
|
||||
} {
|
||||
migration, err := os.ReadFile(filepath.Join(migrationDirectory, migrationName))
|
||||
if err != nil {
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode"
|
||||
@@ -22,6 +23,11 @@ type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
const (
|
||||
postgresApplicationName = "easyai-ai-gateway"
|
||||
postgresConnectTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
func defaultAPIKeyScopes() []string {
|
||||
return []string{"chat", "embedding", "rerank", "image", "video", "music", "audio", "voice_clone"}
|
||||
}
|
||||
@@ -67,7 +73,11 @@ var (
|
||||
)
|
||||
|
||||
func Connect(ctx context.Context, databaseURL string) (*Store, error) {
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
config, err := postgresPoolConfig(databaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pool, err := pgxpool.NewWithConfig(ctx, config)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -78,6 +88,38 @@ func Connect(ctx context.Context, databaseURL string) (*Store, error) {
|
||||
return &Store{pool: pool}, nil
|
||||
}
|
||||
|
||||
func postgresPoolConfig(databaseURL string) (*pgxpool.Config, error) {
|
||||
config, err := pgxpool.ParseConfig(databaseURL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
config.ConnConfig.ConnectTimeout = postgresConnectTimeout
|
||||
config.ConnConfig.RuntimeParams["application_name"] = postgresApplicationName
|
||||
return config, nil
|
||||
}
|
||||
|
||||
func IsPostgresUnavailable(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
return true
|
||||
}
|
||||
var connectError *pgconn.ConnectError
|
||||
if errors.As(err, &connectError) {
|
||||
return true
|
||||
}
|
||||
var networkError net.Error
|
||||
if errors.As(err, &networkError) {
|
||||
return true
|
||||
}
|
||||
var postgresError *pgconn.PgError
|
||||
if errors.As(err, &postgresError) {
|
||||
return strings.HasPrefix(postgresError.Code, "08") || postgresError.Code == "53300" || strings.HasPrefix(postgresError.Code, "57P0")
|
||||
}
|
||||
return pgconn.SafeToRetry(err)
|
||||
}
|
||||
|
||||
func (s *Store) Close() {
|
||||
s.pool.Close()
|
||||
}
|
||||
@@ -1519,11 +1561,35 @@ WHERE subject_type = 'api_key' AND subject_id = $1::uuid`, apiKeyID); err != nil
|
||||
}
|
||||
|
||||
func (s *Store) VerifyLocalAPIKey(ctx context.Context, secret string) (*auth.User, error) {
|
||||
return verifyLocalAPIKey(ctx, s.pool, secret)
|
||||
}
|
||||
|
||||
type localAPIKeyDatabase interface {
|
||||
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
|
||||
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
|
||||
}
|
||||
|
||||
type localAPIKeyCandidate struct {
|
||||
apiKeyID string
|
||||
hash string
|
||||
keyPrefix string
|
||||
keyName string
|
||||
scopesBytes []byte
|
||||
userGroupID string
|
||||
gatewayUserID string
|
||||
username string
|
||||
rolesBytes []byte
|
||||
gatewayTenantID string
|
||||
tenantID string
|
||||
tenantKey string
|
||||
}
|
||||
|
||||
func verifyLocalAPIKey(ctx context.Context, database localAPIKeyDatabase, secret string) (*auth.User, error) {
|
||||
prefix := apiKeyPrefix(secret)
|
||||
if prefix == "" {
|
||||
return nil, auth.ErrUnauthorized
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
rows, err := database.Query(ctx, `
|
||||
SELECT k.id::text, k.key_hash, k.key_prefix, k.name, k.scopes, COALESCE(k.user_group_id::text, u.default_user_group_id::text, ''),
|
||||
u.id::text, u.username, u.roles, COALESCE(u.gateway_tenant_id::text, ''),
|
||||
COALESCE(u.tenant_id, ''), COALESCE(u.tenant_key, '')
|
||||
@@ -1538,49 +1604,51 @@ WHERE k.key_prefix = $1
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
candidates, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (localAPIKeyCandidate, error) {
|
||||
var candidate localAPIKeyCandidate
|
||||
err := row.Scan(
|
||||
&candidate.apiKeyID,
|
||||
&candidate.hash,
|
||||
&candidate.keyPrefix,
|
||||
&candidate.keyName,
|
||||
&candidate.scopesBytes,
|
||||
&candidate.userGroupID,
|
||||
&candidate.gatewayUserID,
|
||||
&candidate.username,
|
||||
&candidate.rolesBytes,
|
||||
&candidate.gatewayTenantID,
|
||||
&candidate.tenantID,
|
||||
&candidate.tenantKey,
|
||||
)
|
||||
return candidate, err
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for rows.Next() {
|
||||
var apiKeyID string
|
||||
var hash string
|
||||
var keyPrefix string
|
||||
var keyName string
|
||||
var scopesBytes []byte
|
||||
var userGroupID string
|
||||
var gatewayUserID string
|
||||
var username string
|
||||
var rolesBytes []byte
|
||||
var gatewayTenantID string
|
||||
var tenantID string
|
||||
var tenantKey string
|
||||
if err := rows.Scan(&apiKeyID, &hash, &keyPrefix, &keyName, &scopesBytes, &userGroupID, &gatewayUserID, &username, &rolesBytes, &gatewayTenantID, &tenantID, &tenantKey); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(secret)) != nil {
|
||||
for _, candidate := range candidates {
|
||||
if bcrypt.CompareHashAndPassword([]byte(candidate.hash), []byte(secret)) != nil {
|
||||
continue
|
||||
}
|
||||
if _, err := s.pool.Exec(ctx, `UPDATE gateway_api_keys SET last_used_at = now(), updated_at = now() WHERE id = $1::uuid`, apiKeyID); err != nil {
|
||||
if _, err := database.Exec(ctx, `UPDATE gateway_api_keys SET last_used_at = now(), updated_at = now() WHERE id = $1::uuid`, candidate.apiKeyID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &auth.User{
|
||||
ID: gatewayUserID,
|
||||
Username: username,
|
||||
Roles: decodeStringArray(rolesBytes),
|
||||
TenantID: tenantID,
|
||||
GatewayTenantID: gatewayTenantID,
|
||||
TenantKey: tenantKey,
|
||||
ID: candidate.gatewayUserID,
|
||||
Username: candidate.username,
|
||||
Roles: decodeStringArray(candidate.rolesBytes),
|
||||
TenantID: candidate.tenantID,
|
||||
GatewayTenantID: candidate.gatewayTenantID,
|
||||
TenantKey: candidate.tenantKey,
|
||||
Source: "gateway",
|
||||
GatewayUserID: gatewayUserID,
|
||||
UserGroupID: userGroupID,
|
||||
APIKeyID: apiKeyID,
|
||||
APIKeyName: keyName,
|
||||
APIKeyPrefix: keyPrefix,
|
||||
APIKeyScopes: decodeStringArray(scopesBytes),
|
||||
GatewayUserID: candidate.gatewayUserID,
|
||||
UserGroupID: candidate.userGroupID,
|
||||
APIKeyID: candidate.apiKeyID,
|
||||
APIKeyName: candidate.keyName,
|
||||
APIKeyPrefix: candidate.keyPrefix,
|
||||
APIKeyScopes: decodeStringArray(candidate.scopesBytes),
|
||||
}, nil
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, auth.ErrUnauthorized
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
func TestPostgresPoolConfigSetsDiagnosticAndConnectTimeout(t *testing.T) {
|
||||
config, err := postgresPoolConfig("postgresql://gateway:password@localhost:5432/gateway?sslmode=disable")
|
||||
if err != nil {
|
||||
t.Fatalf("parse PostgreSQL pool config: %v", err)
|
||||
}
|
||||
if config.ConnConfig.ConnectTimeout != 5*time.Second {
|
||||
t.Fatalf("connect timeout = %s, want 5s", config.ConnConfig.ConnectTimeout)
|
||||
}
|
||||
if applicationName := config.ConnConfig.RuntimeParams["application_name"]; applicationName != "easyai-ai-gateway" {
|
||||
t.Fatalf("application_name = %q, want easyai-ai-gateway", applicationName)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgresPoolConfigRejectsMalformedURL(t *testing.T) {
|
||||
if _, err := postgresPoolConfig("://malformed"); err == nil {
|
||||
t.Fatal("expected malformed PostgreSQL URL to fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsPostgresUnavailableClassifiesConnectivityFailures(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
err error
|
||||
}{
|
||||
{name: "deadline", err: context.DeadlineExceeded},
|
||||
{name: "connection exception", err: &pgconn.PgError{Code: "08006"}},
|
||||
{name: "too many connections", err: &pgconn.PgError{Code: "53300"}},
|
||||
{name: "cannot connect now", err: &pgconn.PgError{Code: "57P03"}},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
if !IsPostgresUnavailable(testCase.err) {
|
||||
t.Fatalf("error %v was not classified as PostgreSQL unavailable", testCase.err)
|
||||
}
|
||||
})
|
||||
}
|
||||
if IsPostgresUnavailable(&pgconn.PgError{Code: "42601"}) {
|
||||
t.Fatal("SQL syntax error was incorrectly classified as PostgreSQL unavailable")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
UPDATE model_pricing_rules
|
||||
SET formula_config = jsonb_set(
|
||||
COALESCE(formula_config, '{}'::jsonb),
|
||||
'{formula}',
|
||||
to_jsonb(replace(
|
||||
formula_config->>'formula',
|
||||
'ceil(duration_seconds / 5)',
|
||||
'(duration_seconds / 5)'
|
||||
)),
|
||||
true
|
||||
),
|
||||
updated_at = now()
|
||||
WHERE resource_type = 'video'
|
||||
AND strpos(formula_config->>'formula', 'ceil(duration_seconds / 5)') > 0;
|
||||
@@ -9,12 +9,39 @@ import {
|
||||
getAPITask,
|
||||
getCurrentUser,
|
||||
getOpsManagementSkillMetadata,
|
||||
loginLocalAccount,
|
||||
OIDC_BROWSER_SESSION_CREDENTIAL,
|
||||
startIdentityPairing,
|
||||
retireIdentityPairingSecurityEventConflict,
|
||||
validateIdentityRevision,
|
||||
} from './api';
|
||||
|
||||
describe('local login transport', () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
it('aborts after ten seconds and returns a stable login timeout message', async () => {
|
||||
vi.useFakeTimers();
|
||||
const fetchMock = vi.fn((_url: string, init?: RequestInit) => new Promise<Response>((_resolve, reject) => {
|
||||
init?.signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')));
|
||||
}));
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
|
||||
const login = loginLocalAccount({ account: 'timeout-test-account', password: 'timeout-test-password' });
|
||||
const rejection = expect(login).rejects.toMatchObject({
|
||||
message: '登录请求超时,请稍后重试',
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
await rejection;
|
||||
|
||||
const [, init] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||
expect(init.signal).toBeInstanceOf(AbortSignal);
|
||||
expect(init.signal?.aborted).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Gateway provisioning errors', () => {
|
||||
const cases = [
|
||||
['GATEWAY_USER_NOT_PROVISIONED', '该账号尚未开通 EasyAI Gateway'],
|
||||
|
||||
+30
-15
@@ -121,6 +121,7 @@ export async function loginLocalAccount(input: { account: string; password: stri
|
||||
auth: false,
|
||||
body: input,
|
||||
method: 'POST',
|
||||
timeoutMs: 10_000,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1206,7 +1207,7 @@ export async function deleteFileStorageChannel(token: string, channelId: string)
|
||||
|
||||
async function request<T>(
|
||||
path: string,
|
||||
options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record<string, string>; signal?: AbortSignal } = {},
|
||||
options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record<string, string>; signal?: AbortSignal; timeoutMs?: number } = {},
|
||||
): Promise<T> {
|
||||
const headers: Record<string, string> = { ...(options.headers ?? {}) };
|
||||
if (options.auth !== false && options.token && options.token !== OIDC_BROWSER_SESSION_CREDENTIAL) {
|
||||
@@ -1215,21 +1216,35 @@ async function request<T>(
|
||||
if (options.body !== undefined) {
|
||||
headers['Content-Type'] = 'application/json';
|
||||
}
|
||||
const response = await fetch(`${API_BASE}${path}`, {
|
||||
method: options.method ?? 'GET',
|
||||
headers,
|
||||
body: options.body === undefined ? undefined : JSON.stringify(options.body),
|
||||
credentials: 'include',
|
||||
signal: options.signal,
|
||||
});
|
||||
if (!response.ok) {
|
||||
const body = await response.text();
|
||||
throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`));
|
||||
const controller = options.timeoutMs ? new AbortController() : undefined;
|
||||
const timeout = controller ? globalThis.setTimeout(() => controller.abort(), options.timeoutMs) : undefined;
|
||||
const signal = controller && options.signal
|
||||
? AbortSignal.any([controller.signal, options.signal])
|
||||
: controller?.signal ?? options.signal;
|
||||
try {
|
||||
const response = await fetch(`${API_BASE}${path}`, {
|
||||
method: options.method ?? 'GET',
|
||||
headers,
|
||||
body: options.body === undefined ? undefined : JSON.stringify(options.body),
|
||||
credentials: 'include',
|
||||
signal,
|
||||
});
|
||||
if (!response.ok) {
|
||||
const body = await response.text();
|
||||
throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`));
|
||||
}
|
||||
if (response.status === 204) {
|
||||
return undefined as T;
|
||||
}
|
||||
return response.json() as Promise<T>;
|
||||
} catch (error) {
|
||||
if (controller?.signal.aborted) {
|
||||
throw new GatewayApiError('登录请求超时,请稍后重试');
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
if (timeout !== undefined) globalThis.clearTimeout(timeout);
|
||||
}
|
||||
if (response.status === 204) {
|
||||
return undefined as T;
|
||||
}
|
||||
return response.json() as Promise<T>;
|
||||
}
|
||||
|
||||
function authorizationHeader(token: string): Record<string, string> {
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import type { PlatformModel } from '@easyai-ai-gateway/contracts';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { filterModelsForMode } from './PlaygroundPage';
|
||||
|
||||
function model(id: string, modelType: string[]) {
|
||||
return { id, modelType } as PlatformModel;
|
||||
}
|
||||
|
||||
function modelIds(models: PlatformModel[]) {
|
||||
return models.map((item) => item.id);
|
||||
}
|
||||
|
||||
describe('playground model filtering', () => {
|
||||
const models = [
|
||||
model('image-generation', ['image_generate']),
|
||||
model('legacy-image', ['image']),
|
||||
model('image-edit', ['image_edit']),
|
||||
model('image-to-video', ['video_generate', 'image_to_video']),
|
||||
model('text-to-video', ['text_to_video']),
|
||||
];
|
||||
|
||||
it('only shows image generation models when no reference image is present', () => {
|
||||
expect(modelIds(filterModelsForMode(models, 'image', false, 'text_to_video'))).toEqual([
|
||||
'image-generation',
|
||||
'legacy-image',
|
||||
]);
|
||||
});
|
||||
|
||||
it('only shows image editing models when a reference image is present', () => {
|
||||
expect(modelIds(filterModelsForMode(models, 'image', true, 'text_to_video'))).toEqual([
|
||||
'legacy-image',
|
||||
'image-edit',
|
||||
]);
|
||||
});
|
||||
|
||||
it('does not fall back to image-to-video models when no image model is available', () => {
|
||||
const videoOnlyModels = [
|
||||
model('kling-3-turbo', ['video_generate', 'image_to_video']),
|
||||
model('kling-1-5', ['image_to_video']),
|
||||
];
|
||||
|
||||
expect(filterModelsForMode(videoOnlyModels, 'image', false, 'text_to_video')).toEqual([]);
|
||||
});
|
||||
|
||||
it('keeps image-to-video models available for the matching video mode', () => {
|
||||
expect(modelIds(filterModelsForMode(models, 'video', false, 'first_last_frame'))).toContain('image-to-video');
|
||||
});
|
||||
});
|
||||
@@ -810,7 +810,7 @@ function Composer(props: {
|
||||
<Select className="playgroundModelSelect" value={props.selectedModel ?? ''} disabled={!props.modelOptions.length} onChange={(event) => props.onModelChange(event.target.value)}>
|
||||
{props.modelOptions.length ? props.modelOptions.map((item) => (
|
||||
<option value={item.value} key={item.value}>{modelOptionLabel(item)}</option>
|
||||
)) : <option value="">模型选择</option>}
|
||||
)) : <option value="">{props.compact ? '模型选择' : '暂无可用模型'}</option>}
|
||||
</Select>
|
||||
{props.mode !== 'chat' && props.mediaSettings && props.onMediaSettingsChange && (
|
||||
<MediaSettingsPopover
|
||||
@@ -977,25 +977,25 @@ function mediaPromptPlaceholder(mode: PlaygroundMode) {
|
||||
return placeholderByMode.chat;
|
||||
}
|
||||
|
||||
function filterModelsForMode(models: PlatformModel[], mode: PlaygroundMode, hasReference: boolean, videoMode: VideoCreateMode) {
|
||||
export function filterModelsForMode(models: PlatformModel[], mode: PlaygroundMode, hasReference: boolean, videoMode: VideoCreateMode) {
|
||||
if (mode === 'chat') {
|
||||
return filterWithFallback(models, ['text_generate', 'chat', 'responses', 'text']);
|
||||
return filterModelsByType(models, ['text_generate', 'chat', 'responses', 'text']);
|
||||
}
|
||||
if (mode === 'image') {
|
||||
const preferredTypes = hasReference ? ['image_edit', 'images.edits'] : ['image_generate', 'images.generations'];
|
||||
return filterWithFallback(models, [...preferredTypes, 'image']);
|
||||
return filterModelsByType(models, [...preferredTypes, 'image']);
|
||||
}
|
||||
const videoTypesByMode: Record<VideoCreateMode, string[]> = {
|
||||
first_last_frame: ['video_first_last_frame', 'image_to_video', 'video_generate'],
|
||||
omni_reference: ['omni_video', 'video_reference', 'video_generate'],
|
||||
text_to_video: ['text_to_video', 'video_generate'],
|
||||
};
|
||||
return filterWithFallback(models, [...videoTypesByMode[videoMode], 'video']);
|
||||
return filterModelsByType(models, [...videoTypesByMode[videoMode], 'video']);
|
||||
}
|
||||
|
||||
function filterWithFallback(models: PlatformModel[], modelTypes: string[]) {
|
||||
const exact = models.filter((model) => model.modelType.some((type) => modelTypes.includes(type)));
|
||||
return exact.length ? exact : models.filter((model) => modelTypes.some((type) => model.modelType.some((modelType) => modelType.includes(type) || type.includes(modelType))));
|
||||
function filterModelsByType(models: PlatformModel[], modelTypes: string[]) {
|
||||
const acceptedTypes = new Set(modelTypes);
|
||||
return models.filter((model) => model.modelType.some((type) => acceptedTypes.has(type)));
|
||||
}
|
||||
|
||||
function buildModelOptions(models: PlatformModel[]): ModelOption[] {
|
||||
|
||||
@@ -56,7 +56,7 @@ const modeDefinitions: ModeDefinition[] = [
|
||||
formula: '扣费 = 基础单价 × 生成时长单位 × 数量 × 分辨率、音频、参考视频、音色等计费参数。',
|
||||
match: (rule) => rule.resourceType === 'video',
|
||||
templates: (currency) => [
|
||||
createRule('video', '视频', 'video', '5s', 100, currency, 'duration_weight', 'count * ceil(duration_seconds / 5) * base_price * resolution_factor * audio_factor * reference_video_factor * voice_specified_factor', {
|
||||
createRule('video', '视频', 'video', '5s', 100, currency, 'duration_weight', 'count * (duration_seconds / 5) * base_price * resolution_factor * audio_factor * reference_video_factor * voice_specified_factor', {
|
||||
resolutionWeights: { '480p': 0.75, '720p': 1, '1080p': 1.5, '2160p': 2 },
|
||||
audioWeights: { true: 2, false: 1 },
|
||||
referenceVideoWeights: { true: 1.5, false: 1 },
|
||||
|
||||
@@ -33,6 +33,20 @@ server {
|
||||
return 404;
|
||||
}
|
||||
|
||||
location = /gateway-api/api/v1/auth/login {
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Forwarded-Proto $scheme;
|
||||
proxy_set_header Connection "";
|
||||
proxy_connect_timeout 3s;
|
||||
proxy_read_timeout 15s;
|
||||
proxy_send_timeout 15s;
|
||||
proxy_redirect off;
|
||||
proxy_pass http://api:8088/api/v1/auth/login;
|
||||
}
|
||||
|
||||
location /gateway-api/ {
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Host $host;
|
||||
|
||||
+1
-1
@@ -1430,7 +1430,7 @@ effective price = rule price(request dimensions) * platform/model discount * use
|
||||
- 基础单位建议为 `5s` 或 `second`,与原 provider 配置保持可映射。
|
||||
- 动态权重至少支持:时长、分辨率、是否包含音频、是否使用参考视频、是否指定声音/音色、生成数量。
|
||||
- 分辨率示例:`480p`、`720p`、`1080p`、`2160p`。
|
||||
- 公式示例:`count * ceil(durationSeconds / unitSeconds) * basePrice * resolutionWeight * audioWeight * referenceWeight`。
|
||||
- 公式示例:`count * (durationSeconds / unitSeconds) * basePrice * resolutionWeight * audioWeight * referenceWeight`。
|
||||
- 规则维度:`durationSeconds`、`resolution`、`count`、`hasAudio`、`hasReferenceVideo`、`hasReferenceImage`、`voice`。
|
||||
|
||||
音频、音乐、数字人、3D 模型:
|
||||
|
||||
Reference in New Issue
Block a user