package bootstrap

import (
	"context"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/replay"
	"github.com/tidbcloud/dataflow-service/internal/cmd/bootstrap/server"
	internalContainer "github.com/tidbcloud/dataflow-service/internal/container"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/dataflow-service/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.RegisterBackupInternalServiceServer(rpcServer, &container.Global().BackupInnerController)
	dataflow.RegisterImportInnerServiceServer(rpcServer,
		internalContainer.Global().ImportInnerAPI)
	dataflow.RegisterOpsInternalServiceServer(rpcServer, container.Global().OpsInnerController)

	dataflow.RegisterConnectorServiceServer(rpcServer, container.Global().BizChangefeedController)

	// because serverless-svc use the grpc, he pass the user json to invoke grpc of console api....
	// so just register the grpc api
	dataflow.RegisterImportServiceServer(rpcServer, internalContainer.Global().ImportConsoleAPI)

	// export DM API for OPS calling.
	dmpb.RegisterServiceServer(rpcServer, internalContainer.Global().DMConsoleAPI)
	// export Backup API for OPS calling.
	// v1
	dataflow.RegisterBackupServiceServer(rpcServer, internalContainer.Global().BackupConsoleAPI)
	// v2
	dataflow.RegisterBackupConsoleServiceServer(rpcServer, container.Global().BackupControllerV2)

	dataflow.RegisterSinkEndpointServiceServer(rpcServer, container.Global().SinkEndpointController)

	replay.RegisterReplayServiceServer(rpcServer, container.Global().ReplayInnerController)
}