package premium_dm

import (
	"context"
	"strconv"
	"strings"

	dm "github.com/pingcap/tiflow/dm/openapi"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	modelpremiumdm "github.com/tidbcloud/dataflow-service-ng/app/models/premium_dm"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/premium_dm/v1beta2"
	"github.com/tidbcloud/dataflow-service-ng/internal/bizutils"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const (
	premiumDmDocURL                = "https://docs.pingcap.com/tidbcloud/migrate-from-mysql-using-data-migration"
	premiumDmTroubleshootingDocURL = "https://docs.pingcap.com/tidbcloud/tidb-cloud-dm-precheck-and-troubleshooting"
)

var premiumPrecheckItemHelpURLs = map[v1beta2.PrecheckItemType]string{
	v1beta2.PrecheckItemType_DUMP_PRIVILEGE_CHECKING: premiumDmDocURL +
		"#grant-required-privileges-to-the-upstream-database",
	v1beta2.PrecheckItemType_REPLICATION_PRIVILEGE_CHECKING: premiumDmDocURL +
		"#grant-required-privileges-to-the-upstream-database",
	v1beta2.PrecheckItemType_VERSION_CHECKING: premiumDmDocURL +
		"#make-sure-your-data-source-and-version-are-supported",
	v1beta2.PrecheckItemType_SERVER_ID_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-whether-mysql-server_id-has-been-greater-than-0",
	v1beta2.PrecheckItemType_BINLOG_ENABLE_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-whether-mysql-binlog-is-enabled",
	v1beta2.PrecheckItemType_BINLOG_FORMAT_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-whether-mysql-binlog_format-is-row",
	v1beta2.PrecheckItemType_BINLOG_ROW_IMAGE_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-whether-mysql-binlog_row_image-is-full",
	v1beta2.PrecheckItemType_TABLE_SCHEMA_CHECKING: premiumDmDocURL,
	v1beta2.PrecheckItemType_BINLOG_DB_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-whether-migrated-dbs-are-in-binlog_do_dbbinlog_ignore_db",
	v1beta2.PrecheckItemType_CONN_NUMBER_CHECKING: premiumDmTroubleshootingDocURL +
		"#error-message-check-if-connection-concurrency-exceeds-databases-maximum-connection-limit",
	v1beta2.PrecheckItemType_TARGET_DB_PRIVILEGE_CHECKING: premiumDmDocURL +
		"#grant-required-privileges-to-the-downstream-tidb-cloud-cluster",
	v1beta2.PrecheckItemType_META_POSITION_CHECKING: premiumDmDocURL,
	v1beta2.PrecheckItemType_PRIMARY_KEY_CHECKING: "https://docs.pingcap.com/tidb/dev/dm-faq/" +
		"#does-dm-support-migrating-data-from-alibaba-rds-or-other-cloud-databases",
	v1beta2.PrecheckItemType_LIGHTNING_TABLE_EMPTY_CHECKING: premiumDmDocURL,
}

// premiumPrecheckTypeToV1beta2 maps domain PremiumPrecheckItemType to the current public API enum (v1beta2).
var premiumPrecheckTypeToV1beta2 = map[entity.PremiumPrecheckItemType]v1beta2.PrecheckItemType{
	entity.PremiumPrecheckItemTypeDumpPrivilege:        v1beta2.PrecheckItemType_DUMP_PRIVILEGE_CHECKING,
	entity.PremiumPrecheckItemTypeReplicationPrivilege: v1beta2.PrecheckItemType_REPLICATION_PRIVILEGE_CHECKING,
	entity.PremiumPrecheckItemTypeVersion:              v1beta2.PrecheckItemType_VERSION_CHECKING,
	entity.PremiumPrecheckItemTypeServerID:             v1beta2.PrecheckItemType_SERVER_ID_CHECKING,
	entity.PremiumPrecheckItemTypeBinlogEnable:         v1beta2.PrecheckItemType_BINLOG_ENABLE_CHECKING,
	entity.PremiumPrecheckItemTypeBinlogFormat:         v1beta2.PrecheckItemType_BINLOG_FORMAT_CHECKING,
	entity.PremiumPrecheckItemTypeBinlogRowImage:       v1beta2.PrecheckItemType_BINLOG_ROW_IMAGE_CHECKING,
	entity.PremiumPrecheckItemTypeTableSchema:          v1beta2.PrecheckItemType_TABLE_SCHEMA_CHECKING,
	entity.PremiumPrecheckItemTypeBinlogDB:             v1beta2.PrecheckItemType_BINLOG_DB_CHECKING,
	entity.PremiumPrecheckItemTypeConnNumber:           v1beta2.PrecheckItemType_CONN_NUMBER_CHECKING,
	entity.PremiumPrecheckItemTypeTargetDBPrivilege:    v1beta2.PrecheckItemType_TARGET_DB_PRIVILEGE_CHECKING,
	entity.PremiumPrecheckItemTypeMetaPosition:         v1beta2.PrecheckItemType_META_POSITION_CHECKING,
	entity.PremiumPrecheckItemTypeLightningTableEmpty:  v1beta2.PrecheckItemType_LIGHTNING_TABLE_EMPTY_CHECKING,
	entity.PremiumPrecheckItemTypePrimaryKey:           v1beta2.PrecheckItemType_PRIMARY_KEY_CHECKING,
}

func premiumPrecheckItemTypeForAPI(item *entity.PremiumDMPrecheckItem) v1beta2.PrecheckItemType {
	if item == nil {
		return v1beta2.PrecheckItemType_PRECHECK_ITEM_TYPE_UNSPECIFIED
	}
	if item.Type != entity.PremiumPrecheckItemTypeUnspecified {
		if t, ok := premiumPrecheckTypeToV1beta2[item.Type]; ok {
			return t
		}
	}
	if k, ok := modelpremiumdm.PremiumPrecheckItemTypeForDMCheckerTaskName(item.Description); ok {
		if t, ok2 := premiumPrecheckTypeToV1beta2[k]; ok2 {
			return t
		}
	}
	if _, ok := v1beta2.PrecheckItemType_name[int32(item.Type)]; ok && item.Type != 0 {
		return v1beta2.PrecheckItemType(item.Type)
	}
	return v1beta2.PrecheckItemType_PRECHECK_ITEM_TYPE_UNSPECIFIED
}

// CreateV1beta1TaskReqToDMTask converts v1beta1 CreateMigrationReq to entity DMTask
func CreateV1beta1TaskReqToDMTask(
	ctx context.Context,
	req *v1beta2.CreatePremiumMigrationReq,
	cluster *tidbmgmtservicev1.Cluster,
	endpointMap map[uint64]*entity.BizSinkEndpoint,
) (*entity.PremiumDMTask, bizerr.BizError) {
	mode, err := ConvertV1beta1ModeToDMJobMode(req.GetMode())
	if err != nil {
		return nil, err
	}
	user, userErr := bizutils.GetUserFromContext(ctx)
	if userErr != nil {
		return nil, bizerr.New(bizerr.ErrCodePermission, "user not in context")
	}
	taskID := entity.GenerateDMIDWithPrefix(entity.PremiumDMTaskPrefix)

	return &entity.PremiumDMTask{
		ID:                      taskID,
		Name:                    req.GetDisplayName(),
		Stage:                   entity.PremiumDMTaskStageSourceCreating,
		TiDBID:                  req.GetTidbId(),
		TenantID:                strconv.FormatUint(cluster.OrganizationId, 10),
		ProjectID:               strconv.FormatUint(cluster.ProjectId, 10),
		CreatorID:               user.ID,
		Creator:                 user.Email,
		CloudProvider:           cluster.GetCloudProvider(),
		Region:                  cluster.GetRegion(),
		Mode:                    mode,
		TargetUser:              req.GetTarget().GetUser(),
		TargetEncryptedPassword: req.GetTarget().GetPassword(),
		ImportMode:              V1beta1ImportModeToDMJobImportMode(req.GetImportMode()),
		DMSources: V1beta1SourcesToEntitySources(cluster.OrganizationId,
			cluster.ProjectId,
			cluster.GetClusterId(),
			taskID,
			user.Email,
			req.GetSources(),
			endpointMap),
	}, nil
}

// CreateV1beta2TaskReqToPremiumDMPrecheck converts v1beta2 CreatePremiumMigrationReq to entity PremiumDMPrecheck
func CreateV1beta2TaskReqToPremiumDMPrecheck(
	ctx context.Context,
	req *v1beta2.CreatePremiumMigrationReq,
	cluster *tidbmgmtservicev1.Cluster,
	skipConfigs config.PrecheckSkipConfigs,
) (*entity.PremiumDMPrecheck, bizerr.BizError) {
	mode, err := ConvertV1beta1ModeToDMJobMode(req.GetMode())
	if err != nil {
		return nil, err
	}

	precheckID := entity.GenerateDMIDWithPrefix("pck")
	reqBytes, marshalErr := protojson.Marshal(req)
	if marshalErr != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "failed to marshal request: %v", marshalErr)
	}

	hasCheckpoint := false
	needsPrimaryKeyCheck := false
	for _, source := range req.GetSources() {
		if source == nil {
			continue
		}
		if source.GetSourceType() == v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL {
			needsPrimaryKeyCheck = true
		}
		if source.GetBinlogName() != "" || source.GetBinlogGtid() != "" {
			hasCheckpoint = true
		}
	}

	return &entity.PremiumDMPrecheck{
		ID:        precheckID,
		TiDBID:    req.GetTidbId(),
		TenantID:  strconv.FormatUint(cluster.OrganizationId, 10),
		ProjectID: strconv.FormatUint(cluster.ProjectId, 10),
		Request:   reqBytes,
		TotalItems: int32(len(GetPremiumPrecheckItemList(
			mode, hasCheckpoint,
			V1beta1ImportModeToDMJobImportMode(req.GetImportMode()),
			needsPrimaryKeyCheck,
			skipConfigs,
			cluster.OrganizationId, cluster.ProjectId, cluster.ClusterId,
		))),
		Stage: string(entity.PremiumDMPrecheckStagePending),
	}, nil
}

