package entity

import (
	"fmt"
	"net/url"
	"path"
	"path/filepath"
	"regexp"
	"sort"
	"strings"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dpapi "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	mysqlschema "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
)

func GetUserNamespace(tenantId string) string {
	return fmt.Sprintf("%s-%s", schemav1.ClusterNGServiceType, tenantId)
}

func GetNamespace() string {
	return defaultNamespace
}

func GetDataPlaneNamespace(sharedPoolId string) string {
	return fmt.Sprintf("tidb%s", sharedPoolId)
}

// ResolveSharedPoolTenantID returns the tenant identity from the Dataplane tag,
// which is the authoritative source for SharedPools loaded from storage.
func ResolveSharedPoolTenantID(sharedPool *SharedPool) (string, error) {
	if sharedPool == nil {
		return "", fmt.Errorf("shared pool is nil")
	}

	tenantID := sharedPool.GetProvider().GetTags()[schemav1.LabelUserTenantKey]
	if tenantID != "" {
		return tenantID, nil
	}
	return "", fmt.Errorf("shared pool tenant id is unavailable")
}

// HasConfigCenterMetaProfile reports whether a non-empty profile ID is configured.
func (sharedPool *SharedPool) HasConfigCenterMetaProfile() bool {
	return sharedPool != nil && sharedPool.ConfigCenterMetaProfileId != ""
}

func (sharedPool *SharedPool) GetDataPlaneNamespace() string {
	return GetDataPlaneNamespace(sharedPool.GetId())
}

// func (sharedPool *SharedPool) GetDataPlaneWorkerNamespace() string {
// 	return GetDataPlaneWorkerNmaespace()
// }

func (sharedPool *SharedPool) GetNodeRoleName() string {
	return fmt.Sprintf("node-role-%s", sharedPool.Id)
}

func (sharedPool *SharedPool) GetDefaultComponenId() string {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_PD {
			return comp.GetId()
		}
	}
	return ""
}

// GetPerPoolHPAComponentIds returns the component IDs that
// ApplyComponentHpaAction should target at SHARED_POOL scope: every
// per-pool worker component (TIDB_WORKER / TIKV_WORKER /
// COPROCESSOR_WORKER) PLUS the pool-shared TIPROXY when
// HasSharedTiProxy(sharedPool) is true.
//
// Pre-fix (and the wrong behavior this helper replaces): callers
// used GetDefaultComponenId(), which returns the PD component ID.
// CheckIsTargetComponent then filtered every worker out — the 3
// worker HPAs were never created in K8s. See PR-C.1 commit message
// for context.
//
// Component IDs come from each ComponentGroup entry's comp.GetId();
// pool-shared TIPROXY uses utils.FnvHash(sharedPool.GetId()) per
// generatedSharedPoolComponent. The helper is self-correcting: if
// a future topology adds a new per-pool HPA-eligible component
// type to EnableHAPComponent, this helper picks it up automatically.
func (sharedPool *SharedPool) GetPerPoolHPAComponentIds() []string {
	if sharedPool == nil {
		return nil
	}
	hapTypeSet := make(map[apiv1.ComponentType]struct{}, len(EnableHAPComponent))
	for _, t := range EnableHAPComponent {
		hapTypeSet[t] = struct{}{}
	}
	var ids []string
	for _, comp := range sharedPool.GetComponentGroups() {
		if _, ok := hapTypeSet[comp.GetComponentType()]; !ok {
			continue
		}
		ids = append(ids, comp.GetId())
	}
	if HasSharedTiProxy(sharedPool) {
		ids = append(ids, utils.FnvHash(sharedPool.GetId()))
	}
	return ids
}

func (tidbGroup *TiDBGroup) GetTiproxyId() string {
	return tidbGroup.GetClusterId()
}

// HasSharedTiProxy check whether the tiproxy is shared in shared pool
// NOTE: It only relies on persisted schema state.
func HasSharedTiProxy(sharedPool *SharedPool) bool {
	return sharedPool.DeployType == mysqlschema.SharedPoolDeployTypeSharedTiProxy
}

// IsEssentialV2 reports whether the shared pool belongs to the Essential v2
// service plan. Callers reconstructing a pool from workflow args must restore
// DeployType before using this predicate.
func IsEssentialV2(sharedPool *SharedPool) bool {
	return HasSharedTiProxy(sharedPool)
}

// UsesTiFlashColumnar reports whether the shared pool uses the TiFlash
// Columnar architecture (schema-manager + tiflash-compute, no tiflash-write).
// Essential v2 pools predate the tiflash_mode marker, so shared-TiProxy pools
// are treated as columnar for backward compatibility.
// NOTE: It only relies on persisted schema state.
func UsesTiFlashColumnar(sharedPool *SharedPool) bool {
	return sharedPool.TiflashMode == mysqlschema.SharedPoolTiflashModeColumnar ||
		HasSharedTiProxy(sharedPool)
}

func ResolveTiProxyID(sharedPool *SharedPool, clusterId string) string {
	if !HasSharedTiProxy(sharedPool) {
		return clusterId
	}

	if defaultCompID := sharedPool.GetDefaultComponenId(); defaultCompID != "" {
		return defaultCompID
	}

	return clusterId
}

