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"
	"google.golang.org/protobuf/types/known/structpb"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/cmek"
	"github.com/tidbcloud/tidb-management-service/dedicated-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/featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/randstr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	minClusterPasswordLength = 8
	maxClusterPasswordLength = 64
	defaultTiDBPort          = 4000
	defaultRootPasswdLen     = 16 // default length of root password
	defaultTidbGroupName     = "DefaultGroup"

	raftDirPathKey      = "raft-engine.dir"
	raftSpillDirPathKey = "raft-engine.spill-dir"
)

func (s *serviceImpl) Create(
	ctx context.Context,
	req *globaldedicatedv1.CreateClusterRequest,
) (*entity.Cluster, bizerr.BizError) {
	req.GetCluster().ClusterComponents = lo.Filter(
		req.GetCluster().GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent, _ int) bool {
			return comp.GetCount() > 0
		},
	)
	cluster := &entity.Cluster{Cluster: req.Cluster}

	err := s.transaction.Transaction(ctx, func(ctx context.Context) error {
		// Idempotency check: if cluster ID is provided and cluster already exists, skip creation
		if req.GetCluster().GetClusterId() != 0 {
			existingCluster, err := s.clusterRepo.Get(ctx, req.GetCluster().GetClusterId(), false)
			if err != nil {
				// If error is not "record not found", return the error
				if err.RawErrCode() != bizerr.ErrCodeRecordNotFound {
					return err
				}
				// Otherwise, cluster doesn't exist, continue with creation
			} else if existingCluster != nil {
				// Cluster already exists, this is a duplicate request
				tilog.Info(ctx, "cluster already exists, skipping creation", zap.Uint64("cluster_id", req.GetCluster().GetClusterId()))
				// Copy the existing cluster to our local variable so it can be returned
				cluster = existingCluster
				return nil
			}
		}

		// Load and check whether cloud provider and region are active.
		cloudProvider, region, err := s.checkCloudProviderAndRegion(ctx, req)
		if err != nil {
			return err
		}

		// Check whether the cloud provider is supported for the cluster.
		if err := checkSupportedCloudProvider(cluster.GetClusterComponents(), cloudProvider.GetId()); err != nil {
			return err
		}

		// Get project and organization from account service.
		tenant, project, err := s.getTenantAndProjectMeta(ctx, req.Cluster.ProjectId)
		if err != nil {
			return err
		}

		// Validate cmek.
		cmekId, err := s.validateCmek(
			ctx,
			tenant,
			project,
			region,
			cloudProvider,
			req.Cluster.CustomerManagedEncryptionKeyId,
		)
		if err != nil {
			return err
		}

		// Validate port.
		if err := s.validatePort(req.Cluster.Port); err != nil {
			return err
		}

		// Validate password.
		if err := s.validatePassword(req.Cluster.RootPassword); err != nil {
			return err
		}

		// Hack here, if cluster id is empty, then skip cidr, billing plan and tidb group check. Only for calc cluster fee scenario.
		if req.GetCluster().GetClusterId() != 0 {
			// Check whether region CIDR is set.
			if err := s.checkCIDR(ctx, project, region); err != nil {
				return err
			}

			// Check whether organization allow to create dedicated cluster.
			if err := s.checkTenantBillingPlan(ctx, tenant, cloudProvider, region); err != nil {
				return err
			}

			// check whether tidb group component is valid.
			// only support one tidb group component and one tiproxy group component.
			if err := checkTidbGroupComp(cluster); err != nil {
				return err
			}

			// Check TiProxy replica count only for real cluster creation (cluster_id != 0).
			// For calc-cluster-fee requests, cluster_id is 0 and we intentionally skip this check.
			if err := tiproxyReplicaCountCheck(cluster); err != nil {
				return err
			}
		}

		cluster.RegionDisplayName = region.DisplayName
		cluster.CustomerManagedEncryptionKeyId = cmekId

		// Get cluster version is not specified.
		// only support set tidb version for test-infra or restore yet.
		if cluster.Version == "" {
			clusterVersion, err := s.clusterConfigCli.GetClusterVersion(ctx, cluster)
			if err != nil {
				return err
			}
			cluster.Version = clusterVersion
		}

		cluster.DesiredComponents, err = s.buildDesiredComponents(
			ctx,
			[]*entity.TidbNodeGroup{{}},
			cluster,
			cluster.GetClusterComponents(),
			true,
		)
		if err != nil {
			return err
		}

		if err := s.CheckCreateDesiredComponents(ctx, cluster, cluster.DesiredComponents); err != nil {
			return err
		}

		// Generate cluster dns id for dns record.
		cluster.ClusterIdentifierHash = randstr.GenDnsIdentity()

		if cluster.Port == 0 {
			cluster.Port = defaultTiDBPort
		}

		// Save root password.
		var rootPassword string
		if cluster.RootPassword == "" {
			// generate a temp random password for root
			rootPassword = randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
		} else {
			// set root password on request, use it directly
			rootPassword = cluster.RootPassword
			// Root password not returned to the user.
			cluster.RootPassword = ""
			cluster.HasRootPassword = true
		}

		encryptedRootPassword, err := s.encryptor.EncryptBase64(ctx, rootPassword)
		if err != nil {
			return bizerr.WrapError(err, "encrypt root password failed")
		}

		// set encrypted cloud_admin password
		if req.Cluster.GetEncryptedCloudAdminPassword() != "" {
			cluster.EncryptedCloudAdminPassword = req.Cluster.GetEncryptedCloudAdminPassword()
		} else {
			cloudAdminPassword := randstr.GenWithCharset(defaultRootPasswdLen, randstr.AlphanumericCharset)
			cluster.EncryptedCloudAdminPassword, err = s.encryptor.EncryptBase64(ctx, cloudAdminPassword)
			if err != nil {
				return bizerr.WrapError(err, "failed to encrypt cloud admin password")
			}
		}

		// set cluster to creating state.
		cluster.State = globaldedicatedv1.Cluster_CREATING
		cluster.HealthState = globaldedicatedv1.Cluster_UNHEALTHY

		cluster.OnNewArch = true

		// Fast return and prevent repo saving if validate only.
		if req.ValidateOnly {
			return nil
		}

		// Save cluster record to db.
		if err := s.clusterRepo.Save(ctx, cluster); err != nil {
			return err
		}
		if err := s.clusterRepo.SetEncryptedRootPassword(ctx, cluster.ClusterId, encryptedRootPassword); err != nil {
			return err
		}

		// Create default work pool.
		defaultTidbGroupId, err := s.createDefaultTidbGroup(ctx, cluster)
		if err != nil {
			return err
		}
		if err := s.componentRepo.SaveAll(ctx, cluster.DesiredComponents, cluster.ClusterId, defaultTidbGroupId); err != nil {
			return err
		}

		// Create cluster event.
		afterTopology := entity.BuildClusterTopology(cluster, defaultTidbGroupId)
		creatingEvent := entity.NewClusterEventBuilder(ctx, cluster, clusterevent.EventTypeCreating).
			SetTopologyAfter(afterTopology).
			Build()
		// Initialize cluster event.
		initializingEvent := entity.NewClusterEventBuilder(ctx, cluster, clusterevent.EventTypeInitializing).
			SetTopologyAfter(afterTopology).
			Build()
		if err := s.eventCommitter.BatchWrite(
			ctx,
			clusterevent.TopicClusterEvent,
			[]clusterevent.Event{creatingEvent, initializingEvent},
		); err != nil {
			return err
		}

		// Cluster configuration
		clusterConfiguration, err := s.patchSeparatedRaftDiskConfiguration(cluster, req.ClusterConfiguration)
		if err != nil {
			return err
		}

		// metadata
		metadataJson, merr := metadata.MarshalRequestMetaFromCtx(ctx)
		if merr != nil {
			tilog.Warn(ctx, "failed to marshal request metadata", zap.Error(merr))
		}

		// Start async reconciliation workflow.
		workFlowErr := s.workflow.RunDagWithBizId(
			ctx,
			common.CreateCluster,
			strconv.FormatUint(cluster.OrganizationId, 10),
			strconv.FormatUint(cluster.ProjectId, 10),
			fmt.Sprintf("%d", cluster.ClusterId),
			map[string]string{
				workflow.KeyTraceId:                 workflow.GetTraceId(ctx),
				workflow.KeyRequestMeta:             string(metadataJson),
				common.KeyClusterId:                 strconv.FormatUint(req.Cluster.ClusterId, 10),
				common.KeyTargetStatus:              globaldedicatedv1.Cluster_AVAILABLE.String(),
				common.KeyTidbGroupId:               strconv.FormatUint(defaultTidbGroupId, 10),
				common.KeyClusterConfiguration:      clusterConfiguration,
				common.KeyOverwriteTenantInitConfig: strconv.FormatBool(req.OverwriteTenantInitConfig),
			},
		)
		if workFlowErr != nil {
			return bizerr.WrapErrorWithCode(workFlowErr, bizerr.ErrCodeUnknown, "create async workflow failed")
		}

		return nil
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "create cluster failed")
	}

	if req.ValidateOnly {
		// For validation-only requests (dry run), return the cluster with desired components directly
		// since no actual cluster creation will occur. The desired components represent the complete
		// set of components that would be created if this were a real cluster creation request.
		cluster.ClusterComponents = lo.Map(
			cluster.DesiredComponents,
			func(desiredComp *entity.DesiredClusterComponent, _ int) *globaldedicatedv1.Cluster_ClusterComponent {
				return desiredComp.ClusterComponent
			},
		)
		return &entity.Cluster{Cluster: cluster.Cluster}, nil
	}

	newCluster, err := s.clusterRepo.GetAndLoadComponents(ctx, req.Cluster.ClusterId)
	if err != nil {
		return nil, bizerr.WrapGormError(err, "find cluster failed")
	}

	return newCluster, nil
}