// V1beta1SourcesToEntitySources converts v1beta1 Source slice to entity DMSource slice
func V1beta1SourcesToEntitySources(
	orgId,
	projectId,
	TiDBID uint64,
	taskId,
	creator string,
	sources []*v1beta2.PremiumDMSource,
	endpointMap map[uint64]*entity.BizSinkEndpoint,
) []entity.PremiumDMSource {
	if len(sources) == 0 {
		return nil
	}
	result := make([]entity.PremiumDMSource, 0, len(sources))
	for _, s := range sources {
		if s == nil {
			continue
		}
		result = append(result, entity.PremiumDMSource{
			ID:             entity.GenerateDMIDWithPrefix(entity.PremiumDMSourcePrefix),
			Stage:          entity.PremiumDMSourceStageCreating,
			Creator:        creator,
			TiDBID:         strconv.FormatUint(TiDBID, 10),
			TenantID:       strconv.FormatUint(orgId, 10),
			SourceType:     v1beta1SourceTypeToEntity(s.GetSourceType()),
			DMTaskID:       taskId,
			ConnProfile:    V1beta1ConnProfileToEntity(s.GetConnProfile(), endpointMap),
			MigrationRules: V1beta1MigrationRulesToEntity(s.GetPremiumMigrationRules()),
			BinlogName:     s.GetBinlogName(),
			BinlogPos:      s.GetBinlogPosition(),
			BinlogGTID:     s.GetBinlogGtid(),
		})
	}
	return result
}

