package s3client

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/aws-sdk-go-v2/service/sts"
	"github.com/aws/smithy-go"
)

const maxAppendLogObjectBytes int64 = 5_000_000_000

type AWSS3Client struct {
	client  *s3.Client
	presign *s3.PresignClient
	bucket  string
	prefix  string
}

type AWSConfig struct {
	Region          string
	Bucket          string
	Prefix          string // key prefix, e.g. "tenants/<id>/" — keys already contain "blobs/"
	RoleARN         string
	Endpoint        string
	ForcePathStyle  bool
	AccessKeyID     string
	SecretAccessKey string
	SessionToken    string
}

func (cfg AWSConfig) validate() error {
	if cfg.Bucket == "" {
		return fmt.Errorf("s3 bucket is required")
	}
	hasAccessKey := cfg.AccessKeyID != ""
	hasSecretKey := cfg.SecretAccessKey != ""
	if hasAccessKey != hasSecretKey {
		return fmt.Errorf("s3 access key id and secret access key must be set together")
	}
	if cfg.SessionToken != "" && !hasAccessKey {
		return fmt.Errorf("s3 session token requires access key id and secret access key")
	}
	return nil
}

// Validate checks whether the S3 config has a legal combination of required
// fields before any SDK clients are constructed.
func (cfg AWSConfig) Validate() error {
	return cfg.validate()
}

// CredentialLogValue reports the credential source label for startup logs.
func CredentialLogValue(accessKeyID string) string {
	if accessKeyID != "" {
		return "static"
	}
	return "default-credentials"
}

// RoleLogValue reports the role-assumption label for startup logs.
func RoleLogValue(roleARN string) string {
	if roleARN == "" {
		return "none"
	}
	return roleARN
}

func applyS3Options(cfg AWSConfig) func(*s3.Options) {
	return func(o *s3.Options) {
		if cfg.Endpoint != "" {
			o.BaseEndpoint = aws.String(cfg.Endpoint)
		}
		o.UsePathStyle = cfg.ForcePathStyle
		if isTencentEndpoint(cfg.Endpoint) {
			o.HTTPClient = &http.Client{
				Transport: &contentMD5Transport{base: http.DefaultTransport},
			}
		}
	}
}

type contentMD5Transport struct {
	base http.RoundTripper
}

func (t *contentMD5Transport) RoundTrip(req *http.Request) (*http.Response, error) {
	if req.Method == http.MethodPost && req.URL.Query().Has("delete") && req.Body != nil {
		bodyBytes, err := io.ReadAll(req.Body)
		_ = req.Body.Close()
		if err != nil {
			return nil, fmt.Errorf("contentMD5Transport: read body: %w", err)
		}
		req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
		hash := md5.Sum(bodyBytes)
		req.Header.Set("Content-MD5", base64.StdEncoding.EncodeToString(hash[:]))
		req.ContentLength = int64(len(bodyBytes))
	}
	if t.base != nil {
		return t.base.RoundTrip(req)
	}
	return http.DefaultTransport.RoundTrip(req)
}

// New creates an S3-compatible client. The endpoint in cfg determines which
// credential chain is used:
//   - Aliyun OSS endpoints (*.aliyuncs.com): explicit key → RRSA → ALIBABA_CLOUD env
//     → AWS SDK default chain (if none of the above are configured)
//   - Tencent COS endpoints (*.myqcloud.com): explicit key → TENCENTCLOUD env → CAM role
//   - All other endpoints: explicit key → AWS SDK default chain
func New(ctx context.Context, cfg AWSConfig) (*AWSS3Client, error) {
	if err := cfg.Validate(); err != nil {
		return nil, err
	}
	if isAliyunEndpoint(cfg.Endpoint) {
		return newAliyun(ctx, cfg)
	}
	if isTencentEndpoint(cfg.Endpoint) {
		return newTencent(ctx, cfg)
	}
	return newAWS(ctx, cfg)
}

// newAWS builds an AWSS3Client using AWS credentials. When cfg.AccessKeyID is
// set, static credentials are used; otherwise the SDK default chain
// (env → profile → instance metadata) applies.
func newAWS(ctx context.Context, cfg AWSConfig) (*AWSS3Client, error) {
	var provider aws.CredentialsProvider
	if cfg.AccessKeyID != "" {
		provider = credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, cfg.SessionToken)
	}
	return buildS3Client(ctx, cfg, provider)
}

