feat(drivers/s3): support direct multipart upload

drivers/s3/driver.go

@@ -224,11 +224,25 @@ func (d *S3) GetDirectUploadTools() []string {
 	return []string{"HttpDirect"}
 }
 
-func (d *S3) GetDirectUploadInfo(ctx context.Context, _ string, dstDir model.Obj, fileName string, _ int64) (any, error) {
+func (d *S3) GetDirectUploadInfo(ctx context.Context, _ string, dstDir model.Obj, fileName string, fileSize int64) (any, error) {
 	if !d.EnableDirectUpload {
 		return nil, errs.NotImplement
 	}
+	maxParts := d.DirectUploadMaxParts
+	if maxParts == 0 {
+		maxParts = maxCopyParts
+	}
+	return d.getDirectUploadInfo(ctx, dstDir, fileName, fileSize, maxParts, d.DirectUploadMinPartSize)
+}
+
+func (d *S3) getDirectUploadInfo(ctx context.Context, dstDir model.Obj, fileName string, fileSize, maxParts, chunkSize int64) (any, error) {
 	path := getKey(stdpath.Join(dstDir.GetPath(), fileName), false)
+	if maxParts > 1 && fileSize > minMultipartUploadPartSize {
+		return d.getMultipartDirectUploadInfo(ctx, path, fileSize, maxParts, chunkSize)
+	}
+	if fileSize > maxMultipartUploadPartSize {
+		return nil, fmt.Errorf("object size %d exceeds direct upload limit", fileSize)
+	}
 	req, _ := d.directUploadClient.PutObjectRequest(&s3.PutObjectInput{
 		Bucket: &d.Bucket,
 		Key:    &path,
@@ -246,6 +260,83 @@ func (d *S3) GetDirectUploadInfo(ctx context.Context, _ string, dstDir model.Obj
 	}, nil
 }
 
+func (d *S3) getMultipartDirectUploadInfo(ctx context.Context, key string, fileSize, maxParts, chunkSize int64) (*model.S3MultipartDirectUploadInfo, error) {
+	partSize, err := getMultipartUploadPartSize(fileSize, maxParts, chunkSize)
+	if err != nil {
+		return nil, err
+	}
+	created, err := d.directUploadClient.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
+		Bucket: &d.Bucket,
+		Key:    &key,
+	})
+	if err != nil {
+		return nil, err
+	}
+	uploadID := aws.StringValue(created.UploadId)
+	if uploadID == "" {
+		return nil, fmt.Errorf("create multipart upload returned an empty upload ID")
+	}
+	createdSuccessfully := false
+	defer func() {
+		if createdSuccessfully {
+			return
+		}
+		_, _ = d.directUploadClient.AbortMultipartUploadWithContext(context.WithoutCancel(ctx), &s3.AbortMultipartUploadInput{
+			Bucket:   &d.Bucket,
+			Key:      &key,
+			UploadId: &uploadID,
+		})
+	}()
+
+	partCount := (fileSize + partSize - 1) / partSize
+	uploadURLs := make([]string, 0, partCount)
+	for partNumber := int64(1); partNumber <= partCount; partNumber++ {
+		req, _ := d.directUploadClient.UploadPartRequest(&s3.UploadPartInput{
+			Bucket:     &d.Bucket,
+			Key:        &key,
+			PartNumber: &partNumber,
+			UploadId:   &uploadID,
+		})
+		if req == nil {
+			return nil, fmt.Errorf("failed to create multipart upload request for part %d", partNumber)
+		}
+		url, err := req.Presign(time.Hour * time.Duration(d.SignURLExpire))
+		if err != nil {
+			return nil, err
+		}
+		uploadURLs = append(uploadURLs, url)
+	}
+	completeReq, _ := d.directUploadClient.CompleteMultipartUploadRequest(&s3.CompleteMultipartUploadInput{
+		Bucket:   &d.Bucket,
+		Key:      &key,
+		UploadId: &uploadID,
+	})
+	abortReq, _ := d.directUploadClient.AbortMultipartUploadRequest(&s3.AbortMultipartUploadInput{
+		Bucket:   &d.Bucket,
+		Key:      &key,
+		UploadId: &uploadID,
+	})
+	if completeReq == nil || abortReq == nil {
+		return nil, fmt.Errorf("failed to create multipart completion requests")
+	}
+	completeURL, err := completeReq.Presign(time.Hour * time.Duration(d.SignURLExpire))
+	if err != nil {
+		return nil, err
+	}
+	abortURL, err := abortReq.Presign(time.Hour * time.Duration(d.SignURLExpire))
+	if err != nil {
+		return nil, err
+	}
+	info := &model.S3MultipartDirectUploadInfo{
+		ChunkSize:   partSize,
+		UploadURLs:  uploadURLs,
+		CompleteURL: completeURL,
+		AbortURL:    abortURL,
+	}
+	createdSuccessfully = true
+	return info, nil
+}
+
 // implements driver.Getter interface
 func (d *S3) Get(ctx context.Context, path string) (model.Obj, error) {
 	// try to get object as a file using HeadObject

drivers/s3/meta.go

@@ -23,6 +23,8 @@ type Addition struct {
 	AddFilenameToDisposition bool   `json:"add_filename_to_disposition" help:"Add filename to Content-Disposition header."`
 	EnableDirectUpload       bool   `json:"enable_direct_upload" default:"false"`
 	DirectUploadHost         string `json:"direct_upload_host" required:"false"`
+	DirectUploadMaxParts     int64  `json:"direct_upload_max_parts" type:"number" default:"10000" help:"Maximum number of parts for frontend multipart upload. Set to 1 to disable multipart upload."`
+	DirectUploadMinPartSize  int64  `json:"direct_upload_min_part_size" type:"number" default:"104857600" help:"Minimum part size for frontend multipart upload, in bytes."`
 	UserAgent                string `json:"user_agent" required:"false" default:"" help:"Custom User-Agent for S3 requests."`
 }
 

drivers/s3/util.go

@@ -25,6 +25,10 @@ const (
 	defaultCopyPartSize int64 = 100 * 1024 * 1024
 	maxCopyPartSize     int64 = 5 * 1024 * 1024 * 1024
 	maxCopyParts        int64 = 10000
+
+	minMultipartUploadPartSize     int64 = 5 * 1024 * 1024
+	defaultMultipartUploadPartSize int64 = 100 * 1024 * 1024
+	maxMultipartUploadPartSize     int64 = 5 * 1024 * 1024 * 1024
 )
 
 // do others that not defined in Driver interface
@@ -79,7 +83,9 @@ func (d *S3) getClient(clientType int) *s3.S3 {
 	}
 	if clientType == ClientTypeDirectUpload && d.DirectUploadHost != "" {
 		client.Handlers.Build.PushBack(func(r *request.Request) {
-			if r.HTTPRequest.Method != http.MethodPut {
+			switch r.HTTPRequest.Method {
+			case http.MethodPut, http.MethodPost, http.MethodDelete:
+			default:
 				return
 			}
 			split := strings.SplitN(d.DirectUploadHost, "://", 2)
@@ -102,6 +108,24 @@ func getKey(path string, dir bool) string {
 	return path
 }
 
+func getMultipartUploadPartSize(size, maxParts, chunkSize int64) (int64, error) {
+	if maxParts <= 1 {
+		if size > maxMultipartUploadPartSize {
+			return 0, fmt.Errorf("object size %d exceeds direct upload limit", size)
+		}
+		return size, nil
+	}
+	maxParts = min(maxParts, maxCopyParts)
+	if size > maxMultipartUploadPartSize*maxParts {
+		return 0, fmt.Errorf("object size %d exceeds multipart upload limit", size)
+	}
+	if chunkSize <= 0 {
+		chunkSize = defaultMultipartUploadPartSize
+	}
+	chunkSize = min(chunkSize, maxMultipartUploadPartSize)
+	return max(chunkSize, (size+maxParts-1)/maxParts, minMultipartUploadPartSize), nil
+}
+
 var defaultPlaceholderName = ".openlist"
 
 func getPlaceholderName(placeholder string) string {