package service

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

	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless/dm"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"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/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/utils"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"k8s.io/utils/pointer"
)

var CreateVpcEndpointVisibleErrCodes = []string{
	"EndpointServiceNotAuthorized",
	"EndpointServiceDeleted",
	"IncorrectStatus.VpcEndpointService",
	"EndpointServiceNotFound",
}

type DMService interface {
	CreateTask(ctx context.Context, task *entity.DMTask) (*entity.DMTask, bizerr.BizError)
	GetTask(
		ctx context.Context,
		cluster *tidbmgmtservicev1.Cluster,
		ID string,
	) (*entity.DMTask, bizerr.BizError)
	ListTasks(ctx context.Context, filter *entity.DMTaskFilter) (int64, []*entity.DMTask, bizerr.BizError)
	DeleteTask(ctx context.Context, clusterID string, taskID string) bizerr.BizError
	PauseTask(ctx context.Context, cluster *tidbmgmtservicev1.Cluster, taskID string) bizerr.BizError
	ResumeTask(ctx context.Context, cluster *tidbmgmtservicev1.Cluster, taskID string) bizerr.BizError
	CreatePrecheck(ctx context.Context, precheck *entity.DMPrecheck) bizerr.BizError
	GetPrecheck(ctx context.Context, clusterID, id string) (*entity.DMPrecheck, bizerr.BizError)
	CancelPrecheck(ctx context.Context, clusterID, id string) bizerr.BizError
	CreateVpcEndpoint(ctx context.Context, clusterID, serviceName string) (string, bizerr.BizError)
	// DescribeVpcEndpoint calls mgmt svc to get endpoint details and persists the endpoint record.
	DescribeVpcEndpoint(ctx context.Context, clusterID, endpointId string) (*entity.DMVpcEndpoint, bizerr.BizError)
}

type dmService struct {
	dmTaskRepo                 repo.DMTaskRepo
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	transProvider              db.TransProvider
	lockRepo                   repo.LockRepo
	mgmtSvcFacade              facade.MgmtSvcFacade
	privateLinkConnectionSvc   PrivateLinkConnectionService
	eventbusService            *eventbus.EventbusService
}

func NewDMService(
	dmTaskRepo repo.DMTaskRepo,
	transProvider db.TransProvider,
	lockRepo repo.LockRepo,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	mgmtSvcFacade facade.MgmtSvcFacade,
	privateLinkConnectionSvc PrivateLinkConnectionService,
	eventbusService *eventbus.EventbusService,
) DMService {
	return &dmService{
		dmTaskRepo:                 dmTaskRepo,
		transProvider:              transProvider,
		lockRepo:                   lockRepo,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		mgmtSvcFacade:              mgmtSvcFacade,
		privateLinkConnectionSvc:   privateLinkConnectionSvc,
		eventbusService:            eventbusService,
	}
}

func (s *dmService) exceedNumberLimit(ctx context.Context, tenantId string) bizerr.BizError {
	cfg := config.Global().BizConfig
	var quota int64
	quota, ok := cfg.DMTaskOrgNumberQuota[tenantId]
	if !ok {
		quota = cfg.MaxDMTaskNumberPerOrg
	}
	count, bizErr := s.dmTaskRepo.CountTasks(ctx, &entity.DMTaskFilter{
		TenantID: tenantId,
		Stages:   constant.ActiveDMTaskStages,
	})
	if bizErr != nil {
		return bizErr
	}
	if count >= quota {
		return bizerr.New(errorcode.ErrCodeGreaterThanDMTaskNumbersLimit,
			"This organization can create up to %d DM tasks.", quota)
	}
	return nil
}

