package options

import (
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/spf13/pflag"
	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/config"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tikv"
	v1 "github.com/tidbcloud/infra-api/api/v1"
	"k8s.io/client-go/tools/clientcmd"
)

// Options is flags of manager.
type Options struct {
	Kubeconfig string

	// TODO: extract common options
	MetricsBindAddress       string
	WebhookPort              int
	CertDir                  string
	InfraRedisClientCertDir  string
	ManagerServerBindAddress string
	EnableGrpcReflection     bool
	EnableLeaderElection     bool

	ClusterCAPath   string
	ClusterCertPath string
	ClusterKeyPath  string

	ImportEnableTLS bool

	S3BucketName                 string
	S3BackupRoleArn              string
	AWSAccountID                 string
	AWSSDKLogMode                string
	BaseProxy                    string
	Region                       string
	CloudProviderType            string
	KubeClusterName              string
	TidbPoolGCInterval           string
	KeyspacePoolNs               string
	EncryptKeyspacePoolNs        string
	KeyspacePoolEnableTLS        bool
	EncryptKeyspacePoolEnableTLS bool
	TidbVersion                  string
	TidbPoolNamespace            string
	ActivateTidbImage            string
	EtcdPrefix                   string
	EtcdCheckConnPrefix          string
	EtcdEndpoints                string
	ServerlessLightningImage     string
	ImportJobsNamespace          string
	ImportSAName                 string
	ImportRoleName               string
	TiDBWorkerNS                 string
	CentralRoleName              string
	DisableNodeRecover           bool
	DisableTidbReschedule        bool
	DisableNodeRebalance         bool
	DisableNodeDrainer           bool
	DisableRZEvicter             bool
	DisableTidbPodEvicter        bool
	SkipClusterDeleteEtcdAndPd   bool

	KeyspacePoolCreateRatePerSecond float64

	// o11y
	O11yAgentEnabled        bool
	O11yAgentAccessArn      string
	O11yEnv                 string
	O11yOpenAPIAddress      string
	O11yCreateO11yInfraPath string
	O11yDeleteO11yInfraPath string
	O11yGetO11yInfraPath    string
	O11yAuth0GetTokenURL    string
	O11yAuth0ClientID       string
	O11yAuth0ClientSecret   string
	O11yAuth0Audience       string
	O11yVersion             int

	// export
	ExportJobCM        string
	ExportJobNS        string
	DumplingImage      string
	ExportStashingArea string
	ExportDebugMode    bool
	ExportEnableTLS    bool

	AuditLogKmsAliasArn          string
	AuditLogEncryptKeyCiphertext string

	InfraRedisAddr        string
	InfraClusterTable     string
	OtsInstanceName       string
	RocketmqInstanceID    string
	RocketmqEndpoint      string
	RocketmqConsumerGroup string
	RocketmqUsername      string
	RocketmqPassword      string
	ClusterQueueURL       string
	RdsSecretArn          string
	EnableVersionUpdater  bool
	DisableAuditLog       bool
	RdsSecretRegion       string

	EnableInfraMysqlClient bool
	EnableDoubleWrite      bool
	InfraRdsSecretArn      string
	CacheServerAddr        string

	BranchRestoreMode string
}

// New returns default running options
//
//nolint:mnd // all constants are default config
func New() *Options {
	opt := &Options{
		MetricsBindAddress:       ":8080",
		CertDir:                  "/certs",
		InfraRedisClientCertDir:  "/infra-redis-client-cert",
		ManagerServerBindAddress: ":8000",
		WebhookPort:              9443,
		EnableLeaderElection:     false,
		TidbPoolNamespace:        "tidb-pool",
		TidbPoolGCInterval:       "5m",
		EtcdPrefix:               "/tidbcloud/serverless/",
		EtcdEndpoints:            "etcd-svc.tidb-admin:2379",
		BranchRestoreMode:        tikv.RestoreModePitr,
	}

	return opt
}

