package objectstorage import ( "bytes" "context" "fmt" "io" "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 { if err := c.Validate(); err != nil { return err } if strings.TrimSpace(objectKey) == "" { return fmt.Errorf("object key is required") } exists, err := c.client.IsBucketExist(c.bucket) 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); 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 } if err := bucket.PutObject(objectKey, bytes.NewReader(content), options...); err != nil { c.logError(ctx, "aliyun put object failed", zap.String("bucket", c.bucket), zap.String("object_key", objectKey), zap.Int("content_size", len(content)), 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 aliyunPutObjectOptions(cfg config.ObjectStorageConfig, contentType string) ([]oss.Option, error) { options := []oss.Option{oss.ContentType(contentType)} acl := strings.ToLower(strings.TrimSpace(cfg.ObjectACL)) switch acl { case "", "default": return options, nil case "private": return append(options, oss.ObjectACL(oss.ACLPrivate)), nil case "public-read": return append(options, oss.ObjectACL(oss.ACLPublicRead)), nil case "public-read-write": return append(options, oss.ObjectACL(oss.ACLPublicReadWrite)), nil default: return 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) 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) 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); 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) 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) (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())) } 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" }