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/app/container"
	importadminbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/adminapi/v1beta1"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"

	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	"github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	"github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server/auth"
	"github.com/tidbcloud/dataflow-service/internal/config"
	internalContainer "github.com/tidbcloud/dataflow-service/internal/container"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/idl/pbgen/proto/openapi/dp"
	accountAuth "github.com/tidbcloud/tiinf/sdk/auth"
	"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.registerOpenAPI,
		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) registerOpenAPI(rpcServer *grpc.Server) {
	s.registerBROpenAPI(rpcServer)
	s.registerImportOpenAPI(rpcServer)
	s.registerImportAdminOpenAPI(rpcServer)
	s.registerChangefeedOpenAPI(rpcServer)
}

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

func (s *DataflowHttpServer) registerImportOpenAPI(rpcServer *grpc.Server) {
	dp.RegisterImportServer(rpcServer, internalContainer.Global().ImportOpenAPI)
}

func (s *DataflowHttpServer) registerImportAdminOpenAPI(rpcServer *grpc.Server) {
	importadminbp.RegisterImportServiceServer(rpcServer, internalContainer.Global().ImportAdminAPI)
}

func (s *DataflowHttpServer) registerChangefeedOpenAPI(rpcServer *grpc.Server) {
	changefeedopenbp.RegisterChangefeedServiceServer(rpcServer, container.Global().ChangefeedOpenAPI)
}

func (s *DataflowHttpServer) registerBROpenAPI(rpcServer *grpc.Server) {
	dp.RegisterBackupServer(rpcServer, internalContainer.Global().BackupOpenAPI)
	dp.RegisterRestoreServer(rpcServer, container.Global().RestoreOpenController)
	importadminbp.RegisterBackupServiceServer(rpcServer, container.Global().BackupOpenControllerV2)
}

func (s *DataflowHttpServer) registerConsoleAPI(rpcServer *grpc.Server) {
	s.registerCommonConsoleAPI(rpcServer)
	s.registerImportDMConsoleAPI(rpcServer)
	s.registerBRConsoleAPI(rpcServer)
	s.registerChangefeedConsoleAPI(rpcServer)
	s.registerPrivateLinkConsoleAPI(rpcServer)
}

func (s *DataflowHttpServer) registerCommonConsoleAPI(rpcServer *grpc.Server) {
	dataflow.RegisterDPFeatureServiceServer(rpcServer, internalContainer.Global().DPFeatureConsoleAPI)
}

func (s *DataflowHttpServer) registerChangefeedConsoleAPI(rpcServer *grpc.Server) {
	dataflow.RegisterChangefeedInstanceTypeServiceServer(
		rpcServer, container.Global().BizChangefeedSpecController)
	dataflow.RegisterConnectorServiceServer(rpcServer, container.Global().BizChangefeedController)
}

func (s *DataflowHttpServer) registerPrivateLinkConsoleAPI(rpcServer *grpc.Server) {
	dataflow.RegisterSinkEndpointServiceServer(rpcServer, container.Global().SinkEndpointController)
}

func (s *DataflowHttpServer) registerBRConsoleAPI(rpcServer *grpc.Server) {
	// v1
	dataflow.RegisterBackupServiceServer(rpcServer, internalContainer.Global().BackupConsoleAPI)
	// v2
	dataflow.RegisterBackupConsoleServiceServer(rpcServer, container.Global().BackupControllerV2)
	// CRRF
	dataflow.RegisterCRRFConsoleServiceServer(rpcServer, container.Global().CRRFController)
}

