package importcfg

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

	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	corepb "github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dfconfig "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/pingcap/util/config"
	"github.com/tidbcloud/pkgv2/bizerr"
	"k8s.io/utils/ptr"
)

const (
	DefaultSQLFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.sql(?:\.(\w+))?$`
	DefaultCSVFilePatternWithTargetTable            = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.csv(?:\.(\w+))?$`
	DefaultRawParquetFilePatternWithTargetTable     = `(?i)^%s.%s(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
	DefaultAuroraSnapshotFilePatternWithTargetTable = `(?i)^%s/%s.%s/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`

	DefaultDBSchemaFilePattern    = `(?i)^(?:[^/]*/)*([^/.]+)-schema-create\.sql(?:\.(\w+))?$`
	DefaultTableSchemaFilePattern = `(?i)^(?:[^/]*/)*([^/.]+)\.(.*?)-schema\.sql(?:\.(\w+))?$`

	DefaultFilePattern = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?\.%s(?:\.(\w+))?$`

	DefaultAuroraSnapshotFilePattern = `(?i)^(?:[^/]*/)*([a-z0-9_]+)\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\-_.]+\.parquet)$`
	DefaultRawParquetFilePattern     = `(?i)^(?:[^/]*/)*([^/.]+)\.([^/.]+?)(?:(?:\.\d+)|(?:\.?(?:_\d+){3}))?(?:\.(?:snappy|gz|zst|zstd|gzip))?\.parquet$`
)

var (
	DefaultSQLFilePattern = fmt.Sprintf(DefaultFilePattern, "sql")
	DefaultCSVFilePattern = fmt.Sprintf(DefaultFilePattern, "csv")
)

func buildSQLFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(DefaultSQLFilePatternWithTargetTable, schema, table)
}

func buildCSVFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(DefaultCSVFilePatternWithTargetTable, schema, table)
}

func buildRawParquetFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(DefaultRawParquetFilePatternWithTargetTable, schema, table)
}

func buildAuroraSnapshotFilePatternByTargetTable(schema, table string) string {
	return fmt.Sprintf(DefaultAuroraSnapshotFilePatternWithTargetTable, schema, schema, table)
}

func GetLightningConfigTemplateByCluster(ctx context.Context, configManager config_manager.ConfigCenterManager, cluster *core.Cluster) (*cloudv1.Configuration, error) {
	return getLightningConfigTemplate(ctx, configManager, cluster)
}

func getLightningConfigTemplate(ctx context.Context, configManager config_manager.ConfigCenterManager, cluster *core.Cluster) (*cloudv1.Configuration, error) {
	failpoint.Inject("mockLightningCfg", func() {
		baseImage, tag := "baseimage", "v1"
		failpoint.Return(&cloudv1.Configuration{
			BaseImage: &baseImage,
			Tag:       &tag,
			Config: &cloudv1.TomlConfig{
				MP: map[string]interface{}{},
			},
		}, nil)
	})

	if cluster.GetAttributes().GetUseNewDedicatedArch() {
		getConfigReq := &domain.GetConfigVersionReq{
			ClusterID: cluster.Id,
			Component: domain.ConfigComponentLightning,
		}
		conf, bizErr := configManager.GetConfigVersion(ctx, getConfigReq)
		if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, errors.Annotate(bizErr, "failed get lightning config")
		}
		if bizErr != nil {
			initConfigReq := &domain.InitConfigVersionReq{
				ClusterID:     cluster.Id,
				TenantID:      cluster.TenantId,
				Component:     domain.ConfigComponentLightning,
				Version:       cluster.Version,
				CloudProvider: domain.CloudProviderType(cluster.CloudProviderName),
			}
			conf, bizErr = configManager.GetAndInitConfigVersion(ctx, initConfigReq)
			if bizErr != nil {
				return nil, errors.Annotate(bizErr, "failed init and get lightning config")
			}
		}

		impConfigTOML := new(cloudv1.TomlConfig)
		if err := impConfigTOML.UnmarshalTOML([]byte(conf.Config)); err != nil {
			return nil, errors.Annotate(err, "parse lightning config TOML data error")
		}
		return &cloudv1.Configuration{
			BaseImage: ptr.To(conf.BaseImage),
			Tag:       ptr.To(conf.Tag),
			Config:    impConfigTOML,
		}, nil
	}
	panic("not support cluster with old dedicated arch")
}

