package vo

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless/dm"
	"github.com/tidbcloud/pkgv2/log"
	v1beta1 "github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/utils"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// CreateV1beta1TaskReqToDMTask converts v1beta1 CreateMigrationReq to entity DMTask
func CreateV1beta1TaskReqToDMTask(
	ctx context.Context,
	req *v1beta1.CreateMigrationReq,
	cluster *tidbmgmtservicev1.Cluster,
) (*entity.DMTask, bizerr.BizError) {
	mode, err := ConvertV1beta1ModeToDMJobMode(req.GetMode())
	if err != nil {
		return nil, err
	}
	shardMode, err := ConvertV1beta1ShardModeToDMShardMode(req.ShardMode)
	if err != nil {
		return nil, err
	}
	taskID := entity.GenerateIDWithPrefix(constant.DMTaskIDPrefix)
	creator := utils.BuildCreatorID(ctx)

	return &entity.DMTask{
		ID:                      taskID,
		Name:                    req.GetDisplayName(),
		Stage:                   constant.DMTaskStageSourceCreating,
		Creator:                 creator,
		ClusterID:               req.GetClusterId(),
		TenantID:                strconv.FormatUint(cluster.OrganizationId, 10),
		CloudProvider:           cluster.GetCloudProvider(),
		Region:                  cluster.GetRegion(),
		Mode:                    mode,
		ShardMode:               shardMode,
		TargetUser:              req.GetTarget().GetUser(),
		TargetEncryptedPassword: req.GetTarget().GetPassword(),
		DMSources: V1beta1SourcesToEntitySources(cluster.OrganizationId,
			cluster.ProjectId,
			cluster.ClusterId,
			taskID,
			creator,
			req.GetSources()),
	}, nil
}

// V1beta1SourcesToEntitySources converts v1beta1 Source slice to entity DMSource slice
func V1beta1SourcesToEntitySources(
	orgId,
	projectId,
	clusterId uint64,
	taskId,
	creator string,
	sources []*v1beta1.Source,
) []entity.DMSource {
	if len(sources) == 0 {
		return nil
	}
	result := make([]entity.DMSource, 0, len(sources))
	for _, s := range sources {
		if s == nil {
			continue
		}
		result = append(result, entity.DMSource{
			ID:             entity.GenerateIDWithPrefix(constant.DMSourceIDPrefix),
			Stage:          constant.DMSourceStageCreating,
			Creator:        creator,
			ClusterID:      strconv.FormatUint(clusterId, 10),
			TenantID:       strconv.FormatUint(orgId, 10),
			SourceType:     v1beta1SourceTypeToEntity(s.GetSourceType()),
			DMTaskID:       taskId,
			ConnProfile:    V1beta1ConnProfileToEntity(s.GetConnProfile()),
			MigrationRules: V1beta1MigrationRulesToEntity(s.GetMigrationRules()),
			BinlogName:     s.GetBinlogName(),
			BinlogPos:      s.GetBinlogPosition(),
			BinlogGTID:     s.GetBinlogGtid(),
		})
	}
	return result
}

// V1beta1ConnProfileToEntity converts v1beta1 ConnProfile to entity ConnProfile
func V1beta1ConnProfileToEntity(pb *v1beta1.ConnProfile) entity.ConnProfile {
	if pb == nil {
		return entity.ConnProfile{}
	}
	switch pb.ConnType {
	case v1beta1.ConnType_PUBLIC:
		return entity.ConnProfile{
			Type:     constant.ConnTypePublic,
			Target:   pb.GetHost(),
			Port:     pb.GetPort(),
			User:     pb.GetUser(),
			Password: pb.GetPassword(),
			Security: V1beta1SecurityToEntity(pb.GetSecurity()),
		}
	case v1beta1.ConnType_PRIVATE_LINK:
		return entity.ConnProfile{
			Type:     constant.ConnTypePrivateLink,
			Target:   pb.GetEndpointId(),
			Port:     pb.GetPort(),
			User:     pb.GetUser(),
			Password: pb.GetPassword(),
			Security: V1beta1SecurityToEntity(pb.GetSecurity()),
		}
	}
	return entity.ConnProfile{}
}

