package console

import (
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/planner"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func createDataPipelineRequestDataflow2Domain(
	req *dataflow.CreateDataPipelineRequest,
	userID uint64,
	userEmail string,
) *domain.CreateDataPipelineReq {
	return &domain.CreateDataPipelineReq{
		ClusterID:               req.GetClusterId(),
		UserID:                  userID,
		UserEmail:               userEmail,
		Name:                    req.GetName(),
		Source:                  dataPipelineSourceInputDataflow2Domain(req.GetSource()),
		Destination:             dataPipelineDestinationInputDataflow2Domain(req.GetDestination()),
		Staging:                 dataPipelineStagingInputDataflow2Domain(req.GetStaging()),
		SyncMode:                dataPipelineSyncModeDataflow2Domain(req.GetSyncMode()),
		ChangefeedSpecification: req.GetChangefeedSpecification(),
	}
}

func validateDataPipelineDestinationRequestDataflow2Domain(
	req *dataflow.ValidateDataPipelineDestinationRequest,
	userID uint64,
) *domain.ValidateDataPipelineDestinationReq {
	return &domain.ValidateDataPipelineDestinationReq{
		ClusterID:   req.GetClusterId(),
		UserID:      userID,
		SyncMode:    dataPipelineSyncModeDataflow2Domain(req.GetSyncMode()),
		Destination: dataPipelineDestinationInputDataflow2Domain(req.GetDestination()),
	}
}

func scaleDataPipelineRequestDataflow2Domain(
	req *dataflow.ScaleDataPipelineRequest,
	userID uint64,
) *domain.ScaleDataPipelineReq {
	return &domain.ScaleDataPipelineReq{
		ClusterID:               req.GetClusterId(),
		PipelineID:              req.GetPipelineId(),
		UserID:                  userID,
		ChangefeedSpecification: req.GetChangefeedSpecification(),
	}
}

func editDataPipelineRequestDataflow2Domain(
	req *dataflow.EditDataPipelineRequest,
	userID uint64,
) *domain.EditDataPipelineReq {
	result := &domain.EditDataPipelineReq{
		ClusterID:  req.GetClusterId(),
		PipelineID: req.GetPipelineId(),
		UserID:     userID,
		Name:       req.GetName(),
	}
	if source := req.GetSource(); source != nil {
		result.Source = domain.DataPipelineEditableSourceInput{
			TableFilter:   append([]string(nil), source.GetTableFilter()...),
			CaseSensitive: source.GetCaseSensitive(),
		}
	}
	if destination := req.GetDestination(); destination != nil {
		result.Destination = domain.DataPipelineEditableDestinationInput{
			Warehouse:            destination.GetWarehouse(),
			TargetNaming:         dataPipelineTargetNamingDataflow2Domain(destination.GetTargetNaming()),
			FlushIntervalSeconds: destination.GetFlushIntervalSeconds(),
		}
	}
	// Absent detaches the queue, like every other absent field in this
	// full-replacement payload.
	if staging := req.GetStaging(); staging != nil {
		result.Staging = domain.DataPipelineEditableStagingInput{
			SQSQueueURL: staging.GetSqsQueueUrl(),
			SQSRegion:   staging.GetSqsRegion(),
		}
	}
	return result
}

func dataPipelineSourceInputDataflow2Domain(
	source *dataflow.DataPipelineSourceInput,
) domain.DataPipelineSourceInput {
	if source == nil {
		return domain.DataPipelineSourceInput{}
	}
	return domain.DataPipelineSourceInput{
		Username:      source.GetUsername(),
		Password:      source.GetPassword(),
		TableFilter:   append([]string(nil), source.GetTableFilter()...),
		CaseSensitive: source.GetCaseSensitive(),
	}
}

