package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"os"
	"reflect"
	"runtime/debug"
	"strconv"
	"strings"
	"time"

	"github.com/getsentry/sentry-go"
	sentryhttp "github.com/getsentry/sentry-go/http"
	"github.com/gin-gonic/gin"
	"github.com/gogo/grpc-example/insecure"
	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth"
	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/validator"
	grpcprometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	grpcruntime "github.com/grpc-ecosystem/grpc-gateway/runtime"
	"github.com/mailgun/mailgun-go/v4"
	"github.com/patrickmn/go-cache"
	centralapi "github.com/pingcap/dbaas/central/api"
	"github.com/pingcap/dbaas/central/config"
	cdcdao "github.com/pingcap/dbaas/central/db/dbutil/cdc"
	clusterdao "github.com/pingcap/dbaas/central/db/dbutil/cluster"
	replicationdao "github.com/pingcap/dbaas/central/db/dbutil/replication"
	dbtenant "github.com/pingcap/dbaas/central/db/tenant"
	"github.com/pingcap/dbaas/central/dbinit"
	grpchttp "github.com/pingcap/dbaas/central/grpc/http"
	api "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/mesh"
	"github.com/pingcap/dbaas/central/metrics"
	backupmetrics "github.com/pingcap/dbaas/central/metrics/backup"
	clustermetrics "github.com/pingcap/dbaas/central/metrics/cluster"
	importmetrics "github.com/pingcap/dbaas/central/metrics/import"
	restoremetrics "github.com/pingcap/dbaas/central/metrics/restore"
	tenantmetrics "github.com/pingcap/dbaas/central/metrics/tenant"
	"github.com/pingcap/dbaas/central/migrate"
	adminmodel "github.com/pingcap/dbaas/central/model/admin"
	"github.com/pingcap/dbaas/central/model/admin/grafana"
	backupmodel "github.com/pingcap/dbaas/central/model/backup"
	billingmodel "github.com/pingcap/dbaas/central/model/billing"
	billingworkermodel "github.com/pingcap/dbaas/central/model/billing/worker"
	cdcmodel "github.com/pingcap/dbaas/central/model/cdc"
	clustermodel "github.com/pingcap/dbaas/central/model/cluster"
	debugmodel "github.com/pingcap/dbaas/central/model/debug"
	topsqlmodel "github.com/pingcap/dbaas/central/model/debug/topsql"
	importmodel "github.com/pingcap/dbaas/central/model/import"
	integrationmodel "github.com/pingcap/dbaas/central/model/integration"
	replicationmodel "github.com/pingcap/dbaas/central/model/replication"
	tokenSession "github.com/pingcap/dbaas/central/model/session"
	"github.com/pingcap/dbaas/central/model/tenant"
	tenantmodel "github.com/pingcap/dbaas/central/model/tenant"
	"github.com/pingcap/dbaas/central/model/testtoken"
	runtimeutil "github.com/pingcap/dbaas/central/runtime/util"
	"github.com/pingcap/dbaas/central/service"
	"github.com/pingcap/dbaas/central/service/auth"
	"github.com/pingcap/dbaas/central/service/backup"
	"github.com/pingcap/dbaas/central/service/billing"
	"github.com/pingcap/dbaas/central/service/cluster"
	connectorservice "github.com/pingcap/dbaas/central/service/connector"
	"github.com/pingcap/dbaas/central/service/debugportal"
	importservice "github.com/pingcap/dbaas/central/service/import"
	"github.com/pingcap/dbaas/central/service/integration"
	svctenant "github.com/pingcap/dbaas/central/service/tenant"
	testokenservice "github.com/pingcap/dbaas/central/service/testtoken"
	opworker "github.com/pingcap/dbaas/central/worker"
	"github.com/pingcap/dbaas/pkg/auth/authuser"
	debugAuth "github.com/pingcap/dbaas/pkg/auth/debug"
	"github.com/pingcap/dbaas/pkg/auth/token"
	grpc_errcode "github.com/pingcap/dbaas/pkg/grpc/middleware/errcode"
	"github.com/pingcap/dbaas/pkg/grpc/middleware/ratelimit"
	grpc_sentry "github.com/pingcap/dbaas/pkg/grpc/middleware/sentry"
	"github.com/pingcap/dbaas/pkg/http/proxy"
	gcpauth "github.com/pingcap/dbaas/pkg/infra/provider/gcp/auth"
	"github.com/pingcap/dbaas/pkg/version"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/sendgrid/sendgrid-go"
	configServiceV1 "github.com/tidbcloud/config-service/api/v1alpha1"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"github.com/tidbcloud/pkgv2/auth/util"
	"github.com/tidbcloud/pkgv2/bizerr"
	grpcbizerr "github.com/tidbcloud/pkgv2/grpc/middleware/bizerr"
	"github.com/tidbcloud/pkgv2/infra/aws"
	pkglog "github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	"github.com/tidbcloud/pkgv2/salesforce"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	secretprovider "github.com/tidbcloud/pkgv2/secret/provider"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/pkgv2/zapext/zapsentry"
	"github.com/tidbcloud/pkgv2/zendesk"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"gopkg.in/auth0.v4/management"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	defaultQPS        = 500
	defaultBurst      = 1000
	FailedStatus      = "failed"
	TokenExpiredError = "expired"
	SystemError       = "system"
	ErrorType         = "err_type"
	defaultTimeout    = 5 * time.Second

	defaultConfigPath = "/etc/dbaas/config.yaml"

	allRole     = "all"
	centralRole = "CENTRAL_ROLE"
	apiService  = "API_SERVICE"
	worker      = "WORKER"
)

