package vo

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

	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless/dm"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"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"
)

// CreateConsoleTaskReqToDMTask converts console CreateDMTaskReq to entity DMTask
func CreateConsoleTaskReqToDMTask(
	ctx context.Context,
	req *console.CreateMigrationTaskReq,
	cluster *tidbmgmtservicev1.Cluster,
) (*entity.DMTask, bizerr.BizError) {
	mode, bizErr := ConvertConsoleModeToDMJobMode(req.GetMode())
	if bizErr != nil {
		return nil, bizErr
	}
	taskID := entity.GenerateIDWithPrefix(constant.DMTaskIDPrefix)
	creator := utils.BuildCreatorID(ctx)
	clusterID := req.GetClusterId()
	orgID := req.GetOrgId()

	return &entity.DMTask{
		ID:                      taskID,
		Name:                    req.GetName(),
		Stage:                   constant.DMTaskStageSourceCreating,
		Creator:                 creator,
		ClusterID:               clusterID,
		TenantID:                orgID,
		CloudProvider:           cluster.GetCloudProvider(),
		Region:                  cluster.GetRegion(),
		Mode:                    mode,
		TargetUser:              req.GetTarget().GetUser(),
		TargetEncryptedPassword: req.GetTarget().GetPassword(),
		DMSources: ConsoleSourcesToEntitySources(
			orgID,
			clusterID,
			creator,
			taskID,
			req.GetSources(),
		),
	}, nil
}

// CreateConsoleTaskReqToDMPrecheck converts console CreateDMTaskReq to entity DMPrecheck
func CreateConsoleTaskReqToDMPrecheck(
	ctx context.Context,
	req *console.CreateMigrationTaskReq,
	getEndpointDomain 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()
	orgID := req.GetOrgId()

	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:                  ConvertConsoleDMMode(req.Mode),
			FullInstanceMigration: req.FullInstanceMigration,
		},
		ClusterId: clusterID,
	}
	sources, bizErr := ConsoleSourcesToDMPrecheckSources(ctx, req.Sources, clusterID, getEndpointDomain)
	if bizErr != nil {
		return nil, bizErr
	}
	precheckReq.DmPrecheck.Sources = sources

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

// ConsoleSourcesToEntitySources converts console Source slice to entity DMSource slice
func ConsoleSourcesToEntitySources(
	orgID string,
	clusterID string,
	creator string,
	taskID string,
	sources []*console.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,
			DMTaskID:        taskID,
			Creator:         creator,
			ClusterID:       clusterID,
			TenantID:        orgID,
			ConnProfile:     ConsoleConnProfileToEntity(s.GetConnProfile()),
			BlockAllowRules: ConsoleBlockAllowRulesToEntity(s.GetBaRules()),
			SourceType:      consoleSourceTypeIdToEntity(s.GetSourceTypeId()),
			BinlogName:      s.GetBinlogName(),
			BinlogPos:       s.GetBinlogPos(),
			BinlogGTID:      s.GetBinlogGtid(),
		})
	}
	return result
}

// ConsoleConnProfileToEntity converts console ConnProfile to entity ConnProfile
func ConsoleConnProfileToEntity(pb *console.ConnProfile) entity.ConnProfile {
	if pb == nil {
		return entity.ConnProfile{}
	}

	switch pb.ConnType {
	case console.ConnType_PUBLIC:
		return entity.ConnProfile{
			Type:     constant.ConnTypePublic,
			Target:   pb.GetHost(),
			Port:     pb.GetPort(),
			User:     pb.GetUser(),
			Password: pb.GetPassword(),
			Security: ConsoleSecurityToEntity(pb.GetSecurity()),
		}
	case console.ConnType_PRIVATE_LINK:
		return entity.ConnProfile{
			Type:     constant.ConnTypePrivateLink,
			Target:   pb.GetEndpointId(),
			Port:     pb.GetPort(),
			User:     pb.GetUser(),
			Password: pb.GetPassword(),
			Security: ConsoleSecurityToEntity(pb.GetSecurity()),
		}
	}

	return entity.ConnProfile{}
}

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

