simples3/multipart.go

package simples3

import (
	"bytes"
	"context"
	"crypto/md5"
	"crypto/sha256"
	"encoding/base64"
	"encoding/xml"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"strconv"
	"sync"
	"time"
)

// Multipart upload constants
const (
	MinPartSize          = 5 * 1024 * 1024        // 5 MB
	MaxPartSize          = 5 * 1024 * 1024 * 1024 // 5 GB
	MaxParts             = 10000
	DefaultPartSize      = 5 * 1024 * 1024 // 5 MB
	DefaultMaxRetries    = 3
	DefaultRetryBaseWait = 100 * time.Millisecond
	DefaultRetryMaxWait  = 5 * time.Second
)

// InitiateMultipartUploadInput contains parameters for initiating a multipart upload
type InitiateMultipartUploadInput struct {
	Bucket         string            // Required: bucket name
	ObjectKey      string            // Required: object key
	ContentType    string            // Optional: content type
	CustomMetadata map[string]string // Optional: x-amz-meta-* headers
	ACL            string            // Optional: x-amz-acl

	// Optional: Server-side encryption (e.g. "AES256" or "aws:kms")
	ServerSideEncryption string
	// Optional: KMS Key ID (ARN or ID) when ServerSideEncryption is "aws:kms"
	SSEKMSKeyId string
}

// InitiateMultipartUploadOutput contains the response from initiating a multipart upload
type InitiateMultipartUploadOutput struct {
	Bucket   string
	Key      string
	UploadID string
}

// initiateMultipartUploadResult is the internal XML response structure
type initiateMultipartUploadResult struct {
	XMLName  xml.Name `xml:"InitiateMultipartUploadResult"`
	Bucket   string   `xml:"Bucket"`
	Key      string   `xml:"Key"`
	UploadID string   `xml:"UploadId"`
}

// InitiateMultipartUpload initiates a multipart upload and returns an upload ID
func (s3 *S3) InitiateMultipartUpload(input InitiateMultipartUploadInput) (InitiateMultipartUploadOutput, error) {
	if input.Bucket == "" {
		return InitiateMultipartUploadOutput{}, fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return InitiateMultipartUploadOutput{}, fmt.Errorf("object key is required")
	}

	if err := s3.renewIAMToken(); err != nil {
		return InitiateMultipartUploadOutput{}, err
	}

	urlStr := s3.getURL(input.Bucket, input.ObjectKey) + "?uploads"

	req, err := http.NewRequest(http.MethodPost, urlStr, nil)
	if err != nil {
		return InitiateMultipartUploadOutput{}, err
	}

	// Set headers
	if input.ContentType != "" {
		req.Header.Set("Content-Type", input.ContentType)
	}

	if input.ACL != "" {
		req.Header.Set("x-amz-acl", input.ACL)
	}

	// Set server-side encryption
	if input.ServerSideEncryption != "" {
		req.Header.Set("x-amz-server-side-encryption", input.ServerSideEncryption)
	}
	if input.SSEKMSKeyId != "" {
		req.Header.Set("x-amz-server-side-encryption-aws-kms-key-id", input.SSEKMSKeyId)
	}

	// Set custom metadata
	for k, v := range input.CustomMetadata {
		req.Header.Set(AMZMetaPrefix+k, v)
	}

	// Empty body hash for POST with no body
	req.Header.Set("x-amz-content-sha256", "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855")

	if err := s3.signRequest(req); err != nil {
		return InitiateMultipartUploadOutput{}, err
	}

	client := s3.getClient()
	res, err := client.Do(req)
	if err != nil {
		return InitiateMultipartUploadOutput{}, err
	}
	defer func() {
		res.Body.Close()
		io.Copy(io.Discard, res.Body)
	}()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return InitiateMultipartUploadOutput{}, err
	}

	if res.StatusCode != http.StatusOK {
		return InitiateMultipartUploadOutput{}, fmt.Errorf("status code: %s: %s", res.Status, string(body))
	}

	var result initiateMultipartUploadResult
	if err := xml.Unmarshal(body, &result); err != nil {
		return InitiateMultipartUploadOutput{}, err
	}

	return InitiateMultipartUploadOutput{
		Bucket:   result.Bucket,
		Key:      result.Key,
		UploadID: result.UploadID,
	}, nil
}

