package dataflowcli

import (
	"context"
	"os"
	"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"
	"google.golang.org/protobuf/proto"

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

type TimeoutCallOption struct {
	grpc.EmptyCallOption
}

// clientPayloadLogInterceptor keeps the standard payload logging for generated
// protobuf messages. Compatibility wire messages intentionally implement the
// legacy protobuf interface so this client can call a newer Dataflow API before
// its generated dependency is published; the shared logger only accepts v2
// messages and would panic before invoking the RPC.
func clientPayloadLogInterceptor(
	ctx context.Context,
	method string,
	req, reply any,
	cc *grpc.ClientConn,
	invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption,
) error {
	if _, ok := req.(proto.Message); !ok {
		return invoker(ctx, method, req, reply, cc, opts...)
	}
	return grpcutil.ClientPayloadLogInterceptor(ctx, method, req, reply, cc, invoker, opts...)
}

func NewClient(cfg *config.Config) (dataflowcli.DataFlow, error) {
	var (
		conn *grpc.ClientConn
		err  error
	)

	dfConf := cfg.ServerConfig.DependentSvcAddresses.DataFlow

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

	opts := []tigrpcclient.Option{tigrpcclient.WithPrometheus(), tigrpcclient.WithTracing(), interceptors}
	// If mirrord is executing, we are in a local run, so we should not use kubernetes resolver
	isLocalRun := os.Getenv("MIRRORD_EXECUTING") == "true"
	if strings.HasPrefix(dfConf.Url, "kubernetes://") && !isLocalRun {
		kuberesolver.RegisterInCluster()
		opts = append(opts, tigrpcclient.WithLoadBalance())
	}

	if conn, err = tigrpcclient.NewClientConn(dfConf.Url, opts...); err != nil {
		return nil, err
	}
	dataflowCli := dataflow.NewBackupInternalServiceClient(conn)

	c := clientImpl{BackupInternalServiceClient: dataflowCli, conn: conn}
	return &c, nil
}