package region

import (
	"cmp"
	"context"
	"fmt"
	"slices"
	"strconv"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"google.golang.org/grpc/codes"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const (
	defaultNodeSpecPageSize int32 = 10
	maximumNodeSpecPageSize int32 = 100
)

func (s *service) ListNodeSpecs(
	ctx context.Context,
	request *globalngv1.ListNodeSpecsRequest,
) (*globalngv1.ListNodeSpecsResponse, bizerr.BizError) {
	if err := validateListNodeSpecsRequest(request); err != nil {
		return nil, err
	}

	regionalComponent, err := regionalNodeSpecComponent(request.GetComponentType())
	if err != nil {
		return nil, err
	}

	cloudRegion, err := s.regionRepo.SelectRegion(ctx, regionrepo.Filter{
		CloudProvider: request.GetCloudProviderId(),
		RegionId:      request.GetRegionId(),
		ServicePlan:   globalcommonv1.ServicePlan_BYOC,
		TenantId:      request.GetOrganizationId(),
	})
	if err != nil {
		return nil, err
	}

	regionalResponse, err := s.nextgenCli.ListNodeSpecs(
		ctx,
		cloudRegion,
		&clusterngv1beta.ListNodeSpecsRequest{
			ComponentType: regionalComponent,
			TenantId:      strconv.FormatUint(request.GetOrganizationId(), 10),
		},
	)
	if err != nil {
		return nil, err
	}

	nodeSpecs, err := convertRegionalNodeSpecs(
		regionalResponse.GetNodeSpecOptions(),
		request.GetComponentType(),
	)
	if err != nil {
		return nil, err
	}
	nodeCountOptions, err := convertRegionalNodeCountOptions(regionalResponse.GetNodeCountOptions())
	if err != nil {
		return nil, err
	}

	slices.SortFunc(nodeSpecs, compareNodeSpecs)
	pageStart, err := nodeSpecPageStart(request, nodeSpecs)
	if err != nil {
		return nil, err
	}
	pageSize := normalizedNodeSpecPageSize(request.GetPageSize())
	pageEnd := min(pageStart+pageSize, len(nodeSpecs))
	page := slices.Clone(nodeSpecs[pageStart:pageEnd])

	var nextPageToken string
	if pageEnd < len(nodeSpecs) {
		nextPageToken, err = buildNodeSpecPageToken(request, nodeSpecKey(page[len(page)-1]))
		if err != nil {
			return nil, err
		}
	}

	return &globalngv1.ListNodeSpecsResponse{
		NodeSpecs:        page,
		NodeCountOptions: nodeCountOptions,
		TotalSize:        int32(len(nodeSpecs)), // #nosec G115 -- regional configuration cannot approach MaxInt32.
		NextPageToken:    nextPageToken,
	}, nil
}

func validateListNodeSpecsRequest(request *globalngv1.ListNodeSpecsRequest) bizerr.BizError {
	if request == nil {
		return bizerr.New(bizerr.ErrCodeParams, "request is required")
	}
	if request.GetOrganizationId() == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "organization id is required")
	}
	if request.GetCloudProviderId() == globalcommonv1.CloudProvider_NAME_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "cloud provider id is required")
	}
	if request.GetRegionId() == globalcommonv1.Region_NAME_UNSPECIFIED {
		return bizerr.New(bizerr.ErrCodeParams, "region id is required")
	}
	if request.GetPageSize() < 0 {
		return bizerr.New(bizerr.ErrCodeParams, "page size must not be negative")
	}
	_, err := regionalNodeSpecComponent(request.GetComponentType())
	return err
}

func regionalNodeSpecComponent(
	componentType globalngv1.NodeSpecComponentType,
) (clusterngv1beta.ComponentType, bizerr.BizError) {
	switch componentType {
	case globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIKV:
		return clusterngv1beta.ComponentType_TIKV, nil
	case globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIFLASH:
		return clusterngv1beta.ComponentType_TIFLASH_COMPUTE, nil
	default:
		return clusterngv1beta.ComponentType_COMPONENT_TYPE_UNSPECIFIED,
			bizerr.New(bizerr.ErrCodeParams, "unsupported node spec component type %q", componentType.String())
	}
}

