feat(image-storage): 异步生图对象存储改为后台配置,保存即生效

此前开启异步生图必须改服务器上的 config.yaml 并重启容器(#4542),且若想
复用已配置的备份 S3,还得把同一套凭证再填一遍(#4458)。

- 新增 ImageStorageSettingService:配置存 settings 表,SecretAccessKey 经
  SecretEncryptor 加密落库、读回脱敏、留空表示沿用旧值,与备份 S3 配置同一套做法。
- reuse_backup_s3(默认开)直接借用 backup_s3_config 的端点与密钥,只用自己的
  bucket/prefix 区分对象,因此备份走 backups/、图片走 images/,且密钥不会在库里存两份。
- ImageTaskService 的启用状态改由 ImageStorageResolver 在运行时解析并缓存,
  保存设置后 Invalidate 使下次请求重建客户端——不再需要重启。
- repository 侧由提供实例改为提供工厂,客户端才可能在运行期重建。
- 轮询接口的门控从 enabled() 放宽为 Pollable():关掉开关只拒绝新提交,
  已受理的任务仍可取回结果,不再被中途吞掉。
- config.yaml 的 image_storage 保留为回落,后台从未保存过时沿用,
  升级前已用配置文件开启的部署不受影响。
- 管理端 GET/PUT/POST /admin/backups/image-storage,PUT 与备份 S3 配置一样要求
  step-up 2FA:改写存储目标同样能把生成内容导向外部账号。

注:go generate ./cmd/server 在当前 upstream 基线上即失败(securityaudit.
PromptAdminService 缺 provider),故 wire_gen.go 为手工同步。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHreE5pzCkSYz7J45fmd2Y
This commit is contained in:
haruka
2026-07-19 00:56:21 -07:00
parent 37db8d031b
commit b08cab91a9
11 changed files with 853 additions and 36 deletions
+4 -6
View File
@@ -198,7 +198,9 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
backupObjectStoreFactory := repository.NewS3BackupStoreFactory()
dbDumper := repository.NewPgDumper(configConfig)
backupService := service.ProvideBackupService(settingRepository, configConfig, secretEncryptor, backupObjectStoreFactory, dbDumper)
backupHandler := admin.NewBackupHandler(backupService, userService)
imageStorageFactory := repository.ProvideImageStorageFactory()
imageStorageSettingService := service.ProvideImageStorageSettingService(settingRepository, secretEncryptor, backupService, imageStorageFactory, configConfig)
backupHandler := admin.NewBackupHandler(backupService, userService, imageStorageSettingService)
oAuthHandler := admin.NewOAuthHandler(oAuthService)
openAIOAuthHandler := admin.NewOpenAIOAuthHandler(openAIOAuthService, adminService, openAIQuotaService)
geminiOAuthHandler := admin.NewGeminiOAuthHandler(geminiOAuthService)
@@ -277,11 +279,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
paymentWebhookHandler := handler.NewPaymentWebhookHandler(paymentService, registry)
availableChannelHandler := handler.NewAvailableChannelHandler(channelService, apiKeyService, settingService)
imageTaskStore := repository.NewImageTaskStore(redisClient)
imageStorage, err := repository.ProvideImageStorage(configConfig)
if err != nil {
return nil, err
}
imageTaskService := service.ProvideImageTaskService(imageTaskStore, imageStorage, configConfig)
imageTaskService := service.ProvideImageTaskService(imageTaskStore, imageStorageSettingService)
asyncImageHandler := handler.NewAsyncImageHandler(imageTaskService, openAIGatewayHandler)
batchImageRepository := repository.NewBatchImageRepository(db)
batchImageQueue := repository.NewBatchImageQueue(redisClient, configConfig)
@@ -10,12 +10,14 @@ import (
type BackupHandler struct {
backupService *service.BackupService
userService *service.UserService
imageStorage *service.ImageStorageSettingService
}
func NewBackupHandler(backupService *service.BackupService, userService *service.UserService) *BackupHandler {
func NewBackupHandler(backupService *service.BackupService, userService *service.UserService, imageStorage *service.ImageStorageSettingService) *BackupHandler {
return &BackupHandler{
backupService: backupService,
userService: userService,
imageStorage: imageStorage,
}
}
@@ -203,3 +205,48 @@ func (h *BackupHandler) RestoreBackup(c *gin.Context) {
}
response.Accepted(c, record)
}
// ─── 异步生图对象存储配置 ───
//
// 与备份共用一套 S3 客户端构造,因此放在同一个页面下:勾选"复用备份 S3"即可直接
// 借用备份已配置的端点与密钥,只用不同的前缀区分对象(备份走 backups/,图片走 images/)。
func (h *BackupHandler) GetImageStorageConfig(c *gin.Context) {
ctx := c.Request.Context()
cfg, err := h.imageStorage.Get(ctx)
if err != nil {
response.ErrorFrom(c, err)
return
}
response.Success(c, gin.H{
"config": cfg,
"secret_configured": h.imageStorage.SecretConfigured(ctx),
})
}
func (h *BackupHandler) UpdateImageStorageConfig(c *gin.Context) {
var req service.ImageStorageSettings
if err := c.ShouldBindJSON(&req); err != nil {
response.BadRequest(c, "Invalid request: "+err.Error())
return
}
cfg, err := h.imageStorage.Update(c.Request.Context(), req)
if err != nil {
response.ErrorFrom(c, err)
return
}
response.Success(c, cfg)
}
func (h *BackupHandler) TestImageStorageConnection(c *gin.Context) {
var req service.ImageStorageSettings
if err := c.ShouldBindJSON(&req); err != nil {
response.BadRequest(c, "Invalid request: "+err.Error())
return
}
if err := h.imageStorage.TestConnection(c.Request.Context(), req); err != nil {
response.Success(c, gin.H{"ok": false, "message": err.Error()})
return
}
response.Success(c, gin.H{"ok": true, "message": "connection successful"})
}
@@ -0,0 +1,134 @@
//go:build unit
package handler
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
type toggleSettingRepo struct {
mu sync.Mutex
values map[string]string
}
func (r *toggleSettingRepo) Get(context.Context, string) (*service.Setting, error) { return nil, nil }
func (r *toggleSettingRepo) GetValue(_ context.Context, key string) (string, error) {
r.mu.Lock()
defer r.mu.Unlock()
return r.values[key], nil
}
func (r *toggleSettingRepo) Set(_ context.Context, key, value string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.values[key] = value
return nil
}
func (r *toggleSettingRepo) GetMultiple(context.Context, []string) (map[string]string, error) {
return map[string]string{}, nil
}
func (r *toggleSettingRepo) SetMultiple(context.Context, map[string]string) error { return nil }
func (r *toggleSettingRepo) GetAll(context.Context) (map[string]string, error) {
return map[string]string{}, nil
}
func (r *toggleSettingRepo) Delete(context.Context, string) error { return nil }
type passthroughEncryptor struct{}
func (passthroughEncryptor) Encrypt(plaintext string) (string, error) { return plaintext, nil }
func (passthroughEncryptor) Decrypt(ciphertext string) (string, error) { return ciphertext, nil }
type noopImageStorage struct{}
func (noopImageStorage) Save(context.Context, string, string, []byte) (string, error) {
return "https://cdn.example.test/object.png", nil
}
// TestAsyncImageEnablesWithoutRestart drives the actual HTTP path for the bug behind
// #4458 and #4542: with object storage unconfigured the async endpoint 404s, and the
// only way to turn it on used to be editing config.yaml and restarting the container.
// Flipping the admin setting must flip the endpoint over in the same process.
func TestAsyncImageEnablesWithoutRestart(t *testing.T) {
gin.SetMode(gin.TestMode)
repo := &toggleSettingRepo{values: map[string]string{}}
backup := service.NewBackupService(repo, &config.Config{}, passthroughEncryptor{}, nil, nil)
factory := func(context.Context, *config.ImageStorageConfig) (service.ImageStorage, error) {
return noopImageStorage{}, nil
}
settings := service.NewImageStorageSettingService(repo, passthroughEncryptor{}, backup, factory, config.ImageStorageConfig{})
store := &asyncImageMemoryStore{tasks: make(map[string]*service.ImageTaskRecord)}
tasks := service.NewImageTaskServiceWithResolver(store, settings.Resolver(), time.Hour, time.Minute)
h := &AsyncImageHandler{tasks: tasks}
h.execute = func(_ string, c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"created": 1, "data": []gin.H{{"url": "https://upstream.test/i.png"}}})
}
router := gin.New()
router.Use(func(c *gin.Context) {
groupID := int64(3)
c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{
ID: 9, UserID: 7, GroupID: &groupID,
Group: &service.Group{ID: groupID, Platform: service.PlatformOpenAI, AllowImageGeneration: true},
})
c.Next()
})
router.POST("/v1/images/generations/async", h.Submit)
router.GET("/v1/images/tasks/:task_id", h.Get)
submit := func() *httptest.ResponseRecorder {
req := httptest.NewRequest(http.MethodPost, "/v1/images/generations/async",
strings.NewReader(`{"model":"gpt-image-1","prompt":"a lighthouse"}`))
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
return rec
}
rec := submit()
require.Equal(t, http.StatusNotFound, rec.Code, "disabled until an admin configures object storage")
require.Contains(t, rec.Body.String(), "async image tasks are not enabled")
// The admin saves the setting — no restart, same process.
_, err := settings.Update(context.Background(), service.ImageStorageSettings{
Enabled: true, Bucket: "my-images",
Endpoint: "https://acct.r2.cloudflarestorage.com", AccessKeyID: "ak", SecretAccessKey: "sk",
})
require.NoError(t, err)
rec = submit()
require.Equal(t, http.StatusAccepted, rec.Code, "the endpoint must go live as soon as the setting is saved")
var accepted struct {
TaskID string `json:"task_id"`
PollURL string `json:"poll_url"`
}
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &accepted))
require.NotEmpty(t, accepted.TaskID)
// Turning the feature back off must not strand a task that was already accepted.
_, err = settings.Update(context.Background(), service.ImageStorageSettings{Enabled: false})
require.NoError(t, err)
require.Equal(t, http.StatusNotFound, submit().Code, "new submissions are refused again")
pollRec := httptest.NewRecorder()
router.ServeHTTP(pollRec, httptest.NewRequest(http.MethodGet, accepted.PollURL, nil))
require.Equal(t, http.StatusOK, pollRec.Code, "an already-accepted task stays pollable after the switch is turned off")
}
+12 -1
View File
@@ -39,6 +39,13 @@ func (h *AsyncImageHandler) enabled() bool {
return h != nil && h.tasks != nil && h.tasks.Enabled()
}
// pollable reports whether task lookups can be served. It is deliberately weaker
// than enabled(): results already written to Redis stay readable after the
// feature is switched off, so an in-flight task is never stranded.
func (h *AsyncImageHandler) pollable() bool {
return h != nil && h.tasks != nil && h.tasks.Pollable()
}
// Submit accepts the same payload as the synchronous Images endpoint and
// returns before the upstream image generation begins.
func (h *AsyncImageHandler) Submit(c *gin.Context) {
@@ -155,7 +162,11 @@ func (h *AsyncImageHandler) checkSecurityAuditBeforeSubmit(c *gin.Context, apiKe
}
func (h *AsyncImageHandler) Get(c *gin.Context) {
if !h.enabled() {
// Polling deliberately does not require the feature to be enabled, only that
// the task store is reachable. Turning the switch off in the admin UI must not
// strand tasks that were already accepted — their results are still in Redis
// and their submitters are still polling.
if !h.pollable() {
imageTaskJSONError(c, http.StatusNotFound, "not_found_error", "async image tasks are not enabled")
return
}
+8 -11
View File
@@ -141,7 +141,7 @@ var ProviderSet = wire.NewSet(
NewS3BackupStoreFactory,
// Image storage (async image task result offload)
ProvideImageStorage,
ProvideImageStorageFactory,
// HTTP service ports (DI Strategy A: return interface directly)
NewTurnstileVerifier,
@@ -174,17 +174,14 @@ func ProvideEnt(cfg *config.Config) (*ent.Client, error) {
return client, err
}
// ProvideImageStorage 提供异步图片任务结果转存所用的对象存储实现
// 仅当开关打开且 S3 凭证齐全时返回具体实现,否则返回 nil(功能整体禁用)。
func ProvideImageStorage(cfg *config.Config) (service.ImageStorage, error) {
if !cfg.ImageStorage.Active() {
return nil, nil
// ProvideImageStorageFactory 提供按需构造对象存储客户端的工厂
//
// 这里返回工厂而不是实例:异步生图的开关与凭证可以在后台随时改动,客户端必须能在
// 设置保存后重建,而不是在启动时定死一份。
func ProvideImageStorageFactory() service.ImageStorageFactory {
return func(ctx context.Context, cfg *config.ImageStorageConfig) (service.ImageStorage, error) {
return NewS3ImageStorage(ctx, cfg)
}
store, err := NewS3ImageStorage(context.Background(), &cfg.ImageStorage)
if err != nil {
return nil, err
}
return store, nil
}
// ProvideSQLDB 从 Ent 客户端提取底层的 *sql.DB 连接。
+6
View File
@@ -574,6 +574,12 @@ func registerBackupRoutes(admin *gin.RouterGroup, h *handler.Handlers, stepUpAut
backup.PUT("/s3-config", gin.HandlerFunc(stepUpAuth), h.Admin.Backup.UpdateS3Config)
backup.POST("/s3-config/test", h.Admin.Backup.TestS3Connection)
// 异步生图对象存储配置(与备份共用 S3 客户端,可直接复用备份凭证)
backup.GET("/image-storage", h.Admin.Backup.GetImageStorageConfig)
// 同 S3 配置:改写对象存储目标可将生成内容导向外部账号——要求 step-up 2FA
backup.PUT("/image-storage", gin.HandlerFunc(stepUpAuth), h.Admin.Backup.UpdateImageStorageConfig)
backup.POST("/image-storage/test", h.Admin.Backup.TestImageStorageConnection)
// 定时备份配置
backup.GET("/schedule", h.Admin.Backup.GetSchedule)
backup.PUT("/schedule", h.Admin.Backup.UpdateSchedule)
@@ -0,0 +1,342 @@
package service
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
"go.uber.org/zap"
)
const settingKeyImageStorageConfig = "image_storage_config"
// ErrImageStorageIncomplete 表示开关已打开但凭证不全,无法启用异步生图。
var ErrImageStorageIncomplete = errors.New("image storage is enabled but bucket/access_key_id/secret_access_key are incomplete")
// ImageStorageFactory 由 repository 层提供,把配置变成一个可用的对象存储实现。
// 与 BackupObjectStoreFactory 同样的注入方式,避免 service 反向依赖 repository。
type ImageStorageFactory func(ctx context.Context, cfg *config.ImageStorageConfig) (ImageStorage, error)
// ImageStorageSettings 是后台可编辑的异步生图对象存储配置。
//
// ReuseBackupS3 为真时不保存自己的凭证,直接借用数据库备份已配置的 S3 端点与密钥,
// 只用自己的 Bucket/Prefix 区分对象;这样"数据走 backups/、图片走 images/"无需重复配置。
type ImageStorageSettings struct {
Enabled bool `json:"enabled"`
ReuseBackupS3 bool `json:"reuse_backup_s3"`
Bucket string `json:"bucket"` // 留空且复用备份时,沿用备份桶
Prefix string `json:"prefix"`
PublicBaseURL string `json:"public_base_url"`
PresignExpiry int `json:"presign_expiry_hours"`
MaxDownloadBytes int64 `json:"max_download_bytes"`
// 以下仅在 ReuseBackupS3 为假时使用
Endpoint string `json:"endpoint"`
Region string `json:"region"`
AccessKeyID string `json:"access_key_id"`
SecretAccessKey string `json:"secret_access_key,omitempty"` //nolint:revive // field name follows AWS convention
ForcePathStyle bool `json:"force_path_style"`
}
// ImageStorageSettingService 读写后台设置,并把结果解析成一个可直接使用的 uploader。
//
// 解析结果带缓存:网关每次请求都要判断功能是否开启,不能每次都查库。保存设置时调用
// Invalidate 清缓存,下一次请求即重建客户端——这是"后台开关立即生效、无需重启"的实现。
type ImageStorageSettingService struct {
settingRepo SettingRepository
encryptor SecretEncryptor
backup *BackupService
factory ImageStorageFactory
// fallback 是 config.yaml 里的配置。后台从未保存过设置时沿用它,
// 保证升级前已用配置文件开启该功能的部署不被打断。
fallback config.ImageStorageConfig
mu sync.Mutex
resolved bool
uploader *ImageResultUploader
enabled bool
}
func NewImageStorageSettingService(
settingRepo SettingRepository,
encryptor SecretEncryptor,
backup *BackupService,
factory ImageStorageFactory,
fallback config.ImageStorageConfig,
) *ImageStorageSettingService {
return &ImageStorageSettingService{
settingRepo: settingRepo,
encryptor: encryptor,
backup: backup,
factory: factory,
fallback: fallback,
}
}
// Resolver 返回可注入 ImageTaskService 的解析函数。
func (s *ImageStorageSettingService) Resolver() ImageStorageResolver {
return func() (*ImageResultUploader, bool) {
return s.resolve()
}
}
func (s *ImageStorageSettingService) resolve() (*ImageResultUploader, bool) {
if s == nil {
return nil, false
}
s.mu.Lock()
defer s.mu.Unlock()
if s.resolved {
return s.uploader, s.enabled
}
ctx := context.Background()
s.resolved = true
s.uploader, s.enabled = nil, false
cfg, err := s.effectiveConfig(ctx)
if err != nil {
logger.L().Warn("image_storage.settings_load_failed; async image tasks stay disabled", zap.Error(err))
return nil, false
}
if !cfg.Enabled {
return nil, false
}
if !cfg.IsConfigured() {
logger.L().Warn("image_storage is enabled but not fully configured; async image tasks are disabled",
zap.Strings("missing_keys", cfg.MissingCredentialKeys()))
return nil, false
}
storage, err := s.factory(ctx, cfg)
if err != nil {
logger.L().Error("image_storage.client_build_failed; async image tasks stay disabled", zap.Error(err))
return nil, false
}
s.uploader = NewImageResultUploader(storage, cfg.Prefix, cfg.MaxDownloadByte, nil)
s.enabled = true
return s.uploader, true
}
// Invalidate 丢弃缓存,使下一次请求按最新设置重新解析。
func (s *ImageStorageSettingService) Invalidate() {
if s == nil {
return
}
s.mu.Lock()
s.resolved = false
s.uploader = nil
s.enabled = false
s.mu.Unlock()
}
// Get 返回后台设置(SecretAccessKey 已脱敏)。从未保存过时返回 config.yaml 的等价值。
func (s *ImageStorageSettingService) Get(ctx context.Context) (*ImageStorageSettings, error) {
settings, err := s.load(ctx)
if err != nil {
return nil, err
}
if settings == nil {
settings = settingsFromConfig(s.fallback)
}
settings.SecretAccessKey = ""
return settings, nil
}
// SecretConfigured 供前端展示"已配置"占位符。
func (s *ImageStorageSettingService) SecretConfigured(ctx context.Context) bool {
settings, err := s.load(ctx)
if err != nil || settings == nil {
return s.fallback.SecretAccessKey != ""
}
if settings.ReuseBackupS3 {
cfg, err := s.backupCredentials(ctx)
return err == nil && cfg != nil && cfg.SecretAccessKey != ""
}
return settings.SecretAccessKey != ""
}
// Update 保存设置并立即生效。SecretAccessKey 留空表示沿用已保存的值。
func (s *ImageStorageSettingService) Update(ctx context.Context, in ImageStorageSettings) (*ImageStorageSettings, error) {
normalizeImageStorageSettings(&in)
if in.ReuseBackupS3 {
// 复用备份凭证时不落自己的密钥,避免同一份密钥在库里存两份。
in.Endpoint, in.Region, in.AccessKeyID, in.SecretAccessKey = "", "", "", ""
in.ForcePathStyle = false
} else if in.SecretAccessKey == "" {
if old, err := s.load(ctx); err == nil && old != nil {
in.SecretAccessKey = old.SecretAccessKey
}
} else {
encrypted, err := s.encryptor.Encrypt(in.SecretAccessKey)
if err != nil {
return nil, fmt.Errorf("encrypt secret: %w", err)
}
in.SecretAccessKey = encrypted
}
data, err := json.Marshal(in)
if err != nil {
return nil, fmt.Errorf("marshal image storage settings: %w", err)
}
if err := s.settingRepo.Set(ctx, settingKeyImageStorageConfig, string(data)); err != nil {
return nil, fmt.Errorf("save image storage settings: %w", err)
}
s.Invalidate()
in.SecretAccessKey = ""
return &in, nil
}
// TestConnection 用给定设置试建一次客户端,用于后台的"测试连接"按钮。
// 与 Update 一样支持留空 SecretAccessKey 表示沿用已保存的值。
func (s *ImageStorageSettingService) TestConnection(ctx context.Context, in ImageStorageSettings) error {
normalizeImageStorageSettings(&in)
if !in.ReuseBackupS3 && in.SecretAccessKey == "" {
old, err := s.load(ctx)
if err == nil && old != nil {
in.SecretAccessKey = old.SecretAccessKey
}
}
cfg, err := s.toImageStorageConfig(ctx, &in)
if err != nil {
return err
}
if !cfg.IsConfigured() {
return ErrImageStorageIncomplete
}
if _, err := s.factory(ctx, cfg); err != nil {
return err
}
return nil
}
// effectiveConfig 把后台设置(或 config.yaml 回落)解析成运行时配置。
func (s *ImageStorageSettingService) effectiveConfig(ctx context.Context) (*config.ImageStorageConfig, error) {
settings, err := s.load(ctx)
if err != nil {
return nil, err
}
if settings == nil {
fallback := s.fallback
return &fallback, nil
}
return s.toImageStorageConfig(ctx, settings)
}
func (s *ImageStorageSettingService) toImageStorageConfig(ctx context.Context, in *ImageStorageSettings) (*config.ImageStorageConfig, error) {
cfg := &config.ImageStorageConfig{
Enabled: in.Enabled,
Bucket: in.Bucket,
Prefix: in.Prefix,
PublicBaseURL: in.PublicBaseURL,
PresignExpiry: in.PresignExpiry,
MaxDownloadByte: in.MaxDownloadBytes,
Endpoint: in.Endpoint,
Region: in.Region,
AccessKeyID: in.AccessKeyID,
SecretAccessKey: in.SecretAccessKey,
ForcePathStyle: in.ForcePathStyle,
}
if in.ReuseBackupS3 {
backupCfg, err := s.backupCredentials(ctx)
if err != nil {
return nil, err
}
if backupCfg == nil {
return nil, errors.New("image storage is set to reuse the backup S3 configuration, but no backup S3 configuration exists")
}
cfg.Endpoint = backupCfg.Endpoint
cfg.Region = backupCfg.Region
cfg.AccessKeyID = backupCfg.AccessKeyID
cfg.SecretAccessKey = backupCfg.SecretAccessKey
cfg.ForcePathStyle = backupCfg.ForcePathStyle
if cfg.Bucket == "" {
cfg.Bucket = backupCfg.Bucket
}
} else if cfg.SecretAccessKey != "" {
decrypted, err := s.encryptor.Decrypt(cfg.SecretAccessKey)
if err != nil {
// 兼容未加密的旧数据,与备份配置的处理保持一致。
logger.L().Warn("image_storage secret decrypt failed; treating the stored value as plaintext", zap.Error(err))
} else {
cfg.SecretAccessKey = decrypted
}
}
return cfg, nil
}
// backupCredentials 取备份已配置的 S3 凭证(已解密)。
func (s *ImageStorageSettingService) backupCredentials(ctx context.Context) (*BackupS3Config, error) {
if s.backup == nil {
return nil, errors.New("backup service is unavailable")
}
return s.backup.loadS3Config(ctx)
}
// load 读出后台设置;从未保存过时返回 nil。
func (s *ImageStorageSettingService) load(ctx context.Context) (*ImageStorageSettings, error) {
if s.settingRepo == nil {
return nil, nil //nolint:nilnil // no repository means no stored settings
}
raw, err := s.settingRepo.GetValue(ctx, settingKeyImageStorageConfig)
if err != nil || strings.TrimSpace(raw) == "" {
return nil, nil //nolint:nilnil // never configured is a valid state
}
var settings ImageStorageSettings
if err := json.Unmarshal([]byte(raw), &settings); err != nil {
return nil, fmt.Errorf("parse image storage settings: %w", err)
}
return &settings, nil
}
func settingsFromConfig(cfg config.ImageStorageConfig) *ImageStorageSettings {
return &ImageStorageSettings{
Enabled: cfg.Enabled,
Bucket: cfg.Bucket,
Prefix: cfg.Prefix,
PublicBaseURL: cfg.PublicBaseURL,
PresignExpiry: cfg.PresignExpiry,
MaxDownloadBytes: cfg.MaxDownloadByte,
Endpoint: cfg.Endpoint,
Region: cfg.Region,
AccessKeyID: cfg.AccessKeyID,
SecretAccessKey: cfg.SecretAccessKey,
ForcePathStyle: cfg.ForcePathStyle,
}
}
func normalizeImageStorageSettings(in *ImageStorageSettings) {
in.Bucket = strings.TrimSpace(in.Bucket)
in.Endpoint = strings.TrimSpace(in.Endpoint)
in.Region = strings.TrimSpace(in.Region)
in.AccessKeyID = strings.TrimSpace(in.AccessKeyID)
in.SecretAccessKey = strings.TrimSpace(in.SecretAccessKey)
in.PublicBaseURL = strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(in.PublicBaseURL), "/"))
in.Prefix = strings.TrimSpace(in.Prefix)
if in.Prefix == "" {
in.Prefix = "images/"
}
if !strings.HasSuffix(in.Prefix, "/") {
in.Prefix += "/"
}
if in.Region == "" {
in.Region = "auto"
}
if in.PresignExpiry <= 0 {
in.PresignExpiry = 24
}
if in.MaxDownloadBytes <= 0 {
in.MaxDownloadBytes = defaultImageMaxDownloadBytes
}
}
@@ -0,0 +1,220 @@
//go:build unit
package service
import (
"context"
"encoding/json"
"errors"
"strings"
"sync"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/stretchr/testify/require"
)
type stubSettingRepo struct {
mu sync.Mutex
values map[string]string
}
func newStubSettingRepo() *stubSettingRepo {
return &stubSettingRepo{values: map[string]string{}}
}
func (r *stubSettingRepo) Get(context.Context, string) (*Setting, error) { return nil, nil }
func (r *stubSettingRepo) GetValue(_ context.Context, key string) (string, error) {
r.mu.Lock()
defer r.mu.Unlock()
return r.values[key], nil
}
func (r *stubSettingRepo) Set(_ context.Context, key, value string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.values[key] = value
return nil
}
func (r *stubSettingRepo) GetMultiple(context.Context, []string) (map[string]string, error) {
return map[string]string{}, nil
}
func (r *stubSettingRepo) SetMultiple(context.Context, map[string]string) error { return nil }
func (r *stubSettingRepo) GetAll(context.Context) (map[string]string, error) {
return map[string]string{}, nil
}
func (r *stubSettingRepo) Delete(context.Context, string) error { return nil }
// reversibleEncryptor stands in for AES: prefixed so a test can tell ciphertext
// from plaintext, and so decrypting a plaintext value fails like the real one.
type reversibleEncryptor struct{}
func (reversibleEncryptor) Encrypt(plaintext string) (string, error) {
return "enc:" + plaintext, nil
}
func (reversibleEncryptor) Decrypt(ciphertext string) (string, error) {
rest, ok := strings.CutPrefix(ciphertext, "enc:")
if !ok {
return "", errors.New("not encrypted")
}
return rest, nil
}
type recordingStorage struct{ saved []string }
func (s *recordingStorage) Save(_ context.Context, key, _ string, _ []byte) (string, error) {
s.saved = append(s.saved, key)
return "https://cdn.example.com/" + key, nil
}
func newImageStorageFixture(t *testing.T, fallback config.ImageStorageConfig) (*ImageStorageSettingService, *stubSettingRepo, *[]config.ImageStorageConfig) {
t.Helper()
repo := newStubSettingRepo()
encryptor := reversibleEncryptor{}
backup := NewBackupService(repo, &config.Config{}, encryptor, nil, nil)
var built []config.ImageStorageConfig
factory := func(_ context.Context, cfg *config.ImageStorageConfig) (ImageStorage, error) {
built = append(built, *cfg)
return &recordingStorage{}, nil
}
return NewImageStorageSettingService(repo, encryptor, backup, factory, fallback), repo, &built
}
func seedBackupS3(t *testing.T, repo *stubSettingRepo, cfg BackupS3Config) {
t.Helper()
cfg.SecretAccessKey = "enc:" + cfg.SecretAccessKey
data, err := json.Marshal(cfg)
require.NoError(t, err)
require.NoError(t, repo.Set(context.Background(), settingKeyBackupS3Config, string(data)))
}
// The admin switch must take effect without a restart: that is the entire point
// of moving image_storage out of config.yaml (#4542).
func TestImageStorageSettingsToggleTakesEffectWithoutRestart(t *testing.T) {
svc, repo, built := newImageStorageFixture(t, config.ImageStorageConfig{})
ctx := context.Background()
seedBackupS3(t, repo, BackupS3Config{
Endpoint: "https://acct.r2.cloudflarestorage.com", Region: "auto",
Bucket: "backup-bucket", AccessKeyID: "ak", SecretAccessKey: "sk",
Prefix: "backups/",
})
uploader, enabled := svc.resolve()
require.False(t, enabled, "disabled until an admin turns it on")
require.Nil(t, uploader)
_, err := svc.Update(ctx, ImageStorageSettings{Enabled: true, ReuseBackupS3: true})
require.NoError(t, err)
uploader, enabled = svc.resolve()
require.True(t, enabled, "saving the setting must enable the feature immediately")
require.NotNil(t, uploader)
_, err = svc.Update(ctx, ImageStorageSettings{Enabled: false, ReuseBackupS3: true})
require.NoError(t, err)
_, enabled = svc.resolve()
require.False(t, enabled, "turning it back off must also apply immediately")
require.Len(t, *built, 1, "the S3 client is built only when the feature is on")
}
func TestImageStorageSettingsReuseBackupCredentials(t *testing.T) {
svc, repo, built := newImageStorageFixture(t, config.ImageStorageConfig{})
ctx := context.Background()
seedBackupS3(t, repo, BackupS3Config{
Endpoint: "https://acct.r2.cloudflarestorage.com", Region: "wnam",
Bucket: "backup-bucket", AccessKeyID: "backup-ak", SecretAccessKey: "backup-sk",
Prefix: "backups/", ForcePathStyle: true,
})
_, err := svc.Update(ctx, ImageStorageSettings{Enabled: true, ReuseBackupS3: true, Prefix: "images"})
require.NoError(t, err)
_, enabled := svc.resolve()
require.True(t, enabled)
require.Len(t, *built, 1)
got := (*built)[0]
require.Equal(t, "https://acct.r2.cloudflarestorage.com", got.Endpoint)
require.Equal(t, "wnam", got.Region)
require.Equal(t, "backup-ak", got.AccessKeyID)
require.Equal(t, "backup-sk", got.SecretAccessKey, "the backup secret must be decrypted before use")
require.True(t, got.ForcePathStyle)
require.Equal(t, "backup-bucket", got.Bucket, "an empty bucket falls back to the backup bucket")
require.Equal(t, "images/", got.Prefix, "images stay under their own prefix so they never collide with backups/")
// Reusing must not duplicate the secret into a second row.
raw, err := repo.GetValue(ctx, settingKeyImageStorageConfig)
require.NoError(t, err)
require.NotContains(t, raw, "backup-sk")
require.NotContains(t, raw, "enc:")
}
func TestImageStorageSettingsOwnCredentialsAreEncryptedAndMasked(t *testing.T) {
svc, repo, built := newImageStorageFixture(t, config.ImageStorageConfig{})
ctx := context.Background()
saved, err := svc.Update(ctx, ImageStorageSettings{
Enabled: true, Bucket: "my-images",
Endpoint: "https://acct.r2.cloudflarestorage.com",
AccessKeyID: "ak", SecretAccessKey: "super-secret",
})
require.NoError(t, err)
require.Empty(t, saved.SecretAccessKey, "the response must never echo the secret back")
raw, err := repo.GetValue(ctx, settingKeyImageStorageConfig)
require.NoError(t, err)
require.NotContains(t, raw, `"secret_access_key":"super-secret"`, "the secret must be encrypted at rest")
require.Contains(t, raw, "enc:super-secret")
fetched, err := svc.Get(ctx)
require.NoError(t, err)
require.Empty(t, fetched.SecretAccessKey)
require.True(t, svc.SecretConfigured(ctx))
_, enabled := svc.resolve()
require.True(t, enabled)
require.Equal(t, "super-secret", (*built)[0].SecretAccessKey, "the stored secret must be decrypted before use")
// An update that omits the secret keeps the stored one rather than wiping it.
_, err = svc.Update(ctx, ImageStorageSettings{
Enabled: true, Bucket: "my-images",
Endpoint: "https://acct.r2.cloudflarestorage.com", AccessKeyID: "ak",
})
require.NoError(t, err)
svc.resolve()
require.Equal(t, "super-secret", (*built)[1].SecretAccessKey)
}
func TestImageStorageSettingsIncompleteStaysDisabled(t *testing.T) {
svc, _, built := newImageStorageFixture(t, config.ImageStorageConfig{})
ctx := context.Background()
_, err := svc.Update(ctx, ImageStorageSettings{Enabled: true, Bucket: "my-images"})
require.NoError(t, err)
_, enabled := svc.resolve()
require.False(t, enabled, "missing credentials must not enable the feature")
require.Empty(t, *built, "no client is built from an incomplete configuration")
}
// Deployments that already enabled the feature through config.yaml must keep
// working after the setting moves into the database.
func TestImageStorageSettingsFallBackToConfigFile(t *testing.T) {
svc, _, built := newImageStorageFixture(t, config.ImageStorageConfig{
Enabled: true, Endpoint: "https://acct.r2.cloudflarestorage.com", Region: "auto",
Bucket: "yaml-bucket", AccessKeyID: "yaml-ak", SecretAccessKey: "yaml-sk",
Prefix: "images/", MaxDownloadByte: 1024,
})
_, enabled := svc.resolve()
require.True(t, enabled, "config.yaml still enables the feature when nothing is stored yet")
require.Equal(t, "yaml-bucket", (*built)[0].Bucket)
fetched, err := svc.Get(context.Background())
require.NoError(t, err)
require.True(t, fetched.Enabled)
require.Equal(t, "yaml-bucket", fetched.Bucket)
require.Empty(t, fetched.SecretAccessKey)
}
+40 -3
View File
@@ -69,10 +69,17 @@ type ImageTaskStore interface {
Get(ctx context.Context, id string) (*ImageTaskRecord, error)
}
// ImageStorageResolver reports the currently effective object-storage binding.
// It exists so the async image feature can be switched on and off from the admin
// UI without a restart: the wiring below is fixed at startup, but the answer to
// "is object storage configured right now" is re-read (and cached) per call.
type ImageStorageResolver func() (uploader *ImageResultUploader, enabled bool)
type ImageTaskService struct {
store ImageTaskStore
uploader *ImageResultUploader
enabled bool
resolve ImageStorageResolver
ttl time.Duration
executionTimeout time.Duration
}
@@ -100,10 +107,40 @@ func NewImageTaskServiceWithUploader(store ImageTaskStore, uploader *ImageResult
return s
}
// NewImageTaskServiceWithResolver 构造一个由 resolver 决定启用状态的服务:
// 开关与凭证来自后台设置,保存后立即生效,无需重启。
func NewImageTaskServiceWithResolver(store ImageTaskStore, resolve ImageStorageResolver, ttl, executionTimeout time.Duration) *ImageTaskService {
s := NewImageTaskServiceWithOptions(store, ttl, executionTimeout)
s.resolve = resolve
return s
}
// current 返回当前生效的 uploader 与启用状态。
// 注入了 resolver 时以 resolver 为准(后台设置可热切换),否则回落到构造时固定的值。
func (s *ImageTaskService) current() (*ImageResultUploader, bool) {
if s == nil {
return nil, false
}
if s.resolve != nil {
return s.resolve()
}
return s.uploader, s.enabled
}
// Enabled 表示异步图片任务功能是否可用(总开关 + 凭证齐全)。
// 关闭时 handler 直接返回 404,不创建任务、不写 Redis。
func (s *ImageTaskService) Enabled() bool {
return s != nil && s.enabled && s.store != nil
if s == nil || s.store == nil {
return false
}
_, enabled := s.current()
return enabled
}
// Pollable 表示已创建的任务能否被查询。
// 比 Enabled 弱:只要 store 可用即可,从而在功能被关掉后仍能取回进行中的任务结果。
func (s *ImageTaskService) Pollable() bool {
return s != nil && s.store != nil
}
func (s *ImageTaskService) ExecutionTimeout() time.Duration {
@@ -154,8 +191,8 @@ func (s *ImageTaskService) Complete(ctx context.Context, id string, statusCode i
if !json.Valid(result) {
return s.Fail(ctx, id, http.StatusBadGateway, imageTaskErrorJSON("api_error", "upstream returned a non-JSON image response"))
}
if s.uploader != nil {
rewritten, err := s.uploader.Rewrite(ctx, id, result)
if uploader, _ := s.current(); uploader != nil {
rewritten, err := uploader.Rewrite(ctx, id, result)
if err != nil {
// 转存失败不回退存 base64,避免大 blob 撑爆 Redis:直接把任务标记为失败。
logger.L().Error("image_task.offload_failed", zap.String("task_id", id), zap.Error(err))
+25 -14
View File
@@ -522,23 +522,33 @@ func ProvideAPIKeyAuthCacheInvalidator(apiKeyService *APIKeyService) APIKeyAuthC
return apiKeyService
}
// ProvideImageStorageSettingService 构造异步生图对象存储的后台设置服务。
//
// config.yaml 里的 image_storage 作为回落:后台从未保存过设置时沿用它,
// 使升级前已通过配置文件开启该功能的部署不被打断。
func ProvideImageStorageSettingService(
settingRepo SettingRepository,
encryptor SecretEncryptor,
backup *BackupService,
factory ImageStorageFactory,
cfg *config.Config,
) *ImageStorageSettingService {
if cfg.ImageStorage.Enabled && !cfg.ImageStorage.Active() {
// 列出具体缺失的键。若这些键其实已在环境变量里设过,说明它们没被读进来,
// 请确认 setDefaults 中已为其注册默认值(见 config.setEnvReachableDefaults)。
logger.L().Warn("image_storage.enabled is true in config but object storage is not fully configured; configure it in the admin UI or complete the config file",
zap.Strings("missing_keys", cfg.ImageStorage.MissingCredentialKeys()))
}
return NewImageStorageSettingService(settingRepo, encryptor, backup, factory, cfg.ImageStorage)
}
// ProvideImageTaskService 构造异步图片任务服务。
//
// 对象存储是异步图片任务的启用前提:仅当 image_storage 开关打开且凭证齐全时,
// 服务才启用,并挂上把结果转存到对象存储的 uploader;否则功能整体禁用
// 对象存储是异步图片任务的启用前提:仅当开关打开且凭证齐全时功能才可用,否则整体禁用
// handler 返回 404,不创建任务、不写 Redis),从而避免大 base64 结果撑爆 Redis。
func ProvideImageTaskService(store ImageTaskStore, storage ImageStorage, cfg *config.Config) *ImageTaskService {
if !cfg.ImageStorage.Active() {
if cfg.ImageStorage.Enabled {
// 列出具体缺失的键。若这些键其实已在环境变量里设过,说明它们没被读进来,
// 请确认 setDefaults 中已为其注册默认值(见 config.setEnvReachableDefaults)。
logger.L().Warn("image_storage.enabled is true but object storage is not fully configured; async image tasks are disabled",
zap.Strings("missing_keys", cfg.ImageStorage.MissingCredentialKeys()))
}
return NewImageTaskService(store)
}
uploader := NewImageResultUploader(storage, cfg.ImageStorage.Prefix, cfg.ImageStorage.MaxDownloadByte, nil)
return NewImageTaskServiceWithUploader(store, uploader, defaultImageTaskTTL, defaultImageTaskExecutionTimeout)
// 启用状态由 settings 服务在运行时解析,因此后台改开关后无需重启即可生效。
func ProvideImageTaskService(store ImageTaskStore, settings *ImageStorageSettingService) *ImageTaskService {
return NewImageTaskServiceWithResolver(store, settings.Resolver(), defaultImageTaskTTL, defaultImageTaskExecutionTimeout)
}
// ProvideBackupService creates and starts BackupService
@@ -665,6 +675,7 @@ var ProviderSet = wire.NewSet(
NewAdminService,
NewGatewayService,
NewOpenAIGatewayService,
ProvideImageStorageSettingService,
ProvideImageTaskService,
ProvideBatchImageModelPricingResolver,
NewBatchImagePublicService,
+14
View File
@@ -20,6 +20,20 @@ Only OpenAI and Grok groups are supported. Requests use the same JSON or multipa
Asynchronous image tasks are **disabled by default** and gated on object storage. When the switch is off — or the S3 credentials are incomplete — the async endpoints return `404` and never create a task or write to Redis. This is deliberate: without offloading, large `b64_json` results (several MB each, e.g. `gpt-image-1`) would accumulate in Redis and exhaust its memory.
### From the admin UI (recommended)
**Admin → Backup → Async image object storage.** Saving the form takes effect immediately — the object-storage client is rebuilt on the next request, so there is no container restart.
Because the async image storage and the database backup share one S3 client, the form defaults to **reusing the backup S3 configuration**: it borrows the endpoint, region and credentials already configured above and keeps only its own bucket and prefix, so backups stay under `backups/` while images go to `images/`. Leave the bucket empty to use the backup bucket as well. Untick the box to point images at a completely separate account.
Saving requires step-up 2FA when that gate is enabled, for the same reason the backup S3 form does: changing the target redirects generated content to another account.
Turning the switch off stops new submissions but keeps already-accepted tasks pollable, so nothing in flight is stranded.
### From the config file
The admin setting takes precedence. When nothing has ever been saved there, the `image_storage` block in `config.yaml` is used instead, so deployments that enabled the feature before the admin UI existed keep working untouched.
Configure an S3-compatible object store (AWS S3, Cloudflare R2, Aliyun OSS, MinIO, …) in `config.yaml` (all keys also accept the `IMAGE_STORAGE_*` environment overrides):
```yaml