package router

import (
	"context"
	"errors"
	"net"

	"buf.build/go/protovalidate"
	awssdkconfig "github.com/aws/aws-sdk-go-v2/config"
	grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
	grpclogging "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	"github.com/pingcap/log"
	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	aliyuncloud "github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aliyun"
	cloudaws "github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/aws"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	bucketcontroller "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket"
	dataplanecontroller "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/dataplane"
	keycontroller "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/key"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/service/bucket"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/service/dataplane"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/service/key"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/service/keyspacerole"
	serlb "github.com/tidbcloud/infra-provider/regional-server/pkg/service/loadbalancer"
	serpls "github.com/tidbcloud/infra-provider/regional-server/pkg/service/privatelink"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/service/proxy"
	servpcpeering "github.com/tidbcloud/infra-provider/regional-server/pkg/service/vpcpeering"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"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/codes"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/status"
	"k8s.io/client-go/dynamic"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func SetupAndRunGRPCServer(
	ctx context.Context,
	cli client.Client,
	dynamicCli dynamic.Interface,
	config rsconfig.RootConfig,
	tmplconfig rsconfig.TemplatesConfig,
	addr string,
) {
	validator, err := protovalidate.New()
	if err != nil {
		tilog.Fatal(ctx, "Failed to create proto validator", zap.Error(err))
	}
	lis, err := net.Listen("tcp", addr)
	if err != nil {
		tilog.Fatal(ctx, "Failed to listen for gRPC server", zap.String("addr", addr), zap.Error(err))
	}
	s := grpc.NewServer(
		grpc.StatsHandler(otelgrpc.NewServerHandler()),
		grpc.ChainUnaryInterceptor(
			grpcrecovery.UnaryServerInterceptor(grpcrecovery.WithRecoveryHandlerContext(recoveryHandler)),
			errorHandlerUnaryServerInterceptor(),
			protovalidate_middleware.UnaryServerInterceptor(validator),
			grpclogging.UnaryServerInterceptor(interceptorLogger(log.L())),
		),
	)
	var cloudProviderExtractor dataplane.CloudProviderExtractor
	var externalPeeringQueryer servpcpeering.ExternalPeeringQueryer
	var awsCli cloudaws.Client
	var aliyunCli aliyuncloud.Client
	switch config.Provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		awsCfg, loadErr := awssdkconfig.LoadDefaultConfig(ctx)
		if loadErr != nil {
			tilog.Fatal(ctx, "failed to init aws client set", zap.Error(loadErr))
		}
		awsCli = cloudaws.NewClientSet(awsCfg)
		var err error
		cloudProviderExtractor, err = dataplane.NewAWSCloudProviderExtractor(
			cli, config.DataPlaneRoleName, config.AWSDataPlaneExternalID, config.DNSProviderConfig.AwsRoute53.Domain)
		if err != nil {
			tilog.Fatal(ctx, "failed to init aws cloud provider extractor", zap.Error(err))
		}
		if config.BYOC.BYOCPrivateLink.IsEnabled() {
			externalPeeringQueryer = servpcpeering.NewAWSExternalPeeringQueryer(awsCli)
		}
	case consts.ProviderTypeACK:
		aliyunCli = aliyuncloud.NewClientSet()
		cloudProviderExtractor = dataplane.NewAliCloudProviderExtractor(
			cli, config.DataPlaneRoleName, config.DNSProviderConfig.AliCloudDNS.Domain)
		if config.BYOC.BYOCPrivateLink.IsEnabled() {
			externalPeeringQueryer = servpcpeering.NewAliyunExternalPeeringQueryer(aliyunCli)
		}
	case consts.ProviderTypeAKS:
		cloudProviderExtractor = dataplane.NewAzureCloudProviderExtractor(
			cli, config.DNSProviderConfig.AzureDNSZone.Domain)
	default:
		tilog.Fatal(ctx, "unsupported cloud provider", zap.String("Provider", config.Provider))
	}
	var dataPlaneOptions []dataplane.DataPlaneServiceOption
	if config.Provider == consts.ProviderTypeEKS {
		resolver, resolverErr := dataplanecontroller.NewAWSPrivateLinkEndpointZoneResolver(awsCli)
		if resolverErr != nil {
			tilog.Fatal(ctx, "failed to initialize private link endpoint zone resolver", zap.Error(resolverErr))
		}
		dataPlaneOptions = append(dataPlaneOptions, dataplane.WithPrivateLinkEndpointZoneResolver(resolver))
	}
	dataPlaneSvc := dataplane.NewDataPlaneService(
		cli,
		repository.NewDataPlaneReaderWriter(cli),
		repository.NewVPCReaderWriter(cli),
		repository.NewInternetReaderWriter(cli),
		repository.NewAPPBoxReaderWriter(cli),
		cloudProviderExtractor,
		config,
		tmplconfig,
		dataPlaneOptions...,
	)
	lbSvc := serlb.NewLoadBalancerService(
		repository.NewLoadBalancerReaderWriter(cli),
		repository.NewDataPlaneReader(cli),
		config,
	)
	plsSvc := serpls.NewService(
		repository.NewPrivateLinkServiceReaderWriter(cli),
		repository.NewPrivateLinkEndpointReaderWriter(cli),
		repository.NewDataPlaneReader(cli),
		repository.NewVPCReader(cli),
		config,
		awsCli,
		aliyunCli,
	)
	bucketSvc := bucket.NewBucketService(
		repository.NewDataPlaneReaderWriter(cli),
		repository.NewAPPBoxReaderWriter(cli),
		repository.NewBucketReaderWriter(cli),
		bucketcontroller.NewBucketQueryer(cli),
		config,
	)
	keySvc := key.NewService(
		repository.NewDataPlaneReaderWriter(cli),
		repository.NewAPPBoxReaderWriter(cli),
		repository.NewKeyReaderWriter(cli),
		keycontroller.NewKeyQueryer(cli, config),
		config,
	)
	keyspaceRoleSvc := keyspacerole.NewService(
		cli,
		repository.NewDataPlaneReader(cli),
		repository.NewAPPBoxReader(cli),
		repository.NewBucketReader(cli),
		config,
	)
	vpcPeeringSvc := servpcpeering.NewService(
		repository.NewVPCPeeringReaderWriter(cli),
		repository.NewDataPlaneReaderWriter(cli),
		repository.NewVPCReaderWriter(cli),
		externalPeeringQueryer,
		config,
	)
	controlPlaneSvc := proxy.NewK8sServer(dynamicCli)
	dataPlaneProxySvc := proxy.NewDataPlaneK8sServer(
		cli,
		repository.NewDataPlaneReaderWriter(cli),
		repository.NewAPPBoxReaderWriter(cli),
		config.Namespace,
	)

	dmpbv1.RegisterDataPlaneServiceServer(s, dataPlaneSvc)
	dmpbv1.RegisterLoadBalancerServiceServer(s, lbSvc)
	dmpbv1.RegisterPrivateLinkConnectionServiceServer(s, plsSvc)
	dmpbv1.RegisterBucketServiceServer(s, bucketSvc)
	dmpbv1.RegisterKeyServiceServer(s, keySvc)
	dmpbv1.RegisterKeyspaceRoleServiceServer(s, keyspaceRoleSvc)
	dmpbv1.RegisterVpcPeeringServiceServer(s, vpcPeeringSvc)
	dmpbv1.RegisterControlPlaneK8SProxyServiceServer(s, controlPlaneSvc)
	dmpbv1.RegisterDataPlaneK8SProxyServiceServer(s, dataPlaneProxySvc)

	// refer to https://grpc.io/docs/guides/health-checking/
	grpc_health_v1.RegisterHealthServer(s, health.NewServer())
	tilog.Info(ctx, "Starting internal gRPC server", zap.String("addr", addr))
	go func() {
		if err2 := s.Serve(lis); err2 != nil {
			tilog.Info(ctx, "Internal gRPC server returned with error", zap.String("addr", addr), zap.Error(err2))
		}
	}()

	<-ctx.Done()

	tilog.Info(ctx, "Stopping internal gRPC server...", zap.String("addr", addr))
	s.GracefulStop()
	tilog.Info(ctx, "Internal gRPC server stopped", zap.String("addr", addr))
}

