package cluster

import (
	"context"
	"encoding/json"
	"fmt"
	"regexp"
	"strconv"

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

	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/billing"
	"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/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/serverless"
	svcworkflow "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/service/workflow"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"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"
)

func (svc *service) CreateCluster(ctx context.Context, cluster *entity.Cluster) (*entity.Cluster, error) {
	spanName := "ClusterService.CreteCluster"
	ctx, span := serverless.Tracer.Start(ctx, spanName)
	defer span.End()

	// check third party account provider
	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
	}

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

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

	// transaction
	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		var err error
		cluster, err = svc.createSingleCluster(ctx, cluster, spanName, false)
		if err != nil {
			return err
		}
		return nil
	}); err != nil {
		return cluster, err
	}

	return cluster, nil
}

func checkServerlessRestriction(cluster *entity.Cluster) error {
	serverlessCluster, err := cluster.GetServerlessCluster()
	if err != nil {
		return err
	}

	if serverlessCluster.GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
		serverlessCluster.GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		/* essential plan */
		if serverlessCluster.GetCluster().
			GetAutoScaling().
			GetMinRcu() >
			serverlessCluster.GetCluster().
				GetAutoScaling().
				GetMaxRcu() ||
			serverlessCluster.GetCluster().GetAutoScaling().GetMinRcu() < common.ServerlessAutoScalingMinRcu ||
			serverlessCluster.GetCluster().GetAutoScaling().GetMaxRcu() > common.ServerlessAutoScalingMaxRcu {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"valid rcu range %v~%v",
				common.ServerlessAutoScalingMinRcu,
				common.ServerlessAutoScalingMaxRcu,
			)
		}
	} else if serverlessCluster.GetSpendLimit().GetMonthly() <= 0 {
		/* starter plan without spendinglimit setting */
		if serverlessCluster.GetCluster().GetEnhancedEncryptionEnabled() {
			return bizerr.New(bizerr.ErrCodeParams, "Enabling enhanced encryption requires setting a monthly spending limit.")
		}
		if serverlessCluster.GetCluster().GetZoneRedundantEnabled() {
			return bizerr.New(bizerr.ErrCodeParams, "Enabling regional availability requires setting a monthly spending limit.")
		}
	}
	return nil
}

// CheckBackupPolicy needs full view of serverlessCluster, the check is the same as serverless-svc
func CheckBackupPolicy(serverlessCluster *globalserverlessv1.ServerlessServiceCluster) error {
	if serverlessCluster.GetBackupPolicy() == nil {
		return nil
	}

	if serverlessCluster.GetCluster().GetAutoScaling().GetMinRcu() <= 0 &&
		serverlessCluster.GetCluster().GetAutoScaling().GetMaxRcu() <= 0 &&
		serverlessCluster.GetSpendLimit().GetMonthly() <= 0 {
		return bizerr.New(bizerr.ErrCodeParams,
			"not support update backup policy for starter plan cluster without spending limit")
	}

	if serverlessCluster.GetBackupPolicy().GetBackupTime() == "" {
		return bizerr.New(bizerr.ErrCodeParams, "backup time is required when backup policy is set")
	}
	matched, err := regexp.MatchString(
		`^([0-1][0-9]|2[0-3]):[03]0$`,
		serverlessCluster.GetBackupPolicy().GetBackupTime(),
	)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "failed to parse backup time")
	}
	if !matched {
		return bizerr.New(bizerr.ErrCodeParams, "invalid start time")
	}

	// Retention days is optional for compatibility.
	if serverlessCluster.GetBackupPolicy().GetRetentionDays() > 30 {
		return bizerr.New(bizerr.ErrCodeParams, "invalid retention days, maximum is 30")
	}
	return nil
}