var (
	scheme = runtime.NewScheme()
)

func init() {
	_ = clientgoscheme.AddToScheme(scheme)
	_ = configServiceV1.AddToScheme(scheme)
}

// server represents the central server instance
type server struct {
	centralAuth0     auth0.Conf
	portalAuth0      auth0.Conf
	stage            config.Stage
	session          *tokenSession.Session
	mailClient       mailv2.Facade
	zendeskClient    *zendesk.Client
	salesforceClient *salesforce.Client
	tenantModel      *tenant.Model
}

func main() {
	var configPath string
	flag.StringVar(&configPath, "config-path", defaultConfigPath, "dbaas configuration file path")
	flag.Parse()

	err := config.InitConfig(configPath)
	if err != nil {
		// TODO: unexpected error like this, should we throw a alert/page, just log seems not enough ?
		log.Fatal("Initialize configuration failed", zap.Error(err))
	}

	s := &server{
		centralAuth0: auth0.Conf{
			Audience: config.Global().Auth0Config.Auth0Audience,
			Secret:   []byte(config.Global().Auth0Config.Auth0Secret),
			Domain:   config.Global().Auth0Config.Auth0Domain,
		},
		portalAuth0: auth0.Conf{
			Audience:                 config.Global().Auth0Config.ManagementPortalAuth0Audience,
			Secret:                   []byte(config.Global().Auth0Config.ManagementPortalAuth0Secret),
			Domain:                   config.Global().Auth0Config.ManagementPortalAuth0Domain,
			ClientCredentialAudience: config.Global().Auth0Config.ManagementPortalAuth0ClientAudience,
		},
		stage: config.Stage(config.Global().ServerConfig.Stage),
	}

	err = s.serve()
	if err != nil {
		// TODO: unexpected error like this, should we throw a alert/page, just log seems not enough ?
		log.Fatal("failed to serve", zap.Error(err))
	}
}

func (s *server) serve() error {
	if err := setupSentry(); err != nil {
		return errors.Annotate(err, "setupSentry failed")
	}
	var level zapcore.Level
	if err := level.UnmarshalText([]byte(config.Global().ServerConfig.SentryConf.LogLevel)); err != nil {
		return errors.Annotate(err, "invalid sentry.LogLevel")
	}
	sentryOpt := zap.WrapCore(func(core zapcore.Core) zapcore.Core {
		return zapcore.NewTee(core, zapsentry.NewCore(level, sentry.CurrentHub().Client()))
	})

	err := setupLogger(sentryOpt)
	if err != nil {
		return errors.Annotate(err, "initialize logger failed")
	}
	// flush all buffering log
	defer func() {
		if err := log.L().Sync(); err != nil {
			fmt.Println(err.Error())
		}
	}()

	err = gcpauth.InitWebIdentityEnv()
	if err != nil {
		return errors.Annotate(err, "initialize the web identity environment variables failed")
	}

	return s.startup()
}

func setupLogger(opts ...zap.Option) error {
	cfg := &log.Config{
		Level: config.Global().ServerConfig.LogLevel,
	}
	gl, props, err := log.InitLogger(cfg, append(opts, zap.AddStacktrace(zapcore.ErrorLevel))...)
	if err != nil {
		return errors.Trace(err)
	}
	log.ReplaceGlobals(gl, props)

	return nil
}

func setupSentry() error {
	cfg := config.Global().ServerConfig.SentryConf
	opts := sentry.ClientOptions{
		Dsn:         cfg.DSN,
		Environment: cfg.Environment,
	}
	if err := sentry.Init(opts); err != nil {
		log.Error("setupSentry failed", zap.String("dsn", cfg.DSN), zap.String("env", cfg.Environment))
		return err
	}
	return nil
}

