/* * Copyright 2025 coze-dev Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package s3 import ( "bytes" "context" "fmt" "io" "time" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/coze-dev/coze-studio/backend/infra/contract/storage" "github.com/coze-dev/coze-studio/backend/infra/impl/storage/proxy" "github.com/coze-dev/coze-studio/backend/pkg/logs" ) type s3Client struct { client *s3.Client bucketName string } func New(ctx context.Context, ak, sk, bucketName, endpoint, region string) (storage.Storage, error) { t, err := getS3Client(ctx, ak, sk, bucketName, endpoint, region) if err != nil { return nil, err } return t, nil } func getS3Client(ctx context.Context, ak, sk, bucketName, endpoint, region string) (*s3Client, error) { creds := credentials.NewStaticCredentialsProvider(ak, sk, "") customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) { return aws.Endpoint{ PartitionID: "aws", URL: endpoint, SigningRegion: region, HostnameImmutable: true, Source: aws.EndpointSourceCustom, }, nil }) cfg, err := config.LoadDefaultConfig( context.TODO(), config.WithCredentialsProvider(creds), config.WithEndpointResolverWithOptions(customResolver), config.WithRegion("auto"), ) if err != nil { return nil, fmt.Errorf("init config failed, bucketName: %s, endpoint: %s, region: %s, err: %v", bucketName, endpoint, region, err) } c := s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = false // virtual-host mode o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired }) t := &s3Client{ client: c, bucketName: bucketName, } err = t.CheckAndCreateBucket(ctx) if err != nil { return nil, err } return t, nil } func (t *s3Client) test() { // test upload objectKey := fmt.Sprintf("test-%s.txt", time.Now().Format("20060102150405")) err := t.PutObject(context.Background(), objectKey, []byte("hello world")) if err != nil { logs.CtxErrorf(context.Background(), "PutObject failed, objectKey: %s, err: %v", objectKey, err) } // test download content, err := t.GetObject(context.Background(), objectKey) if err != nil { logs.CtxErrorf(context.Background(), "GetObject failed, objectKey: %s, err: %v", objectKey, err) } logs.CtxInfof(context.Background(), "GetObject content: %s", string(content)) // test get presigned url url, err := t.GetObjectUrl(context.Background(), objectKey) if err != nil { logs.CtxErrorf(context.Background(), "GetObjectUrl failed, objectKey: %s, err: %v", objectKey, err) } logs.CtxInfof(context.Background(), "GetObjectUrl url: %s", url) // test delete err = t.DeleteObject(context.Background(), objectKey) if err != nil { logs.CtxErrorf(context.Background(), "DeleteObject failed, objectKey: %s, err: %v", objectKey, err) } } func (t *s3Client) CheckAndCreateBucket(ctx context.Context) error { client := t.client bucket := t.bucketName _, err := client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(bucket)}) if err == nil { return nil // already exist } awsErr, ok := err.(interface{ ErrorCode() string }) if !ok || awsErr.ErrorCode() != "404" { return err } // bucket not exist input := &s3.CreateBucketInput{ Bucket: aws.String(bucket), } _, err = client.CreateBucket(ctx, input) return err } func (t *s3Client) PutObject(ctx context.Context, objectKey string, content []byte, opts ...storage.PutOptFn) error { opts = append(opts, storage.WithObjectSize(int64(len(content)))) return t.PutObjectWithReader(ctx, objectKey, bytes.NewReader(content), opts...) } func (t *s3Client) PutObjectWithReader(ctx context.Context, objectKey string, content io.Reader, opts ...storage.PutOptFn) error { client := t.client bucket := t.bucketName option := storage.PutOption{} for _, opt := range opts { opt(&option) } input := &s3.PutObjectInput{ Bucket: aws.String(bucket), Key: aws.String(objectKey), Body: content, } if option.ContentType != nil { input.ContentType = option.ContentType } if option.ContentEncoding != nil { input.ContentEncoding = option.ContentEncoding } if option.ContentDisposition != nil { input.ContentDisposition = option.ContentDisposition } if option.ContentLanguage != nil { input.ContentLanguage = option.ContentLanguage } if option.Expires != nil { input.Expires = option.Expires } if option.ObjectSize > 0 { input.ContentLength = aws.Int64(option.ObjectSize) } // upload object _, err := client.PutObject(ctx, input) return err } func (t *s3Client) GetObject(ctx context.Context, objectKey string) ([]byte, error) { client := t.client bucket := t.bucketName result, err := client.GetObject(ctx, &s3.GetObjectInput{ Bucket: aws.String(bucket), Key: aws.String(objectKey), }) if err != nil { return nil, fmt.Errorf("get object failed : %v", err) } defer result.Body.Close() body, err := io.ReadAll(result.Body) if err != nil { return nil, err } return body, nil } func (t *s3Client) DeleteObject(ctx context.Context, objectKey string) error { client := t.client bucket := t.bucketName _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{ Bucket: aws.String(bucket), Key: aws.String(objectKey), }) return err } func (t *s3Client) GetObjectUrl(ctx context.Context, objectKey string, opts ...storage.GetOptFn) (string, error) { client := t.client bucket := t.bucketName presignClient := s3.NewPresignClient(client) req, err := presignClient.PresignGetObject(ctx, &s3.GetObjectInput{ Bucket: aws.String(bucket), Key: aws.String(objectKey), }, func(options *s3.PresignOptions) { options.Expires = time.Duration(60*60*24) * time.Second }) if err != nil { return "", fmt.Errorf("get object presigned url failed: %v", err) } ok, proxyURL := proxy.CheckIfNeedReplaceHost(ctx, req.URL) if ok { return proxyURL, nil } return req.URL, nil } func (t *s3Client) ListObjects(ctx context.Context, prefix string) ([]*storage.FileInfo, error) { client := t.client bucket := t.bucketName const ( DefaultPageSize = 100 MaxListObjects = 10000 ) input := &s3.ListObjectsV2Input{ Bucket: aws.String(bucket), Prefix: aws.String(prefix), MaxKeys: aws.Int32(DefaultPageSize), } paginator := s3.NewListObjectsV2Paginator(client, input) var files []*storage.FileInfo for paginator.HasMorePages() { page, err := paginator.NextPage(ctx) if err != nil { return nil, fmt.Errorf("failed to get page, %v", err) } for _, obj := range page.Contents { f := &storage.FileInfo{} if obj.Key != nil { f.Key = *obj.Key } if obj.LastModified != nil { f.LastModified = *obj.LastModified } if obj.ETag != nil { f.ETag = *obj.ETag } if obj.Size != nil { f.Size = *obj.Size } files = append(files, f) } if len(files) >= MaxListObjects { logs.CtxErrorf(ctx, "[ListObjects] max list objects reached, total: %d", len(files)) break } } return files, nil } func (t *s3Client) ListObjectsPaginated(ctx context.Context, input *storage.ListObjectsPaginatedInput) (*storage.ListObjectsPaginatedOutput, error) { if input == nil { return nil, fmt.Errorf("input cannot be nil") } if input.PageSize <= 0 { return nil, fmt.Errorf("page size must be positive") } client := t.client bucket := t.bucketName listObjectsInput := &s3.ListObjectsV2Input{ Bucket: aws.String(bucket), Prefix: aws.String(input.Prefix), MaxKeys: aws.Int32(int32(input.PageSize)), ContinuationToken: aws.String(input.Cursor), } p, err := client.ListObjectsV2(ctx, listObjectsInput) if err != nil { return nil, err } var files []*storage.FileInfo for _, obj := range p.Contents { f := &storage.FileInfo{} if obj.Key != nil { f.Key = *obj.Key } if obj.LastModified != nil { f.LastModified = *obj.LastModified } if obj.ETag != nil { f.ETag = *obj.ETag } if obj.Size != nil { f.Size = *obj.Size } files = append(files, f) } output := &storage.ListObjectsPaginatedOutput{ Files: files, } if p.IsTruncated != nil { output.IsTruncated = *p.IsTruncated } if p.NextContinuationToken != nil { output.Cursor = *p.NextContinuationToken } return output, nil }