package console

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"github.com/AlekSi/pointer"
	mapset "github.com/deckarep/golang-set/v2"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
)

var (
	ListSupportedOrderByField = []string{"create_time"}
	ListOrderByFieldMap       = map[string]string{
		"create_time": "created_at",
	}

	awsSupportedSources = []*console.SupportedSource{
		{
			Id:          string(constant.DBTypeMySQL),
			DisplayName: constant.DBTypeMySQL.DisplayName(),
			ConnMethods: []*console.ConnMethod{
				{
					ConnType:    console.ConnType_PUBLIC,
					DisplayName: "Public",
				},
				{
					ConnType:    console.ConnType_PRIVATE_LINK,
					DisplayName: "Private Link",
				},
			},
		},
	}

	aliCloudSupportedSources = []*console.SupportedSource{
		{
			Id:          string(constant.DBTypeMySQL),
			DisplayName: constant.DBTypeMySQL.DisplayName(),
			ConnMethods: []*console.ConnMethod{
				{
					ConnType:    console.ConnType_PUBLIC,
					DisplayName: "Public",
				},
				{
					ConnType:    console.ConnType_PRIVATE_LINK,
					DisplayName: "Private Link",
				},
			},
		},
		{
			Id:          string(constant.DBTypeAliCloudRDSMySQL),
			DisplayName: constant.DBTypeAliCloudRDSMySQL.DisplayName(),
			ConnMethods: []*console.ConnMethod{
				{
					ConnType:    console.ConnType_PUBLIC,
					DisplayName: "Public",
				},
				{
					ConnType:    console.ConnType_PRIVATE_LINK,
					DisplayName: "Private Link",
				},
			},
		},
	}
)

var Tracer = otel.Tracer("dm.console")

type DMService struct {
	mgmtSvc                    facade.MgmtSvcFacade
	importSvc                  service.ImportService
	dmService                  service.DMService
	encryptorFacade            facade.EncryptorFacade
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	privateConnSvc             service.PrivateLinkConnectionService
	console.UnimplementedMigrationServiceServer
}

func NewDMService(
	mgmtSvc facade.MgmtSvcFacade,
	importSvc service.ImportService,
	dmService service.DMService,
	encryptorFacade facade.EncryptorFacade,
	privateConnSvc service.PrivateLinkConnectionService,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
) console.MigrationServiceServer {
	return &DMService{
		mgmtSvc:                    mgmtSvc,
		importSvc:                  importSvc,
		dmService:                  dmService,
		encryptorFacade:            encryptorFacade,
		privateConnSvc:             privateConnSvc,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
	}
}

func (s *DMService) Precheck(
	ctx context.Context,
	req *console.CreateMigrationTaskReq,
) (_ *console.CreateMigrationPrecheckResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.PrecheckTask")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("project_id", req.ProjectId),
		zap.String("cluster_id", req.ClusterId),
	)

	if bizErr := validateCreateTaskReq(req); bizErr != nil {
		return nil, bizErr
	}
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	err = s.encryptPasswords(logger, req)
	if err != nil {
		return nil, err
	}
	precheck, bizErr := vo.CreateConsoleTaskReqToDMPrecheck(
		ctx,
		req,
		s.getEndpointDomain(),
	)
	if bizErr != nil {
		logger.Error("failed to convert request to precheck entity", zap.Error(bizErr))
		return nil, bizErr
	}
	err = s.dmService.CreatePrecheck(ctx, precheck)
	if err != nil {
		return nil, err
	}

	return &console.CreateMigrationPrecheckResp{
		Id: precheck.ID,
	}, nil
}