// ConsoleBlockAllowRulesToEntity converts console BlockAllowRules to entity BlockAllowRules
func ConsoleBlockAllowRulesToEntity(pb *console.BlockAllowRules) entity.BlockAllowRules {
	if pb == nil {
		return entity.BlockAllowRules{}
	}
	doTables := make([]entity.Table, 0, len(pb.GetDoTables()))
	for _, t := range pb.GetDoTables() {
		if t == nil {
			continue
		}
		doTables = append(doTables, entity.Table{
			Schema: t.GetSchema(),
			Table:  t.GetTable(),
		})
	}
	return entity.BlockAllowRules{
		DoDBs:    pb.GetDoDbs(),
		DoTables: doTables,
	}
}

// DMTaskAndStatusToConsolePbTask converts entity DMTask and status to console DMTask
func DMTaskAndStatusToConsolePbTask(ctx context.Context, task *entity.DMTask) (*console.MigrationTask, error) {
	if task == nil {
		return nil, nil
	}
	pbTask := &console.MigrationTask{
		Id:         task.ID,
		Name:       task.Name,
		Mode:       DMJobModeToConsoleMode(task.Mode),
		TargetUser: task.TargetUser,
		CreateTime: timestamppb.New(task.CreatedAt),
		State:      DMTaskStageToConsole(task.Stage),
	}

	pbTask.SubTasks = make([]*console.SubTask, 0, len(task.DMSources))
	for i := range task.DMSources {
		// Build base source info
		pbSource := buildConsoleSourceFromEntity(&task.DMSources[i])

		pbSubTask := &console.SubTask{Source: pbSource}
		// Fill runtime status if available
		if task.DMSources[i].SubTaskStatus != nil {
			if err := fillSubTaskStatus(ctx, task.DMSources[i].SubTaskStatus, pbSubTask, task.Mode); err != nil {
				return nil, err
			}
		}
		pbTask.SubTasks = append(pbTask.SubTasks, pbSubTask)
	}
	return pbTask, nil
}

// buildConsoleSourceFromEntity converts an entity.DMSource to console.Source, including optional binlog fields.
func buildConsoleSourceFromEntity(s *entity.DMSource) *console.Source {
	if s == nil {
		return &console.Source{}
	}
	src := &console.Source{
		ConnProfile:  EntityConnProfileToConsole(&s.ConnProfile),
		BaRules:      EntityBlockAllowRulesToConsole(s.BlockAllowRules, s.MigrationRules),
		SourceTypeId: string(s.SourceType),
	}
	if s.BinlogName != "" {
		src.BinlogName = &s.BinlogName
	}
	if s.BinlogPos != 0 {
		src.BinlogPos = &s.BinlogPos
	}
	if s.BinlogGTID != "" {
		src.BinlogGtid = &s.BinlogGTID
	}
	return src
}

// fillSubTaskStatus populates SubTask runtime details and applies step-completion overrides for MODE_ALL.
func fillSubTaskStatus(
	ctx context.Context,
	st *entity.SubTaskStatus,
	pbSubTask *console.SubTask,
	mode constant.DMTaskMode,
) error {
	var err error
	pbSubTask.LoadDetail, err = ConvertLoadStatusToConsole(ctx, st.LoadStatus)
	if err != nil {
		return err
	}
	pbSubTask.DumpDetail, err = ConvertDumpStatusToConsole(ctx, st.DumpStatus)
	if err != nil {
		return err
	}
	pbSubTask.SyncDetail = ConvertSyncStatusToConsole(st.SyncStatus)
	pbSubTask.CurrentStep = ConvertToConsoleStep(st.Unit)

	// In MODE_ALL, mark previous steps 100% when current step has advanced.
	if mode == constant.DMTaskModeAll {
		ensureStepCompletionForModeAll(pbSubTask)
	}
	pbSubTask.ErrorMsg = st.ErrorMsg
	pbSubTask.Stage = convertStageToConsole(st.Stage)
	if pbSubTask.ErrorMsg != nil && *pbSubTask.ErrorMsg != "" {
		pbSubTask.Stage = console.SubTask_STAGE_FAILED
	}
	return nil
}