func FindTiProxyComponent(sharedPool *SharedPool, clusterId string) *Component {
	tiproxyID := ResolveTiProxyID(sharedPool, clusterId)
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_TIPROXY && comp.GetId() == tiproxyID {
			return comp
		}
	}
	return nil
}

const (
	TiFlashWriteType      = "write-tiflash"
	TiFlashComputeType    = "compute-tiflash"
	TiDBWorkerType        = "worker-tidb"
	TiDBPrivilegedType    = "privileged-tidb"
	TiDBDiagnosisType     = "diagnosis-tidb"
	TiKVWorkerType        = "default-tikv-worker"
	CoprocessorWorkerType = "coprocessor-tikv-worker"
	SchemaManagerType     = "schema-manager-tikv-worker"
)

func GetComponentTypeName(comp *Component) string {
	componentType := comp.GetComponentType().String()
	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIFLASH_COMPUTE:
		componentType = TiFlashComputeType
	case apiv1.ComponentType_TIFLASH_WRITE:
		componentType = TiFlashWriteType
	case apiv1.ComponentType_TIDB_WORKER:
		componentType = TiDBWorkerType
	case apiv1.ComponentType_PD_TSO:
		componentType = "tso"
	case apiv1.ComponentType_PD_SCHEDULING:
		componentType = "scheduling"
	case apiv1.ComponentType_TIKV_WORKER:
		componentType = TiKVWorkerType
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		componentType = CoprocessorWorkerType
	case apiv1.ComponentType_SCHEMA_MANAGER:
		componentType = SchemaManagerType
	}

	return strings.ToLower(strings.ReplaceAll(componentType, "_", "-"))
}

func GetUpdateComponentKey(comp *Component) string {
	return fmt.Sprintf("%s-%s", GetComponentTypeName(comp), comp.GetId())
}

func GetPublicSvcName(compId string) string {
	return fmt.Sprintf("public-%s", compId)
}

func GetPrivateSvcName(compId string) string {
	return fmt.Sprintf("private-%s", compId)
}

const (
	DataBucketSuffix     = "data"
	InternalBucketSuffix = "internal"
)

func (sharedPool *SharedPool) GetDfsPrefix() string {
	return filepath.Join(sharedPool.GetObjectStorage().GetPrefix(), "cse")
}

func (sharedPool *SharedPool) GetTiFlashRootPrefix() string {
	return filepath.Join(sharedPool.GetObjectStorage().GetPrefix(), "tiflash-root")
}

func (sharedPool *SharedPool) GetTiCIPrefix() string {
	return filepath.Join(sharedPool.GetObjectStorage().GetPrefix(), "tici")
}

func (sharedPool *SharedPool) GetDXFGlobalSortPath() (string, error) {
	ob := sharedPool.GetObjectStorage()

	if sharedPool.GetInternalStorage() != nil && sharedPool.GetInternalStorage().GetBucket() != "" {
		ob = sharedPool.GetInternalStorage()
	}

	switch sharedPool.GetProvider().GetType() {
	case apiv1.ProviderType_AWS:
		return fmt.Sprintf("s3://%s", filepath.Join(ob.GetBucket(), ob.GetPrefix(), "global-sort")), nil
	case apiv1.ProviderType_GCP:
		return fmt.Sprintf("gs://%s", filepath.Join(ob.GetBucket(), ob.GetPrefix(), "global-sort")), nil
	case apiv1.ProviderType_AZURE:
		account, container, err := splitAzureBucket(ob.GetBucket())
		if err != nil {
			return "", err
		}
		query := make(url.Values)
		query.Add("account-name", account)
		return fmt.Sprintf("azure://%s?%s", path.Join(container, ob.GetPrefix(), "global-sort"), query.Encode()), nil
	case apiv1.ProviderType_ALICLOUD:
		return fmt.Sprintf("oss://%s", filepath.Join(ob.GetBucket(), ob.GetPrefix(), "global-sort")), nil
	default:
		return "", fmt.Errorf("not supported provider type %s", sharedPool.GetProvider().GetType())
	}
}

func (sharedPool *SharedPool) GetDfxEndpoint() string {
	switch sharedPool.GetProvider().GetType() {
	case apiv1.ProviderType_AWS:
		return ""
	case apiv1.ProviderType_GCP:
		return "storage.googleapis.com"
	case apiv1.ProviderType_AZURE:
		account, _, err := splitAzureBucket(sharedPool.GetObjectStorage().GetBucket())
		if err != nil {
			return ""
		}
		return fmt.Sprintf("%s.blob.core.windows.net", account)
	case apiv1.ProviderType_ALICLOUD:
		return fmt.Sprintf("oss-%s-internal.aliyuncs.com", sharedPool.GetProvider().GetRegion())
	default:
		return ""
	}
}

func (sharedPool *SharedPool) GetAzureStorageAccountAndContainer() (string, string, error) {
	if sharedPool.GetProvider().GetType() != apiv1.ProviderType_AZURE {
		return "", "", fmt.Errorf("provider type %s is not azure", sharedPool.GetProvider().GetType())
	}
	return splitAzureBucket(sharedPool.GetObjectStorage().GetBucket())
}

