geo/server/internal/shared/objectstorage/aliyun.go

package objectstorage

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
	"go.uber.org/zap"

	"github.com/geo-platform/tenant-api/internal/shared/config"
	"github.com/geo-platform/tenant-api/internal/shared/middleware"
)

type aliyunClient struct {
	client *oss.Client
	bucket string
	logger *zap.Logger
	cfg    config.ObjectStorageConfig
}

func NewAliyunClient(cfg config.ObjectStorageConfig, logger *zap.Logger) Client {
	endpoint := strings.TrimSpace(cfg.Endpoint)
	accessKey := strings.TrimSpace(cfg.AccessKey)
	secretKey := strings.TrimSpace(cfg.SecretKey)
	bucket := strings.TrimSpace(cfg.Bucket)
	region := strings.TrimSpace(cfg.Region)
	if endpoint == "" || accessKey == "" || secretKey == "" || bucket == "" || region == "" {
		return disabledClient{reason: "missing object_storage aliyun endpoint/access_key/secret_key/bucket/region"}
	}

	endpointURL, err := parseURLWithScheme(endpoint, boolScheme(cfg.UseSSL))
	if err != nil {
		return disabledClient{reason: fmt.Sprintf("parse aliyun endpoint failed: %v", err)}
	}

	client, err := oss.New(
		endpointURL.String(),
		accessKey,
		secretKey,
		oss.AuthVersion(oss.AuthV4),
		oss.Region(region),
	)
	if err != nil {
		return disabledClient{reason: fmt.Sprintf("init aliyun oss client failed: %v", err)}
	}

	return &aliyunClient{
		client: client,
		bucket: bucket,
		logger: logger,
		cfg:    cfg,
	}
}

func (c *aliyunClient) Validate() error {
	if c == nil || c.client == nil {
		return fmt.Errorf("%w: aliyun oss client is not initialized", ErrNotConfigured)
	}
	if strings.TrimSpace(c.bucket) == "" {
		return fmt.Errorf("%w: aliyun bucket is empty", ErrNotConfigured)
	}
	if strings.TrimSpace(c.cfg.Region) == "" {
		return fmt.Errorf("%w: aliyun region is empty", ErrNotConfigured)
	}
	return nil
}

func (c *aliyunClient) PutBytes(ctx context.Context, objectKey string, content []byte, contentType string) error {
	return c.PutReader(ctx, objectKey, bytes.NewReader(content), int64(len(content)), contentType)
}