func convertRegionalNodeSpecs(
	regionalOptions []*clusterngv1beta.NodeSpecOptions,
	componentType globalngv1.NodeSpecComponentType,
) ([]*globalngv1.ResourcePoolNodeSpec, bizerr.BizError) {
	result := make([]*globalngv1.ResourcePoolNodeSpec, 0, len(regionalOptions))
	seenTuples := make(map[[2]uint32]struct{}, len(regionalOptions))

	for index, regionalOption := range regionalOptions {
		regionalSpec := regionalOption.GetNodeSpec()
		if regionalSpec == nil || regionalSpec.GetCpu() == 0 || regionalSpec.GetMemoryGi() == 0 {
			return nil, invalidRegionalNodeSpec("node spec must contain positive CPU and memory")
		}

		tuple := [2]uint32{regionalSpec.GetCpu(), regionalSpec.GetMemoryGi()}
		if _, exists := seenTuples[tuple]; exists {
			return nil, invalidRegionalNodeSpec(
				fmt.Sprintf("duplicate node spec tuple %d CPU/%d GiB", tuple[0], tuple[1]),
			)
		}
		seenTuples[tuple] = struct{}{}

		storages, complete, err := convertRegionalStorageOptions(
			regionalOption.GetStorageOptions(),
			componentType,
		)
		if err != nil {
			return nil, err
		}
		if !complete {
			continue
		}

		result = append(result, &globalngv1.ResourcePoolNodeSpec{
			NodeSpecKey:   fmt.Sprintf("%dC%dG", regionalSpec.GetCpu(), regionalSpec.GetMemoryGi()),
			DisplayName:   fmt.Sprintf("%d vCPU, %d GiB", regionalSpec.GetCpu(), regionalSpec.GetMemoryGi()),
			ComponentType: componentType,
			Cpu:           int64(regionalSpec.GetCpu()),
			MemoryGi:      int64(regionalSpec.GetMemoryGi()),
			IsDefault:     index == 0,
			Storages:      storages,
		})
	}

	return result, nil
}

func convertRegionalStorageOptions(
	regionalOptions []*clusterngv1beta.StorageSizeOptions,
	componentType globalngv1.NodeSpecComponentType,
) ([]*globalngv1.NodeSpecStorage, bool, bizerr.BizError) {
	requiredNames := requiredRegionalStorageNames(componentType)
	publicNames := publicStorageNames(componentType)
	byName := make(map[string]*globalngv1.NodeSpecStorage, len(requiredNames))
	for _, regionalOption := range regionalOptions {
		if regionalOption == nil {
			return nil, false, invalidRegionalNodeSpec("storage option is missing")
		}
		if regionalOption.GetMedium() != clusterngv1beta.StorageMedium_STORAGE_MEDIUM_CLOUD {
			continue
		}

		name, recognized := publicStorageName(regionalOption.GetName())
		if !recognized || !slices.Contains(requiredNames, name) {
			continue
		}
		if _, exists := byName[name]; exists {
			return nil, false, invalidRegionalNodeSpec(
				fmt.Sprintf("duplicate cloud storage option %q", name),
			)
		}
		minimum := regionalOption.GetMinSizeGi()
		defaultSize := regionalOption.GetDefaultSizeGi()
		maximum := regionalOption.GetMaxSizeGi()
		if minimum == 0 || defaultSize < minimum || maximum < defaultSize {
			return nil, false, invalidRegionalNodeSpec(
				fmt.Sprintf("cloud storage option %q has inconsistent size bounds", name),
			)
		}
		byName[name] = &globalngv1.NodeSpecStorage{
			Name:          name,
			DefaultSizeGi: int64(defaultSize),
			MinSizeGi:     int64(minimum),
			MaxSizeGi:     int64(maximum),
		}
	}

	for _, name := range requiredNames {
		if byName[name] == nil {
			return []*globalngv1.NodeSpecStorage{}, false, nil
		}
	}

	result := make([]*globalngv1.NodeSpecStorage, 0, len(publicNames))
	for _, name := range publicNames {
		result = append(result, byName[name])
	}
	return result, true, nil
}

func requiredRegionalStorageNames(
	componentType globalngv1.NodeSpecComponentType,
) []string {
	switch componentType {
	case globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIKV:
		return []string{"data", "raftlog"}
	case globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIFLASH:
		return []string{"data"}
	default:
		return []string{}
	}
}

func publicStorageNames(
	componentType globalngv1.NodeSpecComponentType,
) []string {
	switch componentType {
	case globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIKV,
		globalngv1.NodeSpecComponentType_NODE_SPEC_COMPONENT_TYPE_TIFLASH:
		return []string{"data"}
	default:
		return []string{}
	}
}