// ensureStepCompletionForModeAll ensures dump/load show 100% once the pipeline advances beyond them.
func ensureStepCompletionForModeAll(pbSubTask *console.SubTask) {
	switch pbSubTask.CurrentStep {
	case console.SubTask_STEP_SYNC:
		// LOAD should be complete
		if pbSubTask.LoadDetail != nil {
			pbSubTask.LoadDetail.Progress = 100
			pbSubTask.LoadDetail.FinishedBytes = pbSubTask.LoadDetail.TotalBytes
		} else {
			pbSubTask.LoadDetail = &console.LoadDetail{Progress: 100}
		}
		// DUMP should be complete
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = 100
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &console.DumpDetail{Progress: 100}
		}
	case console.SubTask_STEP_LOAD:
		// DUMP should be complete
		if pbSubTask.DumpDetail != nil {
			pbSubTask.DumpDetail.Progress = 100
			pbSubTask.DumpDetail.CompletedTables = pbSubTask.DumpDetail.TotalTables
		} else {
			pbSubTask.DumpDetail = &console.DumpDetail{Progress: 100}
		}
	}
}

func DMTaskStageToConsole(stage constant.DMTaskStage) console.MigrationTask_State {
	switch stage {
	case constant.DMTaskStageTaskCreating, constant.DMTaskStageSourceCreating, constant.DMTaskStageSourceBounding:
		return console.MigrationTask_CREATING
	case constant.DMTaskStageStopped:
		return console.MigrationTask_PAUSED
	case constant.DMTaskStageRunning:
		return console.MigrationTask_RUNNING
	case constant.DMTaskStageFailed:
		return console.MigrationTask_FAILED
	case constant.DMTaskStageDeleting:
		return console.MigrationTask_DELETING
	default:
		return console.MigrationTask_STATE_UNSPECIFIED
	}
}

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

// DMJobModeToConsoleMode converts entity DMTaskMode to console TaskMode
func DMJobModeToConsoleMode(mode constant.DMTaskMode) console.TaskMode {
	switch mode {
	case constant.DMTaskModeAll:
		return console.TaskMode_MODE_ALL
	case constant.DMTaskModeIncr:
		return console.TaskMode_MODE_INCREMENTAL
	default:
		return console.TaskMode_MODE_UNSPECIFIED
	}
}

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

	return &console.ConnProfile{}
}

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

// EntityBlockAllowRulesToConsole converts entity BlockAllowRules to console BlockAllowRules
func EntityBlockAllowRulesToConsole(bar entity.BlockAllowRules, rules entity.MigrationRules) *console.BlockAllowRules {
	// migration created by openapi, use migration rules to build block/allow rules
	if len(rules) != 0 {
		doTables := make([]*console.BlockAllowRules_Table, 0)
		doDbs := make([]string, 0)
		for _, rule := range rules {
			if rule.Source.TablePattern != "" {
				doTables = append(doTables, &console.BlockAllowRules_Table{
					Schema: rule.Source.SchemaPattern,
					Table:  rule.Source.TablePattern,
				})
			} else {
				doDbs = append(doDbs, rule.Source.SchemaPattern)
			}
		}
		return &console.BlockAllowRules{
			DoDbs:    doDbs,
			DoTables: doTables,
		}
	}

	doTables := make([]*console.BlockAllowRules_Table, 0, len(bar.DoTables))
	for _, t := range bar.DoTables {
		doTables = append(doTables, &console.BlockAllowRules_Table{
			Schema: t.Schema,
			Table:  t.Table,
		})
	}
	return &console.BlockAllowRules{
		DoDbs:    bar.DoDBs,
		DoTables: doTables,
	}
}