// buildS3Client assembles the AWS SDK config and constructs the final
// AWSS3Client. provider may be nil to use the SDK default credential chain.
func buildS3Client(ctx context.Context, cfg AWSConfig, provider aws.CredentialsProvider) (*AWSS3Client, error) {
	var transport *http.Transport
	if t, ok := http.DefaultTransport.(*http.Transport); ok {
		transport = t.Clone()
	} else {
		transport = &http.Transport{}
	}
	// S3 data-plane calls fan out more than control-plane API calls:
	// multipart uploads use uploadMaxConcurrency workers and FUSE reads can
	// issue direct range/prefetch requests concurrently. Keep this pool 2x
	// the dat9 client pool so hot S3 paths avoid repeated TLS handshakes.
	transport.MaxIdleConns = 512
	transport.MaxIdleConnsPerHost = 128

	loadOptions := []func(*awsconfig.LoadOptions) error{
		awsconfig.WithRegion(cfg.Region),
		awsconfig.WithResponseChecksumValidation(aws.ResponseChecksumValidationWhenRequired),
		awsconfig.WithHTTPClient(&http.Client{Transport: transport}),
	}

	if provider != nil {
		loadOptions = append(loadOptions, awsconfig.WithCredentialsProvider(provider))
	}

	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOptions...)
	if err != nil {
		return nil, fmt.Errorf("load aws config: %w", err)
	}

	if cfg.RoleARN != "" {
		stsClient := sts.NewFromConfig(awsCfg)
		awsCfg.Credentials = aws.NewCredentialsCache(
			stscreds.NewAssumeRoleProvider(stsClient, cfg.RoleARN,
				func(o *stscreds.AssumeRoleOptions) {
					o.RoleSessionName = "drive9-server"
				},
			),
		)
	}

	client := s3.NewFromConfig(awsCfg, applyS3Options(cfg))
	markS3ClientAvailable()
	return &AWSS3Client{
		client:  client,
		presign: s3.NewPresignClient(client),
		bucket:  cfg.Bucket,
		prefix:  normalizePrefix(cfg.Prefix),
	}, nil
}

func normalizePrefix(p string) string {
	p = strings.TrimLeft(p, "/")
	if p == "" {
		return ""
	}
	if !strings.HasSuffix(p, "/") {
		p += "/"
	}
	return p
}

func (c *AWSS3Client) fullKey(key string) string {
	if c.prefix == "" {
		return key
	}
	return c.prefix + key
}

func (c *AWSS3Client) CreateMultipartUpload(ctx context.Context, key string, algo ChecksumAlgo, encOpts EncryptionOpts) (*MultipartUpload, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("create_multipart_upload", result, start) }()

	in := &s3.CreateMultipartUploadInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
	}
	if err := applyEncryptionToCreateMultipartUploadInput(in, encOpts); err != nil {
		result = "error"
		return nil, err
	}
	awsAlgo, ok, err := checksumAlgorithmForAWS(algo)
	if err != nil {
		result = "error"
		return nil, err
	}
	if ok {
		in.ChecksumAlgorithm = awsAlgo
	}
	out, err := c.client.CreateMultipartUpload(ctx, in)
	if err != nil {
		result = "error"
		return nil, fmt.Errorf("create multipart upload: %w", err)
	}
	return &MultipartUpload{
		UploadID: aws.ToString(out.UploadId),
		Key:      key,
	}, nil
}

func checksumAlgorithmForAWS(algo ChecksumAlgo) (types.ChecksumAlgorithm, bool, error) {
	switch algo {
	case ChecksumAlgoNone:
		return "", false, nil
	case ChecksumAlgoCRC32C:
		return types.ChecksumAlgorithmCrc32c, true, nil
	case ChecksumAlgoSHA256:
		return types.ChecksumAlgorithmSha256, true, nil
	default:
		return "", false, fmt.Errorf("unsupported checksum algorithm: %q", algo)
	}
}

