package server

import (
	"context"
	"crypto/tls"
	"fmt"
	"net"
	"net/http"
	"path/filepath"
	"time"

	sql "github.com/go-sql-driver/mysql"
	externalworkloadpb "github.com/pingcap/kvproto/pkg/externalworkloadpb"
	"github.com/pingcap/tidb/pkg/util"
	"github.com/soheilhy/cmux"
	"github.com/tidbcloud/aws-shared-provider/pkg/scheme"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/admin"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/autoanalyze"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/bgtask"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/gc"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/gcv2"
	scalerpb "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/proto/scaler"
	remotequery "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/remote-query"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/poolmanager"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/scheduler"
	rm "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/server/service"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/ttl_task"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/types"
	pd "github.com/tikv/pd/client"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"google.golang.org/grpc"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	ctrl "sigs.k8s.io/controller-runtime"
	zapopt "sigs.k8s.io/controller-runtime/pkg/log/zap"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
)

type Server struct {
	manager.Manager
	apiServer             *http.ServeMux
	grpcServer            *grpc.Server
	tlsConfig             *tls.Config
	db                    *gorm.DB
	cfg                   *tidbworker.Config
	workerResourceManager *scheduler.Scheduler
	poolManager           rm.PoolManager
}

const (
	leaderElectionID = "tidb-worker"
	defaultDBName    = "serverless"
)

const (
	webhookPrefix     = "/scaler/api/v1/"
	connectionTimeout = 3 * time.Second
)

