package pimport

import (
	"regexp"
	"strings"
	"unicode/utf8"

	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/service/utils"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/central"
)

var (
	ErrBadCSVFormat = errors.BadRequestf("bad csv format")

	targetTableRegexp = regexp.MustCompile("^(`.+`)\\.(`.+`)|(.+)\\.(.+)$")
)

func generateObjectNamePatternRegexp(wildcards string) string {
	// regexp syntax: https://en.wikipedia.org/wiki/Regular_expression
	var sb strings.Builder

	sb.WriteRune('^')

	cursor := 0
	maxlen := len(wildcards)

	for cursor < maxlen {
		r, sz := utf8.DecodeRuneInString(wildcards[cursor:])

		switch r {
		// defined in https://en.wikipedia.org/wiki/Glob_(programming)#Syntax
		case '{', '}', '(', ')', '^', '$', '.', '|', '+', '\\':
			sb.WriteString(`\`)
			sb.WriteRune(r)
		case '?':
			sb.WriteRune('.')
		case '*':
			sb.WriteString(".*")
		default:
			sb.WriteRune(r)
		}

		cursor += sz
	}

	sb.WriteRune('$')
	return sb.String()
}

func parseTargetTable(targetTable string) (string, string, error) {
	var result []string

	for _, i := range targetTableRegexp.FindStringSubmatch(targetTable) {
		if i != "" {
			result = append(result, i)
		}
	}

	if len(result) != 3 {
		return "", "", errors.Errorf("Target table pattern does not match the requirement, please refer the document.")
	}

	return result[1], result[2], nil
}

func validateReq(req *central.ImportCreateReq) error {
	err := utils.ValidateBucketURL(req.DataSourceType, req.SourceUrl)
	if err != nil {
		return err
	}

	// TODO(chenfei): check if the cluster is non-xaccount-deployed

	//if entity.DataSource(req.DataSourceType) == entity.DataSourceTypeS3 && req.AwsRoleArn == "" {
	//	return errors.Errorf("role ARN is required")
	//}

	switch entity.DataFormat(req.DataFormat) {
	case entity.DataFormatSQL:
	case entity.DataFormatParquet:
	case entity.DataFormatRawParquet:
	case entity.DataFormatCustomCSV:
		return validateCustomCSV(req.CsvFormat)
	default:
		return errors.Errorf("unsupported data format type: %d", req.DataFormat)
	}

	if (req.ObjectNamePattern == "" && req.TargetTablePattern != "") || (req.ObjectNamePattern != "" && req.TargetTablePattern == "") {
		return errors.Errorf("Object name pattern and target table should be provided at the same time")
	}

	if req.ObjectNamePattern != "" && req.TargetTablePattern != "" {
		switch entity.DataFormat(req.DataFormat) {
		case entity.DataFormatRawParquet, entity.DataFormatCustomCSV:
			// valid, skip this check
		default:
			return errors.Errorf("Object name pattern only support CSV and Parquet file formats.")
		}

		if _, _, err := parseTargetTable(req.TargetTablePattern); err != nil {
			return err
		}

		generatedRegexp := generateObjectNamePatternRegexp(req.ObjectNamePattern)
		if _, err := regexp.Compile(generatedRegexp); err != nil {
			return errors.Errorf("Object name pattern does not match the requirement, please refer the document.")
		}
	}

	return nil
}

func validateCustomCSV(csvFormat *central.CustomCSVFormat) error {
	if csvFormat == nil {
		return ErrBadCSVFormat
	}

	if csvFormat.Separator == "" {
		return errors.Annotatef(ErrBadCSVFormat, "invalid separator")
	}
	if csvFormat.Delimiter == "" {
		return errors.Annotatef(ErrBadCSVFormat, "invalid delimiter")
	}

	return nil
}