func (s *DMService) CreateTask(
	ctx context.Context,
	req *console.CreateMigrationTaskReq,
) (resp *console.MigrationTask, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.CreateTask")
	var cluster *tidbmgmtservicev1.Cluster
	var dmTask *entity.DMTask
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
		go recordCreateMigrationTaskAuditLog(ctx, req, cluster, dmTask, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("project_id", req.ProjectId),
		zap.String("cluster_id", req.ClusterId),
	)

	// Validate the request
	if bizErr := validateCreateTaskReq(req); bizErr != nil {
		return nil, bizErr
	}

	var bizErr bizerr.BizError
	cluster, bizErr = s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	logger.Info("before encryptPasswords", zap.Any("req", req))
	err = s.encryptPasswords(logger, req)
	if err != nil {
		return nil, err
	}
	logger.Info("after encryptPasswords", zap.Any("req", req))
	task, bizErr := vo.CreateConsoleTaskReqToDMTask(ctx, req, cluster)
	if bizErr != nil {
		return nil, bizErr
	}
	logger.Info("CreateTask", zap.Any("task", task))
	dmTask, bizErr = s.dmService.CreateTask(ctx, task)
	if bizErr != nil {
		return nil, bizErr
	}

	return vo.DMTaskAndStatusToConsolePbTask(ctx, dmTask)
}

func (s *DMService) GetTask(
	ctx context.Context,
	req *console.GetMigrationTaskReq,
) (_ *console.MigrationTask, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.GetTask")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("task_id", req.Id),
	)
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	task, bizErr := s.dmService.GetTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	return vo.DMTaskAndStatusToConsolePbTask(ctx, task)
}

func (s *DMService) ListTasks(
	ctx context.Context,
	req *console.ListMigrationTasksReq,
) (_ *console.ListMigrationTasksResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.ListTasks")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	filter := &entity.DMTaskFilter{
		ClusterID: req.ClusterId,
	}
	pager := utils.NewPagerNumberPager()
	err = pager.Normalize(req)
	if err != nil {
		return nil, err
	}
	filter.Offset = pointer.To(int(pager.Offset()))
	filter.Limit = pointer.To(int(pager.PageSize()))

	// check orderBy
	err = utils.CheckOrderBy(req, ListSupportedOrderByField)
	if err != nil {
		return nil, bizerr.New(
			errorcode.ErrCodeInvalidOrderby,
			"%s",
			errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidOrderby]+" Error: "+err.Error(),
		)
	}
	orderBy := req.OrderBy
	for k, v := range ListOrderByFieldMap {
		orderBy = strings.ReplaceAll(orderBy, k, v)
	}
	filter.OrderBy = orderBy
	filter.Stages = constant.NonDeletedDMTaskStages

	span.AddEvent("call ListTasks")
	total, tasks, bizErr := s.dmService.ListTasks(ctx, filter)
	if bizErr != nil {
		return nil, bizErr
	}
	nextPageToken := pager.NextPageToken(uint32(total))
	resp := &console.ListMigrationTasksResp{
		Tasks:         make([]*console.MigrationTask, 0, len(tasks)),
		TotalSize:     uint32(total),
		NextPageToken: nextPageToken,
	}
	for _, t := range tasks {
		task, err := vo.DMTaskAndStatusToConsolePbTask(ctx, t)
		if err != nil {
			return nil, err
		}
		resp.Tasks = append(resp.Tasks, task)
	}
	return resp, nil
}

func (s *DMService) DeleteTask(
	ctx context.Context,
	req *console.DeleteMigrationTaskReq,
) (resp *console.MigrationTask, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.DeleteTask")
	var task *entity.DMTask
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
		go recordDeleteMigrationTaskAuditLog(ctx, req, task, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("task_id", req.Id),
	)
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	task, bizErr = s.dmService.GetTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	bizErr = s.dmService.DeleteTask(ctx, req.ClusterId, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	return vo.DMTaskAndStatusToConsolePbTask(ctx, task)
}

func (s *DMService) PauseTask(
	ctx context.Context,
	req *console.PauseMigrationTaskReq,
) (resp *console.EmptyResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.PauseTask")
	var task *entity.DMTask
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
		go recordPauseMigrationTaskAuditLog(ctx, req, task, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("task_id", req.Id),
	)
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	task, bizErr = s.dmService.GetTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	bizErr = s.dmService.PauseTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}
	return &console.EmptyResp{}, nil
}

func (s *DMService) ResumeTask(
	ctx context.Context,
	req *console.ResumeMigrationTaskReq,
) (resp *console.EmptyResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.ResumeTask")
	var task *entity.DMTask
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
		go recordResumeMigrationTaskAuditLog(ctx, req, task, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("task_id", req.Id),
	)
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	task, bizErr = s.dmService.GetTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	bizErr = s.dmService.ResumeTask(ctx, cluster, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}
	return &console.EmptyResp{}, nil
}