func checkTidbGroupComp(cluster *entity.Cluster) bizerr.BizError {
	tidbComps := lo.Filter(
		cluster.GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent, _ int) bool {
			return comp.ComponentType == globaldedicatedv1.ComponentType_TIDB
		},
	)
	if len(tidbComps) == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "tidb component must be specified")
	}
	if len(tidbComps) > 1 {
		return bizerr.New(bizerr.ErrCodeParams, "only one tidb component is allowed")
	}
	if tidbComps[0].GetTidbGroupId() != 0 {
		return bizerr.New(bizerr.ErrCodeParams, "tidb group id must not be specified when creating cluster request")
	}

	tiproxyComps := lo.Filter(
		cluster.GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent, _ int) bool {
			return comp.ComponentType == globaldedicatedv1.ComponentType_TIPROXY
		},
	)
	if len(tiproxyComps) > 1 {
		return bizerr.New(bizerr.ErrCodeParams, "only one tiproxy component is allowed")
	}
	if len(tiproxyComps) == 1 && tiproxyComps[0].GetTidbGroupId() != 0 {
		return bizerr.New(bizerr.ErrCodeParams, "tiproxy group id must not be specified when creating cluster request")
	}
	return nil
}

func (s *serviceImpl) getTenantAndProjectMeta(
	ctx context.Context,
	projectId uint64,
) (*account.TenantInfo, *account.ProjectInfo, bizerr.BizError) {
	project, err := s.accountCli.GetProject(ctx, projectId)
	if err != nil {
		return nil, nil, err
	}

	tenant, err := s.accountCli.GetTenant(ctx, project.TenantID)
	if err != nil {
		return nil, nil, err
	}

	return tenant, project, nil
}