// UploadPartInput contains parameters for uploading a part
type UploadPartInput struct {
	Bucket     string    // Required: bucket name
	ObjectKey  string    // Required: object key
	UploadID   string    // Required: upload ID from InitiateMultipartUpload
	PartNumber int       // Required: part number (1-10000)
	Body       io.Reader // Required: part data
	Size       int64     // Required: size of part for Content-Length
}

// UploadPartOutput contains the response from uploading a part
type UploadPartOutput struct {
	ETag       string // Required for CompleteMultipartUpload
	PartNumber int
}

// UploadPart uploads a single part for a multipart upload
func (s3 *S3) UploadPart(input UploadPartInput) (UploadPartOutput, error) {
	if input.Bucket == "" {
		return UploadPartOutput{}, fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return UploadPartOutput{}, fmt.Errorf("object key is required")
	}
	if input.UploadID == "" {
		return UploadPartOutput{}, fmt.Errorf("upload ID is required")
	}
	if input.PartNumber < 1 || input.PartNumber > MaxParts {
		return UploadPartOutput{}, fmt.Errorf("part number must be between 1 and %d", MaxParts)
	}
	if input.Body == nil {
		return UploadPartOutput{}, fmt.Errorf("body is required")
	}
	if input.Size <= 0 {
		return UploadPartOutput{}, fmt.Errorf("size must be greater than 0")
	}

	if err := s3.renewIAMToken(); err != nil {
		return UploadPartOutput{}, err
	}

	// Build URL with query parameters
	urlStr := s3.getURL(input.Bucket, input.ObjectKey)
	params := url.Values{}
	params.Set("partNumber", strconv.Itoa(input.PartNumber))
	params.Set("uploadId", input.UploadID)
	urlStr += "?" + params.Encode()

	// Read body and compute SHA256
	content, err := io.ReadAll(input.Body)
	if err != nil {
		return UploadPartOutput{}, err
	}

	req, err := http.NewRequest(http.MethodPut, urlStr, bytes.NewReader(content))
	if err != nil {
		return UploadPartOutput{}, err
	}

	// Set headers
	req.ContentLength = input.Size
	h := sha256.New()
	h.Write(content)
	req.Header.Set("x-amz-content-sha256", fmt.Sprintf("%x", h.Sum(nil)))

	if err := s3.signRequest(req); err != nil {
		return UploadPartOutput{}, err
	}

	client := s3.getClient()
	res, err := client.Do(req)
	if err != nil {
		return UploadPartOutput{}, err
	}
	defer func() {
		res.Body.Close()
		io.Copy(io.Discard, res.Body)
	}()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return UploadPartOutput{}, err
	}

	if res.StatusCode != http.StatusOK {
		return UploadPartOutput{}, fmt.Errorf("status code: %s: %s", res.Status, string(body))
	}

	etag := res.Header.Get("ETag")
	if etag == "" {
		return UploadPartOutput{}, fmt.Errorf("ETag not found in response")
	}

	return UploadPartOutput{
		ETag:       etag,
		PartNumber: input.PartNumber,
	}, nil
}

// uploadPartWithRetry uploads a part with retry logic
func (s3 *S3) uploadPartWithRetry(input UploadPartInput, maxRetries int) (UploadPartOutput, error) {
	if maxRetries <= 0 {
		maxRetries = DefaultMaxRetries
	}

	var lastErr error
	for attempt := 0; attempt <= maxRetries; attempt++ {
		if attempt > 0 {
			// Exponential backoff
			waitTime := time.Duration(math.Pow(2, float64(attempt-1))) * DefaultRetryBaseWait
			if waitTime > DefaultRetryMaxWait {
				waitTime = DefaultRetryMaxWait
			}
			time.Sleep(waitTime)
		}

		output, err := s3.UploadPart(input)
		if err == nil {
			return output, nil
		}

		lastErr = err

		// Check if error is retryable
		if !isRetryableError(err) {
			return UploadPartOutput{}, err
		}
	}

	return UploadPartOutput{}, fmt.Errorf("failed after %d retries: %w", maxRetries, lastErr)
}