func (s *DataflowHttpServer) registerImportDMConsoleAPI(rpcServer *grpc.Server) {
	dmpb.RegisterServiceServer(rpcServer, internalContainer.Global().DMConsoleAPI)
	dataflow.RegisterImportServiceServer(rpcServer, internalContainer.Global().ImportConsoleAPI)
	dataflow.RegisterConnProfileServiceServer(rpcServer, internalContainer.Global().ConnProfileConsoleAPI)
}

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

	// Add explicit route to avoid Gin's default 301 redirect from `/v1beta1/backups` to `/v1beta1/backups/`
	registerOpenAPIV1beta1PassthroughRoutes(eng, gwmuvV1beta1)

	eng.Any("/api/v1/orgs/*any", gin.WrapH(gwmux))
	eng.Any("/api/v2/orgs/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1/admin/orgs/*any", gin.WrapH(gwmux))
	eng.Any("/external-api/v1/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1/dataflow/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1/changefeed/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1/dw_replication/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1beta/projects/*any", gin.WrapH(gwmux))
	eng.Any("/api/v1beta1/*any", gin.WrapH(gwmux))
	v1beta2API := eng.Group("/v1beta2")
	v1beta2API.Use(
		func(c *gin.Context) {
			ctx := log.WithLogID(context.Background())
			logger := log.GetLogger(ctx)
			err = auth.SetAuthOrgIdWithTiDBId(c)
			if err != nil {
				logger.Warn("set auth orgId with TiDB failed", zap.Error(err))
			}
		},
		accountAuth.GinGetIdentityInfoInterceptor(&accountAuth.IdentityAuthOption{
			EnableLog: true,
		}),
		accountAuth.GinPermissionCheckInterceptor(&accountAuth.PermissionCheckerCommonParam{
			IdentityF:               accountAuth.GinGetIdentityFunc,
			PathPattern2ResourceMap: config.AuthSDKRBACPathPattern2ResourceMap,
			GenAccessParam4GinF:     accountAuth.GenAccessParam4GinWithPrefixOrgId,
		}),
	)
	v1beta2API.Any("*any", gin.WrapH(gwmuvV1beta1))
	v1beta2UIAPI := eng.Group("/api/v2025/cloud/v1beta2")
	v1beta2UIAPI.Use(
		func(c *gin.Context) {
			c.Request.URL.Path = strings.TrimPrefix(c.Request.URL.Path, "/api/v2025/cloud")
			ctx := log.WithLogID(context.Background())
			logger := log.GetLogger(ctx)
			err = auth.SetAuthOrgIdWithTiDBId(c)
			if err != nil {
				logger.Warn("set auth orgId with TiDB failed", zap.Error(err))
			}
			c.Next()
		},
		accountAuth.GinGetIdentityInfoInterceptor(&accountAuth.IdentityAuthOption{
			EnableLog: true,
		}),
		accountAuth.GinPermissionCheckInterceptor(&accountAuth.PermissionCheckerCommonParam{
			IdentityF:               accountAuth.GinGetIdentityFunc,
			PathPattern2ResourceMap: config.AuthSDKRBACPathPattern2ResourceMap,
			GenAccessParam4GinF:     accountAuth.GenAccessParam4GinWithPrefixOrgId,
		}),
	)
	v1beta2UIAPI.Any("*any", gin.WrapH(gwmuvV1beta1))
	if err := s.registerInnerRequestMappingForE2E(ctx, eng, gwmux, conn); err != nil {
		return err
	}
	return nil
}

func registerOpenAPIV1beta1PassthroughRoutes(eng *gin.Engine, gateway http.Handler) {
	eng.Any("/v1beta1/backups", gin.WrapH(gateway))
	eng.Any("/v1beta1/backups/*any", gin.WrapH(gateway))
	eng.Any("/v1beta1/changefeeds", gin.WrapH(gateway))
	eng.Any("/v1beta1/changefeeds/*any", gin.WrapH(gateway))
	eng.Any("/v1beta1/changefeedRCUs", gin.WrapH(gateway))
	eng.Any("/v1beta1/clusters/*any", 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,
	_ *grpcruntime.ServeMux,
	conn *grpc.ClientConn,
) error {
	if err := dataflow.RegisterOpsInternalServiceHandler(ctx, gwmux, conn); err != nil {
		return err
	}
	eng.Any("/api/v1/dataflow/ops/*any", gin.WrapH(gwmux))
	return nil
}

func (s *DataflowHttpServer) registerRPCGateWayHandler(
	ctx context.Context, mux *grpcruntime.ServeMux, muxV1beta1 *grpcruntime.ServeMux, conn *grpc.ClientConn,
) error {
	err := s.registerOpenAPIGateWayHandler(ctx, mux, conn)
	if err != nil {
		return err
	}
	if err = s.registerOpenAPIV1beta1GateWayHandler(ctx, muxV1beta1, conn); err != nil {
		return err
	}
	return s.registerConsoleAPIGateWayHandler(ctx, mux, conn)
}

func (s *DataflowHttpServer) registerConsoleAPIGateWayHandler(
	ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn,
) error {
	if err := dataflow.RegisterConnectorServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterBackupServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterBackupConsoleServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterCRRFConsoleServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterDPFeatureServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterConnProfileServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dmpb.RegisterServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterChangefeedInstanceTypeServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dataflow.RegisterSinkEndpointServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	return dataflow.RegisterImportServiceHandler(ctx, mux, conn)
}

func (s *DataflowHttpServer) registerOpenAPIGateWayHandler(
	ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn,
) error {
	if err := dp.RegisterRestoreHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dp.RegisterBackupHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := dp.RegisterImportHandler(ctx, mux, conn); err != nil {
		return err
	}
	return nil
}

func (s *DataflowHttpServer) registerOpenAPIV1beta1GateWayHandler(
	ctx context.Context, mux *grpcruntime.ServeMux, conn *grpc.ClientConn,
) error {
	if err := importadminbp.RegisterBackupServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := importadminbp.RegisterImportServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	if err := changefeedopenbp.RegisterChangefeedServiceHandler(ctx, mux, conn); err != nil {
		return err
	}
	return nil
}