// V1beta1SecurityToEntity converts v1beta1 Security to entity Security
func V1beta1SecurityToEntity(pb *v1beta1.Security) *entity.Security {
	if pb == nil {
		return nil
	}
	return &entity.Security{
		CertAllowedCn:  pb.GetCertAllowedCn(),
		SslCaContent:   pb.GetSslCaContent(),
		SslCertContent: pb.GetSslCertContent(),
		SslKeyContent:  pb.GetSslKeyContent(),
	}
}

// DMTaskAndStatusToV1beta1PbTask converts entity DMTask and status to v1beta1 DMTask
func DMTaskAndStatusToV1beta1PbTask(ctx context.Context, task *entity.DMTask) (*v1beta1.Migration, error) {
	if task == nil {
		return nil, nil
	}
	pbTask := &v1beta1.Migration{
		MigrationId: task.ID,
		DisplayName: task.Name,
		Mode:        DMJobModeToV1beta1Mode(task.Mode),
		TargetUser:  task.TargetUser,
		CreateTime:  timestamppb.New(task.CreatedAt),
		State:       DMTaskStageToV1beta1(task.Stage),
	}
	pbTask.SubTasks = make([]*v1beta1.SubTask, 0, len(task.DMSources))
	for i := range task.DMSources {
		pbSource := buildV1beta1SourceFromEntity(&task.DMSources[i])
		pbSubTask := &v1beta1.SubTask{Source: pbSource}
		if task.DMSources[i].SubTaskStatus != nil {
			if err := fillV1beta1SubTaskStatus(ctx, task.DMSources[i].SubTaskStatus, pbSubTask, task.Mode); err != nil {
				return nil, err
			}
		}
		pbTask.SubTasks = append(pbTask.SubTasks, pbSubTask)
	}
	return pbTask, nil
}

// buildV1beta1SourceFromEntity converts an entity.DMSource to v1beta1.Source
func buildV1beta1SourceFromEntity(s *entity.DMSource) *v1beta1.Source {
	if s == nil {
		return &v1beta1.Source{}
	}
	src := &v1beta1.Source{
		ConnProfile: EntityConnProfileToV1beta1(&s.ConnProfile),
		SourceType:  entitySourceType2V1beta1(s.SourceType),
	}

	if len(s.MigrationRules) > 0 {
		src.MigrationRules = EntityMigrationRulesToV1beta1(s.MigrationRules)
	} else {
		src.MigrationRules = entityBaRulesToV1beta1(s.BlockAllowRules)
	}
	if s.BinlogName != "" {
		src.BinlogName = &s.BinlogName
	}
	if s.BinlogPos != 0 {
		src.BinlogPosition = &s.BinlogPos
	}
	if s.BinlogGTID != "" {
		src.BinlogGtid = &s.BinlogGTID
	}
	return src
}

// entityBaRulesToV1beta1 converts BlockAllowRules to v1beta1 MigrationRules.
func entityBaRulesToV1beta1(rules entity.BlockAllowRules) []*v1beta1.MigrationRule {
	res := make([]*v1beta1.MigrationRule, 0)
	for _, db := range rules.DoDBs {
		res = append(res, &v1beta1.MigrationRule{
			SourceTable: &v1beta1.MigrationRule_Source{
				SchemaPattern: db,
			},
			TargetTable: &v1beta1.MigrationRule_Table{
				Schema: db,
			},
		})
	}
	for _, table := range rules.DoTables {
		r := &v1beta1.MigrationRule{
			SourceTable: &v1beta1.MigrationRule_Source{
				SchemaPattern: table.Schema,
				TablePattern:  table.Table,
			},
			TargetTable: &v1beta1.MigrationRule_Table{
				Schema: table.Schema,
			},
		}
		if table.Table != "~.*" && table.Table != "*" {
			r.TargetTable.Table = table.Table
		}
		res = append(res, r)
	}

	return res
}