// isRetryableError determines if an error should be retried
func isRetryableError(err error) bool {
	if err == nil {
		return false
	}

	errStr := err.Error()
	// Retry on 5xx errors, timeouts, and connection resets
	return contains(errStr, "status code: 5") ||
		contains(errStr, "timeout") ||
		contains(errStr, "connection reset") ||
		contains(errStr, "EOF")
}

// contains checks if a string contains a substring
func contains(s, substr string) bool {
	return len(s) >= len(substr) && (s == substr || len(s) > len(substr) && (s[:len(substr)] == substr || s[len(s)-len(substr):] == substr || containsMiddle(s, substr)))
}

func containsMiddle(s, substr string) bool {
	for i := 0; i <= len(s)-len(substr); i++ {
		if s[i:i+len(substr)] == substr {
			return true
		}
	}
	return false
}

// CompletedPart represents a part that has been uploaded
type CompletedPart struct {
	PartNumber int
	ETag       string
}

// CompleteMultipartUploadInput contains parameters for completing a multipart upload
type CompleteMultipartUploadInput struct {
	Bucket    string          // Required: bucket name
	ObjectKey string          // Required: object key
	UploadID  string          // Required: upload ID
	Parts     []CompletedPart // Required: list of parts, ordered by PartNumber
}

// CompleteMultipartUploadOutput contains the response from completing a multipart upload
type CompleteMultipartUploadOutput struct {
	Location string
	Bucket   string
	Key      string
	ETag     string
}

// completeMultipartUploadRequest is the XML request structure
type completeMultipartUploadRequest struct {
	XMLName xml.Name       `xml:"CompleteMultipartUpload"`
	XMLNS   string         `xml:"xmlns,attr"`
	Parts   []completePart `xml:"Part"`
}

type completePart struct {
	PartNumber int    `xml:"PartNumber"`
	ETag       string `xml:"ETag"`
}

// completeMultipartUploadResult is the XML response structure
type completeMultipartUploadResult struct {
	XMLName  xml.Name `xml:"CompleteMultipartUploadResult"`
	Location string   `xml:"Location"`
	Bucket   string   `xml:"Bucket"`
	Key      string   `xml:"Key"`
	ETag     string   `xml:"ETag"`
}

// CompleteMultipartUpload completes a multipart upload
func (s3 *S3) CompleteMultipartUpload(input CompleteMultipartUploadInput) (CompleteMultipartUploadOutput, error) {
	if input.Bucket == "" {
		return CompleteMultipartUploadOutput{}, fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return CompleteMultipartUploadOutput{}, fmt.Errorf("object key is required")
	}
	if input.UploadID == "" {
		return CompleteMultipartUploadOutput{}, fmt.Errorf("upload ID is required")
	}
	if len(input.Parts) == 0 {
		return CompleteMultipartUploadOutput{}, fmt.Errorf("parts list cannot be empty")
	}

	if err := s3.renewIAMToken(); err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	// Build URL with query parameter
	urlStr := s3.getURL(input.Bucket, input.ObjectKey) + "?uploadId=" + url.QueryEscape(input.UploadID)

	// Build XML request body
	parts := make([]completePart, len(input.Parts))
	for i, p := range input.Parts {
		parts[i] = completePart{
			PartNumber: p.PartNumber,
			ETag:       p.ETag,
		}
	}

	completeReq := completeMultipartUploadRequest{
		XMLNS: "http://s3.amazonaws.com/doc/2006-03-01/",
		Parts: parts,
	}

	xmlBody, err := xml.Marshal(completeReq)
	if err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	req, err := http.NewRequest(http.MethodPost, urlStr, bytes.NewReader(xmlBody))
	if err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	// Set headers
	req.Header.Set("Content-Type", "application/xml")
	req.ContentLength = int64(len(xmlBody))

	h := sha256.New()
	h.Write(xmlBody)
	req.Header.Set("x-amz-content-sha256", fmt.Sprintf("%x", h.Sum(nil)))

	// Calculate Content-MD5
	md5Hash := md5.Sum(xmlBody)
	contentMD5 := base64.StdEncoding.EncodeToString(md5Hash[:])
	req.Header.Set("Content-MD5", contentMD5)

	if err := s3.signRequest(req); err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	client := s3.getClient()
	res, err := client.Do(req)
	if err != nil {
		return CompleteMultipartUploadOutput{}, err
	}
	defer func() {
		res.Body.Close()
		io.Copy(io.Discard, res.Body)
	}()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	if res.StatusCode != http.StatusOK {
		return CompleteMultipartUploadOutput{}, fmt.Errorf("status code: %s: %s", res.Status, string(body))
	}

	var result completeMultipartUploadResult
	if err := xml.Unmarshal(body, &result); err != nil {
		return CompleteMultipartUploadOutput{}, err
	}

	return CompleteMultipartUploadOutput{
		Location: result.Location,
		Bucket:   result.Bucket,
		Key:      result.Key,
		ETag:     result.ETag,
	}, nil
}

