package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/gin-contrib/pprof"
	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpcctxtags "github.com/grpc-ecosystem/go-grpc-middleware/tags"
	"github.com/spf13/cobra"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	execv1pb "github.com/tidbcloud/cluster-service-ng/api/executor/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/adapter"
	"github.com/tidbcloud/cluster-service-ng/domain/app"
	"github.com/tidbcloud/cluster-service-ng/domain/app/cron"
	"github.com/tidbcloud/cluster-service-ng/domain/app/proxy"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/server"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr"
	"github.com/tidbcloud/cluster-service-ng/pkg/executorclient"
	grpcutils "github.com/tidbcloud/cluster-service-ng/pkg/grpc"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	stgdplanemgr "github.com/tidbcloud/cluster-service-ng/pkg/storage/dataplanemanager"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql"
	autoscale_policy_bundle "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/autoscale_policy_bundle"
	autoscale_runtime_guard "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/autoscale_runtime_guard"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/backup"
	cassandraproxystaterds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/cassandra_proxy_state"
	dmclusterrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/dmcluster"
	exportrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/export"
	lockrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/lock"
	dbmigrate "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/migrate/db"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/migrate/deploy"
	nodegrouprds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/node_instance_group"
	pauseresumeoperationrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/pause_resume_operation"
	reservednodemeteringcheckpoint "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/reserved_node_metering_checkpoint"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/restore"
	ruconfigrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/ruconfig"
	scaleoperationrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/scale_operation"
	sharedpoolrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/sharedpool"
	templateregistryrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/template_registry"
	tidbclusterds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/tidbcluster"
	tidbgrouprds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/tidbgroup"
	workernodemeteringcheckpoint "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/worker_node_metering_checkpoint"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	o11yv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/tiinf"
	"github.com/tidbcloud/tiinf/transport/http/tigin"
	tigrpcclient "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	tigrpcserver "github.com/tidbcloud/tiinf/transport/tigrpc/server"
	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
	"gorm.io/gorm"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/clientcmd"
	kubeclient "sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	defaultConfigPath = "/etc/cluster-service-ng/config.yaml"
)

type serverMeteringWriter interface {
	Write(ctx context.Context, data any) error
}

var (
	externalKubeConfigPath string
	configPath             string
	mockMode               = false
	cmd                    = &cobra.Command{}

	serveCmd = &cobra.Command{
		Use:   "server",
		Short: "serve a sharedPool ng server",
		Run: func(cmd *cobra.Command, args []string) {
			cobra.CheckErr(serve())
		},
	}
)

func init() {
	cmd.PersistentFlags().StringVar(&configPath, "config", defaultConfigPath, "config file")
	cmd.PersistentFlags().StringVar(&externalKubeConfigPath, "kubeconfig", "", "external kube config file path")
	cmd.PersistentFlags().BoolVar(&mockMode, "mock", false, "enable mock mode")
	cmd.AddCommand(serveCmd)
}

func main() {
	if err := cmd.Execute(); err != nil {
		tilog.PlainError("serve return err", zap.Error(err))
		os.Exit(1)
	}
}

