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) { 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) } 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 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" }