package app

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"runtime/debug"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/gogo/grpc-example/insecure"
	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/validator"
	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/tidbcloud/pkgv2/auth/auth0"
	pkglog "github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/serverless"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/controller"
	grpc_http "github.com/tidbcloud/serverless-service/internal/grpc/http"
	"github.com/tidbcloud/serverless-service/internal/grpc/middlewares/allowlist"
	"github.com/tidbcloud/serverless-service/internal/grpc/middlewares/auth"
	grpc_baseresp "github.com/tidbcloud/serverless-service/internal/grpc/middlewares/baseresp"
	grpc_errcode "github.com/tidbcloud/serverless-service/internal/grpc/middlewares/errcode"
	grpc_log "github.com/tidbcloud/serverless-service/internal/grpc/middlewares/log"
	grpc_metric "github.com/tidbcloud/serverless-service/internal/grpc/middlewares/metric"
	grpc_ratelimit "github.com/tidbcloud/serverless-service/internal/grpc/middlewares/ratelimit"
	"github.com/tidbcloud/serverless-service/internal/grpc/middlewares/rbac"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"github.com/tidbcloud/serverless-service/pkg/version"
	"github.com/tidbcloud/tiinf/monitor/titracing"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

type server struct {
	portalAuth0 auth0.Conf

	rpcAddr   string
	rpcServer *grpc.Server

	httpAddr   string
	httpServer *gin.Engine
}

func NewServer() *server { //nolint:revive
	return &server{
		portalAuth0: auth0.Conf{
			Audience:                 config.Global().Auth0Config.Auth0Audience,
			Secret:                   []byte(config.Global().Auth0Config.Auth0Secret),
			Domain:                   config.Global().Auth0Config.Auth0Domain,
			ClientCredentialAudience: config.Global().Auth0Config.ManagementPortalAuth0ClientAudience,
		},
	}
}