func recoveryHandler(ctx context.Context, p interface{}) error {
	tilog.Error(ctx, "panic triggered", zap.Any("panic", p))
	return status.Errorf(codes.Internal, "panic triggered: %v", p)
}

// interceptorLogger adapts zap logger to interceptor logger.
// This code is simple enough to be copied and not imported.
func interceptorLogger(logger *zap.Logger) grpclogging.Logger {
	return grpclogging.LoggerFunc(func(ctx context.Context, lvl grpclogging.Level, msg string, fields ...any) {
		f := make([]zap.Field, 0, len(fields)/2)

		for i := 0; i < len(fields); i += 2 {
			key := fields[i]
			value := fields[i+1]

			switch v := value.(type) {
			case string:
				f = append(f, zap.String(key.(string), v))
			case int:
				f = append(f, zap.Int(key.(string), v))
			case bool:
				f = append(f, zap.Bool(key.(string), v))
			default:
				f = append(f, zap.Any(key.(string), v))
			}
		}

		logger := logger.WithOptions(zap.AddCallerSkip(1)).With(f...).With(tilog.FieldsFromContext(ctx)...)

		levelMap := map[grpclogging.Level]zapcore.Level{
			grpclogging.LevelDebug: zap.DebugLevel,
			grpclogging.LevelInfo:  zap.DebugLevel, // there are too many useless grpc request logs, resulting in hard to search controller logs
			grpclogging.LevelWarn:  zap.WarnLevel,
			grpclogging.LevelError: zap.ErrorLevel,
		}
		zapLevel, ok := levelMap[lvl]
		if !ok {
			zapLevel = zap.InfoLevel
		}
		logger.Log(zapLevel, msg)
	})
}

func errorHandlerUnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(
		ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
	) (interface{}, error) {
		resp, err := handler(ctx, req)
		if err == nil {
			return resp, nil
		}

		tilog.Warn(ctx, "grpc call error",
			zap.Error(err),
			zap.String("method", info.FullMethod),
			zap.Any("req", req))

		var bizErr bizerr.Error
		if errors.As(err, &bizErr) {
			return resp, status.Error(bizErr.Code().GetGrpcCode(), bizErr.Error())
		}
		return resp, err
	}
}