func dataPipelineDestinationInputDataflow2Domain(
	destination *dataflow.DataPipelineDestination,
) domain.DataPipelineDestinationInput {
	if destination == nil {
		return domain.DataPipelineDestinationInput{}
	}
	return domain.DataPipelineDestinationInput{
		Type:                 dataPipelineDestinationTypeDataflow2Domain(destination.GetType()),
		LakeOrgID:            destination.GetLakeOrgId(),
		Warehouse:            destination.GetWarehouse(),
		TargetNaming:         dataPipelineTargetNamingDataflow2Domain(destination.GetTargetNaming()),
		FlushIntervalSeconds: destination.GetFlushIntervalSeconds(),
	}
}

func dataPipelineTargetNamingDataflow2Domain(
	targetNaming *dataflow.DataPipelineTargetNaming,
) domain.DataPipelineTargetNaming {
	if targetNaming == nil {
		return domain.DataPipelineTargetNaming{}
	}
	return domain.DataPipelineTargetNaming{
		DatabasePrefix: targetNaming.GetDatabasePrefix(),
		DatabaseSuffix: targetNaming.GetDatabaseSuffix(),
		TablePrefix:    targetNaming.GetTablePrefix(),
		TableSuffix:    targetNaming.GetTableSuffix(),
	}
}

func dataPipelineStagingInputDataflow2Domain(
	staging *dataflow.DataPipelineStagingInput,
) domain.DataPipelineStagingInput {
	if staging == nil {
		return domain.DataPipelineStagingInput{}
	}
	result := domain.DataPipelineStagingInput{
		Type: dataPipelineStagingTypeDataflow2Domain(staging.GetType()),
	}
	if s3 := staging.GetS3(); s3 != nil {
		result.S3 = &domain.DataPipelineS3StagingInput{
			URI:             s3.GetUri(),
			Region:          s3.GetRegion(),
			Endpoint:        s3.GetEndpoint(),
			AuthType:        dataPipelineS3AuthTypeDataflow2Domain(s3.GetAuthType()),
			AccessKeyID:     s3.GetAccessKeyId(),
			SecretAccessKey: s3.GetSecretAccessKey(),
			RoleARN:         s3.GetRoleArn(),
			SQSQueueURL:     s3.GetSqsQueueUrl(),
			SQSRegion:       s3.GetSqsRegion(),
			Provider:        dataPipelineS3ProviderDataflow2Domain(s3.GetProvider()),
		}
	}
	return result
}

func dataPipelineDomain2Dataflow(details *domain.DataPipelineDetails) *dataflow.DataPipeline {
	pipeline := details.Pipeline
	return &dataflow.DataPipeline{
		PipelineId:              pipeline.ID,
		ClusterId:               pipeline.ClusterID,
		Name:                    pipeline.Name,
		Source:                  dataPipelineSourceDomain2Dataflow(pipeline),
		Destination:             dataPipelineDestinationDomain2Dataflow(pipeline),
		Staging:                 dataPipelineStagingDomain2Dataflow(pipeline),
		SyncMode:                dataPipelineSyncModeDomain2Dataflow(pipeline.Params.SyncMode),
		ChangefeedSpecification: pipeline.ChangefeedSpecification,
		Status:                  dataPipelineStatusDomain2Dataflow(pipeline.Status),
		ExecutionPlan:           dataPipelineExecutionPlanDomain2Dataflow(pipeline),
		DeliveryIntervalPlan:    dataPipelineIntervalPlanDomain2Dataflow(pipeline.IntervalPlan),
		ChildResources:          dataPipelineChildrenDomain2Dataflow(details),
		Observation:             dataPipelineObservationDomain2Dataflow(details.Observation),
		Error:                   dataPipelineErrorDomain2Dataflow(pipeline.LastError),
		CreatedAt:               dataPipelineTime2Dataflow(&pipeline.CreatedAt),
		UpdatedAt:               dataPipelineTime2Dataflow(pipeline.UpdatedAt),
		Phase:                   dataPipelinePhaseDomain2Dataflow(pipeline),
		CreatorId:               pipeline.CreatorID,
		Creator:                 pipeline.Creator,
	}
}

