package dataflowcli

import (
	"context"

	"braces.dev/errtrace"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
)

type clientImpl struct {
	backupInternalCli         dataflow.BackupInternalServiceClient
	connectorCli              dataflow.ConnectorServiceClient
	changefeedInstanceTypeCli dataflow.ChangefeedInstanceTypeServiceClient
	sinkEndpointCli           dataflow.SinkEndpointServiceClient
}

func (c clientImpl) CreateFinalBackupIgnoreExist(ctx context.Context, clusterID uint64) (uint64, error) {
	req := &dataflow.CreateFinalBackupReq{ClusterId: clusterID}
	resp, err := c.backupInternalCli.CreateFinalBackup(ctx, req)
	if err != nil {
		return 0, errtrace.Wrap(err)
	}

	return resp.BackupId, nil
}

func (c clientImpl) GetBackup(ctx context.Context, req *dataflow.GetBackupReq) (*dataflow.GetBackupResp, error) {
	return c.backupInternalCli.GetBackup(ctx, req)
}

func (c clientImpl) Restore(ctx context.Context, req *dataflow.BackupInternalRestoreReq) (*dataflow.RestoreResp, error) {
	resp, err := c.backupInternalCli.Restore(ctx, req)
	if err != nil {
		return nil, errtrace.Wrap(err)
	}
	return resp, nil
}

func (c clientImpl) GetRestore(ctx context.Context, req *dataflow.GetRestoreReq) (*dataflow.GetRestoreResp, error) {
	return c.backupInternalCli.GetRestore(ctx, req)
}

func (c clientImpl) CreateConnector(ctx context.Context, req *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, error) {
	return c.connectorCli.CreateConnector(ctx, req)
}

func (c clientImpl) GetConnector(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, error) {
	return c.connectorCli.GetConnector(ctx, req)
}

func (c clientImpl) ListConnectors(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) {
	return c.connectorCli.ListConnectors(ctx, req)
}

func (c clientImpl) DeleteConnector(ctx context.Context, req *dataflow.ConnectorID) error {
	_, err := c.connectorCli.DeleteConnector(ctx, req)
	return err
}

func (c clientImpl) ListChangefeedInstanceTypes(
	ctx context.Context,
	req *dataflow.ChangefeedInstanceTypeListReq,
) (*dataflow.ChangefeedInstanceTypes, error) {
	return c.changefeedInstanceTypeCli.List(ctx, req)
}

func (c clientImpl) CreateSinkEndpoint(ctx context.Context, req *dataflow.CreateSinkEndpointRequest) (*dataflow.SinkEndpoint, error) {
	return c.sinkEndpointCli.CreateSinkEndpoint(ctx, req)
}

func (c clientImpl) GetSinkEndpoint(ctx context.Context, req *dataflow.GetSinkEndpointRequest) (*dataflow.SinkEndpoint, error) {
	return c.sinkEndpointCli.GetSinkEndpoint(ctx, req)
}

func (c clientImpl) DeleteSinkEndpoint(ctx context.Context, req *dataflow.DeleteSinkEndpointRequest) error {
	_, err := c.sinkEndpointCli.DeleteSinkEndpoint(ctx, req)
	return err
}

func (c clientImpl) ListSinkAvailabilityZones(
	ctx context.Context,
	req *dataflow.ListSinkAvailabilityZonesRequest,
) (*dataflow.ListSinkAvailabilityZonesResponse, error) {
	return c.sinkEndpointCli.ListSinkAvailabilityZones(ctx, req)
}