// fillV1beta1SubTaskStatus populates SubTask runtime details and applies step-completion overrides for MODE_ALL.
func fillV1beta1SubTaskStatus(
	ctx context.Context,
	st *entity.SubTaskStatus,
	pbSubTask *v1beta1.SubTask,
	mode constant.DMTaskMode,
) error {
	var err error
	pbSubTask.LoadDetail, err = ConvertLoadStatusToV1beta1(ctx, st.LoadStatus)
	if err != nil {
		return err
	}
	pbSubTask.DumpDetail, err = ConvertDumpStatusToV1beta1(ctx, st.DumpStatus)
	if err != nil {
		return err
	}
	pbSubTask.SyncDetail = ConvertSyncStatusToV1beta1(st.SyncStatus)
	pbSubTask.CurrentStep = ConvertToV1beta1Step(st.Unit)

	if mode == constant.DMTaskModeAll {
		ensureV1beta1StepCompletionForModeAll(pbSubTask)
	}
	pbSubTask.ErrorMsg = st.ErrorMsg
	pbSubTask.Stage = convertStageToV1beta1(st.Stage)
	if pbSubTask.ErrorMsg != nil && *pbSubTask.ErrorMsg != "" {
		pbSubTask.Stage = v1beta1.SubTask_FAILED
	}
	return nil
}

// ensureV1beta1StepCompletionForModeAll ensures dump/load show 100% once the pipeline advances beyond them.
func ensureV1beta1StepCompletionForModeAll(pbSubTask *v1beta1.SubTask) {
	switch pbSubTask.CurrentStep {
	case v1beta1.SubTask_SYNC:
		if pbSubTask.LoadDetail != nil {
			pbSubTask.LoadDetail.Progress = 100
			pbSubTask.LoadDetail.FinishedBytes = pbSubTask.LoadDetail.TotalBytes
		} else {
			pbSubTask.LoadDetail = &v1beta1.LoadDetail{Progress: 100}
		}
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = 100
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &v1beta1.DumpDetail{Progress: 100}
		}
	case v1beta1.SubTask_LOAD:
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = 100
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &v1beta1.DumpDetail{Progress: 100}
		}
	}
}

func DMTaskStageToV1beta1(stage constant.DMTaskStage) v1beta1.Migration_State {
	switch stage {
	case constant.DMTaskStageTaskCreating, constant.DMTaskStageSourceCreating, constant.DMTaskStageSourceBounding:
		return v1beta1.Migration_CREATING
	case constant.DMTaskStageStopped:
		return v1beta1.Migration_PAUSED
	case constant.DMTaskStageRunning:
		return v1beta1.Migration_RUNNING
	case constant.DMTaskStageFailed:
		return v1beta1.Migration_FAILED
	case constant.DMTaskStageDeleting:
		return v1beta1.Migration_DELETING
	default:
		return v1beta1.Migration_STATE_UNSPECIFIED
	}
}

// ConvertV1beta1ModeToDMJobMode converts v1beta1 TaskMode to entity DMTaskMode
func ConvertV1beta1ModeToDMJobMode(mode v1beta1.TaskMode) (constant.DMTaskMode, bizerr.BizError) {
	switch mode {
	case v1beta1.TaskMode_ALL:
		return constant.DMTaskModeAll, nil
	case v1beta1.TaskMode_INCREMENTAL:
		return constant.DMTaskModeIncr, nil
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "invalid mode: %s", mode.String())
	}
}

// ConvertV1beta1ShardModeToDMShardMode converts v1beta1 TaskShardMode to DM task shard mode.
func ConvertV1beta1ShardModeToDMShardMode(mode *v1beta1.TaskShardMode) (*dm.TaskShardMode, bizerr.BizError) {
	if mode == nil {
		return nil, nil
	}
	switch *mode {
	case v1beta1.TaskShardMode_PESSIMISTIC:
		return dm.TASKSHARDMODE_PESSIMISTIC.Ptr(), nil
	case v1beta1.TaskShardMode_OPTIMISTIC:
		return dm.TASKSHARDMODE_OPTIMISTIC.Ptr(), nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid shard mode: %s", mode.String())
	}
}

// DMJobModeToV1beta1Mode converts entity DMTaskMode to v1beta1 TaskMode
func DMJobModeToV1beta1Mode(mode constant.DMTaskMode) v1beta1.TaskMode {
	switch mode {
	case constant.DMTaskModeAll:
		return v1beta1.TaskMode_ALL
	case constant.DMTaskModeIncr:
		return v1beta1.TaskMode_INCREMENTAL
	default:
		return v1beta1.TaskMode_MODE_UNSPECIFIED
	}
}

