package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"os"
	"reflect"
	"runtime/debug"

	"github.com/getsentry/sentry-go"
	"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/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	configServiceV1 "github.com/tidbcloud/config-service/api/v1alpha1"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/central"
	"github.com/tidbcloud/internal-portal/core-portal/admin/api"
	"github.com/tidbcloud/internal-portal/core-portal/admin/models"
	"github.com/tidbcloud/internal-portal/core-portal/auth"
	"github.com/tidbcloud/internal-portal/core-portal/central/config"
	clusterdao "github.com/tidbcloud/internal-portal/core-portal/central/db/dbutil/cluster"
	replicationdao "github.com/tidbcloud/internal-portal/core-portal/central/db/dbutil/replication"
	grpchttp "github.com/tidbcloud/internal-portal/core-portal/central/grpc/http"
	clustermodel "github.com/tidbcloud/internal-portal/core-portal/central/model/cluster"
	"github.com/tidbcloud/internal-portal/core-portal/central/model/testtoken"
	grpc_errcode "github.com/tidbcloud/internal-portal/core-portal/central/pkg/grpc/middleware/errcode"
	"github.com/tidbcloud/internal-portal/core-portal/central/pkg/grpc/middleware/ratelimit"
	grpc_sentry "github.com/tidbcloud/internal-portal/core-portal/central/pkg/grpc/middleware/sentry"
	"github.com/tidbcloud/internal-portal/core-portal/central/pkg/version"
	"github.com/tidbcloud/internal-portal/core-portal/central/service"
	"github.com/tidbcloud/internal-portal/db/migrate"
	"github.com/tidbcloud/internal-portal/idl/pbgen/proto/useralert"
	"github.com/tidbcloud/internal-portal/temporary/centralsvc"
	"github.com/tidbcloud/internal-portal/temporary/rbac"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"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"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"github.com/tidbcloud/pkgv2/zapext/zapsentry"
	"github.com/tidbcloud/tiinf/monitor/titracing"
	"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"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
)

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
}

func initCentralServer() error {
	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),
	}

	return s.serve()
}

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

	return s.startup()
}

// func initK8sClient() (client.Client, error) {
// 	cfg, err := getK8sCfg()
// 	if err != nil {
// 		return nil, errors.Annotate(err, "initialize the base kubeclient 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 getK8sCfg() (*rest.Config, error) {
	// 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 cfg, errors.Annotate(err, "initialize the kubernetes config failed")
	}
	log.Info("initialize the kubernetes config successful")

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

	return cfg, nil
}

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

	serverConfig := config.Global().ServerConfig
	aesCrypt, err := encrypt.NewAESCrypt([]byte(serverConfig.SaltSecret))
	if err != nil {
		return errors.Trace(err)
	}

	auth0Config := config.Global().Auth0Config

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

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

	// TODO; combine proxy http client as one handler instead of pass tlsConfig to each model
	clusterModel := clustermodel.NewModel(log.L(), infraAPI, encryptor, tlsConfig, clusterdao, replicationdao)
	testTokenModel := testtoken.NewModel(log.L(), auth0Config.Auth0Domain, auth0Config.Auth0ManagementClientID, auth0Config.Auth0ManagementClientSecret, aesCrypt, infraAPI.GetK8sCli())
	log.Info("initialize Cls model successful")

	if _, err = titracing.InitJaeger("internal-portal-central"); err != nil {
		log.Error("init tiinf jaeger provider failed", zap.Error(err))
	}

	authFunc := auth.NewAuthFunc(tokenValidator, &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())
	// }

	useralert.RegisterUserAlertServiceServer(server, centralsvc.NewService(centralService, clusterModel))

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

	log.Info("registering Central grpchttp-Handler")
	gw, err := grpchttp.ServeHTTPGateway(log.L(), addr, grpchttp.ProdGatewayOptions, registerUserAlertServiceHandler)
	if err != nil {
		return errors.Trace(err)
	}
	log.Info("register Central grpchttp-Handler successful")

	gatewayHttpAddr, ginEng := gw.HTTP()

	ginEng.GET("/healthz", func(c *gin.Context) {
		c.Status(200)
	})
	ginEng.GET("/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
}

type versionInfo struct {
	Hash             string `json:"hash"`
	CentralDBVersion int    `json:"central_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.GetInternalDBVersion(),
	})
	_, _ = w.Write(data)
}

func registerUserAlertServiceHandler(ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn) error {
	return useralert.RegisterUserAlertServiceHandler(ctx, mux, conn)
}

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),
				rbac.UnaryServerInterceptor(),
				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),
				rbac.StreamServerInterceptor(),
				grpcprometheus.StreamServerInterceptor,
				// otgrpc.OpenTracingStreamServerInterceptor(tracer),
				grpc_recovery.StreamServerInterceptor(opts...),
			),
		),
	)
}

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

// 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:                 config.Global().Auth0Config.ManagementPortalAuth0Audience,
		Secret:                   []byte(config.Global().Auth0Config.ManagementPortalAuth0Secret),
		Domain:                   config.Global().Auth0Config.ManagementPortalAuth0Domain,
		ClientCredentialAudience: config.Global().Auth0Config.ManagementPortalAuth0ClientAudience,
	}
	if err := api.InitTokenValidator(portalAuth0); err != nil {
		return err
	}

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

	api.InitAdminRouter(ginEng)
	return nil
}