package main

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"net/textproto"
	"strconv"
	"time"

	"buf.build/go/protovalidate"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	awsConfig "github.com/aws/aws-sdk-go-v2/config"
	"github.com/gin-gonic/gin"
	grpclogging "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	grpcrecovery "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
	gwruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/pingcap/log"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	tigrpcc "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"go.opentelemetry.io/otel/trace"
	"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"
	"google.golang.org/protobuf/proto"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/cluster-controller/server/middlewares"
	"github.com/tidbcloud/cluster-controller/server/pkg/cloudprovider"
	"github.com/tidbcloud/cluster-controller/server/pkg/config"
	"github.com/tidbcloud/cluster-controller/server/server"
	"github.com/tidbcloud/cluster-controller/server/server/operation"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/cloud/azure"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	utiltime "github.com/tidbcloud/infra-provider/pkg/util/time"
)

const (
	baseAPIPath           = "/api/v1"
	baseAPIPathWithPrefix = "/cluster-server/api/v1"
	healthzEndpoint       = "/healthz"
)

func setupAndRunGRPCServer(ctx context.Context, cfg *config.Config, mgr ctrl.Manager) {
	addr := cfg.InternalGRPCAddr
	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)),
			protovalidate_middleware.UnaryServerInterceptor(validator),
			grpclogging.UnaryServerInterceptor(InterceptorLogger(log.L())),
		),
	)

	var (
		auditLogViewer server.AuditLogViewer
		azureCli       azure.Client
		awsClient      aws.Client
	)
	switch provider := cfg.Provider; provider {
	case "aws":
		awsClient, err = initAWSClientSet(cfg)
		if err != nil {
			tilog.Fatal(ctx, "init AWS Client set", zap.Error(err))
		}
		auditLogViewer = server.NewAwsAuditLogViewer(awsClient)
	case "gcp":
		gcpCli, err := gcp.NewClientSet()
		if err != nil {
			tilog.Fatal(ctx, "init GCP Client set", zap.Error(err))
		}
		auditLogViewer = server.NewGcpAuditLogViewer(gcpCli)
	case "azure":
		cred, err := azidentity.NewDefaultAzureCredential(nil)
		if err != nil {
			tilog.Fatal(ctx, "failed to create Azure credential", zap.Error(err))
		}
		azureCli, err = azure.NewClientSet(cred)
		if err != nil {
			tilog.Fatal(ctx, "failed to create Azure Client set", zap.Error(err))
		}
		auditLogViewer = server.NewAzureAuditLogViewer()
	case "alicloud":
		// nothing to do
	case "kind":
		auditLogViewer = server.NewDummyAuditLogViewer()
	default:
		panic(fmt.Sprintf("unknown provider: %s", provider))
	}

	instanceTypeProvider, err := cloudprovider.NewInstanceTypeProvider(context.Background(), mgr.GetClient())
	if err != nil {
		panic(fmt.Sprintf("failed to create instance type provider: %v", err))
	}
	nodePoolClient, err := client.New(mgr.GetConfig(), client.Options{
		Scheme: mgr.GetScheme(),
		Mapper: mgr.GetRESTMapper(),
	})
	if err != nil {
		tilog.Fatal(ctx, "failed to create direct nodepool client", zap.Error(err))
	}
	apiv1.RegisterClusterServiceServer(
		s,
		server.NewClusterServer(
			mgr.GetClient(),
			auditLogViewer,
			instanceTypeProvider,
			server.WithAzureCli(azureCli),
			server.WithLabelSelectorReader(),
			server.WithAWSCli(awsClient),
			server.WithNodePoolClient(nodePoolClient),
		),
	)
	apiv1.RegisterClusterOperationServiceServer(s, operation.NewServer(mgr.GetClient(), utiltime.RealClock{}))
	apiv1.RegisterTidbGroupServiceServer(s, server.NewTidbGroupServer(mgr.GetClient()))
	apiv1.RegisterPrivateLinkServiceServer(s, &server.PrivateLinkServer{KubeClient: mgr.GetClient()})
	apiv1.RegisterReplayServiceServer(s, server.NewReplayServer(mgr.GetClient(), server.WithLabelSelectorReplayReader()))
	apiv1.RegisterBRServiceServer(s, server.NewBRServer(mgr.GetClient(), instanceTypeProvider))
	// 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 setupAndRunGRPCGateway(ctx context.Context, addr, grpcAddr string) {
	conn, err := tigrpcc.NewClientConn(grpcAddr, tigrpcc.WithDefault())
	if err != nil {
		tilog.Fatal(ctx, "Failed to dial gRPC server", zap.String("addr", grpcAddr), zap.Error(err))
	}
	defer func() {
		go func() {
			<-ctx.Done()
			if cerr := conn.Close(); cerr != nil {
				tilog.Info(ctx, "Failed to close conn to %s: %v", zap.String("addr", grpcAddr), zap.Error(cerr))
			}
		}()
	}()

	mux := gwruntime.NewServeMux(
		gwruntime.WithForwardResponseOption(httpResponseModifier),
		// refer to https://grpc-ecosystem.github.io/grpc-gateway/docs/operations/health_check/
		gwruntime.WithHealthEndpointAt(grpc_health_v1.NewHealthClient(conn), healthzEndpoint),
	)
	if err := apiv1.RegisterClusterServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for cluster service", zap.Error(err))
	}
	if err := apiv1.RegisterClusterOperationServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for cluster operation service", zap.Error(err))
	}
	if err := apiv1.RegisterReplayServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for replay service", zap.Error(err))
	}
	if err := apiv1.RegisterTidbGroupServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for load balancer service", zap.Error(err))
	}
	if err := apiv1.RegisterPrivateLinkServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for private link service", zap.Error(err))
	}
	if err := apiv1.RegisterBRServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for private link service", zap.Error(err))
	}

	apiRouter := gin.New()
	// log with custom format; strip the base path (/api/v1) from the request path
	apiRouter.Use(
		gin.CustomRecovery(func(c *gin.Context, err any) {
			tilog.Error(c.Request.Context(), "panic triggered", zap.String("panic", fmt.Sprintf("%v", err)))
		}),
		otelgin.Middleware(serviceName),
		middlewares.CommonLogging,
		gin.Logger(),
		middlewares.ServePathStartFromAny,
		//tigin.RecordMetrics,
		SetResponseTraceID,
	)
	apiRouter.Any(fmt.Sprintf("%s/*any", baseAPIPath), gin.WrapH(mux))
	apiRouter.Any(fmt.Sprintf("%s/*any", baseAPIPathWithPrefix), gin.WrapH(mux))
	apiRouter.Any(healthzEndpoint, gin.WrapH(mux))
	srv := &http.Server{
		Addr:    addr,
		Handler: apiRouter,
	}

	lis, err := net.Listen("tcp", addr)
	if err != nil {
		tilog.Fatal(ctx, "Failed to listen for gRPC gateway", zap.String("addr", addr), zap.Error(err))
	}
	tilog.Info(ctx, "Starting gRPC gateway", zap.String("addr", addr))
	go func() {
		if err2 := srv.Serve(lis); err2 != nil && !errors.Is(err2, http.ErrServerClosed) {
			tilog.Info(ctx, "The gRPC gateway returned with error", zap.String("addr", addr), zap.Error(err2))
		}
	}()

	<-ctx.Done()

	// The context is used to inform the server it has 5 seconds to finish
	// the request it is currently handling
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	tilog.Info(ctx, "Stopping gRPC gatway...", zap.String("addr", addr))
	if err := srv.Shutdown(ctx); err != nil {
		tilog.Error(ctx, "gRPC gateway forced to shutdown", zap.String("addr", addr), zap.Error(err))
	} else {
		tilog.Info(ctx, "gRPC gateway stopped", zap.String("addr", addr))
	}
}