// AbortMultipartUploadInput contains parameters for aborting a multipart upload
type AbortMultipartUploadInput struct {
	Bucket    string // Required: bucket name
	ObjectKey string // Required: object key
	UploadID  string // Required: upload ID
}

// AbortMultipartUpload aborts a multipart upload and cleans up parts
func (s3 *S3) AbortMultipartUpload(input AbortMultipartUploadInput) error {
	if input.Bucket == "" {
		return fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return fmt.Errorf("object key is required")
	}
	if input.UploadID == "" {
		return fmt.Errorf("upload ID is required")
	}

	if err := s3.renewIAMToken(); err != nil {
		return err
	}

	// Build URL with query parameter
	urlStr := s3.getURL(input.Bucket, input.ObjectKey) + "?uploadId=" + url.QueryEscape(input.UploadID)

	req, err := http.NewRequest(http.MethodDelete, urlStr, nil)
	if err != nil {
		return err
	}

	// Empty body hash
	req.Header.Set("x-amz-content-sha256", "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855")

	if err := s3.signRequest(req); err != nil {
		return err
	}

	client := s3.getClient()
	res, err := client.Do(req)
	if err != nil {
		return err
	}
	defer func() {
		res.Body.Close()
		io.Copy(io.Discard, res.Body)
	}()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return err
	}

	if res.StatusCode != http.StatusNoContent {
		return fmt.Errorf("status code: %s: %s", res.Status, string(body))
	}

	return nil
}

// ListPartsInput contains parameters for listing parts
type ListPartsInput struct {
	Bucket           string // Required: bucket name
	ObjectKey        string // Required: object key
	UploadID         string // Required: upload ID
	MaxParts         int    // Optional: default 1000, max 1000
	PartNumberMarker int    // Optional: for pagination
}

// Part represents an uploaded part
type Part struct {
	PartNumber   int
	ETag         string
	Size         int64
	LastModified time.Time
}

// ListPartsOutput contains the response from listing parts
type ListPartsOutput struct {
	Bucket               string
	Key                  string
	UploadID             string
	Parts                []Part
	IsTruncated          bool
	NextPartNumberMarker int
	MaxParts             int
}

// listPartsResult is the XML response structure
type listPartsResult struct {
	XMLName              xml.Name   `xml:"ListPartsResult"`
	Bucket               string     `xml:"Bucket"`
	Key                  string     `xml:"Key"`
	UploadID             string     `xml:"UploadId"`
	Parts                []partInfo `xml:"Part"`
	IsTruncated          bool       `xml:"IsTruncated"`
	NextPartNumberMarker int        `xml:"NextPartNumberMarker"`
	MaxParts             int        `xml:"MaxParts"`
}

type partInfo struct {
	PartNumber   int       `xml:"PartNumber"`
	ETag         string    `xml:"ETag"`
	Size         int64     `xml:"Size"`
	LastModified time.Time `xml:"LastModified"`
}