func getFilePatternsInConfig(cfg *config.GenericConfig) []map[string]string {
	val := cfg.Get("mydumper.files")
	if val == nil {
		return nil
	}
	jsonBytes, err := json.Marshal(val.Interface())
	if err != nil {
		return nil
	}
	result := []map[string]string{}
	if err := json.Unmarshal(jsonBytes, &result); err != nil {
		return nil
	}
	return result
}

func GenerateImportConfig(ctx context.Context, configCenter config_manager.ConfigCenterManager, imp entity.Import, sourceURL string, cluster *core.Cluster) (*cloudv1.Configuration, []byte, error) {
	// the git-ops config will be override by input of user
	lightningCfg, err := getLightningConfigTemplate(ctx, configCenter, cluster)
	if err != nil {
		return nil, nil, errors.Annotatef(err, "failed to get lightning config template")
	}

	cfg := config.New(lightningCfg.Config.MP)

	cfg, err = setTidbConfig(cfg, cluster)
	if err != nil {
		return nil, nil, errors.Annotatef(err, "failed to set tidb config")
	}

	// seems not used, maybe remove later
	if imp.OptionalArgs != nil {
		switch imp.OptionalArgs.DuplicationHandlingForSQL {
		case "replace", "ignore", "error":
			cfg.Set("tikv-importer.on-duplicate", imp.OptionalArgs.DuplicationHandlingForSQL)
		}
		if imp.OptionalArgs.SkipChecksum {
			cfg.Set("post-restore.checksum", false)
		}
		if imp.OptionalArgs.StoreWriteBWLimit > 0 {
			cfg.Set("tikv-importer.store-write-bwlimit", imp.OptionalArgs.StoreWriteBWLimit)
		}
	}

	cfg.Set("tidb.password", imp.TargetPassword)
	cfg.Set("mydumper.data-source-dir", sourceURL)

	cfg, err = setCsvCfg(cfg, imp)
	if err != nil {
		return nil, nil, errors.Annotatef(err, "failed to set csv config")
	}
	cfg, err = setFilePattern(cfg, imp)
	if err != nil {
		return nil, nil, errors.Annotatef(err, "failed to set file pattern")
	}

	cfg = setFilter(cfg, imp.TargetTableInfos.Infos)
	cfgbs, err := cfg.MarshalTOML()
	if err != nil {
		return nil, nil, errors.Annotatef(err, "failed to set marshal config to toml formate")
	}
	return lightningCfg, cfgbs, nil
}

func setTidbConfig(cfg *config.GenericConfig, cluster *core.Cluster) (*config.GenericConfig, error) {
	cloneCfg := cfg.DeepCopy()

	var (
		// these config are set in https://github.com/tidbcloud/tidb-configuration/blob/1620db7fe562fc17f401677e4ea49172f322f969/versions/v7.1.0/base/lightning.toml#L29-L35
		// TODO: maybe do not set them by code
		tidbPort     = int32(4000)
		tidbHost     = "db-tidb"
		tidbUserName = "cloud_admin"
		pdAddr       = "db-pd:2379"
		backend      = "local"
	)

	switch cluster.InfraProviderType {
	case cloudv1.ProviderTypeAWS, cloudv1.ProviderTypeGCP:
		for _, connection := range cluster.Connections {
			// gardener cluster will use the public port
			if connection.Type == corepb.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC {
				tidbPort = connection.Port
				break
			}
		}
	case cloudv1.ProviderTypeAWSFreeTier:
		// these fields will be override by https://github.com/tidbcloud/aws-shared-provider/blob/63ab5a4bf581bd936a4365fb24a4ef356abaff97/pkg/controllers/imports/env.go#L180
		// NOTE: override not work for precheck
		// TODO: remove the aws-shared-provider code
		// TODO: replace of hard code
		tidbHost = "tidb-gateway-inner.tidb-admin.svc.cluster.local"
		backend = "tidb"
		pdAddr = "serverless-cluster-pd.tidb-serverless.svc.cluster.local:2379"
		if cluster.ServerlessToken != nil {
			tidbUserName = *cluster.ServerlessToken + ".cloud_admin"
		} else {
			return nil, errors.New("serverless token is nil")
		}
		cloneCfg.Del("security")
	case cloudv1.ProviderTypeEKS, cloudv1.ProviderTypeGKE, cloudv1.ProviderTypeAKS:
		// new arch cluster will always use 4000 port
	default:
		return nil, errors.Errorf("not support infra provider type: %s", cluster.InfraProviderType)
	}

	if importConfig := dfconfig.Global().ImportConfig.TaskConfigs.GetMatchedConfig(
		cluster.GetTenantId(), cluster.GetProjectId(), cluster.GetId()); importConfig != nil {
		if importConfig.TiKVBackend != nil {
			backend = *importConfig.TiKVBackend
		}
	}

	// these fields in gitops will be override
	if tidbUserName != "" {
		cloneCfg.Set("tidb.user", tidbUserName)
	}
	if tidbHost != "" {
		cloneCfg.Set("tidb.host", tidbHost)
	}
	if tidbPort != 0 {
		cloneCfg.Set("tidb.port", tidbPort)
	}
	if pdAddr != "" {
		cloneCfg.Set("tidb.pd-addr", pdAddr)
	}
	if backend != "" {
		cloneCfg.Set("tikv-importer.backend", backend)
	}

	return cloneCfg, nil
}