// EntityConnProfileToV1beta1 converts entity ConnProfile to v1beta1 ConnProfile
func EntityConnProfileToV1beta1(cp *entity.ConnProfile) *v1beta1.ConnProfile {
	switch cp.Type {
	case constant.ConnTypePublic:
		return &v1beta1.ConnProfile{
			ConnType: v1beta1.ConnType_PUBLIC,
			Endpoint: &v1beta1.ConnProfile_Host{Host: cp.Target},
			Port:     cp.Port,
			User:     cp.User,
			Password: "",
			Security: EntitySecurityToV1beta1(cp.Security),
		}
	case constant.ConnTypePrivateLink:
		return &v1beta1.ConnProfile{
			ConnType: v1beta1.ConnType_PRIVATE_LINK,
			Endpoint: &v1beta1.ConnProfile_EndpointId{EndpointId: cp.Target},
			Port:     cp.Port,
			User:     cp.User,
			Password: "",
			Security: EntitySecurityToV1beta1(cp.Security),
		}
	}
	return &v1beta1.ConnProfile{}
}

// EntitySecurityToV1beta1 converts entity Security to v1beta1 Security
func EntitySecurityToV1beta1(sec *entity.Security) *v1beta1.Security {
	if sec == nil {
		return nil
	}
	return &v1beta1.Security{
		CertAllowedCn:  sec.CertAllowedCn,
		SslCaContent:   sec.SslCaContent,
		SslCertContent: sec.SslCertContent,
		SslKeyContent:  sec.SslKeyContent,
	}
}

// V1beta1MigrationRulesToEntity converts v1beta1 MigrationRules slice to entity MigrationRules
func V1beta1MigrationRulesToEntity(pb []*v1beta1.MigrationRule) entity.MigrationRules {
	if len(pb) == 0 {
		return nil
	}
	rules := make(entity.MigrationRules, 0, len(pb))
	for _, r := range pb {
		if r == nil {
			continue
		}
		rule := entity.MigrationRule{}
		if r.SourceTable != nil { // proto field name source_table -> SourceTable in Go
			rule.Source.SchemaPattern = r.SourceTable.GetSchemaPattern()
			rule.Source.TablePattern = r.SourceTable.GetTablePattern()
		}
		rule.Target = v1beta1MigrationRuleTargetToEntity(r.GetTargetTable())
		rules = append(rules, rule)
	}
	return rules
}

// EntityMigrationRulesToV1beta1 converts entity MigrationRules to v1beta1 MigrationRule slice
func EntityMigrationRulesToV1beta1(rules entity.MigrationRules) []*v1beta1.MigrationRule {
	if len(rules) == 0 {
		return nil
	}
	ret := make([]*v1beta1.MigrationRule, 0, len(rules))
	for _, r := range rules {
		pb := &v1beta1.MigrationRule{}
		// allocate nested messages only if fields are non-empty
		if r.Source.SchemaPattern != "" || r.Source.TablePattern != "" {
			pb.SourceTable = &v1beta1.MigrationRule_Source{
				SchemaPattern: r.Source.SchemaPattern,
				TablePattern:  r.Source.TablePattern,
			}
		}
		pb.TargetTable = entityMigrationRuleTargetToV1beta1(r.Target)
		ret = append(ret, pb)
	}
	return ret
}

func entityMigrationRuleTargetToV1beta1(target entity.MigrationRuleTarget) *v1beta1.MigrationRule_Table {
	if target.Schema == "" && target.Table == "" {
		return nil
	}
	return &v1beta1.MigrationRule_Table{Schema: target.Schema, Table: target.Table}
}

func v1beta1MigrationRuleTargetToEntity(pbTarget *v1beta1.MigrationRule_Table) entity.MigrationRuleTarget {
	if pbTarget == nil {
		return entity.MigrationRuleTarget{}
	}
	return entity.MigrationRuleTarget{Schema: pbTarget.GetSchema(), Table: pbTarget.GetTable()}
}

func entitySourceType2V1beta1(src constant.DBType) v1beta1.Source_SourceType {
	switch src {
	case constant.DBTypeMySQL:
		return v1beta1.Source_MYSQL
	case constant.DBTypeAliCloudRDSMySQL:
		return v1beta1.Source_ALICLOUD_RDS_MYSQL
	default:
		return v1beta1.Source_SOURCE_TYPE_UNSPECIFIED
	}
}