func httpResponseModifier(ctx context.Context, w http.ResponseWriter, p proto.Message) error {
	md, ok := gwruntime.ServerMetadataFromContext(ctx)
	if !ok {
		return nil
	}

	// set http status code
	if vals := md.HeaderMD.Get(server.ResponseKeyStatusCode); len(vals) > 0 {
		code, err := strconv.Atoi(vals[0])
		if err != nil {
			return err
		}
		// delete the headers to not expose any grpc-metadata in http response
		delete(md.HeaderMD, server.ResponseKeyStatusCode)
		delete(w.Header(), fmt.Sprintf("Grpc-Metadata-%s", textproto.CanonicalMIMEHeaderKey(server.ResponseKeyStatusCode)))
		w.WriteHeader(code)
	}

	return nil
}

func recoveryHandler(ctx context.Context, p any) (err error) {
	tilog.Error(ctx, "panic triggered", zap.Any("panic", p))
	return status.Errorf(codes.Unknown, "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.InfoLevel,
			grpclogging.LevelWarn:  zap.WarnLevel,
			grpclogging.LevelError: zap.ErrorLevel,
		}
		zapLevel, ok := levelMap[lvl]
		if !ok {
			zapLevel = zap.InfoLevel
		}
		logger.Log(zapLevel, msg)
	})
}

// SetResponseTraceID is copied from tiinf/transport/tigin/middlewares.go to avoid import too many useless packages
func SetResponseTraceID(c *gin.Context) {
	spanCtx := trace.SpanContextFromContext(c.Request.Context())
	if spanCtx.HasTraceID() {
		c.Header("x-debug-trace-id", spanCtx.TraceID().String())
	}
	c.Next()
}

func initAWSClientSet(c *config.Config) (aws.Client, error) {
	cfg, err := awsConfig.LoadDefaultConfig(context.TODO(), awsConfig.WithRegion(c.Region))
	if err != nil {
		return nil, err
	}
	return aws.NewClientSet(cfg), nil
}