func setCsvCfg(cfg *config.GenericConfig, imp entity.Import) (*config.GenericConfig, error) {
	if imp.DataFormat != entity.ImportDataFormatCsv || imp.CustomCSVFormat == "" {
		return cfg, nil
	}
	cloneCfg := cfg.DeepCopy()

	var csvCfg dataflow.CustomCSVFormat
	if err := json.Unmarshal([]byte(imp.CustomCSVFormat), &csvCfg); err != nil {
		return nil, err
	}
	cloneCfg.Set("mydumper.csv.separator", csvCfg.Separator)
	cloneCfg.Set("mydumper.csv.delimiter", csvCfg.Delimiter)
	cloneCfg.Set("mydumper.csv.header", csvCfg.Header)
	cloneCfg.Set("mydumper.csv.backslash-escape", csvCfg.BackslashEscape)
	if csvCfg.Null != nil {
		cloneCfg.Set("mydumper.csv.null", *csvCfg.Null)
	}
	if csvCfg.NotNull != nil {
		cloneCfg.Set("mydumper.csv.not-null", *csvCfg.NotNull)
	}
	if csvCfg.TrimLastSeparator != nil {
		cloneCfg.Set("mydumper.csv.trim-last-separator", *csvCfg.TrimLastSeparator)
	}
	if csvCfg.HeaderSchemaMatch != nil {
		cloneCfg.Set("mydumper.csv.header-schema-match", *csvCfg.HeaderSchemaMatch)
	}
	return cloneCfg, nil
}

func setFilePattern(cfg *config.GenericConfig, imp entity.Import) (*config.GenericConfig, error) {
	cloneCfg := cfg.DeepCopy()

	filePatterns := make([]map[string]string, 0)
	if len(imp.TargetTableInfos.Infos) == 0 {
		// the file pattern from git-ops is prior to the default file pattern
		filePatterns = append(filePatterns, getFilePatternsInConfig(cfg)...)
		filePatterns = append(filePatterns, setFilePatternWithNoTargetTable(imp)...)
	} else {
		filePatternsWithTargetTable, err := setFilePatternWithTargetTable(imp)
		if err != nil {
			return nil, err
		}
		// the file pattern from users is prior to the file pattern from git-ops
		filePatterns = append(filePatterns, filePatternsWithTargetTable...)
		filePatterns = append(filePatterns, getFilePatternsInConfig(cfg)...)
	}

	cloneCfg.Set("mydumper.files", filePatterns)
	return cloneCfg, nil
}