func consoleSourceTypeIdToEntity(id string) constant.DBType {
	switch id {
	case string(constant.DBTypeMySQL):
		return constant.DBTypeMySQL
	case string(constant.DBTypeAliCloudRDSMySQL):
		return constant.DBTypeAliCloudRDSMySQL
	default:
		return constant.DBType(id)
	}
}

// ConvertDumpStatusToConsole converts internal DumpStatus to console DumpDetail
func ConvertDumpStatusToConsole(ctx context.Context, ds *dm.DumpStatus) (*console.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 &console.DumpDetail{
		Bps:             ds.Bps,
		TotalTables:     ds.TotalTables,
		CompletedTables: int64(ds.CompletedTables),
		Progress:        p,
		FinishedBytes:   int64(ds.FinishedBytes),
		FinishedRows:    int64(ds.FinishedRows),
	}, nil
}

// ConvertLoadStatusToConsole converts internal LoadStatus to console LoadDetail
func ConvertLoadStatusToConsole(ctx context.Context, ls *dm.LoadStatus) (*console.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 &console.LoadDetail{
		Bps:           ls.Bps,
		Progress:      p,
		FinishedBytes: ls.FinishedBytes,
		TotalBytes:    ls.TotalBytes,
	}, nil
}

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

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

func ConvertToConsoleStep(unit string) console.SubTask_Step {
	switch strings.ToLower(unit) {
	case "dump":
		return console.SubTask_STEP_DUMP
	case "sync":
		return console.SubTask_STEP_SYNC
	case "load":
		return console.SubTask_STEP_LOAD
	default:
		return console.SubTask_STEP_UNSPECIFIED
	}
}

// convertStageToConsole converts dm.TaskStage to console SubTask_Stage
func convertStageToConsole(stage dm.TaskStage) console.SubTask_Stage {
	switch stage {
	case dm.TASKSTAGE_FINISHED:
		return console.SubTask_STAGE_FINISHED
	case dm.TASKSTAGE_PAUSED, dm.TASKSTAGE_STOPPED:
		return console.SubTask_STAGE_PAUSED
	case dm.TASKSTAGE_RUNNING:
		return console.SubTask_STAGE_RUNNING
	default:
		return console.SubTask_STAGE_UNKNOWN
	}
}

// ConsoleSourcesToDMPrecheckSources converts console Source slice to DMPrecheck Sources
func ConsoleSourcesToDMPrecheckSources(
	ctx context.Context,
	pbSources []*console.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{}

		// Convert connection profile
		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.ConnType == console.ConnType_PRIVATE_LINK {
				d, err := getEndpointDomain(ctx, clusterId, pbSource.ConnProfile.GetEndpointId())
				if err != nil {
					return nil, err
				}
				source.ConnProfile.Host = d
			}
			if pbSource.SourceTypeId == string(constant.DBTypeAliCloudRDSMySQL) {
				source.NeedPkCheck = true
			}

			// Convert security settings if present
			if pbSource.ConnProfile.Security != nil {
				source.ConnProfile.Security = &regionalserverlessv1.DMPrecheck_Security{
					SslCaContent:   pbSource.ConnProfile.Security.SslCaContent,
					SslCertContent: pbSource.ConnProfile.Security.SslCertContent,
					SslKeyContent:  pbSource.ConnProfile.Security.SslKeyContent,
					CertAllowedCn:  pbSource.ConnProfile.Security.CertAllowedCn,
				}
			}
		}

		// Convert block/allow rules
		if pbSource.BaRules != nil {
			source.Rules = &regionalserverlessv1.DMPrecheck_BlockAllowRules{
				DoTables: convertConsoleTableSchemas(pbSource.BaRules.DoTables),
				DoDbs:    pbSource.BaRules.DoDbs,
			}
		}

		// Convert binlog settings
		source.BinlogName = pbSource.BinlogName
		source.BinlogPos = pbSource.BinlogPos
		source.BinlogGtid = pbSource.BinlogGtid

		sources = append(sources, source)
	}

	return sources, nil
}