func (svc *service) createSingleCluster(
	ctx context.Context,
	cluster *entity.Cluster,
	spanName string,
	skipValidate bool,
) (*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()

	// Resolve ru_version from organization config
	orgConfig, err := svc.orgRuConfigRepo.GetByOrganizationId(ctx, cluster.OrganizationId)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to get organization ru config")
	}
	cluster.RuVersion = orgConfig.DefaultRuVersion

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

	if createReq.GetSpendLimit().GetMonthly() == 0 && createReq.GetAutoScaling().GetMaxRcu() == 0 {
		// If need lock, we need lock at the beginning of the transaction, otherwise the REPEATABLE READ isolation level
		// will make query clusters in hasFreeCreditQuota() can't see concurrently created clusters.
		createClusterLock := &entity.Lock{
			Key:         fmt.Sprintf("create_serverless_cluster_%d", cluster.OrganizationId),
			TimeoutSecs: 29, // db timeout is 30s
		}

		if err := svc.lockRepo.Lock(ctx, createClusterLock); err != nil {
			tilog.Error(ctx, "create cluster lock failed", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "create cluster failed, please try again")
		}

		defer func() {
			// use new ctx to prevent user ctx timeout
			rctx := context.WithoutCancel(ctx)
			if err := svc.lockRepo.Release(rctx, createClusterLock); err != nil {
				tilog.Error(
					rctx,
					"release create cluster lock failed",
					zap.Any("lock", createClusterLock),
					zap.Error(err),
				)
			} else {
				tilog.Debug(rctx, "release cluster lock success", zap.Uint64("cluster id", cluster.ClusterId), zap.Any("lock", createClusterLock))
			}
		}()

		hasFreeQuota, berr := svc.hasFreeCreditQuota(ctx, cluster.OrganizationId)
		if berr != nil {
			return nil, berr
		}
		if !hasFreeQuota {
			return nil, bizerr.New(
				bizerr.ErrCodeServerlessReachMaxFreeNum,
				"Serverless clusters have reached the maximum number of free clusters. Please set a monthly Spending Limit to continue.",
			)
		}
	}

	// Check cluster name is unique 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, skipValidate)
	if berr != nil {
		return nil, berr
	}

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

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

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

	// run dag instance
	requestJson, err := protojson.Marshal(createReq)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to marshal create cluster request: %w", err)
	}
	// metadata
	metadataJson, err := metadata.MarshalRequestMetaFromCtx(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(
			err,
			bizerr.ErrCodeJsonMarshal,
			"failed to marshal request metadata: %w",
			err,
		)
	}
	if err := svc.workflow.RunDagWithBizId(
		ctx,
		constant.DagServerlessCreateCluster,
		strconv.FormatUint(cluster.OrganizationId, 10),
		strconv.FormatUint(cluster.ProjectId, 10),
		fmt.Sprintf("%d", cluster.ClusterId),
		map[string]string{
			constant.KeyRequest:         string(requestJson),
			utilworkflow.KeyTraceId:     utilworkflow.GetTraceId(ctx),
			utilworkflow.KeySpanName:    spanName,
			utilworkflow.KeyRequestMeta: string(metadataJson),
		},
	); err != nil {
		return nil, bizerr.WrapError(err, "failed to run create serverless cluster dag task: %w", err)
	}

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

	cluster = cvtServerlessCluster(cluster, globalCluster)

	return cluster, nil
}

func (svc *service) validateGlobalCreate(
	ctx context.Context,
	cluster *entity.Cluster,
	createReq *globalserverlessv1.ServerlessServiceCreateRequest,
	skipValidate bool,
) (*globalserverlessv1.ServerlessServiceCluster, bizerr.BizError) {
	if skipValidate {
		sc := defaultServerlessCluster(ctx, cluster)
		return sc, nil
	}

	req, ok := proto.Clone(createReq).(*globalserverlessv1.ServerlessServiceCreateRequest)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "failed to clone create request")
	}
	req.ValidateOnly = true

	resp, err := svc.serverlessCli.CreateCluster(ctx, req)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "failed to create serverless cluster")
	}
	if resp.Cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster in global create response should not be nil")
	}

	return resp.GetCluster(), nil
}

func buildCreateRequest(cluster *entity.Cluster) (*globalserverlessv1.ServerlessServiceCreateRequest, error) {
	serverlessCluster, err := cluster.GetServerlessCluster()
	if err != nil {
		return nil, err
	}

	// OriginalProjectId is only written to the DB via the BeforeCreate hook,
	// which runs after this function is called. Fall back to ProjectId so the
	// request (and the serialized DAG payload) always carries the correct value.
	originalProjectId := cluster.OriginalProjectId
	if originalProjectId == 0 {
		originalProjectId = cluster.ProjectId
	}

	createReq := &globalserverlessv1.ServerlessServiceCreateRequest{
		TenantId:             serverlessCluster.TenantId,
		ProjectId:            serverlessCluster.ProjectId,
		OriginalProjectId:    originalProjectId,
		Id:                   strconv.FormatUint(cluster.ClusterId, 10),
		CloudProvider:        serverlessCluster.CloudProvider,
		Region:               serverlessCluster.Region,
		ZoneRedundantEnabled: serverlessCluster.GetCluster().GetZoneRedundantEnabled(),
		AuditLogConfig:       serverlessCluster.GetCluster().GetAuditLogConfig(),
	}
	engineType := cluster.EngineType
	if engineType == openapicommonv1beta1.EngineType_ENGINE_TYPE_UNSPECIFIED {
		engineType = openapicommonv1beta1.EngineType_MYSQL_COMPATIBLE
	}
	createReq.EngineType = engineType.String()

	if serverlessCluster.SpendLimit != nil {
		createReq.SpendLimit = serverlessCluster.SpendLimit
	}

	if serverlessCluster.BackupPolicy != nil {
		createReq.BackupPolicy = serverlessCluster.BackupPolicy
	}

	if serverlessCluster.Cluster.Endpoints != nil {
		createReq.Endpoints = serverlessCluster.Cluster.Endpoints
	}

	createReq.EncryptionConfig = &globalserverlessv1.EncryptionConfig{
		EnhancedEncryptionEnabled: serverlessCluster.Cluster.EnhancedEncryptionEnabled,
	}

	createReq.RestoreCluster = lo.ToPtr(serverlessCluster.Cluster.RestoreCluster)
	createReq.RootPassword = lo.ToPtr(serverlessCluster.RootPassword)
	createReq.Kubernetes = serverlessCluster.Cluster.Kubernetes

	if serverlessCluster.GetCluster().GetAutoScaling() != nil {
		createReq.AutoScaling = serverlessCluster.GetCluster().GetAutoScaling()
	}

	if len(cluster.InitSQLs) > 0 {
		createReq.InitSqls = cluster.InitSQLs
	}

	createReq.RuVersion = cluster.RuVersion

	return createReq, nil
}

