package managerserver

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

	grpcprom "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/config"
	pbm "github.com/tidbcloud/aws-shared-provider/pbgen/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/controllers/keyspacepoolconfig"
	"github.com/tidbcloud/aws-shared-provider/pkg/infra"
	"github.com/tidbcloud/aws-shared-provider/pkg/keyspacemigration"
	wrapmanager "github.com/tidbcloud/aws-shared-provider/pkg/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/notifier"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbmanager"
	"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"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	clientv3 "go.etcd.io/etcd/client/v3"
	"golang.org/x/net/http2"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/certwatcher"
	"sigs.k8s.io/controller-runtime/pkg/manager"
)

const (
	MaxConcurrentStreams = 10000
)

var (
	log = ctrl.Log.WithName("managerserver")

	noTokenErr     = errors.New("no token")
	noTidbPoolErr  = errors.New("no tidb pool")
	metaNotSyncErr = errors.New("cluster meta not synced yet")
)

type RunEveryManager interface {
	manager.Runnable
	manager.LeaderElectionRunnable
}

type server struct {
	ctx                context.Context
	manager            wrapmanager.Manager
	tManagers          map[string]tidbmanager.TidbManager
	notifier           *notifier.Notifier
	etcdKV             clientv3.KV
	etcdLease          clientv3.Lease
	tidbController     tidb.Controller
	infraClusterCli    infraclient.Client
	infraApi           infra.InfraAPI
	etcdTokenPrefixMap map[string]string
	cfg                *config.Config

	pdClient        pd.HttpPdClient
	tidbWorkerApi   tidbworker.Api
	migrationLocker keyspacemigration.Locker
	// Shared with keyspace-pool reconcilers so API writes update the runtime rate immediately.
	createRateStore *keyspacepoolconfig.CreateRateStore

	defaultTidbPool string

	certDir               string
	etcdCheckConnPrefix   string
	httpServerBindAddress string
	enableGrpcReflection  bool
}

func New(
	cfg *config.Config,
	manager wrapmanager.Manager,
	tManagers map[string]tidbmanager.TidbManager,
	notifier *notifier.Notifier,
	etcdClient *clientv3.Client,
	tidbController tidb.Controller,
	infraClusterCli infraclient.Client,
	etcdTokenPrefixMap map[string]string,
	infraApi infra.InfraAPI,
	pdClient pd.HttpPdClient,
	tidbWorkerApi tidbworker.Api,
	migrationLocker keyspacemigration.Locker,
	createRateStore *keyspacepoolconfig.CreateRateStore,
) (RunEveryManager, error) {
	return &server{
		manager:            manager,
		tManagers:          tManagers,
		notifier:           notifier,
		etcdKV:             etcdClient.KV,
		etcdLease:          etcdClient.Lease,
		tidbController:     tidbController,
		infraClusterCli:    infraClusterCli,
		etcdTokenPrefixMap: etcdTokenPrefixMap,

		pdClient:        pdClient,
		tidbWorkerApi:   tidbWorkerApi,
		migrationLocker: migrationLocker,
		createRateStore: createRateStore,

		// The first element is default TidbPool
		defaultTidbPool: cfg.TidbPoolNamespaceList[0],

		certDir:               cfg.CertDir,
		etcdCheckConnPrefix:   cfg.EtcdCheckConnPrefix,
		httpServerBindAddress: cfg.HttpServerBindAddress,
		enableGrpcReflection:  cfg.EnableGrpcReflection,
		infraApi:              infraApi,
		cfg:                   cfg,
	}, nil
}

func (s *server) Start(ctx context.Context) error {
	s.ctx = ctx

	certWatcher, err := certwatcher.New(filepath.Join(s.certDir, "tls.crt"), filepath.Join(s.certDir, "tls.key"))
	if err != nil {
		return err
	}
	go func() {
		if err := certWatcher.Start(ctx); err != nil {
			log.Error(err, "certificate watcher error")
		}
	}()
	go func() {
		if err := s.infraApi.Start(ctx); err != nil {
			log.Error(err, "infra API start error")
		}
	}()
	sync := s.infraApi.WaitForCacheSync(ctx)
	log.Info("infra api cache sync", "result", sync)

	gServer := grpc.NewServer(
		grpc.StreamInterceptor(grpcprom.StreamServerInterceptor),
		grpc.UnaryInterceptor(grpcprom.UnaryServerInterceptor),
	)
	pbm.RegisterTidbBackendServer(gServer, &grpcServer{ctx: ctx, tManagers: s.tManagers, notifier: s.notifier})
	if s.enableGrpcReflection {
		log.Info("enable grpc reflection")
		reflection.Register(gServer)
	}

	mux := http.NewServeMux()
	mux.HandleFunc("/healthz", s.healthz)
	mux.HandleFunc("/readyz", s.readyz)
	mux.HandleFunc("/api/tidb/assign", s.assign)
	mux.HandleFunc("/api/tidb/check", s.check)
	mux.HandleFunc("/api/tidb/free", s.free)
	mux.HandleFunc("/api/tidb/replace", s.replace)
	mux.HandleFunc("/api/tidb/checkconn", s.checkconn)
	mux.HandleFunc("/api/tidb/resource_group", s.resourceGroup)
	mux.HandleFunc("/api/tidb/status", s.status)
	mux.HandleFunc("/api/tidb/owner", s.getTiDBOwner)

	mux.HandleFunc("/api/import/status", s.importStatus)

	mux.HandleFunc("/api/tidb/get_assign_tidb", s.getAssignTiDB)
	mux.HandleFunc("/api/tidb/kill_assign_tidb", s.killAssignTiDBByKeyspaceName)
	mux.HandleFunc("/api/tidb/get_active_tidb", s.getActiveTiDB)

	mux.HandleFunc("/api/tidb/get_active_import_jobs", s.getImportJobs)
	mux.HandleFunc("/api/tidb/get_active_export_jobs", s.getExportJobs)

	mux.HandleFunc("/api/keyspace/migrate_meta_service_group", s.migrateMetaServiceGroup)
	mux.HandleFunc("/api/keyspace/update_pool_create_rate", s.updateKeyspacePoolCreateRate)

	mux.HandleFunc("/debug/pprof/", pprof.Index)
	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)

	server := &http.Server{
		BaseContext: func(net.Listener) context.Context { return ctx },
		Addr:        s.httpServerBindAddress,
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if r.ProtoMajor == 2 && strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
				log.V(3).Info("grpc request", "url", r.URL, "Head", r.Header, "Proto", r.Proto, "remote-addr", r.RemoteAddr)
				gServer.ServeHTTP(w, r)
			} else {
				log.V(3).Info("http request", "url", r.URL, "Head", r.Header, "Proto", r.Proto, "remote-addr", r.RemoteAddr)
				mux.ServeHTTP(w, r)
			}
		}),
		TLSConfig: &tls.Config{GetCertificate: certWatcher.GetCertificate}, //nolint: gosec // not need
	}
	if err := http2.ConfigureServer(server, &http2.Server{MaxConcurrentStreams: MaxConcurrentStreams}); err != nil {
		return err
	}
	go func() {
		log.Info("start managerserver", "bind-addr", s.httpServerBindAddress)
		_ = server.ListenAndServeTLS("", "")
	}()

	<-ctx.Done()
	//nolint: gomnd // not need
	delayCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	err = server.Shutdown(delayCtx)
	log.Info("stop managerserver", "return err", err)
	return err
}

func (s *server) NeedLeaderElection() bool {
	return false
}