func serve() error {
	tilog.PlainInfo("start to server")

	// Init config.
	cfg, err := config.InitConfigFromFile(configPath)
	if err != nil {
		return err
	}
	// initialize with config file
	inf, err := setupTiInf(cfg)
	if err != nil {
		return fmt.Errorf("init tiinf failed. %w", err)
	}
	if mockMode {
		reflection.Register(inf.GrpcServer())
	}

	// init dependencies
	// init dataplane manager client

	dataplaneMgrClient, err := dataplanemgr.NewClient(cfg.ServerConfig.DataPlaneManagerAddr)
	if err != nil {
		return fmt.Errorf("failed to create dataplane manager client: %w", err)
	}

	// init k8s(etcd) client
	restCfg, err := clientcmd.BuildConfigFromFlags("", externalKubeConfigPath)
	if err != nil {
		return fmt.Errorf("build k8s config failed. %w", err)
	}
	kubeScheme := runtime.NewScheme()
	if err := o11yv2.AddToScheme(kubeScheme); err != nil {
		return fmt.Errorf("add o11y v2 scheme failed. %w", err)
	}
	kubeClient, err := kubeclient.New(restCfg, kubeclient.Options{Scheme: kubeScheme})
	if err != nil {
		return fmt.Errorf("new kube client failed. %w", err)
	}

	dataPlaneRepo, err := stgdplanemgr.NewDataplaneRepo(
		cfg.ServerConfig.Region,
		cfg.ServerConfig.CloudProvider,
		dataplaneMgrClient,
	)
	if err != nil {
		return fmt.Errorf("new dplanemgr dataplane repo failed. %w", err)
	}
	var (
		dbIns            *gorm.DB
		dbProvider       db.Provider
		dbTransactioner  db.Transactioner
		dataplaneManager dataplane.Manager
	)
	// if RDS is enabled, we use mysql storage
	dbIns = mysql.NewDB(&cfg.ServerConfig.MysqlConfig)

	// init db
	if err := dbmigrate.RunDBMigrate(dbIns); err != nil {
		return fmt.Errorf("run db migrate failed. %w", err)
	}
	dbProvider, _, dbTransactioner = mysql.NewDBProvider(dbIns)
	dataplaneManager = dataplane.NewManager(restCfg, dataPlaneRepo) // init dataplane client manager

	var clusterConfigClient clusterconfig.Client
	if cfg.ServerConfig.DevConfig != nil && cfg.ServerConfig.DevConfig.MockConfigCenter {
		tilog.PlainWarn("using mock config center client (dev mode)")
		clusterConfigClient = clusterconfig.NewMockClient()
	} else {
		var err error
		clusterConfigClient, err = clusterconfig.NewClient(cfg.ServerConfig.ConfigCenterAddr)
		if err != nil {
			return fmt.Errorf("failed to init cluster config client: %w", err)
		}
	}

	var (
		sharedPoolRepo            repo.SharedPoolRepo
		cassandraProxyStateRepo   repo.CassandraProxyStateRepo
		tidbGroupRepo             repo.TiDBGroupRepo
		tidbClusterRepo           repo.TiDBClusterRepo
		dmClusterRepo             repo.DMClusterRepo
		exportRepo                repo.ExportRepo
		lockRepo                  repo.LockRepo
		ruConfigRepo              repo.RuConfigRepo
		scaleOperationRepo        repo.ScaleOperationRepo
		pauseResumeOperationRepo  repo.PauseResumeOperationRepo
		backupGcSettingRepo       repo.BackupGcSettingRepo
		lwRestoreRepo             repo.LightweightRestoreRepo
		classicRestoreRepo        repo.ClassicRestoreRepo
		manualBackupRepo          repo.ManualBackupRepo
		scaleTikvDecisionRepo     repo.ScaleTikvDecisionRepo
		sharedPoolObjectStoreRepo repo.SharedPoolObjectStoreRepo
	)

	lockRepo = lockrds.NewLockRepo(dbProvider)

	templateMap, err := templateversion.LoadTemplatesFromPath(cfg.ServerConfig.TemplatePath)
	if err != nil {
		return fmt.Errorf("load templates from path failed: %w", err)
	}

	templateRegistryRepo := templateregistryrds.NewTemplateRegistryRepo(dbProvider)
	templateVersionSvc := templateversion.NewService(templateRegistryRepo, cfg.ServerConfig.BizConfig.TemplateVersionMaxLag)
	unlockTemplateBootstrap, err := lockRepo.Lock(context.Background(), &entity.Lock{
		Key:         entity.LockKeyTemplateBootstrap,
		TimeoutSecs: 600,
	})
	if err != nil {
		return fmt.Errorf("failed to lock template bootstrap: %w", err)
	}
	if err := dbTransactioner.Transaction(context.Background(), func(ctx context.Context) error {
		return templateVersionSvc.Bootstrap(ctx, templateMap)
	}); err != nil {
		_ = unlockTemplateBootstrap()
		return fmt.Errorf("template bootstrap failed: %w", err)
	}
	if unlockErr := unlockTemplateBootstrap(); unlockErr != nil {
		tilog.PlainWarn("failed to unlock template bootstrap lock", zap.Error(unlockErr))
	}

	backupGcSettingRepo = backup.NewBackupGcSettingRepo(dbProvider)
	cassandraProxyStateRepo = cassandraproxystaterds.NewCassandraProxyStateRepo(dbProvider)
	sharedPoolRepo = sharedpoolrds.NewSharedPoolRepo(dbProvider, dataPlaneRepo, backupGcSettingRepo, dataplaneManager, dataplaneMgrClient.GetBucketServiceClient())
	tidbClusterRepo = tidbclusterds.NewClusterRepo(dbProvider, clusterConfigClient, dataplaneManager)
	tidbGroupRepo = tidbgrouprds.NewTiDBGroupRepo(dbProvider, dataplaneManager, dataplaneMgrClient)
	dmClusterRepo = dmclusterrds.NewDMClusterRepo(dbProvider, dataplaneManager)
	exportRepo = exportrds.NewExportRepo(dbProvider)
	lockRepo = lockrds.NewLockRepo(dbProvider)
	lwRestoreRepo = restore.NewLwRestoreRepo(dbProvider)
	classicRestoreRepo = restore.NewClassicRestoreRepo(dbProvider)
	manualBackupRepo = backup.NewManualBackupRepo(dbProvider)
	manualBackupRestoreRepo := backup.NewManualBackupRestoreRepo(dbProvider)
	scaleTikvDecisionRepo = restore.NewScaleTikvDecisionRepo(dbProvider)
	sharedPoolObjectStoreRepo = sharedpoolrds.NewSharedPoolObjectStoreRepo(dbProvider)
	scaleOperationRepo = scaleoperationrds.NewScaleOperationRepo(dbProvider)
	pauseResumeOperationRepo = pauseresumeoperationrds.New(dbProvider)
	nodeGroupStateRepo := nodegrouprds.NewDataplaneNodeInstanceGroupStateRepo(dbProvider)
	nodeGroupLockRepo := nodegrouprds.NewDataplaneNodeInstanceGroupLockRepo(dbProvider)
	nodeGroupRepo := nodegrouprds.NewNodeInstanceGroupRepo(dbProvider)
	dataplaneNodeGroupRepo := nodegrouprds.NewDataplaneNodeInstanceGroupRepo(dbProvider)
	ruConfigRepo, err = ruconfigrds.NewRuConfigRepo(dbProvider, cfg.ServerConfig.RUConfig)
	if err != nil {
		return fmt.Errorf("new ru config repo failed. %w", err)
	}
	autoscalePolicyBundleRepo := autoscale_policy_bundle.NewAutoscalePolicyBundleRepo(dbProvider)
	autoscaleRuntimeGuardRepo := autoscale_runtime_guard.NewAutoscaleRuntimeGuardRepo(dbProvider)

	clustergatewayCli := clustergateway.NewClusterGatewayImpl()

	meteringMetaWriter, meteringDataWriter, err := buildServerMeteringWriters(cfg.ServerConfig.ControlPlaneMetering)
	if err != nil {
		return err
	}

	// init deploy migration
	mg := deploy.NewMigrator(
		sharedPoolRepo,
		tidbGroupRepo,
		tidbClusterRepo,
		dataplaneMgrClient,
		dataplaneManager,
		clusterConfigClient,
		cfg.ServerConfig,
		templateVersionSvc,
		kubeClient,
	)
	err = mg.Migrate(context.Background(), dbIns)
	if err != nil {
		return fmt.Errorf("run deploy migration failed. %w", err)
	}

	// init dag executor
	var dagExecutor executor.RemoteExecutor
	conn, err := tigrpcclient.NewClientConn(cfg.ServerConfig.ExecutorGrpcAddr,
		tigrpcclient.WithDefault(),
		tigrpcclient.WithUnaryInterceptors(
			grpcutils.ClientPayloadLogInterceptor,
			grpcutils.CallingTimeoutInterceptor(5*time.Second),
		),
	)
	if err != nil {
		return fmt.Errorf("new grpc conn failed. %w", err)
	}
	execClient, err := executorclient.New(
		dag.ExecutorScope,
		execv1pb.NewExecutorServiceClient(conn),
	)
	if err != nil {
		return fmt.Errorf("new executor client failed. %w", err)
	}
	dagExecutor = execClient

	ngServer, err := server.NewServer(
		cfg.ServerConfig.CloudProvider,
		cfg.ServerConfig.Region,
		dbTransactioner,
		sharedPoolRepo,
		pauseResumeOperationRepo,
		dataPlaneRepo,
		cassandraProxyStateRepo,
		nodeGroupRepo,
		nodeGroupStateRepo,
		dataplaneNodeGroupRepo,
		tidbClusterRepo,
		tidbGroupRepo,
		dmClusterRepo,
		exportRepo,
		lockRepo,
		ruConfigRepo,
		autoscalePolicyBundleRepo,
		autoscaleRuntimeGuardRepo,
		scaleOperationRepo,
		lwRestoreRepo,
		classicRestoreRepo,
		manualBackupRepo,
		manualBackupRestoreRepo,
		scaleTikvDecisionRepo,
		backupGcSettingRepo,
		templateVersionSvc,
		dataplaneManager,
		dataplaneMgrClient,
		clustergatewayCli,
		clusterConfigClient,
		dagExecutor,
		&cfg.ServerConfig.ImageConfig,
		&cfg.ServerConfig.BizConfig,
		cfg.ServerConfig.RUConfig,
		&cfg.ServerConfig.ScaleOperation,
		cfg.ServerConfig.DataPlaneMetering, // this metering config is set into kernel component's config, so use the dataplane metering config
		meteringMetaWriter,
		meteringDataWriter,
	)
	if err != nil {
		return fmt.Errorf("new cluster ng service server failed. %w", err)
	}

	// run cron jobs
	if err := runCronJobs(
		cfg,
		lockRepo,
		ngServer,
		templateVersionSvc,
		backupGcSettingRepo,
		sharedPoolObjectStoreRepo,
		sharedPoolRepo,
		dmClusterRepo,
		clustergatewayCli,
		dataplaneMgrClient.GetBucketServiceClient(),
		dataplaneManager,
		dataplaneMgrClient,
		nodeGroupStateRepo,
		nodeGroupLockRepo,
		nodeGroupRepo,
		dataplaneNodeGroupRepo,
		manualBackupRepo,
		tidbClusterRepo,
		dbProvider,
	); err != nil {
		return fmt.Errorf("run cron jobs failed: %w", err)
	}

	// register grpc server.
	app.RegisterGrpc(cfg, inf, ngServer)

	// Register http gateway.
	err = app.RegisterGateway(cfg, inf)
	if err != nil {
		return fmt.Errorf("register http gateway failed. %w", err)
	}

	// register proxy server.
	sharedPoolAdapter := adapter.NewSharedPoolAdapter(tidbClusterRepo, sharedPoolRepo, dataplaneMgrClient)
	proxy.SetUpReverseProxy(inf.GinEngine(), sharedPoolAdapter)

	if err := inf.Run(); err != nil {
		return fmt.Errorf("run tiinf failed. %w", err)
	}

	return nil
}