func (s *serviceImpl) validateCmek(
	ctx context.Context,
	tenant *account.TenantInfo,
	project *account.ProjectInfo,
	region *entity.Region,
	provider *entity.CloudProvider,
	clusterCmekId string,
) (string, bizerr.BizError) {
	// skip if aws cmek is not enabled for the project
	if !project.AwsCmekEnabled {
		return "", nil
	}

	// check whether aws cmek feature is enabled for the project and region.
	awsCmekFeature := s.featGateCli.CheckFeatureEnabled(
		featuregate.AwsCmek,
		tenant.ID,
		project.ID,
		region.CloudProviderId.String(),
		region.Id.String(),
	)
	if !awsCmekFeature {
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"CMEK feature gate is not enabled for project %d, region %s",
			project.ID,
			region.Id,
		)
	}

	// only support AWS and AZURE CMEK yet
	if provider.Id != globalcommonv1.CloudProvider_AWS && provider.Id != globalcommonv1.CloudProvider_AZURE {
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"CMEK feature is enabled, creation for clusters on %s is not supported",
			provider.Id,
		)
	}

	// check whether the cmek id is valid
	filter := cmek.Filter{
		OrganizationIds:    []uint64{tenant.ID},
		ProjectIds:         []uint64{project.ID},
		RegionNames:        []entity.RegionName{region.Id},
		CloudProviderNames: []entity.CloudProviderName{region.CloudProviderId},
	}
	cmeks, err := s.cmekRepo.Query(ctx, &filter)
	if err != nil {
		return "", bizerr.WrapGormError(err, "query CMEK failed")
	}
	if len(cmeks) == 0 {
		return "", bizerr.New(bizerr.ErrCodeParams, "no CMEK found for project %d, region %s", project.ID, region.Id)
	}

	// set default cmek id if not specified ?
	if clusterCmekId == "" {
		// return "", bizerr.New(bizerr.ErrCodeParams, "CMEK id is required")
		tilog.Warn(ctx, "CMEK id is not specified, use the first one as default")
		return strconv.FormatUint(cmeks[0].Id, 10), nil
	}

	cmekId, serr := strconv.ParseUint(clusterCmekId, 10, 64)
	if serr != nil {
		return "", bizerr.New(bizerr.ErrCodeParams, "CMEK id %s is invalid", clusterCmekId)
	}

	// check whether the cmek id exists
	for _, cmekEn := range cmeks {
		if cmekEn.Id == cmekId {
			return clusterCmekId, nil
		}
	}

	return "", bizerr.New(bizerr.ErrCodeParams, "CMEK id %s not found", clusterCmekId)
}