// dataPipelinePhaseDomain2Dataflow derives the Console-facing phase. The
// orchestration stage stays internal: it names Fastflow actions, changes
// whenever the workflow is restructured, and must not become an API contract.
func dataPipelinePhaseDomain2Dataflow(
	pipeline *domain.DataPipeline,
) dataflow.DataPipelinePhase {
	switch pipeline.Status {
	case domain.DataPipelineStatusCreating:
		// Child identities are settled when the row is written, so they say
		// nothing about progress. What has actually been observed does, and it
		// needs nothing kept in sync.
		switch {
		case pipeline.Observation.LakeIntegrationStatus != "":
			return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_CREATING_DESTINATION
		case pipeline.Observation.ExportStatus != "":
			return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_SNAPSHOTTING
		default:
			return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_PREPARING
		}
	case domain.DataPipelineStatusRunning:
		// The destination keeps loading the snapshot after the Pipeline is
		// Running, so only the destination can tell these two phases apart.
		if pipeline.Observation.LakeFullSyncCompleted {
			return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_STREAMING
		}
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_INITIAL_LOADING
	case domain.DataPipelineStatusPausing:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_PAUSING
	case domain.DataPipelineStatusPaused:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_PAUSED
	case domain.DataPipelineStatusEditing:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_EDITING
	case domain.DataPipelineStatusResuming:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_RESUMING
	case domain.DataPipelineStatusDeleting:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_CLEANING_UP
	case domain.DataPipelineStatusCompleted:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_COMPLETE
	case domain.DataPipelineStatusFailed:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_ERROR
	default:
		return dataflow.DataPipelinePhase_DATA_PIPELINE_PHASE_UNSPECIFIED
	}
}

func dataPipelineSourceDomain2Dataflow(pipeline *domain.DataPipeline) *dataflow.DataPipelineSource {
	result := &dataflow.DataPipelineSource{
		Username:      pipeline.Params.Source.Username,
		TableFilter:   append([]string(nil), pipeline.Params.Source.TableFilter...),
		CaseSensitive: pipeline.Params.Source.CaseSensitive,
	}
	// Copied rather than aliased: the read model must not share a pointer with
	// the loaded row.
	if count := pipeline.Params.Source.MatchedTableCount; count != nil {
		matched := *count
		result.MatchedTableCount = &matched
	}
	return result
}

func dataPipelineDestinationDomain2Dataflow(pipeline *domain.DataPipeline) *dataflow.DataPipelineDestination {
	return &dataflow.DataPipelineDestination{
		Type:      dataPipelineDestinationTypeDomain2Dataflow(pipeline.Params.Destination.Type),
		LakeOrgId: pipeline.Params.Destination.LakeOrgID,
		Warehouse: pipeline.Params.Destination.Warehouse,
		TargetNaming: &dataflow.DataPipelineTargetNaming{
			DatabasePrefix: pipeline.Params.Destination.TargetNaming.DatabasePrefix,
			DatabaseSuffix: pipeline.Params.Destination.TargetNaming.DatabaseSuffix,
			TablePrefix:    pipeline.Params.Destination.TargetNaming.TablePrefix,
			TableSuffix:    pipeline.Params.Destination.TargetNaming.TableSuffix,
		},
		FlushIntervalSeconds: nonNegativeInt32ToUint32(pipeline.IntervalPlan.EndToEndSeconds),
	}
}

// dataPipelineS3AuthTypeDataflow2Domain maps the wire enum. An unset value is
// left unset rather than defaulted, so the model rejects it with a message
// about the auth type instead of a confusing missing-credential error.
func dataPipelineS3AuthTypeDataflow2Domain(
	authType dataflow.DataPipelineS3AuthType,
) domain.DataPipelineS3AuthType {
	switch authType {
	case dataflow.DataPipelineS3AuthType_DATA_PIPELINE_S3_AUTH_TYPE_ACCESS_KEY:
		return domain.DataPipelineS3AuthTypeAccessKey
	case dataflow.DataPipelineS3AuthType_DATA_PIPELINE_S3_AUTH_TYPE_ROLE_ARN:
		return domain.DataPipelineS3AuthTypeRoleARN
	default:
		return ""
	}
}

