package server

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

	"github.com/bufbuild/protovalidate-go"
	"github.com/gin-gonic/gin"
	"github.com/go-resty/resty/v2"
	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"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/infra-provider/pkg/auth"
	"github.com/tidbcloud/ticdc-service/internal/server/middleware"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy"
	proxycli "github.com/tidbcloud/ticdc-service/pkg/clusterproxy/client"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/tidbcloud/ticdc-service/pkg/service"
	"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"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	baseAPIPath     = "/api/v1"
	healthzEndpoint = "/healthz"
	ServiceName     = "ticdc-service"

	// ResponseKeyStatusCode is the key of the HTTP status code in the response header of gRPC.
	// if we return an error in the gRPC service, the HTTP response body will be something like
	// `{"code":2, "message":"...", "details":[]}`.
	// so we choose to return a nil error and set the HTTP status code in the response header of gRPC.
	ResponseKeyStatusCode = "x-http-code"
)

func SetupAndRunGRPCServer(ctx context.Context, cfg *config.Config, cli client.Client) {
	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())),
		),
	)

	regionalCli := regionalservice.NewRegionalServiceClient(cfg.RegionalServiceBaseURL)

	auth, err := initProxyAuth(ctx, cfg)
	if err != nil {
		tilog.Fatal(ctx, "Failed to init proxy auth", zap.Error(err))
	}
	opts := []proxycli.Option{proxycli.WithAuth(auth)}
	if cfg.Debug.Enable && cfg.Debug.ClusterProxy != "" {
		opts = append(opts, proxycli.WithEndpoint(cfg.Debug.ClusterProxy))
	}

	proxyModel := clusterproxy.NewClusterProxyModelImpl(cfg, regionalCli, proxycli.DefaultClientCreator, opts...)
	apiv1.RegisterTiCDCClusterServiceServer(s, service.NewTiCDCClusterServiceServer(cli, cfg, proxyModel, regionalCli))
	apiv1.RegisterTiCDCPrecheckServiceServer(s, service.NewPreCheckServiceServer(proxyModel))
	apiv1.RegisterTiCDCSinkEndpointServiceServer(s, service.NewSinkEndpointServiceServer(cli, regionalCli, cfg))
	apiv1.RegisterTiCDCCertificateServiceServer(s, service.NewCertificateServer(cli, regionalCli, cfg))

	// 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.RegisterTiCDCPrecheckServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for ticdc precheck service", zap.Error(err))
	}
	if err := apiv1.RegisterTiCDCSinkEndpointServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for ticdc sink endpoint service", zap.Error(err))
	}
	if err := apiv1.RegisterTiCDCClusterServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for ticdc cluster service", zap.Error(err))
	}
	if err := apiv1.RegisterTiCDCCertificateServiceHandler(ctx, mux, conn); err != nil {
		tilog.Fatal(ctx, "Failed to register gRPC gateway for ticdc certificate 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(ctx, "panic triggered", zap.String("panic", fmt.Sprintf("%v", err)))
		}),
		otelgin.Middleware(ServiceName),
		middleware.CommonLogging,
		gin.Logger(),
		middleware.ServePathStartFromAny,
		// tigin.RecordMetrics,
		SetResponseTraceID,
	)
	apiRouter.Any(fmt.Sprintf("%s/*any", baseAPIPath), 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))
	}
}

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

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(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, ResponseKeyStatusCode)
		delete(w.Header(), fmt.Sprintf("Grpc-Metadata-%s", textproto.CanonicalMIMEHeaderKey(ResponseKeyStatusCode)))
		w.WriteHeader(code)
	}

	return nil
}

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

func initProxyAuth(ctx context.Context, cfg *config.Config) (func(r *resty.Request) error, error) {
	// ignore auth if the dev mode is enabled
	if cfg.Proxy.EnableDevMode {
		tilog.Info(ctx, "dev mode is enabled for cluster-proxy, disable auth0 M2M")
		return nil, nil
	}

	proxyAuth0ClientId, ok1 := os.LookupEnv("PROXY_AUTH0_CLIENT_ID")
	proxyAuth0ClientSecret, ok2 := os.LookupEnv("PROXY_AUTH0_CLIENT_SECRET")
	if ok1 && ok2 {
		// verify auth0 credentials and warmup token cache
		_, err := auth.GetClientToken(cfg.Proxy.Auth0Domain, proxyAuth0ClientId, proxyAuth0ClientSecret, cfg.Proxy.Auth0Audience)
		if err != nil {
			return nil, err
		}
		opt := auth.NewAuth0RestyAuth(cfg.Proxy.Auth0Domain, proxyAuth0ClientId, proxyAuth0ClientSecret, cfg.Proxy.Auth0Audience)
		return opt, nil
	}

	return nil, nil
}