// AddFlags adds flags for manager.
//
//nolint:funlen // by design
func (opt *Options) AddFlags(fs *pflag.FlagSet) {
	fs.StringVar(&opt.Kubeconfig, "kubeconfig", opt.Kubeconfig,
		"kubeconfig for user cluster",
	)

	fs.StringVar(&opt.MetricsBindAddress, "metrics-bind-address", opt.MetricsBindAddress,
		"The address the metric endpoint binds to.",
	)
	fs.IntVar(&opt.WebhookPort, "webhook-port", opt.WebhookPort,
		"The port of manager webhook.",
	)
	fs.StringVar(&opt.CertDir, "cert-dir", opt.CertDir,
		"The directory that contains the server certificate.",
	)
	fs.StringVar(&opt.InfraRedisClientCertDir, "infra-redis-client-cert-dir", opt.InfraRedisClientCertDir,
		"The directory that contains the infra redis client certificate.",
	)
	fs.StringVar(&opt.ManagerServerBindAddress, "managerserver-bind-address", opt.ManagerServerBindAddress,
		"The address the managerserver endpoint binds to.",
	)
	fs.BoolVar(&opt.EnableGrpcReflection, "enable-grpc-reflection", opt.EnableGrpcReflection,
		"Enable grpc reflection for manager server.",
	)
	fs.BoolVar(&opt.EnableLeaderElection, "enable-leader-election", opt.EnableLeaderElection,
		"Enable leader election for controller manager. "+
			"Enabling this will ensure there is only one active controller manager.",
	)

	fs.StringVar(&opt.ClusterCAPath, "cluster-ca-path", opt.ClusterCAPath, "The path of cluster ca.")
	fs.StringVar(&opt.ClusterCertPath, "cluster-cert-path", opt.ClusterCertPath, "The path of cluster cert.")
	fs.StringVar(&opt.ClusterKeyPath, "cluster-key-path", opt.ClusterKeyPath, "The path of cluster key.")

	fs.BoolVar(&opt.ImportEnableTLS, "import-enable-tls", opt.ImportEnableTLS, "Enable TLS for pods created by manager.")

	fs.StringVar(&opt.S3BucketName, "s3-bucket", opt.S3BucketName, "The S3 bucket name.")
	fs.StringVar(&opt.S3BackupRoleArn, "s3-backup-role-arn", opt.S3BackupRoleArn, "The S3 backup role arn.")
	fs.StringVar(&opt.AWSAccountID, "aws-account-id", opt.AWSAccountID, "The AWS account ID.")
	fs.StringVar(&opt.AWSSDKLogMode, "aws-sdk-log-mode", opt.AWSSDKLogMode, "The AWS SDK log mode.")
	fs.StringVar(&opt.BaseProxy, "base-proxy", opt.BaseProxy, "The address the proxy binds to.")

	fs.StringVar(&opt.Region, "region", opt.Region, "The region where serverless cluster in.")
	fs.StringVar(&opt.CloudProviderType, "cloud-provider-type", opt.CloudProviderType, "The cloud provider type.")
	fs.StringVar(&opt.KubeClusterName, "kube-cluster-name", opt.KubeClusterName,
		"The name of current kubernetes cluster.",
	)
	fs.StringVar(&opt.ServerlessLightningImage, "serverless-lightning-image", opt.ServerlessLightningImage,
		"The image of serverless lightning.")
	fs.StringVar(&opt.ImportJobsNamespace, "import-jobs-ns", opt.ImportJobsNamespace, "The import jobs namespace.")
	fs.StringVar(&opt.ImportSAName, "import-sa-name", opt.ImportSAName, "The import service account name.")
	fs.StringVar(&opt.ImportRoleName, "import-role-name", opt.ImportRoleName, "The import role name.")
	fs.StringVar(&opt.TiDBWorkerNS, "tidb-worker-ns", opt.TiDBWorkerNS, "The TiDB Worker namespace.")
	fs.StringVar(&opt.CentralRoleName, "central-role-name", opt.CentralRoleName, "The central role name.")
	fs.StringVar(&opt.KeyspacePoolNs, "keyspace-pool-ns", opt.KeyspacePoolNs, "The keyspace pool namespace.")
	fs.StringVar(&opt.EncryptKeyspacePoolNs, "encrypt-keyspace-pool-ns", opt.EncryptKeyspacePoolNs,
		"The encrypt keyspace pool namespace.")
	fs.Float64Var(&opt.KeyspacePoolCreateRatePerSecond, "keyspace-pool-create-rate-per-second",
		opt.KeyspacePoolCreateRatePerSecond, "The max create speed for keyspace-pool, in keyspaces per second. 0 means unlimited.")
	fs.BoolVar(&opt.KeyspacePoolEnableTLS, "keyspace-pool-enable-tls",
		opt.KeyspacePoolEnableTLS, "If activate keyspace using TLS")
	fs.BoolVar(&opt.EncryptKeyspacePoolEnableTLS, "encrypt-keyspace-pool-enable-tls",
		opt.EncryptKeyspacePoolEnableTLS, "If activate encrypt keyspace using TLS")
	fs.StringVar(&opt.TidbPoolNamespace, "tidb-pool-ns", opt.TidbPoolNamespace,
		"The tidb pod pool namespace list, separate by comma")
	fs.StringVar(&opt.TidbPoolGCInterval, "tidb-pool-gc-interval", opt.TidbPoolGCInterval,
		"The interval list that run gc in tidb pool. 0 means not to gc, separate by comma")
	fs.StringVar(&opt.TidbVersion, "tidb-version", opt.TidbVersion,
		"The tidb version list should be specified, separate by comma")
	fs.StringVar(&opt.EtcdPrefix, "etcd-prefix", opt.EtcdPrefix,
		"The etcd prefix list, separate by comma")
	fs.StringVar(&opt.EtcdCheckConnPrefix, "etcd-checkconn-prefix", opt.EtcdCheckConnPrefix,
		"The etcd checkconn prefix path")
	fs.StringVar(&opt.EtcdEndpoints, "etcd-endpoints", opt.EtcdEndpoints,
		"The endpoints the etcd connect to, join with ';'.")
	fs.StringVar(&opt.ActivateTidbImage, "activate-tidb-image", opt.ActivateTidbImage,
		"The activate tidb image")
	fs.BoolVar(&opt.DisableNodeRecover, "disable-node-recover", false, "Disable node recover")
	fs.BoolVar(&opt.DisableTidbReschedule, "disable-tidb-reschedule", false, "Disable tidb reschedule")
	fs.BoolVar(&opt.DisableNodeRebalance, "disable-node-rebalance", false, "Disable node rebalance")
	fs.BoolVar(&opt.DisableNodeDrainer, "disable-node-drainer", false, "Disable node drainer")
	fs.BoolVar(&opt.DisableRZEvicter, "disable-zr-evicter", false, "Disable zone redundant evicter")
	fs.BoolVar(&opt.DisableTidbPodEvicter, "disable-tidb-pod-evicter", false, "Disable tidb pod evicter")
	fs.BoolVar(&opt.SkipClusterDeleteEtcdAndPd, "skip-cluster-delete-etcd-and-pd", false,
		"Skip etcd, Keyspace CR and PD requests when deleting clusters.")
	fs.BoolVar(&opt.O11yAgentEnabled, "o11y-enable", false, "Enable o11y agent.")
	fs.StringVar(&opt.O11yOpenAPIAddress, "o11y-openapi-address", opt.O11yOpenAPIAddress,
		"The address the o11y openapi.")
	fs.StringVar(&opt.O11yCreateO11yInfraPath, "o11y-create-path", "/api/v1/clusters",
		"The path the create o11y cluster.")
	fs.StringVar(&opt.O11yDeleteO11yInfraPath, "o11y-delete-path", "/api/v1/clusters",
		"The path the delete o11y cluster.")
	fs.StringVar(&opt.O11yGetO11yInfraPath, "o11y-get-path", "/api/v1/clusters",
		"The path the get o11y cluster.")
	fs.StringVar(&opt.O11yAuth0GetTokenURL, "o11y-auth0-token-url", "https://tidb-soc2.us.auth0.com/oauth/token",
		"The URL the o11y auth0 token api.")
	fs.StringVar(&opt.O11yAuth0ClientID, "o11y-auth0-client-id", opt.O11yAuth0ClientID,
		"The o11y auth0 client ID.")
	fs.StringVar(&opt.O11yAuth0ClientSecret, "o11y-auth0-client-secret", opt.O11yAuth0ClientSecret,
		"The o11y auth0 client secret.")
	fs.StringVar(&opt.O11yAuth0Audience, "o11y-auth0-audience", opt.O11yAuth0Audience,
		"The o11y auth0 audience.")
	fs.StringVar(&opt.O11yAgentAccessArn, "o11y-agent-access-arn", opt.O11yAgentAccessArn,
		"The o11y agent access arn.")
	fs.StringVar(&opt.O11yEnv, "o11y-env", " dev",
		"The o11y cluster env.")
	fs.IntVar(&opt.O11yVersion, "o11y-version", 0,
		"The o11y version.")
	fs.StringVar(&opt.ExportJobCM, "export-job-cm", "export-job-template",
		"The export job configmap name.")
	fs.StringVar(&opt.ExportJobNS, "export-job-ns", "export-job",
		"The export job namespace")
	fs.StringVar(&opt.DumplingImage, "dumpling-image", "",
		"The dumpling image")
	fs.StringVar(&opt.ExportStashingArea, "export-stashing-area", "",
		"The export local s3 uri")
	fs.BoolVar(&opt.ExportDebugMode, "export-debug-mode", false,
		"The export debug mode")
	fs.BoolVar(&opt.ExportEnableTLS, "export-enable-tls", false,
		"If enable the export tls")
	fs.StringVar(&opt.AuditLogKmsAliasArn, "audit-log-kms-alias-arn", "",
		"The kms alias arn to encrypt/decrypt audit log key")
	fs.StringVar(&opt.AuditLogEncryptKeyCiphertext, "audit-log-encrypt-key-ciphertext", "",
		"The ciphertext key to encrypt audit log")

	fs.StringVar(&opt.InfraRedisAddr, "infra-redis-addr", "",
		"The infra redis addr in regional.")
	fs.StringVar(&opt.InfraClusterTable, "infra-cluster-table", "",
		"The infra cluster table in dynamodb.")
	fs.StringVar(&opt.OtsInstanceName, "ots-instance-name", "", "The alicloud tablestore instance name")
	fs.StringVar(&opt.RocketmqInstanceID, "rocketmq-instance-id", "", "The alicloud rocketmq instance id")
	fs.StringVar(&opt.RocketmqEndpoint, "rocketmq-endpoint", "", "The alicloud rocketmq endpoint")
	fs.StringVar(&opt.RocketmqConsumerGroup, "rocketmq-consumer-group", "", "The alicloud rocketmq consumer group")
	fs.StringVar(&opt.RocketmqUsername, "rocketmq-username", "", "The alicloud rocketmq username")
	fs.StringVar(&opt.RocketmqPassword, "rocketmq-password", "", "The alicloud rocketmq password")
	fs.StringVar(&opt.ClusterQueueURL, "cluster-queue-url", "", "The cluster queue url from aws sqs.")
	fs.StringVar(&opt.RdsSecretArn, "rds-secret-arn", "", "The rds secret arn.")
	fs.BoolVar(&opt.EnableVersionUpdater, "enable-version-updater", false, "Enable version updater.")
	fs.BoolVar(&opt.DisableAuditLog, "disable-audit-log", false, "Disable audit log.")
	fs.StringVar(&opt.RdsSecretRegion, "rds-secret-region", "", "region for rds secret.")
	fs.BoolVar(&opt.EnableInfraMysqlClient, "enable-infra-mysql-client", false,
		"Enable mysql client for serverless infra.")
	fs.BoolVar(&opt.EnableDoubleWrite, "enable-double-write", false,
		"Enable double write for serverless infra.")
	fs.StringVar(&opt.InfraRdsSecretArn, "infra-rds-secret-arn", "", "The infra rds secret arn.")
	fs.StringVar(&opt.CacheServerAddr, "cache-server-addr", "", "The cache server address.")
	fs.StringVar(&opt.BranchRestoreMode, "branch-restore-mode", tikv.RestoreModePitr,
		"Branch keyspace restore mode: pitr or online_snapshot (snapshot TS chosen by tikv-worker at task start)")
}

