fm_be/internal/repository/s3/file_object_storage.go

package s3

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"

	"wucher/internal/config"
	filemanager "wucher/internal/domain/file_manager"
	"wucher/internal/shared/pkg/metrics"
)

type FileObjectStorage struct {
	cfg       config.FileStorageConfig
	client    *awss3.Client
	presigner *awss3.PresignClient
	component string
}

func NewFileObjectStorage(ctx context.Context, cfg config.FileStorageConfig) (*FileObjectStorage, error) {
	if err := validateConfig(cfg); err != nil {
		return nil, err
	}

	loadOptions := make([]func(*awsconfig.LoadOptions) error, 0, 3)
	if cfg.Region != "" {
		loadOptions = append(loadOptions, awsconfig.WithRegion(cfg.Region))
	}
	if strings.TrimSpace(cfg.Endpoint) != "" &&
		strings.TrimSpace(cfg.AccessKeyID) != "" &&
		strings.TrimSpace(cfg.SecretAccessKey) != "" {
		loadOptions = append(loadOptions, awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
			strings.TrimSpace(cfg.AccessKeyID),
			strings.TrimSpace(cfg.SecretAccessKey),
			"",
		)))
	}
	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOptions...)
	if err != nil {
		return nil, err
	}

	client := awss3.NewFromConfig(awsCfg, func(options *awss3.Options) {
		options.UsePathStyle = cfg.ForcePathStyle
		if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
			options.BaseEndpoint = awsv2.String(endpoint)
		}
	})

	storage := &FileObjectStorage{
		cfg:       cfg,
		client:    client,
		presigner: awss3.NewPresignClient(client),
		component: "api",
	}

	if cfg.AutoCreateBucket {
		if err := storage.EnsureBucket(ctx); err != nil {
			return nil, err
		}
	}

	return storage, nil
}

func (s *FileObjectStorage) EnsureBucket(ctx context.Context) error {
	headStarted := time.Now()
	if _, err := s.client.HeadBucket(ctx, &awss3.HeadBucketInput{
		Bucket: awsv2.String(s.cfg.Bucket),
	}); err == nil {
		metrics.ObserveAWSRequest(s.component, "s3", "HeadBucket", nil, headStarted)
		return nil
	} else {
		metrics.ObserveAWSRequest(s.component, "s3", "HeadBucket", err, headStarted)
	}

	input := &awss3.CreateBucketInput{
		Bucket: awsv2.String(s.cfg.Bucket),
	}
	if s.cfg.Region != "" && s.cfg.Region != "us-east-1" {
		input.CreateBucketConfiguration = &types.CreateBucketConfiguration{
			LocationConstraint: types.BucketLocationConstraint(s.cfg.Region),
		}
	}

	createStarted := time.Now()
	_, err := s.client.CreateBucket(ctx, input)
	metrics.ObserveAWSRequest(s.component, "s3", "CreateBucket", err, createStarted)
	if err == nil {
		return nil
	}
	if isBucketAlreadyOwnedByYou(err) {
		return nil
	}
	return err
}

func (s *FileObjectStorage) PutObject(ctx context.Context, input filemanager.PutObjectInput) (filemanager.ObjectMetadata, error) {
	objectKey := buildObjectKey(s.cfg.ObjectKeyPrefix, input.ObjectKey)

	started := time.Now()
	output, err := s.client.PutObject(ctx, &awss3.PutObjectInput{
		Bucket:      awsv2.String(s.cfg.Bucket),
		Key:         awsv2.String(objectKey),
		Body:        input.Body,
		ContentType: optionalString(input.ContentType),
		Metadata:    input.Metadata,
	})
	metrics.ObserveAWSRequest(s.component, "s3", "PutObject", err, started)
	metrics.ObserveFileUpload(s.component, err)
	if err == nil {
		metrics.ObserveAWSTransferredBytes(s.component, "s3", "PutObject", "upload", input.SizeBytes)
	}
	if err != nil {
		return filemanager.ObjectMetadata{}, err
	}

	return filemanager.ObjectMetadata{
		Bucket:    s.cfg.Bucket,
		ObjectKey: objectKey,
		ETag:      awsv2.ToString(output.ETag),
		VersionID: awsv2.ToString(output.VersionId),
	}, nil
}

func (s *FileObjectStorage) CopyObject(ctx context.Context, sourceObjectKey, destinationObjectKey string) (filemanager.ObjectMetadata, error) {
	sourceKey := buildObjectKey(s.cfg.ObjectKeyPrefix, sourceObjectKey)
	destinationKey := buildObjectKey(s.cfg.ObjectKeyPrefix, destinationObjectKey)

	started := time.Now()
	_, err := s.client.CopyObject(ctx, &awss3.CopyObjectInput{
		Bucket:            awsv2.String(s.cfg.Bucket),
		CopySource:        awsv2.String(url.PathEscape(s.cfg.Bucket + "/" + sourceKey)),
		Key:               awsv2.String(destinationKey),
		MetadataDirective: types.MetadataDirectiveCopy,
	})
	metrics.ObserveAWSRequest(s.component, "s3", "CopyObject", err, started)
	if err != nil {
		return filemanager.ObjectMetadata{}, err
	}
	return s.HeadObject(ctx, destinationObjectKey)
}

