package imports

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

	"github.com/pingcap/errors"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/consts"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
	"github.com/tidbcloud/infra-api/pkg/pingcap/util/config"
)

type TCNameGetter interface {
	// TCName get the tidbcluster name create by the specified cluster.
	TCName(clusterName string) string
}

func getLightningConfigTemplate(gitops gitopsv2.Interface, c *cloudv1.Cluster) (*cloudv1.TomlConfig, error) {
	cc, err := gitops.DescribeClusterConfiguration(c)
	if err != nil {
		return nil, err
	}
	if cc.Spec.Lightning != nil {
		return cc.Spec.Lightning.Config, nil
	}
	return &cloudv1.TomlConfig{}, nil
}

func buildSourceUrl(i *cloudv1.Import, c *cloudv1.Cluster) (string, error) {
	var url string
	var region string
	switch c.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeAWS, consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeEKS: // also for shared AWS clusters
		url = i.Spec.Source.Storage.S3.URL
		region = i.Spec.Source.Storage.S3.Region
	case consts.ProviderTypeGCP, consts.ProviderTypeGCPSharedTier, consts.ProviderTypeGKE:
		url = i.Spec.Source.Storage.GCS.URL
		region = i.Spec.Source.Storage.GCS.Region
	default:
		return "", fmt.Errorf("unknown provider %s", c.Labels[cloudv1.ProviderTypeLabel])
	}

	if strings.Contains(url, "?") {
		url = fmt.Sprintf("%s&region=%s", url, region)
	} else {
		url = fmt.Sprintf("%s?region=%s", url, region)
	}

	return url, nil
}

func genServerlessCfg(token, tidbUserName string) (string, string, string) {
	tidbUserName = token + "." + tidbUserName
	tidbHost := "tidb-gateway-inner.tidb-admin.svc.cluster.local"
	pdAddr := "serverless-cluster-pd.tidb-serverless.svc.cluster.local:2379"
	return tidbHost, tidbUserName, pdAddr
}

func newLightningConfig(source cloudv1.Source, lightningConfig cloudv1.TomlConfig, delTLS bool, sourceUrl, tcName string, port int32, tidbHost string, tidbUserName string, pdAddr string) (string, error) {
	if lightningConfig.MP == nil {
		return "", errors.New("lightning config can not be empty")
	}
	cfg := config.New(lightningConfig.MP)
	if delTLS {
		cfg.Del("security")
	}
	filter := source.Filters
	if len(filter) == 1 && filter[0] == "" {
		filter = nil
	}
	if len(filter) != 0 {
		cfg.Set("mydumper.filter", filter)
	}

	if tidbHost == "" {
		tidbHost = tcName + "-tidb"
	}

	if tidbUserName == "" {
		tidbUserName = source.UserName
	}

	if pdAddr == "" {
		pdAddr = tcName + "-pd:2379"
	}

	if pdAddr != "" {
		cfg.Set("tidb.pd-addr", pdAddr)
	}
	if tidbHost != "" {
		cfg.Set("tidb.host", tidbHost)
	}
	if port != 0 {
		cfg.Set("tidb.port", port)
	}
	cfg.Set("tidb.user", tidbUserName)
	cfg.Set("tidb.password", source.Password)

	cfg.Set("mydumper.data-source-dir", sourceUrl)

	// regexp comes from: https://github.com/pingcap/tidb/blob/e7b522b665e9527597c3eb6f5613e26099536b7b/br/pkg/lightning/mydump/router.go#L109-L115
	fileSchema := map[string]string{
		"pattern": `(?i)^(?:[^/]*/)*([^/.]+)-schema-create\.sql$`,
		"schema":  "$1",
		"type":    "schema-schema",
	}
	fileTableSchema := map[string]string{
		"pattern": `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema\.sql$`,
		"schema":  "$1",
		"table":   "$2",
		"type":    "table-schema",
	}

	var err error
	switch source.Format {
	case cloudv1.DataFormatSQL:
		fileSQL := map[string]string{
			"pattern": `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)(?:\.([0-9]+))?\.sql$`,
			"schema":  "$1",
			"table":   "$2",
			"type":    "sql",
		}

		cfg.Set("mydumper.files", []map[string]string{fileSchema, fileTableSchema, fileSQL})
	case cloudv1.DataFormatCSV:
		cfg.Set("mydumper.no-schema", false)
		// Lightning has default values for the CSV format, so when the field is not set,
		// we simply leave it unset.
		if source.CsvFormat.Separator != nil {
			cfg.Set("mydumper.csv.separator", *source.CsvFormat.Separator)
		}
		if source.CsvFormat.Delimiter != nil {
			cfg.Set("mydumper.csv.delimiter", *source.CsvFormat.Delimiter)
		}
		if source.CsvFormat.Header != nil {
			cfg.Set("mydumper.csv.header", *source.CsvFormat.Header)
		}
		if source.CsvFormat.NotNull != nil {
			cfg.Set("mydumper.csv.not-null", *source.CsvFormat.NotNull)
		}
		if source.CsvFormat.Null != nil {
			cfg.Set("mydumper.csv.null", *source.CsvFormat.Null)
		}
		if source.CsvFormat.BackslashEscape != nil {
			cfg.Set("mydumper.csv.backslash-escape", *source.CsvFormat.BackslashEscape)
		}
		if source.CsvFormat.TrimLastSeparator != nil {
			cfg.Set("mydumper.csv.trim-last-separator", *source.CsvFormat.TrimLastSeparator)
		}

		fileCSV := map[string]string{
			"pattern": `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)(?:\.([0-9]+))?\.csv$`,
			"schema":  "$1",
			"table":   "$2",
			"type":    "csv",
		}

		if source.ObjectNamePattern != "" && source.TargetTablePattern != "" {
			fileCSV, err = generateFileRoute(source.Format, source.ObjectNamePattern, source.TargetTablePattern)
			if err != nil {
				return "", err
			}
			fileSchema, fileTableSchema, err = generateSingleTableSchemaFileRoute(source.TargetTablePattern)
			if err != nil {
				return "", errors.Annotate(err, "generate single table schema file route failed")
			}
		}

		cfg.Set("mydumper.files", []map[string]string{fileSchema, fileTableSchema, fileCSV})
	case cloudv1.DataFormatAuroraParquet:
		cfg.Set("mydumper.no-schema", true)
		fileAuroraParquet := map[string]string{
			"pattern": `(?i)^(?:[^/]*/)*([a-z0-9_]+)\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.(parquet))$`,
			"schema":  "$1",
			"table":   "$2",
			"type":    "$3",
		}
		cfg.Set("mydumper.files", []map[string]string{fileSchema, fileTableSchema, fileAuroraParquet})
	case cloudv1.DataFormatRawParquet:
		cfg.Set("mydumper.no-schema", true)
		fileRawParquet := map[string]string{
			"pattern": `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)(?:\.([0-9]+))?(?:\.(?:snappy|gz|lzo))?\.parquet$`,
			"schema":  "$1",
			"table":   "$2",
			"key":     "$3",
			"type":    "parquet",
		}

		if source.ObjectNamePattern != "" && source.TargetTablePattern != "" {
			fileRawParquet, err = generateFileRoute(source.Format, source.ObjectNamePattern, source.TargetTablePattern)
			if err != nil {
				return "", err
			}
			fileSchema, fileTableSchema, err = generateSingleTableSchemaFileRoute(source.TargetTablePattern)
			if err != nil {
				return "", errors.Annotate(err, "generate single table schema file route failed")
			}
		}

		cfg.Set("mydumper.files", []map[string]string{fileSchema, fileTableSchema, fileRawParquet})
	default:
		cfg.Set("mydumper.no-schema", false)
	}

	data, err := cfg.MarshalTOML()
	if err != nil {
		return "", err
	}

	return string(data), nil
}