// splitAzureBucket splits an Azure bucket into storage account and container
// azure bucket format: <storage account>/<container>
// azure blob storage has storage account and container, we use container corresponding to bucket of other cloud providers.
// to keep compatibility with current fields, we add storage account and container to the bucket field.
func splitAzureBucket(bucket string) (string, string, error) {
	parts := strings.SplitN(bucket, "/", 2)
	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
		return "", "", fmt.Errorf("invalid azure bucket format %s, expected <account>/<container>", bucket)
	}
	return parts[0], parts[1], nil
}

func GetDataBucketId(sharedPoolId, region string) string {
	return fmt.Sprintf("%s-%s-%s", sharedPoolId, region, DataBucketSuffix)
}

func GetInternalBucketId(sharedPoolId, region string) string {
	return fmt.Sprintf("%s-%s-%s", sharedPoolId, region, InternalBucketSuffix)
}

func GetComponentGroupName(comp *Component) string {
	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIKV_WORKER:
		return fmt.Sprintf("db-%s-default", comp.GetId())
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		return fmt.Sprintf("db-%s-coprocessor", comp.GetId())
	case apiv1.ComponentType_TIFLASH_COMPUTE:
		return fmt.Sprintf("db-%s-compute", comp.GetId())
	case apiv1.ComponentType_TIFLASH_WRITE:
		return fmt.Sprintf("db-%s-write", comp.GetId())
	case apiv1.ComponentType_TIDB_WORKER:
		return fmt.Sprintf("db-%s-worker", comp.GetId())
	case apiv1.ComponentType_TICI_META:
		return fmt.Sprintf("db-%s-tici-meta", comp.GetId())
	case apiv1.ComponentType_TICI_WORKER:
		return fmt.Sprintf("db-%s-tici-worker", comp.GetId())
	case apiv1.ComponentType_SCHEMA_MANAGER:
		return fmt.Sprintf("db-%s-schema-manager", comp.GetId())
	default:
		return fmt.Sprintf("db-%s", comp.GetId())
	}
}

func GetComponentTemplateName(comp *Component) (string, error) {
	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER:
		return "tikv_worker_group", nil
	case apiv1.ComponentType_TIFLASH_COMPUTE, apiv1.ComponentType_TIFLASH_WRITE:
		return "tiflash_group", nil
	case apiv1.ComponentType_TIDB:
		return "tidb_group", nil
	case apiv1.ComponentType_TIDB_WORKER:
		return "tidb_worker_group", nil
	case apiv1.ComponentType_PD:
		return "pd_group", nil
	case apiv1.ComponentType_PD_SCHEDULING:
		return "scheduling_group", nil
	case apiv1.ComponentType_PD_TSO:
		return "tso_group", nil
	case apiv1.ComponentType_TIKV:
		return "tikv_group", nil
	case apiv1.ComponentType_TIPROXY:
		return "tiproxy_group", nil
	case apiv1.ComponentType_RESOURCE_MANAGER:
		return "resource_manager_group", nil
	case apiv1.ComponentType_CASSANDRA_PROXY:
		return "cassandra_proxy", nil
	case apiv1.ComponentType_SCHEMA_MANAGER:
		return "tikv_worker_group", nil
	}
	return "", fmt.Errorf("not found template for component type %s", comp.GetComponentType())
}

func GetComponentHPATemplateName(comp *Component) (string, error) {
	switch comp.GetComponentType() {
	case apiv1.ComponentType_TIKV_WORKER:
		return "tikv_worker_hpa", nil
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		return "copr_worker_hpa", nil
	case apiv1.ComponentType_TIDB:
		return "tidb_group_hpa", nil
	case apiv1.ComponentType_TIDB_WORKER:
		return "tidb_worker_hpa", nil
	case apiv1.ComponentType_TIPROXY:
		return "tiproxy_group_hpa", nil
	case apiv1.ComponentType_CASSANDRA_PROXY:
		return "cassandra_proxy_hpa", nil
	case apiv1.ComponentType_TIFLASH_COMPUTE, apiv1.ComponentType_TIFLASH_WRITE:
		return "", fmt.Errorf("hpa is not supported for component type %s", comp.GetComponentType())
	case apiv1.ComponentType_PD, apiv1.ComponentType_PD_SCHEDULING, apiv1.ComponentType_PD_TSO:
		return "", fmt.Errorf("hpa is not supported for component type %s", comp.GetComponentType())
	case apiv1.ComponentType_TIKV:
		return "", fmt.Errorf("hpa is not supported for component type %s", comp.GetComponentType())
	case apiv1.ComponentType_SCHEMA_MANAGER:
		return "", fmt.Errorf("hpa is not supported for component type %s", comp.GetComponentType())
	}
	return "", fmt.Errorf("not found template for component type %s", comp.GetComponentType())
}

