package pimport

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

	"github.com/pingcap/errors"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"go.uber.org/zap"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
)

func ValidteS3URL(s string) error {
	u, err := url.Parse(s)
	if err != nil {
		return errors.Annotatef(err, "parse URL '%s' failed", s)
	}

	if u.Scheme != "s3" {
		return errors.Errorf("source url schema should be s3, but is: %s", u.Scheme)
	}

	return nil
}

func ValidateGSURL(s string) error {
	u, err := url.Parse(s)
	if err != nil {
		return errors.Annotatef(err, "parse URL '%s' failed", s)
	}

	if u.Scheme != "gs" {
		return errors.Errorf("source url schema should be gs, but is: %s", u.Scheme)
	}

	return nil
}

func ValidateAzureBlobURL(s string) error {
	u, err := url.Parse(s)
	if err != nil {
		return errors.Annotatef(err, "parse URL '%s' failed", s)
	}

	// s should be like:
	// https://<account-name>.blob.core.windows.net/<container-name>/<blob-prefix>/

	if u.Scheme != constants.AzblobScheme {
		return errors.Errorf("source url schema should be https, but is: %s", u.Scheme)
	}

	if !strings.HasSuffix(u.Host, ".blob.core.windows.net") {
		return errors.New("account-name is required for Azure Blob source")
	}

	return nil
}

func ValidateBucketURL(sourceType dataflow.ImportDataSource, sourceURL string) error {
	// todo: check url finely grained
	switch sourceType {
	case dataflow.ImportDataSource_S3_SOURCE:
		err := ValidteS3URL(sourceURL)
		if err != nil {
			return err
		}
	case dataflow.ImportDataSource_GCS_SOURCE:
		err := ValidateGSURL(sourceURL)
		if err != nil {
			return err
		}
	case dataflow.ImportDataSource_AZBLOB_SOURCE:
		if err := ValidateAzureBlobURL(sourceURL); err != nil {
			return err
		}
	case dataflow.ImportDataSource_LOCAL_SOURCE:
		return nil
	default:
		return errors.Errorf("unsupported source type type: %d", sourceType)
	}

	return nil
}

func validateReq(req *dataflow.ImportCreateReq, logger *zap.Logger) error {
	if req.LocalImportCfg != nil && req.LocalImportCfg.TargetTable == nil {
		return errors.New("Must set local import configuration when import local file.")
	}

	if err := ValidateBucketURL(dataflow.ImportDataSource(req.DataSourceType), req.SourceUrl); err != nil {
		return err
	}

	if req.ConnMethod == string(entity.ConnMethodPrivateLink) {
		if req.EndpointServiceName == nil || len(*req.EndpointServiceName) == 0 {
			return errors.New("The endpoint service name cannot be empty when the connMethod is private-link")
		}
	}

	switch dataflow.ImportDataFormat(req.DataFormat) {
	case dataflow.ImportDataFormat_SQL:
	case dataflow.ImportDataFormat_AURORA_SNAPSHOT:
	case dataflow.ImportDataFormat_PARQUET:
	case dataflow.ImportDataFormat_CSV:
		return validateCustomCSV(req.CsvFormat)
	default:
		return errors.Errorf("unsupported data format type: %d", req.DataFormat)
	}

	// if len(req.TargetTableInfos) == 0 {
	// 	return errors.New("Please select the target table.")
	// }

	for _, info := range req.TargetTableInfos {
		if info.TargetTable == nil {
			return errors.New("The target table can not be empty")
		}
	}

	if req.DataSourceType == uint32(dataflow.ImportDataSource_AZBLOB_SOURCE) {
		if req.AzblobSasToken == "" {
			return errors.New("The sas token cannot be empty.")
		}
	}

	return nil
}

func validateCustomCSV(csvFormat *dataflow.CustomCSVFormat) error {
	if csvFormat == nil {
		return errors.New("The csv configuration cannot be empty.")
	}

	if csvFormat.Separator == "" {
		return errors.New("The parameter(separator) value in the csv configuration cannot be empty. Please click and edit the 'CSV configuration' correctly.")
	}
	if csvFormat.Delimiter == "" {
		return errors.New("The parameter(delimiter) value in the csv configuration cannot be empty. Please click and edit the 'CSV configuration' correctly.")
	}

	return nil
}

