package v1beta1

import (
	"context"
	"strings"

	"github.com/AlekSi/pointer"
	mapset "github.com/deckarep/golang-set/v2"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	v1beta1pb "github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"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"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

var tracer = otel.Tracer("dm.v1beta1")

var (
	listSupportedOrderByField = []string{"create_time"}
	listOrderByFieldMap       = map[string]string{
		"create_time": "created_at",
	}
)

type DMService struct {
	mgmtSvc         facade.MgmtSvcFacade
	dmService       service.DMService
	encryptorFacade facade.EncryptorFacade
	privateConnSvc  service.PrivateLinkConnectionService
	v1beta1pb.UnimplementedMigrationServiceServer
}

func NewDMService(
	mgmtSvc facade.MgmtSvcFacade,
	dmService service.DMService,
	encryptorFacade facade.EncryptorFacade,
	privateConnSvc service.PrivateLinkConnectionService,
) v1beta1pb.MigrationServiceServer {
	return &DMService{
		mgmtSvc:         mgmtSvc,
		dmService:       dmService,
		encryptorFacade: encryptorFacade,
		privateConnSvc:  privateConnSvc,
	}
}

func (s *DMService) CreateMigration(
	ctx context.Context,
	req *v1beta1pb.CreateMigrationReq,
) (_ *v1beta1pb.Migration, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.CreateMigration")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.GetClusterId()),
		zap.String("name", strings.TrimSpace(req.GetDisplayName())),
	)

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

	if err := s.encryptPasswords(logger, req); err != nil {
		return nil, err
	}

	task, bizErr := vo.CreateV1beta1TaskReqToDMTask(ctx, req, cluster)
	if bizErr != nil {
		return nil, bizErr
	}

	dmTask, bizErr := s.dmService.CreateTask(ctx, task)
	if bizErr != nil {
		return nil, bizErr
	}
	return vo.DMTaskAndStatusToV1beta1PbTask(ctx, dmTask)
}

func (s *DMService) GetMigration(
	ctx context.Context,
	req *v1beta1pb.GetMigrationReq,
) (_ *v1beta1pb.Migration, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.GetMigration")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("migration_id", req.GetMigrationId()),
	)
	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.GetMigrationId())
	if bizErr != nil {
		return nil, bizErr
	}
	return vo.DMTaskAndStatusToV1beta1PbTask(ctx, task)
}

func (s *DMService) ListMigrations(
	ctx context.Context,
	req *v1beta1pb.ListMigrationsReq,
) (_ *v1beta1pb.ListMigrationsResp, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.ListMigrations")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()

	filter := &entity.DMTaskFilter{ClusterID: req.ClusterId}
	pager := utils.NewAipPager()
	err = pager.Normalize(req)
	if err != nil {
		return nil, err
	}
	filter.Offset = pointer.To(int(pager.Offset()))
	filter.Limit = pointer.To(int(pager.PageSize()))

	if err = utils.CheckOrderBy(req, listSupportedOrderByField); 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

	total, tasks, bizErr := s.dmService.ListTasks(ctx, filter)
	if bizErr != nil {
		return nil, bizErr
	}
	nextPageToken := pager.NextPageToken(uint32(total))
	resp := &v1beta1pb.ListMigrationsResp{
		Migrations:    make([]*v1beta1pb.Migration, 0, len(tasks)),
		TotalSize:     uint32(total),
		NextPageToken: nextPageToken,
	}
	for _, t := range tasks {
		task, convErr := vo.DMTaskAndStatusToV1beta1PbTask(ctx, t)
		if convErr != nil {
			return nil, convErr
		}
		resp.Migrations = append(resp.Migrations, task)
	}
	return resp, nil
}

func (s *DMService) DeleteMigration(
	ctx context.Context,
	req *v1beta1pb.DeleteMigrationReq,
) (_ *v1beta1pb.Migration, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.DeleteMigration")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("migration_id", req.GetMigrationId()),
	)
	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 := s.dmService.DeleteTask(ctx, req.ClusterId, req.GetMigrationId()); bizErr != nil {
		return nil, bizErr
	}
	task, bizErr := s.dmService.GetTask(ctx, cluster, req.GetMigrationId())
	if bizErr != nil {
		return nil, bizErr
	}
	return vo.DMTaskAndStatusToV1beta1PbTask(ctx, task)
}

func (s *DMService) PauseMigration(
	ctx context.Context,
	req *v1beta1pb.PauseMigrationReq,
) (_ *emptypb.Empty, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.PauseMigration")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("migration_id", req.GetMigrationId()),
	)
	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 := s.dmService.PauseTask(ctx, cluster, req.GetMigrationId()); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (s *DMService) ResumeMigration(
	ctx context.Context,
	req *v1beta1pb.ResumeMigrationReq,
) (_ *emptypb.Empty, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.ResumeMigration")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("migration_id", req.GetMigrationId()),
	)
	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 := s.dmService.ResumeTask(ctx, cluster, req.GetMigrationId()); bizErr != nil {
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

// Precheck creates a migration precheck task based on the request and returns the precheck id
func (s *DMService) Precheck(
	ctx context.Context,
	req *v1beta1pb.CreateMigrationReq,
) (_ *v1beta1pb.CreateMigrationPrecheckResp, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.PrecheckTask")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareDMErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(
		zap.String("cluster_id", req.GetClusterId()),
		zap.String("name", strings.TrimSpace(req.GetDisplayName())),
	)

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

	if err := s.encryptPasswords(logger, req); err != nil {
		return nil, err
	}

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

	return &v1beta1pb.CreateMigrationPrecheckResp{PrecheckId: precheck.ID}, nil
}

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
		}
		return "", bizerr.New(bizerr.ErrCodeParams, "invalid private link connection id")
	}
}

// CancelPrecheck cancels a running precheck task
func (s *DMService) CancelPrecheck(
	ctx context.Context,
	req *v1beta1pb.CancelMigrationPrecheckReq,
) (_ *emptypb.Empty, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.CancelPrecheck")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()

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

	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 := s.dmService.CancelPrecheck(ctx, req.ClusterId, req.GetPrecheckId()); bizErr != nil {
		logger.Error("failed to cancel precheck", zap.Error(bizErr))
		return nil, bizErr
	}
	logger.Info("precheck canceled successfully")
	return &emptypb.Empty{}, nil
}

// GetPrecheck returns the precheck result
func (s *DMService) GetPrecheck(
	ctx context.Context,
	req *v1beta1pb.GetMigrationPrecheckReq,
) (_ *v1beta1pb.MigrationPrecheck, err error) {
	ctx, span := tracer.Start(ctx, "v1beta1.dm.GetPrecheck")
	defer func() { tracing.EndSpan(span, err); err = errorcode.PrepareDMErrorMessage(ctx, err) }()

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

// encryptPasswords encrypts target and sources passwords in-place using encryptor facade.
func (s *DMService) encryptPasswords(logger *zap.Logger, req *v1beta1pb.CreateMigrationReq) 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 enc string
		enc, 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 = enc
	}
	return nil
}

// isWhitelisted checks whether migration is allowed for the given tenant and region.
func isWhitelisted(tenantID uint64, provider, region string) bizerr.BizError {
	tenantWhiteSet := mapset.NewSet(config.Global().BizConfig.MigrationWhitelistTenants...)
	regionWhiteSet := mapset.NewSet(config.Global().BizConfig.MigrationWhitelistRegions...)
	regionKey := provider + "-" + region
	if tenantWhiteSet.IsEmpty() && regionWhiteSet.IsEmpty() {
		return nil
	}
	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",
	)
}