func (c *aliyunClient) PutReader(
	ctx context.Context,
	objectKey string,
	reader io.Reader,
	size int64,
	contentType string,
) error {
	if err := c.Validate(); err != nil {
		return err
	}
	if strings.TrimSpace(objectKey) == "" {
		return fmt.Errorf("object key is required")
	}
	if reader == nil {
		return fmt.Errorf("object reader is required")
	}

	exists, err := c.bucketExists(ctx)
	if err != nil {
		c.logError(ctx, "aliyun bucket exists check failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return fmt.Errorf("check aliyun bucket: %w", err)
	}
	if !exists {
		if err := c.client.CreateBucket(c.bucket, oss.WithContext(ctx)); err != nil {
			c.logError(ctx, "aliyun create bucket failed",
				zap.String("bucket", c.bucket),
				zap.String("object_key", objectKey),
				zap.Error(err),
			)
			return fmt.Errorf("create aliyun bucket: %w", err)
		}
	}

	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		c.logError(ctx, "aliyun get bucket failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return fmt.Errorf("get aliyun bucket: %w", err)
	}

	options, err := aliyunPutObjectOptions(c.cfg, contentType)
	if err != nil {
		return err
	}
	options = append(options, oss.WithContext(ctx))
	if err := bucket.PutObject(objectKey, reader, options...); err != nil {
		c.logError(ctx, "aliyun put object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Int64("content_size", size),
			zap.String("object_acl", strings.TrimSpace(c.cfg.ObjectACL)),
			zap.String("content_type", contentType),
			zap.String("aliyun_error_code", aliyunErrorCode(err)),
			zap.String("aliyun_request_id", aliyunRequestID(err)),
			zap.Int("aliyun_status_code", aliyunStatusCode(err)),
			zap.Error(err),
		)
		return fmt.Errorf("put aliyun object: %w", err)
	}

	return nil
}

func (c *aliyunClient) bucketExists(ctx context.Context) (bool, error) {
	listRes, err := c.client.ListBuckets(oss.Prefix(c.bucket), oss.MaxKeys(1), oss.WithContext(ctx))
	if err != nil {
		return false, err
	}
	return len(listRes.Buckets) == 1 && listRes.Buckets[0].Name == c.bucket, nil
}

func aliyunPutObjectOptions(cfg config.ObjectStorageConfig, contentType string) ([]oss.Option, error) {
	options, _, err := aliyunPutObjectOptionsAndHeaders(cfg, contentType, "")
	return options, err
}

func aliyunPutObjectOptionsAndHeaders(
	cfg config.ObjectStorageConfig,
	contentType string,
	contentMD5Base64 string,
) ([]oss.Option, http.Header, error) {
	contentType = strings.TrimSpace(contentType)
	if contentType == "" {
		contentType = "application/octet-stream"
	}
	options := []oss.Option{oss.ContentType(contentType)}
	headers := http.Header{}
	headers.Set(oss.HTTPHeaderContentType, contentType)
	if contentMD5Base64 = strings.TrimSpace(contentMD5Base64); contentMD5Base64 != "" {
		options = append(options, oss.ContentMD5(contentMD5Base64))
		headers.Set(oss.HTTPHeaderContentMD5, contentMD5Base64)
	}
	acl := strings.ToLower(strings.TrimSpace(cfg.ObjectACL))
	switch acl {
	case "", "default":
		return options, headers, nil
	case "private":
		headers.Set(oss.HTTPHeaderOssObjectACL, string(oss.ACLPrivate))
		return append(options, oss.ObjectACL(oss.ACLPrivate)), headers, nil
	case "public-read":
		headers.Set(oss.HTTPHeaderOssObjectACL, string(oss.ACLPublicRead))
		return append(options, oss.ObjectACL(oss.ACLPublicRead)), headers, nil
	case "public-read-write":
		headers.Set(oss.HTTPHeaderOssObjectACL, string(oss.ACLPublicReadWrite))
		return append(options, oss.ObjectACL(oss.ACLPublicReadWrite)), headers, nil
	default:
		return nil, nil, fmt.Errorf("unsupported aliyun oss object_acl %q", cfg.ObjectACL)
	}
}

func aliyunErrorCode(err error) string {
	serviceErr, ok := err.(oss.ServiceError)
	if !ok {
		return ""
	}
	return serviceErr.Code
}

func aliyunRequestID(err error) string {
	serviceErr, ok := err.(oss.ServiceError)
	if !ok {
		return ""
	}
	return serviceErr.RequestID
}

func aliyunStatusCode(err error) int {
	serviceErr, ok := err.(oss.ServiceError)
	if !ok {
		return 0
	}
	return serviceErr.StatusCode
}

func (c *aliyunClient) GetBytes(ctx context.Context, objectKey string) ([]byte, error) {
	reader, err := c.GetReader(ctx, objectKey)
	if err != nil {
		return nil, err
	}
	defer reader.Close()

	data, err := io.ReadAll(reader)
	if err != nil {
		c.logError(ctx, "aliyun read object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return nil, fmt.Errorf("read aliyun object: %w", err)
	}

	return data, nil
}

func (c *aliyunClient) GetReader(ctx context.Context, objectKey string) (io.ReadCloser, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}

	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		c.logError(ctx, "aliyun get bucket failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return nil, fmt.Errorf("get aliyun bucket: %w", err)
	}

	reader, err := bucket.GetObject(objectKey, oss.WithContext(ctx))
	if err != nil {
		c.logError(ctx, "aliyun get object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		if isAliyunObjectNotFound(err) {
			return nil, fmt.Errorf("%w: %s", ErrObjectNotFound, objectKey)
		}
		return nil, fmt.Errorf("get aliyun object: %w", err)
	}
	return reader, nil
}

func isAliyunObjectNotFound(err error) bool {
	serviceErr, ok := err.(oss.ServiceError)
	if !ok {
		return false
	}
	return serviceErr.StatusCode == 404 ||
		serviceErr.Code == "NoSuchKey" ||
		serviceErr.Code == "NoSuchObject"
}

func (c *aliyunClient) Exists(ctx context.Context, objectKey string) (bool, error) {
	if err := c.Validate(); err != nil {
		return false, err
	}
	if strings.TrimSpace(objectKey) == "" {
		return false, nil
	}

	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		c.logError(ctx, "aliyun get bucket failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return false, fmt.Errorf("get aliyun bucket: %w", err)
	}

	exists, err := bucket.IsObjectExist(objectKey, oss.WithContext(ctx))
	if err != nil {
		c.logError(ctx, "aliyun stat object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return false, fmt.Errorf("stat aliyun object: %w", err)
	}
	return exists, nil
}

func (c *aliyunClient) Delete(ctx context.Context, objectKey string) error {
	if err := c.Validate(); err != nil {
		return err
	}
	if strings.TrimSpace(objectKey) == "" {
		return nil
	}

	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		c.logError(ctx, "aliyun get bucket failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return fmt.Errorf("get aliyun bucket: %w", err)
	}

	if err := bucket.DeleteObject(objectKey, oss.WithContext(ctx)); err != nil {
		c.logError(ctx, "aliyun delete object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		return fmt.Errorf("delete aliyun object: %w", err)
	}

	return nil
}

func (c *aliyunClient) PublicURL(objectKey string) (string, error) {
	if err := c.Validate(); err != nil {
		return "", err
	}
	publicRead, err := c.isPublicReadable()
	if err != nil {
		return "", err
	}
	if !publicRead {
		return c.signedURL(objectKey)
	}
	return buildPublicURL(c.cfg, objectKey, true)
}

func (c *aliyunClient) PublicReadableURL(objectKey string) (string, error) {
	if err := c.Validate(); err != nil {
		return "", err
	}
	publicRead, err := c.isPublicReadable()
	if err != nil {
		return "", err
	}
	if !publicRead {
		return "", ErrObjectNotPublic
	}
	return buildPublicURL(c.cfg, objectKey, true)
}

func (c *aliyunClient) DownloadURL(objectKey string) (string, error) {
	if err := c.Validate(); err != nil {
		return "", err
	}
	return c.signedURL(objectKey, oss.ResponseContentDisposition(downloadContentDisposition(objectKey)))
}

func (c *aliyunClient) PresignedPutURL(
	ctx context.Context,
	objectKey string,
	contentType string,
	contentMD5Base64 string,
	ttl time.Duration,
) (*PresignedPutResult, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}
	objectKey = normalizeObjectKeyPath(objectKey)
	if objectKey == "" {
		return nil, fmt.Errorf("object key is required")
	}
	if ttl <= 0 {
		ttl = c.cfg.SignedURLTTL
	}
	if ttl <= 0 {
		ttl = 30 * time.Minute
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		return nil, fmt.Errorf("get aliyun bucket: %w", err)
	}
	options, headers, err := aliyunPutObjectOptionsAndHeaders(c.cfg, contentType, contentMD5Base64)
	if err != nil {
		return nil, err
	}
	signedURL, err := bucket.SignURL(objectKey, oss.HTTPPut, int64(ttl.Seconds()), options...)
	if err != nil {
		return nil, fmt.Errorf("presign aliyun put url: %w", err)
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return &PresignedPutResult{
		URL:       signedURL,
		Headers:   headers,
		ExpiresAt: time.Now().UTC().Add(ttl),
	}, nil
}

func (c *aliyunClient) Usage(ctx context.Context) (*UsageInfo, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}
	stat, err := c.client.GetBucketStat(c.bucket, oss.WithContext(ctx))
	if err != nil {
		c.logError(ctx, "aliyun get bucket stat failed",
			zap.String("bucket", c.bucket),
			zap.Error(err),
		)
		return nil, fmt.Errorf("get aliyun bucket stat: %w", err)
	}
	return &UsageInfo{
		Available:   true,
		Provider:    "aliyun",
		StorageSize: stat.Storage,
		ObjectCount: stat.ObjectCount,
	}, nil
}

func (c *aliyunClient) List(ctx context.Context, in ListInput) (*ListResult, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}
	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		return nil, fmt.Errorf("get aliyun bucket: %w", err)
	}

	limit := in.Limit
	if limit <= 0 || limit > 1000 {
		limit = 100
	}
	delimiter := strings.TrimSpace(in.Delimiter)
	if delimiter == "" && !in.Recursive {
		delimiter = "/"
	}
	options := []oss.Option{
		oss.Prefix(normalizeObjectKeyPath(in.Prefix)),
		oss.MaxKeys(limit),
	}
	if delimiter != "" {
		options = append(options, oss.Delimiter(delimiter))
	}
	if token := strings.TrimSpace(in.ContinuationToken); token != "" {
		options = append(options, oss.ContinuationToken(token))
	}
	if startAfter := normalizeObjectKeyPath(in.StartAfter); startAfter != "" {
		options = append(options, oss.StartAfter(startAfter))
	}
	options = append(options, oss.WithContext(ctx))

	result, err := bucket.ListObjectsV2(options...)
	if err != nil {
		c.logError(ctx, "aliyun list objects failed",
			zap.String("bucket", c.bucket),
			zap.String("prefix", in.Prefix),
			zap.Error(err),
		)
		return nil, fmt.Errorf("list aliyun objects: %w", err)
	}

	out := &ListResult{
		Objects:               make([]ObjectInfo, 0, len(result.Objects)),
		CommonPrefixes:        append([]string(nil), result.CommonPrefixes...),
		ContinuationToken:     result.ContinuationToken,
		NextContinuationToken: result.NextContinuationToken,
		IsTruncated:           result.IsTruncated,
		Prefix:                result.Prefix,
		Delimiter:             result.Delimiter,
	}
	for _, item := range result.Objects {
		out.Objects = append(out.Objects, ObjectInfo{
			Key:          item.Key,
			Size:         item.Size,
			ETag:         strings.Trim(item.ETag, `"`),
			LastModified: item.LastModified,
			StorageClass: item.StorageClass,
		})
	}
	return out, nil
}

