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/request"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
)

// 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(ctx, "list objects resp", zap.Reflect("req", req), zap.Reflect("resp", resp), zap.Error(err))

	if err != nil {
		if aerr, ok := err.(awserr.Error); ok {
			log.Debug(ctx, "try to parse error of validate s3 access", zap.String("aerr.Code()", aerr.Code()), zap.String("aerr.Error()", aerr.Error()))
			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, "the source url is invalid or we can't access it", 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
}