func v1beta1SourceTypeToEntity(src v1beta1.Source_SourceType) constant.DBType {
	switch src {
	case v1beta1.Source_MYSQL:
		return constant.DBTypeMySQL
	case v1beta1.Source_ALICLOUD_RDS_MYSQL:
		return constant.DBTypeAliCloudRDSMySQL
	default:
		return ""
	}
}

// ConvertDumpStatusToV1beta1 converts internal DumpStatus to v1beta1 DumpDetail
func ConvertDumpStatusToV1beta1(ctx context.Context, ds *dm.DumpStatus) (*v1beta1.DumpDetail, error) {
	if ds == nil {
		return nil, nil
	}
	p, err := ParseProgressString(ds.Progress)
	if err != nil {
		log.WithContextL(ctx).
			Error("parse dump progress string failed", zap.Error(err), zap.String("progress", ds.Progress))
		return nil, err
	}
	return &v1beta1.DumpDetail{
		Bps:             ds.Bps,
		TotalTables:     ds.TotalTables,
		CompletedTables: int64(ds.CompletedTables),
		Progress:        p,
		FinishedBytes:   int64(ds.FinishedBytes),
		FinishedRows:    int64(ds.FinishedRows),
	}, nil
}

// ConvertLoadStatusToV1beta1 converts internal LoadStatus to v1beta1 LoadDetail
func ConvertLoadStatusToV1beta1(ctx context.Context, ls *dm.LoadStatus) (*v1beta1.LoadDetail, error) {
	if ls == nil {
		return nil, nil
	}
	p, err := ParseProgressString(ls.Progress)
	if err != nil {
		log.WithContextL(ctx).
			Error("parse load progress string failed", zap.Error(err), zap.String("progress", ls.Progress))
		return nil, err
	}
	return &v1beta1.LoadDetail{
		Bps:           ls.Bps,
		Progress:      p,
		FinishedBytes: ls.FinishedBytes,
		TotalBytes:    ls.TotalBytes,
	}, nil
}

// ConvertSyncStatusToV1beta1 converts internal SyncStatus to v1beta1 SyncDetail
func ConvertSyncStatusToV1beta1(ss *dm.SyncStatus) *v1beta1.SyncDetail {
	if ss == nil {
		return nil
	}
	return &v1beta1.SyncDetail{Rps: ss.RecentTps, Latency: ss.SecondsBehindMaster, Checkpoint: ss.SyncerBinlog}
}

// ConvertToV1beta1Step maps unit string to v1beta1.SubTask_Step
func ConvertToV1beta1Step(unit string) v1beta1.SubTask_Step {
	switch strings.ToLower(unit) {
	case "dump":
		return v1beta1.SubTask_DUMP
	case "sync":
		return v1beta1.SubTask_SYNC
	case "load":
		return v1beta1.SubTask_LOAD
	default:
		return v1beta1.SubTask_STEP_UNSPECIFIED
	}
}

// convertStageToV1beta1 converts dm.TaskStage to v1beta1 SubTask_Stage
func convertStageToV1beta1(stage dm.TaskStage) v1beta1.SubTask_Stage {
	switch stage {
	case dm.TASKSTAGE_FINISHED:
		return v1beta1.SubTask_FINISHED
	case dm.TASKSTAGE_PAUSED, dm.TASKSTAGE_STOPPED:
		return v1beta1.SubTask_PAUSED
	case dm.TASKSTAGE_RUNNING:
		return v1beta1.SubTask_RUNNING
	default:
		return v1beta1.SubTask_UNKNOWN
	}
}

// ConvertV1beta1DMMode converts v1beta1 TaskMode to regional DM precheck mode
func ConvertV1beta1DMMode(mode v1beta1.TaskMode) regionalserverlessv1.DMPrecheck_TaskMode {
	switch mode {
	case v1beta1.TaskMode_ALL:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_ALL
	case v1beta1.TaskMode_INCREMENTAL:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_INCREMENTAL
	default:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_UNSPECIFIED
	}
}