func (s *DMService) CancelPrecheck(
	ctx context.Context,
	req *console.CancelMigrationPrecheckReq,
) (_ *console.EmptyResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.CancelPrecheck")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()

	logger := log.WithContextL(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("project_id", req.ProjectId),
		zap.String("cluster_id", req.ClusterId),
		zap.String("precheck_id", req.Id),
	)

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	bizErr = s.dmService.CancelPrecheck(ctx, req.ClusterId, req.Id)
	if bizErr != nil {
		logger.Error("failed to cancel precheck", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("precheck canceled successfully")
	return &console.EmptyResp{}, nil
}

func (s *DMService) GetPrecheck(
	ctx context.Context,
	req *console.GetMigrationPrecheckReq,
) (_ *console.MigrationPrecheck, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.GetPrecheck")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()

	precheck, bizErr := s.dmService.GetPrecheck(ctx, req.ClusterId, req.Id)
	if bizErr != nil {
		return nil, bizErr
	}

	return vo.DMPrecheckToConsolePbPrecheck(precheck), nil
}

func (s *DMService) encryptPasswords(logger *zap.Logger, req *console.CreateMigrationTaskReq) error {
	encrypted, err := s.encryptorFacade.EncryptBase64(req.GetTarget().GetPassword())
	if err != nil {
		logger.Error("failed to encrypt target database password", zap.Error(err))
		return err
	}
	req.Target.Password = encrypted

	for i := range req.Sources {
		if req.Sources[i] == nil || req.Sources[i].GetConnProfile() == nil {
			continue
		}
		var encrypted string
		encrypted, err = s.encryptorFacade.EncryptBase64(req.Sources[i].GetConnProfile().GetPassword())
		if err != nil {
			logger.Error("failed to encrypt source database password", zap.Error(err))
			return err
		}
		req.Sources[i].ConnProfile.Password = encrypted
	}
	return nil
}

func (s *DMService) GetSupportedSources(
	ctx context.Context,
	req *console.GetSupportedSourcesReq,
) (_ *console.GetSupportedSourcesResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.GetSupportedSources")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()

	switch req.Provider {
	case console.GetSupportedSourcesReq_CLOUD_PROVIDER_ALICLOUD:
		return &console.GetSupportedSourcesResp{
			Sources: aliCloudSupportedSources,
		}, nil
	case console.GetSupportedSourcesReq_CLOUD_PROVIDER_AWS:
		return &console.GetSupportedSourcesResp{
			Sources: awsSupportedSources,
		}, nil
	}

	return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported cloud provider: %s", req.Provider.String())
}

func (s *DMService) GetSourceTables(
	ctx context.Context,
	req *console.MigrationSourceDBConnProfile,
) (_ *console.GetSourceTablesResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.GetSourceTables")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	if bizErr := validateGetSourceTablesReq(req); bizErr != nil {
		return nil, bizErr
	}

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}

	namespace, service := constant.DMProxyTarget()
	proxy, bizErr := s.clusterProxyProviderFacade.GetServerlessHTTPProxy(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint(),
		namespace,
		service,
	)
	if bizErr != nil {
		logger.Error("failed to get cluster proxy", zap.Error(bizErr))
		return nil, bizErr
	}

	params, bizErr := vo.ConnProfileToConnectionModel(
		ctx,
		req.ClusterId,
		req,
		s.getEndpointDomain(),
	)
	if bizErr != nil {
		logger.Error("failed to build connection params", zap.Error(bizErr))
		return nil, bizErr
	}

	tables, err := proxy.GetSourceTables(ctx, params)
	if err != nil {
		return nil, formatDMSourceConnectionErrorContent(ctx, err)
	}

	respTables := make([]*console.GetSourceTablesResp_Table, 0, len(tables))
	for _, t := range tables {
		respTables = append(respTables, &console.GetSourceTablesResp_Table{
			Schema: t.Schema,
			Table:  t.Table,
		})
	}

	return &console.GetSourceTablesResp{
		Tables: respTables,
	}, nil
}