func (c *AWSS3Client) PresignUploadPart(ctx context.Context, key, uploadID string, partNumber int, partSize int64, algo ChecksumAlgo, checksumValue string, ttl time.Duration) (*UploadPartURL, error) {
	start := time.Now()
	metricResult := "ok"
	defer func() { recordS3Operation("presign_upload_part", metricResult, start) }()

	if ttl > UploadTTL {
		ttl = UploadTTL
	}
	in := &s3.UploadPartInput{
		Bucket:        &c.bucket,
		Key:           aws.String(c.fullKey(key)),
		UploadId:      &uploadID,
		PartNumber:    aws.Int32(int32(partNumber)),
		ContentLength: aws.Int64(partSize),
	}
	var headerKey string
	if checksumValue != "" {
		switch algo {
		case ChecksumAlgoCRC32C:
			in.ChecksumCRC32C = aws.String(checksumValue)
			headerKey = "x-amz-checksum-crc32c"
		default:
			in.ChecksumSHA256 = aws.String(checksumValue)
			headerKey = "x-amz-checksum-sha256"
		}
	}
	partPresigner := v4.NewSigner(func(o *v4.SignerOptions) {
		o.DisableURIPathEscaping = true
		o.DisableHeaderHoisting = true
	})
	out, err := c.presign.PresignUploadPart(ctx, in,
		s3.WithPresignExpires(ttl),
		func(o *s3.PresignOptions) { o.Presigner = partPresigner },
	)
	if err != nil {
		metricResult = "error"
		return nil, fmt.Errorf("presign upload part: %w", err)
	}
	headers := flattenSignedHeaders(out.SignedHeader)
	if checksumValue != "" {
		headers[headerKey] = checksumValue
	}
	urlResult := &UploadPartURL{
		Number:    partNumber,
		URL:       out.URL,
		Size:      partSize,
		Headers:   headers,
		ExpiresAt: time.Now().Add(ttl),
	}
	switch algo {
	case ChecksumAlgoCRC32C:
		urlResult.ChecksumCRC32C = checksumValue
	default:
		urlResult.ChecksumSHA256 = checksumValue
	}
	return urlResult, nil
}

func flattenSignedHeaders(h http.Header) map[string]string {
	if len(h) == 0 {
		return nil
	}
	out := make(map[string]string, len(h))
	for k, vs := range h {
		if len(vs) == 0 {
			continue
		}
		if strings.EqualFold(k, "host") {
			continue
		}
		out[strings.ToLower(k)] = vs[0]
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func (c *AWSS3Client) CompleteMultipartUpload(ctx context.Context, key, uploadID string, parts []Part) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("complete_multipart_upload", result, start) }()

	completed := make([]types.CompletedPart, len(parts))
	for i, p := range parts {
		cp := types.CompletedPart{
			PartNumber: aws.Int32(int32(p.Number)),
			ETag:       aws.String(p.ETag),
		}
		if p.ChecksumCRC32C != "" {
			cp.ChecksumCRC32C = aws.String(p.ChecksumCRC32C)
		} else if p.ChecksumSHA256 != "" {
			cp.ChecksumSHA256 = aws.String(p.ChecksumSHA256)
		}
		completed[i] = cp
	}
	_, err := c.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
		Bucket:   &c.bucket,
		Key:      aws.String(c.fullKey(key)),
		UploadId: &uploadID,
		MultipartUpload: &types.CompletedMultipartUpload{
			Parts: completed,
		},
	})
	if err != nil {
		result = "error"
		return fmt.Errorf("complete multipart upload: %w", err)
	}
	return nil
}

func (c *AWSS3Client) AbortMultipartUpload(ctx context.Context, key, uploadID string) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("abort_multipart_upload", result, start) }()

	_, err := c.client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
		Bucket:   &c.bucket,
		Key:      aws.String(c.fullKey(key)),
		UploadId: &uploadID,
	})
	if err != nil {
		// NoSuchUpload means the multipart upload was already completed or
		// aborted — the desired end state for drive9's best-effort abort
		// flows (expiry cleanup, supersede, client cancel). Treat it as
		// success so benign 404s do not pollute the s3client error metrics
		// and alerting (see Drive9S3ClientErrorRateHigh).
		if isNoSuchUpload(err) {
			return nil
		}
		result = "error"
		return fmt.Errorf("abort multipart upload: %w", err)
	}
	return nil
}