func GetComponentSvcName(cluster *SharedPool, comp *Component) string {
	if comp.GetComponentType() == apiv1.ComponentType_TICI_META ||
		comp.GetComponentType() == apiv1.ComponentType_TICI_WORKER {
		return fmt.Sprintf("%s-peer", GetComponentGroupName(comp))
	}

	if comp.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE ||
		comp.GetComponentType() == apiv1.ComponentType_TIFLASH_WRITE {
		return fmt.Sprintf("%s-tiflash-peer", GetComponentGroupName(comp))
	}

	if comp.GetComponentType() == apiv1.ComponentType_TIDB_WORKER {
		return fmt.Sprintf("%s-tidb-peer", GetComponentGroupName(comp))
	}

	if comp.GetComponentType() == apiv1.ComponentType_COPROCESSOR_WORKER {
		return fmt.Sprintf("%s-tikv-worker-peer", GetComponentGroupName(comp))
	}

	if comp.GetComponentType() == apiv1.ComponentType_TIKV_WORKER {
		return fmt.Sprintf("%s-tikv-worker-peer", GetComponentGroupName(comp))
	}

	if comp.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER {
		return fmt.Sprintf("%s-tikv-worker-peer", GetComponentGroupName(comp))
	}

	return fmt.Sprintf("%s-%s-peer", GetComponentGroupName(comp), GetComponentTypeName(comp))
}

func GetComponentNodeSpec(nodeSpec *NodeSpec) string {
	return fmt.Sprintf("%dC%dG", nodeSpec.GetCpu(), nodeSpec.GetMemoryGi())
}

func (sharedPool *SharedPool) GetPdAddress() string {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_PD {
			return fmt.Sprintf("%s.%s", GetComponentSvcName(sharedPool, comp), sharedPool.GetDataPlaneNamespace())
		}
	}
	return ""
}

func BuildTiDBGroupTags(tidbGroup *TiDBGroup) map[string]string {
	return map[string]string{
		schemav1.LabelServiceTypeKey: schemav1.ClusterNGServiceType,
		schemav1.LabelUsedForKey:     "custom-tidb",
		schemav1.LabelUserTenantKey:  tidbGroup.GetTenantId(),
		schemav1.LabelUserClusterKey: tidbGroup.GetClusterId(),
		schemav1.LabelTiDBGroupKey:   tidbGroup.GetId(),
		schemav1.LabelSharedPoolKey:  tidbGroup.GetSharedPoolId(),
	}
}

// GetCassandraProxyResourceName returns the K8s resource name for cassandra-proxy related resources (e.g., Deployment, ConfigMap, etc.).
// It does not include "proxy" in the name to avoid confusion for customers.
func GetCassandraProxyResourceName(tidbClusterId string) string {
	return "cassandra-" + tidbClusterId
}

// GetCassandraProxyLbName returns the DPM LoadBalancer / PrivateLinkService
// CR name (1:1 pair, both share this name).
func GetCassandraProxyLbName(componentId string) string {
	return "cassandra-private-" + componentId
}

// BuildCassandraProxySelector returns the joint label selector that must
// be used on Pod / Service / LB to scope to a single cluster. A single-key
// `app: cassandra-proxy` selector would cross-route Pods across clusters
// in the same SP namespace.
func BuildCassandraProxySelector(tidbClusterId string) map[string]string {
	return map[string]string{
		"app":                   "cassandra-proxy",
		"tidbcloud.com/cluster": tidbClusterId,
	}
}

// BuildCassandraProxySubIdentity is the LB.Spec.SubIdentity used for the
// PrivateLink DNS prefix. DPM combines it with the cluster gateway identity to
// produce `privatelink-cassandra.{identity}.{domain}`.
//
// Frozen contract: this string flows into the cert SAN baked at apply-group
// time, so renaming it requires a coordinated cert + LB CR change.
func BuildCassandraProxySubIdentity() string {
	return "cassandra"
}

// BuildCassandraProxyTags returns the cost tag map applied to all
// cassandra-proxy K8s resources (mirrors BuildTiDBGroupTags shape).
//
// `used-for` reuses "custom-tidb" to align with TiDBGroup / TiProxy.
// `LabelComponentKey` here writes the `pingcap.com/component` label which
// matches existing tici/pd/tikv conventions; per-component cost-explorer
// breakdown is a non-goal in this PR.
func BuildCassandraProxyTags(comp *Component, tidbCluster *TiDBCluster, sharedPool *SharedPool) map[string]string {
	return map[string]string{
		schemav1.LabelServiceTypeKey: schemav1.ClusterNGServiceType,
		schemav1.LabelUsedForKey:     "custom-tidb",
		schemav1.LabelUserTenantKey:  tidbCluster.GetTenantId(),
		schemav1.LabelUserClusterKey: tidbCluster.GetId(),
		schemav1.LabelSharedPoolKey:  sharedPool.GetId(),
		schemav1.LabelComponentKey:   "cassandra-proxy",
	}
}