func (s *DMService) ValidateMigrationSourceConn(
	ctx context.Context,
	req *console.MigrationSourceDBConnProfile,
) (_ *console.EmptyResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.ValidateDMSourceConn")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	if bizErr := validateMigrationSourceConnReq(req); bizErr != nil {
		return nil, bizErr
	}

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}
	namespace, service := constant.DMProxyTarget()
	proxy, bizErr := s.clusterProxyProviderFacade.GetServerlessHTTPProxy(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint(),
		namespace,
		service,
	)
	if bizErr != nil {
		logger.Error("failed to get cluster proxy", zap.Error(bizErr))
		return nil, bizErr
	}

	params, bizErr := vo.ConnProfileToConnectionModel(
		ctx,
		req.ClusterId,
		req,
		s.getEndpointDomain(),
	)
	if bizErr != nil {
		logger.Error("failed to build connection params", zap.Error(bizErr))
		return nil, bizErr
	}

	err = proxy.TestConnection(ctx, params)
	if err != nil {
		return nil, formatDMSourceConnectionErrorContent(ctx, err)
	}
	return &console.EmptyResp{}, nil
}

func formatDMSourceConnectionErrorContent(ctx context.Context, err error) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Error("formatDMSourceConnectionErrorContent", zap.Error(err))
	// Get the raw error content
	content := err.Error()
	var bizErr bizerr.BizError
	isFormattedError := false

	// Check if error is already a BizError
	if errors.As(err, &bizErr) {
		content = bizErr.ErrMsg()
		isFormattedError = true
	}

	// Handle timeout errors
	if strings.Contains(content, "i/o timeout") || strings.Contains(content, "no such host") {
		return bizerr.New(errorcode.ErrCodeSourceNetworkUnreachable, "")
	}

	// Handle test failure errors
	const testFailedPrefix = "Request fail, 400-"
	if strings.HasPrefix(content, testFailedPrefix) {
		content = strings.TrimPrefix(content, testFailedPrefix)
		isFormattedError = true
	}

	if isFormattedError {
		return bizerr.New(bizerr.ErrCodeParams, "connect to source failed, reason: %s", content)
	}
	// For unknown errors, return unknown error
	return bizerr.New(bizerr.ErrCodeUnknown, "unknown error: %s", content)
}

func (s *DMService) ValidateMigrationTargetConn(
	ctx context.Context,
	req *console.MigrationTargetDBConnProfile,
) (_ *console.EmptyResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.ValidateDMTargetConn")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	if bizErr := validateMigrationTargetConnReq(req); bizErr != nil {
		return nil, bizErr
	}

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}
	if !strings.HasPrefix(req.User, cluster.GetServerlessCluster().GetCluster().UserPrefix) {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"Incorrect user name prefix. See https://docs.pingcap.com/tidbcloud/select-cluster-tier#user-name-prefix",
		)
	}

	operator := &clusterproxy.ClusterOperator{
		Username:   req.User,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: req.Password,
	}
	proxy, bizErr := s.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if bizErr != nil {
		logger.Error("failed to get cluster proxy", zap.Error(bizErr))
		return nil, bizErr
	}

	_, err = proxy.QuerySQL(ctx, "SELECT 1")
	if err != nil {
		logger.Error("failed to test connection", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "Failed to connect to target database, please check your password")
	}

	return &console.EmptyResp{}, nil
}

func (s *DMService) CreateVpcEndpoint(
	ctx context.Context,
	req *console.CreateEndpointReq,
) (_ *console.CreateEndpointResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.CreateVpcInterfaceEndpoint")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("service_name", req.ServiceName),
	)
	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	if bizErr := isWhitelisted(cluster.OrganizationId, cluster.CloudProvider, cluster.Region); bizErr != nil {
		return nil, bizErr
	}
	if bizErr := validateCreateEndpointReq(req); bizErr != nil {
		return nil, bizErr
	}

	endpointId, bizErr := s.dmService.CreateVpcEndpoint(ctx, req.ClusterId, req.ServiceName)
	if bizErr != nil {
		return nil, bizErr
	}
	return &console.CreateEndpointResp{
		EndpointId: endpointId,
	}, nil
}

