778 lines
23 KiB
Go
778 lines
23 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
func (s *Server) health(w http.ResponseWriter, r *http.Request) {
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"ok": true,
|
|
"service": "easyai-ai-gateway",
|
|
"env": s.cfg.AppEnv,
|
|
"identityMode": s.cfg.IdentityMode,
|
|
})
|
|
}
|
|
|
|
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")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
|
|
}
|
|
|
|
func (s *Server) me(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
writeJSON(w, http.StatusOK, user)
|
|
}
|
|
|
|
func (s *Server) register(w http.ResponseWriter, r *http.Request) {
|
|
if !s.localIdentityEnabled() {
|
|
writeError(w, http.StatusForbidden, "local registration is disabled")
|
|
return
|
|
}
|
|
var input store.LocalRegisterInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
user, err := s.store.RegisterLocalUser(r.Context(), input)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrWeakPassword) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
if errors.Is(err, store.ErrInvalidInvitation) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
if errors.Is(err, store.ErrUserAlreadyExists) {
|
|
writeError(w, http.StatusConflict, err.Error())
|
|
return
|
|
}
|
|
s.logger.Error("register local user failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "register local user failed")
|
|
return
|
|
}
|
|
s.writeAuthResponse(w, http.StatusCreated, user)
|
|
}
|
|
|
|
func (s *Server) login(w http.ResponseWriter, r *http.Request) {
|
|
if !s.localIdentityEnabled() {
|
|
writeError(w, http.StatusForbidden, "local login is disabled")
|
|
return
|
|
}
|
|
var input store.LocalLoginInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
user, err := s.store.AuthenticateLocalUser(r.Context(), input)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrInvalidCredentials) {
|
|
writeError(w, http.StatusUnauthorized, "invalid account or password")
|
|
return
|
|
}
|
|
s.logger.Error("login local user failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "login failed")
|
|
return
|
|
}
|
|
s.writeAuthResponse(w, http.StatusOK, user)
|
|
}
|
|
|
|
func (s *Server) localIdentityEnabled() bool {
|
|
mode := strings.ToLower(strings.TrimSpace(s.cfg.IdentityMode))
|
|
return mode == "" || mode == "standalone" || mode == "hybrid"
|
|
}
|
|
|
|
func (s *Server) writeAuthResponse(w http.ResponseWriter, status int, user store.GatewayUser) {
|
|
authUser := authUserFromGatewayUser(user)
|
|
const ttl = 24 * time.Hour
|
|
token, err := s.auth.SignJWT(authUser, ttl)
|
|
if err != nil {
|
|
s.logger.Error("sign local jwt failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "token sign failed")
|
|
return
|
|
}
|
|
writeJSON(w, status, map[string]any{
|
|
"accessToken": token,
|
|
"tokenType": "Bearer",
|
|
"expiresIn": int(ttl.Seconds()),
|
|
"user": authUser,
|
|
})
|
|
}
|
|
|
|
func authUserFromGatewayUser(user store.GatewayUser) *auth.User {
|
|
roles := user.Roles
|
|
if len(roles) == 0 {
|
|
roles = []string{"user"}
|
|
}
|
|
tenantID := user.TenantID
|
|
if tenantID == "" {
|
|
tenantID = user.TenantKey
|
|
}
|
|
return &auth.User{
|
|
ID: user.ID,
|
|
Username: user.Username,
|
|
Roles: roles,
|
|
TenantID: tenantID,
|
|
GatewayTenantID: user.GatewayTenantID,
|
|
TenantKey: user.TenantKey,
|
|
Source: "gateway",
|
|
GatewayUserID: user.ID,
|
|
UserGroupID: user.DefaultUserGroupID,
|
|
}
|
|
}
|
|
|
|
func (s *Server) listPlatforms(w http.ResponseWriter, r *http.Request) {
|
|
platforms, err := s.store.ListPlatforms(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list platforms failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list platforms failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": platforms})
|
|
}
|
|
|
|
func (s *Server) listPlayablePlatforms(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
models, err := s.store.ListAccessiblePlatformModels(r.Context(), user)
|
|
if err != nil {
|
|
s.logger.Error("list playable platform models failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list playable platforms failed")
|
|
return
|
|
}
|
|
allowedPlatformIDs := map[string]bool{}
|
|
for _, model := range models {
|
|
allowedPlatformIDs[model.PlatformID] = true
|
|
}
|
|
platforms, err := s.store.ListPlatforms(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list platforms failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list playable platforms failed")
|
|
return
|
|
}
|
|
filtered := platforms[:0]
|
|
for _, platform := range platforms {
|
|
if platform.Status == "enabled" && allowedPlatformIDs[platform.ID] {
|
|
filtered = append(filtered, platform)
|
|
}
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": filtered})
|
|
}
|
|
|
|
func (s *Server) createPlatform(w http.ResponseWriter, r *http.Request) {
|
|
var input store.CreatePlatformInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
input.Provider = strings.TrimSpace(input.Provider)
|
|
input.Name = strings.TrimSpace(input.Name)
|
|
input.InternalName = strings.TrimSpace(input.InternalName)
|
|
if input.Provider == "" || input.Name == "" {
|
|
writeError(w, http.StatusBadRequest, "provider and name are required")
|
|
return
|
|
}
|
|
if input.AuthType == "" {
|
|
input.AuthType = "bearer"
|
|
}
|
|
platform, err := s.store.CreatePlatform(r.Context(), input)
|
|
if err != nil {
|
|
s.logger.Error("create platform failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "create platform failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, platform)
|
|
}
|
|
|
|
func (s *Server) updatePlatform(w http.ResponseWriter, r *http.Request) {
|
|
var input store.CreatePlatformInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
input.Provider = strings.TrimSpace(input.Provider)
|
|
input.Name = strings.TrimSpace(input.Name)
|
|
input.InternalName = strings.TrimSpace(input.InternalName)
|
|
if input.Provider == "" || input.Name == "" {
|
|
writeError(w, http.StatusBadRequest, "provider and name are required")
|
|
return
|
|
}
|
|
if input.AuthType == "" {
|
|
input.AuthType = "bearer"
|
|
}
|
|
platform, err := s.store.UpdatePlatform(r.Context(), r.PathValue("platformID"), input)
|
|
if err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "platform not found")
|
|
return
|
|
}
|
|
if store.IsUniqueViolation(err) {
|
|
writeError(w, http.StatusConflict, "platform key already exists")
|
|
return
|
|
}
|
|
s.logger.Error("update platform failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "update platform failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, platform)
|
|
}
|
|
|
|
func (s *Server) deletePlatform(w http.ResponseWriter, r *http.Request) {
|
|
if err := s.store.DeletePlatform(r.Context(), r.PathValue("platformID")); err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "platform not found")
|
|
return
|
|
}
|
|
s.logger.Error("delete platform failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "delete platform failed")
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
func (s *Server) createPlatformModel(w http.ResponseWriter, r *http.Request) {
|
|
var input store.CreatePlatformModelInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
if pathPlatformID := r.PathValue("platformID"); pathPlatformID != "" {
|
|
input.PlatformID = pathPlatformID
|
|
}
|
|
if input.PlatformID == "" {
|
|
writeError(w, http.StatusBadRequest, "platformId is required")
|
|
return
|
|
}
|
|
model, err := s.store.CreatePlatformModel(r.Context(), input)
|
|
if err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "base model not found")
|
|
return
|
|
}
|
|
s.logger.Error("create platform model failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "create platform model failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, s.platformModelResponse(r.Context(), model))
|
|
}
|
|
|
|
func (s *Server) replacePlatformModels(w http.ResponseWriter, r *http.Request) {
|
|
platformID := r.PathValue("platformID")
|
|
if platformID == "" {
|
|
writeError(w, http.StatusBadRequest, "platformId is required")
|
|
return
|
|
}
|
|
|
|
var input struct {
|
|
Models []store.CreatePlatformModelInput `json:"models"`
|
|
}
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
|
|
models, err := s.store.ReplacePlatformModels(r.Context(), platformID, input.Models)
|
|
if err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "base model not found")
|
|
return
|
|
}
|
|
s.logger.Error("replace platform models failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "replace platform models failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": s.platformModelResponses(r.Context(), models)})
|
|
}
|
|
|
|
func (s *Server) deletePlatformModel(w http.ResponseWriter, r *http.Request) {
|
|
if err := s.store.DeletePlatformModel(r.Context(), r.PathValue("modelID")); err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "platform model not found")
|
|
return
|
|
}
|
|
s.logger.Error("delete platform model failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "delete platform model failed")
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
func (s *Server) listModels(w http.ResponseWriter, r *http.Request) {
|
|
models, err := s.store.ListModels(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list models failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list models failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": s.platformModelResponses(r.Context(), models)})
|
|
}
|
|
|
|
func (s *Server) listPlayableModels(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
models, err := s.store.ListAccessiblePlatformModels(r.Context(), user)
|
|
if err != nil {
|
|
s.logger.Error("list playable models failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list playable models failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": s.platformModelResponses(r.Context(), models)})
|
|
}
|
|
|
|
func (s *Server) listPricingRules(w http.ResponseWriter, r *http.Request) {
|
|
items, err := s.store.ListPricingRules(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list pricing rules failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list pricing rules failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) listTenants(w http.ResponseWriter, r *http.Request) {
|
|
items, err := s.store.ListTenants(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list tenants failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list tenants failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) listUsers(w http.ResponseWriter, r *http.Request) {
|
|
items, err := s.store.ListUsers(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list users failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list users failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) listUserGroups(w http.ResponseWriter, r *http.Request) {
|
|
items, err := s.store.ListUserGroups(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list user groups failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list user groups failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) listAPIKeys(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
items, err := s.store.ListAPIKeys(r.Context(), user)
|
|
if err != nil {
|
|
s.logger.Error("list api keys failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list api keys failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) listPlayableAPIKeys(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
items, err := s.store.ListPlayableAPIKeys(r.Context(), user)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrLocalUserRequired) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
s.logger.Error("list playable api keys failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list playable api keys failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) createAPIKey(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
var input store.CreateAPIKeyInput
|
|
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
created, err := s.store.CreateAPIKey(r.Context(), input, user)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrLocalUserRequired) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
s.logger.Error("create api key failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "create api key failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, created)
|
|
}
|
|
|
|
func (s *Server) disableAPIKey(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
item, err := s.store.DisableAPIKey(r.Context(), r.PathValue("apiKeyID"), user)
|
|
if err == nil {
|
|
writeJSON(w, http.StatusOK, item)
|
|
return
|
|
}
|
|
if errors.Is(err, store.ErrLocalUserRequired) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "api key not found")
|
|
return
|
|
}
|
|
s.logger.Error("disable api key failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "disable api key failed")
|
|
}
|
|
|
|
func (s *Server) deleteAPIKey(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
err := s.store.DeleteAPIKey(r.Context(), r.PathValue("apiKeyID"), user)
|
|
if err == nil {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
if errors.Is(err, store.ErrLocalUserRequired) {
|
|
writeError(w, http.StatusBadRequest, err.Error())
|
|
return
|
|
}
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "api key not found")
|
|
return
|
|
}
|
|
s.logger.Error("delete api key failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "delete api key failed")
|
|
}
|
|
|
|
func (s *Server) estimatePricing(w http.ResponseWriter, r *http.Request) {
|
|
user, _ := auth.UserFromContext(r.Context())
|
|
var body map[string]any
|
|
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
model, _ := body["model"].(string)
|
|
kind, _ := body["kind"].(string)
|
|
if kind == "" {
|
|
kind = "chat.completions"
|
|
}
|
|
if model == "" {
|
|
writeError(w, http.StatusBadRequest, "model is required")
|
|
return
|
|
}
|
|
if !apiKeyScopeAllowed(user, kind) {
|
|
writeError(w, http.StatusForbidden, "api key scope does not allow this capability")
|
|
return
|
|
}
|
|
estimate, err := s.runner.Estimate(r.Context(), kind, model, body, user)
|
|
if err != nil {
|
|
if errors.Is(err, store.ErrNoModelCandidate) {
|
|
writeError(w, http.StatusNotFound, "no enabled platform model matches request")
|
|
return
|
|
}
|
|
s.logger.Error("estimate pricing failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "estimate pricing failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, estimate)
|
|
}
|
|
|
|
func (s *Server) listRateLimitWindows(w http.ResponseWriter, r *http.Request) {
|
|
items, err := s.store.ListRateLimitWindows(r.Context())
|
|
if err != nil {
|
|
s.logger.Error("list rate limit windows failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list rate limit windows failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"items": items})
|
|
}
|
|
|
|
func (s *Server) createTask(kind string, compatible bool) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
user, ok := auth.UserFromContext(r.Context())
|
|
if !ok {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return
|
|
}
|
|
|
|
var body map[string]any
|
|
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid json body")
|
|
return
|
|
}
|
|
model, _ := body["model"].(string)
|
|
if model == "" {
|
|
writeError(w, http.StatusBadRequest, "model is required")
|
|
return
|
|
}
|
|
if !apiKeyScopeAllowed(user, kind) {
|
|
writeError(w, http.StatusForbidden, "api key scope does not allow this capability")
|
|
return
|
|
}
|
|
|
|
task, err := s.store.CreateTask(r.Context(), store.CreateTaskInput{
|
|
Kind: kind,
|
|
Model: model,
|
|
RunMode: runModeFromRequest(body),
|
|
Request: body,
|
|
}, user)
|
|
if err != nil {
|
|
s.logger.Error("create task failed", "kind", kind, "error", err)
|
|
writeError(w, http.StatusInternalServerError, "create task failed")
|
|
return
|
|
}
|
|
if compatible {
|
|
if boolValue(body, "stream") {
|
|
flusher := prepareCompatibleStream(w)
|
|
result, runErr := s.runner.ExecuteStream(r.Context(), task, user, func(delta string) error {
|
|
writeCompatibleDelta(w, kind, model, delta)
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
return nil
|
|
})
|
|
if runErr != nil {
|
|
sendSSE(w, "error", map[string]any{"error": map[string]any{"message": runErr.Error(), "status": statusFromRunError(runErr)}})
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
return
|
|
}
|
|
writeCompatibleDone(w, kind, model, result.Output)
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
return
|
|
}
|
|
result, runErr := s.runner.Execute(r.Context(), task, user)
|
|
if runErr != nil {
|
|
writeError(w, statusFromRunError(runErr), runErr.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, result.Output)
|
|
return
|
|
}
|
|
result, runErr := s.runner.Execute(r.Context(), task, user)
|
|
if runErr != nil {
|
|
s.logger.Warn("task completed with failure", "kind", kind, "taskId", task.ID, "error", runErr)
|
|
}
|
|
|
|
writeJSON(w, http.StatusAccepted, map[string]any{
|
|
"task": result.Task,
|
|
"next": map[string]string{
|
|
"events": fmt.Sprintf("/api/v1/tasks/%s/events", task.ID),
|
|
"detail": fmt.Sprintf("/api/v1/tasks/%s", task.ID),
|
|
},
|
|
})
|
|
})
|
|
}
|
|
|
|
func apiKeyScopeAllowed(user *auth.User, kind string) bool {
|
|
if user == nil || strings.TrimSpace(user.APIKeyID) == "" || len(user.APIKeyScopes) == 0 {
|
|
return true
|
|
}
|
|
required := scopeForTaskKind(kind)
|
|
for _, scope := range user.APIKeyScopes {
|
|
scope = strings.TrimSpace(strings.ToLower(scope))
|
|
if scope == "*" || scope == "all" || scope == required {
|
|
return true
|
|
}
|
|
if required == "chat" && (scope == "text" || scope == "text_generate") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func scopeForTaskKind(kind string) string {
|
|
switch kind {
|
|
case "chat.completions", "responses":
|
|
return "chat"
|
|
case "images.generations", "images.edits":
|
|
return "image"
|
|
case "videos.generations":
|
|
return "video"
|
|
default:
|
|
return kind
|
|
}
|
|
}
|
|
|
|
func statusFromRunError(err error) int {
|
|
switch {
|
|
case errors.Is(err, store.ErrNoModelCandidate):
|
|
return http.StatusNotFound
|
|
case errors.Is(err, store.ErrRateLimited):
|
|
return http.StatusTooManyRequests
|
|
default:
|
|
return http.StatusBadGateway
|
|
}
|
|
}
|
|
|
|
func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
|
|
user, ok := auth.UserFromContext(r.Context())
|
|
if !ok {
|
|
writeError(w, http.StatusUnauthorized, "unauthorized")
|
|
return
|
|
}
|
|
query := r.URL.Query()
|
|
page, err := positiveQueryInt(query.Get("page"), 1)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid page")
|
|
return
|
|
}
|
|
pageSizeRaw := query.Get("pageSize")
|
|
if pageSizeRaw == "" {
|
|
pageSizeRaw = query.Get("limit")
|
|
}
|
|
pageSize, err := positiveQueryInt(pageSizeRaw, 50)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid pageSize")
|
|
return
|
|
}
|
|
createdFrom, err := parseTaskListTime(query.Get("createdFrom"), query.Get("from"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid createdFrom")
|
|
return
|
|
}
|
|
createdTo, err := parseTaskListTime(query.Get("createdTo"), query.Get("to"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid createdTo")
|
|
return
|
|
}
|
|
result, err := s.store.ListTasks(r.Context(), user, store.TaskListFilter{
|
|
Query: firstNonEmpty(query.Get("q"), query.Get("query")),
|
|
ModelType: firstNonEmpty(query.Get("modelType"), query.Get("type")),
|
|
CreatedFrom: createdFrom,
|
|
CreatedTo: createdTo,
|
|
Page: page,
|
|
PageSize: pageSize,
|
|
})
|
|
if err != nil {
|
|
s.logger.Error("list tasks failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "list tasks failed")
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"items": result.Items,
|
|
"total": result.Total,
|
|
"page": result.Page,
|
|
"pageSize": result.PageSize,
|
|
})
|
|
}
|
|
|
|
func positiveQueryInt(raw string, fallback int) (int, error) {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return fallback, nil
|
|
}
|
|
value, err := strconv.Atoi(raw)
|
|
if err != nil || value <= 0 {
|
|
return 0, fmt.Errorf("invalid positive integer")
|
|
}
|
|
return value, nil
|
|
}
|
|
|
|
func parseTaskListTime(values ...string) (*time.Time, error) {
|
|
raw := strings.TrimSpace(firstNonEmpty(values...))
|
|
if raw == "" {
|
|
return nil, nil
|
|
}
|
|
layouts := []string{time.RFC3339Nano, time.RFC3339, "2006-01-02T15:04", "2006-01-02 15:04:05", "2006-01-02"}
|
|
var lastErr error
|
|
for _, layout := range layouts {
|
|
parsed, err := time.ParseInLocation(layout, raw, time.Local)
|
|
if err == nil {
|
|
return &parsed, nil
|
|
}
|
|
lastErr = err
|
|
}
|
|
return nil, lastErr
|
|
}
|
|
|
|
func firstNonEmpty(values ...string) string {
|
|
for _, value := range values {
|
|
if trimmed := strings.TrimSpace(value); trimmed != "" {
|
|
return trimmed
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func boolValue(body map[string]any, key string) bool {
|
|
value, _ := body[key].(bool)
|
|
return value
|
|
}
|
|
|
|
func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
|
|
task, err := s.store.GetTask(r.Context(), r.PathValue("taskID"))
|
|
if err == nil {
|
|
writeJSON(w, http.StatusOK, task)
|
|
return
|
|
}
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "task not found")
|
|
return
|
|
}
|
|
s.logger.Error("get task failed", "error", err)
|
|
writeError(w, http.StatusInternalServerError, "get task failed")
|
|
}
|
|
|
|
func (s *Server) taskEvents(w http.ResponseWriter, r *http.Request) {
|
|
task, err := s.store.GetTask(r.Context(), r.PathValue("taskID"))
|
|
if err != nil {
|
|
if store.IsNotFound(err) {
|
|
writeError(w, http.StatusNotFound, "task not found")
|
|
return
|
|
}
|
|
writeError(w, http.StatusInternalServerError, "get task failed")
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
|
|
events, err := s.store.ListTaskEvents(r.Context(), task.ID)
|
|
if err != nil {
|
|
s.logger.Error("list task events failed", "error", err)
|
|
return
|
|
}
|
|
for _, event := range events {
|
|
sendSSE(w, event.EventType, event)
|
|
if flusher, ok := w.(http.Flusher); ok {
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
if len(events) == 0 {
|
|
sendSSE(w, "task.accepted", map[string]any{
|
|
"taskId": task.ID,
|
|
"status": task.Status,
|
|
})
|
|
}
|
|
}
|
|
|
|
func runModeFromRequest(body map[string]any) string {
|
|
if value, ok := body["runMode"].(string); ok {
|
|
return value
|
|
}
|
|
if value, ok := body["mode"].(string); ok {
|
|
return value
|
|
}
|
|
if value, ok := body["simulation"].(bool); ok && value {
|
|
return "simulation"
|
|
}
|
|
if value, ok := body["testMode"].(bool); ok && value {
|
|
return "simulation"
|
|
}
|
|
return ""
|
|
}
|