package app

import (
	"context"
	"crypto/tls"
	"encoding/base64"
	"flag"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"

	alicred "github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/kms"
	"github.com/aws/smithy-go/logging"
	"github.com/go-logr/logr"
	"github.com/pingcap/tidb/pkg/util"
	"github.com/pkg/errors"
	"github.com/spf13/cobra"
	"github.com/spf13/pflag"
	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/config"
	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/options"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/repo"
	auditlogutil "github.com/tidbcloud/aws-shared-provider/pkg/auditlog/util"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/assigncachelimit"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/branch"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/cluster"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/export"
	cimports "github.com/tidbcloud/aws-shared-provider/pkg/controllers/imports"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/keyspace"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/keyspacepoolconfig"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/noderebalance"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/noderecover"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/secretsyncer"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/tidbnodescaler"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/tidbpod"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/tidbpodevicter"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/tidbpodrescheduler"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/tidbservertemplate"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/zoneredundantevicter"
	"github.com/tidbcloud/aws-shared-provider/pkg/dmscaler"
	pkginfra "github.com/tidbcloud/aws-shared-provider/pkg/infra"
	"github.com/tidbcloud/aws-shared-provider/pkg/keyspacemigration"
	"github.com/tidbcloud/aws-shared-provider/pkg/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/managerserver"
	"github.com/tidbcloud/aws-shared-provider/pkg/nodedrainer"
	"github.com/tidbcloud/aws-shared-provider/pkg/nodescorer"
	"github.com/tidbcloud/aws-shared-provider/pkg/notifier"
	"github.com/tidbcloud/aws-shared-provider/pkg/scheme"
	"github.com/tidbcloud/aws-shared-provider/pkg/storage"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbautoscaling"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbmanager"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbpdb"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbreclaimer"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbscaler"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbupgrader"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/kubeapi"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidb"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidbworker"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tikv"
	"github.com/tidbcloud/aws-shared-provider/pkg/version"
	"github.com/tidbcloud/aws-shared-provider/pkg/versionupdater"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infrao11y "github.com/tidbcloud/observability/agent-control-server/pkg/o11y/api"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	infracache "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/cache"
	infradynamo "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/dynamo"
	_ "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/metrics/k8s"
	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
	infraots "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/tablestore"
	_ "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/metrics/k8s"
	consumerredis "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/redis"
	unmarshaldynamo "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/dynamo"
	unmarshalots "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/tablestore"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"google.golang.org/grpc"
	"google.golang.org/grpc/backoff"
	"google.golang.org/grpc/keepalive"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	zapopt "sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/webhook"
)

var (
	NoSyncPeriod = time.Duration(0)
)

const (
	serverlessPDEndpoint  = "serverless-cluster-pd.tidb-serverless.svc:2379"
	tikvApiEndpoint       = "tikv-api-service.tidb-serverless.svc.cluster.local:19500"
	tidbWorkerApiEndpoint = "scaler-svc.tidb-worker:9080"
	expectedArnLen        = 6
	infraClusterCacheSize = 10000
)

// NewCommand returns app command.
func NewCommand() *cobra.Command {
	setupLog := ctrl.Log.WithName("setup")

	opts := options.New()
	lvl := zap.NewAtomicLevelAt(zap.FatalLevel)
	zapOpts := zapopt.Options{
		Development:     true,
		StacktraceLevel: &lvl,
		TimeEncoder:     zapcore.ISO8601TimeEncoder,
	}

	cmd := &cobra.Command{
		Use:  "manager",
		Long: "manager watches CRD and implements infra API",
		Run: func(cmd *cobra.Command, args []string) {
			fmt.Printf("%v\n", version.Get())

			ctrl.SetLogger(zapopt.New(zapopt.UseFlagOptions(&zapOpts)))
			printFlags(setupLog, cmd.Flags())

			cfg, err := opts.Config()
			if err != nil {
				setupLog.Error(err, "can't parse options to config")
				os.Exit(-1)
			}

			setupLog.Info("begin to start manager")
			if err := Run(cfg, setupLog); err != nil {
				setupLog.Error(err, "run manager failed")
				os.Exit(-1)
			}
		},
	}
	opts.AddFlags(cmd.Flags())
	zapOpts.BindFlags(flag.CommandLine)
	cmd.FParseErrWhitelist.UnknownFlags = true

	cmd.AddCommand(NewVersionCmd())

	return cmd
}

// NewVersionCmd return cmd reports version.
func NewVersionCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:  "version",
		Long: "manager version",
		Run: func(cmd *cobra.Command, args []string) {
			fmt.Printf("%v\n", version.Get())
		},
	}

	return cmd
}

