package exportaction

import (
	"bufio"
	"bytes"
	"context"
	"encoding/csv"
	"fmt"
	"net/url"
	"regexp"
	"strconv"
	"strings"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	CreateExportDumplingActionName = "create-export-dumpling"
	PollExportTiDBGroupActionName  = "poll-export-tidb-group"
	PollExportDumplingActionName   = "poll-export-dumpling"
	DeleteExportDumplingActionName = "delete-export-dumpling"
)

const (
	defaultTableFilter   = "!/^(mysql|sys|INFORMATION_SCHEMA|PERFORMANCE_SCHEMA|METRICS_SCHEMA|INSPECTION_SCHEMA|lightning_task_info|__TiDB_BR_Temporary_mysql|tidbcloud_dm_meta|tidb_cdc)$/.*"
	tsoPhysicalShiftBits = 18
)

type ExportDumplingConfig struct {
	Node               string
	CPURequest         string
	CPULimit           string
	MemoryRequest      string
	MemoryLimit        string
	TiDBGroupNodeCount uint32
	TiDBGroupNodeSpec  string
	Thread             int
}

func (cfg *ExportDumplingConfig) HasTiDBGroupScaleConfig() bool {
	if cfg == nil {
		return false
	}
	return cfg.TiDBGroupNodeCount > 0 || cfg.TiDBGroupNodeSpec != ""
}

func loadExportByArgs(ctx context.Context, exportRepo repo.ExportRepo, args actions.ActionArgs) (*entity.Export, error) {
	exportID, err := args.GetString(workflow.ExportIdKey)
	if err != nil {
		return nil, fmt.Errorf("failed to get export id from args: %w", err)
	}
	exportIns, err := exportRepo.Load(ctx, exportID, true)
	if err != nil {
		return nil, fmt.Errorf("failed to load export %s: %w", exportID, err)
	}
	return exportIns, nil
}

func buildDumplingJobName(export *entity.Export) string {
	return export.ExportId + "-dumpling"
}

func buildDumplingHost(tidbGroup *entity.TiDBGroup) string {
	serviceName := fmt.Sprintf("db-%s-tidb", tidbGroup.Id)
	namespace := fmt.Sprintf("tidb%s", tidbGroup.SharedPoolId)
	return fmt.Sprintf("%s.%s.svc.cluster.local", serviceName, namespace)
}

func buildDumplingCommand(
	exportIns *entity.Export,
	tidbGroup *entity.TiDBGroup,
	cluster *entity.TiDBCluster,
	sharedPool *entity.SharedPool,
	thread int,
) ([]string, error) {
	t := 2
	if thread > 0 {
		t = thread
	}
	command := []string{
		"-u=" + exportIns.GetUsername(),
		"-p=" + exportIns.GetPassword(),
		"-h=" + buildDumplingHost(tidbGroup),
		"-t=" + strconv.Itoa(t),
		"--filetype=" + dumplingFileType(exportIns.GetExportOptions().GetFileType()),
	}

	// get pd service name
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() != apiv1.ComponentType_PD {
			continue
		}
		pdAddress := fmt.Sprintf("%s.%s.svc:2379", entity.GetComponentSvcName(sharedPool, comp), sharedPool.GetDataPlaneNamespace())
		command = append(command, "--pd="+pdAddress)
		break
	}

	// build output
	outputURI, err := buildExportOutputURI(exportIns.GetCloudStorage(), cluster)
	if err != nil {
		return nil, err
	}
	command = append(command, "-o="+outputURI)
	exportOptions := exportIns.GetExportOptions()

	// build filter
	filterArgs, err := buildDumplingFilterArgs(exportOptions)
	if err != nil {
		return nil, err
	}
	command = append(command, filterArgs...)

	// build compression
	command = append(command, buildDumplingCompressionArgs(exportOptions)...)

	// build snapshot
	command = append(command, "--snapshot="+fmt.Sprintf("%d", exportIns.GetSnapshotTso()))

	if exportOptions != nil {
		command = append(command, buildDumplingEscapeBackslashArgs(exportOptions.EscapeBackslash)...)
	}

	// build csv format
	if exportOptions.GetFileType() == apiv1.ExportOptions_CSV && exportOptions.GetCsvFormat() != nil {
		csvFormat := exportOptions.GetCsvFormat()
		if csvFormat.GetSeparator() != "" {
			command = append(command, "--csv-separator="+csvFormat.GetSeparator())
		}
		if csvFormat.Delimiter != nil {
			command = append(command, "--csv-delimiter="+*csvFormat.Delimiter)
		}
		if csvFormat.NullValue != nil {
			command = append(command, "--csv-null-value="+*csvFormat.NullValue)
		}
		if csvFormat.GetSkipHeader() {
			command = append(command, "--no-header")
		}
		dialectArgs, err := buildDumplingCSVDialectArgs(csvFormat.Dialect)
		if err != nil {
			return nil, err
		}
		command = append(command, dialectArgs...)
	}

	if exportOptions.GetFileType() == apiv1.ExportOptions_PARQUET && exportOptions.GetParquetFormat() != nil {
		if compression := toDumplingParquetCompression(exportOptions.GetParquetFormat().GetCompression()); compression != "" {
			command = append(command, "--parquet-compress="+compression)
		}
	}

	return command, nil
}