func (opt *Options) Config() (*config.Config, error) {
	if opt.BranchRestoreMode != tikv.RestoreModePitr && opt.BranchRestoreMode != tikv.RestoreModeOnlineSnapshot {
		return nil, fmt.Errorf("invalid --branch-restore-mode %q, must be %q or %q",
			opt.BranchRestoreMode, tikv.RestoreModePitr, tikv.RestoreModeOnlineSnapshot)
	}
	kubeconfig, err := clientcmd.BuildConfigFromFlags("", opt.Kubeconfig)
	if err != nil {
		return nil, fmt.Errorf("can't parse kubeconfig: %w", err)
	}
	importMaxConcurrent, err := importMaxConcurrentFromEnv()
	if err != nil {
		return nil, err
	}
	kubeconfig.QPS = 30
	kubeconfig.Burst = 70

	if opt.AWSAccountID == "" {
		return nil, fmt.Errorf("aws-account-id must set")
	}
	if opt.BaseProxy == "" {
		return nil, fmt.Errorf("base-proxy must set")
	}
	if opt.Region == "" {
		return nil, fmt.Errorf("region must set")
	}
	if opt.CloudProviderType == "" {
		return nil, fmt.Errorf("cloud-provider-type must set")
	}
	if opt.KubeClusterName == "" {
		return nil, fmt.Errorf("kube-cluster-name must set")
	}
	if opt.KeyspacePoolNs == "" {
		return nil, fmt.Errorf("keyspace-pool-ns must set")
	}
	if opt.EncryptKeyspacePoolNs == "" {
		return nil, fmt.Errorf("encrypt-keyspace-pool-ns must set")
	}
	if opt.KeyspacePoolCreateRatePerSecond < 0 {
		return nil, fmt.Errorf("keyspace-pool-create-rate-per-second can't be less than 0")
	}

	tidbPoolNamespaceList, err := opt.validateAndFormatTidbPoolNamespace()
	if err != nil {
		return nil, fmt.Errorf("tidb-pool-ns %w", err)
	}
	tidbPoolGCIntervalList, err := opt.validateAndFormatTidbPoolGCInterval()
	if err != nil {
		return nil, fmt.Errorf("tidb-pool-gc-interval %w", err)
	}
	if len(tidbPoolGCIntervalList) != len(tidbPoolNamespaceList) {
		return nil, fmt.Errorf("tidb-pool-gc-interval element count not equal to tidb-pool-ns")
	}
	tidbVersionList, err := opt.validateAndFormatTidbVersion()
	if err != nil {
		return nil, fmt.Errorf("tidb-verion %w", err)
	}
	if len(tidbVersionList) != len(tidbPoolNamespaceList) {
		return nil, fmt.Errorf("tidb-version element count not equal to tidb-pool-ns")
	}
	etcdPrefixList, err := opt.validateAndFormatEtcdPrefix()
	if err != nil {
		return nil, fmt.Errorf("etcd-prefix %w", err)
	}
	if len(etcdPrefixList) != len(tidbPoolNamespaceList) {
		return nil, fmt.Errorf("etcd-prefix element count not equal to tidb-pool-ns")
	}
	if opt.EtcdCheckConnPrefix == "" {
		return nil, fmt.Errorf("etcd-checkconn-prefix must set")
	}
	if opt.EtcdEndpoints == "" {
		return nil, fmt.Errorf("etcd-endpoints must set")
	}
	if opt.ActivateTidbImage == "" {
		return nil, fmt.Errorf("activate-tidb-image must set")
	}

	if opt.O11yAgentEnabled {
		if opt.O11yEnv == "" {
			return nil, fmt.Errorf("o11y-env must set")
		}
		if opt.O11yAgentAccessArn == "" {
			return nil, fmt.Errorf("o11y-agent-access-arn must set")
		}
		if opt.O11yAuth0Audience == "" {
			return nil, fmt.Errorf("o11y-auth0-audience must set")
		}
		if opt.O11yAuth0GetTokenURL == "" {
			return nil, fmt.Errorf("o11y-auth0-get-token-url must set")
		}
		if opt.O11yOpenAPIAddress == "" {
			return nil, fmt.Errorf("o11y-openapi-address must set")
		}
	}

	if err := opt.validateInfraClientOptions(); err != nil {
		return nil, err
	}
	if opt.RdsSecretArn == "" && !opt.DisableAuditLog {
		return nil, fmt.Errorf("rds-secret-arn must be set when audit log is enabled")
	}

	cfg := &config.Config{
		Kubeconfig: kubeconfig,

		MetricsBindAddress:      opt.MetricsBindAddress,
		WebhookPort:             opt.WebhookPort,
		CertDir:                 opt.CertDir,
		InfraRedisClientCertDir: opt.InfraRedisClientCertDir,
		HttpServerBindAddress:   opt.ManagerServerBindAddress,
		EnableGrpcReflection:    opt.EnableGrpcReflection,
		EnableLeaderElection:    opt.EnableLeaderElection,

		ClusterCAPath:   opt.ClusterCAPath,
		ClusterCertPath: opt.ClusterCertPath,
		ClusterKeyPath:  opt.ClusterKeyPath,

		ImportEnableTLS: opt.ImportEnableTLS,

		S3BucketName:    opt.S3BucketName,
		S3BackupRoleArn: opt.S3BackupRoleArn,
		AWSAccountID:    opt.AWSAccountID,
		AWSSDKLogMode:   opt.AWSSDKLogMode,
		BaseProxy:       opt.BaseProxy,

		Region:                       opt.Region,
		CloudProviderType:            opt.CloudProviderType,
		KubeClusterName:              opt.KubeClusterName,
		ServerlessLightningImage:     opt.ServerlessLightningImage,
		ImportJobsNamespace:          opt.ImportJobsNamespace,
		ImportSAName:                 opt.ImportSAName,
		ImportRoleName:               opt.ImportRoleName,
		ImportMaxConcurrent:          importMaxConcurrent,
		TiDBWorkerNS:                 opt.TiDBWorkerNS,
		KeyspacePoolNs:               opt.KeyspacePoolNs,
		EncryptKeyspacePoolNs:        opt.EncryptKeyspacePoolNs,
		KeyspacePoolEnableTLS:        opt.KeyspacePoolEnableTLS,
		EncryptKeyspacePoolEnableTLS: opt.EncryptKeyspacePoolEnableTLS,
		TidbPoolNamespaceList:        tidbPoolNamespaceList,
		TidbPoolGCIntervalList:       tidbPoolGCIntervalList,
		TidbVersionList:              tidbVersionList,
		EtcdPrefixList:               etcdPrefixList,
		EtcdCheckConnPrefix:          opt.EtcdCheckConnPrefix,
		EtcdEndpoints:                opt.EtcdEndpoints,
		ActivateTidbImage:            opt.ActivateTidbImage,
		DisableNodeRecover:           opt.DisableNodeRecover,
		DisableTidbReschedule:        opt.DisableTidbReschedule,
		DisableNodeRebalance:         opt.DisableNodeRebalance,
		DisableNodeDrainer:           opt.DisableNodeDrainer,
		DisableRZEvicter:             opt.DisableRZEvicter,
		DisableTidbPodEvicter:        opt.DisableTidbPodEvicter,
		SkipClusterDeleteEtcdAndPd:   opt.SkipClusterDeleteEtcdAndPd,

		KeyspacePoolCreateRatePerSecond: opt.KeyspacePoolCreateRatePerSecond,

		O11yAgentEnabled:   opt.O11yAgentEnabled,
		O11yAgentAccessArn: opt.O11yAgentAccessArn,

		ObservabilityInfraConfig: &v1.ObservabilityInfraConfig{
			OpenAPIAddress:      opt.O11yOpenAPIAddress,
			CreateO11yInfraPath: opt.O11yCreateO11yInfraPath,
			DeleteO11yInfraPath: opt.O11yDeleteO11yInfraPath,
			GetO11yInfraPath:    opt.O11yGetO11yInfraPath,
			Auth0GetTokenURL:    opt.O11yAuth0GetTokenURL,
			Auth0ClientID:       opt.O11yAuth0ClientID,
			Auth0ClientSecret:   opt.O11yAuth0ClientSecret,
			Auth0Audience:       opt.O11yAuth0Audience,
			Version:             opt.O11yVersion,
			Env:                 opt.O11yEnv,
		},

		ExportJobCM:        opt.ExportJobCM,
		ExportJobNS:        opt.ExportJobNS,
		DumplingImage:      opt.DumplingImage,
		ExportStashingArea: opt.ExportStashingArea,
		ExportDebugMode:    opt.ExportDebugMode,
		ExportEnableTLS:    opt.ExportEnableTLS,

		AuditLogKmsAliasArn:          opt.AuditLogKmsAliasArn,
		AuditLogEncryptKeyCiphertext: opt.AuditLogEncryptKeyCiphertext,

		InfraRedisAddr:        opt.InfraRedisAddr,
		InfraClusterTable:     opt.InfraClusterTable,
		OtsInstanceName:       opt.OtsInstanceName,
		RocketmqInstanceID:    opt.RocketmqInstanceID,
		RocketmqEndpoint:      opt.RocketmqEndpoint,
		RocketmqConsumerGroup: opt.RocketmqConsumerGroup,
		RocketmqUsername:      opt.RocketmqUsername,
		RocketmqPassword:      opt.RocketmqPassword,
		ClusterQueueURL:       opt.ClusterQueueURL,
		RdsSecretArn:          opt.RdsSecretArn,
		EnableVersionUpdater:  opt.EnableVersionUpdater,
		DisableAuditLog:       opt.DisableAuditLog,
		RdsSecretRegion:       opt.RdsSecretRegion,

		EnableInfraMysqlClient: opt.EnableInfraMysqlClient,
		EnableDoubleWrite:      opt.EnableDoubleWrite,
		InfraRdsSecretArn:      opt.InfraRdsSecretArn,
		CacheServerAddr:        opt.CacheServerAddr,

		BranchRestoreMode: opt.BranchRestoreMode,
	}

	return cfg, nil
}