func (s *FileObjectStorage) DeleteObject(ctx context.Context, objectKey string) error {
	started := time.Now()
	_, err := s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{
		Bucket: awsv2.String(s.cfg.Bucket),
		Key:    awsv2.String(buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)),
	})
	metrics.ObserveAWSRequest(s.component, "s3", "DeleteObject", err, started)
	metrics.ObserveFileDelete(s.component, err)
	return err
}

func (s *FileObjectStorage) PutObjectTagging(ctx context.Context, objectKey string, tags map[string]string) error {
	if len(tags) == 0 {
		return nil
	}
	key := buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)
	tagSet := make([]types.Tag, 0, len(tags))
	for k, v := range tags {
		keyTag := strings.TrimSpace(k)
		valueTag := strings.TrimSpace(v)
		if keyTag == "" || valueTag == "" {
			continue
		}
		tagSet = append(tagSet, types.Tag{
			Key:   awsv2.String(keyTag),
			Value: awsv2.String(valueTag),
		})
	}
	if len(tagSet) == 0 {
		return nil
	}

	started := time.Now()
	_, err := s.client.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{
		Bucket:  awsv2.String(s.cfg.Bucket),
		Key:     awsv2.String(key),
		Tagging: &types.Tagging{TagSet: tagSet},
	})
	metrics.ObserveAWSRequest(s.component, "s3", "PutObjectTagging", err, started)
	return err
}

func (s *FileObjectStorage) DeleteObjectTaggingKeys(ctx context.Context, objectKey string, keys []string) error {
	if len(keys) == 0 {
		return nil
	}
	key := buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)

	started := time.Now()
	existing, err := s.client.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{
		Bucket: awsv2.String(s.cfg.Bucket),
		Key:    awsv2.String(key),
	})
	metrics.ObserveAWSRequest(s.component, "s3", "GetObjectTagging", err, started)
	if err != nil {
		return err
	}
	removed := make(map[string]struct{}, len(keys))
	for i := range keys {
		k := strings.TrimSpace(keys[i])
		if k == "" {
			continue
		}
		removed[k] = struct{}{}
	}
	tagSet := make([]types.Tag, 0, len(existing.TagSet))
	for i := range existing.TagSet {
		tagKey := strings.TrimSpace(awsv2.ToString(existing.TagSet[i].Key))
		if _, drop := removed[tagKey]; drop {
			continue
		}
		tagSet = append(tagSet, existing.TagSet[i])
	}

	started = time.Now()
	_, err = s.client.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{
		Bucket:  awsv2.String(s.cfg.Bucket),
		Key:     awsv2.String(key),
		Tagging: &types.Tagging{TagSet: tagSet},
	})
	metrics.ObserveAWSRequest(s.component, "s3", "PutObjectTagging", err, started)
	return err
}

func (s *FileObjectStorage) HeadObject(ctx context.Context, objectKey string) (filemanager.ObjectMetadata, error) {
	key := buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)
	started := time.Now()
	output, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{
		Bucket: awsv2.String(s.cfg.Bucket),
		Key:    awsv2.String(key),
	})
	metrics.ObserveAWSRequest(s.component, "s3", "HeadObject", err, started)
	metrics.ObserveFileHead(s.component, err)
	if err != nil {
		if isObjectNotFoundError(err) {
			return filemanager.ObjectMetadata{}, filemanager.ErrUploadObjectNotFound
		}
		return filemanager.ObjectMetadata{}, err
	}

	return filemanager.ObjectMetadata{
		Bucket:       s.cfg.Bucket,
		ObjectKey:    key,
		ETag:         awsv2.ToString(output.ETag),
		VersionID:    awsv2.ToString(output.VersionId),
		SizeBytes:    awsv2.ToInt64(output.ContentLength),
		LastModified: output.LastModified,
		ContentType:  awsv2.ToString(output.ContentType),
	}, nil
}

func (s *FileObjectStorage) GetObject(ctx context.Context, objectKey string) (io.ReadCloser, filemanager.ObjectMetadata, error) {
	key := buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)
	started := time.Now()
	output, err := s.client.GetObject(ctx, &awss3.GetObjectInput{
		Bucket: awsv2.String(s.cfg.Bucket),
		Key:    awsv2.String(key),
	})
	metrics.ObserveAWSRequest(s.component, "s3", "GetObject", err, started)
	if err != nil {
		if isObjectNotFoundError(err) {
			return nil, filemanager.ObjectMetadata{}, filemanager.ErrUploadObjectNotFound
		}
		return nil, filemanager.ObjectMetadata{}, err
	}

	metrics.ObserveAWSTransferredBytes(s.component, "s3", "GetObject", "download", awsv2.ToInt64(output.ContentLength))
	return output.Body, filemanager.ObjectMetadata{
		Bucket:       s.cfg.Bucket,
		ObjectKey:    key,
		ETag:         awsv2.ToString(output.ETag),
		VersionID:    awsv2.ToString(output.VersionId),
		SizeBytes:    awsv2.ToInt64(output.ContentLength),
		LastModified: output.LastModified,
		ContentType:  awsv2.ToString(output.ContentType),
	}, nil
}