// dataPipelineS3AuthTypeDomain2Dataflow reads an empty stored auth type as
// access key: rows written before role ARN support carry a key pair.
func dataPipelineS3AuthTypeDomain2Dataflow(
	authType domain.DataPipelineS3AuthType,
) dataflow.DataPipelineS3AuthType {
	if authType == domain.DataPipelineS3AuthTypeRoleARN {
		return dataflow.DataPipelineS3AuthType_DATA_PIPELINE_S3_AUTH_TYPE_ROLE_ARN
	}
	return dataflow.DataPipelineS3AuthType_DATA_PIPELINE_S3_AUTH_TYPE_ACCESS_KEY
}

func dataPipelineStagingDomain2Dataflow(pipeline *domain.DataPipeline) *dataflow.DataPipelineStaging {
	s3Config := pipeline.Params.Staging.S3
	return &dataflow.DataPipelineStaging{
		Type: dataPipelineStagingTypeDomain2Dataflow(pipeline.Params.Staging.Type),
		S3: &dataflow.DataPipelineS3Staging{
			Uri:            dataPipelineS3URI(s3Config.Provider, s3Config.Bucket, s3Config.BasePrefix),
			Region:         s3Config.Region,
			AuthType:       dataPipelineS3AuthTypeDomain2Dataflow(s3Config.AuthType),
			AccessKeyId:    s3Config.AccessKeyID,
			RoleArn:        s3Config.RoleARN,
			Endpoint:       s3Config.Endpoint,
			SnapshotUri:    dataPipelineDerivedS3URI(s3Config.Provider, s3Config.Bucket, pipeline.StagingRoot, "snapshot/"),
			IncrementalUri: dataPipelineDerivedS3URI(s3Config.Provider, s3Config.Bucket, pipeline.StagingRoot, "incremental/"),
			SqsQueueUrl:    s3Config.SQSQueueURL,
			SqsRegion:      s3Config.SQSRegion,
			Provider:       dataPipelineS3ProviderDomain2Dataflow(s3Config.Provider),
		},
	}
}

func dataPipelineExecutionPlanDomain2Dataflow(
	pipeline *domain.DataPipeline,
) *dataflow.DataPipelineExecutionPlan {
	result := &dataflow.DataPipelineExecutionPlan{
		SnapshotEngine: dataPipelineSnapshotEngineDomain2Dataflow(pipeline.ExecutionPlan.Snapshot.Engine),
		SnapshotFormat: dataPipelineSnapshotFormatDomain2Dataflow(pipeline.ExecutionPlan.Snapshot.Format),
		SnapshotTso:    pipeline.SnapshotTSO,
	}
	if incremental := pipeline.ExecutionPlan.Incremental; incremental != nil {
		result.IncrementalEngine = dataPipelineIncrementalEngineDomain2Dataflow(incremental.Engine)
		result.IncrementalFormat = dataPipelineIncrementalFormatDomain2Dataflow(incremental.Format)
	}
	return result
}

func dataPipelineIntervalPlanDomain2Dataflow(
	plan domain.DeliveryIntervalPlan,
) *dataflow.DataPipelineDeliveryIntervalPlan {
	return &dataflow.DataPipelineDeliveryIntervalPlan{
		EndToEndSeconds:              nonNegativeInt32ToUint32(plan.EndToEndSeconds),
		ProducerFlushIntervalSeconds: nonNegativeInt32ToUint32(plan.ProducerFlushIntervalSeconds),
		ConsumerPollIntervalSeconds:  nonNegativeInt32ToUint32(plan.ConsumerPollIntervalSeconds),
	}
}