func (s *server) setupLogger() error {
	// Wrap logger core to redact sensitive fields like passwords/tokens in all logs.
	redactOpt := utils.WrapWithRedactor()
	opts := []zap.Option{redactOpt}
	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 (s *server) Setup() error {
	if err := s.setupLogger(); err != nil {
		return errors.Annotate(err, "initialize logger failed")
	}

	serviceName := "serverless-service-" + config.GetWorkMode().String()
	if _, err := titracing.InitJaeger(serviceName); err != nil {
		return errors.Annotate(err, "initialize jaeger failed")
	}

	if config.Global().ServerConfig.EnableAuditLog {
		if err := auditlog.InitAuditLogSDK(context.TODO(), time.Second); err != nil {
			return errors.Annotate(err, "initialize audit log sdk failed")
		}
	} else {
		log.Info("audit log is disabled")
	}

	// init dependencies
	controller.InitDependencies()
	controller.RunConfigWatcher()

	// flush all buffering log
	defer func() {
		if err := log.L().Sync(); err != nil {
			fmt.Println(err.Error())
		}
	}()

	return nil
}

// ServeGRPC runs a gRPC server
func (s *server) ServeGRPC() error {
	controller.NewPubSubManager().StartPublisher()
	s.rpcServer = grpc.NewServer(grpc.UnaryInterceptor(
		grpc_middleware.ChainUnaryServer(
			grpc_baseresp.UnaryServerInterceptor(),
			logid.UnaryServerInterceptor(),
			otelgrpc.UnaryServerInterceptor(),
			grpc_validator.UnaryServerInterceptor(),
			grpc_log.UnaryServerInterceptor(),
			grpc_metric.UnaryServerInterceptor(),
			grpc_recovery.UnaryServerInterceptor(grpc_recovery.WithRecoveryHandlerContext(panicHandler)),
		),
	))
	controller.RegisterGrpcServices(s.rpcServer)
	grpc_prometheus.EnableHandlingTimeHistogram()
	grpc_prometheus.Register(s.rpcServer)

	go runMetricsServer()
	s.rpcAddr = grpc_http.Port2Address(config.Global().ServerConfig.GrpcPort)
	log.Info("starting serverless service grpc server",
		zap.String("gitHash", version.GitHash),
		zap.String("logLevel", config.Global().ServerConfig.LogLevel),
		zap.String("address", s.rpcAddr))
	lis, err := net.Listen("tcp", s.rpcAddr)
	if err != nil {
		return err
	}

	return s.rpcServer.Serve(lis)
}

func (s *server) getTokenValidator() (*auth0.RawTokenValidator, error) {
	managementPortalJWKs, err := auth0.GetJWKS(s.portalAuth0.Domain)
	if err != nil {
		return nil, errors.Trace(err)
	}
	tokenValidator := auth0.NewRawTokenValidator().
		RegisterAuth(auth0.AuthValidatorIDManagement, s.portalAuth0, managementPortalJWKs)
	return tokenValidator, nil
}

// serveLRPC runs a local rpc server as the backend of http-gateway
func (s *server) serveLRPC() error {
	tokenValidator, err := s.getTokenValidator()
	if err != nil {
		log.Error("get token validator failed", zap.Error(err))
		return err
	}

	testTokenChecker, err := auth.NewTestTokenChecker(config.Global().ServerConfig.SaltSecret)
	if err != nil {
		log.Error("get test token checker failed", zap.Error(err))
		return err
	}

	credOpt := grpc.Creds(credentials.NewServerTLSFromCert(&insecure.Cert))
	s.rpcServer = grpc.NewServer(credOpt, grpc.UnaryInterceptor(
		grpc_middleware.ChainUnaryServer(
			grpc_baseresp.UnaryServerInterceptor(),
			logid.UnaryServerInterceptor(),
			otelgrpc.UnaryServerInterceptor(),
			grpc_validator.UnaryServerInterceptor(),
			grpc_errcode.UnaryServerInterceptor(),
			auth.UnaryServerInterceptor(testTokenChecker, tokenValidator),
			rbac.UnaryServerInterceptor(),
			grpc_log.UnaryServerInterceptor(),
			grpc_metric.UnaryServerInterceptor(),
			grpc_ratelimit.UnaryServerInterceptor(grpc_ratelimit.NewRateLimiter(time.Minute, 100)),
			allowlist.UnaryServerInterceptor(),
			grpc_recovery.UnaryServerInterceptor(grpc_recovery.WithRecoveryHandlerContext(panicHandler)),
		),
	))
	controller.RegisterLrpcServices(s.rpcServer)
	grpc_prometheus.EnableHandlingTimeHistogram()
	grpc_prometheus.Register(s.rpcServer)

	s.rpcAddr = grpc_http.Port2Address(config.Global().ServerConfig.GrpcPort)
	log.Info("starting serverless service lrpc server",
		zap.String("gitHash", version.GitHash),
		zap.String("logLevel", config.Global().ServerConfig.LogLevel),
		zap.String("address", s.rpcAddr))
	lis, err := net.Listen("tcp", s.rpcAddr)
	if err != nil {
		return err
	}

	return s.rpcServer.Serve(lis)
}

func (s *server) setupAPIGateway() error {
	if !config.IsTestStage() {
		gin.SetMode(gin.ReleaseMode)
	}
	eng := gin.Default()
	err := grpc_http.SetupHTTPGateway(eng,
		[]grpc_http.RegisterHandler{
			serverless.RegisterServerlessBillingServiceHandler,
			console.RegisterBranchServiceHandler,
			console.RegisterClusterServiceHandler,
			console.RegisterExportServiceHandler,
			console.RegisterBackupRestoreServiceHandler,
			console.RegisterImportServiceHandler,
			console.RegisterDatabaseAuditLogServiceHandler,
			console.RegisterMigrationServiceHandler,
			rpc.RegisterClusterCROperationServiceHandler,
			rpc.RegisterRegionalK8SOperationServiceHandler,
			rpc.RegisterChangefeedOpsServiceHandler,
			console.RegisterChangefeedServiceHandler,
			console.RegisterPrivateLinkConnectionServiceHandler,
		},
		[]grpc_http.RegisterHandler{
			v1beta.RegisterImportServiceHandler,
			v1beta.RegisterBranchServiceHandler,
			v1beta.RegisterUserServiceHandler,
		},
		[]grpc_http.RegisterHandler{
			v1beta1.RegisterBranchServiceHandler,
			v1beta1.RegisterImportServiceHandler,
			v1beta1.RegisterExportServiceHandler,
			v1beta1.RegisterAuditLogServiceHandler,
			v1beta1.RegisterDatabaseAuditLogServiceHandler,
			v1beta1.RegisterChangefeedServiceHandler,
			v1beta1.RegisterPrivateLinkConnectionServiceHandler,
			v1beta1.RegisterMigrationServiceHandler,
		})
	if err != nil {
		return errors.Trace(err)
	}

	eng.Any("/healthz", func(c *gin.Context) { c.Status(200) })

	s.httpAddr = config.Global().ServerConfig.HttpListenAddr
	s.httpServer = eng
	return nil
}

func (s *server) ServeHTTP() error {
	controller.NewPubSubManager().StartPublisher()
	eg := errgroup.Group{}
	eg.Go(func() error {
		return s.serveLRPC()
	})

	eg.Go(func() error {
		if err := s.setupAPIGateway(); err != nil {
			log.Error("setup api gateway failed", zap.Error(err))
			return err
		}
		go runMetricsServer()
		log.Info("starting serverless service http server",
			zap.String("gitHash", version.GitHash),
			zap.String("logLevel", config.Global().ServerConfig.LogLevel),
			zap.String("address", s.httpAddr))
		return s.httpServer.Run(s.httpAddr)
	})

	return eg.Wait()
}

func (s *server) RunWorkers() {
	controller.RunDBMigrate()
	controller.RunReconcilers()
	controller.NewPubSubManager().Start()
	controller.NewGlobalWorkers().Start()
}

func (s *server) RunBRWorkers() {
	controller.NewPubSubManager().StartPublisher()
	controller.NewBRWorkers().Start()
}

func (s *server) RunWorkflow() {
	controller.NewWorkflowRunner().Start()
}

func runMetricsServer() {
	port := config.Global().ServerConfig.MetricsPort
	if port <= 0 {
		panic("invalid metrics port")
	}

	mux := http.NewServeMux()
	mux.Handle("/metrics", promhttp.Handler())
	log.Info("starting metrics server")
	err := http.ListenAndServe(grpc_http.Port2Address(port), mux) //nolint:gosec
	if err != nil {
		panic(fmt.Sprintf("start metrics server failed %+v", err))
	}
}

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

	// In case the usual stack trace is not working, put debug.Stack() into the message
	if !ok {
		err = fmt.Errorf("%+vn%s", p, debug.Stack())
		return errcode.NewInternalErr(err)
	}

	return errcode.NewInternalErr(err)
}