// ConnProfileToConnectionModel converts console MigrationSourceDBConnProfile to clusterproxy.ConnectionModel
func ConnProfileToConnectionModel(
	ctx context.Context,
	clusterId string,
	cp *console.MigrationSourceDBConnProfile,
	getEndpointDomain func(c context.Context, clusterID string, endpointID string) (string, bizerr.BizError),
) (*clusterproxy.ConnectionModel, bizerr.BizError) {
	if cp == nil || cp.GetConnProfile() == nil {
		return nil, nil
	}

	params := &clusterproxy.ConnectionModel{
		Type:       "mysql",
		User:       cp.GetConnProfile().GetUser(),
		Password:   cp.GetConnProfile().GetPassword(),
		MysqlParam: clusterproxy.MysqlParameter{},
	}

	enableTls := false
	if cp.GetConnProfile().GetSecurity() != nil {
		if len(cp.GetConnProfile().GetSecurity().GetSslCaContent()) != 0 {
			params.MysqlParam.SSLCABytes = cp.GetConnProfile().GetSecurity().GetSslCaContent()
			enableTls = true
		}
		if len(cp.GetConnProfile().GetSecurity().GetSslCertContent()) != 0 {
			params.MysqlParam.SSLCertBytes = cp.GetConnProfile().GetSecurity().GetSslCertContent()
			enableTls = true
		}
		if len(cp.GetConnProfile().GetSecurity().GetSslKeyContent()) != 0 {
			params.MysqlParam.SSLKeyBytes = cp.GetConnProfile().GetSecurity().GetSslKeyContent()
			enableTls = true
		}
	}
	params.EnableTls = enableTls

	// Build URL according to conn type. For private link, we may need to describe vpc endpoint to get the domain.
	if cp.GetConnProfile().GetConnType() == console.ConnType_PRIVATE_LINK {
		domain, err := getEndpointDomain(ctx, clusterId, cp.GetConnProfile().GetEndpointId())
		if err != nil {
			return nil, err
		}

		params.Url = fmt.Sprintf("%s:%d", domain, cp.GetConnProfile().GetPort())
	} else if cp.GetConnProfile().GetConnType() == console.ConnType_PUBLIC {
		params.Url = fmt.Sprintf("%s:%d", cp.GetConnProfile().GetHost(), cp.GetConnProfile().GetPort())
	}

	return params, nil
}

// convertConsoleTableSchemas converts console Table slice to regional Table slice
func convertConsoleTableSchemas(pbTables []*console.BlockAllowRules_Table) []*regionalserverlessv1.Table {
	if len(pbTables) == 0 {
		return nil
	}

	tables := make([]*regionalserverlessv1.Table, 0, len(pbTables))
	for _, pbTable := range pbTables {
		if pbTable != nil {
			tables = append(tables, &regionalserverlessv1.Table{
				Schema: pbTable.Schema,
				Table:  pbTable.Table,
			})
		}
	}
	return tables
}

// ConvertConsoleDMMode converts console TaskMode to regional DM mode
func ConvertConsoleDMMode(mode console.TaskMode) regionalserverlessv1.DMPrecheck_TaskMode {
	switch mode {
	case console.TaskMode_MODE_ALL:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_ALL
	case console.TaskMode_MODE_INCREMENTAL:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_INCREMENTAL
	default:
		return regionalserverlessv1.DMPrecheck_TASK_MODE_UNSPECIFIED
	}
}