// V1beta1ConnProfileToEntity converts v1beta1 ConnProfile to entity ConnProfile
func V1beta1ConnProfileToEntity(
	pb *v1beta2.PremiumDMConnProfile,
	endpointMap map[uint64]*entity.BizSinkEndpoint,
) entity.PremiumConnProfile {
	if pb == nil {
		return entity.PremiumConnProfile{}
	}
	switch pb.ConnType {
	case v1beta2.PremiumDMConnType_PUBLIC:
		return entity.PremiumConnProfile{
			Type:     entity.PremiumConnTypePublic,
			Target:   pb.GetHost(),
			Port:     pb.GetPort(),
			User:     pb.GetUser(),
			Password: pb.GetPassword(),
			Security: V1beta1SecurityToEntity(pb.GetSecurity()),
		}
	case v1beta2.PremiumDMConnType_PRIVATELINK:
		sinkID := pb.GetSinkEndpointId()
		target := ""
		if ep, ok := endpointMap[sinkID]; ok && ep != nil {
			target = ep.EndpointAddress
		}
		return entity.PremiumConnProfile{
			Type:           entity.PremiumConnTypePrivateLink,
			Target:         target,
			SinkEndpointID: sinkID,
			Port:           pb.GetPort(),
			User:           pb.GetUser(),
			Password:       pb.GetPassword(),
			Security:       V1beta1SecurityToEntity(pb.GetSecurity()),
		}
	}
	return entity.PremiumConnProfile{}
}