func (c *aliyunClient) Stat(ctx context.Context, objectKey string) (*ObjectInfo, error) {
	if err := c.Validate(); err != nil {
		return nil, err
	}
	objectKey = normalizeObjectKeyPath(objectKey)
	if objectKey == "" {
		return nil, fmt.Errorf("object key is required")
	}
	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		return nil, fmt.Errorf("get aliyun bucket: %w", err)
	}
	header, err := bucket.GetObjectDetailedMeta(objectKey, oss.WithContext(ctx))
	if err != nil {
		c.logError(ctx, "aliyun stat object failed",
			zap.String("bucket", c.bucket),
			zap.String("object_key", objectKey),
			zap.Error(err),
		)
		if isAliyunObjectNotFound(err) {
			return nil, fmt.Errorf("%w: %s", ErrObjectNotFound, objectKey)
		}
		return nil, fmt.Errorf("stat aliyun object: %w", err)
	}
	return aliyunHeaderToObjectInfo(objectKey, header), nil
}

func (c *aliyunClient) Copy(ctx context.Context, sourceKey, destinationKey string) error {
	if err := c.Validate(); err != nil {
		return err
	}
	sourceKey = normalizeObjectKeyPath(sourceKey)
	destinationKey = normalizeObjectKeyPath(destinationKey)
	if sourceKey == "" || destinationKey == "" {
		return fmt.Errorf("source and destination object keys are required")
	}
	if sourceKey == destinationKey {
		return nil
	}
	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		return fmt.Errorf("get aliyun bucket: %w", err)
	}
	if _, err := bucket.CopyObject(sourceKey, destinationKey, oss.WithContext(ctx)); err != nil {
		c.logError(ctx, "aliyun copy object failed",
			zap.String("bucket", c.bucket),
			zap.String("source_key", sourceKey),
			zap.String("destination_key", destinationKey),
			zap.Error(err),
		)
		if isAliyunObjectNotFound(err) {
			return fmt.Errorf("%w: %s", ErrObjectNotFound, sourceKey)
		}
		return fmt.Errorf("copy aliyun object: %w", err)
	}
	return nil
}

