package cluster

import (
	"context"
	"errors"
	"fmt"
	"math"
	"strconv"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"gorm.io/gorm"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/apimiddleware/bizerror"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

// Reserve a high status_id range for dedicated restore idempotency.
// This makes concurrent requests for the same restore task collide on the
// existing unique status_id index instead of creating duplicate clusters.
const restoreStatusIDOffset uint64 = 9_000_000_000_000_000_000

func buildRestoreStatusID(restoreID uint64) (uint64, bizerr.BizError) {
	if restoreID == 0 || restoreID > math.MaxUint64-restoreStatusIDOffset {
		return 0, bizerr.New(bizerr.ErrCodeParams, "restore id is invalid")
	}
	return restoreStatusIDOffset + restoreID, nil
}

func (svc *service) RestoreCluster(
	ctx context.Context,
	restoreCluster *entity.Cluster,
	sourceClusterId, restoreId uint64,
	validateOnly bool,
	clusterConfiguration *globaldedicatedv1.ClusterConfiguration,
) (*entity.Cluster, error) {
	spanName := "ClusterService.RestoreCluster"
	ctx, span := tracer.Start(ctx, spanName)
	defer span.End()

	if restoreCluster == nil {
		return nil, fmt.Errorf("restore cluster is nil")
	}

	if sourceClusterId == 0 {
		return nil, fmt.Errorf("source cluster is nil")
	}

	if restoreCluster.DeployType != entity.DeployTypeDedicated {
		return nil, fmt.Errorf("cluster deploy type is not dedicated")
	}

	restoreCluster.ServicePlan = openapicommonv1beta1.ServicePlan_Dedicated

	// check third party account provider
	thirdProviders, err := svc.accountCli.GetTenantThirdPartyAccountProviders(ctx, restoreCluster.OrganizationId)
	if err != nil {
		return nil, err
	}
	if len(thirdProviders) > 0 && !lo.Contains(thirdProviders, restoreCluster.CloudProvider) {
		return nil, bizerr.New(bizerr.ErrCodePermission, bizerr.ErrMsg3rdPartyAccount, restoreCluster.CloudProvider)
	}

	// if validate only
	// 1. check cluster name is unique under the project
	// 2. check cluster create request
	if validateOnly {
		if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
			// Check cluster name is unique under the project
			count, err := svc.clusterRepo.CountDisplayNameUnderProject(ctx, restoreCluster.ProjectId, restoreCluster.DisplayName)
			if err != nil {
				return err
			}
			if count > 0 {
				return constant.ErrDuplicatedClusterName
			}

			// only validate
			restoreCluster.ClusterId = 0 // fake cluster id
			request, err := buildCreateRequest(restoreCluster, clusterConfiguration)
			if err != nil {
				return err
			}

			// set validate only
			request.ValidateOnly = true
			dedicatedCluster, err := svc.dedicatedCli.CreateCluster(ctx, request)
			if err != nil {
				return err
			}
			restoreCluster = cvtDedicatedCluster(restoreCluster, dedicatedCluster)

			return nil
		}); err != nil {
			return restoreCluster, err
		}

		return restoreCluster, nil
	}

	// transaction
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// check is cluster in restoring
		restoreStatus, err := svc.customStatusRepo.GetRestoreStatusByName(ctx, fmt.Sprintf("restore-%d", restoreId))
		if err != nil {
			if !errors.Is(err, gorm.ErrRecordNotFound) {
				return bizerror.NewInternal(err, "failed to get restore status")
			}
		}

		// cluster in restoring
		if restoreStatus != nil {
			tilog.Warn(ctx, "cluster is restoring", zap.Uint64("cluster_id", restoreStatus.ClusterId))
			restoreCluster, err = svc.GetCluster(ctx, restoreStatus.ClusterId, entity.ClusterViewClusterViewBasic, true, false)
			if err != nil {
				if errors.Is(err, gorm.ErrRecordNotFound) {
					return bizerror.NewNotFound(err, "restore cluster not found")
				}
				return bizerror.NewInternal(err, "failed to get restore cluster")
			}
			return nil
		}

		// Check cluster name is unique under the project
		count, err := svc.clusterRepo.CountDisplayNameUnderProject(ctx, restoreCluster.ProjectId, restoreCluster.DisplayName)
		if err != nil {
			return err
		}
		if count > 0 {
			return constant.ErrDuplicatedClusterName
		}

		// TODO: check cmek by @shirui
		restoreCluster.ClusterId = entity.GenClusterID()

		// Then set restore custom status
		statusId, err := buildRestoreStatusID(restoreId)
		if err != nil {
			return err
		}
		status := &entity.ClusterCustomStatus{
			StatusId:      statusId,
			StatusName:    fmt.Sprintf("restore-%d", restoreId),
			ClusterId:     restoreCluster.ClusterId,
			StatusType:    tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE,
			OverrideState: openapicommonv1beta1.ClusterState_RESTORING.Enum(),
			CustomMutex: &tidbmgmtservicev1.ClusterCustomStatus_CustomMutex{
				StatusType: []tidbmgmtservicev1.CustomStatusType{
					tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT,
					tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE,
					tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_APPLY,
					tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
				},
				ClusterAction: []tidbmgmtservicev1.ClusterAction{
					tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_MODIFY_CLUSTER,
					tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
					tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_SCALE_CLUSTER,
					tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_SET_ROOT_PASSWORD,
				},
			},
		}
		// First set restore custom status
		if _, err = svc.customStatusRepo.NewClusterCustomStatus(ctx, status); err != nil {
			return bizerr.WrapGormError(err, "failed to set cluster custom status for restore task %d", restoreId)
		}

		// Then persist cluster info to database
		if _, err = svc.clusterRepo.InsertNewCluster(ctx, restoreCluster); err != nil {
			return fmt.Errorf("failed to insert cluster record: %w", err)
		}

		if err = svc.accountCli.UpsertClusterRelations(ctx,
			restoreCluster.ClusterId,
			restoreCluster.OrganizationId,
			restoreCluster.ProjectId); err != nil {
			return fmt.Errorf("failed to init cluster permissions: %w", err)
		}

		// Then create cluster in global dedicated service
		request, err := buildCreateRequest(restoreCluster, clusterConfiguration)
		if err != nil {
			return err
		}

		// only validate
		request.ValidateOnly = true
		dedicatedCluster, err := svc.dedicatedCli.CreateCluster(ctx, request)
		if err != nil {
			return err
		}

		if dedicatedCluster == nil {
			return errors.New("cluster in global create response should not be nil")
		}

		// run dag instance
		// encrypt root password
		if request.Cluster.RootPassword != "" {
			encryptedRootPassword, err := svc.encryptor.EncryptBase64(ctx, request.Cluster.RootPassword)
			if err != nil {
				return fmt.Errorf("failed to encrypt root password: %w", err)
			}
			request.Cluster.RootPassword = encryptedRootPassword
		}

		request.ValidateOnly = false
		requestJson, err := protojson.Marshal(request)
		if err != nil {
			return fmt.Errorf("failed to marshal create cluster request: %w", err)
		}
		// metadata
		metadataJson, err := metadata.MarshalRequestMetaFromCtx(ctx)
		if err != nil {
			return fmt.Errorf("failed to marshal request metadata: %w", err)
		}

		err = svc.workflow.RunDagWithBizId(
			ctx,
			constant.DagDedicatedRestoreCluster,
			strconv.FormatUint(restoreCluster.OrganizationId, 10),
			strconv.FormatUint(restoreCluster.ProjectId, 10),
			fmt.Sprintf("%d", restoreCluster.ClusterId),
			map[string]string{
				constant.KeyRestoreFromCluster: fmt.Sprintf("%d", sourceClusterId),
				constant.KeyRequest:            string(requestJson),
				workflow.KeyTraceId:            workflow.GetTraceId(ctx),
				workflow.KeySpanName:           spanName,
				workflow.KeyRequestMeta:        string(metadataJson),
			},
		)
		if err != nil {
			return fmt.Errorf("failed to run create cluster dag task: %w", err)
		}

		// Finally, return cluster info
		restoreCluster, err = svc.clusterRepo.GetCluster(ctx, restoreCluster.ClusterId, false)
		if err != nil {
			return fmt.Errorf("failed to get cluster record: %w", err)
		}

		restoreCluster = cvtDedicatedCluster(restoreCluster, dedicatedCluster)

		return nil
	}); err != nil {
		return restoreCluster, err
	}

	return restoreCluster, nil
}