func initK8sClient() (client.Client, error) {
	var cfg *rest.Config
	var err error
	kubeConfig := config.Global().GardenerConfig.KubeConfigPath
	if kubeConfig == "" {
		cfg, err = rest.InClusterConfig()
	} else {
		cfg, err = clientcmd.BuildConfigFromFlags("", kubeConfig)
	}
	if err != nil {
		return nil, errors.Annotate(err, "initialize the kubernetes config failed")
	}

	c, err := client.New(cfg, client.Options{Scheme: scheme})
	if err != nil {
		return nil, errors.Annotate(err, "initialize the base kubeclient failed")
	}

	return c, nil
}

func (s *server) startup() error {
	log.Info("starting central service",
		zap.String("gitHash", version.GitHash),
		zap.String("logLevel", config.Global().ServerConfig.LogLevel))

	// gardener cluster k8s client
	var cfg *rest.Config
	var err error
	kubeConfig := config.Global().GardenerConfig.KubeConfigPath
	if kubeConfig == "" {
		cfg, err = rest.InClusterConfig()
	} else {
		cfg, err = clientcmd.BuildConfigFromFlags("", kubeConfig)
	}
	if err != nil {
		return errors.Annotate(err, "initialize the kubernetes config failed")
	}

	// Overwrite some configurations.
	cfg.QPS = defaultQPS
	cfg.Burst = defaultBurst
	cfg.Timeout = defaultTimeout

	// Initialize the k8s/gardener clients.
	baseKubeCli, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		return errors.Annotate(err, "initialize the base kubeclient failed")
	}
	// Initialize Top SQL TLS config
	secretKey := config.Global().ServerConfig.TopSQLSecretStore.SecretKey
	region := config.Global().ServerConfig.TopSQLSecretStore.Region
	secretString, err := config.GetSecretFromAwsSecretManager(secretKey, region)
	if err != nil {
		return errors.Annotate(err, "failed to load secret for TopSQL from AWS secretsmanager")
	}
	topSQLTLSConfig := config.TLSConfig{}
	if err := json.Unmarshal([]byte(secretString), &topSQLTLSConfig); err != nil {
		return errors.Annotate(err, "failed to unmarshal json from AWS secretsmanager")
	}
	config.StoreTopSQLTLS(topSQLTLSConfig)

	gardenConfig, err := runtimeutil.GetGardenerKubeConfigFromBase(baseKubeCli)
	if err != nil {
		return errors.Annotate(err, "retrieve gardener kubeconfig failed")
	}
	gardenKube, err := kubernetes.NewForConfig(gardenConfig)
	if err != nil {
		return errors.Annotate(err, "initialize the gardener client failed")
	}

	gardenerConfig := config.Global().GardenerConfig
	infraAPI := api.InitInfraAPI(baseKubeCli)

	// Initialize global database.
	mysqlConfig := config.Global().MysqlConfig
	log.Info("Initialize global database connection", zap.String("user", mysqlConfig.User))
	_, err = dbinit.Initialize(mysqlConfig, enableWorker())
	if err != nil {
		return errors.Annotate(err, "initialize global database connection failed")
	}

	gcpSecret, err := gardenKube.CoreV1().Secrets(fmt.Sprintf("garden-%s", gardenerConfig.GardenerProjectName)).Get(context.TODO(), "gcp", metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return errors.Trace(err)
		}
	} else {
		err := ioutil.WriteFile("/etc/sa.json", gcpSecret.Data["serviceaccount.json"], os.ModePerm)
		if err != nil {
			return errors.Trace(err)
		}
	}

	serverConfig := config.Global().ServerConfig
	CMKArn := config.Global().InfraConfig.CMKArn
	enc, err := secretprovider.NewSecretEncryptor(CMKArn)
	if err != nil {
		return errors.Trace(err)
	}

	aesCrypt, err := encrypt.NewAESCrypt([]byte(serverConfig.SaltSecret))
	if err != nil {
		return errors.Trace(err)
	}
	session := tokenSession.NewSession(aesCrypt)
	mg := mailgun.NewMailgun(config.Domain, serverConfig.MailGunApiKey)
	sgCli := sendgrid.NewSendClient(serverConfig.SendgridApiKey)
	mailFacade := mailv2.NewMailProvider(mg, sgCli, config.SendgridTemplateName2id)
	zendeskConfig := config.Global().ZendeskConfig
	var zendeskClient *zendesk.Client
	if zendeskConfig.EnableZendesk {
		zendeskClient, err = zendesk.NewClient(zendeskConfig.ZendeskDomain, zendeskConfig.ZendeskEmail, zendeskConfig.ZendeskAPIToken)
		if err != nil {
			return errors.Annotate(err, "initialize the zendesk client failed")
		}
	}

	salesforceConfig := config.Global().SalesforceConfig
	var salesforceClient *salesforce.Client
	if salesforceConfig.EnableSalesforce {
		accessTokenCache := cache.New(20*time.Minute, 10*time.Minute)
		salesforceCfg := &salesforce.Config{
			SalesforceDomain:       config.Global().SalesforceConfig.SalesforceDomain,
			SalesforceClientID:     config.Global().SalesforceConfig.SalesforceClientID,
			SalesforceClientSecret: config.Global().SalesforceConfig.SalesforceClientSecret,
			SalesforceUsername:     config.Global().SalesforceConfig.SalesforceUsername,
			SalesforcePassword:     config.Global().SalesforceConfig.SalesforcePassword,
		}
		salesforceClient, err = salesforce.NewClient(accessTokenCache, salesforceCfg)
		if err != nil {
			return err
		}
	}

	// The following variables will be shared between multiple HTTP handlers.
	s.session = session
	s.mailClient = mailFacade
	s.zendeskClient = zendeskClient
	s.salesforceClient = salesforceClient

	auth0Config := config.Global().Auth0Config
	m, err := management.New(auth0Config.Auth0Domain, auth0Config.Auth0ManagementClientID, auth0Config.Auth0ManagementClientSecret)
	if err != nil {
		return errors.Trace(err)
	}

	k8sClient, err := initK8sClient()
	if err != nil {
		return errors.Annotate(err, "failed to init k8s client")
	}

	tlsConfig, err := aws.ProxyTlsConfig(config.Global().AwsConfig.PCAArn)
	if err != nil {
		return errors.Annotate(err, "failed to get proxy TLS config")
	}

	grafanaConfig := config.Global().GrafanaConfig
	grafanaClient := grafana.NewClient(grafanaConfig.Server, grafanaConfig.APIKey, grafana.DefaultHTTPClient)

	jwks, err := auth0.GetJWKS(s.centralAuth0.Domain)
	if err != nil {
		return errors.Trace(err)
	}
	managementPortalJWKs, err := auth0.GetJWKS(s.portalAuth0.Domain)
	if err != nil {
		return errors.Trace(err)
	}

	tokenValidator := auth0.NewRawTokenValidator().
		RegisterAuth(auth0.AuthValidatorIDCentral, s.centralAuth0, jwks).
		RegisterAuth(auth0.AuthValidatorIDManagement, s.portalAuth0, managementPortalJWKs)

	auth0TokenValidator := token.NewValidator(tokenValidator)

	cdcdao := &cdcdao.CDCDao{}
	clusterdao := &clusterdao.ClusterDao{}
	replicationdao := &replicationdao.ReplicationDao{}

	billingModel := billingmodel.NewModel(log.L(), mailFacade, salesforceClient)
	tenantModel := tenantmodel.NewModel(log.L(), session, mailFacade, m.User, zendeskClient, salesforceClient, billingModel, auth0TokenValidator)
	// TODO; combine proxy http client as one handler instead of pass tlsConfig to each model
	clusterModel := clustermodel.NewModel(log.L(), infraAPI, enc, mailFacade, salesforceClient, tenantModel, aesCrypt, tlsConfig, clusterdao)
	backupsModel := backupmodel.NewModel(log.L(), infraAPI, mailFacade, enc, tenantModel)
	importsModel := importmodel.NewModel(log.L(), infraAPI, enc)
	debugModel := debugmodel.NewModel(log.L(), grafanaClient, infraAPI)
	testTokenModel := testtoken.NewModel(log.L(), auth0Config.Auth0Domain, auth0Config.Auth0ManagementClientID, auth0Config.Auth0ManagementClientSecret, aesCrypt, []auth0.TokenExtractor{auth0.TokenRawFromHTTPHeader, auth0.TokenFromParameter("jwt"), auth0.TokenFromCookie}, k8sClient)
	topsqlModel := topsqlmodel.NewTopSQLModel()
	integrationModel := integrationmodel.NewModel(enc)
	replicationModel := replicationmodel.NewModel(log.L(), enc, tlsConfig, infraAPI, tenantModel, replicationdao, clusterdao, cdcdao)
	billingWorkerModel := billingworkermodel.NewModel(billingModel, clusterModel)
	cdcModel := cdcmodel.NewModel(log.L(), enc, infraAPI, cdcdao, clusterdao, replicationdao, billingModel)
	adminModel := adminmodel.NewModel(log.L(), mailFacade, grafanaClient, salesforceClient, infraAPI, tenantModel, billingModel, cdcModel)
	cs := opworker.New(&clusterModel, &backupsModel, &importsModel, &replicationModel, &cdcModel, &billingWorkerModel, infraAPI, mailFacade)

	tracing.InitTracer(log.L(), &tracing.TraceInfo{
		ServiceName: "DBaaS",
		Version:     version.GitHash,
		TraceToken:  os.Getenv("LS_ACCESS_TOKEN"),
	})

	s.tenantModel = &tenantModel

	go func() {
		if enableWorker() {
			log.Info("worker is enabled ", zap.String("role", os.Getenv(centralRole)))
			if err = cs.StartBackgroundWorkers(); err != nil {
				panic(err)
			}
		}
	}()

	//start metric
	registerMetricsHandler(&adminModel)

	// start a http server for healthz and version request
	if !enableApiService() {
		log.Info("API service is not enabled, start healthz", zap.String("role", os.Getenv(centralRole)))
		eng := gin.Default()
		eng.Any("/healthz", gin.WrapF(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) }))
		eng.Any("/version", gin.WrapF(versionHandler))
		panic(eng.Run("0.0.0.0:11000"))
	}
	log.Info("API service is enabled ", zap.String("role", os.Getenv(centralRole)))

	acceptBot := s.stage == config.StageE2E
	getClaim := service.GetUserClaim(acceptBot)

	authFunc := service.MakeAuthGRPC(tokenValidator, getClaim, testTokenModel)
	server := configureServer(authFunc, grpchttp.LogUnaryServerInterceptor(), ratelimit.UnaryServerInterceptor(ratelimit.GlobalLimiters))

	centralService := service.New()

	debugSessionManager := debugAuth.NewSessionManager(&config.Global().SessionConfig)
	debugPortalService := debugportal.NewService(centralService, debugModel, topsqlModel, &config.Global().DebugPortalConfig, debugSessionManager)
	central.RegisterCentralDebugPortalServiceServer(server, debugPortalService)
	central.RegisterBackupServiceServer(server, backup.NewService(centralService, backupsModel))
	central.RegisterConnectorServiceServer(server, connectorservice.NewService(centralService, cdcModel))
	central.RegisterImportServiceServer(server, importservice.NewService(centralService, importsModel))
	central.RegisterBillingServiceServer(server, billing.NewService(centralService, billingModel))
	central.RegisterTenantServiceServer(server, svctenant.NewService(centralService, tenantModel, adminModel))
	central.RegisterClusterServiceServer(server, cluster.NewService(centralService, clusterModel, &billingModel))
	central.RegisterAuthServiceServer(server, auth.NewService(centralService, m))
	central.RegisterTestTokenServiceServer(server, testokenservice.NewService(testTokenModel))
	central.RegisterIntegrationServiceServer(server, integration.NewService(centralService, integrationModel))

	grpcprometheus.EnableHandlingTimeHistogram()
	grpcprometheus.Register(server)
	addr := grpchttp.PortToAddress(config.Global().ServerConfig.GrpcPort)
	go func() {
		log.Info("serving gRPC", zap.String("address", addr))
		lis, err := net.Listen("tcp", addr)
		if err != nil {
			panic(err)
		}
		err = server.Serve(lis)
		if err != nil {
			panic(err)
		}
	}()

	gw, err := grpchttp.ServeHTTPGateway(log.L(), addr, grpchttp.ProdGatewayOptions, registerCentralHandler)
	if err != nil {
		return errors.Trace(err)
	}

	gatewayHttpAddr, ginEng := gw.HTTP()

	// tenantProxy proxies service from the cluster of target tenant.
	// The proxy interface may be called from Central UI or Management UI.
	mwb := auth0.NewHTTPMiddlewareBuilder(tokenValidator, getClaim).
		AcceptToken(auth0.TokenRawFromHTTPHeader).
		AcceptToken(auth0.TokenFromParameter("jwt")).
		AcceptToken(auth0.TokenFromCookie).
		AcceptValidator(auth0.AuthValidatorIDCentral).
		AcceptValidator(auth0.AuthValidatorIDManagement) // proxy interface accept both central and management access
	proxyConfig := config.Global().ProxyConfig
	proxyMetrics := newProxyMetricsHandler()
	proxyAuthenticator := proxy.NewAuth0M2MAuth(proxyConfig.Auth0Domain, proxyConfig.Auth0ClientID, proxyConfig.Auth0ClientSecret, proxyConfig.Auth0Audience)
	tenantProxy := proxy.NewReverseProxy(proxyAuthenticator, mesh.ProxyShootEndpoint(clusterModel, adminModel, debugModel))
	tenantProxy.SetTransport(&http.Transport{
		TLSClientConfig: tlsConfig,
	})

	middlewares := []middleware{
		mwSentry,
		testTokenModel.TryAuth,
		mwb.MWTryAuth,
		debugSessionManager.MWTryAuth,
		util.MWRequireAuth,
		util.MWRemoveAuthHeader,
		proxyMetrics,
		mwCORS,
	}
	middlewares = append(middlewares, rewriteTerminalPathMiddleware)

	mesh.HandleRouteStripGin(ginEng, "/proxy/", Middlewares(
		tenantProxy,
		middlewares...,
	))

	/*
		new api proxy
	*/
	proxyMiddlewares := []middleware{
		testTokenModel.TryAuth,
		mwb.MWTryAuth,
		util.MWRequireAuth,
		util.MWRemoveAuthHeader,
		mwCORS,
	}
	mesh.HandleRouteStripGin(ginEng, "/api_proxy/billing", Middlewares(
		http.HandlerFunc(centralapi.BillingProxy),
		proxyMiddlewares...,
	))

	centralapi.RegisterThirdPartyRedirectRoute(ginEng)

	ginEng.Any("/healthz", func(c *gin.Context) {
		c.Status(200)
	})
	ginEng.Any("/version", gin.WrapF(versionHandler))
	ginEng.Any("/invitation-verification", gin.WrapF(s.invitationVerificationHandler))
	ginEng.Any("/registration-verification", gin.WrapF(s.registrationVerificationHandler(cs)))

	// Serve Gateway
	log.Info("serving HTTP gRPC-Gateway", zap.String("address", gatewayHttpAddr))
	log.Info("serving HTTP OpenAPI Documentation", zap.String("url", fmt.Sprintf("%s/openapi-ui/", gatewayHttpAddr)))

	panic(fmt.Sprintf("grpc HTTP gateway died %+v", ginEng.Run(gatewayHttpAddr)))
}