func buildDumplingFilterArgs(opts *apiv1.ExportOptions) ([]string, error) {
	filterPrefix := "--filter="
	defaultFilters := []string{filterPrefix + "*.*", filterPrefix + defaultTableFilter}
	if opts == nil || opts.GetFilter() == nil {
		return defaultFilters, nil
	}

	// sql filter
	if opts.GetFilter().GetSql() != "" {
		return []string{"--sql=" + opts.GetFilter().GetSql()}, nil
	}

	// table filter
	table := opts.GetFilter().GetTable()
	if table == nil {
		return defaultFilters, nil
	}

	args := make([]string, 0, 3)
	if where := table.GetWhere(); where != "" {
		args = append(args, "--where="+where)
	}

	patterns := table.GetPatterns()
	if len(patterns) == 0 {
		args = append(args, filterPrefix+"*.*")
	} else {
		dumplingFilter, err := buildDumplingFilter(patterns)
		if err != nil {
			return nil, fmt.Errorf("invalid table filters: %w", err)
		}
		args = append(args, filterPrefix+dumplingFilter)
	}

	args = append(args, filterPrefix+defaultTableFilter)
	for _, partition := range table.GetPartitions() {
		args = append(args, "--partitions="+partition)
	}
	return args, nil
}

func buildDumplingFilter(patterns []string) (string, error) {
	buf := &bytes.Buffer{}
	csvWriter := csv.NewWriter(buf)
	if err := csvWriter.Write(patterns); err != nil {
		return "", err
	}
	csvWriter.Flush()
	return strings.TrimSuffix(buf.String(), "\n"), nil
}

func toDumplingCompression(compression apiv1.ExportOptions_CompressionType) string {
	switch compression {
	case apiv1.ExportOptions_GZIP:
		return "gzip"
	case apiv1.ExportOptions_SNAPPY:
		return "snappy"
	case apiv1.ExportOptions_ZSTD:
		return "zstd"
	default:
		return ""
	}
}

func buildDumplingCompressionArgs(opts *apiv1.ExportOptions) []string {
	// Dumpling rejects the general --compress option for Parquet. Parquet
	// compression is configured separately through --parquet-compress.
	if opts == nil || opts.GetFileType() == apiv1.ExportOptions_PARQUET {
		return nil
	}
	if compression := toDumplingCompression(opts.GetCompression()); compression != "" {
		return []string{"--compress=" + compression}
	}
	return nil
}

func buildDumplingEscapeBackslashArgs(escapeBackslash *bool) []string {
	if escapeBackslash == nil {
		return nil
	}
	return []string{"--escape-backslash=" + strconv.FormatBool(*escapeBackslash)}
}

