package grpcli

import (
	"context"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type ChangefeedGrpcClient interface {
	GetClient() dataflow.ConnectorServiceClient

	CreateChangefeed(ctx context.Context, req *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, bizerr.BizError)
	DeleteChangefeed(ctx context.Context, req *dataflow.ConnectorID) bizerr.BizError
	StartChangefeed(ctx context.Context, req *dataflow.ConnectorID) bizerr.BizError
	StopChangefeed(ctx context.Context, req *dataflow.ConnectorID) bizerr.BizError
	GetChangefeed(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, bizerr.BizError)
}

type changefeedGrpcClientImpl struct {
	addr   string
	client dataflow.ConnectorServiceClient
}

func NewChangefeedGrpcClient(ctx context.Context, addr string) ChangefeedGrpcClient {
	logger := log.GetLogger(ctx)
	logger.Info("initializing import client", zap.String("addr", addr))

	conn := MustBuildGrpcClientConnection(ctx, addr)
	coreCli := dataflow.NewConnectorServiceClient(conn)
	return &changefeedGrpcClientImpl{
		addr:   addr,
		client: coreCli,
	}
}

func (c *changefeedGrpcClientImpl) GetClient() dataflow.ConnectorServiceClient {
	return c.client
}

func (c *changefeedGrpcClientImpl) CreateChangefeed(ctx context.Context, req *dataflow.ConnectorCreateReq,
) (*dataflow.ConnectorID, bizerr.BizError) {
	resp, e := c.client.CreateConnector(ctx, req)
	if e != nil {
		return nil, bizerr.WrapGrpcError(e, "CreateChangefeed")
	}
	return resp, nil
}

func (c *changefeedGrpcClientImpl) DeleteChangefeed(ctx context.Context, req *dataflow.ConnectorID,
) bizerr.BizError {
	_, e := c.client.DeleteConnector(ctx, req)
	if e != nil {
		return bizerr.WrapGrpcError(e, "DeleteChangefeed")
	}
	return nil
}

func (c *changefeedGrpcClientImpl) StartChangefeed(ctx context.Context, req *dataflow.ConnectorID,
) bizerr.BizError {
	_, e := c.client.StartConnector(ctx, req)
	if e != nil {
		return bizerr.WrapGrpcError(e, "StartChangefeed")
	}
	return nil
}

func (c *changefeedGrpcClientImpl) StopChangefeed(ctx context.Context, req *dataflow.ConnectorID,
) bizerr.BizError {
	_, e := c.client.StopConnector(ctx, req)
	if e != nil {
		return bizerr.WrapGrpcError(e, "StopChangefeed")
	}
	return nil
}

func (c *changefeedGrpcClientImpl) GetChangefeed(ctx context.Context, req *dataflow.ConnectorID,
) (*dataflow.Connector, bizerr.BizError) {
	resp, e := c.client.GetConnector(ctx, req)
	if e != nil {
		return nil, bizerr.WrapGrpcError(e, "GetChangefeed")
	}
	return resp, nil
}