func setFilePatternWithTargetTable(imp entity.Import) ([]map[string]string, error) {
	filePatterns := make([]map[string]string, 0)

	sourceUrl, err := url.Parse(imp.SourceURL)
	if err != nil {
		return nil, errors.Annotate(err, "failed to parse source url")
	}
	sourceUrl.RawQuery = ""

	// custom file pattern should have a higher priority
	for _, info := range imp.TargetTableInfos.Infos {
		if len(info.CustomFile) == 0 {
			filePattern := map[string]string{
				"schema": info.Schema,
				"table":  info.Table,
			}
			switch imp.DataFormat {
			case entity.ImportDataFormatSql:
				filePattern["pattern"] = buildSQLFilePatternByTargetTable(info.Schema, info.Table)
				filePattern["type"] = "sql"
				filePattern["compression"] = "$1"
			case entity.ImportDataFormatCsv:
				filePattern["pattern"] = buildCSVFilePatternByTargetTable(info.Schema, info.Table)
				filePattern["type"] = "csv"
				filePattern["compression"] = "$1"
			case entity.ImportDataFormatAuroraSnapshot:
				filePattern["pattern"] = buildAuroraSnapshotFilePatternByTargetTable(info.Schema, info.Table)
				filePattern["type"] = "parquet"
			case entity.ImportDataFormatParquet:
				filePattern["pattern"] = buildRawParquetFilePatternByTargetTable(info.Schema, info.Table)
				filePattern["type"] = "parquet"
			}
			filePatterns = append(filePatterns, filePattern)

			continue
		}

		pattern := genCustomNamingRuleRegexp(strings.TrimPrefix(info.CustomFile, sourceUrl.String()))
		filePattern := map[string]string{
			"pattern": pattern,
			"schema":  info.Schema,
			"table":   info.Table,
			"type":    entity.ConvertDataFormat2Str(imp.DataFormat),
		}
		if compression := getFileCompression(info.CustomFile); compression != "" {
			filePattern["compression"] = compression
		}
		filePatterns = append(filePatterns, filePattern)
	}

	return filePatterns, nil
}

func getFileCompression(fileName string) string {
	var compression string
	for _, c := range []string{"gzip", "gz", "snappy", "zstd", "zst"} {
		if strings.HasSuffix(fileName, "."+c) {
			compression = c
			break
		}
	}
	return compression
}

func setFilePatternWithNoTargetTable(imp entity.Import) []map[string]string {
	filePatterns := make([]map[string]string, 0)

	// append the default file pattern
	fileSchema := map[string]string{
		"pattern":     DefaultDBSchemaFilePattern,
		"schema":      "$1",
		"compression": "$2",
		"type":        "schema-schema",
	}
	fileTableSchema := map[string]string{
		"pattern":     DefaultTableSchemaFilePattern,
		"schema":      "$1",
		"table":       "$2",
		"compression": "$3",
		"type":        "table-schema",
	}
	filePatterns = append(filePatterns, fileSchema, fileTableSchema)

	switch imp.DataFormat {
	case entity.ImportDataFormatSql:
		filePatterns = append(filePatterns, map[string]string{
			"pattern":     DefaultSQLFilePattern,
			"schema":      "$1",
			"table":       "$2",
			"type":        "sql",
			"compression": "$3",
		})
	case entity.ImportDataFormatCsv:
		filePatterns = append(filePatterns, map[string]string{
			"pattern":     DefaultCSVFilePattern,
			"schema":      "$1",
			"table":       "$2",
			"type":        "csv",
			"compression": "$3",
		})
	case entity.ImportDataFormatAuroraSnapshot:
		filePatterns = append(filePatterns, map[string]string{
			"pattern": DefaultAuroraSnapshotFilePattern,
			"schema":  "$1",
			"table":   "$2",
			"type":    "parquet",
		})
	case entity.ImportDataFormatParquet:
		filePatterns = append(filePatterns, map[string]string{
			"pattern": DefaultRawParquetFilePattern,
			"schema":  "$1",
			"table":   "$2",
			"type":    "parquet",
		})
	}

	return filePatterns
}

func setFilter(cfg *config.GenericConfig, targetInfos []entity.TargetTableInfo) *config.GenericConfig {
	if len(targetInfos) == 0 {
		return cfg
	}

	cloneCfg := cfg.DeepCopy()

	filters := make([]string, 0, len(targetInfos))
	for _, info := range targetInfos {
		filters = append(filters, GenTableID(info.Schema, info.Table))
	}

	cloneCfg.Set("mydumper.filter", filters)

	return cloneCfg
}

func GenTableID(schema, table string) string {
	return fmt.Sprintf("`%s`.`%s`", EscapeName(schema), EscapeName(table))
}

// EscapeName replaces all "`" in name with double "`"
func EscapeName(name string) string {
	return strings.Replace(name, "`", "``", -1)
}