func dataPipelineChildrenDomain2Dataflow(
	details *domain.DataPipelineDetails,
) *dataflow.DataPipelineChildResources {
	return &dataflow.DataPipelineChildResources{
		Export: &dataflow.DataPipelineExportResource{
			ExportId:     details.Pipeline.Children.ExportID,
			Status:       details.ChildResources.Export.Status,
			ErrorMessage: details.ChildResources.Export.ErrorMessage,
		},
		Changefeed: &dataflow.DataPipelineChangefeedResource{
			ChangefeedId: details.Pipeline.Children.ChangefeedID,
			Status:       details.ChildResources.Changefeed.Status,
			ErrorMessage: details.ChildResources.Changefeed.ErrorMessage,
		},
		Lake: &dataflow.DataPipelineLakeResources{
			DatasourceName:    details.Pipeline.Children.LakeDataSourceName,
			DatasourceStatus:  details.ChildResources.Lake.DataSourceStatus,
			IntegrationName:   details.Pipeline.Children.LakeTaskName,
			IntegrationStatus: details.ChildResources.Lake.IntegrationStatus,
			ErrorMessage:      details.ChildResources.Lake.ErrorMessage,
		},
	}
}

func dataPipelineObservationDomain2Dataflow(
	observation domain.DataPipelineObservationDetails,
) *dataflow.DataPipelineObservation {
	result := &dataflow.DataPipelineObservation{
		ChangefeedCheckpointTso:  observation.ChangefeedCheckpointTSO,
		ChangefeedCheckpointTime: dataPipelineTime2Dataflow(observation.ChangefeedCheckpointTime),
		LakeResolvedTso:          observation.LakeResolvedTSO,
		LakeResolvedTime:         dataPipelineTime2Dataflow(observation.LakeResolvedTime),
		LatencySeconds:           observation.LatencySeconds,
		ObservedAt:               dataPipelineTime2Dataflow(observation.ObservedAt),
		Stale:                    observation.Stale,
		RowsSynced:               observation.RowsSynced,
		InsertCount:              observation.InsertCount,
		UpdateCount:              observation.UpdateCount,
		DeleteCount:              observation.DeleteCount,
		TablesActive:             observation.TablesActive,
		TablesError:              observation.TablesError,
		DatabasesActive:          observation.DatabasesActive,
	}
	if observation.Error != nil {
		result.ObservationError = observation.Error.Message
	}
	return result
}

func dataPipelineErrorDomain2Dataflow(pipelineErr *domain.DataPipelineError) *dataflow.DataPipelineError {
	if pipelineErr == nil {
		return nil
	}
	return &dataflow.DataPipelineError{
		Code:    pipelineErr.Code,
		Message: pipelineErr.Message,
	}
}

func dataPipelineTime2Dataflow(value *time.Time) *timestamppb.Timestamp {
	if value == nil || value.IsZero() {
		return nil
	}
	return timestamppb.New(*value)
}

func dataPipelineDerivedS3URI(
	provider domain.DataPipelineS3Provider, bucket, stagingRoot, suffix string,
) string {
	if strings.TrimSpace(stagingRoot) == "" {
		return ""
	}
	return dataPipelineS3URI(provider, bucket, strings.TrimSuffix(stagingRoot, "/")+"/"+suffix)
}

// dataPipelineS3ProviderDataflow2Domain reads an unspecified provider as AWS,
// so a request written before OSS support keeps its meaning.
func dataPipelineS3ProviderDataflow2Domain(
	provider dataflow.DataPipelineS3Provider,
) domain.DataPipelineS3Provider {
	if provider == dataflow.DataPipelineS3Provider_DATA_PIPELINE_S3_PROVIDER_ALIBABA_OSS {
		return domain.DataPipelineS3ProviderAlibabaOSS
	}
	return domain.DataPipelineS3ProviderAWS
}

// dataPipelineS3ProviderDomain2Dataflow reports AWS explicitly rather than
// leaving the field unspecified: a reader should not have to know that the
// stored zero value means AWS.
func dataPipelineS3ProviderDomain2Dataflow(
	provider domain.DataPipelineS3Provider,
) dataflow.DataPipelineS3Provider {
	if provider == domain.DataPipelineS3ProviderAlibabaOSS {
		return dataflow.DataPipelineS3Provider_DATA_PIPELINE_S3_PROVIDER_ALIBABA_OSS
	}
	return dataflow.DataPipelineS3Provider_DATA_PIPELINE_S3_PROVIDER_AWS
}

