package s3

import (
	"context"
	"fmt"
	"io"
	"net/url"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/pingcap/errors"
	"go.uber.org/multierr"
	"k8s.io/utils/pointer"
)

const (
	defaultRegionHint = "us-east-1"
)

type S3URL struct {
	URL    string
	Prefix string
	Bucket string

	region string
}

func NewS3URL(s3URL string) (*S3URL, error) {
	s3URL = strings.TrimSpace(s3URL)
	u, err := url.Parse(s3URL)
	if err != nil {
		return nil, errors.AddStack(err)
	}

	if u.Scheme != "s3" {
		return nil, errors.New("an invalid S3 endpoint URL")
	}
	bucket := u.Host
	prefix := u.Path
	prefix = strings.TrimLeft(prefix, "/")
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return &S3URL{
		URL:    s3URL,
		Prefix: prefix,
		Bucket: bucket,
		region: "",
	}, nil
}

// Only url starting with S3:// need to get the region
// see more here https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-bucket-intro.html
func (s *S3URL) GetBucketRegion(ctx context.Context) (string, error) {
	if s.region != "" {
		return s.region, nil
	}
	se := session.Must(session.NewSession())
	region, err := s3manager.GetBucketRegion(ctx, se, s.Bucket, defaultRegionHint)
	if err != nil {
		return "", err
	}
	s.region = region
	return s.region, nil
}

func (s *S3URL) GetBucketRegionWithSession(ctx context.Context, se *session.Session) (string, error) {
	if s.region != "" {
		return s.region, nil
	}
	region, err := s3manager.GetBucketRegion(ctx, se, s.Bucket, defaultRegionHint)
	if err != nil {
		return "", err
	}
	s.region = region
	return s.region, nil
}

// Check import need s3 access privileges
// see more in https://docs.pingcap.com/tidbcloud/config-s3-and-gcs-access#configure-amazon-s3-access
func CheckS3Access(
	ctx context.Context,
	s3URL, roleArn, externalID string,
	accessKeyId, secretAccessKey string,
) error {
	u, err := NewS3URL(s3URL)
	if err != nil {
		return err
	}

	client, err := CreateClient(ctx, s3URL, &S3Option{
		RoleArn:         roleArn,
		ExternalID:      externalID,
		AccessKeyId:     accessKeyId,
		SecretAccessKey: secretAccessKey,
	})
	if err != nil {
		return errors.AddStack(err)
	}

	// test s3:ListBucket
	if resp, err := client.ListObjectsV2WithContext(ctx, &s3.ListObjectsV2Input{
		Bucket:  aws.String(u.Bucket),
		Prefix:  aws.String(u.Prefix),
		MaxKeys: aws.Int64(1),
	}); err != nil {
		return errors.Annotate(err, "list objects error")
	} else if len(resp.Contents) == 0 {
		return errors.Errorf("there is no file in this bucket")
	} else {
		// test s3:GetObject
		if _, err := client.GetObjectWithContext(ctx, &s3.GetObjectInput{
			Bucket: &u.Bucket,
			Key:    resp.Contents[0].Key,
		}); err != nil {
			return errors.Annotate(err, "get object error")
		}
	}
	return nil
}

// DeleteS3Dir deletes all objects in the given S3 directory and the
// directory itself. If there are no matching directories or no objects
// in it, this function can still succeed.
func DeleteS3Dir(ctx context.Context, s3URL, bucket, folder string, roleARN, externalID string) error {
	client, err := CreateClient(ctx, s3URL, &S3Option{
		RoleArn:    roleARN,
		ExternalID: externalID,
	})
	if err != nil {
		return errors.Annotate(err, "failed to create client")
	}
	// check files exist
	listReq := s3.ListObjectsInput{
		Bucket:  aws.String(bucket),
		MaxKeys: pointer.Int64(1),
		Prefix:  aws.String(folder),
	}
	var out *s3.ListObjectsOutput
	if out, err = client.ListObjectsWithContext(ctx, &listReq); err != nil {
		return errors.AddStack(err)
	}
	if len(out.Contents) == 0 {
		return nil
	}
	// delete all
	iter := s3manager.NewDeleteListIterator(client, &s3.ListObjectsInput{
		Bucket: aws.String(bucket),
		Prefix: aws.String(folder),
	})
	if err := s3manager.NewBatchDeleteWithClient(client).Delete(ctx, iter); err != nil {
		return errors.AddStack(err)
	}
	return nil
}

// DeleteExpiredObject deletes all expired objects in the given S3 bucket and prefix.
func DeleteExpiredObject(
	ctx context.Context,
	region, bucket string,
	expiration time.Duration,
) error {
	client, err := CreateClientWithAssumeRole(region, "", "")
	if err != nil {
		return errors.Annotate(err, "failed to create client")
	}

	listReq := s3.ListObjectsInput{
		Bucket: aws.String(bucket),
	}
	var out *s3.ListObjectsOutput
	if out, err = client.ListObjectsWithContext(ctx, &listReq); err != nil {
		return errors.AddStack(err)
	}

	deletedObj := &s3.Delete{
		Objects: make([]*s3.ObjectIdentifier, 0),
	}
	// delete all
	for _, obj := range out.Contents {
		if obj.LastModified.Add(expiration).Before(time.Now()) {
			deletedObj.Objects = append(deletedObj.Objects, &s3.ObjectIdentifier{
				Key: obj.Key,
			})
		}
	}

	if len(deletedObj.Objects) == 0 {
		return nil
	}

	output, err := client.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
		Bucket: aws.String(bucket),
		Delete: deletedObj,
	})
	if err != nil {
		return errors.AddStack(err)
	}
	if len(output.Errors) == 0 {
		return nil
	}
	var errs error
	for _, s3Err := range output.Errors {
		errs = multierr.Append(errs, errors.New(s3Err.String()))
	}
	return errors.Annotate(errs, "failed delete some objects")
}