// GetPrecheckItemList returns the list of precheck items based on task mode
func GetPrecheckItemList(precheckReq *entity.PrecheckRequest) []regionalserverlessv1.DMPrecheck_PrecheckItem {
	items := make([]regionalserverlessv1.DMPrecheck_PrecheckItem, 0)
	for _, s := range precheckReq.DmPrecheck.Sources {
		if s.NeedPkCheck {
			items = append(
				items,
				regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_PRIMARY_KEY_CHECKING,
			)
			break
		}
	}
	switch precheckReq.DmPrecheck.Mode {
	case regionalserverlessv1.DMPrecheck_TASK_MODE_ALL:
		return append(items, constant.AllModeCheckList...)
	case regionalserverlessv1.DMPrecheck_TASK_MODE_INCREMENTAL:
		items = append(items, constant.IncrementModeCheckList...)
		for _, s := range precheckReq.DmPrecheck.Sources {
			if s.GetBinlogGtid() != "" || s.GetBinlogName() != "" {
				return append(
					items,
					regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_META_POSITION_CHECKING,
				)
			}
		}
		return items
	default:
		return []regionalserverlessv1.DMPrecheck_PrecheckItem{}
	}
}

// DMPrecheckToConsolePbPrecheck converts entity DMPrecheck to console DMPrecheck
func DMPrecheckToConsolePbPrecheck(precheck *entity.DMPrecheck) *console.MigrationPrecheck {
	if precheck == nil {
		return nil
	}

	resp := &console.MigrationPrecheck{}
	resp.Status = string(precheck.Stage)
	resp.Total = precheck.TotalItems
	var finishedCnt, 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 := formatConsoleCheckItemResult(item); respItem != nil {
				resp.Items = append(resp.Items, respItem)
				failedCnt++
				finishedCnt++
				continue
			}
			resp.Items = append(resp.Items, &console.PrecheckItem{
				Desc:           item.Description,
				Status:         string(item.Stage),
				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:           ConvertConsolePrecheckItemType(item.Type),
			})
			failedCnt++
			finishedCnt++
			continue
		} else if item.Stage == constant.DMPrecheckStageRunning {
			continue
		}
		respItem := formatConsoleCheckItemResult(item)
		if respItem == nil {
			// should not happen
			continue
		}
		resp.Items = append(resp.Items, respItem)
		switch respItem.Status {
		case string(checker.StateFailure):
			failedCnt++
		case string(checker.StateWarning):
			warnCnt++
		case string(checker.StateSuccess):
			successCnt++
		}
		finishedCnt++
	}
	resp.FailedCnt = failedCnt
	resp.WarnCnt = warnCnt
	resp.SuccessCnt = successCnt
	if failedCnt > 0 {
		resp.Status = string(constant.DMPrecheckStageFailed)
	}
	return resp
}

// formatConsoleCheckItemResult formats a precheck item result for console response
func formatConsoleCheckItemResult(item *entity.DMPrecheckItem) *console.PrecheckItem {
	raw := item.Results
	if len(raw.Results) == 0 {
		return nil
	}

	first := raw.Results[0]
	ret := &console.PrecheckItem{
		Desc: first.Desc,
	}

	// initial status
	status := string(first.State)

	// collect reasons (short errors) and instructions
	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))
	}

	// compact subsequent results into the collections without mutating original structs
	for i := 1; i < len(raw.Results); i++ {
		r := raw.Results[i]
		switch r.State {
		case checker.StateFailure:
			status = string(checker.StateFailure)
		case checker.StateWarning:
			if status != string(checker.StateFailure) {
				status = string(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))
		}
	}

	// compose final fields
	ret.Status = status
	if len(reasons) > 0 {
		ret.Reason = strings.Join(reasons, "; ")
	}
	if len(instrs) > 0 {
		ret.Solution = strings.Join(instrs, "; ")
	}

	// only set SolutionDocUrl when not success and mapping exists
	if checker.State(ret.Status) != checker.StateSuccess {
		if url, ok := constant.PrecheckItemHelpUrls[item.Type]; ok {
			ret.SolutionDocUrl = url
		}
	}

	// Limit lengths of Reason and Solution to avoid extremely long responses.
	// Keep the same maximum for both fields.
	const maxReasonSolutionLen = 1024
	truncate := func(s string, n int) string {
		rs := []rune(s)
		if len(rs) <= n {
			return s
		}
		return string(rs[:n])
	}
	if ret.Reason != "" {
		ret.Reason = truncate(ret.Reason, maxReasonSolutionLen)
	}
	if ret.Solution != "" {
		ret.Solution = truncate(ret.Solution, maxReasonSolutionLen)
	}

	ret.Type = ConvertConsolePrecheckItemType(item.Type)
	return ret
}