func generateSingleTableSchemaFileRoute(targetTablePattern string) (map[string]string, map[string]string, error) {
	dbName, tableName, err := cloudv1.ParseTargetTable(targetTablePattern)
	if err != nil {
		return nil, nil, err
	}

	return map[string]string{
			"pattern": fmt.Sprintf(`(?i)^(?:[^/]*/)*%s-schema-create\.sql$`, regexp.QuoteMeta(dbName)),
			"schema":  dbName,
			"type":    "schema-schema",
		}, map[string]string{
			"pattern": fmt.Sprintf(`(?i)^(?:[^/]*/)*%s\.%s-schema\.sql$`, regexp.QuoteMeta(dbName), regexp.QuoteMeta(tableName)),
			"schema":  dbName,
			"table":   tableName,
			"type":    "table-schema",
		}, nil
}

func generateFileRoute(format cloudv1.DataFormat, objectNamePattern, targetTablePattern string) (map[string]string, error) {
	var fileType string

	switch format {
	case cloudv1.DataFormatCSV:
		fileType = "csv"
	case cloudv1.DataFormatRawParquet:
		fileType = "parquet"
	}

	dbName, tableName, err := cloudv1.ParseTargetTable(targetTablePattern)
	if err != nil {
		return nil, err
	}

	pattern := generateObjectNamePatternRegexp(objectNamePattern)

	return map[string]string{
		"pattern": pattern,
		"schema":  dbName,
		"table":   tableName,
		"type":    fileType,
	}, nil
}

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 GenerateLightningConfig(gitops gitopsv2.Interface, c *cloudv1.Cluster, i *cloudv1.Import, tcNameGetter TCNameGetter, isDevTier bool) (string, error) {
	lightningCfgTemplate, err := getLightningConfigTemplate(gitops, c)
	if err != nil {
		return "", errors.Trace(err)
	}
	sourceUrl, err := buildSourceUrl(i, c)
	if err != nil {
		return "", errors.Trace(err)
	}
	tcName := tcNameGetter.TCName(c.Name)
	port := c.GetPublicSQLEndpointPort()
	delTLS := false

	if isDevTier {
		port = 4000
		delTLS = true
		token := c.Status.SqlEndpoints.PublicEndpoint.Token
		tidbHost, tidbUserName, pdAddr := genServerlessCfg(token, i.Spec.Source.UserName)
		if _, ok := c.Annotations["free-tier.tidbcloud.com/serverless"]; !ok {
			pdAddr = ""
		}
		return newLightningConfig(i.Spec.Source, *lightningCfgTemplate, delTLS, sourceUrl, tcName, port, tidbHost, tidbUserName, pdAddr)
	}
	return newLightningConfig(i.Spec.Source, *lightningCfgTemplate, delTLS, sourceUrl, tcName, port, "", "", "")
}