// CreateV1beta1TaskReqToDMPrecheck converts v1beta1 CreateMigrationReq to entity DMPrecheck
func CreateV1beta1TaskReqToDMPrecheck(
	ctx context.Context,
	req *v1beta1.CreateMigrationReq,
	cluster *tidbmgmtservicev1.Cluster,
	describeEndpoint func(c context.Context, clusterID string, endpointID string) (string, bizerr.BizError),
) (*entity.DMPrecheck, bizerr.BizError) {
	precheckID := entity.GenerateIDWithPrefix(constant.DMPrecheckIDPrefix)
	creator := utils.BuildCreatorID(ctx)
	clusterID := req.GetClusterId()

	precheckReq := &entity.PrecheckRequest{
		DmPrecheck: &regionalserverlessv1.DMPrecheck{
			TargetDb: &regionalserverlessv1.DMPrecheck_TargetDatabase{
				Host:     constant.DMTaskTargetHost,
				Port:     constant.DMTaskTargetPort,
				User:     req.GetTarget().GetUser(),
				Password: req.GetTarget().GetPassword(),
			},
			Mode: ConvertV1beta1DMMode(req.GetMode()),
		},
		ClusterId: clusterID,
	}
	if len(req.Sources[0].MigrationRules) == 0 {
		precheckReq.DmPrecheck.FullInstanceMigration = true
	}
	sources, bizErr := V1beta1SourcesToDMPrecheckSources(ctx, req.GetSources(), clusterID, describeEndpoint)
	if bizErr != nil {
		return nil, bizErr
	}
	precheckReq.DmPrecheck.Sources = sources

	return &entity.DMPrecheck{
		ID:         precheckID,
		Creator:    creator,
		ClusterID:  clusterID,
		TenantID:   strconv.FormatUint(cluster.OrganizationId, 10),
		Request:    precheckReq,
		TotalItems: int32(len(GetPrecheckItemList(precheckReq))),
		Stage:      constant.DMPrecheckStagePending,
	}, nil
}

// V1beta1SourcesToDMPrecheckSources converts v1beta1 Source slice to regional DMPrecheck Sources
func V1beta1SourcesToDMPrecheckSources(
	ctx context.Context,
	pbSources []*v1beta1.Source,
	clusterId string,
	getEndpointDomain func(c context.Context, clusterID string, endpointID string) (string, bizerr.BizError),
) ([]*regionalserverlessv1.DMPrecheck_Source, bizerr.BizError) {
	if len(pbSources) == 0 {
		return nil, nil
	}
	sources := make([]*regionalserverlessv1.DMPrecheck_Source, 0, len(pbSources))
	for _, pbSource := range pbSources {
		if pbSource == nil {
			continue
		}
		source := &regionalserverlessv1.DMPrecheck_Source{}
		if pbSource.ConnProfile != nil {
			source.ConnProfile = &regionalserverlessv1.DMPrecheck_ConnProfile{
				Host:     pbSource.ConnProfile.GetHost(),
				Port:     pbSource.ConnProfile.GetPort(),
				User:     pbSource.ConnProfile.GetUser(),
				Password: pbSource.ConnProfile.GetPassword(),
			}
			if pbSource.ConnProfile.GetConnType() == v1beta1.ConnType_PRIVATE_LINK {
				domain, err := getEndpointDomain(ctx, clusterId, pbSource.ConnProfile.GetEndpointId())
				if err != nil {
					return nil, err
				}
				source.ConnProfile.Host = domain
			}
			if pbSource.SourceType == v1beta1.Source_ALICLOUD_RDS_MYSQL {
				source.NeedPkCheck = true
			}

			if cp := pbSource.GetConnProfile(); cp != nil && cp.GetSecurity() != nil {
				sec := cp.GetSecurity()
				source.ConnProfile.Security = &regionalserverlessv1.DMPrecheck_Security{
					SslCaContent:   sec.GetSslCaContent(),
					SslCertContent: sec.GetSslCertContent(),
					SslKeyContent:  sec.GetSslKeyContent(),
					CertAllowedCn:  sec.GetCertAllowedCn(),
				}
			}
		}
		if pbSource.MigrationRules != nil {
			source.RouteRules = convertToPrecheckMigrationRules(pbSource.MigrationRules)
		}
		source.BinlogName = pbSource.BinlogName
		source.BinlogPos = pbSource.BinlogPosition
		source.BinlogGtid = pbSource.BinlogGtid
		sources = append(sources, source)
	}
	return sources, nil
}