// isNoSuchUpload reports whether err is the NoSuchUpload error, which AWS S3,
// Tencent COS, and Aliyun OSS all return (same XML error code) when the
// referenced multipart upload was already completed or aborted.
func isNoSuchUpload(err error) bool {
	var apiErr smithy.APIError
	return errors.As(err, &apiErr) && apiErr.ErrorCode() == "NoSuchUpload"
}

func (c *AWSS3Client) ListParts(ctx context.Context, key, uploadID string) ([]Part, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("list_parts", result, start) }()

	var parts []Part
	var partMarker *string

	for {
		out, err := c.client.ListParts(ctx, &s3.ListPartsInput{
			Bucket:           &c.bucket,
			Key:              aws.String(c.fullKey(key)),
			UploadId:         &uploadID,
			PartNumberMarker: partMarker,
		})
		if err != nil {
			result = "error"
			return nil, fmt.Errorf("list parts: %w", err)
		}
		for _, p := range out.Parts {
			parts = append(parts, Part{
				Number:         int(aws.ToInt32(p.PartNumber)),
				Size:           aws.ToInt64(p.Size),
				ETag:           aws.ToString(p.ETag),
				ChecksumSHA256: aws.ToString(p.ChecksumSHA256),
				ChecksumCRC32C: aws.ToString(p.ChecksumCRC32C),
			})
		}
		if !aws.ToBool(out.IsTruncated) {
			break
		}
		partMarker = out.NextPartNumberMarker
	}
	return parts, nil
}

func (c *AWSS3Client) PresignGetObject(ctx context.Context, key string, ttl time.Duration) (string, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("presign_get_object", result, start) }()

	if ttl > DownloadTTL {
		ttl = DownloadTTL
	}
	out, err := c.presign.PresignGetObject(ctx, &s3.GetObjectInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
	}, s3.WithPresignExpires(ttl))
	if err != nil {
		result = "error"
		return "", fmt.Errorf("presign get object: %w", err)
	}
	return out.URL, nil
}

func (c *AWSS3Client) PutObject(ctx context.Context, key string, body io.Reader, size int64, encOpts EncryptionOpts) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("put_object", result, start) }()

	in := &s3.PutObjectInput{
		Bucket:        &c.bucket,
		Key:           aws.String(c.fullKey(key)),
		Body:          body,
		ContentLength: aws.Int64(size),
	}
	if err := applyEncryptionToPutObjectInput(in, encOpts); err != nil {
		result = "error"
		return err
	}
	_, err := c.client.PutObject(ctx, in)
	if err != nil {
		result = "error"
		return fmt.Errorf("put object: %w", err)
	}
	return nil
}

// AppendObject appends exactly size bytes to a Directory Bucket object at offset.
func (c *AWSS3Client) AppendObject(ctx context.Context, key string, body io.Reader, size, offset int64) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("append_object", result, start) }()

	if size < 0 || offset < 0 || size > maxAppendLogObjectBytes-offset {
		result = "error"
		return fmt.Errorf("append object size exceeds %d bytes", maxAppendLogObjectBytes)
	}
	_, err := c.client.PutObject(ctx, &s3.PutObjectInput{
		Bucket:           &c.bucket,
		Key:              aws.String(c.fullKey(key)),
		Body:             body,
		ContentLength:    aws.Int64(size),
		WriteOffsetBytes: aws.Int64(offset),
	})
	if err != nil {
		result = "error"
	}
	return classifyAppendLogError(err)
}

// GetObjectRange reads an inclusive byte range without presigning it.
func (c *AWSS3Client) GetObjectRange(ctx context.Context, key string, startByte, endByte int64) (io.ReadCloser, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("get_object_range", result, start) }()

	if startByte < 0 || endByte < startByte {
		result = "error"
		return nil, fmt.Errorf("invalid object range %d-%d", startByte, endByte)
	}
	out, err := c.client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
		Range:  aws.String(fmt.Sprintf("bytes=%d-%d", startByte, endByte)),
	})
	if err != nil {
		result = "error"
		return nil, err
	}
	return out.Body, nil
}