func New(cfg *tidbworker.Config) (*Server, error) { //nolint:funlen // for readability
	zapOpts := zapopt.Options{
		Development: true,
		TimeEncoder: zapcore.ISO8601TimeEncoder,
		ZapOpts:     []zap.Option{zap.AddCaller()},
	}
	ctrl.SetLogger(zapopt.New(zapopt.UseFlagOptions(&zapOpts)))

	tlsConfig, err := util.NewTLSConfig(
		util.WithCAPath(cfg.Security.CA),
		util.WithCertAndKeyPath(cfg.Security.Cert, cfg.Security.Key),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to load tls config: %w", err)
	}
	if tlsConfig != nil {
		tlsConfig.NextProtos = []string{"http/1.1", "h2"}
	}

	var certDir, certName, keyName string
	if tlsConfig != nil {
		certDir = filepath.Dir(cfg.Security.CA)
		certName = filepath.Base(cfg.Security.Cert)
		keyName = filepath.Base(cfg.Security.Key)
	}
	mgr, err := ctrl.NewManager(cfg.KubeConfig, manager.Options{
		Scheme:           scheme.Scheme,
		LeaderElection:   cfg.EnableLeaderElection,
		LeaderElectionID: leaderElectionID,
		Metrics: metricsserver.Options{
			BindAddress: cfg.MetricBindAddr,
			// Set metric server tls config.
			SecureServing: tlsConfig != nil,
			CertDir:       certDir,
			CertName:      certName,
			KeyName:       keyName,
			TLSOpts: []func(t *tls.Config){
				func(t *tls.Config) {
					t.ClientAuth = tls.VerifyClientCertIfGiven
					t.MinVersion = tls.VersionTLS12
					t.Certificates = tlsConfig.Certificates
					t.RootCAs = tlsConfig.RootCAs
					t.ClientCAs = tlsConfig.ClientCAs
					t.NextProtos = tlsConfig.NextProtos
				},
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create manager: %w", err)
	}

	db, err := initDB(cfg.DSN)
	if err != nil {
		return nil, err
	}

	svr := &Server{
		Manager:    mgr,
		apiServer:  http.NewServeMux(),
		grpcServer: grpc.NewServer(),
		tlsConfig:  tlsConfig,
		db:         db,
		cfg:        cfg,
	}

	pdCli, err := pd.NewClient(cfg.PDAddr, pd.SecurityOption{
		KeyPath:  cfg.Security.Key,
		CertPath: cfg.Security.Cert,
		CAPath:   cfg.Security.CA,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create pd client: %w", err)
	}

	adminConfig, err := admin.NewReconciler(mgr, db)
	if err != nil {
		return nil, fmt.Errorf("failed to create config watcher: %w", err)
	}
	svr.apiServer.Handle(webhookPrefix+"admin/", adminConfig)

	if cfg.EnableResourceManager {
		svr.poolManager = poolmanager.NewManager(cfg.Namespace, mgr.GetClient(), &rm.PoolManagerConfig{
			TLSConfig: tlsConfig,
			ConfigWatcherConfig: &rm.ConfigWatcherConfig{
				PollInterval:    10 * time.Second, //nolint:mnd // 10s is reasonable
				WorkerConfigMap: "worker-config",
				PoolsConfigMap:  "pools-config",
				WorkerSecret:    cfg.WorkerCertSecret,
			},
			StandbyWatcherConfig: &rm.StandbyWatcherConfig{
				ProbeInterval: 15 * time.Second, //nolint:mnd // 10s is reasonable
			},
			ReconcilerConfig: &rm.ReconcilerConfig{
				ReconcileInterval: 15 * time.Second, //nolint:mnd // 10s is reasonable
			},
		})
		svr.workerResourceManager, err = scheduler.NewScheduler(db, svr.poolManager)
		if err != nil {
			return nil, fmt.Errorf("failed to resource manager: %w", err)
		}

		if err := svr.Add(svr.workerResourceManager); err != nil {
			return nil, fmt.Errorf("failed to add resource manager: %w", err)
		}
	}

	if cfg.EnableGcV1 {
		err = svr.registerHandler(webhookPrefix+"gc/v1/", gc.NewHandler(
			cfg.Namespace,
			mgr,
			db,
			pdCli,
			gc.WithAdminConfig(adminConfig),
			gc.WithWorkerCertSecret(cfg.WorkerCertSecret),
			gc.WithJobLimit(cfg.GcV1JobLimit),
			gc.WithResourceManager(svr.workerResourceManager, cfg.EnableGCResourceManager),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add gc handler: %w", err)
		}
	}

	if cfg.EnableGcV2 {
		err = svr.registerHandler(webhookPrefix+"gc/v2/", gcv2.NewHandler(
			cfg.Namespace,
			mgr,
			db,
			pdCli,
			gcv2.WithAdminConfig(adminConfig),
			gcv2.WithWorkerCertSecret(cfg.WorkerCertSecret),
			gcv2.WithJobLimit(cfg.GcV2JobLimit),
			gcv2.WithResourceManager(svr.workerResourceManager, cfg.EnableGCV2ResourceManager),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add gc v2 handler: %w", err)
		}
	}

	if cfg.EnableBgTask {
		err = svr.registerHandler(webhookPrefix+"bgtask/", bgtask.NewHandler(
			cfg.Namespace,
			mgr,
			db,
			pdCli,
			bgtask.WithAdminConfig(adminConfig),
			bgtask.WithWorkerCertSecret(cfg.WorkerCertSecret),
			bgtask.WithBatchJobLimit(cfg.BatchJobLimit),
			bgtask.WithDDLJobLimit(cfg.DDLJobLimit),
			bgtask.WithImportIntoJobLimit(cfg.ImportIntoJobLimit),
			bgtask.WithBatchWorkerCount(cfg.BatchWorkerCount),
			bgtask.WithDDLWorkerCount(cfg.DDLWorkerCount),
			bgtask.WithImportIntoWorkerCount(cfg.ImportIntoWorkerCount),
			bgtask.WithComponentNode(cfg.BgTaskNodeGroup),
			bgtask.WithEnableWorkerAutoScale(cfg.EnableWorkerAutoScale),
			bgtask.WithDisableBatchWorkerAutoScale(cfg.DisableBatchWorkerAutoScale),
			bgtask.WithDisableDDLWorkerAutoScale(cfg.DisableDDLWorkerAutoScale),
			bgtask.WithDisableImportIntoWorkerAutoScale(cfg.DisableImportIntoWorkerAutoScale),
			bgtask.WithMaxSharedWorkerCount(cfg.MaxSharedWorkerCount),
			bgtask.WithMaxSharedWorkerCountPerKeyspace(cfg.MaxSharedWorkerCountPerKeyspace),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add bg handler: %w", err)
		}
	}

	if cfg.EnableTTLTask {
		err = svr.registerHandler(webhookPrefix+"ttlTask/", ttl_task.NewHandler(
			cfg.Namespace,
			mgr,
			db,
			pdCli,
			svr.workerResourceManager,
			ttl_task.WithAdminConfig(adminConfig),
			ttl_task.WithWorkerCertSecret(cfg.WorkerCertSecret),
			ttl_task.WithJobLimit(cfg.TTLJobLimit),
			ttl_task.WithTestKeyspace(cfg.TestRMKeyspaces),
			ttl_task.WithEnableResourceManager(cfg.EnableTTLResourceManager),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add bg handler: %w", err)
		}
	}

	if cfg.EnableRemoteQuery {
		err = svr.registerHandler(webhookPrefix+"remote-query/", remotequery.NewHandler(
			cfg.Namespace,
			mgr,
			remotequery.WithAdminConfig(adminConfig),
			remotequery.WithWorkerCertSecret(cfg.WorkerCertSecret),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add remote query handler: %w", err)
		}
	}

	if cfg.EnableAutoAnalyze {
		err = svr.registerHandler(webhookPrefix+"auto-analyze/", autoanalyze.NewHandler(
			cfg.Namespace,
			mgr,
			db,
			pdCli,
			autoanalyze.WithAdminConfig(adminConfig),
			autoanalyze.WithWorkerCertSecret(cfg.WorkerCertSecret),
			autoanalyze.WithJobLimit(cfg.AutoAnalyzeLimit),
		))
		if err != nil {
			return nil, fmt.Errorf("failed to add auto analyze handler: %w", err)
		}
	}

	return svr, nil
}

func initDB(dsn string) (*gorm.DB, error) {
	cfg, err := sql.ParseDSN(dsn)
	if err != nil {
		return nil, fmt.Errorf("failed to parse dsn: %w", err)
	}

	dbName := defaultDBName
	if cfg.DBName != "" {
		dbName = cfg.DBName
	} else {
		cfg.DBName = dbName
	}

	// Create database `dbName` if not exists
	baseCfg := cfg.Clone()
	baseCfg.DBName = ""
	baseDsn := baseCfg.FormatDSN()
	dialer := mysql.Open(baseDsn)
	db, err := gorm.Open(dialer)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to base database: %w", err)
	}

	err = db.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s", dbName)).Error
	if err != nil {
		return nil, fmt.Errorf("failed to create database `%s`: %w", dbName, err)
	}

	// Connect to database `dbName`
	dialer = mysql.Open(cfg.FormatDSN())
	db, err = gorm.Open(dialer)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to database `%s`: %w", dbName, err)
	}

	return db, nil
}

func (s *Server) registerHandler(path string, handler types.Handler) error {
	err := s.Add(handler)
	if err != nil {
		return err
	}
	if h, ok := handler.(http.Handler); ok {
		s.apiServer.Handle(path, h)
	}
	return nil
}

func (s *Server) Start(ctx context.Context, apiServerAddr string) error {
	if apiServerAddr == "" {
		return s.Manager.Start(ctx)
	}

	// Setup listener.
	ln, err := net.Listen("tcp", apiServerAddr)
	if err != nil {
		ctrl.Log.Error(err, "failed to listen")
		return err
	}
	secure := s.tlsConfig != nil
	// Upgrade the listener to TLS if necessary.
	if secure {
		ln = tls.NewListener(ln, s.tlsConfig)
	}
	mux := cmux.New(ln)
	mux.SetReadTimeout(connectionTimeout)
	httpListener := mux.Match(cmux.HTTP1Fast())
	grpcListener := mux.Match(cmux.Any())

	// Start http server.
	go func() {
		ctrl.Log.Info("starting api server", "addr", apiServerAddr, "tls", secure)
		if err := http.Serve(httpListener, s.apiServer); err != nil {
			ctrl.Log.Error(err, "failed to start api server")
		}
	}()

	// Start gRPC server.
	go func() {
		ctrl.Log.Info("starting grpc server", "addr", apiServerAddr, "tls", secure)
		scalerService := &service.ScalerService{Server: s}
		scalerpb.RegisterScalerServer(s.grpcServer, scalerService)
		externalworkloadpb.RegisterExternalWorkloadControllerServer(
			s.grpcServer,
			service.NewExternalWorkloadService(scalerService),
		)
		if err := s.grpcServer.Serve(grpcListener); err != nil {
			ctrl.Log.Error(err, "failed to start grpc server")
		}
	}()

	// Start cmux.
	go func() {
		ctrl.Log.Info("starting mux server", "tls", secure)
		if err := mux.Serve(); err != nil {
			ctrl.Log.Error(err, "failed to start mux server")
		}
	}()

	return s.Manager.Start(ctx)
}

func (s *Server) GetDB() *gorm.DB {
	return s.db
}

func (s *Server) GetConfig() *tidbworker.Config {
	return s.cfg
}