func (opt *Options) validateInfraClientOptions() error {
	requirements := config.ResolveInfraClientRequirements(opt.EnableInfraMysqlClient, opt.EnableDoubleWrite)
	if requirements.LegacyCache && opt.InfraRedisAddr == "" {
		return fmt.Errorf("infra-redis-addr must set")
	}
	if requirements.LegacyClient && opt.InfraClusterTable == "" {
		return fmt.Errorf("infra-cluster-table must set")
	}
	if requirements.MySQLClient && opt.InfraRdsSecretArn == "" {
		return fmt.Errorf("infra-rds-secret-arn must be set when enable-infra-mysql-client or enable-double-write is true")
	}
	return nil
}

const (
	defaultImportMaxConcurrent = 10
	importMaxConcurrentEnv     = "IMPORT_MAX_CONCURRENT"
)

func importMaxConcurrentFromEnv() (int, error) {
	value := os.Getenv(importMaxConcurrentEnv)
	if value == "" {
		return defaultImportMaxConcurrent, nil
	}

	concurrent, err := strconv.Atoi(value)
	if err != nil {
		return 0, fmt.Errorf("%s must be an integer: %w", importMaxConcurrentEnv, err)
	}
	if concurrent <= 0 {
		return 0, fmt.Errorf("%s must be greater than 0", importMaxConcurrentEnv)
	}

	return concurrent, nil
}

