package cvt

import (
	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"

	"google.golang.org/protobuf/types/known/structpb"
)

type toDataflow struct{}
type fromDataflow struct{}

func (c toDataflow) NamedParameter(item ccent.NamedParameter) (*dataflow.ParamItem, error) {
	val, err := structpb.NewValue(item.Value)
	if err != nil {
		return nil, err
	}

	return &dataflow.ParamItem{
		Value:   val,
		Comment: item.Comment,
	}, nil
}
func (c toDataflow) ClusterType(typ ccent.ClusterType) dataflow.ConfigClusterType {
	m := map[ccent.ClusterType]dataflow.ConfigClusterType{
		ccent.ClusterTypeDedicatedTiDB: dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
		ccent.ClusterTypeTiCDC:         dataflow.ConfigClusterType_TICDC_CLUSTER,
		ccent.ClusterTypeChangefeed:    dataflow.ConfigClusterType_CHANGEFEED_CLUSTER,
	}
	t, exist := m[typ]
	if !exist {
		return dataflow.ConfigClusterType_CONFIG_CLUSTER_TYPE_UNSPECIFIED
	}
	return t
}

func (c toDataflow) ConfigType(configType string) dataflow.ConfigFileType {
	m := map[string]dataflow.ConfigFileType{
		ccproto.DEDICATED_TIDB_CONFIG_TICDC:                 dataflow.ConfigFileType_TICDC_FILE,
		ccproto.DEDICATED_TIDB_CONFIG_RESTORE:               dataflow.ConfigFileType_RESTORE_FILE,
		ccproto.DEDICATED_TIDB_CONFIG_BACKUP:                dataflow.ConfigFileType_BACKUP_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_KAFKA:          dataflow.ConfigFileType_CHANGEFEED_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_MYSQL:          dataflow.ConfigFileType_CHANGEFEED_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_S3:             dataflow.ConfigFileType_CHANGEFEED_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_GCS:            dataflow.ConfigFileType_CHANGEFEED_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_SERVERLESS:     dataflow.ConfigFileType_CHANGEFEED_FILE,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_PULSAR:         dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_PULSAR:     dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_KAFKA:      dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_SERVERLESS: dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_GCS:        dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_S3:         dataflow.ConfigFileType_SINK_URI,
		ccproto.CHANGEFEED_CONFIG_CHANGEFEED_URI_MYSQL:      dataflow.ConfigFileType_SINK_URI,
	}
	t, exist := m[configType]
	if !exist {
		return dataflow.ConfigFileType_CONFIG_FILE_TYPE_UNSPECIFIED
	}
	return dataflow.ConfigFileType(t)
}

func (c toDataflow) Component(comp string) dataflow.ConfigFileType {
	m := map[string]dataflow.ConfigFileType{
		ccproto.DEDICATED_TIDB_COMPONENT_TICDC:   dataflow.ConfigFileType_TICDC_FILE,
		ccproto.DEDICATED_TIDB_COMPONENT_RESTORE: dataflow.ConfigFileType_RESTORE_FILE,
		ccproto.DEDICATED_TIDB_COMPONENT_BACKUP:  dataflow.ConfigFileType_BACKUP_FILE,
		// use OnlyConfigCenterClusterApplier instead
		// ccproto.DEDICATED_TIDB_COMPONENT_COMPACT_BACKUP_BR:   dataflow.ConfigFileType_COMPACT_BACKUP_BR_FILE,
		// ccproto.DEDICATED_TIDB_COMPONENT_COMPACT_BACKUP_TIKV: dataflow.ConfigFileType_COMPACT_BACKUP_TIKV_FILE,
	}
	t, exist := m[comp]
	if !exist {
		return dataflow.ConfigFileType_CONFIG_FILE_TYPE_UNSPECIFIED
	}
	return dataflow.ConfigFileType(t)
}