// ConvertConsolePrecheckItemType converts regional precheck item type to console precheck item type
func ConvertConsolePrecheckItemType(item regionalserverlessv1.DMPrecheck_PrecheckItem) console.PrecheckItemType {
	switch item {
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_DUMP_PRIVILEGE_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_DUMP_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_REPLICATION_PRIVILEGE_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_REPLICATION_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_VERSION_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_VERSION_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_SERVER_ID_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_SERVER_ID_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_ENABLE_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_BINLOG_ENABLE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_FORMAT_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_BINLOG_FORMAT_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_ROW_IMAGE_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_BINLOG_ROW_IMAGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_TABLE_SCHEMA_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_TABLE_SCHEMA_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_BINLOG_DB_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_BINLOG_DB_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_CONN_NUMBER_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_CONN_NUMBER_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_TARGET_DB_PRIVILEGE_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_TARGET_DB_PRIVILEGE_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_META_POSITION_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_META_POSITION_CHECKING
	case regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_PRIMARY_KEY_CHECKING:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_PRIMARY_KEY_CHECKING
	default:
		return console.PrecheckItemType_PRECHECK_ITEM_TYPE_UNSPECIFIED
	}
}

func ConvertEndpointState(state regionalserverlessv1.VpcEndpointState) constant.DMVpcEndpointState {
	switch state {
	case regionalserverlessv1.VpcEndpointState_VPC_ENDPOINT_STATE_NOT_READY:
		return constant.DMVpcEndpointStatePending
	case regionalserverlessv1.VpcEndpointState_VPC_ENDPOINT_STATE_ACTIVE:
		return constant.DMVpcEndpointStateAvailable
	case regionalserverlessv1.VpcEndpointState_VPC_ENDPOINT_STATE_DELETED:
		return constant.DMVpcEndpointStateDeleted
	case regionalserverlessv1.VpcEndpointState_VPC_ENDPOINT_STATE_PENDING_APPROVE:
		return constant.DMVpcEndpointStatePendingAcceptance
	case regionalserverlessv1.VpcEndpointState_VPC_ENDPOINT_STATE_FAILED:
		return constant.DMVpcEndpointStateFailed
	default:
		return constant.DMVpcEndpointStatePending
	}
}

func DescribeEndpointRespFromEntity(endpoint *entity.DMVpcEndpoint) *console.DescribeEndpointResp {
	if endpoint == nil {
		return &console.DescribeEndpointResp{}
	}

	var phase console.DescribeEndpointResp_Phase
	switch endpoint.State {
	case constant.DMVpcEndpointStatePending:
		phase = console.DescribeEndpointResp_Pending
	case constant.DMVpcEndpointStatePendingAcceptance:
		phase = console.DescribeEndpointResp_PendingAcceptance
	case constant.DMVpcEndpointStateAvailable:
		phase = console.DescribeEndpointResp_Available
	case constant.DMVpcEndpointStateFailed:
		phase = console.DescribeEndpointResp_Failed
	default:
		phase = console.DescribeEndpointResp_Pending
	}

	resp := &console.DescribeEndpointResp{
		Phase:        phase,
		ErrorMessage: "",
		EndpointId:   endpoint.EndpointId,
		ServiceName:  endpoint.ServiceName,
	}

	if endpoint.CreationTime != "" {
		if t, terr := time.Parse(time.RFC3339, endpoint.CreationTime); terr == nil {
			resp.CreateTime = timestamppb.New(t)
		}
	}

	return resp
}