type middleware func(http.Handler) http.Handler

func Middlewares(h http.Handler, m ...middleware) http.Handler {
	if len(m) < 1 {
		return h
	}
	wrapped := h
	// loop in reverse to preserve middleware order
	for i := len(m) - 1; i >= 0; i-- {
		wrapped = m[i](wrapped)
	}
	return wrapped
}

// rewriteTerminalPath rewrite the old version path:
// /terminal/19/root/
// as the new version proxy support:
// /cluster/19/terminal/root/
func rewriteTerminalPath(path string) string {
	if !strings.HasPrefix(path, "/terminal/") && !strings.HasPrefix(path, "terminal/") {
		return path
	}

	originPath := path

	appendLeadingSlash := false
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
		appendLeadingSlash = true
	}

	ar := strings.Split(path, "/")
	if len(ar) < 4 {
		return originPath
	}

	clusterID := ar[2]
	user := ar[3]

	ret := fmt.Sprintf("/cluster/%s/terminal/%s", clusterID, user)
	if len(ar[4:]) > 0 {
		ret += "/" + strings.Join(ar[4:], "/")
	}

	if appendLeadingSlash {
		ret = ret[1:]
	}

	return ret
}

func rewriteTerminalPathMiddleware(handler http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		origin := r.URL.Path
		r.URL.Path = rewriteTerminalPath(r.URL.Path)

		if r.URL.Path != origin {
			log.Debug("rewrite terminal path", zap.String("old", origin), zap.String("new", r.URL.Path))
		}

		handler.ServeHTTP(w, r)
	})
}