func buildServerMeteringWriters(cfg *config.MeteringConfig) (serverMeteringWriter, serverMeteringWriter, error) {
	var metadataWriter serverMeteringWriter
	if cfg.MetadataEnabled() {
		writer, err := cfg.BuildMeteringMetaWriter()
		if err != nil {
			return nil, nil, fmt.Errorf("build metering metadata writer failed. %w", err)
		}
		metadataWriter = writer
	}

	var dataWriter serverMeteringWriter
	if cfg != nil && cfg.Enable {
		writer, err := cfg.BuildMeteringDataWriter()
		if err != nil {
			return nil, nil, fmt.Errorf("build metering data writer failed. %w", err)
		}
		dataWriter = writer
	}

	return metadataWriter, dataWriter, nil
}

// setupTiInf sets up the tiinf instance.
// Tiinf instance has both gin and grpc server.
func setupTiInf(_ *config.Config) (tiinf.TiInf, error) {
	inf := tiinf.NewInf()

	interceptors := []grpc.UnaryServerInterceptor{
		grpcctxtags.UnaryServerInterceptor(),
		grpcutils.ServerPayloadLogInterceptor,
	}

	if err := inf.Init(
		tiinf.WithConfigFile(configPath),
		// We do not use default tiinf setup here.
		tiinf.WithGrpcOptions(
			tigrpcserver.WithPrometheus(),
			tigrpcserver.WithTracing(),
			tigrpcserver.WithRecovery(grpcrecovery.WithRecoveryHandlerContext(tigrpcserver.DefaultPanicHandler)),
			tigrpcserver.WithUnaryInterceptors(interceptors...),
		),
	); err != nil {
		return nil, err
	}

	// Do not use default tiinf gin setup.
	inf.GinEngine().Handlers = make([]gin.HandlerFunc, 0)
	serviceName, err := tiinf.ServiceNameFromEnv()
	if err != nil {
		return nil, err
	}
	inf.GinEngine().Use(gin.Recovery(), tigin.RecordMetrics, otelgin.Middleware(serviceName), tigin.SetResponseTraceID)

	pprof.Register(inf.GinEngine())

	return inf, nil
}