func publicStorageName(name clusterngv1beta.Component_Storage_Name) (string, bool) {
	switch name {
	case clusterngv1beta.Component_Storage_DATA:
		return "data", true
	case clusterngv1beta.Component_Storage_RAFTLOG:
		return "raftlog", true
	default:
		return "", false
	}
}

func convertRegionalNodeCountOptions(
	regionalOptions *clusterngv1beta.NodeCountOptions,
) (*globalngv1.NodeCountOptions, bizerr.BizError) {
	if regionalOptions == nil || regionalOptions.GetRule() == nil {
		return nil, invalidRegionalNodeSpec("node count options and rule are required")
	}

	ruleType, err := convertRegionalNodeCountRuleType(regionalOptions.GetRule().GetType())
	if err != nil {
		return nil, err
	}
	ruleValue := regionalOptions.GetRule().GetValue()
	if (ruleType == globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_MULTIPLE_OF ||
		ruleType == globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_FIXED) && ruleValue == 0 {
		return nil, invalidRegionalNodeSpec("node count rule value must be positive")
	}
	minimum := regionalOptions.GetMinNodeCount()
	maximum := regionalOptions.GetMaxNodeCount()
	defaultCount := regionalOptions.GetDefaultNodeCount()
	if minimum == 0 || maximum < minimum || defaultCount < minimum || defaultCount > maximum {
		return nil, invalidRegionalNodeSpec("node count options are inconsistent")
	}

	return &globalngv1.NodeCountOptions{
		DefaultNodeCount: int64(defaultCount),
		MinNodeCount:     int64(minimum),
		MaxNodeCount:     int64(maximum),
		Rule: &globalngv1.NodeCountRule{
			Type:  ruleType,
			Value: int64(ruleValue),
		},
	}, nil
}

func convertRegionalNodeCountRuleType(
	ruleType clusterngv1beta.NodeCountRuleType,
) (globalngv1.NodeCountRuleType, bizerr.BizError) {
	switch ruleType {
	case clusterngv1beta.NodeCountRuleType_NODE_COUNT_RULE_TYPE_NONE:
		return globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_NONE, nil
	case clusterngv1beta.NodeCountRuleType_NODE_COUNT_RULE_TYPE_MULTIPLE_OF:
		return globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_MULTIPLE_OF, nil
	case clusterngv1beta.NodeCountRuleType_NODE_COUNT_RULE_TYPE_FIXED:
		return globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_FIXED, nil
	default:
		return globalngv1.NodeCountRuleType_NODE_COUNT_RULE_TYPE_UNSPECIFIED,
			invalidRegionalNodeSpec(fmt.Sprintf("unsupported node count rule %q", ruleType.String()))
	}
}

func normalizedNodeSpecPageSize(pageSize int32) int {
	switch {
	case pageSize == 0:
		return int(defaultNodeSpecPageSize)
	case pageSize > maximumNodeSpecPageSize:
		return int(maximumNodeSpecPageSize)
	default:
		return int(pageSize)
	}
}

func nodeSpecPageStart(
	request *globalngv1.ListNodeSpecsRequest,
	nodeSpecs []*globalngv1.ResourcePoolNodeSpec,
) (int, bizerr.BizError) {
	lastKey, err := parseNodeSpecPageToken(request)
	if err != nil || lastKey == nil {
		return 0, err
	}

	index := slices.IndexFunc(nodeSpecs, func(nodeSpec *globalngv1.ResourcePoolNodeSpec) bool {
		return nodeSpecKey(nodeSpec) == *lastKey
	})
	if index < 0 {
		return 0, invalidNodeSpecPageToken()
	}
	return index + 1, nil
}

func compareNodeSpecs(left, right *globalngv1.ResourcePoolNodeSpec) int {
	if result := cmp.Compare(left.GetCpu(), right.GetCpu()); result != 0 {
		return result
	}
	if result := cmp.Compare(left.GetMemoryGi(), right.GetMemoryGi()); result != 0 {
		return result
	}
	return cmp.Compare(left.GetNodeSpecKey(), right.GetNodeSpecKey())
}

func nodeSpecKey(nodeSpec *globalngv1.ResourcePoolNodeSpec) nodeSpecSortKey {
	return nodeSpecSortKey{
		CPU:         nodeSpec.GetCpu(),
		MemoryGi:    nodeSpec.GetMemoryGi(),
		NodeSpecKey: nodeSpec.GetNodeSpecKey(),
	}
}

func invalidRegionalNodeSpec(message string) bizerr.BizError {
	return bizerr.New(
		bizerr.ErrCode(codes.Unavailable),
		"invalid regional node spec response: %s",
		message,
	)
}