// BuildCassandraProxyCertDNSNames composes the cert SAN list. The shape
// matches `pkg/cert.getTiProxyClusterServerDNSWildcard`: four in-cluster
// service-local names plus a `*.{gatewayIdentity}.{domain}` wildcard that
// covers the customer-facing PrivateLink endpoint
// `privatelink-{subIdentity}.{gatewayIdentity}.{domain}` produced by the
// dataplane PrivateLinkService controller.
//
// The wildcard is only emitted when both `gatewayIdentity` and `domain` are
// non-empty, mirroring the tiproxy helper.
func BuildCassandraProxyCertDNSNames(
	tidbCluster *TiDBCluster,
	sharedPool *SharedPool,
	dp *dpapi.DataPlane,
) []string {
	ns := GetDataPlaneNamespace(sharedPool.GetId())
	svc := GetCassandraProxyResourceName(tidbCluster.GetId())
	names := []string{
		svc,
		fmt.Sprintf("%s.%s", svc, ns),
		fmt.Sprintf("%s.%s.svc", svc, ns),
		fmt.Sprintf("%s.%s.svc.cluster.local", svc, ns),
	}
	if id, dom := tidbCluster.GetGatewayIdentity(), dp.GetDomain(); id != "" && dom != "" {
		names = append(names, fmt.Sprintf("*.%s.%s", id, dom))
	}
	return names
}

// SelectCustomerFacingIssuer picks the cert-manager issuer for
// customer-facing TLS based on dataplane configuration.
//
//	prod         -> AWSPCAClusterIssuer (cluster-issuer)
//	dev/staging  -> tidb-components-self-signed-issuer (per-namespace Issuer)
//
// Gating signal: DataPlane.Pca presence. Dev/staging dataplanes have no PCA
// config; prod always does (provisioned in CreateDataPlane action). The
// returned values mirror the issuerRef used by `pkg/cert.TiProxyServerCerts`
// so that cassandra-proxy and tiproxy share the same trust chain on each
// environment.
func SelectCustomerFacingIssuer(dp *dpapi.DataPlane) (name, kind, group string) {
	if dp.GetPca() == nil {
		return "tidb-components-self-signed-issuer", "Issuer", "cert-manager.io"
	}
	return "cluster-issuer", "AWSPCAClusterIssuer", "awspca.cert-manager.io"
}

// HasComponent reports whether the component type has RU ranges configured in
// the RU config. Internal components such as SCHEMA_MANAGER are intentionally
// absent from the RU config because they do not host user RU directly.
func (c *RuConfig) HasComponent(componentName apiv1.ComponentType) bool {
	_, ok := c.Components[strings.ToLower(componentName.String())]
	return ok
}

func (c *RuConfig) getNodeSpecRuRange(componentName apiv1.ComponentType, ru uint32) (string, *apiv1.RuConfig_RuRange, error) {
	componentCfg, ok := c.Components[strings.ToLower(componentName.String())]
	if !ok {
		return "", nil, fmt.Errorf("component %s not found", componentName.String())
	}

	for nodeSpec, ruRange := range componentCfg.GetNodeSpecRuRange() {
		maxRU := ruRange.GetMax()
		if maxRU == 0 {
			// max ru is 0 means max is unlimited
			maxRU = 0xffffffff // use a large number to represent unlimited
		}

		if ru >= ruRange.GetMin() && ru <= maxRU {
			return nodeSpec, ruRange, nil
		}
	}

	return "", nil, fmt.Errorf("no matching node for RU %d in component %s", ru, componentName.String())
}

// GetNodeSpec from component name and RU.
func (c *RuConfig) GetNodeSpec(componentName apiv1.ComponentType, ru uint32) (*apiv1.NodeSpec, error) {
	nodeSpec, _, err := c.GetNodeSpecAndDefaultNodeCount(componentName, ru)
	return nodeSpec, err
}

func (c *RuConfig) GetNodeSpecAndDefaultNodeCount(componentName apiv1.ComponentType, ru uint32) (*apiv1.NodeSpec, *uint32, error) {
	nodeSpec, ruRange, err := c.getNodeSpecRuRange(componentName, ru)
	if err != nil {
		return nil, nil, err
	}

	cpu, memoryGi, err := utils.ParseNodeSpec(nodeSpec)
	if err != nil {
		return nil, nil, fmt.Errorf("invalid node spec for RU %d in component %s: %w", ru, componentName.String(), err)
	}
	return &apiv1.NodeSpec{
		Cpu:      uint32(cpu),
		MemoryGi: uint32(memoryGi),
	}, ruRange.DefaultNodeCount, nil
}

// DefaultNodeCountQuota is the legacy fallback quota used when a component has
// no maxNodeCount configured in RuConfig.
const DefaultNodeCountQuota uint32 = 60

func (c *RuConfig) GetNodeCountQuota(componentName apiv1.ComponentType) uint32 {
	componentCfg, ok := c.Components[strings.ToLower(componentName.String())]
	if !ok {
		return DefaultNodeCountQuota
	}

	if componentCfg.GetMaxNodeCount() == 0 {
		return DefaultNodeCountQuota
	}

	return componentCfg.GetMaxNodeCount()
}

// SupportsDefaultNodeCount reports whether phase 1 allows defaultNodeCount for the component.
func SupportsDefaultNodeCount(componentName apiv1.ComponentType) bool {
	switch componentName {
	case apiv1.ComponentType_TIDB,
		apiv1.ComponentType_TIPROXY,
		apiv1.ComponentType_TIKV,
		apiv1.ComponentType_TIFLASH_COMPUTE,
		apiv1.ComponentType_TIFLASH_WRITE:
		return true
	default:
		return false
	}
}