// PutAppendLogObject writes a replacement Directory Bucket object without
// general-purpose S3 encryption headers.
func (c *AWSS3Client) PutAppendLogObject(ctx context.Context, key string, body io.Reader, size int64) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("put_append_log_object", result, start) }()

	if size < 0 || size > maxAppendLogObjectBytes {
		result = "error"
		return fmt.Errorf("append object size exceeds %d bytes", maxAppendLogObjectBytes)
	}
	_, err := c.client.PutObject(ctx, &s3.PutObjectInput{
		Bucket:        &c.bucket,
		Key:           aws.String(c.fullKey(key)),
		Body:          body,
		ContentLength: aws.Int64(size),
	})
	if err != nil {
		result = "error"
	}
	return err
}

// CopyAppendLogObject rebases a Directory Bucket object into a fresh key.
func (c *AWSS3Client) CopyAppendLogObject(ctx context.Context, sourceKey, destKey string) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("copy_append_log_object", result, start) }()

	copySource := url.PathEscape(c.bucket + "/" + c.fullKey(sourceKey))
	_, err := c.client.CopyObject(ctx, &s3.CopyObjectInput{
		Bucket:     &c.bucket,
		Key:        aws.String(c.fullKey(destKey)),
		CopySource: aws.String(copySource),
	})
	if err != nil {
		result = "error"
	}
	return err
}

func classifyAppendLogError(err error) error {
	if err == nil {
		return nil
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "TooManyParts":
			return fmt.Errorf("%w: %v", ErrAppendTooManyParts, err)
		case "InvalidWriteOffset":
			return fmt.Errorf("%w: %v", ErrAppendOffsetMismatch, err)
		}
	}
	return err
}

func applyEncryptionToCreateMultipartUploadInput(in *s3.CreateMultipartUploadInput, encOpts EncryptionOpts) error {
	fields, err := awsEncryptionFields(encOpts)
	if err != nil {
		return err
	}
	if fields.serverSideEncryption == "" {
		return nil
	}
	in.ServerSideEncryption = fields.serverSideEncryption
	in.SSEKMSKeyId = fields.kmsKeyID
	in.BucketKeyEnabled = fields.bucketKeyEnabled
	in.SSEKMSEncryptionContext = fields.encryptionContext
	return nil
}

func applyEncryptionToPutObjectInput(in *s3.PutObjectInput, encOpts EncryptionOpts) error {
	fields, err := awsEncryptionFields(encOpts)
	if err != nil {
		return err
	}
	if fields.serverSideEncryption == "" {
		return nil
	}
	in.ServerSideEncryption = fields.serverSideEncryption
	in.SSEKMSKeyId = fields.kmsKeyID
	in.BucketKeyEnabled = fields.bucketKeyEnabled
	in.SSEKMSEncryptionContext = fields.encryptionContext
	return nil
}

type awsEncryptionFieldSet struct {
	serverSideEncryption types.ServerSideEncryption
	kmsKeyID             *string
	bucketKeyEnabled     *bool
	encryptionContext    *string
}

func awsEncryptionFields(encOpts EncryptionOpts) (awsEncryptionFieldSet, error) {
	var fields awsEncryptionFieldSet
	switch encOpts.Mode {
	case "", EncryptionModeLegacy, EncryptionModeNone:
		if err := rejectUnusedEncryptionFields(encOpts); err != nil {
			return fields, err
		}
		return fields, nil
	case EncryptionModeSSES3:
		if err := rejectUnusedEncryptionFields(encOpts); err != nil {
			return fields, err
		}
		fields.serverSideEncryption = types.ServerSideEncryptionAes256
		return fields, nil
	case EncryptionModeSSEKMS:
		if encOpts.KMSKeyID == "" {
			return fields, fmt.Errorf("sse-kms encryption requires KMS key ID")
		}
		fields.serverSideEncryption = types.ServerSideEncryptionAwsKms
		fields.kmsKeyID = aws.String(encOpts.KMSKeyID)
		fields.bucketKeyEnabled = aws.Bool(encOpts.BucketKeyEnabled)
	case EncryptionModeDSSEKMS:
		if encOpts.KMSKeyID == "" {
			return fields, fmt.Errorf("dsse-kms encryption requires KMS key ID")
		}
		if encOpts.BucketKeyEnabled {
			return fields, fmt.Errorf("bucket key is not supported for dsse-kms encryption")
		}
		fields.serverSideEncryption = types.ServerSideEncryptionAwsKmsDsse
		fields.kmsKeyID = aws.String(encOpts.KMSKeyID)
	default:
		return fields, fmt.Errorf("unsupported encryption mode: %q", encOpts.Mode)
	}
	contextValue, err := encodeKMSEncryptionContext(encOpts.EncryptionContext)
	if err != nil {
		return fields, err
	}
	fields.encryptionContext = contextValue
	return fields, nil
}

