package bootstrap

import (
	"context"

	"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/metrics"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"
)

type DataflowGrpcServer struct {
	sever        *server.Server
	grpcServer   *server.GrpcServer
	metricsSerer *server.MetricsServer
}

func NewDataflowGrpcServer() (*DataflowGrpcServer, error) {
	s := &DataflowGrpcServer{}
	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.registerInnerAPI})
	if err != nil {
		logger.Error("init grpc server error", zap.Error(err))
		return nil, err
	}
	s.grpcServer = grpcServer

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

func (s *DataflowGrpcServer) RunDataflowGrpcServer() error {
	ctx := context.Background()
	eg := errgroup.Group{}
	eg.Go(func() error {
		return s.grpcServer.ServerGrpc(ctx)
	})
	eg.Go(func() error {
		return s.metricsSerer.ServerMetrics(ctx)
	})
	return eg.Wait()
}

func (s *DataflowGrpcServer) registerInnerAPI(rpcServer *grpc.Server) {
	dataflow.RegisterOpsInternalServiceServer(rpcServer, container.Global().OpsInnerController)

	dataflow.RegisterChangefeedServiceServer(rpcServer, container.Global().BizChangefeedController)
	dataflow.RegisterDataPipelineServiceServer(rpcServer, container.Global().DataPipelineController)
	changefeedopenpb.RegisterChangefeedServiceServer(rpcServer, container.Global().OpenChangefeedAPI)

	importintopb.RegisterImportIntoServiceServer(rpcServer, container.Global().ImportIntoAPI)

	premiumdmpb.RegisterPremiumMigrationServiceServer(rpcServer, container.Global().PremiumDMAPI)
}