// ListParts lists the parts that have been uploaded for a multipart upload
func (s3 *S3) ListParts(input ListPartsInput) (ListPartsOutput, error) {
	if input.Bucket == "" {
		return ListPartsOutput{}, fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return ListPartsOutput{}, fmt.Errorf("object key is required")
	}
	if input.UploadID == "" {
		return ListPartsOutput{}, fmt.Errorf("upload ID is required")
	}

	if err := s3.renewIAMToken(); err != nil {
		return ListPartsOutput{}, err
	}

	// Build URL with query parameters
	urlStr := s3.getURL(input.Bucket, input.ObjectKey)
	params := url.Values{}
	params.Set("uploadId", input.UploadID)
	if input.MaxParts > 0 {
		params.Set("max-parts", strconv.Itoa(input.MaxParts))
	}
	if input.PartNumberMarker > 0 {
		params.Set("part-number-marker", strconv.Itoa(input.PartNumberMarker))
	}
	urlStr += "?" + params.Encode()

	req, err := http.NewRequest(http.MethodGet, urlStr, nil)
	if err != nil {
		return ListPartsOutput{}, err
	}

	// Empty body hash
	req.Header.Set("x-amz-content-sha256", "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855")

	if err := s3.signRequest(req); err != nil {
		return ListPartsOutput{}, err
	}

	client := s3.getClient()
	res, err := client.Do(req)
	if err != nil {
		return ListPartsOutput{}, err
	}
	defer func() {
		res.Body.Close()
		io.Copy(io.Discard, res.Body)
	}()

	body, err := io.ReadAll(res.Body)
	if err != nil {
		return ListPartsOutput{}, err
	}

	if res.StatusCode != http.StatusOK {
		return ListPartsOutput{}, fmt.Errorf("status code: %s: %s", res.Status, string(body))
	}

	var result listPartsResult
	if err := xml.Unmarshal(body, &result); err != nil {
		return ListPartsOutput{}, err
	}

	parts := make([]Part, len(result.Parts))
	for i, p := range result.Parts {
		parts[i] = Part{
			PartNumber:   p.PartNumber,
			ETag:         p.ETag,
			Size:         p.Size,
			LastModified: p.LastModified,
		}
	}

	return ListPartsOutput{
		Bucket:               result.Bucket,
		Key:                  result.Key,
		UploadID:             result.UploadID,
		Parts:                parts,
		IsTruncated:          result.IsTruncated,
		NextPartNumberMarker: result.NextPartNumberMarker,
		MaxParts:             result.MaxParts,
	}, nil
}

// ProgressFunc is called during upload with progress information
type ProgressFunc func(info ProgressInfo)

// ProgressInfo contains progress information for a multipart upload
type ProgressInfo struct {
	TotalBytes     int64 // Total bytes to upload (if known)
	UploadedBytes  int64 // Bytes uploaded so far
	CurrentPart    int   // Current part number
	TotalParts     int   // Total parts (if known)
	BytesPerSecond int64 // Current upload speed
}

// MultipartUploadInput contains parameters for the high-level multipart upload
type MultipartUploadInput struct {
	Bucket         string            // Required: bucket name
	ObjectKey      string            // Required: object key
	Body           io.Reader         // Required: file/data to upload
	ContentType    string            // Optional: content type
	CustomMetadata map[string]string // Optional: x-amz-meta-* headers
	ACL            string            // Optional: x-amz-acl
	PartSize       int64             // Optional: default 5MB, min 5MB
	MaxRetries     int               // Optional: default 3
	Concurrency    int               // Optional: default 1 (sequential)
	OnProgress     ProgressFunc      // Optional: progress callback

	// Optional: Server-side encryption (e.g. "AES256" or "aws:kms")
	ServerSideEncryption string
	// Optional: KMS Key ID (ARN or ID) when ServerSideEncryption is "aws:kms"
	SSEKMSKeyId string
}

// MultipartUploadOutput contains the response from a multipart upload
type MultipartUploadOutput struct {
	Location string
	Bucket   string
	Key      string
	ETag     string
	UploadID string
}