func (s *dmService) CreateTask(ctx context.Context, task *entity.DMTask) (_ *entity.DMTask, bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("dm_task_id", task.ID), zap.String("cluster_id", task.ClusterID))
	uint64ClusterId, err := strconv.ParseUint(task.ClusterID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}

	statusID, bizErr := s.mgmtSvcFacade.SetClusterMigrationStatus(ctx, uint64ClusterId)
	if bizErr != nil {
		if bizErr.ErrCode() == CustomStatusMutexErrorCode {
			return nil, bizErr
		}
		logger.Error("failed to set cluster migration status", zap.Error(bizErr))
		return nil, bizErr
	}
	logger.Info("set cluster migration status successfully", zap.Uint64("status_id", statusID))
	task.ClusterStatusID = statusID

	defer func() {
		if bizErr != nil {
			rctx := context.WithoutCancel(ctx)
			err = s.mgmtSvcFacade.DeleteClusterCustomStatus(rctx, uint64ClusterId, statusID)
			if err != nil {
				logger.Error("failed to delete cluster custom status after create dm task failed", zap.Error(err))
			}
			logger.Info("deleted cluster custom status after create dm task failed")
		}
	}()

	var result *entity.DMTask
	err = s.transProvider.Transaction(ctx, func(ctx context.Context) (gerr error) {
		lock := &entity.Lock{
			Key:         fmt.Sprintf("create-dm-org-%s", task.TenantID),
			TimeoutSecs: 10,
		}
		err := s.lockRepo.Lock(ctx, lock)
		if err != nil {
			logger.Error("failed to get lock", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeDB, "failed to create dm task")
		}
		logger.Info("get lock successfully")
		defer func() {
			rctx := context.WithoutCancel(ctx)
			err := s.lockRepo.Release(rctx, lock)
			if err != nil {
				logger.Error("failed to release lock", zap.Error(err))
			} else {
				logger.Info("release lock successfully")
			}
		}()

		if bizErr := s.exceedNumberLimit(ctx, task.TenantID); bizErr != nil {
			return bizErr
		}
		if bizErr := s.tryToRegisterPrivateLink(ctx, task, logger); bizErr != nil {
			return bizErr
		}
		if err := s.dmTaskRepo.CreateTask(ctx, task); err != nil {
			logger.Error("create dm task failed", zap.Error(err))
			return err
		}
		result = task
		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if stderrors.As(err, &e) {
			if e.RawErrCode() == bizerr.ErrCodeRecordExist {
				return nil, bizerr.New(bizerr.ErrCodeParams, "migration name '%s' already exists", task.Name)
			}
			return nil, e
		}
		logger.Error("failed to create dm task", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeDB, "failed to create migration")
	}
	s.publishDMTaskEvent(
		ctx,
		result,
		event_constant.EventTypeCreateMigration,
		fmt.Sprintf("Migration job %s created.", result.Name),
	)
	return result, nil
}

func (s *dmService) tryToRegisterPrivateLink(
	ctx context.Context,
	task *entity.DMTask,
	logger *zap.Logger,
) bizerr.BizError {
	for i := range task.DMSources {
		logger := logger.With(zap.String("dm_source_id", task.DMSources[i].ID))
		if task.DMSources[i].ConnProfile.Type == constant.ConnTypePrivateLink {
			domain, id, bizErr := s.registerPrivateLink(ctx, &task.DMSources[i], logger)
			if bizErr != nil {
				logger.Error("failed to resolve private link host", zap.Error(bizErr))
				return bizErr
			}
			task.DMSources[i].EndpointDomain = domain
			task.DMSources[i].EndpointID = id
			logger.Info(
				"register private link host successfully",
				zap.String("endpoint_domain", domain),
				zap.String("endpoint_id", id),
			)
		}
	}
	return nil
}

func (s *dmService) GetTask(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	taskID string,
) (*entity.DMTask, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("dm_task_id", taskID), zap.Uint64("cluster_id", cluster.ClusterId))
	task, bizErr := s.dmTaskRepo.GetTask(
		ctx,
		strconv.FormatUint(cluster.ClusterId, 10),
		taskID,
		constant.NonDeletedDMTaskStages,
	)
	if bizErr != nil {
		logger.Error("get dm task failed", zap.Error(bizErr))
		return nil, bizErr
	}

	return task, nil
}

// ListTasks returns a list of DMTask entities matching the filter.
func (s *dmService) ListTasks(
	ctx context.Context,
	filter *entity.DMTaskFilter,
) (int64, []*entity.DMTask, bizerr.BizError) {
	tasks, err := s.dmTaskRepo.ListTasks(ctx, filter)
	if err != nil {
		return 0, nil, err
	}
	total, err := s.dmTaskRepo.CountTasks(ctx, filter)
	if err != nil {
		return 0, nil, err
	}
	return total, tasks, nil
}