// Run runs the operator.
func Run(cfg *config.Config, logger logr.Logger) error {
	mgr, err := newManager(cfg)
	if err != nil {
		return err
	}

	infraRequirements := config.ResolveInfraClientRequirements(cfg.EnableInfraMysqlClient, cfg.EnableDoubleWrite)
	var infraRedisTLSConfig *tls.Config
	if infraRequirements.LegacyCache {
		infraRedisTLSConfig, err = infraRedisClientTLSConfig(cfg.InfraRedisClientCertDir)
		if err != nil {
			return fmt.Errorf("load infra redis client tls config err: %w", err)
		}
	}

	clusterTLSConfig, err := util.NewTLSConfig(
		util.WithCAPath(cfg.ClusterCAPath),
		util.WithCertAndKeyPath(
			cfg.ClusterCertPath, cfg.ClusterKeyPath,
		),
	)
	if err != nil {
		return fmt.Errorf("load cluster tls config err: %w", err)
	}

	pdClient := pd.NewClient(serverlessPDEndpoint, clusterTLSConfig)
	tikvApi := tikv.NewApi(tikvApiEndpoint, clusterTLSConfig)
	tidbWorkerApi := tidbworker.NewApi(tidbWorkerApiEndpoint, clusterTLSConfig)

	// only for regional infra
	var infraClusterCli infraclient.Client
	var infraDoubleWriteCli infraclient.Client
	var redisConsumer *consumerredis.Consumer
	var pAwsCfg *aws.Config
	var alicloudCred alicred.Credential

	if infraRequirements.MySQLClient {
		rdsConfig, err := auditlogutil.GetRdsConfig(cfg.CloudProviderType, cfg.InfraRdsSecretArn, cfg.Region)
		if err != nil {
			return fmt.Errorf("get rds config err: %w", err)
		}
		underlayClient, err := inframysql.NewMySQLClient(
			rdsConfig.ToGormutilConfig(),
			"manager",
			cfg.KubeClusterName,
			inframysql.DefaultIndexField2name,
			infra.DefaultClusterEventTTL,
			true,
		)
		if err != nil {
			return fmt.Errorf("create infra mysql client err: %w", err)
		}
		underlayClient.SetLogger(ctrl.Log)

		if cfg.EnableInfraMysqlClient {
			index2key := map[string]string{
				"status.keyspace_token-index":         "keyspace_token",
				"status.zone_redundant_enabled-index": "zone_redundant_enabled",
			}
			infraDoubleWriteCli, _ = infracache.NewCacheClientV2(
				cfg.CacheServerAddr, index2key, []string{"status.keyspace_token-index"},
				infraClusterCacheSize, underlayClient, false,
			)
			infraDoubleWriteCli.SetLogger(ctrl.Log)
		} else {
			infraDoubleWriteCli = underlayClient
		}
	}

	switch cfg.CloudProviderType {
	case types.ProviderTypeAliCloudFreeTier:
		if !infraRequirements.LegacyClient {
			break
		}
		if cfg.OtsInstanceName != "" {
			if alicloudCred, err = alicred.NewCredential(nil); err != nil {
				return fmt.Errorf("load alicloud credential err: %w", err)
			}
			otsEndpoint := fmt.Sprintf("https://%s.%s.ots.aliyuncs.com", cfg.OtsInstanceName, cfg.Region)
			underlayClient := infraots.NewTableStoreClient(
				alicloudCred, otsEndpoint, cfg.OtsInstanceName, cfg.InfraClusterTable,
				map[string]string{
					"status.keyspace_token-index":         "status.keyspace_token",
					"status.zone_redundant_enabled-index": "status.zone_redundant_enabled",
				},
			)
			underlayClient.SetLogger(ctrl.Log)
			if !infraRequirements.LegacyCache {
				infraClusterCli = underlayClient
				break
			}

			var infraClusterCo infraclient.CacheOperate
			infraClusterCli, infraClusterCo = infracache.NewCacheClient(
				infraClusterCacheSize, underlayClient, false,
			)
			infraClusterCli.SetLogger(ctrl.Log)

			redisConsumer = &consumerredis.Consumer{
				Addr:         cfg.InfraRedisAddr,
				TLSConfig:    infraRedisTLSConfig,
				Stream:       cfg.InfraClusterTable + "_" + cfg.KubeClusterName,
				CacheOperate: infraClusterCo,
				Unmarshal: func(data any) (infra.Object, error) {
					return unmarshalots.Unmarshal(data, func() infra.Object { return &infra.Cluster{} })
				},
			}
			redisConsumer.SetLogger(ctrl.Log)
			if err = redisConsumer.Init(context.Background()); err != nil {
				return err
			}
			if err = mgr.Add(manager.RunAnyWhere(redisConsumer)); err != nil {
				return err
			}
			break
		}
		logger.Info("--ots-instance-name is empty in alicloud, use aws for its regional infra!")
		fallthrough
	case types.ProviderTypeAWSFreeTier:
		if !infraRequirements.LegacyClient {
			break
		}
		awsSDKLogger, awsSDKLogMode := logging.Logger(logging.Nop{}), aws.ClientLogMode(0)
		if cfg.AWSSDKLogMode != "" {
			awsSDKLogger = logging.NewStandardLogger(os.Stderr)
			for _, mode := range strings.Split(cfg.AWSSDKLogMode, ",") {
				switch strings.TrimSpace(mode) {
				case "signing":
					awsSDKLogMode |= aws.LogSigning
				case "retries":
					awsSDKLogMode |= aws.LogRetries
				case "request":
					awsSDKLogMode |= aws.LogRequest
				case "requestWithBody":
					awsSDKLogMode |= aws.LogRequestWithBody
				case "response":
					awsSDKLogMode |= aws.LogResponse
				case "responseWithBody":
					awsSDKLogMode |= aws.LogResponseWithBody
				default:
					return fmt.Errorf("unknown aws sdk log mode: %s", mode)
				}
			}
		}
		awsCfg, err := awsconfig.LoadDefaultConfig(context.TODO(),
			awsconfig.WithLogger(awsSDKLogger),
			awsconfig.WithClientLogMode(awsSDKLogMode),
		)
		if err != nil {
			return fmt.Errorf("load aws configuration err: %w", err)
		}

		pAwsCfg = &awsCfg
		underlayClient := infradynamo.NewDynamoClient(
			pAwsCfg, cfg.InfraClusterTable, map[string]string{
				"status.keyspace_token-index":         "status.keyspace_token",
				"status.zone_redundant_enabled-index": "status.zone_redundant_enabled",
			},
		)
		underlayClient.SetLogger(ctrl.Log)
		if !infraRequirements.LegacyCache {
			infraClusterCli = underlayClient
			break
		}

		var infraClusterCo infraclient.CacheOperate
		infraClusterCli, infraClusterCo = infracache.NewCacheClient(
			infraClusterCacheSize, underlayClient, false,
		)
		infraClusterCli.SetLogger(ctrl.Log)

		redisConsumer = &consumerredis.Consumer{
			Addr:         cfg.InfraRedisAddr,
			TLSConfig:    infraRedisTLSConfig,
			Stream:       cfg.InfraClusterTable + "_" + cfg.KubeClusterName,
			CacheOperate: infraClusterCo,
			Unmarshal: func(data any) (infra.Object, error) {
				return unmarshaldynamo.Unmarshal(data, func() infra.Object { return &infra.Cluster{} })
			},
		}
		redisConsumer.SetLogger(ctrl.Log)
		if err = redisConsumer.Init(context.Background()); err != nil {
			return err
		}
		if err = mgr.Add(manager.RunAnyWhere(redisConsumer)); err != nil {
			return err
		}
	default:
		return fmt.Errorf("unsupported cloud provider type: %s", cfg.CloudProviderType)
	}

	if cfg.EnableInfraMysqlClient {
		logger.Info("enable infra mysql client")
		infraClusterCli, infraDoubleWriteCli = infraDoubleWriteCli, infraClusterCli
	}

	// set auditlog scheduler repo
	var auditLogScheduleRepo repo.AuditLogScheduleRepo
	if !cfg.DisableAuditLog {
		rdsSecretRegion := cfg.Region
		if cfg.RdsSecretRegion != "" {
			rdsSecretRegion = cfg.RdsSecretRegion
		}
		rdsConfig, err := auditlogutil.GetRdsConfig(cfg.CloudProviderType, cfg.RdsSecretArn, rdsSecretRegion)
		if err != nil {
			return fmt.Errorf("get rds config err: %w", err)
		}
		db, err := storage.NewGormDB(rdsConfig)
		if err != nil {
			return fmt.Errorf("new gorm db err: %w", err)
		}
		dbProvider, _ := storage.NewDBProvier(db)
		auditLogScheduleRepo = repo.NewAuditLogScheduleRepo(dbProvider)
	}

	tidbController := tidb.NewController(clusterTLSConfig, "")

	//nolint:mnd // ignore
	etcdClient, err := clientv3.New(clientv3.Config{
		Endpoints:   strings.Split(cfg.EtcdEndpoints, ";"),
		DialTimeout: 5 * time.Second,
		DialOptions: []grpc.DialOption{
			grpc.WithBlock(),
			grpc.WithConnectParams(grpc.ConnectParams{
				Backoff: backoff.Config{
					BaseDelay:  100 * time.Millisecond, // Default was 1s.
					Multiplier: 1.6,                    // Default
					Jitter:     0.2,                    // Default
					MaxDelay:   3 * time.Second,        // Default was 120s.
				},
				MinConnectTimeout: 2 * time.Second,
			}),
			grpc.WithKeepaliveParams(keepalive.ClientParameters{
				Time:    10 * time.Second,
				Timeout: 2 * time.Second,
			}),
		},
	})
	if err != nil {
		return fmt.Errorf("new etcd clientv3: %w", err)
	}
	defer etcdClient.Close() //nolint:errcheck // no need check error

	tNotifier := notifier.New()
	tManagers := map[string]tidbmanager.TidbManager{}
	etcdTokenPrefixMap := map[string]string{}
	tidbVersionMap := map[string]string{}
	migrationLocker := keyspacemigration.New()
	safeToEvict := types.CASafeToEvict
	if cfg.CloudProviderType == types.ProviderTypeAliCloudFreeTier {
		safeToEvict = types.GoatSafeToEvict
	}

	infraConfig := &pkginfra.InfraConfig{
		Rest:              cfg.Kubeconfig,
		CloudProviderType: cfg.CloudProviderType,
		KubeClusterName:   cfg.KubeClusterName,
	}
	infraApi, err := pkginfra.NewInfraAPI(context.Background(), infraConfig)
	if err != nil {
		return err
	}
	for i, tidbPoolNamespace := range cfg.TidbPoolNamespaceList {
		tManager, err := tidbmanager.New(
			mgr,
			safeToEvict,
			tidbPoolNamespace,
			etcdClient,
			cfg.EtcdPrefixList[i],
			tNotifier,
			tidbController,
			infraClusterCli,
			infraApi,
			cfg,
			migrationLocker,
		)
		if err != nil {
			return err
		}
		if err = mgr.Add(tManager); err != nil {
			return err
		}
		tManagers[tidbPoolNamespace] = tManager

		tidbScaler := tidbscaler.New(mgr.GetClient(), tManager)
		if err = mgr.Add(tidbScaler); err != nil {
			return err
		}

		// tidbmanager can set Scaler interface here
		tManager.SetTidbScaler(tidbScaler)

		tidbUpgrader := tidbupgrader.New(mgr.GetClient(), tManager, tidbController)
		if err = mgr.Add(tidbUpgrader); err != nil {
			return err
		}

		tidbAutoscaling := tidbautoscaling.New(tidbPoolNamespace, mgr.GetClient(), mgr.GetClient(),
			infraClusterCli, tManager)
		if err = mgr.Add(tidbAutoscaling); err != nil {
			return err
		}

		tidbPDB := tidbpdb.New(tidbPoolNamespace, mgr.GetClient(), tManager)
		if err = mgr.Add(tidbPDB); err != nil {
			return err
		}

		if cfg.TidbPoolGCIntervalList[i] > 0 {
			tidbReclaimer := tidbreclaimer.New(
				tidbPoolNamespace,
				mgr.GetClient(),
				tManager,
				tidbController,
				cfg.TidbPoolGCIntervalList[i],
			)
			if err = mgr.Add(tidbReclaimer); err != nil {
				return err
			}
		}

		if err = (&tidbservertemplate.Reconciler{
			Client:      mgr.GetClient(),
			TidbManager: tManager,
			Namespace:   tidbPoolNamespace,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create TidbServerTemplate controller(%s): %w", tidbPoolNamespace, err)
		}

		if err = (&assigncachelimit.Reconciler{
			Client:      mgr.GetClient(),
			TidbManager: tManager,
			Namespace:   tidbPoolNamespace,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create AssignCacheLimit controller(%s): %w", tidbPoolNamespace, err)
		}

		if err = (&tidbpod.SyncCacheReconciler{
			Client:      mgr.GetClient(),
			TidbManager: tManager,
			Namespace:   tidbPoolNamespace,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create Tidb-Pod-SyncCache controller(%s): %w", tidbPoolNamespace, err)
		}

		etcdTokenPrefixMap[tidbPoolNamespace] = cfg.EtcdPrefixList[i] + "token/"
		tidbVersionMap[tidbPoolNamespace] = cfg.TidbVersionList[i]
	}

	dmScaler := dmscaler.New(mgr.GetClient(), clusterTLSConfig)
	if err = mgr.Add(dmScaler); err != nil {
		return err
	}

	// Seed the shared in-memory runtime rate from startup config/flags.
	createRateStore := keyspacepoolconfig.NewCreateRateStore(cfg.KeyspacePoolCreateRatePerSecond)

	server, err := managerserver.New(
		cfg,
		mgr,
		tManagers,
		tNotifier,
		etcdClient,
		tidbController,
		infraClusterCli,
		etcdTokenPrefixMap,
		infraApi,
		pdClient,
		tidbWorkerApi,
		migrationLocker,
		createRateStore,
	)
	if err != nil {
		return err
	}
	if err = mgr.Add(server); err != nil {
		return err
	}

	nodeScorer := nodescorer.New(mgr.GetClient(), 1*time.Minute)
	if err = mgr.Add(nodeScorer); err != nil {
		return err
	}

	if !cfg.DisableNodeDrainer {
		nodeDrainer := nodedrainer.New(mgr, 1*time.Minute)
		if err = mgr.Add(nodeDrainer); err != nil {
			return err
		}
	}

	if cfg.EnableVersionUpdater {
		versionUpdater := versionupdater.NewTiDBVersionUpdater(infraClusterCli, tidbVersionMap, mgr.Elected())
		versionUpdater.Start(context.Background())
	}

	// Setup o11y agent controllers
	var o11yApi infrao11y.Api
	if cfg.O11yAgentEnabled {
		// Init o11y api
		o11yApi, err = infrao11y.InitApi(
			cfg.ObservabilityInfraConfig.Auth0GetTokenURL,
			cfg.ObservabilityInfraConfig.Auth0ClientID,
			cfg.ObservabilityInfraConfig.Auth0ClientSecret,
			cfg.ObservabilityInfraConfig.Auth0Audience,
			cfg.ObservabilityInfraConfig.OpenAPIAddress,
			cfg.ObservabilityInfraConfig.CreateO11yInfraPath,
			cfg.ObservabilityInfraConfig.DeleteO11yInfraPath,
			cfg.ObservabilityInfraConfig.GetO11yInfraPath,
			"/api/v1/clusters", "/api/v1/clusters",
			cfg.ObservabilityInfraConfig.Env,
		)
		if err != nil {
			return fmt.Errorf("unable to setup o11y client error: %v", err)
		}

		O11yGlobalClusterID := os.Getenv(types.O11yGlobalClusterID)
		if O11yGlobalClusterID == "" {
			return errors.Errorf("env %s is empty", types.O11yGlobalClusterID)
		}
		cfg.O11yGlobalClusterID = O11yGlobalClusterID
	}

	if err = (&cluster.Reconciler{
		InfraClusterCli:        infraClusterCli,
		InfraDoubleWriteCli:    infraDoubleWriteCli,
		EtcdKV:                 etcdClient.KV,
		EtcdLease:              etcdClient.Lease,
		AwsCfg:                 pAwsCfg,
		AlicloudCred:           alicloudCred,
		S3BucketName:           cfg.S3BucketName,
		S3BackupRoleArn:        cfg.S3BackupRoleArn,
		BaseProxy:              cfg.BaseProxy,
		KubeClusterName:        cfg.KubeClusterName,
		ImportJobsNamespace:    cfg.ImportJobsNamespace,
		KeyspacePoolNs:         cfg.KeyspacePoolNs,
		EncryptKeyspacePoolNs:  cfg.EncryptKeyspacePoolNs,
		EtcdTokenPrefixMap:     etcdTokenPrefixMap,
		TidbVersionMap:         tidbVersionMap,
		EnableDoubleWrite:      cfg.EnableDoubleWrite,
		EnableInfraMysqlClient: cfg.EnableInfraMysqlClient,
		TidbPools:              cfg.TidbPoolNamespaceList,
		ProviderType:           cfg.CloudProviderType,
		O11yAgentEnabled:       cfg.O11yAgentEnabled,
		O11yConfig:             cfg.ObservabilityInfraConfig,
		O11yApi:                o11yApi,
		O11yAgentAccessArn:     cfg.O11yAgentAccessArn,
		O11yGlobalClusterID:    cfg.O11yGlobalClusterID,
		O11yEnv:                cfg.ObservabilityInfraConfig.Env,
		PdClient:               pdClient,
		TidbWorkerApi:          tidbWorkerApi,
		TidbManagers:           tManagers,
		AuditLogScheduleRepo:   auditLogScheduleRepo,
		TidbController:         tidbController,

		SkipClusterDeleteEtcdAndPd: cfg.SkipClusterDeleteEtcdAndPd,
	}).SetupWithManager(mgr, cfg, redisConsumer); err != nil {
		return fmt.Errorf("unable to create Cluster controller: %w", err)
	}

	if err = (&branch.Reconciler{
		InfraClusterCli:          infraClusterCli,
		EtcdKV:                   etcdClient.KV,
		EtcdLease:                etcdClient.Lease,
		S3BucketName:             cfg.S3BucketName,
		BaseProxy:                cfg.BaseProxy,
		KubeClusterName:          cfg.KubeClusterName,
		KeyspacePoolNs:           cfg.KeyspacePoolNs,
		EncryptKeyspacePoolNs:    cfg.EncryptKeyspacePoolNs,
		EtcdTokenPrefixMap:       etcdTokenPrefixMap,
		TidbVersionMap:           tidbVersionMap,
		TidbPools:                cfg.TidbPoolNamespaceList,
		ProviderType:             cfg.CloudProviderType,
		ServerlessLightningImage: cfg.ServerlessLightningImage,
		PdClient:                 pdClient,
		TikvApi:                  tikvApi,
		TidbWorkerApi:            tidbWorkerApi,
		O11yAgentEnabled:         cfg.O11yAgentEnabled,
		O11yConfig:               cfg.ObservabilityInfraConfig,
		O11yApi:                  o11yApi,
		O11yAgentAccessArn:       cfg.O11yAgentAccessArn,
		O11yGlobalClusterID:      cfg.O11yGlobalClusterID,
		O11yEnv:                  cfg.ObservabilityInfraConfig.Env,
		TidbManagers:             tManagers,
		TidbController:           tidbController,

		BranchRestoreMode:          cfg.BranchRestoreMode,
		SkipClusterDeleteEtcdAndPd: cfg.SkipClusterDeleteEtcdAndPd,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create Branch controller: %w", err)
	}

	if err = (&export.Reconciler{
		Logger:             ctrl.Log.WithName("controllers").WithName("Export").WithName("regional"),
		InfraClusterCli:    infraClusterCli,
		Client:             mgr.GetClient(),
		WorkloadCache:      mgr.GetCache(),
		EtcdKV:             etcdClient.KV,
		KubeClusterName:    cfg.KubeClusterName,
		EtcdTokenPrefixMap: etcdTokenPrefixMap,
		TidbVersionMap:     tidbVersionMap,
		TidbPools:          cfg.TidbPoolNamespaceList,
		ProviderType:       cfg.CloudProviderType,
		PdClient:           pdClient,
		ExportJobNS:        cfg.ExportJobNS,
		ExportJobCM:        cfg.ExportJobCM,
		DumplingImage:      cfg.DumplingImage,
		ExportStashingArea: cfg.ExportStashingArea,
		DebugMode:          cfg.ExportDebugMode,
		AWSAccountID:       cfg.AWSAccountID,
		Codecs:             serializer.NewCodecFactory(scheme.Scheme),
		EnableTLS:          cfg.ExportEnableTLS,
		MigrationLocker:    migrationLocker,
	}).SetupWithManager(mgr, "export-regional"); err != nil {
		return fmt.Errorf("unable to create export controller: %w", err)
	}

	if err = (&keyspace.Reconciler{
		Client:            mgr.GetClient(),
		InfraClusterCli:   infraClusterCli,
		InfraClient:       mgr.GetClient(),
		PdClient:          pdClient,
		ActivateTidbImage: cfg.ActivateTidbImage,
		Namespace:         cfg.KeyspacePoolNs,
		EnableTLS:         cfg.KeyspacePoolEnableTLS,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create %s keyspace controller: %w", cfg.KeyspacePoolNs, err)
	}
	if err = (&keyspacepoolconfig.Reconciler{
		Client:          mgr.GetClient(),
		Namespace:       cfg.KeyspacePoolNs,
		KubeClusterName: cfg.KubeClusterName,
		CreateRateStore: createRateStore,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create %s keyspace pool config controller: %w", cfg.KeyspacePoolNs, err)
	}

	if err = (&keyspace.Reconciler{
		Client:            mgr.GetClient(),
		InfraClusterCli:   infraClusterCli,
		InfraClient:       mgr.GetClient(),
		PdClient:          pdClient,
		ActivateTidbImage: cfg.ActivateTidbImage,
		Namespace:         cfg.EncryptKeyspacePoolNs,
		EnableTLS:         cfg.EncryptKeyspacePoolEnableTLS,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create %s keyspace controller: %w", cfg.EncryptKeyspacePoolNs, err)
	}
	if err = (&keyspacepoolconfig.Reconciler{
		Client:          mgr.GetClient(),
		Namespace:       cfg.EncryptKeyspacePoolNs,
		KubeClusterName: cfg.KubeClusterName,
		Encrypt:         true,
		CreateRateStore: createRateStore,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create %s keyspace pool config controller: %w", cfg.EncryptKeyspacePoolNs, err)
	}

	if err = addKeyspaceIndexFields(mgr, func(object client.Object) bool {
		return object.GetNamespace() == cfg.KeyspacePoolNs || object.GetNamespace() == cfg.EncryptKeyspacePoolNs
	}); err != nil {
		return err
	}

	if err = (&secretsyncer.Reconciler{
		Client:     mgr.GetClient(),
		TidbPools:  cfg.TidbPoolNamespaceList,
		Namespace:  os.Getenv("NAMESPACE"),
		SecretName: secretsyncer.TiDBSessCertName,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create secret syncer controller: %w", err)
	}

	if err = (&tidbpod.CleanReconciler{
		Client:    mgr.GetClient(),
		TidbPools: cfg.TidbPoolNamespaceList,
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create Tidb-Pod-Clean controller: %w", err)
	}

	if !cfg.DisableRZEvicter {
		if err = (&zoneredundantevicter.Reconciler{
			ProviderType:    cfg.CloudProviderType,
			Client:          mgr.GetClient(),
			InfraClusterCli: infraClusterCli,
			InfraClient:     mgr.GetClient(),
			Namespace:       os.Getenv("NAMESPACE"),
			CheckInterval:   80 * time.Second,
			TidbManagers:    tManagers,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create zoneredundantevicter controller: %w", err)
		}
	}

	if err = (&tidbnodescaler.Reconciler{
		ProviderType: cfg.CloudProviderType,
		Client:       mgr.GetClient(),
		Namespace:    os.Getenv("NAMESPACE"),
	}).SetupWithManager(mgr); err != nil {
		return fmt.Errorf("unable to create tidbnodescaler controller: %w", err)
	}

	if !cfg.DisableTidbPodEvicter {
		if err = (&tidbpodevicter.Reconciler{
			Client:       mgr.GetClient(),
			Namespace:    os.Getenv("NAMESPACE"),
			TidbManagers: tManagers,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create tidbpodevicter controller: %w", err)
		}
	}

	if !cfg.DisableTidbReschedule {
		if err = (&tidbpodrescheduler.Reconciler{
			Client:       mgr.GetClient(),
			Namespace:    os.Getenv("NAMESPACE"),
			TidbManagers: tManagers,
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create tidbpodrescheduler controller: %w", err)
		}
	}

	if !cfg.DisableNodeRebalance {
		if err = (&noderebalance.Reconciler{
			Client: mgr.GetClient(),
		}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create noderebalance controller: %w", err)
		}
	}

	if !cfg.DisableNodeRecover {
		if err = (&noderecover.Reconciler{Client: mgr.GetClient()}).SetupWithManager(mgr); err != nil {
			return fmt.Errorf("unable to create noderecover controller: %w", err)
		}
	}

	if err = cimports.NewImportReconciler(
		mgr, ctrl.Log.WithName("controllers").WithName("Import").WithName("regional"),
		cimports.NewServerlessTierEnvBuilder(infraClusterCli, mgr.GetClient(), cfg, tikvApi, pdClient),
		migrationLocker,
		cfg.ImportMaxConcurrent,
	).SetupWithManager(mgr, "import-regional"); err != nil {
		return fmt.Errorf("unable to create Import controller: %w", err)
	}
	if err = addImportIndexFields(mgr.GetFieldIndexer()); err != nil {
		return err
	}

	if err = addPodIndexFields(mgr); err != nil {
		return err
	}

	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		return fmt.Errorf("problem running manager: %w", err)
	}

	return nil
}

func infraRedisClientTLSConfig(certDir string) (*tls.Config, error) {
	if strings.TrimSpace(certDir) == "" {
		return nil, nil
	}
	return util.NewTLSConfig(
		util.WithCAPath(filepath.Join(certDir, "ca.crt")),
		util.WithCertAndKeyPath(
			filepath.Join(certDir, "tls.crt"),
			filepath.Join(certDir, "tls.key"),
		),
	)
}

func newManager(cfg *config.Config) (manager.Manager, error) {
	mgr, err := manager.New(cfg.Kubeconfig, manager.Options{
		Manager: ctrl.Options{
			Scheme:  scheme.Scheme,
			Metrics: metricsserver.Options{BindAddress: cfg.MetricsBindAddress},
			WebhookServer: webhook.NewServer(webhook.Options{
				Port:    cfg.WebhookPort,
				CertDir: cfg.CertDir,
			}),
			LeaderElection:                cfg.EnableLeaderElection,
			LeaderElectionID:              "7dd07112.pingcap.com",
			LeaderElectionReleaseOnCancel: true,
			Cache: cache.Options{
				SyncPeriod: &NoSyncPeriod,
				ByObject: map[client.Object]cache.ByObject{
					&corev1.Pod{}: {UnsafeDisableDeepCopy: ptr.To(true)},
				},
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create manager: %w", err)
	}

	return mgr, nil
}

func addKeyspaceIndexFields(mgr manager.Manager, filter func(object client.Object) bool) error {
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &serverlessv1alpha1.Keyspace{},
		"status.phase", func(obj client.Object) []string {
			// filter out BindPhase, it's too more
			if filter(obj) && obj.(*serverlessv1alpha1.Keyspace).Status.Phase != serverlessv1alpha1.BindPhase {
				return []string{string(obj.(*serverlessv1alpha1.Keyspace).Status.Phase)}
			}
			return nil
		}); err != nil {
		return fmt.Errorf("unable to add index for Keyspace status.phase: %w", err)
	}
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &serverlessv1alpha1.Keyspace{},
		"status.inPool", func(obj client.Object) []string {
			if filter(obj) && obj.(*serverlessv1alpha1.Keyspace).Status.InPool {
				return []string{"true"}
			}
			return nil
		}); err != nil {
		return fmt.Errorf("unable to add index for Keyspace status.inPool: %w", err)
	}
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &serverlessv1alpha1.Keyspace{},
		"spec.keyspaceName", func(obj client.Object) []string {
			if filter(obj) {
				if keyspaceName := obj.(*serverlessv1alpha1.Keyspace).Spec.KeyspaceName; keyspaceName != "" {
					return []string{keyspaceName}
				}
			}
			return nil
		}); err != nil {
		return fmt.Errorf("unable to add index for Keyspace spec.keyspaceName: %w", err)
	}
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &serverlessv1alpha1.Keyspace{},
		"status.bindCluster", func(obj client.Object) []string {
			if filter(obj) {
				if bindCluster := obj.(*serverlessv1alpha1.Keyspace).Status.BindCluster; bindCluster != "" {
					return []string{bindCluster}
				}
			}
			return nil
		}); err != nil {
		return fmt.Errorf("unable to add index for Keyspace status.bindCluster: %w", err)
	}
	return nil
}

func addImportIndexFields(indexer client.FieldIndexer) error {
	if err := indexer.IndexField(context.TODO(), &cloudv1.Import{},
		"spec.clusterName", func(obj client.Object) []string {
			importObj := obj.(*cloudv1.Import)
			return []string{importObj.Spec.ClusterName}
		}); err != nil {
		return fmt.Errorf("unable to add index for Import spec.ClusterName: %w", err)
	}
	if err := indexer.IndexField(context.TODO(), &cloudv1.Import{},
		"status.taskId", func(obj client.Object) []string {
			importObj := obj.(*cloudv1.Import)
			return []string{fmt.Sprintf("%d", importObj.Status.TaskId)}
		}); err != nil {
		return fmt.Errorf("unable to add index for Import spec.TaskId: %w", err)
	}
	return nil
}

func addPodIndexFields(mgr manager.Manager) error {
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &corev1.Pod{},
		"spec.nodeName", func(obj client.Object) []string {
			pod := obj.(*corev1.Pod)
			return []string{pod.Spec.NodeName}
		}); err != nil {
		return fmt.Errorf("unable to add index for pod spec.nodeName: %w", err)
	}
	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &corev1.Pod{},
		"status.unschedulable", func(obj client.Object) []string {
			pod := obj.(*corev1.Pod)
			if kubeapi.IsPodUnschedulable(pod) {
				return []string{"unschedulable"}
			}
			return nil
		}); err != nil {
		return fmt.Errorf("unable to add index for pod status unschedulable: %w", err)
	}
	return nil
}

func printFlags(logger logr.Logger, fs *pflag.FlagSet) {
	fs.VisitAll(func(f *pflag.Flag) {
		logger.Info("FLAG", f.Name, f.Value)
	})
}

// nolint: unused // not need
func decryptAuditLogKey(cfg *aws.Config, kmsAliasArn, ciphertext string) (string, error) {
	ctx := context.TODO()
	if kmsAliasArn == "" || ciphertext == "" {
		return "", fmt.Errorf("got empty config: arn %s, cipher %s", kmsAliasArn, ciphertext)
	}

	if len(strings.Split(kmsAliasArn, ":")) != expectedArnLen {
		return "", fmt.Errorf("bad arn: %s", kmsAliasArn)
	}

	cipher, err := base64.StdEncoding.DecodeString(ciphertext)
	if err != nil {
		return "", fmt.Errorf("ciphertext is not base64: %w", err)
	}

	svc := kms.NewFromConfig(*cfg)
	result, err := svc.Decrypt(ctx, &kms.DecryptInput{
		KeyId:          aws.String(kmsAliasArn),
		CiphertextBlob: cipher,
	})
	if err != nil {
		return "", fmt.Errorf("failed to decrypt ciphertext: %w", err)
	}

	return string(result.Plaintext), err
}