func (s *FileObjectStorage) PresignPutObject(ctx context.Context, input filemanager.PresignPutObjectInput) (filemanager.PresignedPutObject, error) {
	key := buildObjectKey(s.cfg.ObjectKeyPrefix, input.ObjectKey)
	ttl := input.TTL
	if ttl <= 0 {
		ttl = s.cfg.PresignPutTTL
	}
	if ttl <= 0 {
		ttl = 15 * time.Minute
	}

	putInput := &awss3.PutObjectInput{
		Bucket:      awsv2.String(s.cfg.Bucket),
		Key:         awsv2.String(key),
		ContentType: optionalString(input.ContentType),
		Metadata:    input.Metadata,
	}

	output, err := s.presigner.PresignPutObject(ctx, putInput, func(options *awss3.PresignOptions) {
		options.Expires = ttl
	})
	if err != nil {
		return filemanager.PresignedPutObject{}, err
	}

	return filemanager.PresignedPutObject{
		Bucket:    s.cfg.Bucket,
		ObjectKey: key,
		Method:    output.Method,
		URL:       output.URL,
		Headers:   canonicalSignedHeaders(output.SignedHeader),
		ExpiresAt: time.Now().UTC().Add(ttl),
	}, nil
}

func (s *FileObjectStorage) PresignGetObject(ctx context.Context, objectKey string, ttl time.Duration) (string, error) {
	if ttl <= 0 {
		ttl = s.cfg.PresignGetTTL
	}

	started := time.Now()
	output, err := s.presigner.PresignGetObject(ctx, &awss3.GetObjectInput{
		Bucket: awsv2.String(s.cfg.Bucket),
		Key:    awsv2.String(buildObjectKey(s.cfg.ObjectKeyPrefix, objectKey)),
	}, func(options *awss3.PresignOptions) {
		options.Expires = ttl
	})
	metrics.ObserveAWSRequest(s.component, "s3", "GetObject", err, started)
	metrics.ObserveFilePresignDownload(s.component, err)
	if err != nil {
		return "", err
	}
	return output.URL, nil
}

func validateConfig(cfg config.FileStorageConfig) error {
	if strings.ToLower(strings.TrimSpace(cfg.Provider)) != "s3" {
		return errors.New("unsupported file storage provider")
	}
	if strings.TrimSpace(cfg.Bucket) == "" {
		return errors.New("file storage bucket is required")
	}
	if strings.TrimSpace(cfg.Region) == "" {
		return errors.New("file storage region is required")
	}
	if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
		parsed, err := url.Parse(endpoint)
		if err != nil {
			return err
		}
		if parsed.Scheme != "https" && !cfg.AllowInsecureEndpoint {
			return errors.New("insecure file storage endpoint is disabled")
		}
	}
	return nil
}

func buildObjectKey(prefix, objectKey string) string {
	objectKey = strings.TrimSpace(objectKey)
	objectKey = strings.TrimPrefix(objectKey, "/")
	objectKey = strings.TrimPrefix(objectKey, "./")
	if objectKey == "" {
		return objectKey
	}

	prefix = strings.TrimSpace(prefix)
	prefix = strings.Trim(prefix, "/")
	if prefix == "" {
		return objectKey
	}
	if objectKey == prefix || strings.HasPrefix(objectKey, prefix+"/") {
		return objectKey
	}
	return prefix + "/" + objectKey
}

func optionalString(value string) *string {
	value = strings.TrimSpace(value)
	if value == "" {
		return nil
	}
	return awsv2.String(value)
}

func isBucketAlreadyOwnedByYou(err error) bool {
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		return false
	}
	return apiErr.ErrorCode() == "BucketAlreadyOwnedByYou" || apiErr.ErrorCode() == "BucketAlreadyExists"
}

func isObjectNotFoundError(err error) bool {
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		return false
	}
	switch strings.TrimSpace(apiErr.ErrorCode()) {
	case "NotFound", "NoSuchKey", "404":
		return true
	default:
		return false
	}
}

func canonicalSignedHeaders(header http.Header) map[string]string {
	if len(header) == 0 {
		return nil
	}
	out := make(map[string]string, len(header))
	for key, values := range header {
		k := strings.TrimSpace(strings.ToLower(key))
		if k == "" || len(values) == 0 {
			continue
		}
		out[k] = strings.TrimSpace(values[0])
	}
	return out
}