func runCronJobs(
	cfg *config.Config,
	lockRepo repo.LockRepo,
	ngService server.Service,
	templateVersionSvc *templateversion.Service,
	backupGcSettingRepo repo.BackupGcSettingRepo,
	sharedPoolObjectStoreRepo repo.SharedPoolObjectStoreRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	dmClusterRepo repo.DMClusterRepo,
	clusterGateway clustergateway.Client,
	bucketSvc dplanemgrv1.BucketServiceClient,
	dataplaneManager dataplane.Manager,
	dataplaneMgrClient dataplanemgr.Client,
	nodeGroupStateRepo repo.DataplaneNodeInstanceGroupStateRepo,
	nodeGroupLockRepo repo.DataplaneNodeInstanceGroupLockRepo,
	nodeGroupRepo repo.NodeInstanceGroupRepo,
	dataplaneNodeGroupRepo repo.DataplaneNodeInstanceGroupRepo,
	manualBackupRepo repo.ManualBackupRepo,
	tidbClusterRepo repo.TiDBClusterRepo,
	dbProvider db.Provider,
) error {
	cronMgr := cron.NewCronJobManager()

	// create a clean shared pool job
	if cfg.ServerConfig.CronJob.EnableCleanSharedPool {
		if cfg.ServerConfig.CronJob.CleanSharedPoolCron == "" {
			cfg.ServerConfig.CronJob.CleanSharedPoolCron = "0 0 */6 * * *" // default to run every 6 hours
		}

		tilog.PlainInfo("clean shared pool job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.CleanSharedPoolCron),
		)
		cleanSharedPoolJob := cron.NewCleanSharedPoolJob(
			ngService,
			lockRepo,
			cfg.ServerConfig.CronJob.CleanSharedPoolCron,
		)
		if err := cronMgr.AddJob(cleanSharedPoolJob); err != nil {
			return fmt.Errorf("failed to add clean shared pool job: %w", err)
		}
	}

	// create a clean data bucket job
	if cfg.ServerConfig.CronJob.CleanDataBucketCron == "" {
		cfg.ServerConfig.CronJob.CleanDataBucketCron = "0 0 */6 * * *" // default to run every 6 hours
	}
	tilog.PlainInfo("clean data bucket job is enabled",
		zap.String("cron", cfg.ServerConfig.CronJob.CleanDataBucketCron),
	)
	cleanDataBucketJob := cron.NewCleanDataBucketJob(
		cfg.ServerConfig.CronJob.CleanDataBucketCron,
		backupGcSettingRepo,
		manualBackupRepo,
		sharedPoolObjectStoreRepo,
		sharedPoolRepo,
		bucketSvc)
	if err := cronMgr.AddJob(cleanDataBucketJob); err != nil {
		return fmt.Errorf("failed to add clean data bucket job: %w", err)
	}

	if cfg.ServerConfig.CronJob.CleanInternalBucketCron == "" {
		cfg.ServerConfig.CronJob.CleanInternalBucketCron = "0 0 */6 * * *" // default to run every 6 hours
	}
	tilog.PlainInfo("clean internal bucket job is enabled",
		zap.String("cron", cfg.ServerConfig.CronJob.CleanInternalBucketCron),
	)
	cleanInternalBucketJob := cron.NewCleanInternalBucketJob(
		cfg.ServerConfig.CronJob.CleanInternalBucketCron,
		sharedPoolRepo,
		sharedPoolObjectStoreRepo,
		bucketSvc)
	if err := cronMgr.AddJob(cleanInternalBucketJob); err != nil {
		return fmt.Errorf("failed to add clean internal bucket job: %w", err)
	}

	// create a clean expired auto backup data job
	if cfg.ServerConfig.CronJob.CleanExpiredAutoBackupDataCron == "" {
		cfg.ServerConfig.CronJob.CleanExpiredAutoBackupDataCron = "0 0 */6 * * *" // default to run every 6 hours
	}
	tilog.PlainInfo("clean expired auto backup data job is enabled",
		zap.String("cron", cfg.ServerConfig.CronJob.CleanExpiredAutoBackupDataCron),
	)
	cleanExpiredAutoBackupDataJob := cron.NewCleanExpiredAutoBackupDataJob(
		cfg.ServerConfig.CronJob.CleanExpiredAutoBackupDataCron,
		backupGcSettingRepo,
		sharedPoolRepo,
		sharedPoolObjectStoreRepo,
		bucketSvc,
	)
	if err := cronMgr.AddJob(cleanExpiredAutoBackupDataJob); err != nil {
		return fmt.Errorf("failed to add clean expired auto backup data job: %w", err)
	}

	if cfg.ServerConfig.CronJob.CleanManualBackupDeletionRulesCron == "" {
		cfg.ServerConfig.CronJob.CleanManualBackupDeletionRulesCron = "0 0 */6 * * *"
	}
	tilog.PlainInfo("clean manual backup deletion rules job is enabled",
		zap.String("cron", cfg.ServerConfig.CronJob.CleanManualBackupDeletionRulesCron),
	)
	cleanManualBackupDeletionRulesJob := cron.NewCleanManualBackupDeletionRulesJob(
		cfg.ServerConfig.CronJob.CleanManualBackupDeletionRulesCron,
		manualBackupRepo,
		sharedPoolRepo,
		sharedPoolObjectStoreRepo,
		bucketSvc,
	)
	if err := cronMgr.AddJob(cleanManualBackupDeletionRulesJob); err != nil {
		return fmt.Errorf("failed to add clean manual backup deletion rules job: %w", err)
	}

	// create metering disk info job
	if cfg.ServerConfig.ControlPlaneMetering.Enable {
		controlPlaneMeteringDataWriter, err := cfg.ServerConfig.ControlPlaneMetering.BuildMeteringDataWriter()
		if err != nil {
			return fmt.Errorf("failed to build metering data writer: %w", err)
		}

		meteringDiskInfoJob := cron.NewMeteringDiskInfoJob(
			ngService,
			lockRepo,
			controlPlaneMeteringDataWriter,
			"@every 1m",
		)
		if err := cronMgr.AddJob(meteringDiskInfoJob); err != nil {
			return fmt.Errorf("failed to add metering disk info job: %w", err)
		}

		manualBackupMeteringJob := cron.NewManualBackupMeteringJob(
			manualBackupRepo,
			lockRepo,
			controlPlaneMeteringDataWriter,
			"@every 1h",
		)
		if err := cronMgr.AddJob(manualBackupMeteringJob); err != nil {
			return fmt.Errorf("failed to add manual backup metering job: %w", err)
		}

	}

	if cfg.ServerConfig.ControlPlaneMetering.WorkerNodeEnabled() {
		workerNodeMeteringDataWriter, err := cfg.ServerConfig.ControlPlaneMetering.BuildMeteringDataWriter()
		if err != nil {
			return fmt.Errorf("failed to build worker-node metering data writer: %w", err)
		}

		workerNodeCheckpointRepo := workernodemeteringcheckpoint.NewWorkerNodeMeteringCheckpointRepo(dbProvider)

		tilog.PlainInfo("worker-node metering job is enabled")
		workerNodeMeteringJob := cron.NewWorkerNodeMeteringJob(
			workerNodeCheckpointRepo,
			dataplaneManager,
			sharedPoolRepo,
			lockRepo,
			workerNodeMeteringDataWriter,
			cfg.ServerConfig.Region,
			cfg.ServerConfig.CloudProvider,
			cfg.ServerConfig.ControlPlaneMetering.WorkerNode.CompensationWindow,
			"@every 1m",
		)
		if err := cronMgr.AddJob(workerNodeMeteringJob); err != nil {
			return fmt.Errorf("failed to add worker-node metering job: %w", err)
		}
	}

	if cron.ShouldEnableReservedNodeMetering(cfg.ServerConfig.ControlPlaneMetering) {
		reservedNodeMeteringDataWriter, err := cfg.ServerConfig.ControlPlaneMetering.BuildMeteringDataWriter()
		if err != nil {
			return fmt.Errorf("failed to build reserved-node metering data writer: %w", err)
		}

		reservedNodeCheckpointRepo := reservednodemeteringcheckpoint.NewReservedNodeMeteringCheckpointRepo(dbProvider)

		tilog.PlainInfo("reserved-node metering job is enabled")
		reservedNodeMeteringJob := cron.NewReservedNodeMeteringJob(
			reservedNodeCheckpointRepo,
			dataplaneManager,
			sharedPoolRepo,
			tidbClusterRepo,
			lockRepo,
			reservedNodeMeteringDataWriter,
			cfg.ServerConfig.Region,
			cfg.ServerConfig.CloudProvider,
			cfg.ServerConfig.ControlPlaneMetering.ReservedNode.CompensationWindow,
			"@every 1m",
		)
		if err := cronMgr.AddJob(reservedNodeMeteringJob); err != nil {
			return fmt.Errorf("failed to add reserved-node metering job: %w", err)
		}
	}
	// sync standby tidbgroup job
	syncStandbyTiDBGroupJob := cron.NewSyncStandbyTiDBGroupJob(
		dataplaneManager,
		ngService,
		lockRepo,
		"@every 1m",
		cfg.ServerConfig.RUConfig.DefaultStandbyReplicas,
		cfg.ServerConfig.RUConfig.TenantStandbyReplicas,
	)
	if err := cronMgr.AddJob(syncStandbyTiDBGroupJob); err != nil {
		return fmt.Errorf("failed to add sync standby tidbgroup job: %w", err)
	}

	if cfg.ServerConfig.CronJob.EnableTemplateVersionUsageDetector {
		tilog.PlainInfo("template version usage detector job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.TemplateVersionUsageDetectorCron),
		)
		templateVersionUsageDetectorJob := cron.NewTemplateVersionUsageDetectorJob(
			ngService,
			templateVersionSvc,
			lockRepo,
			cfg.ServerConfig.CronJob.TemplateVersionUsageDetectorCron,
		)
		if err := cronMgr.AddJob(templateVersionUsageDetectorJob); err != nil {
			return fmt.Errorf("failed to add template version usage detector job: %w", err)
		}
	}

	// create shortage detector job
	if cfg.ServerConfig.CronJob.EnableShortageDetector {
		// Generate unique owner ID for this instance
		ownerID := os.Getenv("POD_NAME")
		if ownerID == "" {
			ownerID = uuid.New().String()
		}

		tilog.PlainInfo("shortage detector job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.ShortageDetectorCron),
			zap.String("ownerID", ownerID),
		)
		shortageDetectorJob := cron.NewShortageDetectorJob(
			sharedPoolRepo,
			nodeGroupStateRepo,
			nodeGroupLockRepo,
			nodeGroupRepo,
			dataplaneNodeGroupRepo,
			dataplaneMgrClient,
			dataplaneManager,
			ownerID,
			cfg.ServerConfig.CronJob.ShortageDetectorCron,
		)
		if err := cronMgr.AddJob(shortageDetectorJob); err != nil {
			return fmt.Errorf("failed to add shortage detector job: %w", err)
		}
	}

	// create retiring processor job
	if cfg.ServerConfig.CronJob.EnableRetiringProcessor {
		tilog.PlainInfo("retiring processor job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.RetiringProcessorCron),
		)
		retiringProcessorJob := cron.NewRetiringProcessorJob(
			nodeGroupRepo,
			dataplaneNodeGroupRepo,
			dataplaneMgrClient,
			lockRepo,
			cfg.ServerConfig.CronJob.RetiringProcessorCron,
		)
		if err := cronMgr.AddJob(retiringProcessorJob); err != nil {
			return fmt.Errorf("failed to add retiring processor job: %w", err)
		}
	}

	// RCU scale down job
	if cfg.ServerConfig.CronJob.EnableRcuScaleDown {
		tilog.PlainInfo("rcu scale down job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.RcuScaleDownCron),
		)
		rcuScaleDownJob := cron.NewRcuScaleDownJob(
			ngService,
			lockRepo,
			cfg.ServerConfig.CronJob.RcuScaleDownCron,
			cfg.ServerConfig.CronJob.RcuScaleDownMaxRu,
		)
		if err := cronMgr.AddJob(rcuScaleDownJob); err != nil {
			return fmt.Errorf("failed to add rcu scale down job: %w", err)
		}
	}

	if cfg.ServerConfig.CronJob.EnableDMWorkerAutoscaler {
		tilog.PlainInfo("DM worker autoscaler job is enabled",
			zap.String("cron", cfg.ServerConfig.CronJob.DMWorkerAutoscalerCron),
			zap.Duration("scaleDownStability", cfg.ServerConfig.CronJob.DMWorkerScaleDownStability),
		)
		dmWorkerAutoscalerJob := cron.NewDMWorkerAutoscalerJob(
			dmClusterRepo,
			sharedPoolRepo,
			lockRepo,
			ngService,
			clusterGateway,
			cfg.ServerConfig.CronJob.DMWorkerAutoscalerCron,
			cfg.ServerConfig.CronJob.DMWorkerScaleDownStability,
		)
		if err := cronMgr.AddJob(dmWorkerAutoscalerJob); err != nil {
			return fmt.Errorf("failed to add DM worker autoscaler job: %w", err)
		}
	}

	cronMgr.Start()
	return nil
}