func (s *serviceImpl) validatePort(port uint32) bizerr.BizError {
	// zero means don't specific port
	if port == 0 {
		return nil
	}
	if port == 10080 || port < 1024 || port > 65535 {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"invalid port: %d, port must be in the range of 1024-65535 except 10080",
			port,
		)
	}
	return nil
}

func (s *serviceImpl) validatePassword(password string) bizerr.BizError {
	// means use auto-generated password
	if password == "" {
		return nil
	}

	passwordLen := len(password)
	if passwordLen < minClusterPasswordLength || passwordLen > maxClusterPasswordLength {
		return bizerr.New(bizerr.ErrCodeParams, "password must be 8-64 characters")
	}
	return nil
}

func (s *serviceImpl) checkTenantBillingPlan(
	ctx context.Context,
	tenant *account.TenantInfo,
	cloudProvider *entity.CloudProvider,
	region *entity.Region,
) bizerr.BizError {
	result, err := s.billingCli.BillingActionCheck(ctx, billing.Action_Create, tenant.ID, &billing.Resource{
		Type:     billing.ResourceType_DedicatedCluster,
		Region:   lo.ToPtr(region.GetId().String()),
		Provider: lo.ToPtr(billing.ProviderMap[cloudProvider.Id]),
		IsFree:   false,
	})
	if err != nil {
		return err
	}
	if result.Pass {
		return nil
	}
	if result.Reason != nil {
		// Use the reason from billing service if it is set as the error message to the user.
		return bizerr.New(bizerr.ErrCodeParams, "%s", *result.Reason)
	}
	// If the reason is not set, return a generic error message.
	return bizerr.New(bizerr.ErrCodeParams, "Unknown tenant plan %s", tenant.Plan)
}

func (s *serviceImpl) checkCIDR(
	ctx context.Context,
	project *account.ProjectInfo,
	region *entity.Region,
) bizerr.BizError {
	_, err := s.cidrRepo.LoadByCloudProviderAndRegion(
		ctx,
		project.ID,
		project.TenantID,
		region.CloudProviderId,
		region.Id,
	)
	if err != nil {
		if err.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"The CIDR is not set for the region and the cluster cannot be created",
			)
		}
		return err
	}
	return nil
}

func (s *serviceImpl) createDefaultTidbGroup(ctx context.Context, cluster *entity.Cluster) (uint64, bizerr.BizError) {
	id := s.idGeneratorAdapter.MustNew()
	group := &entity.TidbNodeGroup{
		Id:          id,
		DisplayName: defaultTidbGroupName,
		Default:     true,
		ClusterId:   cluster.ClusterId,
	}
	return id, s.tidbGroupRepo.Create(ctx, group)
}

func (s *serviceImpl) checkCloudProviderAndRegion(
	ctx context.Context,
	req *globaldedicatedv1.CreateClusterRequest,
) (*entity.CloudProvider, *entity.Region, bizerr.BizError) {
	region, bErr := s.regionRepo.SelectRegionById(ctx, req.Cluster.RegionId, req.Cluster.OrganizationId)
	if bErr != nil {
		return nil, nil, bErr
	}
	cloudProvider, bErr := s.regionRepo.GetCloudProviderById(
		ctx,
		region.CloudProviderId,
		req.GetCluster().GetOrganizationId(),
	)
	if bErr != nil {
		return nil, nil, bErr
	}

	if cloudProvider.State != globalcommonv1.CloudProvider_ACTIVE {
		return nil, nil, bizerr.New(bizerr.ErrCodeParams, "inactive cloud provider %s", cloudProvider.Id)
	}

	if region.State != globalcommonv1.Region_ACTIVE {
		return nil, nil, bizerr.New(
			bizerr.ErrCodeParams,
			"inactive region %s in %s cloud provider",
			region.Id,
			cloudProvider.Id,
		)
	}

	if region.CloudProviderId != req.Cluster.CloudProviderId {
		return nil, nil, bizerr.New(
			bizerr.ErrCodeParams,
			"region %s is not in cloud provider %s",
			region.Id,
			cloudProvider.Id,
		)
	}

	return cloudProvider, region, nil
}