// FileUploadMultipart handles the entire multipart upload workflow
func (s3 *S3) FileUploadMultipart(input MultipartUploadInput) (MultipartUploadOutput, error) {
	if input.Bucket == "" {
		return MultipartUploadOutput{}, fmt.Errorf("bucket name is required")
	}
	if input.ObjectKey == "" {
		return MultipartUploadOutput{}, fmt.Errorf("object key is required")
	}
	if input.Body == nil {
		return MultipartUploadOutput{}, fmt.Errorf("body is required")
	}

	// Set defaults
	partSize := input.PartSize
	if partSize == 0 {
		partSize = DefaultPartSize
	}
	if partSize < MinPartSize {
		return MultipartUploadOutput{}, fmt.Errorf("part size must be at least %d bytes", MinPartSize)
	}

	maxRetries := input.MaxRetries
	if maxRetries == 0 {
		maxRetries = DefaultMaxRetries
	}

	concurrency := input.Concurrency
	if concurrency <= 0 {
		concurrency = 1
	}

	// Initiate multipart upload
	initOutput, err := s3.InitiateMultipartUpload(InitiateMultipartUploadInput{
		Bucket:               input.Bucket,
		ObjectKey:            input.ObjectKey,
		ContentType:          input.ContentType,
		CustomMetadata:       input.CustomMetadata,
		ACL:                  input.ACL,
		ServerSideEncryption: input.ServerSideEncryption,
		SSEKMSKeyId:          input.SSEKMSKeyId,
	})
	if err != nil {
		return MultipartUploadOutput{}, err
	}

	// Read entire body to determine size and split into parts
	bodyData, err := io.ReadAll(input.Body)
	if err != nil {
		s3.AbortMultipartUpload(AbortMultipartUploadInput{
			Bucket:    input.Bucket,
			ObjectKey: input.ObjectKey,
			UploadID:  initOutput.UploadID,
		})
		return MultipartUploadOutput{}, err
	}

	totalSize := int64(len(bodyData))
	totalParts := int(math.Ceil(float64(totalSize) / float64(partSize)))

	if totalParts > MaxParts {
		s3.AbortMultipartUpload(AbortMultipartUploadInput{
			Bucket:    input.Bucket,
			ObjectKey: input.ObjectKey,
			UploadID:  initOutput.UploadID,
		})
		return MultipartUploadOutput{}, fmt.Errorf("file too large: requires %d parts, maximum is %d", totalParts, MaxParts)
	}

	startTime := time.Now()
	var uploadedBytes int64

	// Upload parts
	var completedParts []CompletedPart
	var err2 error

	if concurrency <= 1 {
		// Sequential upload
		completedParts, err2 = s3.uploadPartsSequential(bodyData, partSize, totalParts, initOutput.UploadID, input.Bucket, input.ObjectKey, maxRetries, input.OnProgress, &uploadedBytes, totalSize, startTime)
	} else {
		// Parallel upload
		completedParts, err2 = s3.uploadPartsParallel2(bodyData, partSize, totalParts, initOutput.UploadID, input.Bucket, input.ObjectKey, maxRetries, concurrency, input.OnProgress, &uploadedBytes, totalSize, startTime)
	}

	if err2 != nil {
		s3.AbortMultipartUpload(AbortMultipartUploadInput{
			Bucket:    input.Bucket,
			ObjectKey: input.ObjectKey,
			UploadID:  initOutput.UploadID,
		})
		return MultipartUploadOutput{}, err2
	}

	// Complete multipart upload
	completeOutput, err := s3.CompleteMultipartUpload(CompleteMultipartUploadInput{
		Bucket:    input.Bucket,
		ObjectKey: input.ObjectKey,
		UploadID:  initOutput.UploadID,
		Parts:     completedParts,
	})
	if err != nil {
		s3.AbortMultipartUpload(AbortMultipartUploadInput{
			Bucket:    input.Bucket,
			ObjectKey: input.ObjectKey,
			UploadID:  initOutput.UploadID,
		})
		return MultipartUploadOutput{}, err
	}

	return MultipartUploadOutput{
		Location: completeOutput.Location,
		Bucket:   completeOutput.Bucket,
		Key:      completeOutput.Key,
		ETag:     completeOutput.ETag,
		UploadID: initOutput.UploadID,
	}, nil
}

