feat(infra): support uploading files via io.Reader

Author: fanlv

backend/domain/knowledge/service/retrieve_test.go

@@ -24,6 +24,11 @@ import (
 	"testing"
 
 	"github.com/cloudwego/eino/schema"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/mock/gomock"
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+
 	"github.com/coze-dev/coze-studio/backend/domain/knowledge/entity"
 	"github.com/coze-dev/coze-studio/backend/domain/knowledge/internal/dal/model"
 	"github.com/coze-dev/coze-studio/backend/domain/knowledge/repository"
@@ -35,10 +40,6 @@ import (
 	mock_db "github.com/coze-dev/coze-studio/backend/internal/mock/infra/contract/rdb"
 	"github.com/coze-dev/coze-studio/backend/pkg/lang/ptr"
 	"github.com/coze-dev/coze-studio/backend/pkg/lang/sets"
-	"github.com/stretchr/testify/assert"
-	"go.uber.org/mock/gomock"
-	"gorm.io/driver/mysql"
-	"gorm.io/gorm"
 )
 
 func TestAddSliceIdColumn(t *testing.T) {

backend/infra/contract/storage/option.go

@@ -38,6 +38,7 @@ type PutOption struct {
 	ContentDisposition *string
 	ContentLanguage    *string
 	Expires            *time.Time
+	ObjectSize         int64
 }
 
 type PutOptFn func(option *PutOption)
@@ -48,6 +49,12 @@ func WithContentType(v string) PutOptFn {
 	}
 }
 
+func WithObjectSize(v int64) PutOptFn {
+	return func(o *PutOption) {
+		o.ObjectSize = v
+	}
+}
+
 func WithContentEncoding(v string) PutOptFn {
 	return func(o *PutOption) {
 		o.ContentEncoding = &v

backend/infra/contract/storage/storage.go

@@ -16,11 +16,15 @@
 
 package storage
 
-import "context"
+import (
+	"context"
+	"io"
+)
 
 //go:generate  mockgen -destination ../../../internal/mock/infra/contract/storage/storage_mock.go -package mock -source storage.go Factory
 type Storage interface {
 	PutObject(ctx context.Context, objectKey string, content []byte, opts ...PutOptFn) error
+	PutObjectWithReader(ctx context.Context, objectKey string, content io.Reader, opts ...PutOptFn) error
 	GetObject(ctx context.Context, objectKey string) ([]byte, error)
 	DeleteObject(ctx context.Context, objectKey string) error
 	GetObjectUrl(ctx context.Context, objectKey string, opts ...GetOptFn) (string, error)

backend/infra/impl/storage/minio/minio.go

@@ -138,6 +138,11 @@ func (m *minioClient) test() {
 }
 
 func (m *minioClient) PutObject(ctx context.Context, objectKey string, content []byte, opts ...storage.PutOptFn) error {
+	opts = append(opts, storage.WithObjectSize(int64(len(content))))
+	return m.PutObjectWithReader(ctx, objectKey, bytes.NewReader(content), opts...)
+}
+
+func (m *minioClient) PutObjectWithReader(ctx context.Context, objectKey string, content io.Reader, opts ...storage.PutOptFn) error {
 	option := storage.PutOption{}
 	for _, opt := range opts {
 		opt(&option)
@@ -165,7 +170,7 @@ func (m *minioClient) PutObject(ctx context.Context, objectKey string, content [
 	}
 
 	_, err := m.client.PutObject(ctx, m.bucketName, objectKey,
-		bytes.NewReader(content), int64(len(content)), minioOpts)
+		content, option.ObjectSize, minioOpts)
 	if err != nil {
 		return fmt.Errorf("PutObject failed: %v", err)
 	}

backend/infra/impl/storage/s3/s3.go

@@ -152,15 +152,18 @@ func (t *s3Client) CheckAndCreateBucket(ctx context.Context) error {
 }
 
 func (t *s3Client) PutObject(ctx context.Context, objectKey string, content []byte, opts ...storage.PutOptFn) error {
+	return t.PutObjectWithReader(ctx, objectKey, bytes.NewReader(content), opts...)
+}
+
+func (t *s3Client) PutObjectWithReader(ctx context.Context, objectKey string, content io.Reader, opts ...storage.PutOptFn) error {
 	client := t.client
-	body := bytes.NewReader(content)
 	bucket := t.bucketName
 
 	// upload object
 	_, err := client.PutObject(ctx, &s3.PutObjectInput{
 		Bucket: aws.String(bucket),
 		Key:    aws.String(objectKey),
-		Body:   body,
+		Body:   content,
 	})
 	return err
 }

backend/infra/impl/storage/tos/tos.go

@@ -140,18 +140,20 @@ func (t *tosClient) CheckAndCreateBucket(ctx context.Context) error {
 
 	return err
 }
-
 func (t *tosClient) PutObject(ctx context.Context, objectKey string, content []byte, opts ...storage.PutOptFn) error {
+	return t.PutObjectWithReader(ctx, objectKey, bytes.NewReader(content), opts...)
+}
+
+func (t *tosClient) PutObjectWithReader(ctx context.Context, objectKey string, content io.Reader, opts ...storage.PutOptFn) error {
 	client := t.client
-	body := bytes.NewReader(content)
 	bucketName := t.bucketName
 
 	_, err := client.PutObjectV2(ctx, &tos.PutObjectV2Input{
 		PutObjectBasicInput: tos.PutObjectBasicInput{
 			Bucket: bucketName,
 			Key:    objectKey,
 		},
-		Content: body,
+		Content: content,
 	})
 
 	// logs.CtxDebugf(ctx, "PutObject resp: %v, err: %v", conv.DebugJsonToStr(output), err)