func isRaftDiskSeparated(cluster *entity.Cluster) bool {
	tikvComp, ok := lo.Find(cluster.DesiredComponents, func(comp *entity.DesiredClusterComponent) bool {
		return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
	})
	if !ok {
		return false
	}
	// Only check tikv component.
	return tikvComp.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].GetVolumeType() != globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED
}

func (s *serviceImpl) patchSeparatedRaftDiskConfiguration(
	cluster *entity.Cluster,
	cfg *globaldedicatedv1.ClusterConfiguration,
) (string, bizerr.BizError) {
	raftDiskSeparated := isRaftDiskSeparated(cluster)

	// If raft disk is not separated, return marshalled configuration directly.
	if !raftDiskSeparated {
		// no custom configuration, return empty string
		if cfg == nil {
			return "", nil
		}

		// return marshalled custom configuration.
		cfgJson, err := protojson.Marshal(cfg)
		if err != nil {
			return "", bizerr.New(bizerr.ErrCodeUnknown, "marshal cluster configuration failed")
		}
		return string(cfgJson), nil
	}

	// Only patch raft log dir if the selected desired tikv component has separated raft log disk.
	if cfg == nil {
		cfg = &globaldedicatedv1.ClusterConfiguration{}
	}

	raftDirCfg := &globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem{
		Value:   structpb.NewStringValue(entity.DefaultRaftLogDirPath),
		Comment: "the path to store raft engine logs with",
	}
	raftSpillDirCfg := &globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem{
		Value:   structpb.NewStringValue(entity.DefaultRaftLogSpillPath),
		Comment: "the path to store raft engine spill logs with",
	}

	// add raftstore config
	var hasTiKVConfig bool
	for i, compCfg := range cfg.GetComponentConfigurations() {
		if compCfg.GetComponentType() != globaldedicatedv1.ComponentType_TIKV {
			continue
		}

		// set raftstore config
		if cfg.GetComponentConfigurations()[i].CustomParams == nil {
			cfg.GetComponentConfigurations()[i].CustomParams = make(
				map[string]*globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem,
			)
		}

		cfg.GetComponentConfigurations()[i].CustomParams[raftDirPathKey] = raftDirCfg
		cfg.GetComponentConfigurations()[i].CustomParams[raftSpillDirPathKey] = raftSpillDirCfg
		hasTiKVConfig = true
	}

	// append raftstore config
	if !hasTiKVConfig {
		cfg.ComponentConfigurations = append(
			cfg.ComponentConfigurations,
			&globaldedicatedv1.ClusterConfiguration_ComponentConfiguration{
				ComponentType: globaldedicatedv1.ComponentType_TIKV,
				CustomParams: map[string]*globaldedicatedv1.ClusterConfiguration_ComponentConfiguration_ParamItem{
					raftDirPathKey:      raftDirCfg,
					raftSpillDirPathKey: raftSpillDirCfg,
				},
			},
		)
	}

	cfgJson, err := protojson.Marshal(cfg)
	if err != nil {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "marshal cluster configuration failed")
	}

	return string(cfgJson), nil
}

// checkSupportedCloudProvider checks whether the cloud provider is supported for the cluster.
// Currently, only AWS is supported for the tiproxy component.
func checkSupportedCloudProvider(
	comps []*globaldedicatedv1.Cluster_ClusterComponent,
	cloudProvider globalcommonv1.CloudProvider_Name,
) bizerr.BizError {
	tiproxyComp, ok := lo.Find(comps, func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
		return comp.ComponentType == globaldedicatedv1.ComponentType_TIPROXY
	})
	if ok && tiproxyComp != nil {
		if cloudProvider != globalcommonv1.CloudProvider_AWS {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"tiproxy component is not supported in %s cloud provider",
				cloudProvider.String(),
			)
		}
	}
	return nil
}