package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"reflect"
	"regexp"
	"runtime/debug"
	"strings"
	"time"

	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/getsentry/sentry-go"
	sentryhttp "github.com/getsentry/sentry-go/http"
	"github.com/gin-gonic/gin"
	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/v2/runtime"
	"github.com/mailgun/mailgun-go/v4"
	"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"
	api2 "github.com/tidbcloud/core-portal/internal/admin/api"
	"github.com/tidbcloud/core-portal/internal/admin/models"
	config2 "github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-portal/internal/central/db"
	clusterdao "github.com/tidbcloud/core-portal/internal/central/db/dbutil/cluster"
	replicationdao "github.com/tidbcloud/core-portal/internal/central/db/dbutil/replication"
	grpchttp "github.com/tidbcloud/core-portal/internal/central/grpc/http"
	api "github.com/tidbcloud/core-portal/internal/central/infra-api"
	"github.com/tidbcloud/core-portal/internal/central/mesh"
	metrics2 "github.com/tidbcloud/core-portal/internal/central/metrics"
	clustermetrics "github.com/tidbcloud/core-portal/internal/central/metrics/cluster"
	billingmodel "github.com/tidbcloud/core-portal/internal/central/model/billing"
	clustermodel "github.com/tidbcloud/core-portal/internal/central/model/cluster"
	"github.com/tidbcloud/core-portal/internal/central/model/testtoken"
	"github.com/tidbcloud/core-portal/internal/central/pkg/auth/authuser"
	debugAuth "github.com/tidbcloud/core-portal/internal/central/pkg/auth/debug"
	"github.com/tidbcloud/core-portal/internal/central/pkg/grpc/middleware/errcode"
	"github.com/tidbcloud/core-portal/internal/central/pkg/grpc/middleware/ratelimit"
	"github.com/tidbcloud/core-portal/internal/central/pkg/grpc/middleware/sentry"
	"github.com/tidbcloud/core-portal/internal/central/pkg/http/proxy"
	"github.com/tidbcloud/core-portal/internal/central/pkg/version"
	"github.com/tidbcloud/core-portal/internal/central/service"
	"github.com/tidbcloud/core-portal/internal/central/service/cluster"
	coresvc "github.com/tidbcloud/core-portal/internal/central/service/core"
	testokenservice "github.com/tidbcloud/core-portal/internal/central/service/testtoken"
	"github.com/tidbcloud/core-portal/internal/central/worker"
	openapiservice "github.com/tidbcloud/core-portal/internal/openapi/service"
	openapiCluster "github.com/tidbcloud/core-portal/internal/openapi/service/cluster"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/central"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/idl/pbgen/proto/openapi/cm"
	"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/secret/encrypt"
	secretprovider "github.com/tidbcloud/pkgv2/secret/provider"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/pkgv2/zapext/zapsentry"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"google.golang.org/grpc"
	"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
	defaultTimeout = 5 * time.Second

	defaultConfigPath = "/etc/core-portal/config.yaml"

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

	regexpDebugAPI          = `^cluster/\d+/pd/dashboard/api/debug_api/`
	regexpQueryEditor       = `^cluster/\d+/pd/dashboard/api/query_editor/`
	regexpEditConfiguration = `^cluster/\d+/pd/dashboard/api/configuration/edit`
)

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        config2.Stage
}

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

	err := config2.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: config2.Global().Auth0Config.Auth0Audience,
			Secret:   []byte(config2.Global().Auth0Config.Auth0Secret),
			Domain:   config2.Global().Auth0Config.Auth0Domain,
		},
		portalAuth0: auth0.Conf{
			Audience:                 config2.Global().Auth0Config.ManagementPortalAuth0Audience,
			Secret:                   []byte(config2.Global().Auth0Config.ManagementPortalAuth0Secret),
			Domain:                   config2.Global().Auth0Config.ManagementPortalAuth0Domain,
			ClientCredentialAudience: config2.Global().Auth0Config.ManagementPortalAuth0ClientAudience,
		},
		stage: config2.Stage(config2.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(config2.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())
		}
	}()

	return s.startup()
}