func toDumplingParquetCompression(compression apiv1.ExportParquetCompressionType_Enum) string {
	switch compression {
	case apiv1.ExportParquetCompressionType_GZIP:
		return "gzip"
	case apiv1.ExportParquetCompressionType_SNAPPY:
		return "snappy"
	case apiv1.ExportParquetCompressionType_ZSTD:
		return "zstd"
	case apiv1.ExportParquetCompressionType_NONE:
		return "no-compression"
	default:
		return ""
	}
}

// dumplingFileType renders the file type the way Dumpling compares it.
//
// The enum name is upper case ("CSV"), and Dumpling normalizes --filetype to
// lower case - but only in a later adjust step. Its --csv-output-dialect guard
// runs before that, during flag parsing, and compares the raw value against the
// literal "csv". An upper-case file type is therefore harmless on its own and
// fails argument parsing outright as soon as a dialect is also requested:
//
//	csv-output-dialect is only supported when dumping whole table to csv,
//	not compatible with CSV
//
// Lowering it here keeps both paths working whatever Dumpling does internally.
func dumplingFileType(fileType apiv1.ExportOptions_FileType) string {
	return strings.ToLower(fileType.String())
}

func buildDumplingCSVDialectArgs(dialect *apiv1.ExportOptions_ExportCSVDialect_Enum) ([]string, error) {
	if dialect == nil || *dialect == apiv1.ExportOptions_ExportCSVDialect_EXPORT_CSV_DIALECT_UNSPECIFIED {
		return nil, nil
	}

	var value string
	switch *dialect {
	case apiv1.ExportOptions_ExportCSVDialect_DEFAULT:
		value = ""
	case apiv1.ExportOptions_ExportCSVDialect_SNOWFLAKE:
		value = "snowflake"
	case apiv1.ExportOptions_ExportCSVDialect_REDSHIFT:
		value = "redshift"
	case apiv1.ExportOptions_ExportCSVDialect_BIGQUERY:
		value = "bigquery"
	default:
		return nil, fmt.Errorf("unsupported csv dialect: %s", dialect.String())
	}
	return []string{"--csv-output-dialect=" + value}, nil
}

func buildExportOutputURI(storage *apiv1.CloudStorage, cluster *entity.TiDBCluster) (string, error) {
	if storage == nil {
		return "", fmt.Errorf("cloud storage is nil")
	}

	switch storage.GetType() {
	case apiv1.CloudStorageType_S3:
		s3 := storage.GetS3()
		if s3 == nil || s3.GetUri() == "" {
			return "", fmt.Errorf("s3 uri is empty")
		}
		escapedURI, err := url.Parse(escapeBucketURI(s3.GetUri()))
		if err != nil {
			return "", fmt.Errorf("invalid s3 uri: %w", err)
		}
		q := escapedURI.Query()
		if s3.GetRoleArn() != "" {
			q.Set("role-arn", s3.GetRoleArn())
			q.Set("external-id", cluster.GetId())
		}
		if s3.GetAccessKey() != nil && s3.GetAccessKey().GetId() != "" && s3.GetAccessKey().GetSecret() != "" {
			q.Set("access-key", s3.GetAccessKey().GetId())
			q.Set("secret-access-key", s3.GetAccessKey().GetSecret())
		}
		escapedURI.RawQuery = q.Encode()
		return escapedURI.String(), nil

	case apiv1.CloudStorageType_GCS:
		gcs := storage.GetGcs()
		if gcs == nil || gcs.GetUri() == "" {
			return "", fmt.Errorf("gcs uri is empty")
		}
		escapedURI, err := url.Parse(escapeBucketURI(gcs.GetUri()))
		if err != nil {
			return "", fmt.Errorf("invalid gcs uri: %w", err)
		}
		q := escapedURI.Query()
		if gcs.GetServiceAccountKey() != "" {
			q.Set("credentials", gcs.GetServiceAccountKey())
		}
		escapedURI.RawQuery = q.Encode()
		return escapedURI.String(), nil

	case apiv1.CloudStorageType_AZURE_BLOB:
		azureBlob := storage.GetAzureBlob()
		if azureBlob == nil || azureBlob.GetUri() == "" {
			return "", fmt.Errorf("azure blob uri is empty")
		}
		escapedURI, err := url.Parse(escapeBucketURI(azureBlob.GetUri()))
		if err != nil {
			return "", fmt.Errorf("invalid azure blob uri: %w", err)
		}

		accountName, escapeUri, err := ParseAzureURI(escapedURI.String())
		if err != nil {
			return "", fmt.Errorf("parse azure blob uri: %w", err)
		}
		q := escapeUri.Query()
		if accountName != "" {
			q.Set("account-name", accountName)
		}
		if azureBlob.GetSasToken() != "" {
			q.Set("sas-token", azureBlob.GetSasToken())
		}
		escapeUri.RawQuery = q.Encode()
		return escapeUri.String(), nil
	case apiv1.CloudStorageType_OSS:
		oss := storage.GetOss()
		if oss == nil || oss.GetUri() == "" {
			return "", fmt.Errorf("oss uri is empty")
		}
		if oss.GetRoleArn() != "" {
			return "", fmt.Errorf("oss role arn is not supported turns to uri yet")
		}
		escapedURI, err := parseEscapedOSSURI(oss.GetUri())
		if err != nil {
			return "", fmt.Errorf("invalid oss uri: %w", err)
		}

		q := escapedURI.Query()
		if oss.GetAccessKey() != nil && oss.GetAccessKey().GetId() != "" && oss.GetAccessKey().GetSecret() != "" {
			q.Set("access-key", oss.GetAccessKey().GetId())
			q.Set("secret-access-key", oss.GetAccessKey().GetSecret())
		}

		escapedURI.RawQuery = q.Encode()
		return escapedURI.String(), nil
	default:
		return "", fmt.Errorf("unsupported cloud storage type: %s", storage.GetType().String())
	}
}

