package bootstrap

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

	"github.com/gin-gonic/gin"
	grpcruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service-ng/app/container"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	importintopb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/import_into/v1beta2"
	changefeedopenpb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
	premiumdmpb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/premium_dm/v1beta2"
	"github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service-ng/internal/cmd/bootstrap/server/auth"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/dataflow-service-ng/internal/metrics"

	"github.com/tidbcloud/dataflow-service-ng/pkg/log"

	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
)

type DataflowHttpServer struct {
	sever           *server.Server
	grpcServer      *server.GrpcServer
	httpServer      *server.HttpServer
	metricsSerer    *server.MetricsServer
	innerHttpServer *server.HttpServer
}

func NewDataflowHttpServer() (*DataflowHttpServer, error) {
	s := &DataflowHttpServer{}
	ctx := log.WithLogID(context.Background())
	logger := log.GetLogger(ctx)
	ser, err := server.NewServer(ctx, server.ServerOption{
		EnableEventBusConsumer: true,
	})
	if err != nil {
		logger.Error("setup server error", zap.Error(err))
		return nil, err
	}
	s.sever = ser

	grpcServer, err := server.NewGrpcServer(ctx, []server.RegisterGrpcServer{
		s.registerConsoleAPI,
		s.registerOpsAPI,
	})
	if err != nil {
		logger.Error("init grpc server error", zap.Error(err))
		return nil, err
	}
	s.grpcServer = grpcServer

	httpServer, err := server.NewHttpServer(ctx, []server.RegisterRequestMappingFunc{
		s.registerRequestMapping,
	})
	if err != nil {
		logger.Error("init http server error", zap.Error(err))
		return nil, err
	}
	s.httpServer = httpServer

	metricsServer, err := server.NewMetricsServer(ctx)
	prometheus.MustRegister(metrics.HttpErrorResultCounter)
	prometheus.MustRegister(metrics.DownstreamRequestDuration)
	if err != nil {
		logger.Error("init metrics server error", zap.Error(err))
		return nil, err
	}
	s.metricsSerer = metricsServer

	innerHttpServer, err := server.NewHttpServer(ctx, []server.RegisterRequestMappingFunc{
		s.registerInnerRequestMapping,
	})
	if err != nil {
		logger.Error("init inner http server error", zap.Error(err))
		return nil, err
	}
	s.innerHttpServer = innerHttpServer
	return s, nil
}

func (s *DataflowHttpServer) RunDataflowHttpServer() error {
	ctx := context.Background()
	eg := errgroup.Group{}
	eg.Go(func() error {
		return s.grpcServer.ServerGrpc(ctx)
	})
	eg.Go(func() error {
		return s.httpServer.ServerHttpGateWayForGrpc(ctx, config.Global().ServerConfig.HttpPort)
	})
	eg.Go(func() error {
		return s.metricsSerer.ServerMetrics(ctx)
	})
	eg.Go(func() error {
		return s.innerHttpServer.ServerHttpGateWayForGrpc(ctx, config.Global().ServerConfig.InnerHttpPort)
	})
	return eg.Wait()
}

func (s *DataflowHttpServer) registerOpsAPI(rpcServer *grpc.Server) {
	dataflow.RegisterOpsInternalServiceServer(rpcServer, container.Global().OpsInnerController)
}

func (s *DataflowHttpServer) registerConsoleAPI(rpcServer *grpc.Server) {
	s.registerChangefeedConsoleAPI(rpcServer)
	dataflow.RegisterDataPipelineServiceServer(rpcServer, container.Global().DataPipelineController)
	changefeedopenpb.RegisterChangefeedServiceServer(rpcServer, container.Global().OpenChangefeedAPI)
	s.registerImportIntoAPI(rpcServer)
	s.registerPremiumDMAPI(rpcServer)
}

func (s *DataflowHttpServer) registerChangefeedConsoleAPI(rpcServer *grpc.Server) {
	dataflow.RegisterChangefeedSpecServiceServer(
		rpcServer, container.Global().BizChangefeedSpecController)
	dataflow.RegisterChangefeedServiceServer(rpcServer, container.Global().BizChangefeedController)
	dataflow.RegisterSinkEndpointServiceServer(rpcServer, container.Global().SinkEndpointController)
}