func convertToPrecheckMigrationRules(pb []*v1beta1.MigrationRule) []*regionalserverlessv1.DMPrecheck_TableRouteRule {
	out := make([]*regionalserverlessv1.DMPrecheck_TableRouteRule, 0, len(pb))
	for _, r := range pb {
		if r == nil {
			continue
		}
		rr := &regionalserverlessv1.DMPrecheck_TableRouteRule{}
		if src := r.GetSourceTable(); src != nil {
			rr.SchemaPattern = src.GetSchemaPattern()
			rr.TablePattern = src.GetTablePattern()
		}
		if tgt := r.GetTargetTable(); tgt != nil {
			target := v1beta1MigrationRuleTargetToEntity(tgt)
			rr.TargetSchema = target.Schema
			rr.TargetTable = target.Table
		}
		out = append(out, rr)
	}
	return out
}

// DMPrecheckToV1beta1PbPrecheck converts entity DMPrecheck to v1beta1 MigrationPrecheck
func DMPrecheckToV1beta1PbPrecheck(precheck *entity.DMPrecheck) *v1beta1.MigrationPrecheck {
	if precheck == nil {
		return nil
	}
	resp := &v1beta1.MigrationPrecheck{}
	resp.PrecheckId = precheck.ID
	resp.Status = dmPrecheckStageToV1beta1(precheck.Stage)
	resp.Total = precheck.TotalItems
	var failedCnt, warnCnt, successCnt int32
	for i := range precheck.DMPrecheckItems {
		item := &precheck.DMPrecheckItems[i]
		if item.Stage == constant.DMPrecheckStageFailed {
			// Internal worker failure may still carry persisted results (e.g. legacy rows).
			if respItem := formatV1beta1CheckItemResult(item); respItem != nil {
				resp.Items = append(resp.Items, respItem)
				failedCnt++
				continue
			}
			resp.Items = append(resp.Items, &v1beta1.PrecheckItem{
				Description:    item.Description,
				Status:         v1beta1.PrecheckItem_FAILED,
				Solution:       fmt.Sprintf("Precheck(id = %s) failed, please contact the support team", precheck.ID),
				Reason:         "internal error: precheck item failed unexpectedly",
				SolutionDocUrl: constant.DmDocUrl,
				Type:           ConvertV1beta1PrecheckItemType(item.Type),
			})
			failedCnt++
			continue
		} else if item.Stage == constant.DMPrecheckStageRunning {
			continue
		}
		respItem := formatV1beta1CheckItemResult(item)
		if respItem == nil {
			continue
		}
		resp.Items = append(resp.Items, respItem)
		switch respItem.Status {
		case v1beta1.PrecheckItem_FAILED:
			failedCnt++
		case v1beta1.PrecheckItem_WARNING:
			warnCnt++
		case v1beta1.PrecheckItem_SUCCESS:
			successCnt++
		}
	}
	resp.FailedCnt = failedCnt
	resp.WarnCnt = warnCnt
	resp.SuccessCnt = successCnt
	if failedCnt > 0 {
		resp.Status = v1beta1.MigrationPrecheck_FAILED
	}
	return resp
}