func generateInvalidBucketURIMessage(uri string, dataSourceType dataflow.ImportDataSource) string {
	var exampleURI, uriType string
	switch dataSourceType {
	case dataflow.ImportDataSource_S3_SOURCE:
		uriType = "S3 URI"
		exampleURI = "s3://tidbcloud-samples/data-ingestion/"
	case dataflow.ImportDataSource_GCS_SOURCE:
		uriType = "gsutil URI"
		exampleURI = "gs://tidbcloud-samples/test/"
	case dataflow.ImportDataSource_AZBLOB_SOURCE:
		uriType = "Azure Blob URI"
		exampleURI = "https://<account-name>.blob.core.windows.net/<container-name>/<blob-prefix>/"
	default:
		return ""
	}
	return fmt.Sprintf("%s '%s' is not valid.  Please provide a correct %s, such as the template format '%s'",
		uriType, uri, uriType, exampleURI)
}

// checkValidationRequest checks the import validation request.
// The returned error's message will be exposed to the user, so the wording should be obvious.
// TODO: duplicate with validateReq, should merge them later.
func checkValidationRequest(ctx context.Context, req *dataflow.ImportValidateReq, logger *zap.Logger) error {
	if err := ValidateBucketURL(dataflow.ImportDataSource(req.DataSourceType), req.SourceUrl); err != nil {
		errMsg := generateInvalidBucketURIMessage(req.SourceUrl, dataflow.ImportDataSource(req.DataSourceType))
		logger.Error(errMsg, zap.Error(err))
		if err := bizutils.AppendErrorDetails(ctx, &errdetails.ErrorInfo{
			Reason: err.Error(),
		}); err != nil {
			logger.Error("append error detail error", zap.Error(err))
		}
		return errors.New(errMsg)
	}

	if dataflow.ImportDataSource(req.DataSourceType) == dataflow.ImportDataSource_S3_SOURCE {
		needCheckAuthArgs := false
	iterate_check_items:
		for _, checkItem := range req.CheckItems {
			switch checkItem {
			case dataflow.ImportValidationItem_SOURCE_ACCESS_CHECK,
				dataflow.ImportValidationItem_IMPORT_PRECHECK:
				needCheckAuthArgs = true
				break iterate_check_items
			}
		}
		if len(req.CheckItems) == 0 {
			// if no check items are provided, all the check will be executed
			needCheckAuthArgs = true
		}
		if needCheckAuthArgs {
			switch {
			case len(req.AwsAccessKeyId) > 0 && len(req.AwsSecretAccessKey) > 0:
				if err := validateAwsAkSk(req.AwsAccessKeyId, req.AwsSecretAccessKey); err != nil {
					return err
				}
			case len(req.AwsRoleArn) > 0:
				// todo: check AwsRoleArn content
			default:
				return errors.New("role_arn or aws_access_key_id/aws_secret_access_key must be set when data source is S3.")
			}
		}
	} else if dataflow.ImportDataSource(req.DataSourceType) == dataflow.ImportDataSource_AZBLOB_SOURCE {
		if req.AzblobSasToken == "" {
			return errors.New("The sas token cannot be empty.")
		}
	}

	if req.DataFormat == uint32(dataflow.ImportDataFormat_CSV) {
		if err := validateCustomCSV(req.CsvFormat); err != nil {
			return err
		}
	}

	// if len(req.TargetTableInfos) == 0 {
	// 	return errors.New("Please select the target table.")
	// }

	for _, info := range req.TargetTableInfos {
		if info.TargetTable == nil {
			return errors.New("The target table can not be empty")
		}
	}

	return nil
}

func validateAwsAkSk(ak, sk string) error {
	if len(ak) != 20 {
		return errors.New("Access Key ID must be 20 characters in length.")
	}

	// the detail of prefix: https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-about
	akPrefix := ak[:4]
	if akPrefix == "ASIA" {
		return errors.New("The import of TiDB Cloud doesn't support aws temporary credentials access keys temporarily.")
	}
	if akPrefix != "AKIA" {
		return errors.New("AWS long-term access key must start with AKIA.")
	}

	skPattern := regexp.MustCompile("^[0-9a-zA-z+/]{40}$")

	if !skPattern.MatchString(sk) {
		return errors.New("An AWS secret access key must consist of 40 characters, including '+', '/', numbers and letters.")
	}

	return nil
}

func validateCustomFile(sourceUri string, targetTableInfos []*dataflow.ImportTargetTableInfo) error {
	for _, info := range targetTableInfos {
		if info.CustomFile != "" && !strings.HasPrefix(info.CustomFile, sourceUri) {
			return errors.New(fmt.Sprintf("`%s`.`%s`'s custom file '%s' must be in the same directory as the source uri '%s'",
				info.TargetTable.GetSchema(), info.TargetTable.GetTable(), info.CustomFile, sourceUri))
		}
	}
	return nil
}