// ValidateDefaultNodeCount validates an explicit per-RU-tier node count baseline.
func ValidateDefaultNodeCount(componentName apiv1.ComponentType, defaultNodeCount *uint32, maxNodeCount uint32) error {
	if defaultNodeCount == nil {
		return nil
	}
	if !SupportsDefaultNodeCount(componentName) {
		return fmt.Errorf("component %s does not support defaultNodeCount in phase 1", componentName.String())
	}

	minNodeCount := MinDefaultNodeCount(componentName)
	if *defaultNodeCount < minNodeCount {
		return fmt.Errorf(
			"default node count %d is less than minimum %d for component %s",
			*defaultNodeCount,
			minNodeCount,
			componentName.String(),
		)
	}

	maxDefaultNodeCount := EffectiveMaxDefaultNodeCount(componentName, maxNodeCount)
	if *defaultNodeCount > maxDefaultNodeCount {
		return fmt.Errorf(
			"default node count %d exceeds max node count %d for component %s",
			*defaultNodeCount,
			maxDefaultNodeCount,
			componentName.String(),
		)
	}

	if componentName == apiv1.ComponentType_TIKV && *defaultNodeCount%3 != 0 {
		return fmt.Errorf("default node count %d for component %s must be a multiple of 3", *defaultNodeCount, componentName.String())
	}

	return nil
}

// MinDefaultNodeCount returns the minimum allowed explicit default node count for a component.
func MinDefaultNodeCount(componentName apiv1.ComponentType) uint32 {
	switch componentName {
	case apiv1.ComponentType_TIFLASH_COMPUTE, apiv1.ComponentType_TIFLASH_WRITE:
		return 0
	case apiv1.ComponentType_TIKV:
		return 3
	default:
		return 1
	}
}

// EffectiveMaxDefaultNodeCount returns the maximum explicit default node count after component-specific quota transforms.
func EffectiveMaxDefaultNodeCount(componentName apiv1.ComponentType, maxNodeCount uint32) uint32 {
	switch componentName {
	case apiv1.ComponentType_TIKV:
		if maxNodeCount < 3 {
			return 3
		}
		return maxNodeCount - (maxNodeCount % 3)
	default:
		return maxNodeCount
	}
}

// LowercaseAlphanumericPattern is a regex pattern for validating IDs that only allow lowercase alphanumeric characters
// This pattern is specifically used for tikv_group_id which requires strict format (no hyphens or underscores)
// Note: Other IDs like shared_pool_id, cluster_id, tidb_group_id, dm_cluster_id may allow hyphens
var LowercaseAlphanumericPattern = regexp.MustCompile(`^[a-z0-9]+$`)

// ValidateLowercaseAlphanumericId validates that an ID matches the pattern ^[a-z0-9]+$
// (lowercase alphanumeric only, no hyphens, underscores, or uppercase letters)
// This is specifically used for tikv_group_id validation
func ValidateLowercaseAlphanumericId(idType, id string) error {
	if id == "" {
		return fmt.Errorf("%s cannot be empty", idType)
	}
	if !LowercaseAlphanumericPattern.MatchString(id) {
		return fmt.Errorf("%s '%s' is invalid: must match pattern ^[a-z0-9]+$ (lowercase alphanumeric only, no hyphens or underscores)", idType, id)
	}
	return nil
}

// StorageValidationError represents storage validation error with structured information
type StorageValidationError struct {
	StorageName string
	Message     string
}

func (e *StorageValidationError) Error() string {
	if e.StorageName != "" {
		return fmt.Sprintf("%s %s", e.StorageName, e.Message)
	}
	return e.Message
}

// StorageParams holds storage parameters for validation
// This abstraction allows validating different storage types (entity.Storage, InitSharedPoolConfig_Storage, etc.)
type StorageParams struct {
	Type          string
	SizeGi        uint32
	Iops          uint32
	ThroughputMbs uint32
}

// StorageValidationOptions holds additional options for storage validation
type StorageValidationOptions struct {
	// AllowZeroSize allows size to be zero (for LocalSSD where size is derived from node)
	AllowZeroSize bool
	// MinSize is the minimum required size (0 means no minimum check beyond > 0)
	MinSize uint32
}