func escapeBucketURI(uri string) string {
	var schema, bucketURI string
	if idx := strings.Index(uri, "://"); idx >= 0 {
		schema = uri[:idx+3]
		bucketURI = uri[idx+3:]
	}
	pathPart := bucketURI
	suffix := ""
	if idx := strings.IndexAny(bucketURI, "?#"); idx >= 0 {
		pathPart = bucketURI[:idx]
		suffix = bucketURI[idx:]
	}

	pathSegments := strings.Split(pathPart, "/")
	var builder strings.Builder
	builder.WriteString(schema)
	for i, segment := range pathSegments {
		if i > 0 {
			builder.WriteByte('/')
		}
		builder.WriteString(url.PathEscape(segment))
	}
	builder.WriteString(suffix)
	return builder.String()
}

func parseEscapedOSSURI(uri string) (*url.URL, error) {
	escapedURI, err := url.Parse(escapeBucketURI(uri))
	if err != nil {
		return nil, err
	}
	if !strings.EqualFold(escapedURI.Scheme, "oss") {
		return nil, fmt.Errorf("unsupported scheme %q", escapedURI.Scheme)
	}
	return escapedURI, nil
}

func ParseExportDumplingConfig(cfgContent string) (*ExportDumplingConfig, error) {
	cfg := &ExportDumplingConfig{}
	if strings.TrimSpace(cfgContent) == "" {
		return cfg, nil
	}

	scanner := bufio.NewScanner(strings.NewReader(cfgContent))
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}

		key, value, ok := strings.Cut(line, "=")
		if !ok {
			return nil, fmt.Errorf("invalid config line: %q", line)
		}

		key = strings.TrimSpace(key)
		value = strings.Trim(strings.TrimSpace(value), "\"")

		switch key {
		case "node":
			cfg.Node = strings.ToUpper(value)
		case "cpuRequest":
			cfg.CPURequest = value
		case "cpuLimit":
			cfg.CPULimit = value
		case "memoryRequest":
			cfg.MemoryRequest = value
		case "memoryLimit":
			cfg.MemoryLimit = value
		case "tidbGroupNodeCount":
			nodeCount, err := strconv.ParseUint(value, 10, 32)
			if err != nil {
				return nil, fmt.Errorf("invalid tidbGroupNodeCount %q: %w", value, err)
			}
			cfg.TiDBGroupNodeCount = uint32(nodeCount)
		case "tidbGroupNode":
			cfg.TiDBGroupNodeSpec = strings.ToUpper(value)
		case "thread":
			thread, err := strconv.ParseInt(value, 10, 64)
			if err != nil {
				return nil, fmt.Errorf("invalid thread %q: %w", value, err)
			}
			cfg.Thread = int(thread)
		}
	}

	if err := scanner.Err(); err != nil {
		return nil, fmt.Errorf("scan dumpling config: %w", err)
	}
	return cfg, nil
}

