package pimport

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

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/request"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

// central need authorized to AssumeRole on roleARN
// or will meet error: *is not authorized to perform: sts:AssumeRole on*
func validateReq(
	ctx context.Context,
	sourceURL string,
	region string,
	roleARN string,
) (valid bool, reason string, err error) {
	client, err := createClient(region, roleARN)
	if err != nil {
		return false, "", err
	}

	valid, reason, err = validateS3Access(ctx, client, sourceURL, region)
	return
}

// s3client used for easy test
type s3client interface {
	ListObjectsWithContext(ctx aws.Context, input *s3.ListObjectsInput, opts ...request.Option) (*s3.ListObjectsOutput, error)
}

func validateS3Access(ctx context.Context, client s3client, rawURL string, region string) (valid bool, reason string, err error) {
	u, err := url.Parse(rawURL)
	if err != nil {
		return false, "invalid url: " + rawURL, errors.AddStack(err)
	}

	bucket := u.Host
	prefix := u.Path
	prefix = strings.TrimLeft(prefix, "/")
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}

	req := &s3.ListObjectsInput{
		Bucket:  aws.String(bucket),
		Prefix:  aws.String(prefix),
		MaxKeys: aws.Int64(1),
	}

	// For information on other S3 API error codes see:
	// http://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
	resp, err := client.ListObjectsWithContext(ctx, req)
	log.Debug("list objects resp", zap.Reflect("req", req), zap.Reflect("resp", resp), zap.Error(err))

	if err != nil {
		if aerr, ok := err.(awserr.Error); ok {
			switch aerr.Code() {
			case s3.ErrCodeNoSuchBucket:
				return false, fmt.Sprintf("bucket %s does not exist", bucket), nil
			case "BucketRegionError":
				return false, fmt.Sprintf("incorrect region, the bucket is not in '%s'", region), nil
			case "AccessDenied":
				// we can not make sure whether if it is valid or not if we can not assume the import role
				if strings.Contains(err.Error(), "sts:AssumeRole") {
					return false, "", errors.AddStack(err)
				}
				// when calling the ListObjectsV2 operation: Access Denied
				return false, "when calling the ListObjectsV2 operation: Access Denied", nil
				// see https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/http-400-bad-request.html
				// one example:
				// [error="AuthorizationHeaderMalformed: The authorization header is malformed; the region 'us-east-1' is wrong; expecting 'us-west-2'\n\ts tatus code: 400, request id: 7P5YDT2DAKD1S1S0, host id: JVa8HklZH6EIZek/IiB4OvdJyJU8ojZ76kv0Fl+8q4mk+7i9g6HsVFaazuRgN12wDdFWh7tQZmo="]
			case "AuthorizationHeaderMalformed":
				if strings.Contains(aerr.Message(), "the region") &&
					strings.Contains(aerr.Message(), "is wrong") {
					reason := strings.TrimPrefix(aerr.Message(), "The authorization header is malformed;")
					reason = strings.TrimSpace(reason)
					// the final reason return will like:
					// the region 'us-east-1' is wrong; expecting 'us-west-2'
					return false, reason, nil
				}
			}
		}
		return false, "", errors.AddStack(err)
	}

	// no any object at this prefix
	if !aws.BoolValue(resp.IsTruncated) && len(resp.Contents) == 0 {
		return false, fmt.Sprintf("no any objects under: s3://%s/%s", bucket, prefix), nil
	}

	return true, "", nil
}

func createClient(region string, roleARN string) (*s3.S3, error) {
	cfg := aws.NewConfig().WithRegion(region)

	awsSessionOpts := session.Options{
		Config: *cfg,
	}

	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) {})
	client := s3.New(ses, &aws.Config{Credentials: creds})

	return client, nil
}

func parseBucketName(rawURL string) string {
	u, err := url.Parse(rawURL)
	if err != nil {
		return ""
	}

	return u.Host
}

func isMyBuckets(region string, name string) (bool, error) {
	client, err := createClient(region, "")
	if err != nil {
		return false, err
	}

	output, err := client.ListBuckets(&s3.ListBucketsInput{})
	if err != nil {
		return false, errors.AddStack(err)
	}

	for _, b := range output.Buckets {
		// We need to skip tidbcloud-sample check for internal bucket check.
		if name == *b.Name && name != "tidbcloud-samples" {
			return true, nil
		}
	}

	return false, nil
}