func (s *DMService) DescribeVpcEndpoint(
	ctx context.Context,
	req *console.DescribeEndpointReq,
) (_ *console.DescribeEndpointResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.DescribeVpcEndpoint")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()

	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("endpoint_id", req.EndpointId),
	)

	endpoint, bizErr := s.dmService.DescribeVpcEndpoint(ctx, req.ClusterId, req.EndpointId)
	if bizErr != nil {
		logger.Error("failed to describe vpc endpoint", zap.Error(bizErr))
		return nil, bizErr
	}

	return vo.DescribeEndpointRespFromEntity(endpoint), nil
}

func (s *DMService) GetVpcInfo(ctx context.Context, req *console.GetVpcInfoReq) (_ *console.GetVpcInfoResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.dm.GetVpcInfo")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, logger)
	if bizErr != nil {
		return nil, bizErr
	}
	p, err := s.importSvc.GetDataIngestionProfile(ctx, cluster)
	if err != nil {
		return nil, bizErr
	}

	var azs []string
	private := cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPrivate()
	switch cluster.CloudProvider {
	case config.ProviderAliCloud:
		azs = private.GetAlicloud().GetAvailabilityZone()
	case config.ProviderAWS:
		azs = private.GetAws().GetAvailabilityZone()
	case config.ProviderGCP:
		// Cluster_Endpoints_Private_GCP only carries the PSC service
		// attachment name; there is no AZ list on the cluster endpoint.
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported cloud provider: %s", cluster.CloudProvider)
	}
	return &console.GetVpcInfoResp{
		AccountId:         p.AccountId,
		AvailabilityZones: azs,
	}, nil
}

func (s *DMService) GetIpWhitelist(
	ctx context.Context,
	req *console.GetIpWhitelistReq,
) (*console.GetIpWhitelistResp, error) {
	return &console.GetIpWhitelistResp{
		Ips: []string{"0.0.0.0"},
	}, nil
}

// isWhitelisted checks whether migration is allowed for the given tenant and region.
//
// Rules:
//   - If both tenant and region whitelists are empty, allow all.
//   - If a whitelist dimension is empty, it is treated as "no restriction" for that dimension.
//   - Otherwise, the request is allowed only if BOTH conditions hold:
//     tenant is in tenant whitelist AND provider-region is in region whitelist.
func isWhitelisted(tenantID uint64, provider, region string) bizerr.BizError {
	tenantWhiteSet := mapset.NewSet(config.Global().BizConfig.MigrationWhitelistTenants...)
	regionWhiteSet := mapset.NewSet(config.Global().BizConfig.MigrationWhitelistRegions...)
	regionKey := fmt.Sprintf("%s-%s", provider, region)

	// No whitelist configured: allow all
	if tenantWhiteSet.IsEmpty() && regionWhiteSet.IsEmpty() {
		return nil
	}

	// Empty list means no restriction for that dimension
	tenantAllowed := tenantWhiteSet.IsEmpty() || tenantWhiteSet.Contains(tenantID)
	regionAllowed := regionWhiteSet.IsEmpty() || regionWhiteSet.Contains(regionKey)

	if tenantAllowed && regionAllowed {
		return nil
	}
	return bizerr.New(
		bizerr.ErrCodeParams,
		"migration service is not available for your organization in the current region",
	)
}

func (s *DMService) getEndpointDomain() func(c context.Context, clusterID string, endpointID string) (string, bizerr.BizError) {
	return func(c context.Context, clusterID, endpointID string) (string, bizerr.BizError) {
		if strings.HasPrefix(endpointID, constant.PrivateLinkConnectionIDPrefix) {
			domains, err := s.privateConnSvc.RegisterPrivateLinkConnection(c, clusterID, endpointID, "", true)
			if err != nil {
				return "", err
			}
			if len(domains) == 0 {
				return "", bizerr.New(
					errorcode.ErrCodeInternalError,
					"no domain found for private link connection %s",
					endpointID,
				)
			}
			return domains[0], nil
		} else if strings.HasPrefix(endpointID, constant.DMVpcEndpointPrefix) {
			e, err := s.dmService.DescribeVpcEndpoint(c, clusterID, endpointID)
			if err != nil {
				return "", err
			}
			return e.Domain, nil
		}
		return "", bizerr.New(bizerr.ErrCodeParams, "invalid private link connection id")
	}
}