fix(media): 资源分发 - 统一 CDN 读取签名与源站降级边界
This commit is contained in:
1 parent
62dd62a96c
commit
6c013ebf7e
6 files changed
+114
-15
No files matched your search
@@ -15,6 +15,7 @@ import (
|
||||
_ "image/png"
|
||||
"infinite-canvas/backend/internal/kernel"
|
||||
"io"
|
||||
"log"
|
||||
"mime"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
@@ -324,7 +325,7 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran
|
||||
setting.Provider = firstNonEmpty(resource.Provider, setting.Provider)
|
||||
setting.Endpoint = firstNonEmpty(resource.Endpoint, setting.Endpoint)
|
||||
setting.Bucket = firstNonEmpty(resource.Bucket, setting.Bucket)
|
||||
stream, err := getOSSObjectRange(setting, resource.ObjectKey, normalizeSingleByteRange(rangeHeader))
|
||||
stream, err := getOriginOSSObjectRange(setting, resource.ObjectKey, normalizeSingleByteRange(rangeHeader))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -432,6 +433,9 @@ func writeLocalResourceObject(filePath string, body io.Reader) error {
|
||||
}
|
||||
|
||||
// A cloud write is successful only after the configured origin accepts it.
|
||||
// storeResourceObject writes to the configured origin and degrades to local storage
|
||||
// when the external origin is unavailable. The resource binding is rewritten before
|
||||
// the caller persists the ready state, so later reads follow the actual object location.
|
||||
func (s *Service) storeResourceObject(resource *model.Resource, fileName string, body io.Reader) (string, error) {
|
||||
if resource == nil {
|
||||
return "", errors.New("资源不存在")
|
||||
@@ -439,11 +443,36 @@ func (s *Service) storeResourceObject(resource *model.Resource, fileName string,
|
||||
if resource.Provider == "local" {
|
||||
return "", writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(resource.ObjectKey)), body)
|
||||
}
|
||||
setting, err := s.ossSettingForResource(resource.UserID, resource)
|
||||
if err != nil {
|
||||
return "", err
|
||||
setting, settingErr := s.ossSettingForResource(resource.UserID, resource)
|
||||
var etag string
|
||||
var putErr error
|
||||
if settingErr == nil {
|
||||
etag, putErr = putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
}
|
||||
return putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
if putErr == nil && settingErr == nil {
|
||||
return etag, nil
|
||||
}
|
||||
fallbackErr := putErr
|
||||
if fallbackErr == nil {
|
||||
fallbackErr = settingErr
|
||||
}
|
||||
if seeker, ok := body.(io.Seeker); ok {
|
||||
if _, seekErr := seeker.Seek(0, io.SeekStart); seekErr != nil {
|
||||
return "", errors.Join(fallbackErr, fmt.Errorf("降级本地存储时重置读取位置失败:%w", seekErr))
|
||||
}
|
||||
}
|
||||
localKey := localObjectKey(resource.UserID, resource.Kind, fileName, resource.MimeType, time.Now())
|
||||
resource.Provider = "local"
|
||||
resource.ObjectKey = localKey
|
||||
resource.Endpoint = ""
|
||||
resource.Bucket = ""
|
||||
resource.StorageSettingID = ""
|
||||
resource.ETag = ""
|
||||
if localErr := writeLocalResourceObject(filepath.Join(s.dataDir, "resources", filepath.FromSlash(localKey)), body); localErr != nil {
|
||||
return "", errors.Join(fallbackErr, fmt.Errorf("降级本地存储失败:%w", localErr))
|
||||
}
|
||||
log.Printf("object storage upload degraded to local storage: resource=%s error=%v", resource.ID, fallbackErr)
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func (s *Service) retryStoredResource(userID string, resource *model.Resource, kind string, mimeType string, size int64, body io.Reader) (*model.Resource, error) {
|
||||
|
||||
@@ -35,6 +35,7 @@ func newResourceFallbackTestService(t *testing.T) (*Service, *gorm.DB) {
|
||||
// 配置了 OSS(enabled=1、endpoint 不可达)时上传应回退本地存储成功,
|
||||
// 而不是把上传标记为失败——本地媒体导入不应因外部存储故障整体失败。
|
||||
func TestStoreResourceDegradesToLocalWhenOSSUnavailable(t *testing.T) {
|
||||
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
|
||||
service, db := newResourceFallbackTestService(t)
|
||||
seedOSSEnabled(t, db, "user-1", "http://127.0.0.1:1")
|
||||
|
||||
|
||||
@@ -731,8 +731,9 @@ func TestNormalizeSingleByteRange(t *testing.T) {
|
||||
func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
|
||||
svc := newResourceTestService(t)
|
||||
settingJSON, _ := json.Marshal(ossSettingValue{
|
||||
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", Bucket: "private-bucket",
|
||||
AccessKeyID: "access-id", AccessKeySecret: "secret-value",
|
||||
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", CDNBaseURL: "https://media.example.com",
|
||||
Bucket: "private-bucket", AccessKeyID: "access-id", AccessKeySecret: "secret-value",
|
||||
Delivery: storage.DeliverySettings{CDNAuthMode: "public"},
|
||||
})
|
||||
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -749,7 +750,7 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
|
||||
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{requireURL: true}); err != nil {
|
||||
t.Fatalf("hydrateProviderMedia() error = %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(media.URL, "https://private-bucket.oss-cn-test.aliyuncs.com/") || media.DataURL != "" || !strings.Contains(media.URL, "Signature=") {
|
||||
if media.URL != "https://media.example.com/users/user-1/image/reference.png" || media.DataURL != "" {
|
||||
t.Fatalf("media = %#v", media)
|
||||
}
|
||||
if err := svc.hydrateProviderMedia("other-user", &providerMedia{StorageKey: "resource:resource-1"}, providerMediaHydrationPolicy{requireURL: true}); err == nil {
|
||||
@@ -759,10 +760,11 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedOSSURL(t *testing.T) {
|
||||
|
||||
func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
|
||||
t.Setenv("CANVAS_ALLOW_PRIVATE_UPSTREAMS", "true")
|
||||
t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
|
||||
server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
|
||||
defer server.Close()
|
||||
svc := newResourceTestService(t)
|
||||
settingJSON, _ := json.Marshal(ossSettingValue{Provider: "aliyun", PublicBaseURL: server.URL})
|
||||
settingJSON, _ := json.Marshal(ossSettingValue{Provider: "aliyun", PublicBaseURL: "https://127.0.0.1"})
|
||||
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -774,7 +776,7 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
|
||||
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{requireURL: true}); err != nil {
|
||||
t.Fatalf("hydrateProviderMedia() error = %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(media.URL, server.URL+"/api/public/resources/resource-local/file/resource-local.png?") || !strings.Contains(media.URL, "signature=") || media.DataURL != "" {
|
||||
if !strings.HasPrefix(media.URL, "https://127.0.0.1/api/public/resources/resource-local/file?") || !strings.Contains(media.URL, "signature=") || media.DataURL != "" {
|
||||
t.Fatalf("media = %#v", media)
|
||||
}
|
||||
stored, err := svc.repo.Resource("resource-local")
|
||||
@@ -786,8 +788,9 @@ func TestHydrateNewAPIChannel1ResourceUsesSignedLocalURL(t *testing.T) {
|
||||
func TestHydratePreferredURLUsesObjectStorageAndFallsBackLocal(t *testing.T) {
|
||||
svc := newResourceTestService(t)
|
||||
settingJSON, _ := json.Marshal(ossSettingValue{
|
||||
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", Bucket: "private-bucket",
|
||||
AccessKeyID: "access-id", AccessKeySecret: "secret-value",
|
||||
Enabled: true, Provider: "aliyun", Endpoint: "https://oss-cn-test.aliyuncs.com", CDNBaseURL: "https://media.example.com",
|
||||
Bucket: "private-bucket", AccessKeyID: "access-id", AccessKeySecret: "secret-value",
|
||||
Delivery: storage.DeliverySettings{CDNAuthMode: "public"},
|
||||
})
|
||||
if err := svc.repo.SaveSystemSetting(&model.SystemSetting{Key: ossSettingKey, ValueJSON: string(settingJSON)}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -804,7 +807,7 @@ func TestHydratePreferredURLUsesObjectStorageAndFallsBackLocal(t *testing.T) {
|
||||
if err := svc.hydrateProviderMedia("user-1", &media, providerMediaHydrationPolicy{preferURL: true}); err != nil {
|
||||
t.Fatalf("hydrateProviderMedia(prefer object) error = %v", err)
|
||||
}
|
||||
if !strings.HasPrefix(media.URL, "https://private-bucket.oss-cn-test.aliyuncs.com/") || media.DataURL != "" || !strings.Contains(media.URL, "Signature=") {
|
||||
if media.URL != "https://media.example.com/users/user-1/image/prefer.png" || media.DataURL != "" {
|
||||
t.Fatalf("object media = %#v", media)
|
||||
}
|
||||
|
||||
|
||||
@@ -12,6 +12,8 @@ var (
|
||||
getOSSObjectRange = storage.GetOSSObjectRange
|
||||
getAliyunOSSObjectRange = storage.GetAliyunOSSObjectRange
|
||||
signedOSSObjectURL = storage.SignedOSSObjectURL
|
||||
signedOriginObjectURL = storage.SignedOriginObjectURL
|
||||
getOriginOSSObjectRange = storage.GetOriginObjectRange
|
||||
signedAliyunOSSObjectURL = storage.SignedAliyunOSSObjectURL
|
||||
putCOSObject = storage.PutCOSObject
|
||||
putQiniuObject = storage.PutQiniuObject
|
||||
|
||||
Reference in new issue
Block a user