package server

import (
	"context"
	"net"

	"github.com/bufbuild/protovalidate-go"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	grpcprometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/app/controllers/interceptor"
	"github.com/tidbcloud/dataflow-service/internal/config"
	grpchttp "github.com/tidbcloud/dataflow-service/internal/grpc/http"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/server"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
)

type RegisterGrpcServer func(rpcServer *grpc.Server)

type GrpcServer struct {
	registerGrpcServerHooks []RegisterGrpcServer
	rpcServer               *grpc.Server
}

func NewGrpcServer(
	ctx context.Context,
	registerGrpcServerHooks []RegisterGrpcServer,
) (*GrpcServer, error) {
	s := &GrpcServer{
		registerGrpcServerHooks: registerGrpcServerHooks,
	}
	validator, err := protovalidate.New()
	if err != nil {
		tilog.Fatal(ctx, "Failed to create proto validator", zap.Error(err))
	}

	unaryInterceptors := s.buildGrpcUnaryInterceptor()
	rpcServer := tigrpc.NewServer(
		tigrpc.WithTracing(),
		tigrpc.WithPrometheus(),
		tigrpc.WithValidator(),
		tigrpc.WithUnaryInterceptors(protovalidate_middleware.UnaryServerInterceptor(validator)),
		tigrpc.WithUnaryInterceptors(unaryInterceptors...))
	for _, hook := range registerGrpcServerHooks {
		hook(rpcServer)
	}

	if err := s.registerPrometheus(ctx, rpcServer); err != nil {
		return nil, err
	}
	s.rpcServer = rpcServer
	reflection.Register(rpcServer)
	return s, nil
}

func (s *GrpcServer) ServerGrpc(ctx context.Context) error {
	rpcAddr := grpchttp.PortToAddress(config.Global().ServerConfig.GrpcPort)
	lis, err := net.Listen("tcp", rpcAddr)
	if err != nil {
		return err
	}
	return s.rpcServer.Serve(lis)
}

func (s *GrpcServer) buildGrpcUnaryInterceptor() []grpc.UnaryServerInterceptor {
	return []grpc.UnaryServerInterceptor{
		interceptor.UserPasserUnaryServerInterceptor(),
		interceptor.ErrorHandlerUnaryServerInterceptor(),
		interceptor.RequestLogUnaryServerInterceptor(),
		interceptor.ClusterCheckerUnaryServerInterceptor(container.Global().CoreGrpcClient),
		interceptor.RecoverPanicInterceptor(),
	}
}

func (s *GrpcServer) registerPrometheus(ctx context.Context, rpcServer *grpc.Server) error {
	grpcprometheus.EnableHandlingTimeHistogram()
	grpcprometheus.Register(rpcServer)
	return nil
}