// ValidateStorageParams validates storage parameters based on cloud provider.
// storageName is optional and used for better error messages (e.g., "DATA", "RAFTLOG").
// Returns StorageValidationError with structured error information.
//
// Note on LocalSSD validation:
// For local-ssd storage type, this function requires SizeGi/Iops/ThroughputMbs to be 0
// because these values cannot be user-specified - they are determined by the node's
// local disk hardware. After validation passes, the caller (e.g., CreateTiKVGroup) is
// responsible for populating the actual size from the node spec (fullNodeSpec.GetLocalDiskSizeGi()).
// This two-phase approach ensures:
//  1. Users cannot specify arbitrary size/iops/throughput for local-ssd (validation phase)
//  2. The storage object contains the actual disk size for downstream processing (post-validation phase)
func ValidateStorageParams(params StorageParams, provider, storageName string, opts *StorageValidationOptions) error {
	if opts == nil {
		opts = &StorageValidationOptions{}
	}

	switch strings.ToUpper(provider) {
	case apiv1.ProviderType_AWS.String():
		switch params.Type {
		case GP3:
			if !opts.AllowZeroSize && params.SizeGi == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "gp3 storage size must be greater than zero"}
			}
			if opts.MinSize > 0 && params.SizeGi < opts.MinSize {
				return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("gp3 storage size must be at least %dGi", opts.MinSize)}
			}
			if params.Iops == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "gp3 storage iops must be greater than zero"}
			}
			if params.ThroughputMbs == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "gp3 storage throughput must be greater than zero"}
			}
		case IO2:
			if !opts.AllowZeroSize && params.SizeGi == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "io2 storage size must be greater than zero"}
			}
			if opts.MinSize > 0 && params.SizeGi < opts.MinSize {
				return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("io2 storage size must be at least %dGi", opts.MinSize)}
			}
			if params.Iops == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "io2 storage iops must be greater than zero"}
			}
			if params.ThroughputMbs != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "io2 storage does not support setting throughput"}
			}
		case LocalSSD:
			// For local-ssd, user-provided size/iops/throughput must be 0.
			// The actual size will be populated by the caller after validation,
			// based on the node's local disk size (see CreateTiKVGroup for example).
			// This validation ensures users don't attempt to specify these values.
			if params.SizeGi != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting size (size is determined by node's local disk)"}
			}
			if params.Iops != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting iops"}
			}
			if params.ThroughputMbs != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting throughput"}
			}
		default:
			return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("invalid storage type '%s' for AWS", params.Type)}
		}
	case apiv1.ProviderType_ALICLOUD.String():
		switch params.Type {
		case CloudAuto:
			if !opts.AllowZeroSize && params.SizeGi == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "cloud-auto storage size must be greater than zero"}
			}
			if opts.MinSize > 0 && params.SizeGi < opts.MinSize {
				return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("cloud-auto storage size must be at least %dGi", opts.MinSize)}
			}
			if params.ThroughputMbs != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "cloud-auto storage does not support setting throughput"}
			}
		case LocalSSD:
			// Same as AWS LocalSSD - size will be populated post-validation
			if params.SizeGi != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting size (size is determined by node's local disk)"}
			}
			if params.Iops != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting iops"}
			}
			if params.ThroughputMbs != 0 {
				return &StorageValidationError{StorageName: storageName, Message: "local-ssd storage does not support setting throughput"}
			}
		default:
			return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("invalid storage type '%s' for AliCloud", params.Type)}
		}
	case apiv1.ProviderType_AZURE.String():
		switch params.Type {
		case PremiumSSDv2:
			if !opts.AllowZeroSize && params.SizeGi == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "premium-ssd-v2 storage size must be greater than zero"}
			}
			if opts.MinSize > 0 && params.SizeGi < opts.MinSize {
				return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("premium-ssd-v2 storage size must be at least %dGi", opts.MinSize)}
			}
			if params.Iops == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "premium-ssd-v2 storage iops must be greater than zero"}
			}
			if params.ThroughputMbs == 0 {
				return &StorageValidationError{StorageName: storageName, Message: "premium-ssd-v2 storage throughput must be greater than zero"}
			}
		default:
			return &StorageValidationError{StorageName: storageName, Message: fmt.Sprintf("invalid storage type '%s' for Azure", params.Type)}
		}
	default:
		return &StorageValidationError{Message: fmt.Sprintf("cloud provider '%s' is not supported", provider)}
	}

	return nil
}

// ValidateStorageTypeForProvider validates storage type and parameters based on cloud provider
// storageName is optional and used for better error messages (e.g., "DATA", "RAFTLOG")
// Returns StorageValidationError with structured error information
func ValidateStorageTypeForProvider(storage *Storage, provider, storageName string) error {
	return ValidateStorageParams(StorageParams{
		Type:          storage.GetType(),
		SizeGi:        storage.GetSizeGi(),
		Iops:          storage.GetIops(),
		ThroughputMbs: storage.GetThroughputMbs(),
	}, provider, storageName, nil)
}

// StorageParamsFromInitConfig creates StorageParams from InitSharedPoolConfig_Storage
func StorageParamsFromInitConfig(s *apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage) StorageParams {
	if s == nil {
		return StorageParams{}
	}
	return StorageParams{
		Type:          s.GetType(),
		SizeGi:        s.GetSizeGi(),
		Iops:          s.GetIops(),
		ThroughputMbs: s.GetThroughputMbps(),
	}
}

func GetEffectivePDDeploymentMode(mode apiv1.PDDeploymentMode) apiv1.PDDeploymentMode {
	if mode == apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED {
		return apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MICROSERVICE
	}
	return mode
}

func GetEffectiveCapacityMode(mode apiv1.CapacityMode) apiv1.CapacityMode {
	if mode == apiv1.CapacityMode_CAPACITY_MODE_UNSPECIFIED {
		return apiv1.CapacityMode_RCU_DRIVEN
	}
	return mode
}

const EnabledTiKVRemoteWorkersSeparator = ","

type TiKVRemoteWorkerGroups struct {
	Worker      string
	Coprocessor string
}