// Auth middleware do not authenticate OPTIONS request, which violates the requirements of tenant proxy, so
// we handle CORS of options in prior
func mwCORS(handler http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodOptions {
			w.Header().Set("Access-Control-Allow-Headers", "accept, content-type, authorization")
			w.Header().Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE")
			w.Header().Set("Access-Control-Allow-Origin", "*")
			w.WriteHeader(200)
			return
		}
		w.Header().Set("Access-Control-Allow-Origin", "*")
		handler.ServeHTTP(w, r)
	})
}

func mwSentry(next http.Handler) http.Handler {
	sentryHandler := sentryhttp.New(sentryhttp.Options{})
	return sentryHandler.HandleFunc(func(writer http.ResponseWriter, request *http.Request) {
		next.ServeHTTP(writer, request)
	})
}

type versionInfo struct {
	Hash             string `json:"hash"`
	CentralDBVersion int    `json:"central_db_version"`
	TenantDBVersion  int    `json:"tenant_db_version"`
}

// versionHandler outputs the current central server version information.
func versionHandler(w http.ResponseWriter, _ *http.Request) {
	data, _ := json.Marshal(versionInfo{
		Hash:             version.GitHash,
		CentralDBVersion: migrate.GetCentralDBVersion(),
	})
	_, _ = w.Write(data)
}