func rejectUnusedEncryptionFields(encOpts EncryptionOpts) error {
	if encOpts.KMSKeyID != "" {
		return fmt.Errorf("KMS key ID is not supported for %s encryption", noEncryptionModeLabel(encOpts.Mode))
	}
	if encOpts.BucketKeyEnabled {
		return fmt.Errorf("bucket key is not supported for %s encryption", noEncryptionModeLabel(encOpts.Mode))
	}
	if len(encOpts.EncryptionContext) != 0 {
		return fmt.Errorf("encryption context is not supported for %s encryption", noEncryptionModeLabel(encOpts.Mode))
	}
	return nil
}

func noEncryptionModeLabel(mode EncryptionMode) string {
	if mode == "" {
		return "zero-value"
	}
	return string(mode)
}

func encodeKMSEncryptionContext(contextMap map[string]string) (*string, error) {
	if len(contextMap) == 0 {
		return nil, nil
	}
	payload, err := json.Marshal(contextMap)
	if err != nil {
		return nil, fmt.Errorf("encode KMS encryption context: %w", err)
	}
	return aws.String(base64.StdEncoding.EncodeToString(payload)), nil
}

func (c *AWSS3Client) GetObject(ctx context.Context, key string) (io.ReadCloser, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("get_object", result, start) }()

	out, err := c.client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
	})
	if err != nil {
		result = "error"
		return nil, fmt.Errorf("get object: %w", err)
	}
	return out.Body, nil
}

func (c *AWSS3Client) DeleteObject(ctx context.Context, key string) error {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("delete_object", result, start) }()

	_, err := c.client.DeleteObject(ctx, &s3.DeleteObjectInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
	})
	if err != nil {
		result = "error"
		return fmt.Errorf("delete object: %w", err)
	}
	return nil
}

func (c *AWSS3Client) DeletePrefix(ctx context.Context, prefix string) (PrefixDeleteResult, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("delete_prefix", result, start) }()

	fullPrefix := c.fullKey(strings.TrimLeft(prefix, "/"))
	var out PrefixDeleteResult
	if err := c.abortMultipartUploadsByPrefix(ctx, fullPrefix, &out); err != nil {
		result = "error"
		return out, err
	}
	if err := c.deleteObjectsByPrefix(ctx, fullPrefix, &out); err != nil {
		result = "error"
		return out, err
	}
	return out, nil
}

func (c *AWSS3Client) abortMultipartUploadsByPrefix(ctx context.Context, fullPrefix string, out *PrefixDeleteResult) error {
	var keyMarker *string
	var uploadIDMarker *string
	for {
		res, err := c.client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
			Bucket:         &c.bucket,
			Prefix:         aws.String(fullPrefix),
			KeyMarker:      keyMarker,
			UploadIdMarker: uploadIDMarker,
		})
		if err != nil {
			return fmt.Errorf("list multipart uploads by prefix: %w", err)
		}
		for _, upload := range res.Uploads {
			key := aws.ToString(upload.Key)
			uploadID := aws.ToString(upload.UploadId)
			if key == "" || uploadID == "" {
				continue
			}
			if _, err := c.client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
				Bucket:   &c.bucket,
				Key:      aws.String(key),
				UploadId: aws.String(uploadID),
			}); err != nil {
				// Already-completed/aborted uploads are the desired end state;
				// skip them instead of failing the whole prefix cleanup.
				if isNoSuchUpload(err) {
					continue
				}
				return fmt.Errorf("abort multipart upload %s/%s: %w", key, uploadID, err)
			}
			out.AbortedMultipartUploads++
		}
		if !aws.ToBool(res.IsTruncated) {
			return nil
		}
		keyMarker = res.NextKeyMarker
		uploadIDMarker = res.NextUploadIdMarker
	}
}

