package cluster

import (
	"context"
	"fmt"
	"strconv"

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

	"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/domain/service/serverless"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	utilworkflow "github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

// ClaimClusterFromZero migrates data from a zero (shadow-pool) cluster to a new permanent cluster.
// The claim token is validated against the encryptor and the zero cluster's state is transitioned
// from ShadowStateRented → ShadowStateClaiming → ShadowStateClaimed as the workflow DAG progresses:
// CreateDataMigrationTask action transitions Rented → Claiming, CheckDataMigrationTask transitions Claiming → Claimed.
// While DagServerlessClaimClusterFromZero is running the new cluster shows CREATING state so that
// user-facing update/delete operations are rejected.
func (svc *service) ClaimClusterFromZero(
	ctx context.Context,
	cluster *entity.Cluster,
	zeroId string,
) (*entity.Cluster, error) {
	spanName := "ClusterService.ClaimClusterFromZero"
	ctx, span := serverless.Tracer.Start(ctx, spanName)
	defer span.End()

	var err error
	// validate the zero ID and retrieve the corresponding zero cluster record
	originalZeroClusterId, parseError := svc.parseZeroId(ctx, zeroId)
	if parseError != nil {
		tilog.Error(ctx, "failed to parse zero id", zap.String("zero_id", zeroId), zap.Error(err))
		return nil, parseError
	}

	// Validate that the zero cluster exists and is in the expected state
	zeroCluster, err := svc.clusterRepo.GetCluster(ctx, originalZeroClusterId, false)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get zero cluster")
	}
	if zeroCluster.ShadowState != entity.ShadowStateRented {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"zero cluster %d is not claimable",
			originalZeroClusterId,
		)
	}

	// Check third-party account provider restrictions (same as CreateCluster)
	thirdProviders, err := svc.accountCli.GetTenantThirdPartyAccountProviders(ctx, cluster.OrganizationId)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "failed to get third party account providers")
	}
	if err := checkClusterCreateRestrictions(cluster, thirdProviders); err != nil {
		return nil, err
	}
	if err := svc.checkAlibabaCloudMarketplaceServerlessCreate(ctx, cluster, thirdProviders); err != nil {
		return nil, err
	}

	// Validate serverless cluster restrictions
	if err := checkServerlessRestriction(cluster); err != nil {
		return nil, err
	}

	// Validate backup policy
	serverlessCluster, err := cluster.GetServerlessCluster()
	if err != nil {
		return nil, err
	}
	if err := CheckBackupPolicy(serverlessCluster); err != nil {
		return nil, err
	}

	// Execute the cluster creation and DAG startup in a transaction
	if err := svc.transaction.Transaction(ctx, func(txCtx context.Context) error {
		var txErr error
		cluster, txErr = svc.claimSingleCluster(txCtx, cluster, originalZeroClusterId, spanName)
		return txErr
	}); err != nil {
		return cluster, err
	}

	return cluster, nil
}