func (svc *service) hasFreeCreditQuota(ctx context.Context, tenantId uint64) (bool, bizerr.BizError) {
	count, err := svc.clusterRepo.CountClusterByFilter(
		ctx,
		&repository.ClusterFilter{
			OrganizationIds: []uint64{tenantId},
			QueryDeleted:    false,
			DeployType:      entity.DeployTypeServerless,
		},
	)
	if err != nil {
		return false, bizerr.WrapGormError(err, "failed to query serverless clusters for tenant")
	}

	if int(count) < svc.config.ServerlessFreeQuotaMaxCount {
		return true, nil
	}
	if int(count) > svc.config.ServerlessFreeQuotaMaxCount*100 {
		tilog.Warn(
			ctx,
			"cluster count is more than 100 times the ServerlessFreeQuotaMaxCount, skip the check.",
			zap.Uint64("tenandId", tenantId),
			zap.Int64("clusterCount", count),
			zap.Int("maxFreeClusterCount", svc.config.ServerlessFreeQuotaMaxCount),
		)
		return false, nil
	}
	clusters, err := svc.clusterRepo.QueryClusterByFilterAndPage(
		ctx,
		&repository.ClusterFilter{
			OrganizationIds: []uint64{tenantId},
			QueryDeleted:    false,
			DeployType:      entity.DeployTypeServerless,
		},
		0,
		0,
	)
	if err != nil {
		return false, bizerr.WrapGormError(err, "failed to query serverless clusters for tenant")
	}

	// 1. Get creating free clusters in workflow
	// 2. Get clusters set free spending limit from billing service
	// 3. Union the two parts to get the accurate free clusters count
	// We don't need to worry about free spend limit has already been set or not in the workflow,
	// since we union the workflow and billing results, no matter it's set, not set, or even the
	// workflow finished parallelly.
	// Don't change the order of 1 and 2, otherwise the count will be incorrect.
	// If we do step 2 first and then step 1, there might be clusters finishing the creating
	// process between step 2 and step 1 period. They finish the workflow so we cannot query
	// them from the running dags, and we also passed the querying spending limit stage.
	freeClusters := map[string]bool{}

	instances, err := svc.workflow.ListRunningDagInstancesByOrgId(
		ctx,
		[]string{constant.DagServerlessCreateCluster},
		strconv.FormatUint(tenantId, 10),
	)
	if err != nil {
		return false, bizerr.WrapGormError(err, "failed to list running dag instances for tenant")
	}
	tilog.Debug(ctx, "list running dag instances for tenant", zap.Any("instances", instances))

	for _, ins := range instances {
		out := svcworkflow.Unmarshal(ctx, 0, ins, &globalserverlessv1.ServerlessServiceCreateRequest{})
		if out != nil {
			createReq, ok := out.(*globalserverlessv1.ServerlessServiceCreateRequest)
			if !ok {
				tilog.Error(ctx, "failed to convert to ServerlessServiceCreateRequest", zap.Any("out", out))
				continue
			}
			if createReq.GetSpendLimit().GetMonthly() == 0 && createReq.GetAutoScaling().GetMaxRcu() == 0 {
				freeClusters[createReq.Id] = true
			}
		}
	}

	clusterIdsMap := lo.KeyBy(clusters, func(c *entity.Cluster) uint64 { return c.ClusterId })

	clusterSpendLimitList, berr := svc.billingCli.GetSpendLimitSettings(ctx, tenantId)
	if berr != nil {
		return false, berr
	}

	for _, clusterSpendLimit := range clusterSpendLimitList {
		// check if the cluster is in the mgmt service, billing service might have remaining deleted clusters
		if _, ok := clusterIdsMap[clusterSpendLimit.RangeID]; !ok {
			tilog.Warn(ctx, "the cluster is not in mgmt service", zap.Uint64("cluster", clusterSpendLimit.RangeID))
			continue
		}

		var bizData billing.ServerlessSpendLimitBizData
		if err := json.Unmarshal([]byte(clusterSpendLimit.BizData), &bizData); err != nil {
			tilog.Error(
				ctx,
				"unmarshal biz data failed",
				zap.String("bizData", clusterSpendLimit.BizData),
				zap.Error(err),
			)
			continue
		}

		if bizData.FreeCredits.GreaterThan(decimal.Zero) {
			freeClusters[strconv.FormatUint(clusterSpendLimit.RangeID, 10)] = true
		}
	}

	if len(freeClusters) >= svc.config.ServerlessFreeQuotaMaxCount {
		tilog.Warn(
			ctx,
			"the free cluster count is greater than the max count",
			zap.Int("freeClusterCount", len(freeClusters)),
			zap.Int("maxCount", svc.config.ServerlessFreeQuotaMaxCount),
		)
		return false, nil
	}

	return true, nil
}