func (s *DataflowHttpServer) registerPremiumDMAPI(rpcServer *grpc.Server) {
	premiumdmpb.RegisterPremiumMigrationServiceServer(rpcServer, container.Global().PremiumDMAPI)
}

func (s *DataflowHttpServer) registerImportIntoAPI(rpcServer *grpc.Server) {
	importintopb.RegisterImportIntoServiceServer(rpcServer, container.Global().ImportIntoAPI)
}

func trimConsoleAPIPathPrefixHandler(c *gin.Context) {
	c.Request.URL.Path = strings.TrimPrefix(c.Request.URL.Path, "/api/v2025/cloud")
	c.Next()
}

func (s *DataflowHttpServer) registerRequestMapping(
	ctx context.Context,
	eng *gin.Engine,
	gwmux *grpcruntime.ServeMux,
	conn *grpc.ClientConn,
) error {
	err := s.registerRPCGateWayHandler(ctx, gwmux, conn)
	if err != nil {
		return err
	}
	commomMiddlewares := []gin.HandlerFunc{
		auth.Auth,
		auth.RBAC,
		auth.DataflowUserPasser,
	}

	v1beta2UIAPI := eng.Group("/api/v2025/cloud")
	v1beta2UIAPI.Use(
		append([]gin.HandlerFunc{trimConsoleAPIPathPrefixHandler}, commomMiddlewares...)...,
	)
	v1beta2UIAPI.Any("/v1beta2/tidbs/:cluster_id/*any", gin.WrapH(gwmux))

	// Used by CDC consoleAPI
	v1beta2API := eng.Group("/api/v1beta2")
	v1beta2API.Use(commomMiddlewares...)
	v1beta2API.Any("/tidbs/:cluster_id/*any", gin.WrapH(gwmux))
	v1beta2API.Any("/orgs/:org_id/*any", gin.WrapH(gwmux))

	// used by premium DM, import-into, CDC openAPI
	v1beta2 := eng.Group("/v1beta2")
	v1beta2.Use(auth.OpenAPIMiddlewares...)
	v1beta2.Any("/tidbs/:cluster_id/*any", gin.WrapH(gwmux))
	v1beta2.Any("/changefeeds", gin.WrapH(gwmux))
	v1beta2.Any("/changefeeds/*any", gin.WrapH(gwmux))

	registerOpenAPIV1beta2PassthroughRoutes(eng, gwmux)

	if err := s.registerInnerRequestMappingForE2E(ctx, eng, gwmux, conn); err != nil {
		return err
	}
	return nil
}

func registerOpenAPIV1beta2PassthroughRoutes(eng *gin.Engine, gateway http.Handler) {
	eng.Any("/v1beta2/changefeeds:listCCUs", gin.WrapH(gateway))
}

func (s *DataflowHttpServer) registerInnerRequestMappingForE2E(
	ctx context.Context,
	eng *gin.Engine,
	gwmux *grpcruntime.ServeMux,
	conn *grpc.ClientConn,
) error {
	if config.Global().Env == "prod" {
		return nil
	}

	// only register internal ops API to console API on dev & staging environment to support e2e test
	if err := dataflow.RegisterOpsInternalServiceHandler(ctx, gwmux, conn); err != nil {
		return err
	}
	return nil
}

func (s *DataflowHttpServer) registerInnerRequestMapping(
	ctx context.Context,
	eng *gin.Engine,
	gwmux *grpcruntime.ServeMux,
	conn *grpc.ClientConn,
) error {
	if err := dataflow.RegisterOpsInternalServiceHandler(ctx, gwmux, conn); err != nil {
		return err
	}
	eng.Any("/api/v1beta2/dataflow/ops/*any", gin.WrapH(gwmux))
	return nil
}

func (s *DataflowHttpServer) registerRPCGateWayHandler(
	ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn,
) error {
	if err := dataflow.RegisterChangefeedServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterDPFeatureServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterDataPipelineServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterSinkEndpointServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := premiumdmpb.RegisterPremiumMigrationServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := importintopb.RegisterImportIntoServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := changefeedopenpb.RegisterChangefeedServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	return dataflow.RegisterChangefeedSpecServiceHandler(ctx, mux, conn)
}