package ticdc

import (
	"context"
	"fmt"

	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

type dedicatedBusinessClient interface {
	ListConnectors(ctx context.Context, in *dataflow.ConnectorListReq, opts ...grpc.CallOption) (*dataflow.Connectors, error)
	GetConnector(ctx context.Context, in *dataflow.ConnectorID, opts ...grpc.CallOption) (*dataflow.Connector, error)
}

type dedicatedInternalClient interface {
	ListInternalChangefeeds(ctx context.Context, in *dataflow.ListInternalChangefeedsRequest, opts ...grpc.CallOption) (*dataflow.ListInternalChangefeedsResponse, error)
	GetInternalChangefeed(ctx context.Context, in *dataflow.GetInternalChangefeedRequest, opts ...grpc.CallOption) (*dataflow.InternalChangefeed, error)
}

type nextGenBusinessClient interface {
	ListChangefeeds(ctx context.Context, in *dataflowng.ChangefeedListReq, opts ...grpc.CallOption) (*dataflowng.Changefeeds, error)
	GetChangefeed(ctx context.Context, in *dataflowng.ChangefeedID, opts ...grpc.CallOption) (*dataflowng.Changefeed, error)
}

type nextGenInternalClient interface {
	ListInternalChangefeeds(ctx context.Context, in *dataflowng.ListInternalChangefeedsRequest, opts ...grpc.CallOption) (*dataflowng.ListInternalChangefeedsResponse, error)
	GetInternalChangefeed(ctx context.Context, in *dataflowng.GetInternalChangefeedRequest, opts ...grpc.CallOption) (*dataflowng.InternalChangefeed, error)
}

type essentialChangefeedClient interface {
	ListClusterChangefeeds(ctx context.Context, in *serverlessrpc.ListClusterChangefeedsRequest, opts ...grpc.CallOption) (*serverlessrpc.ListClusterChangefeedsResponse, error)
	GetChangefeed(ctx context.Context, in *serverlessrpc.GetChangefeedRequest, opts ...grpc.CallOption) (*serverlessrpc.ChangefeedInfo, error)
}

type Clients struct {
	dedicatedBusiness dedicatedBusinessClient
	dedicatedInternal dedicatedInternalClient
	nextGenBusiness   nextGenBusinessClient
	nextGenInternal   nextGenInternalClient
	essential         essentialChangefeedClient
}

func NewClients(dataflowAddr, dataflowNGAddr, serverlessAddr string) (*Clients, error) {
	dialOptions := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
	}
	dedicatedConn, err := grpc.NewClient(dataflowAddr, dialOptions...)
	if err != nil {
		return nil, fmt.Errorf("create Dataflow client: %w", err)
	}
	nextGenConn, err := grpc.NewClient(dataflowNGAddr, dialOptions...)
	if err != nil {
		_ = dedicatedConn.Close()
		return nil, fmt.Errorf("create Dataflow-NG client: %w", err)
	}
	serverlessConn, err := grpc.NewClient(serverlessAddr, dialOptions...)
	if err != nil {
		_ = dedicatedConn.Close()
		_ = nextGenConn.Close()
		return nil, fmt.Errorf("create Serverless client: %w", err)
	}
	return &Clients{
		dedicatedBusiness: dataflow.NewConnectorServiceClient(dedicatedConn),
		dedicatedInternal: dataflow.NewOpsInternalServiceClient(dedicatedConn),
		nextGenBusiness:   dataflowng.NewChangefeedServiceClient(nextGenConn),
		nextGenInternal:   dataflowng.NewOpsInternalServiceClient(nextGenConn),
		essential:         serverlessrpc.NewChangefeedOpsServiceClient(serverlessConn),
	}, nil
}