func (c *AWSS3Client) deleteObjectsByPrefix(ctx context.Context, fullPrefix string, out *PrefixDeleteResult) error {
	var continuation *string
	for {
		res, err := c.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket:            &c.bucket,
			Prefix:            aws.String(fullPrefix),
			ContinuationToken: continuation,
		})
		if err != nil {
			return fmt.Errorf("list objects by prefix: %w", err)
		}
		if len(res.Contents) > 0 {
			objects := make([]types.ObjectIdentifier, 0, len(res.Contents))
			for _, obj := range res.Contents {
				key := aws.ToString(obj.Key)
				if key == "" {
					continue
				}
				objects = append(objects, types.ObjectIdentifier{Key: aws.String(key)})
			}
			if len(objects) > 0 {
				del, err := c.client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
					Bucket: &c.bucket,
					Delete: &types.Delete{
						Objects: objects,
						Quiet:   aws.Bool(true),
					},
				})
				if err != nil {
					if !isDeleteObjectsCompatibilityError(err) {
						return fmt.Errorf("delete objects by prefix: %w", err)
					}
					deleted, fallbackErr := c.deleteObjectsOneByOne(ctx, objects)
					out.DeletedObjects += deleted
					if fallbackErr != nil {
						return fmt.Errorf("delete objects by prefix fallback: %w", fallbackErr)
					}
					continue
				}
				if len(del.Errors) > 0 {
					first := del.Errors[0]
					return fmt.Errorf("delete object %q: %s %s", aws.ToString(first.Key), aws.ToString(first.Code), aws.ToString(first.Message))
				}
				out.DeletedObjects += int64(len(objects))
			}
		}
		if !aws.ToBool(res.IsTruncated) {
			return nil
		}
		continuation = res.NextContinuationToken
	}
}

func isDeleteObjectsCompatibilityError(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "MissingArgument") ||
		strings.Contains(msg, "MalformedXML") ||
		strings.Contains(msg, "NotImplemented")
}

func (c *AWSS3Client) deleteObjectsOneByOne(ctx context.Context, objects []types.ObjectIdentifier) (int64, error) {
	var deleted int64
	for _, obj := range objects {
		key := aws.ToString(obj.Key)
		if key == "" {
			continue
		}
		if _, err := c.client.DeleteObject(ctx, &s3.DeleteObjectInput{
			Bucket: &c.bucket,
			Key:    aws.String(key),
		}); err != nil {
			return deleted, fmt.Errorf("delete object %s: %w", key, err)
		}
		deleted++
	}
	return deleted, nil
}

func (c *AWSS3Client) UploadPartCopy(ctx context.Context, destKey, uploadID string, partNumber int, sourceKey string, startByte, endByte int64) (string, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("upload_part_copy", result, start) }()

	copySource := fmt.Sprintf("%s/%s", c.bucket, c.fullKey(sourceKey))
	copyRange := fmt.Sprintf("bytes=%d-%d", startByte, endByte)

	out, err := c.client.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
		Bucket:          &c.bucket,
		Key:             aws.String(c.fullKey(destKey)),
		UploadId:        &uploadID,
		PartNumber:      aws.Int32(int32(partNumber)),
		CopySource:      aws.String(copySource),
		CopySourceRange: aws.String(copyRange),
	})
	if err != nil {
		result = "error"
		return "", fmt.Errorf("upload part copy: %w", err)
	}
	return aws.ToString(out.CopyPartResult.ETag), nil
}

func (c *AWSS3Client) PresignGetObjectRange(ctx context.Context, key string, startByte, endByte int64, ttl time.Duration) (string, error) {
	start := time.Now()
	result := "ok"
	defer func() { recordS3Operation("presign_get_object_range", result, start) }()

	if ttl > DownloadTTL {
		ttl = DownloadTTL
	}
	rangeHeader := fmt.Sprintf("bytes=%d-%d", startByte, endByte)
	out, err := c.presign.PresignGetObject(ctx, &s3.GetObjectInput{
		Bucket: &c.bucket,
		Key:    aws.String(c.fullKey(key)),
		Range:  aws.String(rangeHeader),
	}, s3.WithPresignExpires(ttl))
	if err != nil {
		result = "error"
		return "", fmt.Errorf("presign get object range: %w", err)
	}
	return out.URL, nil
}

var _ S3Client = (*AWSS3Client)(nil)