func CreateClientWithAssumeRole(region string, roleARN string, externalID string) (*s3.S3, error) {
	awsSessionOpts := session.Options{
		Config: *aws.NewConfig().WithRegion(region).WithCredentialsChainVerboseErrors(true),
	}

	ses, err := session.NewSessionWithOptions(awsSessionOpts)
	if err != nil {
		return nil, errors.AddStack(err)
	}

	if roleARN == "" {
		return s3.New(ses), nil
	}
	creds := stscreds.NewCredentials(ses, roleARN, func(p *stscreds.AssumeRoleProvider) {
		p.RoleARN = roleARN
		if externalID != "" {
			p.ExternalID = &externalID
		}
	})
	client := s3.New(ses, &aws.Config{Credentials: creds})

	return client, nil
}

func CreateClientWithStaticCredentials(region string, accessKeyId string, secretAccessKey string) (*s3.S3, error) {
	awsSessionOpts := session.Options{
		Config: *aws.NewConfig().WithRegion(region).WithCredentialsChainVerboseErrors(true),
	}

	ses, err := session.NewSessionWithOptions(awsSessionOpts)
	if err != nil {
		return nil, errors.AddStack(err)
	}

	creds := credentials.NewStaticCredentials(accessKeyId, secretAccessKey, "")

	client := s3.New(ses, &aws.Config{Credentials: creds})

	return client, nil
}

type S3Option struct {
	RoleArn         string
	ExternalID      string
	AccessKeyId     string
	SecretAccessKey string
}

func CreateClient(ctx context.Context, s3URL string, option *S3Option) (*s3.S3, error) {
	u, err := NewS3URL(s3URL)
	if err != nil {
		return nil, errors.Annotate(err, "parse s3 url")
	}

	region, err := u.GetBucketRegion(ctx)
	if err != nil {
		return nil, errors.Annotate(err, "get bucket region")
	}

	if option == nil {
		return CreateClientWithAssumeRole(region, "", "")
	}

	if option.AccessKeyId != "" && option.SecretAccessKey != "" {
		return CreateClientWithStaticCredentials(region, option.AccessKeyId, option.SecretAccessKey)
	} else if option.RoleArn != "" {
		return CreateClientWithAssumeRole(region, option.RoleArn, option.ExternalID)
	}

	return CreateClientWithAssumeRole(region, "", "")
}

func IsNotFoundError(err error) bool {
	if err == nil {
		return false
	}
	if strings.Contains(err.Error(), "failed to get region") {
		// regard as not found error
		return true
	}
	if err1, ok := (err).(awserr.Error); ok {
		switch err1.Code() {
		case s3.ErrCodeNoSuchBucket, s3.ErrCodeNoSuchKey, s3.ErrCodeNoSuchUpload:
			return true
		}
	}
	return false
}

func GetFileContent(ctx context.Context, s3Url, roleARN, externalID string, start, end int64) (string, error) {
	u, err := NewS3URL(s3Url)
	if err != nil {
		return "", err
	}

	client, err := CreateClient(ctx, s3Url, &S3Option{
		RoleArn:    roleARN,
		ExternalID: externalID,
	})
	if err != nil {
		return "", errors.Annotate(err, "failed to create client")
	}
	key := strings.TrimSuffix(u.Prefix, "/")
	getReq := s3.GetObjectInput{
		Bucket: aws.String(u.Bucket),
		Key:    aws.String(key),
		Range:  aws.String(fmt.Sprintf("bytes=%d-%d", start, end)),
	}
	var out *s3.GetObjectOutput
	if out, err = client.GetObjectWithContext(ctx, &getReq); err != nil {
		return "", errors.Annotate(err, "failed to get object")
	}
	ret := make([]byte, *out.ContentLength)
	if _, err = out.Body.Read(ret); err != nil && errors.Cause(err) != io.EOF {
		return "", errors.Annotate(err, "failed to read object")
	}
	return string(ret), nil
}

func GetFileBody(ctx context.Context, s3Url, roleARN, externalID string) (io.ReadCloser, error) {
	u, err := NewS3URL(s3Url)
	if err != nil {
		return nil, err
	}

	client, err := CreateClient(ctx, s3Url, &S3Option{
		RoleArn:    roleARN,
		ExternalID: externalID,
	})
	if err != nil {
		return nil, errors.Annotate(err, "failed to create client")
	}
	key := strings.TrimSuffix(u.Prefix, "/")
	getReq := s3.GetObjectInput{
		Bucket: aws.String(u.Bucket),
		Key:    aws.String(key),
	}
	var out *s3.GetObjectOutput
	if out, err = client.GetObjectWithContext(ctx, &getReq); err != nil {
		return nil, errors.Annotate(err, "failed to get object")
	}
	return out.Body, nil
}