package cluster

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

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

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (svc *clusterServiceImpl) Create(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceCreateRequest,
) (*globalserverlessv1.ServerlessServiceCluster, error) {
	ctx, span := Tracer.Start(ctx, "Domain.Create")
	defer span.End()

	if err := svc.regionLimitations(req); err != nil {
		return nil, err
	}

	cloudProviders, err := svc.cloudProviderRepo.LoadAll(ctx)
	if err != nil {
		return nil, err
	}

	cluster, err := svc.buildClusterEntity(ctx, req, cloudProviders)
	if err != nil {
		return nil, err
	}

	if req.EncryptionConfig == nil {
		req.EncryptionConfig = &globalserverlessv1.EncryptionConfig{EnhancedEncryptionEnabled: false}
	}

	if err := svc.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Idempotency check: if cluster ID is provided and cluster already exists, skip creation
		if req.GetId() != "" {
			existingCluster, err := svc.clusterRepo.Get(ctx, req.GetId(), false)
			if err == nil && existingCluster != nil {
				// Cluster already exists, this is a duplicate request
				tilog.Info(ctx, "cluster already exists, skipping creation", zap.String("cluster_id", req.GetId()))
				cluster = existingCluster
				return nil
			}
			// If error is not "record not found", return the error
			var bizErr bizerr.BizError
			if errors.As(err, &bizErr) && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				return bizErr
			}
			// Otherwise, cluster doesn't exist, continue with creation
		}

		// freecluster quota check
		if req.GetSpendLimit().GetMonthly() == 0 && req.GetAutoScaling().GetMaxRcu() == 0 {
			if req.GetEncryptionConfig().GetEnhancedEncryptionEnabled() {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"the monthly spending limit must be greater than 0, if the cluster has advanced encryption enabled",
				)
			}
		} else {
			// check can create scalable cluster
			resourceType := adapter.BillingResourceTypeServerlessCluster
			if req.GetAutoScaling().GetMinRcu() > 0 || req.GetAutoScaling().GetMaxRcu() > 0 {
				resourceType = adapter.BillingResourceTypeEssentialCluster
			}
			canCreate, err := svc.billingFacade.CanCreateScalableCluster(
				ctx,
				cluster.Region.Id.String(),
				cluster.Region.CloudProviderId.String(),
				resourceType,
				cluster.TenantId,
			)
			if err != nil {
				return err
			}
			if !canCreate.Pass {
				return bizerr.New(bizerr.ErrCodeNoPaymentMethods, "%s", canCreate.Reason)
			}
		}

		// check audit log
		if req.GetAuditLogConfig() != nil {
			return bizerr.New(bizerr.ErrCodeParams, constant.AuditLogDeprecatedMsg)
		}

		if req.ValidateOnly {
			return nil
		}

		if err := svc.clusterRepo.Save(ctx, cluster); err != nil {
			return fmt.Errorf("create cluster failed: %w", err)
		}

		events, err := cluster.CreateEvent(ctx)
		if err != nil {
			return err
		}

		if err := svc.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, events); err != nil {
			return err
		}
		if req.GetAutoScaling().GetMaxRcu() == 0 {
			/*
				max_rcu = 0 means starter plan, max_rcu > 0 means essential plan，essential plan should not call SetSpendLimit
				set 0 to spend limit, because the tenant has free quota, we need set spend limit to 0 now.
			*/
			if req.GetSpendLimit().GetMonthly() == 0 {
				if err := svc.serverlessFacade.SetSpendLimit(
					ctx,
					cluster.ClusterId,
					cluster.TenantId,
					0,
					cluster.Region.Id.String(),
					req.GetEncryptionConfig().GetEnhancedEncryptionEnabled(),
					false,
				); err != nil {
					return fmt.Errorf("set spend limit failed: %w", err)
				}
			}
		}

		requestStr, err := protojson.Marshal(req)
		if err != nil {
			return fmt.Errorf("marshal request failed: %w", err)
		}

		initSQLsJSON, err := json.Marshal(req.GetInitSqls())
		if err != nil {
			return fmt.Errorf("marshal init sqls failed: %w", err)
		}

		if err := svc.flow.RunDagWithBizId(
			ctx,
			common.CreateCluster,
			fmt.Sprintf("%d", cluster.TenantId),
			fmt.Sprintf("%d", cluster.ProjectId),
			cluster.ClusterId,
			map[string]string{
				workflow.KeyTraceId:   workflow.GetTraceId(ctx),
				common.KeyClusterId:   cluster.ClusterId,
				common.KeyRequestBody: string(requestStr),
				common.KeyInitSQLs:    string(initSQLsJSON),
			},
		); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return nil, err
	}

	resp := &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
		ClusterId: cluster.ClusterId,
		CreatedTs: time.Now().Unix(),
		TenantId:  cluster.TenantId,
		ProjectId: cluster.ProjectId,
		Region: &regionalserverlessv1.Region{
			Name:          cluster.Region.Id.String(),
			DisplayName:   cluster.Region.DisplayName,
			CloudProvider: cvt.Name2CloudProviderEnum(cluster.Region.CloudProviderId.String()),
		},
		Endpoints: &regionalserverlessv1.Endpoints{
			Public: &regionalserverlessv1.PublicEndpoint{
				Disabled:          req.GetEndpoints().GetPublic().GetDisabled(),
				AuthorizedNetwork: req.GetEndpoints().GetPublic().GetAuthorizedNetwork(),
			},
			Private: &regionalserverlessv1.PrivateEndpoint{
				Disabled:          req.GetEndpoints().GetPrivate().GetDisabled(),
				AuthorizedNetwork: req.GetEndpoints().GetPrivate().GetAuthorizedNetwork(),
			},
		},
		AuditLogConfig:            req.AuditLogConfig,
		State:                     regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING,
		EnhancedEncryptionEnabled: req.GetEncryptionConfig().GetEnhancedEncryptionEnabled(),
		ZoneRedundantEnabled:      req.GetZoneRedundantEnabled(),
		AutoScaling:               req.GetAutoScaling(),
	}

	return cvt.ClusterRegionalToGlobal(resp), nil
}

