BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: add local static file storage fallback

Browse Source
This commit is contained in:
wangbo
2026-05-13 20:53:07 +08:00
parent fc5dfd6bc5
commit 62d9170e65
7 changed files with 395 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/runner/upload.go
+180 -67
View File
@@ -9,20 +9,29 @@ import (
"encoding/json"
"fmt"
"io"
"mime"
"mime/multipart"
"net/http"
"net/textproto"
"net/url"
"os"
"path/filepath"
"strings"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)
const defaultServerMainOpenAPIUploadURL = "http://127.0.0.1:3001/v1/files/upload"
const maxGeneratedAssetFetchBytes = 256 << 20
const (
localStaticGeneratedPathPrefix = "/static/generated/"
localStaticUploadedPathPrefix = "/static/uploaded/"
)
type FileUploadPayload struct {
ContentType string
FileName string
@@ -32,8 +41,9 @@ type FileUploadPayload struct {
}
type generatedAssetUploadPolicy struct {
UploadInlineMedia bool
UploadURLMedia bool
UploadInlineMedia bool
UploadURLMedia bool
StoreInlineMediaLocally bool
}
type generatedAssetDecision struct {
@@ -96,14 +106,11 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task
return result, nil
}
var channels []store.FileStorageChannel
if needsUpload {
if needsUpload && generatedAssetNeedsChannelLookup(policy, decisions) {
channels, err = s.activeFileStorageChannels(ctx, store.FileStorageSceneImageResult)
if err != nil {
return nil, &clients.ClientError{Code: "upload_config_failed", Message: err.Error(), Retryable: true}
}
if len(channels) == 0 {
return nil, &clients.ClientError{Code: "upload_no_channel", Message: "no enabled file storage channel for generated media results", Retryable: false}
}
}
next := map[string]any{}
for key, value := range result {
@@ -132,13 +139,11 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task
var contentType string
var err error
if decision.Inline != nil {
upload, contentType, kind, err = s.uploadGeneratedAsset(ctx, taskID, decision.Inline, index, channels)
upload, contentType, kind, strategy, err = s.uploadGeneratedAsset(ctx, taskID, decision.Inline, index, channels, policy.StoreInlineMediaLocally)
sourceKey = decision.Inline.SourceKey
strategy = "upload_inline_media"
} else {
upload, contentType, kind, err = s.uploadGeneratedURLAsset(ctx, taskID, decision.URL, index, channels)
upload, contentType, kind, strategy, err = s.uploadGeneratedURLAsset(ctx, taskID, decision.URL, index, channels)
sourceKey = decision.URL.SourceKey
strategy = "upload_url_media"
}
if err != nil {
return nil, err
@@ -191,40 +196,174 @@ func generatedAssetUploadPolicyFromName(policyName string) generatedAssetUploadP
case store.FileStorageResultUploadPolicyUploadAll:
return generatedAssetUploadPolicy{UploadInlineMedia: true, UploadURLMedia: true}
case store.FileStorageResultUploadPolicyUploadNone:
return generatedAssetUploadPolicy{UploadInlineMedia: false, UploadURLMedia: false}
return generatedAssetUploadPolicy{UploadInlineMedia: true, UploadURLMedia: false, StoreInlineMediaLocally: true}
default:
return defaultGeneratedAssetUploadPolicy()
}
}
func (s *Service) uploadGeneratedAsset(ctx context.Context, taskID string, asset *generatedInlineAsset, index int, channels []store.FileStorageChannel) (map[string]any, string, string, error) {
func generatedAssetNeedsChannelLookup(policy generatedAssetUploadPolicy, decisions []generatedAssetDecision) bool {
for _, decision := range decisions {
if decision.URL != nil {
return true
}
if decision.Inline != nil && !policy.StoreInlineMediaLocally {
return true
}
}
return false
}
func (s *Service) uploadGeneratedAsset(ctx context.Context, taskID string, asset *generatedInlineAsset, index int, channels []store.FileStorageChannel, forceLocal bool) (map[string]any, string, string, string, error) {
contentType := resolvedGeneratedAssetContentType(asset.ContentType, asset.Kind, asset.Bytes)
kind := generatedAssetKindFromContentType(asset.Kind, contentType)
upload, err := s.uploadFileWithFailover(ctx, FileUploadPayload{
payload := FileUploadPayload{
Bytes: asset.Bytes,
ContentType: contentType,
FileName: generatedAssetFileName(taskID, index, contentType, kind),
Scene: store.FileStorageSceneImageResult,
Source: "ai-gateway",
}, channels)
return upload, contentType, kind, err
}
if forceLocal || len(channels) == 0 {
upload, err := s.storeFileLocally(payload, s.cfg.LocalGeneratedStorageDir, config.DefaultLocalGeneratedStorageDir, localStaticGeneratedPathPrefix)
return upload, contentType, kind, "local_static_inline_media", err
}
upload, err := s.uploadFileWithFailover(ctx, payload, channels)
return upload, contentType, kind, "upload_inline_media", err
}
func (s *Service) uploadGeneratedURLAsset(ctx context.Context, taskID string, asset *generatedURLAsset, index int, channels []store.FileStorageChannel) (map[string]any, string, string, error) {
func (s *Service) uploadGeneratedURLAsset(ctx context.Context, taskID string, asset *generatedURLAsset, index int, channels []store.FileStorageChannel) (map[string]any, string, string, string, error) {
payload, contentType, err := s.readGeneratedURLAsset(ctx, asset)
if err != nil {
return nil, "", "", err
return nil, "", "", "", err
}
contentType = resolvedGeneratedAssetContentType(firstNonEmptyString(contentType, asset.ContentType), asset.Kind, payload)
kind := generatedAssetKindFromContentType(asset.Kind, contentType)
upload, err := s.uploadFileWithFailover(ctx, FileUploadPayload{
uploadPayload := FileUploadPayload{
Bytes: payload,
ContentType: contentType,
FileName: generatedAssetFileName(taskID, index, contentType, kind),
Scene: store.FileStorageSceneImageResult,
Source: "ai-gateway",
}, channels)
return upload, contentType, kind, err
}
if len(channels) == 0 {
upload, err := s.storeFileLocally(uploadPayload, s.cfg.LocalGeneratedStorageDir, config.DefaultLocalGeneratedStorageDir, localStaticGeneratedPathPrefix)
return upload, contentType, kind, "local_static_url_media", err
}
upload, err := s.uploadFileWithFailover(ctx, uploadPayload, channels)
return upload, contentType, kind, "upload_url_media", err
}
func (s *Service) storeFileLocally(payload FileUploadPayload, storageDir string, fallbackStorageDir string, pathPrefix string) (map[string]any, error) {
storageDir = strings.TrimSpace(storageDir)
if storageDir == "" {
storageDir = fallbackStorageDir
}
if err := os.MkdirAll(storageDir, 0o755); err != nil {
return nil, &clients.ClientError{Code: "local_static_store_failed", Message: err.Error(), Retryable: true}
}
fileName := filepath.Base(strings.TrimSpace(payload.FileName))
if fileName == "" || fileName == "." || fileName == ".." || fileName == string(filepath.Separator) {
kind := generatedAssetKindFromContentType("", payload.ContentType)
fileName = generatedAssetFileName("generated", 0, payload.ContentType, kind)
}
targetPath := filepath.Join(storageDir, fileName)
file, err := os.OpenFile(targetPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
if err != nil {
return nil, &clients.ClientError{Code: "local_static_store_failed", Message: err.Error(), Retryable: true}
}
_, writeErr := file.Write(payload.Bytes)
closeErr := file.Close()
if writeErr != nil {
_ = os.Remove(targetPath)
return nil, &clients.ClientError{Code: "local_static_store_failed", Message: writeErr.Error(), Retryable: true}
}
if closeErr != nil {
_ = os.Remove(targetPath)
return nil, &clients.ClientError{Code: "local_static_store_failed", Message: closeErr.Error(), Retryable: true}
}
return map[string]any{
"url": s.localStaticFileURL(fileName, pathPrefix),
"fileName": fileName,
"contentType": payload.ContentType,
"size": len(payload.Bytes),
"storageChannel": map[string]any{
"id": "local-static",
"channelKey": "local-static",
"name": "AI Gateway local static storage",
"provider": "local_static",
},
}, nil
}
func (s *Service) localStaticFileURL(fileName string, pathPrefix string) string {
if strings.TrimSpace(pathPrefix) == "" {
pathPrefix = localStaticUploadedPathPrefix
}
path := pathPrefix + url.PathEscape(filepath.Base(fileName))
baseURL := strings.TrimRight(strings.TrimSpace(s.cfg.PublicBaseURL), "/")
if baseURL == "" {
return path
}
return baseURL + path
}
func localStaticUploadFileName(originalName string, contentType string) string {
baseName := filepath.Base(strings.TrimSpace(originalName))
originalExt := strings.ToLower(filepath.Ext(baseName))
namePart := strings.TrimSuffix(baseName, originalExt)
namePart = sanitizeGeneratedAssetNamePart(namePart)
if namePart == "" {
namePart = "gateway-upload"
}
if len(namePart) > 48 {
namePart = namePart[:48]
}
return fmt.Sprintf("%s-%s%s", namePart, randomHexSuffix(6), uploadFileExtension(contentType, originalExt))
}
func uploadFileExtension(contentType string, fallbackExt string) string {
normalized := normalizeGeneratedContentType(contentType)
if generatedContentTypeIsMedia(normalized) {
return fileExtensionForContentType(normalized, generatedAssetKindFromContentType("", normalized))
}
if normalized != "" && normalized != "application/octet-stream" {
if extensions, err := mime.ExtensionsByType(normalized); err == nil && len(extensions) > 0 {
if ext := sanitizeFileExtension(extensions[0]); ext != "" {
return ext
}
}
}
if ext := sanitizeFileExtension(fallbackExt); ext != "" {
return ext
}
if normalized == "application/json" {
return ".json"
}
if strings.HasPrefix(normalized, "text/") {
return ".txt"
}
return ".bin"
}
func sanitizeFileExtension(value string) string {
value = strings.ToLower(strings.TrimSpace(value))
if value == "" {
return ""
}
if !strings.HasPrefix(value, ".") {
value = "." + value
}
if len(value) > 16 {
return ""
}
for _, item := range value[1:] {
if (item >= 'a' && item <= 'z') || (item >= '0' && item <= '9') {
continue
}
return ""
}
return value
}
func (s *Service) readGeneratedURLAsset(ctx context.Context, asset *generatedURLAsset) ([]byte, string, error) {
@@ -316,12 +455,25 @@ func (s *Service) UploadFile(ctx context.Context, payload FileUploadPayload) (ma
return nil, &clients.ClientError{Code: "upload_config_failed", Message: err.Error(), Retryable: true}
}
if len(channels) == 0 {
return nil, &clients.ClientError{Code: "upload_no_channel", Message: "no enabled file storage channel", Retryable: false}
payload.FileName = localStaticUploadFileName(payload.FileName, payload.ContentType)
upload, err := s.storeFileLocally(payload, s.cfg.LocalUploadedStorageDir, config.DefaultLocalUploadedStorageDir, localStaticUploadedPathPrefix)
if err != nil {
return nil, err
}
upload["assetStorage"] = map[string]any{
"scene": payload.Scene,
"source": firstNonEmptyString(payload.Source, "ai-gateway-openapi"),
"strategy": "local_static_upload",
}
return upload, nil
}
return s.uploadFileWithFailover(ctx, payload, channels)
}
func (s *Service) activeFileStorageChannels(ctx context.Context, scene string) ([]store.FileStorageChannel, error) {
if s.store == nil {
return nil, nil
}
channels, err := s.store.ListEnabledFileStorageChannelsForScene(ctx, scene)
if err != nil && !store.IsUndefinedDatabaseObject(err) {
return nil, err
@@ -329,33 +481,7 @@ func (s *Service) activeFileStorageChannels(ctx context.Context, scene string) (
if len(channels) > 0 {
return channels, nil
}
fallback := s.fallbackFileStorageChannel()
if fallback == nil {
return nil, nil
}
if !fileStorageChannelSupportsScene(*fallback, scene) {
return nil, nil
}
return []store.FileStorageChannel{*fallback}, nil
}
func (s *Service) fallbackFileStorageChannel() *store.FileStorageChannel {
baseURL := strings.TrimRight(strings.TrimSpace(s.cfg.ServerMainBaseURL), "/")
apiKey := strings.TrimSpace(s.cfg.ServerMainInternalToken)
if baseURL == "" || apiKey == "" {
return nil
}
return &store.FileStorageChannel{
ChannelKey: "server-main-env-fallback",
Name: "server-main env fallback",
Provider: "server_main_openapi",
UploadURL: baseURL + "/v1/files/upload",
APIKey: apiKey,
Scenes: []string{store.FileStorageSceneUpload, store.FileStorageSceneImageResult},
RetryPolicy: defaultUploadRetryPolicy(),
Priority: 100,
Status: "enabled",
}
return nil, nil
}
func (s *Service) uploadFileWithFailover(ctx context.Context, payload FileUploadPayload, channels []store.FileStorageChannel) (map[string]any, error) {
@@ -363,11 +489,15 @@ func (s *Service) uploadFileWithFailover(ctx context.Context, payload FileUpload
for _, channel := range channels {
upload, err := s.uploadWithChannelRetries(ctx, payload, channel)
if err == nil {
_ = s.store.MarkFileStorageChannelSuccess(context.WithoutCancel(ctx), channel.ID)
if s.store != nil {
_ = s.store.MarkFileStorageChannelSuccess(context.WithoutCancel(ctx), channel.ID)
}
return upload, nil
}
lastErr = err
_ = s.store.MarkFileStorageChannelFailure(context.WithoutCancel(ctx), channel.ID, err.Error())
if s.store != nil {
_ = s.store.MarkFileStorageChannelFailure(context.WithoutCancel(ctx), channel.ID, err.Error())
}
}
if lastErr != nil {
return nil, lastErr
@@ -375,23 +505,6 @@ func (s *Service) uploadFileWithFailover(ctx context.Context, payload FileUpload
return nil, &clients.ClientError{Code: "upload_no_channel", Message: "no enabled file storage channel", Retryable: false}
}
func fileStorageChannelSupportsScene(channel store.FileStorageChannel, scene string) bool {
scene = strings.TrimSpace(scene)
if scene == "" {
return true
}
scenes := channel.Scenes
if len(scenes) == 0 {
scenes = []string{store.FileStorageSceneUpload, store.FileStorageSceneImageResult}
}
for _, item := range scenes {
if strings.TrimSpace(item) == scene {
return true
}
}
return false
}
func (s *Service) uploadWithChannelRetries(ctx context.Context, payload FileUploadPayload, channel store.FileStorageChannel) (map[string]any, error) {
maxRetries, delays := uploadRetrySchedule(channel.RetryPolicy)
var lastErr error