// DeleteTask performs a soft delete on the DMTask with the given id.
func (s *dmService) DeleteTask(ctx context.Context, clusterID, taskID string) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("dm_task_id", taskID), zap.String("cluster_id", clusterID))

	task, bizErr := s.dmTaskRepo.GetTask(ctx, clusterID, taskID, nil)
	if bizErr != nil {
		logger.Error("get dm task failed", zap.Error(bizErr))
		return bizErr
	}
	if task.Stage == constant.DMTaskStageDeleted || task.Stage == constant.DMTaskStageDeleting {
		return nil
	}

	if err := s.dmTaskRepo.UpdateTask(ctx, task.ID, constant.ActiveDMTaskStages, map[string]any{
		"stage":        constant.DMTaskStageDeleting,
		"deleted_time": time.Now(),
	}); err != nil {
		return err
	}
	s.publishDMTaskEvent(
		ctx,
		task,
		event_constant.EventTypeDeleteMigration,
		fmt.Sprintf("Migration job %s deleted.", task.Name),
	)
	return nil
}

func (s *dmService) PauseTask(ctx context.Context, cluster *tidbmgmtservicev1.Cluster, taskID string) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("dm_task_id", taskID), zap.Uint64("cluster_id", cluster.ClusterId))

	var task *entity.DMTask
	err := s.transProvider.Transaction(ctx, func(ctx context.Context) (gerr error) {
		// load task within transaction to ensure we operate on the same db state
		var bizErr bizerr.BizError
		task, bizErr = s.dmTaskRepo.GetTask(ctx, strconv.FormatUint(cluster.ClusterId, 10), taskID, nil)
		if bizErr != nil {
			logger.Error("get dm task failed", zap.Error(bizErr))
			return bizErr
		}
		if task.Stage != constant.DMTaskStageRunning {
			return bizerr.New(bizerr.ErrCodeParams, "dm task %s is not running", task.ID)
		}

		// update DB first (will be committed only if transaction returns nil)
		if updErr := s.dmTaskRepo.UpdateTask(
			ctx,
			task.ID,
			[]constant.DMTaskStage{constant.DMTaskStageRunning},
			map[string]any{"stage": constant.DMTaskStageStopped},
		); updErr != nil {
			logger.Error("update dm task stage failed", zap.Error(updErr))
			return updErr
		}

		// build source list
		sourceNames := make([]string, 0, len(task.DMSources))
		for i := range task.DMSources {
			sourceNames = append(sourceNames, task.DMSources[i].ID)
		}

		clusterProxyURL := cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
		// get proxy and call stop
		namespace, service := constant.DMProxyTarget()
		proxy, bizErr := s.clusterProxyProviderFacade.GetServerlessHTTPProxy(
			ctx,
			clusterProxyURL,
			namespace,
			service,
		)
		if bizErr != nil {
			logger.Error("get serverless proxy failed", zap.Error(bizErr))
			return bizErr
		}
		if err := proxy.StopServerlessDMTask(ctx, task.ID, &dm.StopTaskRequest{SourceNameList: sourceNames}); err != nil {
			logger.Error("stop dm task failed", zap.Error(err))
			// return error so transaction will rollback
			return bizerr.New(errorcode.ErrCodeInternalError, "failed to stop dm task")
		}
		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if stderrors.As(err, &e) {
			return e
		}
		logger.Error("failed to pause dm task", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeInternalError, "failed to pause dm task")
	}
	s.publishDMTaskEvent(
		ctx,
		task,
		event_constant.EventTypePauseMigration,
		fmt.Sprintf("Migration job %s paused.", task.Name),
	)
	return nil
}

