package dataflowcli

import (
	"strings"
	"time"

	"github.com/sercand/kuberesolver/v5"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	tigrpcclient "github.com/tidbcloud/tiinf/transport/tigrpc/client"
	"google.golang.org/grpc"

	dataflowcli "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dataflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/config"
	grpcutil "github.com/tidbcloud/tidb-management-service/utilities/grpc"
)

type TimeoutCallOption struct {
	grpc.EmptyCallOption
}

func NewClient(cfg *config.DataFlowService) (dataflowcli.DataFlow, error) {
	// if cfg == nil {
	// 	return nil, errtrace.Wrap(errors.New("config is nil"))
	// }

	if cfg == nil || cfg.Local || cfg.Url == "" {
		return &fakeImpl{}, nil
	}
	var (
		conn *grpc.ClientConn
		err  error
	)

	interceptors := tigrpcclient.WithUnaryInterceptors(
		grpcutil.ClientPayloadLogInterceptor,
		grpcutil.TimeoutInterceptor(time.Duration(cfg.WaitTimeoutInMilliSec)*time.Millisecond),
	)

	opts := []tigrpcclient.Option{tigrpcclient.WithPrometheus(), tigrpcclient.WithTracing(), interceptors}
	if strings.HasPrefix(cfg.Url, "kubernetes://") {
		kuberesolver.RegisterInCluster()
		opts = append(opts, tigrpcclient.WithLoadBalance())
	}

	// More about timeout:
	// We have a server option to set a timeout on connection establishment: https://pkg.go.dev/google.golang.org/grpc#ConnectionTimeout
	//
	// We have ways to configure keepalive parameters on the server: https://pkg.go.dev/google.golang.org/grpc@v1.43.0/keepalive#ServerParameters
	//
	// And if we wish to set timeouts on a per-RPC level, you could register a server interceptor using https://pkg.go.dev/google.golang.org/grpc#UnaryInterceptor and create a new context in there with a different timeout and pass it to the handler.

	if conn, err = tigrpcclient.NewClientConn(cfg.Url, opts...); err != nil {
		return nil, err
	}

	backupInternalCli := dataflow.NewBackupInternalServiceClient(conn)
	connectorCli := dataflow.NewConnectorServiceClient(conn)
	changefeedInstanceTypeCli := dataflow.NewChangefeedInstanceTypeServiceClient(conn)
	sinkEndpointCli := dataflow.NewSinkEndpointServiceClient(conn)

	c := clientImpl{
		backupInternalCli:         backupInternalCli,
		connectorCli:              connectorCli,
		changefeedInstanceTypeCli: changefeedInstanceTypeCli,
		sinkEndpointCli:           sinkEndpointCli,
	}
	return &c, nil
}