package grpcli

import (
	"context"
	"fmt"
	"sync"

	"github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/logid"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"google.golang.org/grpc"
)

var (
	dataflowReplicationSvc sync.Once
	dataflowReplicationCli dataflow.ReplicationServiceClient
)

func getDataflowReplicationSvcClient() dataflow.ReplicationServiceClient {
	dataflowReplicationSvc.Do(func() {
		conn, err := grpc.DialContext(
			context.Background(),
			config.Global().GrpcConfig.DataflowSvcAddr,
			grpc.WithInsecure(),
			grpc.WithChainUnaryInterceptor(
				logid.UnaryClientInterceptor(),
				otelgrpc.UnaryClientInterceptor(),
			),
		)
		if err != nil {
			panic(fmt.Sprintf("dial dataflow-svc: %s", err))
		}

		dataflowReplicationCli = dataflow.NewReplicationServiceClient(conn)
	})

	return dataflowReplicationCli
}

func CreateReplication(ctx context.Context, req *dataflow.CreateReplicationReq) (*dataflow.CreateReplicationResp, bizerr.BizError) {
	client := getDataflowReplicationSvcClient()

	resp, e := client.CreateReplication(ctx, req)
	if e != nil {
		return resp, bizerr.WrapGrpcError(e, "CreateReplication")
	}
	if resp.BaseResp.ErrCode != 0 {
		return resp, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp, nil
}