func dataPipelineS3URI(provider domain.DataPipelineS3Provider, bucket, prefix string) string {
	bucket = strings.TrimSpace(bucket)
	if bucket == "" {
		return ""
	}
	scheme := string(planner.SchemeForProvider(provider)) + "://"
	prefix = strings.TrimLeft(strings.TrimSpace(prefix), "/")
	if prefix == "" {
		return scheme + bucket
	}
	return scheme + bucket + "/" + prefix
}

func nonNegativeInt32ToUint32(value int32) uint32 {
	if value <= 0 {
		return 0
	}
	return uint32(value)
}

func dataPipelineStatusDataflow2Domain(status dataflow.DataPipelineStatus) *domain.DataPipelineStatus {
	var result domain.DataPipelineStatus
	switch status {
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_CREATING:
		result = domain.DataPipelineStatusCreating
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_RUNNING:
		result = domain.DataPipelineStatusRunning
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_PAUSING:
		result = domain.DataPipelineStatusPausing
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_PAUSED:
		result = domain.DataPipelineStatusPaused
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_EDITING:
		result = domain.DataPipelineStatusEditing
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_RESUMING:
		result = domain.DataPipelineStatusResuming
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_DELETING:
		result = domain.DataPipelineStatusDeleting
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_FAILED:
		result = domain.DataPipelineStatusFailed
	case dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_COMPLETED:
		result = domain.DataPipelineStatusCompleted
	default:
		return nil
	}
	return &result
}

func dataPipelineStatusDomain2Dataflow(status domain.DataPipelineStatus) dataflow.DataPipelineStatus {
	switch status {
	case domain.DataPipelineStatusCreating:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_CREATING
	case domain.DataPipelineStatusRunning:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_RUNNING
	case domain.DataPipelineStatusPausing:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_PAUSING
	case domain.DataPipelineStatusPaused:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_PAUSED
	case domain.DataPipelineStatusEditing:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_EDITING
	case domain.DataPipelineStatusResuming:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_RESUMING
	case domain.DataPipelineStatusDeleting:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_DELETING
	case domain.DataPipelineStatusFailed:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_FAILED
	case domain.DataPipelineStatusCompleted:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_COMPLETED
	default:
		return dataflow.DataPipelineStatus_DATA_PIPELINE_STATUS_UNSPECIFIED
	}
}

func dataPipelineSyncModeDataflow2Domain(syncMode dataflow.DataPipelineSyncMode) domain.DataPipelineSyncMode {
	switch syncMode {
	case dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL:
		return domain.DataPipelineSyncModeFull
	case dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL_AND_INCREMENTAL:
		return domain.DataPipelineSyncModeFullAndIncremental
	default:
		return ""
	}
}

func dataPipelineSyncModeDomain2Dataflow(syncMode domain.DataPipelineSyncMode) dataflow.DataPipelineSyncMode {
	switch syncMode {
	case domain.DataPipelineSyncModeFull:
		return dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL
	case domain.DataPipelineSyncModeFullAndIncremental:
		return dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_FULL_AND_INCREMENTAL
	default:
		return dataflow.DataPipelineSyncMode_DATA_PIPELINE_SYNC_MODE_UNSPECIFIED
	}
}

func dataPipelineDestinationTypeDataflow2Domain(
	destinationType dataflow.DataPipelineDestinationType,
) domain.DataPipelineDestinationType {
	switch destinationType {
	case dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_TIDB_LAKE:
		return domain.DataPipelineDestinationTypeTiDBLake
	default:
		return ""
	}
}

func dataPipelineDestinationTypeDomain2Dataflow(
	destinationType domain.DataPipelineDestinationType,
) dataflow.DataPipelineDestinationType {
	switch destinationType {
	case domain.DataPipelineDestinationTypeTiDBLake:
		return dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_TIDB_LAKE
	default:
		return dataflow.DataPipelineDestinationType_DATA_PIPELINE_DESTINATION_TYPE_UNSPECIFIED
	}
}