// uploadPartsSequential uploads parts sequentially
func (s3 *S3) uploadPartsSequential(bodyData []byte, partSize int64, totalParts int, uploadID, bucket, objectKey string, maxRetries int, onProgress ProgressFunc, uploadedBytes *int64, totalSize int64, startTime time.Time) ([]CompletedPart, error) {
	completedParts := make([]CompletedPart, 0, totalParts)

	for partNum := 1; partNum <= totalParts; partNum++ {
		start := int64(partNum-1) * partSize
		end := start + partSize
		if end > totalSize {
			end = totalSize
		}

		partData := bodyData[start:end]

		output, err := s3.uploadPartWithRetry(UploadPartInput{
			Bucket:     bucket,
			ObjectKey:  objectKey,
			UploadID:   uploadID,
			PartNumber: partNum,
			Body:       bytes.NewReader(partData),
			Size:       int64(len(partData)),
		}, maxRetries)

		if err != nil {
			return nil, err
		}

		completedParts = append(completedParts, CompletedPart{
			PartNumber: output.PartNumber,
			ETag:       output.ETag,
		})

		*uploadedBytes += int64(len(partData))

		// Call progress callback
		if onProgress != nil {
			elapsed := time.Since(startTime).Seconds()
			bytesPerSecond := int64(0)
			if elapsed > 0 {
				bytesPerSecond = int64(float64(*uploadedBytes) / elapsed)
			}

			onProgress(ProgressInfo{
				TotalBytes:     totalSize,
				UploadedBytes:  *uploadedBytes,
				CurrentPart:    partNum,
				TotalParts:     totalParts,
				BytesPerSecond: bytesPerSecond,
			})
		}
	}

	return completedParts, nil
}

// partData represents a part to be uploaded
type partData struct {
	partNumber int
	data       []byte
}

// uploadPartsParallel2 uploads parts in parallel using a worker pool
func (s3 *S3) uploadPartsParallel2(bodyData []byte, partSize int64, totalParts int, uploadID, bucket, objectKey string, maxRetries, concurrency int, onProgress ProgressFunc, uploadedBytes *int64, totalSize int64, startTime time.Time) ([]CompletedPart, error) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Create channels
	partsChan := make(chan partData, concurrency)
	resultsChan := make(chan CompletedPart, totalParts)
	errChan := make(chan error, concurrency)

	// Start workers
	var wg sync.WaitGroup
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case part, ok := <-partsChan:
					if !ok {
						return
					}

					output, err := s3.uploadPartWithRetry(UploadPartInput{
						Bucket:     bucket,
						ObjectKey:  objectKey,
						UploadID:   uploadID,
						PartNumber: part.partNumber,
						Body:       bytes.NewReader(part.data),
						Size:       int64(len(part.data)),
					}, maxRetries)

					if err != nil {
						select {
						case errChan <- err:
							cancel()
						default:
						}
						return
					}

					resultsChan <- CompletedPart{
						PartNumber: output.PartNumber,
						ETag:       output.ETag,
					}
				}
			}
		}()
	}

	// Send parts to workers
	go func() {
		for partNum := 1; partNum <= totalParts; partNum++ {
			start := int64(partNum-1) * partSize
			end := start + partSize
			if end > totalSize {
				end = totalSize
			}

			select {
			case <-ctx.Done():
				return
			case partsChan <- partData{
				partNumber: partNum,
				data:       bodyData[start:end],
			}:
			}
		}
		close(partsChan)
	}()

	// Collect results
	go func() {
		wg.Wait()
		close(resultsChan)
		close(errChan)
	}()

	completedParts := make([]CompletedPart, 0, totalParts)
	partsCompleted := 0

	for {
		select {
		case err := <-errChan:
			if err != nil {
				cancel()
				return nil, err
			}
		case part, ok := <-resultsChan:
			if !ok {
				// Sort by part number
				for i := 0; i < len(completedParts); i++ {
					for j := i + 1; j < len(completedParts); j++ {
						if completedParts[i].PartNumber > completedParts[j].PartNumber {
							completedParts[i], completedParts[j] = completedParts[j], completedParts[i]
						}
					}
				}
				return completedParts, nil
			}

			completedParts = append(completedParts, part)
			partsCompleted++
			*uploadedBytes = int64(partsCompleted) * partSize
			if *uploadedBytes > totalSize {
				*uploadedBytes = totalSize
			}

			// Call progress callback
			if onProgress != nil {
				elapsed := time.Since(startTime).Seconds()
				bytesPerSecond := int64(0)
				if elapsed > 0 {
					bytesPerSecond = int64(float64(*uploadedBytes) / elapsed)
				}

				onProgress(ProgressInfo{
					TotalBytes:     totalSize,
					UploadedBytes:  *uploadedBytes,
					CurrentPart:    partsCompleted,
					TotalParts:     totalParts,
					BytesPerSecond: bytesPerSecond,
				})
			}
		}
	}
}