mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-11 17:56:37 +08:00
refactor(storage): update Backend.Put to return PutResult and encapsulate bucket mapping
- Update Backend.Put method signature in storage.go to return (PutResult, error). - Adjust all backend implementations (local, oss, s3, webdav) to return a PutResult enclosing Key and Bucket. - Refactor storeUploadFile in upload routers.go to extract key/bucket from PutResult, eliminating manual config bucket lookups. - Remove the unused cfgBucket helper from storage_ops.go. - Adjust storage_migration_task.go and tests to accommodate the updated method signature.
This commit is contained in:
@@ -404,22 +404,13 @@ func storeUploadFile(ctx context.Context, subPath string, size int64, mimeType s
|
|||||||
logger.ErrorF(ctx, "初始化活动存储失败: %v", err)
|
logger.ErrorF(ctx, "初始化活动存储失败: %v", err)
|
||||||
return "", "", ErrSaveFileFailed
|
return "", "", ErrSaveFileFailed
|
||||||
}
|
}
|
||||||
storedPath, err := backend.Put(ctx, subPath, bytes.NewReader(buf.Bytes()), size, mimeType)
|
result, err := backend.Put(ctx, subPath, bytes.NewReader(buf.Bytes()), size, mimeType)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.ErrorF(ctx, "写入 %s 存储失败: %v", driver, err)
|
logger.ErrorF(ctx, "写入 %s 存储失败: %v", driver, err)
|
||||||
return "", "", ErrSaveFileFailed
|
return "", "", ErrSaveFileFailed
|
||||||
}
|
}
|
||||||
switch driver {
|
meta.Bucket = result.Bucket
|
||||||
case storage.DriverS3:
|
return string(driver), result.Key, ""
|
||||||
meta.Bucket = cfgBucket(ctx, driver)
|
|
||||||
case storage.DriverR2:
|
|
||||||
meta.Bucket = cfgBucket(ctx, driver)
|
|
||||||
case storage.DriverMinIO:
|
|
||||||
meta.Bucket = cfgBucket(ctx, driver)
|
|
||||||
case storage.DriverOSS:
|
|
||||||
meta.Bucket = cfgBucket(ctx, driver)
|
|
||||||
}
|
|
||||||
return string(driver), storedPath, ""
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// isImageExtension 判断文件扩展名是否属于常见图片格式
|
// isImageExtension 判断文件扩展名是否属于常见图片格式
|
||||||
|
|||||||
@@ -287,7 +287,7 @@ func migrateObjects(
|
|||||||
}
|
}
|
||||||
return fmt.Errorf("open source object %q: %w", obj.FilePath, err)
|
return fmt.Errorf("open source object %q: %w", obj.FilePath, err)
|
||||||
}
|
}
|
||||||
targetPath, putErr := targetBackend.Put(ctx, obj.FilePath, source.Body, obj.FileSize, obj.MimeType)
|
targetResult, putErr := targetBackend.Put(ctx, obj.FilePath, source.Body, obj.FileSize, obj.MimeType)
|
||||||
closeErr := source.Body.Close()
|
closeErr := source.Body.Close()
|
||||||
if putErr != nil {
|
if putErr != nil {
|
||||||
return fmt.Errorf("copy object %q: %w", obj.FilePath, putErr)
|
return fmt.Errorf("copy object %q: %w", obj.FilePath, putErr)
|
||||||
@@ -298,7 +298,7 @@ func migrateObjects(
|
|||||||
|
|
||||||
// Data integrity check (SHA-256 hash verification)
|
// Data integrity check (SHA-256 hash verification)
|
||||||
if len(obj.Hash) == sha256HexLength {
|
if len(obj.Hash) == sha256HexLength {
|
||||||
targetObj, getErr := targetBackend.Get(ctx, targetPath)
|
targetObj, getErr := targetBackend.Get(ctx, targetResult.Key)
|
||||||
if getErr != nil {
|
if getErr != nil {
|
||||||
return fmt.Errorf("retrieve target object for verification %q: %w", obj.FilePath, getErr)
|
return fmt.Errorf("retrieve target object for verification %q: %w", obj.FilePath, getErr)
|
||||||
}
|
}
|
||||||
@@ -318,7 +318,7 @@ func migrateObjects(
|
|||||||
Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath).
|
Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath).
|
||||||
Updates(map[string]any{
|
Updates(map[string]any{
|
||||||
"storage_driver": targetDriver,
|
"storage_driver": targetDriver,
|
||||||
"file_path": targetPath,
|
"file_path": targetResult.Key,
|
||||||
}).Error; err != nil {
|
}).Error; err != nil {
|
||||||
return fmt.Errorf("update migrated object %q: %w", obj.FilePath, err)
|
return fmt.Errorf("update migrated object %q: %w", obj.FilePath, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -67,21 +67,4 @@ func currentMigrationTargetConfig(ctx context.Context) (storage.Config, bool, er
|
|||||||
return target, true, nil
|
return target, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func cfgBucket(ctx context.Context, driver storage.Driver) string {
|
|
||||||
cfg, err := storage.LoadConfig(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
switch driver {
|
|
||||||
case storage.DriverS3:
|
|
||||||
return cfg.S3.Bucket
|
|
||||||
case storage.DriverR2:
|
|
||||||
return cfg.R2.Bucket
|
|
||||||
case storage.DriverMinIO:
|
|
||||||
return cfg.MinIO.Bucket
|
|
||||||
case storage.DriverOSS:
|
|
||||||
return cfg.OSS.Bucket
|
|
||||||
default:
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -25,13 +25,13 @@ func newLocalBackend(cfg LocalConfig) (*localBackend, error) {
|
|||||||
return &localBackend{root: root}, nil
|
return &localBackend{root: root}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *localBackend) Put(_ context.Context, key string, body io.Reader, _ int64, _ string) (string, error) {
|
func (b *localBackend) Put(_ context.Context, key string, body io.Reader, _ int64, _ string) (PutResult, error) {
|
||||||
path, err := b.path(key)
|
path, err := b.path(key)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
if err := os.MkdirAll(filepath.Dir(path), storageDirPerm); err != nil {
|
if err := os.MkdirAll(filepath.Dir(path), storageDirPerm); err != nil {
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
file, err := os.OpenFile( //nolint:gosec // path is constrained to the configured storage root.
|
file, err := os.OpenFile( //nolint:gosec // path is constrained to the configured storage root.
|
||||||
path,
|
path,
|
||||||
@@ -39,18 +39,18 @@ func (b *localBackend) Put(_ context.Context, key string, body io.Reader, _ int6
|
|||||||
storageFilePerm,
|
storageFilePerm,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
if _, err := io.Copy(file, body); err != nil {
|
if _, err := io.Copy(file, body); err != nil {
|
||||||
_ = file.Close()
|
_ = file.Close()
|
||||||
_ = os.Remove(path)
|
_ = os.Remove(path)
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
if err := file.Close(); err != nil {
|
if err := file.Close(); err != nil {
|
||||||
_ = os.Remove(path)
|
_ = os.Remove(path)
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
return filepath.ToSlash(key), nil
|
return PutResult{Key: filepath.ToSlash(key)}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *localBackend) Get(_ context.Context, key string) (*Object, error) {
|
func (b *localBackend) Get(_ context.Context, key string) (*Object, error) {
|
||||||
|
|||||||
@@ -19,12 +19,12 @@ func TestLocalBackendRoundTrip(t *testing.T) {
|
|||||||
const key = "uploads/2026/06/13/test.txt"
|
const key = "uploads/2026/06/13/test.txt"
|
||||||
const content = "wavelet storage"
|
const content = "wavelet storage"
|
||||||
|
|
||||||
storedKey, err := backend.Put(ctx, key, bytes.NewBufferString(content), int64(len(content)), "text/plain")
|
storedResult, err := backend.Put(ctx, key, bytes.NewBufferString(content), int64(len(content)), "text/plain")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Put(%q) returned error: %v", key, err)
|
t.Fatalf("Put(%q) returned error: %v", key, err)
|
||||||
}
|
}
|
||||||
if storedKey != key {
|
if storedResult.Key != key {
|
||||||
t.Errorf("Put(%q) key = %q, want %q", key, storedKey, key)
|
t.Errorf("Put(%q) key = %q, want %q", key, storedResult.Key, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
object, err := backend.Get(ctx, key)
|
object, err := backend.Get(ctx, key)
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ func newOSSBackend(cfg ObjectConfig) (*ossBackend, error) {
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *ossBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, _ string) (string, error) {
|
func (b *ossBackend) Put(ctx context.Context, key string, body io.Reader, _ int64, _ string) (PutResult, error) {
|
||||||
key = b.key(key)
|
key = b.key(key)
|
||||||
_, err := b.client.PutObject(ctx, &oss.PutObjectRequest{
|
_, err := b.client.PutObject(ctx, &oss.PutObjectRequest{
|
||||||
Bucket: oss.Ptr(b.bucket),
|
Bucket: oss.Ptr(b.bucket),
|
||||||
@@ -43,9 +43,9 @@ func (b *ossBackend) Put(ctx context.Context, key string, body io.Reader, _ int6
|
|||||||
Body: body,
|
Body: body,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("put OSS object: %w", err)
|
return PutResult{}, fmt.Errorf("put OSS object: %w", err)
|
||||||
}
|
}
|
||||||
return key, nil
|
return PutResult{Key: key, Bucket: b.bucket}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *ossBackend) Get(ctx context.Context, key string) (*Object, error) {
|
func (b *ossBackend) Get(ctx context.Context, key string) (*Object, error) {
|
||||||
|
|||||||
@@ -56,7 +56,7 @@ func newR2Backend(ctx context.Context, cfg ObjectConfig) (*s3Backend, error) {
|
|||||||
return newS3Backend(ctx, cfg)
|
return newS3Backend(ctx, cfg)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *s3Backend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error) {
|
func (b *s3Backend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (PutResult, error) {
|
||||||
key = b.key(key)
|
key = b.key(key)
|
||||||
_, err := b.client.PutObject(ctx, &s3.PutObjectInput{
|
_, err := b.client.PutObject(ctx, &s3.PutObjectInput{
|
||||||
Bucket: aws.String(b.bucket),
|
Bucket: aws.String(b.bucket),
|
||||||
@@ -66,9 +66,9 @@ func (b *s3Backend) Put(ctx context.Context, key string, body io.Reader, size in
|
|||||||
ContentType: aws.String(contentType),
|
ContentType: aws.String(contentType),
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("put S3 object: %w", err)
|
return PutResult{}, fmt.Errorf("put S3 object: %w", err)
|
||||||
}
|
}
|
||||||
return key, nil
|
return PutResult{Key: key, Bucket: b.bucket}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *s3Backend) Get(ctx context.Context, key string) (*Object, error) {
|
func (b *s3Backend) Get(ctx context.Context, key string) (*Object, error) {
|
||||||
|
|||||||
@@ -32,9 +32,15 @@ type Object struct {
|
|||||||
ContentType string
|
ContentType string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// PutResult describes the result of a successful Put operation.
|
||||||
|
type PutResult struct {
|
||||||
|
Key string
|
||||||
|
Bucket string
|
||||||
|
}
|
||||||
|
|
||||||
// Backend defines storage operations used by the upload domain.
|
// Backend defines storage operations used by the upload domain.
|
||||||
type Backend interface {
|
type Backend interface {
|
||||||
Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error)
|
Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (PutResult, error)
|
||||||
Get(ctx context.Context, key string) (*Object, error)
|
Get(ctx context.Context, key string) (*Object, error)
|
||||||
Delete(ctx context.Context, key string) error
|
Delete(ctx context.Context, key string) error
|
||||||
Test(ctx context.Context) error
|
Test(ctx context.Context) error
|
||||||
@@ -173,11 +179,11 @@ type functionBackend struct {
|
|||||||
delete func(context.Context, string) error
|
delete func(context.Context, string) error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *functionBackend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (string, error) {
|
func (b *functionBackend) Put(ctx context.Context, key string, body io.Reader, size int64, contentType string) (PutResult, error) {
|
||||||
if err := b.put(ctx, key, body, size, contentType); err != nil {
|
if err := b.put(ctx, key, body, size, contentType); err != nil {
|
||||||
return "", err
|
return PutResult{}, err
|
||||||
}
|
}
|
||||||
return key, nil
|
return PutResult{Key: key}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *functionBackend) Get(ctx context.Context, key string) (*Object, error) {
|
func (b *functionBackend) Get(ctx context.Context, key string) (*Object, error) {
|
||||||
|
|||||||
@@ -28,17 +28,17 @@ func newWebDAVBackend(cfg WebDAVConfig) (*webDAVBackend, error) {
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *webDAVBackend) Put(_ context.Context, key string, body io.Reader, size int64, _ string) (string, error) {
|
func (b *webDAVBackend) Put(_ context.Context, key string, body io.Reader, size int64, _ string) (PutResult, error) {
|
||||||
key = b.key(key)
|
key = b.key(key)
|
||||||
if dir := path.Dir(key); dir != "." && dir != "/" {
|
if dir := path.Dir(key); dir != "." && dir != "/" {
|
||||||
if err := b.client.MkdirAll(dir, storageDirPerm); err != nil {
|
if err := b.client.MkdirAll(dir, storageDirPerm); err != nil {
|
||||||
return "", fmt.Errorf("create WebDAV directory: %w", err)
|
return PutResult{}, fmt.Errorf("create WebDAV directory: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := b.client.WriteStreamWithLength(key, body, size, storageFilePerm); err != nil {
|
if err := b.client.WriteStreamWithLength(key, body, size, storageFilePerm); err != nil {
|
||||||
return "", fmt.Errorf("put WebDAV object: %w", err)
|
return PutResult{}, fmt.Errorf("put WebDAV object: %w", err)
|
||||||
}
|
}
|
||||||
return key, nil
|
return PutResult{Key: key}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (b *webDAVBackend) Get(_ context.Context, key string) (*Object, error) {
|
func (b *webDAVBackend) Get(_ context.Context, key string) (*Object, error) {
|
||||||
|
|||||||
Reference in New Issue
Block a user