func (opt *Options) validateAndFormatTidbPoolNamespace() ([]string, error) {
	tidbPoolNamespaceList := strings.Split(opt.TidbPoolNamespace, ",")
	for i := range tidbPoolNamespaceList {
		tidbPoolNamespace := strings.Trim(tidbPoolNamespaceList[i], "")
		if tidbPoolNamespace == "" {
			return nil, fmt.Errorf("have empty value")
		}
		tidbPoolNamespaceList[i] = tidbPoolNamespace
		if i+1 >= len(tidbPoolNamespaceList) {
			break
		}
		for _, v := range tidbPoolNamespaceList[i+1:] {
			if strings.Trim(v, " ") == tidbPoolNamespace {
				return nil, fmt.Errorf("have duplicate name `%s`", v)
			}
		}
	}

	return tidbPoolNamespaceList, nil
}

func (opt *Options) validateAndFormatTidbPoolGCInterval() ([]time.Duration, error) {
	var tidbPoolGCIntervalList []time.Duration
	for _, v := range strings.Split(opt.TidbPoolGCInterval, ",") {
		if v = strings.Trim(v, " "); v == "" {
			return nil, fmt.Errorf("have empty value")
		}
		tidbPoolGCInterval, err := time.ParseDuration(v)
		if err != nil {
			return nil, fmt.Errorf("`%s` ParseDuration err: %v", v, err)
		}
		if tidbPoolGCInterval < 0 {
			return nil, fmt.Errorf("`%s` can't be less than 0", v)
		}
		tidbPoolGCIntervalList = append(tidbPoolGCIntervalList, tidbPoolGCInterval)
	}

	return tidbPoolGCIntervalList, nil
}

func (opt *Options) validateAndFormatTidbVersion() ([]string, error) {
	tidbVersionList := strings.Split(opt.TidbVersion, ",")
	for i, v := range tidbVersionList {
		if v = strings.Trim(v, " "); v == "" {
			return nil, fmt.Errorf("have empty value")
		}
		tidbVersionList[i] = v
	}

	return tidbVersionList, nil
}

func (opt *Options) validateAndFormatEtcdPrefix() ([]string, error) {
	etcdPrefixList := strings.Split(opt.EtcdPrefix, ",")
	for i := range etcdPrefixList {
		etcdPrefix := strings.Trim(etcdPrefixList[i], "")
		if etcdPrefix == "" {
			return nil, fmt.Errorf("have empty value")
		}
		if strings.HasSuffix(etcdPrefix, "/") {
			etcdPrefixList[i] = etcdPrefix
		} else {
			etcdPrefixList[i] = etcdPrefix + "/"
		}
		if i+1 >= len(etcdPrefixList) {
			break
		}
		for _, v := range etcdPrefixList[i+1:] {
			if strings.Trim(v, " ") == etcdPrefix {
				return nil, fmt.Errorf("have duplicate name `%s`", v)
			}
		}
	}

	return etcdPrefixList, nil
}