package storage

import (
	"context"
	"fmt"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/br/pkg/mock"
	s3pkg "github.com/tidbcloud/dataflow-service/pkg/s3"
)

type S3Client struct {
	s3iface.S3API
}

func NewMockS3Client(mockS3 *mock.MockS3API) *S3Client {
	return &S3Client{mockS3}
}

func NewS3Client(
	ctx context.Context,
	s3URL string,
	roleArn string, externalID string,
	accessKeyId string, secretAccessKey string,
) (StorageClient, error) {
	// TODO: CreateClient maybe removed to this pkg
	client, err := s3pkg.CreateClient(ctx, s3URL, &s3pkg.S3Option{
		RoleArn:         roleArn,
		ExternalID:      externalID,
		AccessKeyId:     accessKeyId,
		SecretAccessKey: secretAccessKey,
	})
	if err != nil {
		return nil, err
	}
	return &S3Client{client}, nil
}

func (c *S3Client) ListObjects(ctx context.Context, bucket, prefix string) ([]string, error) {
	getReq := s3.ListObjectsInput{
		Bucket: aws.String(bucket),
		Prefix: aws.String(prefix),
	}
	var out *s3.ListObjectsOutput
	var err error
	if out, err = c.ListObjectsWithContext(ctx, &getReq); err != nil {
		return nil, errors.Annotate(err, "failed to get object")
	}

	ret := make([]string, len(out.Contents))
	for i, obj := range out.Contents {
		ret[i] = fmt.Sprintf("s3://%s/%s", bucket, *obj.Key)
	}
	return ret, nil
}