package impl

import (
	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type DownstreamAdaptorFactory struct {
	mysql      biz_changefeed.DownstreamAdaptor
	kafka      biz_changefeed.DownstreamAdaptor
	pulsar     biz_changefeed.DownstreamAdaptor
	s3         biz_changefeed.DownstreamAdaptor
	gcs        biz_changefeed.DownstreamAdaptor
	azBlob     biz_changefeed.DownstreamAdaptor
	serverless biz_changefeed.DownstreamAdaptor
}

func NewDownstreamAdaptorFactory(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
	cdcClusterModel ticdc.ClusterModel,
	encryptor encryptor_wrapper.Encryptor,
	privatelinkEndpointRepo repo.BizPrivateLinkEndpointRepo,
	coreGrpcClient grpcli.CoreGrpcClient,
	tidbClusterInfraCli grpcli.TiDBClusterInfraClient,
) *DownstreamAdaptorFactory {
	return &DownstreamAdaptorFactory{
		mysql:  NewMySQLDownstreamAdaptor(cdcPreCheckCli, encryptor),
		kafka:  NewKafkaDownstreamAdaptor(cdcPreCheckCli, encryptor),
		pulsar: NewPulsarDownstreamAdaptor(cdcPreCheckCli, encryptor),
		s3:     NewS3DownstreamAdaptor(cdcPreCheckCli, cdcClusterModel, encryptor),
		gcs:    NewGCSDownstreamAdaptor(cdcPreCheckCli, tidbClusterInfraCli),
		azBlob: NewAzureBlobStorageDownstreamAdaptor(cdcPreCheckCli, encryptor),
		serverless: NewServerlessDownstreamAdaptor(
			cdcPreCheckCli, encryptor, privatelinkEndpointRepo, coreGrpcClient),
	}
}

func NewMockDownstreamAdaptorFactory(mockMySQL biz_changefeed.DownstreamAdaptor) *DownstreamAdaptorFactory {
	return &DownstreamAdaptorFactory{
		mysql: mockMySQL,
	}
}

func (m *DownstreamAdaptorFactory) Get(
	downstreamType dataflow.ConnectorType,
) (biz_changefeed.DownstreamAdaptor, bizerr.BizError) {
	switch downstreamType {
	case dataflow.ConnectorType_MYSQL:
		return m.mysql, nil
	case dataflow.ConnectorType_KAFKA:
		return m.kafka, nil
	case dataflow.ConnectorType_PULSAR:
		return m.pulsar, nil
	case dataflow.ConnectorType_S3:
		return m.s3, nil
	case dataflow.ConnectorType_GCS:
		return m.gcs, nil
	case dataflow.ConnectorType_AZURE_BLOB_STORAGE:
		return m.azBlob, nil
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return m.serverless, nil
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported downstream type %s", dataflow.ConnectorType_name[int32(downstreamType)])
	}
}