kra-new/internal/integration/storage/s3_storage.go

package storage

import (
	"context"
	"fmt"
	"io"
	"kra/internal/biz/system"
	"net/url"
	"path"
	"strings"

	"kra/internal/conf"

	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
)

type s3Storage struct {
	client                  *minio.Client
	bucket, baseURL, prefix string
}

func newS3Storage(provider string, config *conf.AdminBackend_ObjectStore) (system.FileStorage, error) {
	if config == nil || config.Bucket == "" || config.AccessKey == "" || config.SecretKey == "" {
		return nil, fmt.Errorf("%s storage configuration is incomplete", provider)
	}
	endpoint := config.Endpoint
	if endpoint == "" {
		switch provider {
		case "aws-s3":
			endpoint = "s3." + config.Region + ".amazonaws.com"
		case "cloudflare-r2":
			endpoint = config.AccountId + ".r2.cloudflarestorage.com"
		case "aliyun-oss":
			endpoint = "oss-" + config.Region + ".aliyuncs.com"
		case "huawei-obs":
			endpoint = "obs." + config.Region + ".myhuaweicloud.com"
		case "tencent-cos":
			endpoint = "cos." + config.Region + ".myqcloud.com"
		}
	}
	secure := config.UseSsl
	if parsed, err := url.Parse(endpoint); err == nil && parsed.Host != "" {
		secure = parsed.Scheme == "https"
		endpoint = parsed.Host
	}
	if endpoint == "" {
		return nil, fmt.Errorf("%s endpoint is required", provider)
	}
	client, err := minio.New(strings.TrimSuffix(endpoint, "/"), &minio.Options{Creds: credentials.NewStaticV4(config.AccessKey, config.SecretKey, ""), Secure: secure, Region: config.Region, BucketLookup: func() minio.BucketLookupType {
		if config.ForcePathStyle {
			return minio.BucketLookupPath
		}
		return minio.BucketLookupAuto
	}()})
	if err != nil {
		return nil, err
	}
	return &s3Storage{client: client, bucket: config.Bucket, baseURL: strings.TrimSuffix(config.BaseUrl, "/"), prefix: strings.Trim(config.PathPrefix, "/")}, nil
}
func (s *s3Storage) key(name string) string {
	if s.prefix == "" {
		return strings.TrimPrefix(name, "/")
	}
	return path.Join(s.prefix, strings.TrimPrefix(name, "/"))
}
func (s *s3Storage) unkey(key string) string {
	return strings.TrimPrefix(strings.TrimPrefix(key, s.prefix), "/")
}
func (s *s3Storage) file(key string, size int64) *system.StoredFile {
	name := s.unkey(key)
	rawURL := s.baseURL + "/" + key
	return &system.StoredFile{Name: path.Base(name), Path: name, URL: rawURL, Size: size}
}
func (s *s3Storage) Put(ctx context.Context, name string, reader io.Reader) (*system.StoredFile, error) {
	key := s.key(name)
	info, err := s.client.PutObject(ctx, s.bucket, key, reader, -1, minio.PutObjectOptions{})
	if err != nil {
		return nil, err
	}
	return s.file(key, info.Size), nil
}
func (s *s3Storage) Open(ctx context.Context, name string) (io.ReadCloser, error) {
	obj, err := s.client.GetObject(ctx, s.bucket, s.key(name), minio.GetObjectOptions{})
	if err != nil {
		return nil, err
	}
	if _, err = obj.Stat(); err != nil {
		_ = obj.Close()
		return nil, err
	}
	return obj, nil
}
func (s *s3Storage) Delete(ctx context.Context, name string) error {
	return s.client.RemoveObject(ctx, s.bucket, s.key(name), minio.RemoveObjectOptions{})
}
func (s *s3Storage) Compose(ctx context.Context, names []string, destination string) (*system.StoredFile, string, error) {
	return composeFiles(ctx, s, names, destination)
}
func (s *s3Storage) DeletePrefix(ctx context.Context, prefix string) error {
	prefix, err := normalizeDeletePrefix(prefix)
	if err != nil {
		return err
	}
	items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: boundedPrefix(s.key(prefix+"/"), prefix+"/"), Recursive: true})
	for item := range items {
		if item.Err != nil {
			return item.Err
		}
		if err := s.client.RemoveObject(ctx, s.bucket, item.Key, minio.RemoveObjectOptions{}); err != nil {
			return err
		}
	}
	return nil
}
func (s *s3Storage) List(ctx context.Context, prefix, cursor string, limit int) ([]*system.StoredFile, string, bool, error) {
	if limit <= 0 {
		limit = 100
	}
	items := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: s.key(prefix), Recursive: true, StartAfter: s.key(cursor)})
	out := make([]*system.StoredFile, 0, limit+1)
	for item := range items {
		if item.Err != nil {
			return nil, "", false, item.Err
		}
		file := s.file(item.Key, item.Size)
		file.LastModified = item.LastModified
		file.ContentType = item.ContentType
		out = append(out, file)
		if len(out) > limit {
			break
		}
	}
	more := len(out) > limit
	if more {
		out = out[:limit]
	}
	next := ""
	if len(out) > 0 {
		next = out[len(out)-1].Path
	}
	return out, next, more, nil
}