func (s *server) invitationVerificationHandler(w http.ResponseWriter, r *http.Request) {
	log.Info("api call", zap.String("method", r.Method), zap.String("url", r.URL.Path))
	baseUrl := fmt.Sprintf("https://%s/invite_callback?", config.Global().ServerConfig.FrontendDNS)
	vars := r.URL.Query()
	params := url.Values{}
	if len(vars["token"]) != 1 {
		params.Add("status", FailedStatus)
		params.Add(ErrorType, SystemError)
		http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
		return
	}

	session := s.session
	token := vars["token"][0]
	userInviteToken, err := session.ValidateUserInviteToken(token, true)
	if err != nil {
		if session.IsActivatedError(err) {
			baseUrl = fmt.Sprintf("https://%s", config.Global().ServerConfig.FrontendDNS)
			http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
			return
		}
		params.Add("status", FailedStatus)
		var errType string
		if session.IsTokenExpiredError(err) {
			errType = TokenExpiredError
		} else {
			errType = SystemError
		}
		params.Add(ErrorType, errType)
		log.Error("failed to validate invitation token", zap.String("feature", "invitation"), zap.String("errorType", errType), zap.Error(err))
		http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
		return
	}

	email := userInviteToken.Email
	tenantID := userInviteToken.TenantID
	log.Info("invitation-verification: validate invitation token", zap.String("feature", "invitation"), zap.String("email", email), zap.Uint64("tenantID", tenantID.Uint64()))
	params.Add("status", "success")
	params.Add("org_id", strconv.FormatInt(int64(tenantID), 10))
	params.Add("email", email)
	params.Add("token", token)
	http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
}

