package impl

import (
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/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
}

func NewDownstreamAdaptorFactory(
	cdcPreCheckCli regionalapi.TiCDCPreCheckClient,
	encryptor encryptor_wrapper.Encryptor,
	cdcClusterModel ticdc.ClusterModel,
) *DownstreamAdaptorFactory {
	return &DownstreamAdaptorFactory{
		mysql:  NewMySQLDownstreamAdaptor(cdcPreCheckCli, encryptor),
		kafka:  NewKafkaDownstreamAdaptor(cdcPreCheckCli, encryptor),
		pulsar: NewPulsarDownstreamAdaptor(cdcPreCheckCli, encryptor),
		s3:     NewS3DownstreamAdaptor(cdcPreCheckCli, encryptor, cdcClusterModel),
		gcs:    NewGCSDownstreamAdaptor(cdcPreCheckCli),
	}
}

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

func (m *DownstreamAdaptorFactory) Get(
	downstreamType dataflow.ChangefeedType,
) (biz_changefeed.DownstreamAdaptor, bizerr.BizError) {
	switch downstreamType {
	case dataflow.ChangefeedType_MYSQL:
		return m.mysql, nil
	case dataflow.ChangefeedType_KAFKA:
		return m.kafka, nil
	case dataflow.ChangefeedType_PULSAR:
		return m.pulsar, nil
	case dataflow.ChangefeedType_S3:
		return m.s3, nil
	case dataflow.ChangefeedType_GCS:
		return m.gcs, nil
	default:
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported downstream type %s", dataflow.ChangefeedType_name[int32(downstreamType)])
	}
}