func aliyunHeaderToObjectInfo(objectKey string, header http.Header) *ObjectInfo {
	size, _ := strconv.ParseInt(strings.TrimSpace(header.Get(oss.HTTPHeaderContentLength)), 10, 64)
	lastModified, _ := http.ParseTime(header.Get(oss.HTTPHeaderLastModified))
	return &ObjectInfo{
		Key:          objectKey,
		Size:         size,
		ETag:         strings.Trim(header.Get(oss.HTTPHeaderEtag), `"`),
		LastModified: lastModified,
		ContentType:  header.Get(oss.HTTPHeaderContentType),
		StorageClass: header.Get(oss.HTTPHeaderOssStorageClass),
	}
}

func (c *aliyunClient) isPublicReadable() (bool, error) {
	objectACL := strings.ToLower(strings.TrimSpace(c.cfg.ObjectACL))
	if isPublicReadACL(objectACL) {
		return true, nil
	}
	if objectACL != "" && objectACL != "default" {
		return false, nil
	}

	bucketACL := strings.ToLower(strings.TrimSpace(c.cfg.BucketACL))
	if bucketACL != "" {
		return isPublicReadACL(bucketACL), nil
	}

	result, err := c.client.GetBucketACL(c.bucket)
	if err != nil {
		return false, fmt.Errorf("get aliyun bucket acl: %w", err)
	}
	return isPublicReadACL(result.ACL), nil
}

func (c *aliyunClient) signedURL(objectKey string, options ...oss.Option) (string, error) {
	objectKey = normalizeObjectKeyPath(objectKey)
	if objectKey == "" {
		return "", fmt.Errorf("object key is required")
	}
	bucket, err := c.client.Bucket(c.bucket)
	if err != nil {
		return "", fmt.Errorf("get aliyun bucket: %w", err)
	}
	ttl := c.cfg.SignedURLTTL
	if ttl <= 0 {
		ttl = 30 * time.Minute
	}
	return bucket.SignURL(objectKey, oss.HTTPGet, int64(ttl.Seconds()), options...)
}

func isPublicReadACL(acl string) bool {
	switch strings.ToLower(strings.TrimSpace(acl)) {
	case "public-read", "public-read-write":
		return true
	default:
		return false
	}
}

func (c *aliyunClient) logError(ctx context.Context, msg string, fields ...zap.Field) {
	if c == nil || c.logger == nil {
		return
	}
	if requestID := middleware.RequestIDFromContext(ctx); requestID != "" {
		fields = append(fields, zap.String("request_id", requestID))
	}
	c.logger.Error(msg, fields...)
}

func boolScheme(useSSL bool) string {
	if useSSL {
		return "https"
	}
	return "http"
}