func dataPipelineStagingTypeDataflow2Domain(
	stagingType dataflow.DataPipelineStagingType,
) domain.DataPipelineStagingType {
	switch stagingType {
	case dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_CUSTOMER_MANAGED_S3:
		return domain.DataPipelineStagingTypeCustomerManagedS3
	case dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_TIDB_MANAGED:
		return domain.DataPipelineStagingTypeTiDBManaged
	default:
		return ""
	}
}

func dataPipelineStagingTypeDomain2Dataflow(
	stagingType domain.DataPipelineStagingType,
) dataflow.DataPipelineStagingType {
	switch stagingType {
	case domain.DataPipelineStagingTypeCustomerManagedS3:
		return dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_CUSTOMER_MANAGED_S3
	case domain.DataPipelineStagingTypeTiDBManaged:
		return dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_TIDB_MANAGED
	default:
		return dataflow.DataPipelineStagingType_DATA_PIPELINE_STAGING_TYPE_UNSPECIFIED
	}
}

func dataPipelineSnapshotEngineDomain2Dataflow(
	engine domain.DataPipelineSnapshotEngine,
) dataflow.DataPipelineSnapshotEngine {
	switch engine {
	case domain.DataPipelineSnapshotEngineDumpling:
		return dataflow.DataPipelineSnapshotEngine_DATA_PIPELINE_SNAPSHOT_ENGINE_DUMPLING
	case domain.DataPipelineSnapshotEngineBR:
		return dataflow.DataPipelineSnapshotEngine_DATA_PIPELINE_SNAPSHOT_ENGINE_BR
	default:
		return dataflow.DataPipelineSnapshotEngine_DATA_PIPELINE_SNAPSHOT_ENGINE_UNSPECIFIED
	}
}

func dataPipelineSnapshotFormatDomain2Dataflow(
	format domain.DataPipelineSnapshotFormat,
) dataflow.DataPipelineSnapshotFormat {
	if format == domain.DataPipelineSnapshotFormatCSV {
		return dataflow.DataPipelineSnapshotFormat_DATA_PIPELINE_SNAPSHOT_FORMAT_CSV
	}
	return dataflow.DataPipelineSnapshotFormat_DATA_PIPELINE_SNAPSHOT_FORMAT_UNSPECIFIED
}

func dataPipelineIncrementalEngineDomain2Dataflow(
	engine domain.DataPipelineIncrementalEngine,
) dataflow.DataPipelineIncrementalEngine {
	if engine == domain.DataPipelineIncrementalEngineTiCDC {
		return dataflow.DataPipelineIncrementalEngine_DATA_PIPELINE_INCREMENTAL_ENGINE_TICDC
	}
	return dataflow.DataPipelineIncrementalEngine_DATA_PIPELINE_INCREMENTAL_ENGINE_UNSPECIFIED
}

func dataPipelineIncrementalFormatDomain2Dataflow(
	format domain.DataPipelineIncrementalFormat,
) dataflow.DataPipelineIncrementalFormat {
	if format == domain.DataPipelineIncrementalFormatCanalJSON {
		return dataflow.DataPipelineIncrementalFormat_DATA_PIPELINE_INCREMENTAL_FORMAT_CANAL_JSON
	}
	return dataflow.DataPipelineIncrementalFormat_DATA_PIPELINE_INCREMENTAL_FORMAT_UNSPECIFIED
}

func dataPipelineStagingTrustDomain2Dataflow(
	resp *domain.GetDataPipelineStagingTrustResp,
) *dataflow.DataPipelineStagingTrust {
	if resp == nil {
		return &dataflow.DataPipelineStagingTrust{}
	}
	principals := make([]*dataflow.DataPipelineStagingTrustPrincipal, 0, len(resp.Principals))
	for _, principal := range resp.Principals {
		principals = append(principals, &dataflow.DataPipelineStagingTrustPrincipal{
			Name:       principal.Name,
			RoleArn:    principal.RoleARN,
			ExternalId: principal.ExternalID,
		})
	}
	return &dataflow.DataPipelineStagingTrust{
		Principals:    principals,
		BucketActions: append([]string(nil), resp.BucketActions...),
		QueueActions:  append([]string(nil), resp.QueueActions...),
	}
}