// claimSingleCluster performs the transactional work of creating a cluster record and
// launching the DagServerlessClaimClusterFromZero workflow.
// It intentionally skips the free-credit-quota check because the claim flow involves
// an explicit spending-limit or auto-scaling configuration provided by the user.
func (svc *service) claimSingleCluster(
	ctx context.Context,
	cluster *entity.Cluster,
	zeroClusterID uint64,
	spanName string,
) (*entity.Cluster, bizerr.BizError) {
	if cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster is nil")
	}
	if cluster.DeployType != entity.DeployTypeServerless {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster deploy type is not serverless")
	}

	starterInventoryLock, shouldCheckStarterInventory, berr := svc.lockStarterInventory(ctx, cluster)
	if berr != nil {
		return nil, berr
	}
	defer svc.releaseStarterInventoryLock(ctx, starterInventoryLock)

	if shouldCheckStarterInventory {
		if berr := svc.checkStarterInventoryLimit(ctx, cluster.OrganizationId); berr != nil {
			return nil, berr
		}
	}

	cluster.ClusterId = entity.GenClusterID()
	createReq, err := buildCreateRequest(cluster)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to build create request")
	}

	// Check display name uniqueness under the project
	count, err := svc.clusterRepo.CountDisplayNameUnderProject(ctx, cluster.ProjectId, cluster.DisplayName)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to count cluster display name")
	}
	if count > 0 {
		return nil, bizerr.New(bizerr.ErrCodeDuplicateDisplayName, bizerr.ErrMsgDupDisplayName)
	}

	// Validate create request against global service
	globalCluster, berr := svc.validateGlobalCreate(ctx, cluster, createReq, false)
	if berr != nil {
		return nil, berr
	}

	// Encrypt root password before persisting
	if createReq.GetRootPassword() != "" {
		encryptedRootPassword, encErr := svc.encryptor.EncryptBase64(ctx, createReq.GetRootPassword())
		if encErr != nil {
			return nil, bizerr.WrapError(encErr, "failed to encrypt root password: %w", encErr)
		}
		createReq.RootPassword = lo.ToPtr(encryptedRootPassword)
	}

	err = svc.transaction.Transaction(ctx, func(txCtx context.Context) error {
		// Transition zero cluster shadow state: ShadowStateRented → ShadowStateClaiming
		if err := svc.clusterRepo.UpdateShadowState(txCtx, zeroClusterID,
			entity.ShadowStateClaiming,
			entity.ShadowStateRented); err != nil {
			return bizerr.WrapGormError(err, "failed to update zero cluster shadow state to claiming: %w", err)
		}

		// Persist new cluster record
		if _, err := svc.clusterRepo.InsertNewCluster(txCtx, cluster); err != nil {
			return bizerr.WrapGormError(err, "failed to insert cluster record: %w", err)
		}

		// Marshal create request for workflow vars
		requestJson, err := protojson.Marshal(createReq)
		if err != nil {
			return bizerr.WrapError(err, "failed to marshal create cluster request: %w", err)
		}

		metadataJson, err := metadata.MarshalRequestMetaFromCtx(txCtx)
		if err != nil {
			return bizerr.WrapErrorWithCode(
				err,
				bizerr.ErrCodeJsonMarshal,
				"failed to marshal request metadata: %w",
				err,
			)
		}

		// Launch the claim DAG (includes create + data migration tasks)
		if err := svc.workflow.RunDagWithBizId(
			txCtx,
			constant.DagServerlessClaimClusterFromZero,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			fmt.Sprintf("%d", cluster.ClusterId),
			map[string]string{
				constant.KeyRequest:         string(requestJson),
				constant.KeyZeroClusterID:   strconv.FormatUint(zeroClusterID, 10),
				utilworkflow.KeyTraceId:     utilworkflow.GetTraceId(ctx),
				utilworkflow.KeySpanName:    spanName,
				utilworkflow.KeyRequestMeta: string(metadataJson),
			},
		); err != nil {
			return bizerr.WrapError(err, "failed to run claim serverless cluster from zero dag task: %w", err)
		}

		tilog.Info(txCtx, "Started claim cluster from zero DAG",
			zap.Uint64("new_cluster_id", cluster.ClusterId),
			zap.Uint64("zero_cluster_id", zeroClusterID),
		)

		if err := svc.accountCli.UpsertClusterRelations(txCtx, cluster.ClusterId, cluster.OrganizationId, cluster.ProjectId); err != nil {
			return bizerr.WrapGormError(err, "failed to add cluster relations to account-svc: %w", err)
		}

		return nil
	})

	if err != nil {
		tilog.Error(ctx, "transaction failed for claiming cluster from zero",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.Uint64("zero_cluster_id", zeroClusterID),
			zap.Error(err))
		return nil, bizerr.WrapGrpcError(err, "failed to claim cluster")
	}

	// Return the freshly inserted cluster record (enriched with global-service placeholder)
	cluster, err = svc.clusterRepo.GetCluster(ctx, cluster.ClusterId, false)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "failed to get newly created cluster record: %w", err)
	}

	cluster = cvtServerlessCluster(cluster, globalCluster)

	return cluster, nil
}