func (svc *clusterServiceImpl) regionLimitations(
	req *globalserverlessv1.ServerlessServiceCreateRequest,
) error {
	if req.GetZoneRedundantEnabled() &&
		lo.Contains(constant.ZoneRedundantBlockRegions, strings.ToLower(req.CloudProvider+"/"+req.Region)) {
		return bizerr.New(bizerr.ErrCodeParams, "zone redundant is not supported by cloud provider %s, region %s", req.CloudProvider, req.Region)
	}
	return nil
}
func (svc *clusterServiceImpl) buildClusterEntity(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceCreateRequest,
	cloudProviders []*entity.CloudProvider,
) (*entity.Cluster, error) {
	cloudProvider, ok := lo.Find(cloudProviders, func(c *entity.CloudProvider) bool {
		return c.Id.String() == req.CloudProvider
	})
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported cloud provider: %s", req.CloudProvider)
	}
	servicePlan := globalcommonv1.ServicePlan_Starter
	if req.GetAutoScaling().GetMinRcu() > 0 || req.GetAutoScaling().GetMaxRcu() > 0 {
		servicePlan = globalcommonv1.ServicePlan_Essential
	}
	engineType := req.GetEngineType()
	if engineType == "" {
		engineType = "MYSQL_COMPATIBLE"
		req.EngineType = engineType
	}
	var region *globalcommonv1.Region
	for _, r := range cloudProvider.Regions {
		regionEngineType := r.GetEngineType()
		if regionEngineType == globalcommonv1.Region_ENGINE_TYPE_UNSPECIFIED {
			regionEngineType = globalcommonv1.Region_MYSQL_COMPATIBLE
		}
		if r.Id.String() == req.Region &&
			r.State == globalcommonv1.Region_ACTIVE &&
			r.ServicePlan == servicePlan &&
			strings.EqualFold(regionEngineType.String(), engineType) {
			region = r
			break
		}
	}
	if region == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported region: %s service_plan: %s", req.Region, servicePlan)
	}
	if !region.IsPublic {
		if !lo.Contains(region.OpenedTenantIds, req.TenantId) {
			return nil, bizerr.New(
				bizerr.ErrCodeParams,
				"unsupported region: %s service_plan: %s for tenant %d",
				req.Region,
				servicePlan,
				req.TenantId,
			)
		}
	}

	encryptedCloudAdminPassword, err := svc.encryptorService.EncryptBase64(
		ctx,
		password.Decorate(uuid.NewV4().String()),
	)
	if err != nil {
		return nil, err
	}

	rootPwd := req.GetRootPassword()
	if rootPwd == "" {
		rootPwd = password.Decorate(uuid.NewV4().String())
	}
	encryptedRootPassword, err := svc.encryptorService.EncryptBase64(ctx, rootPwd)
	if err != nil {
		return nil, err
	}

	return &entity.Cluster{
		ServerlessServiceCluster: globalserverlessv1.ServerlessServiceCluster{
			Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
				AutoScaling: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{
					MinRcu: req.GetAutoScaling().GetMinRcu(),
					MaxRcu: req.GetAutoScaling().GetMaxRcu(),
				},
			},
			ClusterId:  req.Id,
			ProjectId:  req.ProjectId,
			TenantId:   req.TenantId,
			HasSetPwd:  req.GetRootPassword() != "",
			UserStatus: globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING,
			RuVersion:  req.RuVersion,
			EngineType: engineType,
		},
		EncryptedCloudAdminPassword: encryptedCloudAdminPassword,
		EncryptedRootPassword:       encryptedRootPassword,
		Region:                      region,
		CloudProvider:               cloudProvider,
		ServicePlan:                 servicePlan,
		EngineType:                  engineType,
	}, nil
}