func buildExportResourceSpec(cfg *ExportDumplingConfig) (cpuRequest, cpuLimit, memoryRequest, memoryLimit string) {
	cpuRequest = exportJobDefaultCPURequest
	cpuLimit = exportJobDefaultCPULimit
	memoryRequest = exportJobDefaultMemoryRequest
	memoryLimit = exportJobDefaultMemoryLimit

	if cfg == nil {
		return cpuRequest, cpuLimit, memoryRequest, memoryLimit
	}

	if cfg.CPURequest != "" {
		cpuRequest = cfg.CPURequest
	}
	if cfg.CPULimit != "" {
		cpuLimit = cfg.CPULimit
	}
	if cfg.MemoryLimit != "" {
		memoryLimit = cfg.MemoryLimit
	}
	if cfg.MemoryRequest != "" {
		memoryRequest = cfg.MemoryRequest
	}

	return cpuRequest, cpuLimit, memoryRequest, memoryLimit
}

func getNodeVersions(ctx context.Context, dataplaneClient dplanemgrv1.DataPlaneServiceClient, dataplaneName string) (map[string]string, error) {
	nodePools, err := dataplaneClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{
		DataPlaneId: dataplaneName,
	})
	if err != nil {
		return nil, err
	}
	nodeVersionMap := make(map[string]string)
	for _, np := range nodePools.GetNodePools() {
		key := fmt.Sprintf("%dC%dG", np.GetCpuCore(), np.GetMemoryGi())
		if np.GetWithLocalDisk() {
			key = fmt.Sprintf("%s-%s", key, "Local")
		}
		if existingVersion, ok := nodeVersionMap[key]; ok {
			cmp, cmpErr := k8s.CompareK8sVersion(np.GetK8SVersion(), existingVersion)
			if cmpErr == nil && cmp <= 0 {
				continue
			}
		}
		nodeVersionMap[key] = np.GetK8SVersion()
	}
	return nodeVersionMap, nil
}

func getDumplingNameSpace(sharedPoolId string) string {
	return fmt.Sprintf("tidb%s", sharedPoolId)
}

var AzureStorageAccountRe = regexp.MustCompile(`^([a-z0-9]+)\.blob\.core\.windows\.net$`)

const InvalidAzureUriErrMsg = "Invalid Azure Blob URI. Expected format: (azure|https)://<account>.blob.core.windows.net/<container>/<path>"

func ParseAzureURI(uri string) (accountName string, containerURI *url.URL, bizErr bizerr.BizError) {
	if !strings.HasPrefix(uri, "azure://") && !strings.HasPrefix(uri, "https://") {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, InvalidAzureUriErrMsg)
	}
	if !strings.HasSuffix(uri, "/") {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, "uri must end with '/'")
	}
	parsedURL, err := url.Parse(uri)
	if err != nil {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, InvalidAzureUriErrMsg)
	}
	sourceURL, err := url.Parse(fmt.Sprintf("azure:/%s", parsedURL.EscapedPath()))
	if err != nil {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, InvalidAzureUriErrMsg)
	}
	subMatch := AzureStorageAccountRe.FindStringSubmatch(parsedURL.Host)
	if len(subMatch) != 2 {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, InvalidAzureUriErrMsg)
	}
	return subMatch[1], sourceURL, nil
}