// ResolveTiKVRemoteWorkerGroups returns the worker references safe to render
// into a TiKVGroup. A missing or zero-capacity default worker disables the
// remoteWorkers block entirely (returns zero value).
func ResolveTiKVRemoteWorkerGroups(sharedPool *SharedPool) (TiKVRemoteWorkerGroups, error) {
	if sharedPool == nil {
		return TiKVRemoteWorkerGroups{}, nil
	}
	defaultWorker := findComponentByType(sharedPool.GetComponentGroups(), apiv1.ComponentType_TIKV_WORKER)
	if defaultWorker == nil || defaultWorker.GetNodeQuota().GetNodeCount() == 0 {
		return TiKVRemoteWorkerGroups{}, nil
	}
	groups := TiKVRemoteWorkerGroups{
		Worker:      GetComponentGroupName(defaultWorker),
		Coprocessor: GetComponentGroupName(defaultWorker),
	}
	enabled, err := NormalizeEnabledTiKVRemoteWorkers(sharedPool.GetEnabledTikvRemoteWorkers())
	if err != nil {
		return TiKVRemoteWorkerGroups{}, err
	}
	if !HasEnabledTiKVRemoteWorker(enabled, apiv1.ComponentType_COPROCESSOR_WORKER) {
		return groups, nil
	}
	coprocessor := findComponentByType(sharedPool.GetComponentGroups(), apiv1.ComponentType_COPROCESSOR_WORKER)
	if coprocessor != nil && coprocessor.GetNodeQuota().GetNodeCount() > 0 {
		groups.Coprocessor = GetComponentGroupName(coprocessor)
	}
	return groups, nil
}

func DefaultEnabledTiKVRemoteWorkers() []apiv1.ComponentType {
	return sortTiKVRemoteWorkers([]apiv1.ComponentType{
		apiv1.ComponentType_TIKV_WORKER,
		apiv1.ComponentType_COPROCESSOR_WORKER,
	})
}

func NormalizeEnabledTiKVRemoteWorkers(workers []apiv1.ComponentType) ([]apiv1.ComponentType, error) {
	if len(workers) == 0 {
		return DefaultEnabledTiKVRemoteWorkers(), nil
	}

	enabled := make(map[apiv1.ComponentType]struct{}, len(workers))
	for _, worker := range workers {
		switch worker {
		case apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER:
			enabled[worker] = struct{}{}
		default:
			return nil, fmt.Errorf(
				"enabled_tikv_remote_workers only supports TIKV_WORKER and COPROCESSOR_WORKER, got %s",
				worker.String(),
			)
		}
	}

	if _, ok := enabled[apiv1.ComponentType_TIKV_WORKER]; !ok {
		return nil, fmt.Errorf("enabled_tikv_remote_workers must include TIKV_WORKER")
	}

	normalized := make([]apiv1.ComponentType, 0, len(enabled))
	for worker := range enabled {
		normalized = append(normalized, worker)
	}
	return sortTiKVRemoteWorkers(normalized), nil
}

func EnabledTiKVRemoteWorkersToString(workers []apiv1.ComponentType) (string, error) {
	normalized, err := NormalizeEnabledTiKVRemoteWorkers(workers)
	if err != nil {
		return "", err
	}

	parts := make([]string, 0, len(normalized))
	for _, worker := range normalized {
		parts = append(parts, worker.String())
	}
	return strings.Join(parts, EnabledTiKVRemoteWorkersSeparator), nil
}

func ParseEnabledTiKVRemoteWorkers(value string, capacityMode apiv1.CapacityMode) []apiv1.ComponentType {
	if GetEffectiveCapacityMode(capacityMode) != apiv1.CapacityMode_RCU_DRIVEN {
		return []apiv1.ComponentType{}
	}
	if strings.TrimSpace(value) == "" {
		return DefaultEnabledTiKVRemoteWorkers()
	}

	parts := strings.Split(value, EnabledTiKVRemoteWorkersSeparator)
	workers := make([]apiv1.ComponentType, 0, len(parts))
	for _, part := range parts {
		name := strings.TrimSpace(part)
		if name == "" {
			continue
		}
		value, ok := apiv1.ComponentType_value[name]
		if !ok {
			continue
		}
		workers = append(workers, apiv1.ComponentType(value))
	}

	normalized, err := NormalizeEnabledTiKVRemoteWorkers(workers)
	if err != nil {
		return DefaultEnabledTiKVRemoteWorkers()
	}
	return normalized
}

func HasEnabledTiKVRemoteWorker(workers []apiv1.ComponentType, worker apiv1.ComponentType) bool {
	normalized, err := NormalizeEnabledTiKVRemoteWorkers(workers)
	if err != nil {
		normalized = workers
	}
	for _, enabledWorker := range normalized {
		if enabledWorker == worker {
			return true
		}
	}
	return false
}

func sortTiKVRemoteWorkers(workers []apiv1.ComponentType) []apiv1.ComponentType {
	normalized := append([]apiv1.ComponentType(nil), workers...)
	sort.Slice(normalized, func(i, j int) bool {
		return normalized[i].String() < normalized[j].String()
	})
	return normalized
}

func findComponentByType(components []*Component, componentType apiv1.ComponentType) *Component {
	for _, component := range components {
		if component != nil && component.GetComponentType() == componentType {
			return component
		}
	}
	return nil
}