package main

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

	"github.com/getsentry/sentry-go"
	"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/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	configServiceV1 "github.com/tidbcloud/config-service/api/v1alpha1"
	"github.com/tidbcloud/metrics-server/central/config"
	"github.com/tidbcloud/metrics-server/central/dbinit"
	grpchttp "github.com/tidbcloud/metrics-server/central/grpc/http"
	"github.com/tidbcloud/metrics-server/central/metrics"
	"github.com/tidbcloud/metrics-server/central/migrate"
	"github.com/tidbcloud/metrics-server/central/model/account/centralmod"
	"github.com/tidbcloud/metrics-server/central/model/account/centralsvc"
	clustermodel "github.com/tidbcloud/metrics-server/central/model/cluster"
	integrationmodel "github.com/tidbcloud/metrics-server/central/model/integration"
	"github.com/tidbcloud/metrics-server/central/service"
	"github.com/tidbcloud/metrics-server/central/service/cluster"
	"github.com/tidbcloud/metrics-server/central/service/integration"
	"github.com/tidbcloud/metrics-server/central/service/useralert"
	"github.com/tidbcloud/metrics-server/idl/pbgen/proto/central"
	"github.com/tidbcloud/metrics-server/pkg/api"
	grpc_errcode "github.com/tidbcloud/metrics-server/pkg/grpc/middleware/errcode"
	"github.com/tidbcloud/metrics-server/pkg/grpc/middleware/ratelimit"
	grpc_sentry "github.com/tidbcloud/metrics-server/pkg/grpc/middleware/sentry"
	"github.com/tidbcloud/metrics-server/pkg/version"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	"github.com/tidbcloud/pkgv2/bizerr"
	grpcbizerr "github.com/tidbcloud/pkgv2/grpc/middleware/bizerr"
	pkglog "github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	secretprovider "github.com/tidbcloud/pkgv2/secret/provider"
	"github.com/tidbcloud/pkgv2/zapext/zapsentry"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/monitor/titracing"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"github.com/tidbcloud/tiinf/sdk/rbac/helperimpl"
	"github.com/tidbcloud/tiinf/sdk/rbac/useadapter"
	"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"
	"k8s.io/apimachinery/pkg/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
)

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

	defaultConfigPath = "/etc/metrics/config.yaml"
)

var (
	scheme = runtime.NewScheme()
)

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

// server represents the central server instance
type server struct {
	centralAuth0 auth0.Conf
	stage        config.Stage
}

func main() {
	var configPath string
	flag.StringVar(&configPath, "config-path", defaultConfigPath, "metrics 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,
		},
		stage: config.Stage(config.Global().ServerConfig.Stage),
	}
	gwChan := make(chan error, 1)

	go func() {
		gwChan <- s.serve()
	}()

	// serve user alert webhook on 0.0.0.0:8081 by default.
	r := api.InitRouter()
	alertsChan := make(chan error, 1)
	go func() {
		alertsChan <- r.Run(config.Global().ServerConfig.AlertsListenAddr)
	}()

	select {
	case err := <-gwChan:
		if err != nil {
			// TODO: unexpected error like this, should we throw a alert/page, just log seems not enough ?
			log.Fatal("failed to start grpc gateway server", zap.Error(err))
		}
		return
	case err := <-alertsChan:
		if err != nil {
			log.Fatal("failed to start alerts webhook server", zap.Error(err))
		}
		return
	}
}

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 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 (s *server) startup() error {
	log.Info("starting central service",
		zap.String("gitHash", version.GitHash),
		zap.String("logLevel", config.Global().ServerConfig.LogLevel))

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

	ctx := context.Background()
	err = auditlog.InitAuditLogSDK(ctx, time.Second)
	if err != nil {
		tilog.Panic(ctx, "failed to init auditlog sdk", zap.Error(err))
	}

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

	testTokenModel := centralmod.NewCoveredTestTokenModel(log.L(), aesCrypt)

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

	//start metric
	registerMetricsHandler()

	log.Info("API service is enabled ")

	acceptBot := s.stage == config.StageE2E
	getClaim := centralsvc.GetUserClaim(acceptBot)
	// TODO: this should probably be provisioned as a secret to handle Auth0 downtime
	jwks, err := auth0.GetJWKS(s.centralAuth0.Domain)
	if err != nil {
		return errors.Trace(err)
	}

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

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

	centralService := service.New()
	clusterModel := clustermodel.NewModelV2(log.L())

	central.RegisterClusterServiceServer(server, cluster.NewService(centralService, clusterModel))
	central.RegisterIntegrationServiceServer(server, integration.NewService(centralService, integrationModel))
	central.RegisterUserAlertServiceServer(server, useralert.NewService(centralService, clusterModel))

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

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

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 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.RegisterUserAlertServiceHandler(ctx, mux, conn); err != nil {
		return err
	}

	return central.RegisterIntegrationServiceHandler(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(serviceName string, authFunc func(context.Context) (context.Context, error), logInterceptor grpc.UnaryServerInterceptor, ratelimitInterceptor grpc.UnaryServerInterceptor) *grpc.Server {
	opts := []grpc_recovery.Option{
		grpc_recovery.WithRecoveryHandlerContext(panicHandler),
	}

	// init tracing
	_, err := titracing.InitJaeger(serviceName)
	if err != nil {
		log.Fatal(err.Error())
	}

	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),
				// add extract user interceptor
				grpc_auth.UnaryServerInterceptor(grpchttp.ExtractIdentityAuthFunc),
				// add permission check interceptor
				useadapter.GrpcGatewayPermissionCheckUnaryServerInterceptor(
					helperimpl.NewContextStoreGetIdentityFunc(grpchttp.CtxIdentityKey),
					grpchttp.PathPattern2ResourceMap,
					grpchttp.PathWhitelist...,
				),
				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),
				// add extract user interceptor
				grpc_auth.StreamServerInterceptor(grpchttp.ExtractIdentityAuthFunc),
				// add permission check interceptor
				useadapter.GrpcGatewayPermissionCheckStreamServerInterceptor(
					helperimpl.NewContextStoreGetIdentityFunc(grpchttp.CtxIdentityKey),
					grpchttp.PathPattern2ResourceMap,
					grpchttp.PathWhitelist...,
				),
				grpcprometheus.StreamServerInterceptor,
				// otgrpc.OpenTracingStreamServerInterceptor(tracer),
				grpc_recovery.StreamServerInterceptor(opts...),
			),
		),
	)
}

func registerMetricsHandler() {
	metrics.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))
			}
		}()
	}
}

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