package main

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

	"github.com/gin-contrib/pprof"
	"github.com/gin-gonic/gin"
	"github.com/go-logr/logr"
	grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpcctxtags "github.com/grpc-ecosystem/go-grpc-middleware/tags"
	meteringsdkmetawriter "github.com/pingcap/metering_sdk/writer/meta"
	"github.com/spf13/cobra"
	execpbv1 "github.com/tidbcloud/cluster-service-ng/api/executor/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/executorgrpc"
	"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/dummyworkflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	stgdplanemgr "github.com/tidbcloud/cluster-service-ng/pkg/storage/dataplanemanager"
	stgetcd "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/runtime"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql"
	autoscalepolicybundle "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/autoscale_policy_bundle"
	"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"
	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"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/restore"
	ruconfigrds "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/ruconfig"
	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"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/tiinf"
	"github.com/tidbcloud/tiinf/transport/http/tigin"
	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"
	"k8s.io/client-go/tools/clientcmd"
	kubeclient "sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

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

var (
	externalKubeConfigPath string
	configPath             string
	enableGRPCReflection   = false
	cleanupInterval        time.Duration
	keepFinishedDuration   time.Duration
	maxFinishedTasks       int64
	cmd                    = &cobra.Command{}

	serveCmd = &cobra.Command{
		Use:   "server",
		Short: "run grcp server for executor",
		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(&enableGRPCReflection, "enable-grpc-reflection", false, "enable gRPC reflection for debugging (DO NOT use in production)")
	cmd.PersistentFlags().DurationVar(&cleanupInterval, "cleanup-interval", 0, "interval for finished task cleanup cycle (e.g. 3h). 0 uses default 3h")
	cmd.PersistentFlags().DurationVar(&keepFinishedDuration, "keep-finished-duration", 0, "how long finished tasks are retained before deletion (e.g. 48h). 0 uses default 48h")
	cmd.PersistentFlags().Int64Var(&maxFinishedTasks, "max-finished-tasks", 1000, "max number of finished tasks to retain, oldest deleted first. 0 means no limit")
	cmd.AddCommand(serveCmd)
}

func main() {
	if err := cmd.Execute(); err != nil {
		tilog.PlainError("execute fialed", 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()
	if err != nil {
		return fmt.Errorf("init tiinf failed. %w", err)
	}

	if enableGRPCReflection {
		reflection.Register(inf.GrpcServer())
	}

	// init dependencies

	execGrpcWrapper, err := executorgrpc.NewExecutorGrpcWrapper()
	if err != nil {
		return fmt.Errorf("new executor grpc wrapper failed. %w", err)
	}
	// init k8s(etcd) client
	restCfg, err := clientcmd.BuildConfigFromFlags("", externalKubeConfigPath)
	if err != nil {
		return fmt.Errorf("build k8s config failed. %w", err)
	}

	kubeClient, err := k8s.NewClient(context.Background(), restCfg)
	if err != nil {
		return fmt.Errorf("new kube client failed. %w", err)
	}
	// disable some noisy log
	log.SetLogger(logr.Discard())

	dataplaneMgrClient, err := dataplanemgr.NewClient(cfg.ServerConfig.DataPlaneManagerAddr)
	if err != nil {
		return fmt.Errorf("failed to create dataplane manager client: %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)
	}
	dataplaneManager := dataplane.NewManager(restCfg, dataplaneRepo) // init dataplane client manager

	meteringWriter, err := cfg.ServerConfig.ControlPlaneMetering.BuildMeteringMetaWriter()
	if err != nil {
		return fmt.Errorf("build metering metadata writer failed. %w", err)
	}
	defer func() { _ = meteringWriter.Close() }()

	{
		// create dag executor executor
		dagExecutor, err := startCSNGDagExecutor(
			cfg.ServerConfig,
			kubeClient,
			dataplaneManager,
			dataplaneRepo,
			dataplaneMgrClient,
			meteringWriter,
		)
		if err != nil {
			return fmt.Errorf("start dag executor failed. %w", err)
		}
		execGrpcWrapper.AddExecutor(dag.ExecutorScope, dagExecutor)
	}
	// {
	// 	// create dummy executor
	// 	dummyExecutor, err := startDummyExecutor(kubeClient)
	// 	if err != nil {
	// 		return fmt.Errorf("start dummy executor failed. %w", err)
	// 	}
	// 	execGrpcWrapper.AddExecutor(dummyworkflow.ExecutorScope, dummyExecutor)
	// }

	// register grpc server.
	execpbv1.RegisterExecutorServiceServer(inf.GrpcServer(), execGrpcWrapper)

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

	return nil
}

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

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

	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
}

// TODO(shenjun): may pass tidbgroup and sharedpool repo in args
func startCSNGDagExecutor(
	cfg *config.ServerConfig,
	kubeClient kubeclient.Client,
	dataplaneManager dataplane.Manager,
	dataplaneRepo repo.DataplaneRepo,
	dataplaneMgrClient dataplanemgr.Client,
	metaWriter *meteringsdkmetawriter.MetaWriter,
) (*executor.Executor, error) {
	var clusterConfigClient clusterconfig.Client
	if cfg.DevConfig != nil && cfg.DevConfig.MockConfigCenter {
		tilog.PlainWarn("using mock config center client (dev mode)")
		clusterConfigClient = clusterconfig.NewMockClient()
	} else {
		var err error
		clusterConfigClient, err = clusterconfig.NewClient(cfg.ConfigCenterAddr)
		if err != nil {
			return nil, 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
		lwRestoreRepo             repo.LightweightRestoreRepo
		classicRestoreRepo        repo.ClassicRestoreRepo
		manualBackupRepo          repo.ManualBackupRepo
		scaleTikvDecisionRepo     repo.ScaleTikvDecisionRepo
		sharedPoolObjectStoreRepo repo.SharedPoolObjectStoreRepo
		backupGcSettingRepo       repo.BackupGcSettingRepo
		nodeGroupStateRepo        repo.DataplaneNodeInstanceGroupStateRepo
	)
	// if RDS is enabled, we use rds storage for shared pool and tidb group
	db := mysql.NewDB(&cfg.MysqlConfig)
	dbProvider, _, transactioner := mysql.NewDBProvider(db)
	templateRegistryRepo := templateregistryrds.NewTemplateRegistryRepo(dbProvider)
	templateVersionSvc := templateversion.NewService(templateRegistryRepo, cfg.BizConfig.TemplateVersionMaxLag)

	// Bootstrap template registry under global lock (same as cluster_ng_server).
	templateMap, loadErr := templateversion.LoadTemplatesFromPath(cfg.TemplatePath)
	if loadErr != nil {
		return nil, fmt.Errorf("load templates from path failed: %w", loadErr)
	}
	lockRepo := lockrds.NewLockRepo(dbProvider)
	unlockTemplateBootstrap, lockErr := lockRepo.Lock(context.Background(), &entity.Lock{
		Key:         entity.LockKeyTemplateBootstrap,
		TimeoutSecs: 600,
	})
	if lockErr != nil {
		return nil, fmt.Errorf("failed to lock template bootstrap: %w", lockErr)
	}
	if bootstrapErr := transactioner.Transaction(context.Background(), func(ctx context.Context) error {
		return templateVersionSvc.Bootstrap(ctx, templateMap)
	}); bootstrapErr != nil {
		_ = unlockTemplateBootstrap()
		return nil, fmt.Errorf("template bootstrap failed: %w", bootstrapErr)
	}
	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())
	pauseResumeOperationRepo := pauseresumeoperationrds.New(dbProvider)
	tidbClusterRepo = tidbclusterds.NewClusterRepo(dbProvider, clusterConfigClient, dataplaneManager)
	tidbGroupRepo = tidbgrouprds.NewTiDBGroupRepo(dbProvider, dataplaneManager, dataplaneMgrClient)
	dmClusterRepo = dmclusterrds.NewDMClusterRepo(dbProvider, dataplaneManager)
	exportRepo = exportrds.NewExportRepo(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)
	nodeGroupStateRepo = nodegrouprds.NewDataplaneNodeInstanceGroupStateRepo(dbProvider)
	nodeGroupLockRepo := nodegrouprds.NewDataplaneNodeInstanceGroupLockRepo(dbProvider)
	dataplaneNodeGroupRepo := nodegrouprds.NewDataplaneNodeInstanceGroupRepo(dbProvider)
	autoscalePolicyBundleRepo := autoscalepolicybundle.NewAutoscalePolicyBundleRepo(dbProvider)
	ruConfigRepo, err := ruconfigrds.NewRuConfigRepo(dbProvider, cfg.RUConfig)
	if err != nil {
		// nil repo keeps TiProxy ScaleDown at Disabled (same as applyTiProxyScaleInGate).
		tilog.PlainWarn("new ru config repo failed, tiproxy scale-in gate keeps baseline", zap.Error(err))
	}

	clustergatewayCli := clustergateway.NewClusterGatewayImpl()
	dataplaneClient, err := dataplanemgr.NewClient(cfg.DataPlaneManagerAddr)
	if err != nil {
		return nil, fmt.Errorf("new dataplane manager client failed. %w", err)
	}

	exec, err := dag.NewDagExecutor(
		sharedPoolRepo,
		pauseResumeOperationRepo,
		lockRepo,
		cassandraProxyStateRepo,
		tidbClusterRepo,
		tidbGroupRepo,
		dmClusterRepo,
		exportRepo,
		lwRestoreRepo,
		classicRestoreRepo,
		manualBackupRepo,
		manualBackupRestoreRepo,
		scaleTikvDecisionRepo,
		sharedPoolObjectStoreRepo,
		backupGcSettingRepo,
		dataplaneRepo,
		nodeGroupStateRepo,
		nodeGroupLockRepo,
		dataplaneNodeGroupRepo,
		autoscalePolicyBundleRepo,
		ruConfigRepo,
		dataplaneManager,
		dataplaneClient,
		clustergatewayCli,
		clusterConfigClient,
		kubeClient,
		metaWriter,
		templateVersionSvc,
		cfg,
		transactioner,
	)
	if err != nil {
		return nil, fmt.Errorf("new dag executor failed. %w", err)
	}
	sched := executor.NewBackOffScheduler(10 * time.Second)
	exec.ChangeActionTaskScheduler(sched)

	// Apply executor cleanup configuration from startup flags.
	if cleanupInterval > 0 {
		exec.SetCleanupInterval(cleanupInterval)
	}
	if keepFinishedDuration > 0 {
		exec.SetKeepDeletedDuration(keepFinishedDuration)
	}
	exec.SetMaxFinishedTasks(maxFinishedTasks)

	err = exec.Start(executor.ExitPolicyNonStop)
	return exec, err
}

//nolint:unused // for debug
func startDummyExecutor(kubeClient kubeclient.Client) (*executor.Executor, error) {
	actionLib := dummyworkflow.BuildActionLib()
	logicPlans := dummyworkflow.BuildPlans()
	stg := stgetcd.NewExecutorTaskStorage(dummyworkflow.ExecutorScope, kubeClient)
	exec, err := executor.NewExecutor(actionLib, logicPlans, stg, stg, 10)
	if err != nil {
		return nil, err
	}
	sched := executor.NewBackOffScheduler(10 * time.Second)
	exec.ChangeActionTaskScheduler(sched)
	logger := executor.NewGeneralLogger("dummy-executor", os.Stdout)
	exec.RegisterEventReceiver(logger)
	err = exec.Start(executor.ExitPolicyNonStop)
	return exec, err
}