wangbo
3053ba4925
引入动态流量门禁、隔离验收身份与协议级 Gemini/Volces 模拟器，覆盖双站点 API、Worker、PostgreSQL、River、账务、回调和媒体物化链路。 新增 P24/P28/P32 容量阶梯、Worker 强杀恢复、真实小流量 canary、CAS 放量和失败保持 validation 的生产编排；Worker 执行槽、连接池、媒体并发和双站点副本数改为环境配置。 验证：Go 全量测试、真实 PostgreSQL 迁移集成测试、迁移安全检查、OpenAPI 生成、ShellCheck、Kustomize、gofmt 和 git diff --check。
78 lines
2.0 KiB
Go
78 lines
2.0 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
type gatewayTaskCreationStage string
|
|
|
|
const (
|
|
gatewayTaskCreationPrepare gatewayTaskCreationStage = "prepare"
|
|
gatewayTaskCreationStore gatewayTaskCreationStage = "store"
|
|
)
|
|
|
|
type gatewayTaskCreationError struct {
|
|
Stage gatewayTaskCreationStage
|
|
Err error
|
|
}
|
|
|
|
func (e *gatewayTaskCreationError) Error() string {
|
|
if e == nil || e.Err == nil {
|
|
return "gateway task creation failed"
|
|
}
|
|
return e.Err.Error()
|
|
}
|
|
|
|
func (e *gatewayTaskCreationError) Unwrap() error {
|
|
if e == nil {
|
|
return nil
|
|
}
|
|
return e.Err
|
|
}
|
|
|
|
func (s *Server) prepareAndCreateGatewayTask(
|
|
ctx context.Context,
|
|
r *http.Request,
|
|
user *auth.User,
|
|
kind string,
|
|
model string,
|
|
body map[string]any,
|
|
async bool,
|
|
) (store.GatewayTask, error) {
|
|
admission, err := s.authorizeTaskTraffic(r, user)
|
|
if err != nil {
|
|
return store.GatewayTask{}, &gatewayTaskCreationError{Stage: gatewayTaskCreationPrepare, Err: err}
|
|
}
|
|
prepared, err := s.prepareTaskRequest(ctx, r, user, body)
|
|
if err != nil {
|
|
return store.GatewayTask{}, &gatewayTaskCreationError{Stage: gatewayTaskCreationPrepare, Err: err}
|
|
}
|
|
runMode, err := s.admittedTaskRunMode(admission, prepared.Body)
|
|
if err != nil {
|
|
return store.GatewayTask{}, &gatewayTaskCreationError{Stage: gatewayTaskCreationPrepare, Err: err}
|
|
}
|
|
task, err := s.store.CreateTask(ctx, store.CreateTaskInput{
|
|
Kind: kind,
|
|
Model: model,
|
|
RunMode: runMode,
|
|
AcceptanceRunID: admission.AcceptanceRunID,
|
|
Async: async,
|
|
Request: prepared.Body,
|
|
ConversationID: prepared.ConversationID,
|
|
NewMessageCount: prepared.NewMessageCount,
|
|
MessageRefs: prepared.MessageRefs,
|
|
}, user)
|
|
if err != nil {
|
|
return store.GatewayTask{}, &gatewayTaskCreationError{
|
|
Stage: gatewayTaskCreationStore,
|
|
Err: fmt.Errorf("create task: %w", err),
|
|
}
|
|
}
|
|
return task, nil
|
|
}
|