// V1beta1SecurityToEntity converts v1beta1 Security to entity Security
func V1beta1SecurityToEntity(pb *v1beta2.Security) *entity.PremiumSecurity {
	if pb == nil {
		return nil
	}
	return &entity.PremiumSecurity{
		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.PremiumDMTask, endpointMap map[uint64]*entity.BizSinkEndpoint,
) (*v1beta2.PremiumMigration, error) {
	if task == nil {
		return nil, nil
	}
	pbTask := &v1beta2.PremiumMigration{
		MigrationId: task.ID,
		DisplayName: task.Name,
		Mode:        DMJobModeToV1beta1Mode(task.Mode),
		TargetUser:  task.TargetUser,
		CreateTime:  timestamppb.New(task.CreatedAt),
		State:       DMTaskStageToV1beta1(task.Stage),
		ImportMode:  DMJobImportModeToV1beta1ImportMode(task.ImportMode),
	}
	pbTask.SubTasks = make([]*v1beta2.PremiumDMSubTask, 0, len(task.DMSources))
	for i := range task.DMSources {
		pbSource := buildV1beta1SourceFromEntity(&task.DMSources[i], endpointMap)
		pbSubTask := &v1beta2.PremiumDMSubTask{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.PremiumDMSource to v1beta2.Source
func buildV1beta1SourceFromEntity(
	s *entity.PremiumDMSource,
	endpointMap map[uint64]*entity.BizSinkEndpoint,
) *v1beta2.PremiumDMSource {
	if s == nil {
		return &v1beta2.PremiumDMSource{}
	}
	src := &v1beta2.PremiumDMSource{
		ConnProfile: EntityConnProfileToV1beta1(&s.ConnProfile, endpointMap),
		SourceType:  entitySourceType2V1beta1(s.SourceType),
	}

	if len(s.MigrationRules) > 0 {
		src.PremiumMigrationRules = EntityMigrationRulesToV1beta1(s.MigrationRules)
	} else {
		src.PremiumMigrationRules = 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.PremiumBlockAllowRules) []*v1beta2.PremiumMigrationRule {
	res := make([]*v1beta2.PremiumMigrationRule, 0)
	for _, db := range rules.DoDBs {
		res = append(res, &v1beta2.PremiumMigrationRule{
			SourceTable: &v1beta2.PremiumMigrationRule_Source{
				SchemaPattern: db,
			},
			TargetTable: &v1beta2.PremiumMigrationRule_Table{
				Schema: db,
			},
		})
	}
	for _, table := range rules.DoTables {
		r := &v1beta2.PremiumMigrationRule{
			SourceTable: &v1beta2.PremiumMigrationRule_Source{
				SchemaPattern: table.Schema,
				TablePattern:  table.Table,
			},
			TargetTable: &v1beta2.PremiumMigrationRule_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.PremiumSubTaskStatus,
	pbSubTask *v1beta2.PremiumDMSubTask,
	mode entity.PremiumDMTaskMode,
) 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 == entity.PremiumDMTaskModeAll {
		ensureV1beta1StepCompletionForModeAll(pbSubTask)
	}
	pbSubTask.ErrorMsg = st.ErrorMsg
	pbSubTask.Stage = convertStageToV1beta1(st.Stage)
	if pbSubTask.ErrorMsg != nil && *pbSubTask.ErrorMsg != "" {
		pbSubTask.Stage = v1beta2.PremiumDMSubTask_FAILED
	}
	return nil
}

// ensureV1beta1StepCompletionForModeAll ensures dump/load show 100% once the pipeline advances beyond them.
func ensureV1beta1StepCompletionForModeAll(pbSubTask *v1beta2.PremiumDMSubTask) {
	const progressComplete float64 = 100
	switch pbSubTask.CurrentStep {
	case v1beta2.PremiumDMSubTask_SYNC:
		if pbSubTask.LoadDetail != nil {
			pbSubTask.LoadDetail.Progress = progressComplete
			pbSubTask.LoadDetail.FinishedBytes = pbSubTask.LoadDetail.TotalBytes
		} else {
			pbSubTask.LoadDetail = &v1beta2.LoadDetail{Progress: progressComplete}
		}
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = progressComplete
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &v1beta2.DumpDetail{Progress: progressComplete}
		}
	case v1beta2.PremiumDMSubTask_LOAD:
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = progressComplete
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &v1beta2.DumpDetail{Progress: progressComplete}
		}
	}
}

func DMTaskStageToV1beta1(stage entity.PremiumDMTaskStage) v1beta2.PremiumMigration_State {
	switch stage {
	case entity.PremiumDMTaskStageTaskCreating,
		entity.PremiumDMTaskStageSourceCreating,
		entity.PremiumDMTaskStageSourceBounding:
		return v1beta2.PremiumMigration_CREATING
	case entity.PremiumDMTaskStageStopped:
		return v1beta2.PremiumMigration_PAUSED
	case entity.PremiumDMTaskStagePausing:
		return v1beta2.PremiumMigration_PAUSING
	case entity.PremiumDMTaskStageRunning:
		return v1beta2.PremiumMigration_RUNNING
	case entity.PremiumDMTaskStageResuming:
		return v1beta2.PremiumMigration_RESUMING
	case entity.PremiumDMTaskStageFailed:
		return v1beta2.PremiumMigration_FAILED
	case entity.PremiumDMTaskStageDeleting:
		return v1beta2.PremiumMigration_DELETING
	default:
		return v1beta2.PremiumMigration_STATE_UNSPECIFIED
	}
}

// ConvertV1beta1ModeToDMJobMode converts v1beta1 TaskMode to entity DMTaskMode
func ConvertV1beta1ModeToDMJobMode(mode v1beta2.PremiumDMTaskMode) (entity.PremiumDMTaskMode, bizerr.BizError) {
	switch mode {
	case v1beta2.PremiumDMTaskMode_ALL:
		return entity.PremiumDMTaskModeAll, nil
	case v1beta2.PremiumDMTaskMode_INCREMENTAL:
		return entity.PremiumDMTaskModeIncremental, nil
	default:
		return "", bizerr.New(bizerr.ErrCodeParams, "invalid mode: %s", mode.String())
	}
}

// DMJobModeToV1beta1Mode converts entity DMTaskMode to v1beta1 TaskMode
func DMJobModeToV1beta1Mode(mode entity.PremiumDMTaskMode) v1beta2.PremiumDMTaskMode {
	switch mode {
	case entity.PremiumDMTaskModeAll:
		return v1beta2.PremiumDMTaskMode_ALL
	case entity.PremiumDMTaskModeIncremental:
		return v1beta2.PremiumDMTaskMode_INCREMENTAL
	default:
		return v1beta2.PremiumDMTaskMode_MODE_UNSPECIFIED
	}
}

// DMJobImportModeToV1beta1ImportMode converts entity DMTaskImportMode to v1beta1 TaskImportMode
func DMJobImportModeToV1beta1ImportMode(mode entity.PremiumDMTaskImportMode) v1beta2.PremiumDMTaskImportMode_ENUM {
	switch mode {
	case entity.PremiumDMTaskImportModeLogical:
		return v1beta2.PremiumDMTaskImportMode_IMPORTMODE_LOGICAL
	case entity.PremiumDMTaskImportModePhysical:
		return v1beta2.PremiumDMTaskImportMode_IMPORTMODE_PHYSICAL
	default:
		return v1beta2.PremiumDMTaskImportMode_IMPORTMODE_UNSPECIFIED
	}
}

func V1beta1ImportModeToDMJobImportMode(mode v1beta2.PremiumDMTaskImportMode_ENUM) entity.PremiumDMTaskImportMode {
	switch mode {
	case v1beta2.PremiumDMTaskImportMode_IMPORTMODE_LOGICAL:
		return entity.PremiumDMTaskImportModeLogical
	case v1beta2.PremiumDMTaskImportMode_IMPORTMODE_PHYSICAL:
		return entity.PremiumDMTaskImportModePhysical
	default:
		return entity.PremiumDMTaskImportModeLogical
	}
}

// EntityConnProfileToV1beta1 converts entity ConnProfile to v1beta1 ConnProfile.
// endpointMap is used to look up the sink endpoint name for PRIVATELINK connections.
func EntityConnProfileToV1beta1(
	cp *entity.PremiumConnProfile,
	endpointMap map[uint64]*entity.BizSinkEndpoint,
) *v1beta2.PremiumDMConnProfile {
	switch cp.Type {
	case entity.PremiumConnTypePublic:
		return &v1beta2.PremiumDMConnProfile{
			ConnType: v1beta2.PremiumDMConnType_PUBLIC,
			Endpoint: &v1beta2.PremiumDMConnProfile_Host{Host: cp.Target},
			Port:     cp.Port,
			User:     cp.User,
			Password: "",
			Security: EntitySecurityToV1beta1(cp.Security),
		}
	case entity.PremiumConnTypePrivateLink:
		endpointName := ""
		if ep, ok := endpointMap[cp.SinkEndpointID]; ok && ep != nil {
			endpointName = ep.Name
		}
		return &v1beta2.PremiumDMConnProfile{
			ConnType: v1beta2.PremiumDMConnType_PRIVATELINK,
			Endpoint: &v1beta2.PremiumDMConnProfile_SinkEndpointId{SinkEndpointId: cp.SinkEndpointID},
			Port:     cp.Port,
			User:     cp.User,
			Password: "",
			Security: EntitySecurityToV1beta1(cp.Security),
			PrivatelinkInfo: &dataflow.PrivatelinkInfo{
				DfPrivatelinkEndpointId: cp.SinkEndpointID,
				EndpointName:            endpointName,
				EndpointAddress:         cp.Target,
			},
		}
	}
	return &v1beta2.PremiumDMConnProfile{}
}

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

// V1beta1MigrationRulesToEntity converts v1beta2 MigrationRules slice to entity MigrationRules
func V1beta1MigrationRulesToEntity(pb []*v1beta2.PremiumMigrationRule) entity.PremiumMigrationRules {
	if len(pb) == 0 {
		return nil
	}
	rules := make(entity.PremiumMigrationRules, 0, len(pb))
	for _, r := range pb {
		if r == nil {
			continue
		}
		rule := entity.PremiumMigrationRule{}
		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.PremiumMigrationRules) []*v1beta2.PremiumMigrationRule {
	if len(rules) == 0 {
		return nil
	}
	ret := make([]*v1beta2.PremiumMigrationRule, 0, len(rules))
	for _, r := range rules {
		pb := &v1beta2.PremiumMigrationRule{}
		// allocate nested messages only if fields are non-empty
		if r.Source.SchemaPattern != "" || r.Source.TablePattern != "" {
			pb.SourceTable = &v1beta2.PremiumMigrationRule_Source{
				SchemaPattern: r.Source.SchemaPattern,
				TablePattern:  r.Source.TablePattern,
			}
		}
		pb.TargetTable = entityMigrationRuleTargetToV1beta1(r.Target)
		ret = append(ret, pb)
	}
	return ret
}

func entityMigrationRuleTargetToV1beta1(target entity.PremiumMigrationRuleTarget) *v1beta2.PremiumMigrationRule_Table {
	if target.Schema == "" && target.Table == "" {
		return nil
	}
	return &v1beta2.PremiumMigrationRule_Table{Schema: target.Schema, Table: target.Table}
}

func v1beta1MigrationRuleTargetToEntity(
	pbTarget *v1beta2.PremiumMigrationRule_Table,
) entity.PremiumMigrationRuleTarget {
	if pbTarget == nil {
		return entity.PremiumMigrationRuleTarget{}
	}
	return entity.PremiumMigrationRuleTarget{Schema: pbTarget.GetSchema(), Table: pbTarget.GetTable()}
}

func entitySourceType2V1beta1(src entity.PremiumDBType) v1beta2.PremiumDMSourceType {
	switch src {
	case entity.PremiumDBTypeMySQL:
		return v1beta2.PremiumDMSourceType_MYSQL
	case entity.PremiumDBTypeAzureFlexibleMySQL:
		return v1beta2.PremiumDMSourceType_AZURE_FLEXIBLE_MYSQL
	case entity.PremiumDBTypeAliCloudRDSMySQL:
		return v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL
	default:
		return v1beta2.PremiumDMSourceType_SOURCE_TYPE_UNSPECIFIED
	}
}

func v1beta1SourceTypeToEntity(src v1beta2.PremiumDMSourceType) entity.PremiumDBType {
	switch src {
	case v1beta2.PremiumDMSourceType_MYSQL:
		return entity.PremiumDBTypeMySQL
	case v1beta2.PremiumDMSourceType_AZURE_FLEXIBLE_MYSQL:
		return entity.PremiumDBTypeAzureFlexibleMySQL
	case v1beta2.PremiumDMSourceType_ALICLOUD_RDS_MYSQL:
		return entity.PremiumDBTypeAliCloudRDSMySQL
	default:
		return ""
	}
}

// ParseProgressString parses a progress string like "100.00  %" or "85.5 %" to float64.
func ParseProgressString(progress string) (float64, error) {
	clean := strings.ReplaceAll(progress, " ", "")
	clean = strings.TrimSuffix(clean, "%")
	if clean == "" {
		return 0, nil
	}
	return strconv.ParseFloat(clean, 64)
}

// ConvertDumpStatusToV1beta1 converts internal DumpStatus to v1beta1 DumpDetail
func ConvertDumpStatusToV1beta1(ctx context.Context, ds *dm.DumpStatus) (*v1beta2.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 &v1beta2.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) (*v1beta2.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 &v1beta2.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) *v1beta2.SyncDetail {
	if ss == nil {
		return nil
	}
	return &v1beta2.SyncDetail{Rps: ss.RecentTps, Latency: ss.SecondsBehindMaster, Checkpoint: ss.SyncerBinlog}
}

// ConvertToV1beta1Step maps unit string to v1beta2.SubTask_Step
func ConvertToV1beta1Step(unit string) v1beta2.PremiumDMSubTask_Step {
	switch strings.ToLower(unit) {
	case "dump":
		return v1beta2.PremiumDMSubTask_DUMP
	case "sync":
		return v1beta2.PremiumDMSubTask_SYNC
	case "load":
		return v1beta2.PremiumDMSubTask_LOAD
	default:
		return v1beta2.PremiumDMSubTask_STEP_UNSPECIFIED
	}
}

// GetPremiumPrecheckItemList returns the list of precheck items to run based on migration mode and binlog checkpoint presence.
func GetPremiumPrecheckItemList(
	mode entity.PremiumDMTaskMode,
	hasCheckpoint bool,
	importMode entity.PremiumDMTaskImportMode,
	needsPrimaryKeyCheck bool,
	skipConfigs config.PrecheckSkipConfigs,
	tenantID, projectID, clusterID uint64,
) []v1beta2.PrecheckItemType {
	precheckTypes := modelpremiumdm.PremiumPrecheckItemTypesForCluster(
		mode, hasCheckpoint, importMode, needsPrimaryKeyCheck, skipConfigs, tenantID, projectID, clusterID,
	)
	if len(precheckTypes) == 0 {
		return []v1beta2.PrecheckItemType{}
	}
	itemTypes := make([]v1beta2.PrecheckItemType, 0, len(precheckTypes))
	for _, typ := range precheckTypes {
		if t, ok := premiumPrecheckTypeToV1beta2[typ]; ok {
			itemTypes = append(itemTypes, t)
		}
	}
	return itemTypes
}

// convertStageToV1beta1 converts dm.TaskStage to v1beta1 SubTask_Stage
func convertStageToV1beta1(stage dm.TaskStage) v1beta2.PremiumDMSubTask_Stage {
	switch stage {
	case dm.TaskStageFinished:
		return v1beta2.PremiumDMSubTask_FINISHED
	case dm.TaskStagePaused, dm.TaskStageStopped:
		return v1beta2.PremiumDMSubTask_PAUSED
	case dm.TaskStageRunning:
		return v1beta2.PremiumDMSubTask_RUNNING
	default:
		return v1beta2.PremiumDMSubTask_UNKNOWN
	}
}

func dmPrecheckStageToV1beta1(stage entity.PremiumDMPrecheckStage) v1beta2.PremiumMigrationPrecheck_Status {
	switch stage {
	case entity.PremiumDMPrecheckStagePending:
		return v1beta2.PremiumMigrationPrecheck_PENDING
	case entity.PremiumDMPrecheckStageRunning:
		return v1beta2.PremiumMigrationPrecheck_RUNNING
	case entity.PremiumDMPrecheckStageFinished:
		return v1beta2.PremiumMigrationPrecheck_FINISHED
	case entity.PremiumDMPrecheckStageFailed:
		return v1beta2.PremiumMigrationPrecheck_FAILED
	case entity.PremiumDMPrecheckStageCanceled:
		return v1beta2.PremiumMigrationPrecheck_CANCELED
	default:
		return v1beta2.PremiumMigrationPrecheck_STATUS_UNSPECIFIED
	}
}

func PremiumDMPrecheckToV1beta2PbPrecheck(precheck *entity.PremiumDMPrecheck) *v1beta2.PremiumMigrationPrecheck {
	if precheck == nil {
		return nil
	}
	resp := &v1beta2.PremiumMigrationPrecheck{
		PrecheckId: precheck.ID,
		Status:     dmPrecheckStageToV1beta1(entity.PremiumDMPrecheckStage(precheck.Stage)),
		Total:      precheck.TotalItems,
	}
	var failedCnt, warnCnt, successCnt int32
	for i := range precheck.DMPrecheckItems {
		item := &precheck.DMPrecheckItems[i]
		if item.Stage == string(entity.PremiumDMPrecheckStageFailed) {
			itemType := premiumPrecheckItemTypeForAPI(item)
			resp.Items = append(resp.Items, &v1beta2.PremiumDMPrecheckItem{
				Description:    item.Description,
				Status:         v1beta2.PremiumDMPrecheckItem_FAILED,
				Solution:       "Precheck(id = " + precheck.ID + ") failed, please contact the support team",
				Reason:         "internal error: precheck item failed unexpectedly",
				SolutionDocUrl: premiumPrecheckItemHelpURL(itemType),
				Type:           itemType,
			})
			failedCnt++
			continue
		} else if item.Stage == string(entity.PremiumDMPrecheckStageRunning) {
			continue
		}
		respItem := formatPremiumCheckItemResult(item)
		if respItem == nil {
			continue
		}
		resp.Items = append(resp.Items, respItem)
		switch respItem.Status {
		case v1beta2.PremiumDMPrecheckItem_FAILED:
			failedCnt++
		case v1beta2.PremiumDMPrecheckItem_WARNING:
			warnCnt++
		case v1beta2.PremiumDMPrecheckItem_SUCCESS:
			successCnt++
		}
	}
	resp.FailedCnt = failedCnt
	resp.WarnCnt = warnCnt
	resp.SuccessCnt = successCnt
	if failedCnt > 0 {
		resp.Status = v1beta2.PremiumMigrationPrecheck_FAILED
	}
	return resp
}

// PremiumDMPrecheckToV1beta2PbSummary converts a single precheck record into a
// lightweight per-row summary used by ListPremiumMigrationPrechecks. Counts
// mirror PremiumDMPrecheckToV1beta2PbPrecheck so listing and detail agree.
func PremiumDMPrecheckToV1beta2PbSummary(precheck *entity.PremiumDMPrecheck) *v1beta2.PremiumMigrationPrecheckSummary {
	if precheck == nil {
		return nil
	}
	summary := &v1beta2.PremiumMigrationPrecheckSummary{
		PrecheckId: precheck.ID,
		CreatedAt:  timestamppb.New(precheck.CreatedAt),
		Status:     dmPrecheckStageToV1beta1(entity.PremiumDMPrecheckStage(precheck.Stage)),
		Total:      precheck.TotalItems,
	}
	var failedCnt, warnCnt, successCnt int32
	for i := range precheck.DMPrecheckItems {
		item := &precheck.DMPrecheckItems[i]
		if item.Stage == string(entity.PremiumDMPrecheckStageFailed) {
			failedCnt++
			continue
		}
		if item.Stage == string(entity.PremiumDMPrecheckStageRunning) {
			continue
		}
		respItem := formatPremiumCheckItemResult(item)
		if respItem == nil {
			continue
		}
		switch respItem.Status {
		case v1beta2.PremiumDMPrecheckItem_FAILED:
			failedCnt++
		case v1beta2.PremiumDMPrecheckItem_WARNING:
			warnCnt++
		case v1beta2.PremiumDMPrecheckItem_SUCCESS:
			successCnt++
		}
	}
	summary.FailedCnt = failedCnt
	summary.WarnCnt = warnCnt
	summary.SuccessCnt = successCnt
	return summary
}

func formatPremiumCheckItemResult(item *entity.PremiumDMPrecheckItem) *v1beta2.PremiumDMPrecheckItem {
	summary, ok := item.SummarizeResults()
	if !ok {
		return nil
	}
	description := summary.Desc
	if description == modelpremiumdm.PremiumDMPrecheckInitializationDescForAPI() && item.Description != "" {
		description = item.Description
	}
	ret := &v1beta2.PremiumDMPrecheckItem{Description: description}
	ret.Status = premiumCheckerStateToV1beta2(summary.State)
	ret.Type = premiumPrecheckItemTypeForAPI(item)
	if len(summary.Reasons) > 0 {
		ret.Reason = strings.Join(summary.Reasons, "; ")
	}
	if len(summary.Instructions) > 0 {
		ret.Solution = strings.Join(summary.Instructions, "; ")
	}
	if ret.Status != v1beta2.PremiumDMPrecheckItem_SUCCESS {
		ret.SolutionDocUrl = premiumPrecheckItemHelpURL(ret.Type)
	}
	return ret
}

func premiumPrecheckItemHelpURL(itemType v1beta2.PrecheckItemType) string {
	if url, ok := premiumPrecheckItemHelpURLs[itemType]; ok {
		return url
	}
	return premiumDmDocURL
}

func premiumCheckerStateToV1beta2(state checker.State) v1beta2.PremiumDMPrecheckItem_Status {
	switch state {
	case checker.StateFailure:
		return v1beta2.PremiumDMPrecheckItem_FAILED
	case checker.StateWarning:
		return v1beta2.PremiumDMPrecheckItem_WARNING
	case checker.StateSuccess:
		return v1beta2.PremiumDMPrecheckItem_SUCCESS
	default:
		return v1beta2.PremiumDMPrecheckItem_STATUS_UNSPECIFIED
	}
}