func (s *dmService) ResumeTask(ctx context.Context, cluster *tidbmgmtservicev1.Cluster, taskID string) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("dm_task_id", taskID), zap.Uint64("cluster_id", cluster.ClusterId))

	var task *entity.DMTask
	err := s.transProvider.Transaction(ctx, func(ctx context.Context) (gerr error) {
		// load task within transaction
		var bizErr bizerr.BizError
		task, bizErr = s.dmTaskRepo.GetTask(ctx, strconv.FormatUint(cluster.ClusterId, 10), taskID, nil)
		if bizErr != nil {
			logger.Error("get dm task failed", zap.Error(bizErr))
			return bizErr
		}
		if task.Stage != constant.DMTaskStageStopped && task.Stage != constant.DMTaskStageFailed {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"dm task %s cannot be resumed, current stage: %s",
				task.ID,
				task.Stage,
			)
		}

		// update DB first
		if updErr := s.dmTaskRepo.UpdateTask(
			ctx,
			task.ID,
			[]constant.DMTaskStage{constant.DMTaskStageStopped, constant.DMTaskStageFailed},
			map[string]any{"stage": constant.DMTaskStageRunning},
		); updErr != nil {
			logger.Error("update dm task stage failed", zap.Error(updErr))
			return updErr
		}

		// build source list
		sourceNames := make([]string, 0, len(task.DMSources))
		for i := range task.DMSources {
			sourceNames = append(sourceNames, task.DMSources[i].ID)
		}

		clusterProxyURL := cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint()
		// get proxy and call start
		namespace, service := constant.DMProxyTarget()
		proxy, bizErr := s.clusterProxyProviderFacade.GetServerlessHTTPProxy(
			ctx,
			clusterProxyURL,
			namespace,
			service,
		)
		if bizErr != nil {
			logger.Error("get serverless proxy failed", zap.Error(bizErr))
			return bizErr
		}
		if err := proxy.StartServerlessDMTask(ctx, task.ID, &dm.StartTaskRequest{SourceNameList: sourceNames}); err != nil {
			logger.Error("start dm task failed", zap.Error(err))
			// return error so transaction will rollback
			return bizerr.New(errorcode.ErrCodeInternalError, "failed to start dm task")
		}
		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if stderrors.As(err, &e) {
			return e
		}
		logger.Error("failed to resume dm task", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeInternalError, "failed to resume dm task")
	}
	s.publishDMTaskEvent(
		ctx,
		task,
		event_constant.EventTypeResumeMigration,
		fmt.Sprintf("Migration job %s resumed.", task.Name),
	)
	return nil
}

func (s *dmService) CreatePrecheck(ctx context.Context, precheck *entity.DMPrecheck) bizerr.BizError {
	return s.dmTaskRepo.CreatePrecheck(ctx, precheck)
}

func (s *dmService) GetPrecheck(
	ctx context.Context,
	clusterID, id string,
) (precheck *entity.DMPrecheck, err bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("dm_precheck_id", id), zap.String("cluster_id", clusterID))
	precheck, bizErr := s.dmTaskRepo.GetPrecheck(ctx, clusterID, id)
	if bizErr != nil {
		logger.Error("get dm precheck failed", zap.Error(bizErr))
		return nil, bizErr
	}
	return precheck, nil
}

func (s *dmService) CancelPrecheck(ctx context.Context, clusterID, id string) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("dm_precheck_id", id), zap.String("cluster_id", clusterID))

	precheck, bizErr := s.dmTaskRepo.GetPrecheck(ctx, clusterID, id)
	if bizErr != nil {
		logger.Error("get dm precheck failed", zap.Error(bizErr))
		return bizErr
	}

	if updErr := s.dmTaskRepo.UpdatePrecheck(ctx, precheck.ID, map[string]any{
		"stage":         constant.DMPrecheckStageCanceled,
		"canceled_time": time.Now(),
	}); updErr != nil {
		logger.Error("update dm precheck stage to canceled failed", zap.Error(updErr))
		return updErr
	}
	return nil
}

func (s *dmService) CreateVpcEndpoint(ctx context.Context, clusterID, serviceName string) (string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterID), zap.String("service_name", serviceName))

	// call mgmt svc
	resp, bizErr := s.mgmtSvcFacade.CreateVpcEndpoint(ctx, &globalserverlessv1.CreateVpcEndpointRequest{
		ClusterId:   clusterID,
		ServiceName: serviceName,
	})
	if bizErr != nil {
		if lo.SomeBy(CreateVpcEndpointVisibleErrCodes, func(item string) bool {
			return strings.Contains(bizErr.ErrMsg(), item)
		}) {
			return "", bizerr.New(bizerr.ErrCodeParams, "%s", bizErr.ErrMsg())
		}
		logger.Error("failed to create vpc endpoint via mgmtsvc", zap.Error(bizErr))
		return "", bizErr
	}

	endpoint := &entity.DMVpcEndpoint{
		ID:           entity.GenerateIDWithPrefix(constant.DMVpcEndpointPrefix),
		ClusterID:    clusterID,
		ServiceName:  resp.EndpointDetails.GetServiceName(),
		EndpointId:   resp.EndpointDetails.GetEndpointId(),
		State:        vo.ConvertEndpointState(resp.EndpointDetails.GetState()),
		VpcId:        resp.EndpointDetails.GetVpcId(),
		CreationTime: resp.EndpointDetails.GetCreationTimestamp(),
		Domain:       resp.EndpointDetails.GetDomain(),
		CreatedAt:    time.Now(),
		Creator:      utils.BuildCreatorID(ctx),
		DMTaskID:     pointer.String(""),
	}
	if err := s.dmTaskRepo.CreateVpcEndpoint(ctx, endpoint); err != nil {
		logger.Error("failed to persist vpc endpoint", zap.Error(err))
		return "", err
	}
	return endpoint.ID, nil
}