// formatV1beta1CheckItemResult composes a PrecheckItem for v1beta1 response
func formatV1beta1CheckItemResult(item *entity.DMPrecheckItem) *v1beta1.PrecheckItem {
	raw := item.Results
	if len(raw.Results) == 0 {
		return nil
	}
	first := raw.Results[0]
	ret := &v1beta1.PrecheckItem{Description: first.Desc}
	status := first.State

	reasons := make([]string, 0)
	if len(first.Errors) > 0 {
		for _, e := range first.Errors {
			if e.ShortErr != "" {
				reasons = append(reasons, e.ShortErr)
			}
		}
	}
	instrs := make([]string, 0)
	if strings.TrimSpace(first.Instruction) != "" {
		instrs = append(instrs, strings.TrimSpace(first.Instruction))
	}
	for i := 1; i < len(raw.Results); i++ {
		r := raw.Results[i]
		switch r.State {
		case checker.StateFailure:
			status = checker.StateFailure
		case checker.StateWarning:
			if status != checker.StateFailure {
				status = checker.StateWarning
			}
		case checker.StateSuccess:
			// no-op
		}
		for _, e := range r.Errors {
			if e.ShortErr != "" {
				reasons = append(reasons, e.ShortErr)
			}
		}
		if strings.TrimSpace(r.Instruction) != "" {
			instrs = append(instrs, strings.TrimSpace(r.Instruction))
		}
	}
	ret.Status = checkerStateToV1beta1(status)
	if len(reasons) > 0 {
		ret.Reason = strings.Join(reasons, "; ")
	}
	if len(instrs) > 0 {
		ret.Solution = strings.Join(instrs, "; ")
	}
	if ret.Status != v1beta1.PrecheckItem_SUCCESS {
		if url, ok := constant.PrecheckItemHelpUrls[item.Type]; ok {
			ret.SolutionDocUrl = url
		}
	}
	ret.Type = ConvertV1beta1PrecheckItemType(item.Type)
	return ret
}

// ConvertV1beta1PrecheckItemType maps regional precheck item type to v1beta1 response enum
func ConvertV1beta1PrecheckItemType(item regionalserverlessv1.DMPrecheck_PrecheckItem) v1beta1.PrecheckItemType {
	switch item {
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_DUMP_PRIVILEGE_CHECKING:
		return v1beta1.PrecheckItemType_DUMP_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_REPLICATION_PRIVILEGE_CHECKING:
		return v1beta1.PrecheckItemType_REPLICATION_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_VERSION_CHECKING:
		return v1beta1.PrecheckItemType_VERSION_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_SERVER_ID_CHECKING:
		return v1beta1.PrecheckItemType_SERVER_ID_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_ENABLE_CHECKING:
		return v1beta1.PrecheckItemType_BINLOG_ENABLE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_FORMAT_CHECKING:
		return v1beta1.PrecheckItemType_BINLOG_FORMAT_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_ROW_IMAGE_CHECKING:
		return v1beta1.PrecheckItemType_BINLOG_ROW_IMAGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_TABLE_SCHEMA_CHECKING:
		return v1beta1.PrecheckItemType_TABLE_SCHEMA_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_DB_CHECKING:
		return v1beta1.PrecheckItemType_BINLOG_DB_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_CONN_NUMBER_CHECKING:
		return v1beta1.PrecheckItemType_CONN_NUMBER_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_TARGET_DB_PRIVILEGE_CHECKING:
		return v1beta1.PrecheckItemType_TARGET_DB_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_META_POSITION_CHECKING:
		return v1beta1.PrecheckItemType_META_POSITION_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_PRIMARY_KEY_CHECKING:
		return v1beta1.PrecheckItemType_PRIMARY_KEY_CHECKING
	default:
		return v1beta1.PrecheckItemType_PRECHECK_ITEM_TYPE_UNSPECIFIED
	}
}

func dmPrecheckStageToV1beta1(stage constant.DMPrecheckStage) v1beta1.MigrationPrecheck_Status {
	switch stage {
	case constant.DMPrecheckStagePending:
		return v1beta1.MigrationPrecheck_PENDING
	case constant.DMPrecheckStageRunning:
		return v1beta1.MigrationPrecheck_RUNNING
	case constant.DMPrecheckStageFinished:
		return v1beta1.MigrationPrecheck_FINISHED
	case constant.DMPrecheckStageFailed:
		return v1beta1.MigrationPrecheck_FAILED
	case constant.DMPrecheckStageCanceled:
		return v1beta1.MigrationPrecheck_CANCELED
	default:
		return v1beta1.MigrationPrecheck_STATUS_UNSPECIFIED
	}
}

func checkerStateToV1beta1(state checker.State) v1beta1.PrecheckItem_Status {
	switch state {
	case checker.StateFailure:
		return v1beta1.PrecheckItem_FAILED
	case checker.StateWarning:
		return v1beta1.PrecheckItem_WARNING
	case checker.StateSuccess:
		return v1beta1.PrecheckItem_SUCCESS
	default:
		return v1beta1.PrecheckItem_STATUS_UNSPECIFIED
	}
}