func (s *server) registrationVerificationHandler(cs *opworker.Manager) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		session := s.session
		logger := pkglog.WithContextL(ctx)
		logger.Info("api call", zap.String("method", r.Method), zap.String("url", r.URL.Path))
		baseUrl := fmt.Sprintf("https://%s/signup?", config.Global().ServerConfig.FrontendDNS)
		vars := r.URL.Query()
		params := url.Values{}
		if len(vars["token"]) != 1 {
			params.Add("status", FailedStatus)
			params.Add(ErrorType, SystemError)
			http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
			return
		}

		token := vars["token"][0]
		user := &dbtenant.User{}
		email, err := s.tenantModel.User(ctx, user).RegistrationVerification(ctx, token)
		if err != nil {
			var errType string
			if session.IsTokenExpiredError(err) {
				errType = TokenExpiredError
			} else {
				errType = SystemError
			}
			log.Error("failed to registration verification", zap.String("feature", "signup"), zap.String("errorType", errType), zap.Error(err))
			params.Add("status", FailedStatus)
			params.Add(ErrorType, errType)
			http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
		}

		params.Add("status", "success")
		if email != nil {
			params.Add("email", *email)
		}
		http.Redirect(w, r, baseUrl+params.Encode(), http.StatusFound)
	}
}

func registerCentralHandler(ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn) error {
	if err := central.RegisterClusterServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterBackupServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterConnectorServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterImportServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterTenantServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterBillingServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterCentralDebugPortalServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterTestTokenServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := central.RegisterIntegrationServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	return central.RegisterAuthServiceHandler(ctx, mux, conn)
}

func newProxyMetricsHandler() func(handler http.Handler) http.Handler {
	return func(handler http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			tBegin := time.Now()
			rl := &grpchttp.ResponseLogger{W: w, Status: 200}
			handler.ServeHTTP(rl, r)
			statusCodeStr := fmt.Sprintf("%d", rl.Status)
			duration := float64(time.Since(tBegin)) / float64(time.Millisecond)
			u, _ := authuser.GetUserFromContext(r.Context())
			if u != nil {
				metrics.ProxyRequestDuration.WithLabelValues(fmt.Sprintf("%d", u.TenantID), statusCodeStr).Observe(duration)
			}
			if rl.Status >= 500 {
				url := r.URL.String()
				log.Error("call central API failed", zap.Int("status code", rl.Status), zap.String("url", url))
			}
		})
	}
}