func setupLogger(opts ...zap.Option) error {
	cfg := &log.Config{
		Level: config2.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 := config2.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 := config2.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 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.RegisterTenantServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	return central.RegisterTestTokenServiceHandler(ctx, mux, conn)
}

func registerOpenapiHandler(ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn) error {
	return cm.RegisterClusterHandler(ctx, mux, conn)
}

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

	// gardener cluster k8s client
	var cfg *rest.Config
	var err error
	kubeConfig := config2.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")
	}
	infraAPI := api.InitInfraAPI(baseKubeCli)

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

	serverConfig := config2.Global().ServerConfig
	CMKArn := config2.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)
	}
	mg := mailgun.NewMailgun(config2.Domain, serverConfig.MailGunApiKey)
	sgCli := sendgrid.NewSendClient(serverConfig.SendgridApiKey)
	mailFacade := mailv2.NewMailProvider(mg, sgCli, config2.SendgridTemplateName2id)

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

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

	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)

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

	billingModel := billingmodel.NewModel(log.L())
	// TODO; combine proxy http client as one handler instead of pass tlsConfig to each model
	clusterModel := clustermodel.NewModel(log.L(), infraAPI, enc, mailFacade, aesCrypt, tlsConfig, clusterdao, replicationdao)
	testTokenModel := testtoken.NewModel(log.L(), auth0Config.Auth0Domain, auth0Config.Auth0ManagementClientID, auth0Config.Auth0ManagementClientSecret, aesCrypt, []auth0.TokenExtractor{auth0.TokenRawFromHTTPHeader, auth0.TokenFromParameter("jwt"), auth0.TokenFromCookie}, k8sClient)
	cs := opworker.New(&clusterModel, infraAPI, mailFacade)

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

	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 and pprof
	registerMetricsPprofHandler()

	// 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 == config2.StageE2E
	getClaim := service.GetUserClaim(acceptBot)

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

	centralService := service.New()

	awsConfig, err := awsconfig.LoadDefaultConfig(context.TODO())
	if err != nil {
		panic("AWS configuration error, " + err.Error())
	}

	debugSessionManager := debugAuth.NewSessionManager(&config2.Global().SessionConfig)
	central.RegisterTenantServiceServer(server, cluster.NewTenantService(centralService, clusterModel)) // still needed for `GetProjectTrafficFilter` and `UpdateProjectTrafficFilter`
	central.RegisterClusterServiceServer(server, cluster.NewService(centralService, clusterModel, &billingModel, awsConfig))
	central.RegisterTestTokenServiceServer(server, testokenservice.NewService(testTokenModel)) // put in core-portal temporarily

	// APIs for billing service, no need to register for HTTP API
	core.RegisterCoreServiceServer(server, coresvc.NewService(log.L(), infraAPI, clusterModel))

	// Register Open API Server
	openAPIService := openapiservice.New()
	cm.RegisterClusterServer(server, openapiCluster.NewService(openAPIService, infraAPI, clusterModel, &billingModel))

	grpcprometheus.EnableHandlingTimeHistogram()
	grpcprometheus.Register(server)
	addr := grpchttp.PortToAddress(config2.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, registerOpenapiHandler)
	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 := config2.Global().ProxyConfig
	proxyMetrics := newProxyMetricsHandler()
	proxyAuthenticator := proxy.NewAuth0M2MAuth(proxyConfig.Auth0Domain, proxyConfig.Auth0ClientID, proxyConfig.Auth0ClientSecret, proxyConfig.Auth0Audience)
	tenantProxy := proxy.NewReverseProxy(proxyAuthenticator, mesh.ProxyShootEndpoint(clusterModel))
	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)

	// For security reasons, we need to temporarily disable access to the
	// `cluster/.../pd/dashboard/api/debug_api/` series of APIs.
	// This is achieved by mounting additional middleware for `/proxy/`.
	reDebugAPI := regexp.MustCompile(regexpDebugAPI)
	middlewares = append(middlewares, mwForbidMatchedURLIfNotDebugSession(func(path string) bool {
		return reDebugAPI.MatchString(path)
	}))

	// In order to enable the experimental features of TiDB Dashboard on cloud,
	// we must disable the following APIs to prevent visitors from executing
	// SQL or modifying the cluster configuration.
	reQueryEditor := regexp.MustCompile(regexpQueryEditor)
	reEditConfiguration := regexp.MustCompile(regexpEditConfiguration)
	middlewares = append(middlewares, mwForbidMatchedURL(func(path string) bool {
		return reQueryEditor.MatchString(path) || reEditConfiguration.MatchString(path)
	}))

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

	ginEng.Any("/healthz", func(c *gin.Context) {
		c.Status(200)
	})
	ginEng.Any("/version", gin.WrapF(versionHandler))

	// 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)))

	// Serve Admin API
	if err := setupAdminAPI(ginEng); err != nil {
		return errors.Trace(err)
	}

	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)
	})
}

// mwForbidMatchedURL returns a middleware that interrupts the
// request and returns 403 when url matched.
func mwForbidMatchedURL(matched func(string) bool) middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if matched(r.URL.Path) {
				http.Error(w, http.StatusText(http.StatusForbidden), http.StatusForbidden)
				return
			}
			next.ServeHTTP(w, r)
		})
	}
}

// mwForbidMatchedURLIfNotDebugSession returns a middleware that
// interrupts the request and returns 403 when url matched and
// the request does not come from the debug portal.
func mwForbidMatchedURLIfNotDebugSession(matched func(string) bool) middleware {
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if matched(r.URL.Path) && !debugAuth.IsDebugSessionAuth(r.Context()) {
				http.Error(w, http.StatusText(http.StatusForbidden), http.StatusForbidden)
				return
			}
			next.ServeHTTP(w, r)
		})
	}
}

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,
	})
	_, _ = w.Write(data)
}

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 {
				metrics2.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(
		// NO TSL for gRPC methods now, as we don't want to use `github.com/gogo/grpc-example/insecure`
		// 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 registerMetricsPprofHandler() {
	metrics2.RegisterMetrics()
	clustermetrics.RegisterMetrics()
	// Start metrics server.
	if config2.Global().ServerConfig.MetricsPort > 0 {
		mux := http.NewServeMux()
		mux.Handle("/metrics", promhttp.Handler())
		setupPprofHandler(mux)
		go func() {
			err := http.ListenAndServe(fmt.Sprintf("0.0.0.0:%d", config2.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
}

// copied from admin/cmd/server/main.go
func setupAdminAPI(ginEng *gin.Engine) error {
	portalAuth0 := auth0.Conf{
		Audience:                 config2.Global().Auth0Config.ManagementPortalAuth0Audience,
		Secret:                   []byte(config2.Global().Auth0Config.ManagementPortalAuth0Secret),
		Domain:                   config2.Global().Auth0Config.ManagementPortalAuth0Domain,
		ClientCredentialAudience: config2.Global().Auth0Config.ManagementPortalAuth0ClientAudience,
	}
	if err := api2.InitTokenValidator(portalAuth0); err != nil {
		return err
	}

	if err := models.InitAdminModel(); err != nil {
		return err
	}

	api2.InitAdminRouter(ginEng)
	return nil
}

func setupPprofHandler(mux *http.ServeMux) {
	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)
}