package server

import (
	"context"
	"fmt"
	"net/http"
	"strings"

	"github.com/gin-gonic/gin"
	grpcruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server/wrapper"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/pkgv2/logid"
	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/encoding/protojson"
)

const (
	setHeaderPrefix    = "x-set-header_"
	queryHeaderPrefix  = "x-query-header_"
	cookieHeaderPrefix = "x-cookie-header_"
	httpMethod         = "x-http-method"
	httpURI            = "x-http-uri"
	httpPath           = "x-forwarded-path"
)

type HttpServer struct {
	registerRequestMappingHooks []RegisterRequestMappingFunc

	httpServer *gin.Engine
}

type RegisterRequestMappingFunc func(ctx context.Context,
	eng *gin.Engine,
	gwmux *grpcruntime.ServeMux,
	conn *grpc.ClientConn,
) error

func NewHttpServer(ctx context.Context, registerRequestMappingHooks []RegisterRequestMappingFunc) (*HttpServer, error) {
	s := &HttpServer{
		registerRequestMappingHooks: registerRequestMappingHooks,
	}
	eng := gin.Default()
	s.httpServer = eng

	eng.Use(
		AllowAll(),
		logid.InjectLogid4GRPCGatewayMW,
		otelgin.Middleware("dataflow-service"),
		wrapper.WriterWrapperHandler, // modify response body middleware
		ModifyGrpcErrorHandler,       // modify grpc error middleware
	)
	return s, nil
}

func (s *HttpServer) ServerHttpGateWayForGrpc(ctx context.Context, port int) error {
	grpcAddress := fmt.Sprintf("127.0.0.1:%d", config.Global().ServerConfig.GrpcPort)
	conn, err := s.gatewayConn(grpcAddress)
	if err != nil {
		return err
	}

	gwmux := s.makeGatewayMux(GRPCGatewayErrorHandler)
	for _, hook := range s.registerRequestMappingHooks {
		if err := hook(ctx, s.httpServer, gwmux, conn); err != nil {
			return err
		}
	}
	s.httpServer.Any("/healthz", func(c *gin.Context) {
		c.Status(200)
	})

	httpAddr := fmt.Sprintf("0.0.0.0:%d", port)
	return s.httpServer.Run(httpAddr)
}

// GatewayConn create a connection that can be used by a gateway
func (s *HttpServer) gatewayConn(addr string) (*grpc.ClientConn, error) {
	// See https://github.com/grpc/grpc/blob/master/doc/naming.md
	// for gRPC naming standard information.
	dialAddr := fmt.Sprintf("passthrough://0.0.0.0/%s", addr)
	dialOpts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithBlock(),
		grpc.WithChainUnaryInterceptor(
			otelgrpc.UnaryClientInterceptor(),
		),
	}
	return grpc.DialContext(
		context.Background(),
		dialAddr,
		dialOpts...,
	)
}

// // makeGatewayMuxForV1beta1 camel case json marshaler with UseProtoNames: false
func (s *HttpServer) makeGatewayMuxForV1beta1(fn grpcruntime.ErrorHandlerFunc) *grpcruntime.ServeMux {
	jsonpb := &grpcruntime.JSONPb{
		MarshalOptions: protojson.MarshalOptions{
			UseProtoNames:   false, // use camel case
			EmitUnpopulated: true,
		},
		UnmarshalOptions: protojson.UnmarshalOptions{
			DiscardUnknown: true,
		},
	}
	// This marshaler enables support to return raw response using HttpBody.
	// Note that this behavior is enabled by default in grpc-gateway v2.
	marshaler := &grpcruntime.HTTPBodyMarshaler{
		Marshaler: jsonpb,
	}
	mux := grpcruntime.NewServeMux(
		grpcruntime.WithMarshalerOption(grpcruntime.MIMEWildcard, marshaler),
		grpcruntime.WithIncomingHeaderMatcher(s.allHeaderMatcher),
		grpcruntime.WithMetadata(s.addHttpDetail),
		grpcruntime.WithOutgoingHeaderMatcher(s.addHeaderInResponseMatcher),
		grpcruntime.WithErrorHandler(fn),
	)
	return mux
}

func (s *HttpServer) makeGatewayMux(fn grpcruntime.ErrorHandlerFunc) *grpcruntime.ServeMux {
	snakeCaseMarshaler := &grpcruntime.JSONPb{
		MarshalOptions: protojson.MarshalOptions{
			UseProtoNames:   true,
			EmitUnpopulated: true,
		},
		UnmarshalOptions: protojson.UnmarshalOptions{
			DiscardUnknown: true,
		},
	}
	camelCaseMarshaler := &grpcruntime.JSONPb{
		MarshalOptions: protojson.MarshalOptions{
			UseProtoNames:   false, // use camel case
			EmitUnpopulated: true,
		},
		UnmarshalOptions: protojson.UnmarshalOptions{
			DiscardUnknown: true,
		},
	}

	mux := grpcruntime.NewServeMux(
		// default to snake case marshaler for backward compatibility, and camel case marshaler for new APIs
		grpcruntime.WithMarshalerOption(grpcruntime.MIMEWildcard, snakeCaseMarshaler),
		grpcruntime.WithMarshalerOption("application/json", snakeCaseMarshaler),
		grpcruntime.WithMarshalerOption("application/json+camel", camelCaseMarshaler),

		grpcruntime.WithIncomingHeaderMatcher(s.allHeaderMatcher),
		grpcruntime.WithMetadata(s.addHttpDetail),
		grpcruntime.WithOutgoingHeaderMatcher(s.addHeaderInResponseMatcher),
		grpcruntime.WithErrorHandler(fn),
	)
	return mux
}

func (s *HttpServer) addHeaderInResponseMatcher(key string) (string, bool) {
	if strings.HasPrefix(key, setHeaderPrefix) {
		return strings.TrimPrefix(key, setHeaderPrefix), true
	}
	return key, false
}

// AllHeaderMatcher add all headers to grpc
func (s *HttpServer) allHeaderMatcher(key string) (string, bool) {
	// We can not pass ”Connection: keep-alive“ to grpc server, which can lead to a 500 error.
	// The example response is listed below:
	// HTTP/1.1 500 Internal Server Error
	// ...
	// {
	// 	"code": 13,
	// 	"message": "stream terminated by RST_STREAM with error code: PROTOCOL_ERROR",
	// 	"details": [
	// 	]
	// }
	if strings.ToLower(key) == "connection" {
		return "", false
	}
	return key, true
}

func (s *HttpServer) addHttpDetail(_ context.Context, r *http.Request) metadata.MD {
	m := make(map[string]string)
	m[httpURI] = r.RequestURI
	m[httpMethod] = r.Method
	for key, value := range r.URL.Query() {
		m[queryHeaderPrefix+key] = value[0]
	}
	for _, c := range r.Cookies() {
		m[cookieHeaderPrefix+c.Name] = c.Value
	}
	return metadata.New(m)
}