// DescribeVpcEndpoint returns the VPC endpoint record from repository only.
func (s *dmService) DescribeVpcEndpoint(
	ctx context.Context,
	clusterID, endpointId string,
) (*entity.DMVpcEndpoint, bizerr.BizError) {
	// Only fetch from repo; external sync is handled by background workers.
	return s.dmTaskRepo.GetVpcEndpoint(ctx, clusterID, endpointId)
}

func (s *dmService) registerPrivateLink(
	ctx context.Context,
	source *entity.DMSource,
	logger *zap.Logger,
) (domain, connectionID string, err bizerr.BizError) {
	// old architecture: migrate later, keep for backward compatibility
	// TODO: Remove this block after migration is complete.
	if strings.HasPrefix(source.ConnProfile.Target, constant.DMVpcEndpointPrefix) {
		// For private link, ConnProfile.Target holds the endpoint id. Look up the persisted
		// DMVpcEndpoint (by cluster id + endpoint id) and use its domain as the host.
		endpoint, bizErr := s.dmTaskRepo.GetVpcEndpoint(ctx, source.ClusterID, source.ConnProfile.Target)
		if bizErr != nil {
			logger.Error(
				"failed to get vpc endpoint for private link",
				zap.Error(bizErr),
				zap.String("endpoint_id", source.ConnProfile.Target),
			)
			return "", "", bizErr
		}
		// should not happen
		if endpoint == nil || endpoint.Domain == "" {
			logger.Error("vpc endpoint not found or domain empty", zap.String("endpoint_id", source.ConnProfile.Target))
			return "", "", bizerr.New(
				errorcode.ErrCodeInternalError,
				"vpc endpoint %s not found or domain empty",
				source.ConnProfile.Target,
			)
		}
		bizErr = s.dmTaskRepo.UpdateVpcEndpoint(ctx, endpoint.ID, map[string]any{
			"dm_task_id": source.DMTaskID,
		})
		if bizErr != nil {
			logger.Error("failed to update vpc endpoint with dm task id", zap.Error(bizErr))
			return "", "", bizErr
		}
		return endpoint.Domain, source.ConnProfile.Target, nil
	}

	// new architecture: register private link connection to get domain
	if strings.HasPrefix(source.ConnProfile.Target, constant.PrivateLinkConnectionIDPrefix) {
		domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
			ctx,
			source.ClusterID,
			source.ConnProfile.Target,
			source.ID,
			false,
		)
		if err != nil {
			logger.Error("failed to register private link connection", zap.Error(err))
			return "", "", err
		}
		if len(domains) == 0 || domains[0] == "" {
			logger.Error("no domain found for private link connection", zap.String("plc_id", source.ConnProfile.Target))
			return "", "", bizerr.New(
				errorcode.ErrCodeInternalError,
				"no domain found for private link connection %s",
				source.ConnProfile.Target,
			)
		}
		return domains[0], source.ConnProfile.Target, nil
	}

	logger.Error("invalid private link target", zap.String("target", source.ConnProfile.Target))
	return "", "", bizerr.New(bizerr.ErrCodeParams, "invalid private link connection id: %s", source.ConnProfile.Target)
}

func (s *dmService) publishDMTaskEvent(
	ctx context.Context,
	task *entity.DMTask,
	eventType string,
	summary string,
) {
	if task == nil {
		return
	}
	publishDisplayEventAsync(
		ctx,
		s.eventbusService,
		func(ctx context.Context) *tidbmgmtservicev1.Cluster {
			return loadClusterForEvent(ctx, s.mgmtSvcFacade, task.ClusterID)
		},
		eventType,
		summary,
		event_constant.MigrationDetail{
			ID:   task.ID,
			Name: task.Name,
			Mode: string(task.Mode),
		},
	)
}