func panicHandler(ctx context.Context, p interface{}) error {
	logger := pkglog.WithContextL(ctx)
	err, ok := p.(error)
	if ok && errors.HasStack(err) {
		logger.Sugar().Warnf("panic, err: %s, stack: %s", err, errors.GetStackTracer(err).StackTrace())
	} else {
		logger.Sugar().Warnf("panic, err: %s, stack: %s", p, debug.Stack())
	}

	// I don't know why the usual stack trace is not working
	// So putting debug.Stack() into the message for now
	if !ok {
		err = fmt.Errorf("%+v\n%s", p, debug.Stack())
		return errcode.NewInternalErr(err)
	}

	// TODO(chenfei): Should replaced by grpc_sentry.UnaryServerInterceptor
	// report panic to sentry
	sentry.CurrentHub().Recover(err)
	return errcode.NewInternalErr(err)
}

// configureServer is an opinionated common configuration for a grpc Server.
func configureServer(authFunc func(context.Context) (context.Context, error), logInterceptor grpc.UnaryServerInterceptor, ratelimitInterceptor grpc.UnaryServerInterceptor) *grpc.Server {
	opts := []grpc_recovery.Option{
		grpc_recovery.WithRecoveryHandlerContext(panicHandler),
	}
	/* TODO: run the closer
	   tracer, _ := tracing.InitJaeger(service)
	*/

	return grpc.NewServer(
		grpc.Creds(credentials.NewServerTLSFromCert(&insecure.Cert)),

		grpc.UnaryInterceptor(
			grpc_middleware.ChainUnaryServer(
				grpcbizerr.UnaryServerInterceptor(func(resp interface{}, err bizerr.BizError) (resp_ interface{}, err_ error) {
					typed := getBaseRespFromResp(resp)
					if typed == nil {
						return resp, err
					}
					typed.ErrCode = int64(err.ErrCode())
					typed.ErrMsg = err.ErrMsg()
					return resp, nil
				}),
				logid.UnaryServerInterceptor(),
				otelgrpc.UnaryServerInterceptor(),
				grpc_sentry.UnaryServerInterceptor(),
				grpc_errcode.UnaryServerInterceptor(),
				grpc_validator.UnaryServerInterceptor(),
				grpc_auth.UnaryServerInterceptor(authFunc),
				logInterceptor,
				ratelimitInterceptor,
				grpcprometheus.UnaryServerInterceptor,
				grpc_recovery.UnaryServerInterceptor(opts...),
			),
		),
		grpc.StreamInterceptor(
			grpc_middleware.ChainStreamServer(
				logid.StreamServerInterceptor(),
				grpc_sentry.StreamServerInterceptor(),
				grpc_errcode.StreamServerInterceptor(),
				grpc_validator.StreamServerInterceptor(),
				grpc_auth.StreamServerInterceptor(authFunc),
				grpcprometheus.StreamServerInterceptor,
				// otgrpc.OpenTracingStreamServerInterceptor(tracer),
				grpc_recovery.StreamServerInterceptor(opts...),
			),
		),
	)
}

func registerMetricsHandler(adminModel *adminmodel.Model) {
	metrics.RegisterMetrics()
	clustermetrics.RegisterMetrics()
	importmetrics.RegisterMetrics()
	tenantmetrics.RegisterMetrics(adminModel)
	backupmetrics.RegisterMetrics()
	restoremetrics.RegisterMetrics()
	// Start metrics server.
	if config.Global().ServerConfig.MetricsPort > 0 {
		mux := http.NewServeMux()
		mux.Handle("/metrics", promhttp.Handler())
		go func() {
			err := http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", config.Global().ServerConfig.MetricsPort), mux)
			if err != nil {
				panic(fmt.Sprintf("metrics server died %+v", err))
			}
		}()
	}
}

func getRoleFromEnv() string {
	role := os.Getenv(centralRole)
	if role == "" {
		role = allRole
	}
	return role
}

func enableApiService() bool {
	role := getRoleFromEnv()
	return role == allRole || role == apiService
}

func enableWorker() bool {
	role := getRoleFromEnv()
	return role == allRole || role == worker
}

// TODO(chenfei): merge with idl/pbgen/util/helper.go:GetBaseRespFromResp
func getBaseRespFromResp(resp interface{}) *central.BaseResp {
	if resp == nil {
		return nil
	}

	baseResp := reflect.ValueOf(resp).Elem().FieldByName("BaseResp")
	if !baseResp.IsValid() { // don't have BaseReq field
		return nil
	}
	if baseResp.IsNil() {
		baseResp.Set(reflect.New(baseResp.Type().Elem()))
	}

	typed, ok := baseResp.Interface().(*central.BaseResp)
	if !ok {
		return nil
	}
	return typed
}