package resourcepool

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"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/utilities/grpc/metadata"
)

const (
	importExportInProgressCheckCode = "IMPORT_EXPORT_IN_PROGRESS"
	changefeedInProgressCheckCode   = "CHANGEFEED_IN_PROGRESS"
)

func (s serverImpl) CreateResourcePool(ctx context.Context,
	request *globalngv1.CreateResourcePoolRequest) (*globalngv1.ResourcePoolProfile, error) {
	return s.nextgenCli.CreateResourcePool(ctx, request)
}

func (s serverImpl) ListResourcePools(ctx context.Context,
	request *globalngv1.ListResourcePoolsRequest) (*globalngv1.ListResourcePoolsResponse, error) {
	return s.nextgenCli.ListResourcePools(ctx, request)
}

func (s serverImpl) UpdateResourcePoolName(ctx context.Context,
	request *globalngv1.UpdateResourcePoolNameRequest) (*globalngv1.ResourcePool, error) {
	return s.nextgenCli.UpdateResourcePoolName(ctx, request)
}

func (s serverImpl) GetResourcePool(ctx context.Context,
	request *globalngv1.GetResourcePoolRequest) (*globalngv1.ResourcePoolProfile, error) {
	response, err := s.nextgenCli.GetResourcePool(ctx, request)
	if err != nil {
		return nil, err
	}

	// collect all instance ids
	instanceIds := make([]uint64, 0, len(response.GetInstances()))
	for _, instance := range response.GetInstances() {
		instanceIds = append(instanceIds, instance.ClusterId)
	}

	// fast return if no instances
	if len(instanceIds) == 0 {
		return response, nil
	}

	// fill in instance display name
	instanceNameMap, err := s.getInstanceDisplayNamesByIds(ctx, instanceIds)
	if err != nil {
		return nil, err
	}
	for _, instance := range response.GetInstances() {
		instance.Name = instanceNameMap[instance.ClusterId]
	}

	return response, nil
}

func (s serverImpl) DeleteResourcePool(ctx context.Context,
	request *globalngv1.DeleteResourcePoolRequest) (*globalngv1.ResourcePool, error) {
	return s.nextgenCli.DeleteResourcePool(ctx, request)
}

func (s serverImpl) GetResourcePoolPauseResumeOperation(
	ctx context.Context,
	request *globalngv1.GetResourcePoolPauseResumeOperationRequest,
) (*globalngv1.ResourcePoolPauseResumeOperation, error) {
	return s.nextgenCli.GetResourcePoolPauseResumeOperation(ctx, request)
}

func (s serverImpl) PauseResourcePool(
	ctx context.Context,
	request *globalngv1.PauseResourcePoolRequest,
) (*globalngv1.PauseResourcePoolResponse, error) {
	if request == nil {
		return nil, fmt.Errorf("pause resource pool request should not be nil")
	}
	ctx = appendResourcePoolRequestMetadata(ctx, request.GetResourcePoolId())
	if request.GetValidateOnly() {
		clusterIDs, err := s.pauseClusterIDs(ctx, request.GetResourcePoolId())
		if err != nil {
			return nil, err
		}
		customStatusResults, err := s.validatePauseCustomStatuses(ctx, clusterIDs, false)
		if err != nil {
			return nil, err
		}
		response, err := s.nextgenCli.PauseResourcePool(ctx, request)
		if err != nil {
			return nil, err
		}
		if response == nil {
			return nil, fmt.Errorf("pause resource pool response should not be nil")
		}
		response.ValidationResults = append(response.GetValidationResults(), customStatusResults...)
		return response, nil
	}

	begin, err := s.nextgenCli.BeginPauseResourcePoolAdmission(ctx, &globalngv1.BeginPauseResourcePoolAdmissionRequest{
		ResourcePoolId:                request.GetResourcePoolId(),
		OperationId:                   request.GetOperationId(),
		CreateManualBackupBeforePause: request.GetCreateManualBackupBeforePause(),
	})
	if err != nil {
		return nil, err
	}
	if begin == nil {
		return nil, fmt.Errorf("begin pause resource pool admission response should not be nil")
	}
	if len(begin.GetValidationResults()) > 0 {
		return &globalngv1.PauseResourcePoolResponse{ValidationResults: begin.GetValidationResults()}, nil
	}
	// Load instances after NG has entered PAUSE_PRECHECKING. Instance creation is
	// rejected in that state, so this is the stable set that must be locked.
	clusterIDs, err := s.pauseClusterIDs(ctx, request.GetResourcePoolId())
	if err != nil {
		return nil, errors.Join(err, s.abortPauseAdmission(ctx, request))
	}

	customStatusResults, err := s.validatePauseCustomStatuses(ctx, clusterIDs, true)
	if err != nil {
		return nil, errors.Join(err, s.abortPauseAdmission(ctx, request))
	}
	if len(customStatusResults) > 0 {
		_, resolveErr := s.nextgenCli.ResolvePauseResourcePoolAdmission(ctx, &globalngv1.ResolvePauseResourcePoolAdmissionRequest{
			ResourcePoolId:    request.GetResourcePoolId(),
			OperationId:       request.GetOperationId(),
			Proceed:           false,
			ValidationResults: customStatusResults,
		})
		if resolveErr != nil {
			return nil, fmt.Errorf("abort pause resource pool admission: %w", resolveErr)
		}
		return &globalngv1.PauseResourcePoolResponse{ValidationResults: customStatusResults}, nil
	}

	resolved, err := s.nextgenCli.ResolvePauseResourcePoolAdmission(ctx, &globalngv1.ResolvePauseResourcePoolAdmissionRequest{
		ResourcePoolId: request.GetResourcePoolId(),
		OperationId:    request.GetOperationId(),
		Proceed:        true,
	})
	if err != nil {
		return nil, errors.Join(err, s.abortPauseAdmission(ctx, request))
	}
	if resolved == nil {
		return nil, errors.Join(
			fmt.Errorf("resolve pause resource pool admission response should not be nil"),
			s.abortPauseAdmission(ctx, request),
		)
	}
	return &globalngv1.PauseResourcePoolResponse{
		OperationId:       resolved.GetOperationId(),
		ValidationResults: resolved.GetValidationResults(),
	}, nil
}

func (s serverImpl) abortPauseAdmission(
	ctx context.Context,
	request *globalngv1.PauseResourcePoolRequest,
) error {
	compensationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer cancel()
	_, err := s.nextgenCli.ResolvePauseResourcePoolAdmission(compensationCtx, &globalngv1.ResolvePauseResourcePoolAdmissionRequest{
		ResourcePoolId: request.GetResourcePoolId(),
		OperationId:    request.GetOperationId(),
		Proceed:        false,
	})
	if err != nil {
		return fmt.Errorf(
			"abort pause admission for resource pool %d operation %s: %w",
			request.GetResourcePoolId(),
			request.GetOperationId(),
			err,
		)
	}
	return nil
}

func (s serverImpl) pauseClusterIDs(
	ctx context.Context,
	resourcePoolID uint64,
) ([]uint64, error) {
	profile, err := s.nextgenCli.GetResourcePool(ctx, &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: resourcePoolID,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get resource pool for custom status validation: %w", err)
	}
	if profile == nil {
		return nil, fmt.Errorf("resource pool profile should not be nil")
	}

	clusterIDs := make([]uint64, 0, len(profile.GetInstances()))
	for _, instance := range profile.GetInstances() {
		if instance == nil || instance.GetClusterId() == 0 {
			continue
		}
		clusterIDs = append(clusterIDs, instance.GetClusterId())
	}
	if len(clusterIDs) == 0 {
		return clusterIDs, nil
	}
	return clusterIDs, nil
}

func (s serverImpl) validatePauseCustomStatuses(
	ctx context.Context,
	clusterIDs []uint64,
	lockClusters bool,
) ([]*globalngv1.ResourcePoolPauseValidationResult, error) {
	if len(clusterIDs) == 0 {
		return nil, nil
	}
	var statusesByCluster map[uint64][]*entity.ClusterCustomStatus
	query := func(ctx context.Context) error {
		if lockClusters {
			if err := s.customStatusRepo.LockClustersForUpdate(ctx, clusterIDs); err != nil {
				return fmt.Errorf("failed to lock resource pool clusters for pause: %w", err)
			}
		}
		var err error
		statusesByCluster, err = s.customStatusRepo.ListClusterCustomStatus(
			ctx,
			clusterIDs,
			[]tidbmgmtservicev1.CustomStatusType{
				tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT,
				tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
			},
			false,
		)
		if err != nil {
			return fmt.Errorf("failed to list cluster custom statuses for resource pool pause: %w", err)
		}
		return nil
	}
	if lockClusters {
		if err := s.transaction.Transaction(ctx, query); err != nil {
			return nil, err
		}
	} else if err := query(ctx); err != nil {
		return nil, err
	}

	var importRunning, changefeedRunning bool
	for _, statuses := range statusesByCluster {
		for _, status := range statuses {
			if status == nil || !status.IsActionMutexed(tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER) {
				continue
			}
			switch status.StatusType {
			case tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT:
				importRunning = true
			case tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC:
				changefeedRunning = true
			}
		}
	}

	results := make([]*globalngv1.ResourcePoolPauseValidationResult, 0, 2)
	if importRunning {
		results = append(results, &globalngv1.ResourcePoolPauseValidationResult{
			CheckCode: importExportInProgressCheckCode,
			Detail:    "resource pool has a running import",
		})
	}
	if changefeedRunning {
		results = append(results, &globalngv1.ResourcePoolPauseValidationResult{
			CheckCode: changefeedInProgressCheckCode,
			Detail:    "resource pool has a running changefeed",
		})
	}
	return results, nil
}

func (s serverImpl) ResumeResourcePool(
	ctx context.Context,
	request *globalngv1.ResumeResourcePoolRequest,
) (*globalngv1.ResumeResourcePoolResponse, error) {
	ctx = appendResourcePoolRequestMetadata(ctx, request.GetResourcePoolId())
	return s.nextgenCli.ResumeResourcePool(ctx, request)
}

func appendResourcePoolRequestMetadata(ctx context.Context, resourcePoolID uint64) context.Context {
	if resourcePoolID == 0 {
		return ctx
	}
	requestMeta := &metadata.RequestMeta{
		ResourceID: strconv.FormatUint(resourcePoolID, 10),
	}
	ctx = metadata.AppendToIncomingMeta(ctx, requestMeta)
	return metadata.AppendToOutgoingMeta(ctx, requestMeta)
}

func (s serverImpl) ListOrgResourcePools(ctx context.Context,
	request *tidbmgmtservicev1.ListOrgResourcePoolsRequest) (*globalngv1.ListOwnerResourcePoolsResponse, error) {
	var providers []globalcommonv1.CloudProvider_Name
	if request.GetProvider() != globalcommonv1.CloudProvider_NAME_UNSPECIFIED {
		providers = []globalcommonv1.CloudProvider_Name{
			request.GetProvider(),
		}
	}

	var regions []globalcommonv1.Region_Name
	if request.GetRegion() != globalcommonv1.Region_NAME_UNSPECIFIED {
		regions = []globalcommonv1.Region_Name{
			request.GetRegion(),
		}
	}
	nextgenCliReq := &globalngv1.ListOwnerResourcePoolsRequest{
		OwnerId:     request.OrganizationId,
		OwnerType:   globalngv1.ResourcePool_ORGANIZATION,
		ProviderIds: providers,
		RegionIds:   regions,
		ServicePlan: request.ServicePlan,
	}

	response, err := s.nextgenCli.ListOwnerResourcePools(ctx, nextgenCliReq)
	if err != nil {
		return nil, err
	}

	// collect all instance ids
	instanceIds := make([]uint64, 0)
	for _, r := range response.RegionResourcePools {
		for _, p := range r.ResourcePools {
			for _, instance := range p.Instances {
				instanceIds = append(instanceIds, instance.ClusterId)
			}
		}
	}

	// fast return if no instances
	if len(instanceIds) == 0 {
		return response, nil
	}

	// fill in instance display name
	instanceNameMap, err := s.getInstanceDisplayNamesByIds(ctx, instanceIds)
	if err != nil {
		return nil, err
	}
	for _, r := range response.RegionResourcePools {
		for _, p := range r.ResourcePools {
			for _, instance := range p.Instances {
				instance.Name = instanceNameMap[instance.ClusterId]
			}
		}
	}

	return response, nil
}

func (s serverImpl) UpdateOrgDefaultResourcePool(ctx context.Context,
	request *tidbmgmtservicev1.UpdateOrgDefaultResourcePoolRequest) (*globalngv1.ResourcePoolStrategy, error) {
	return s.nextgenCli.UpdateOrgDefaultResourcePool(ctx, &globalngv1.UpdateOrgDefaultResourcePoolRequest{
		Strategy: &globalngv1.ResourcePoolStrategy{
			OrganizationId:        request.OrganizationId,
			DefaultResourcePoolId: request.DefaultResourcePoolId,
			ProviderId:            request.ProviderId,
			RegionId:              request.RegionId,
		},
	})
}

func (s serverImpl) UpdateResourcePoolInstanceLimit(ctx context.Context,
	request *globalngv1.UpdateResourcePoolInstanceLimitRequest) (*globalngv1.ResourcePool, error) {
	return s.nextgenCli.UpdateResourcePoolInstanceLimit(ctx, request)
}

func (s serverImpl) getInstanceDisplayNamesByIds(ctx context.Context, instanceIds []uint64) (map[uint64]string, error) {
	if len(instanceIds) == 0 {
		return map[uint64]string{}, nil
	}

	clusters, err := s.clusterRepo.QueryClusterByFilterAndPage(ctx, &repository.ClusterFilter{
		ClusterIds: instanceIds,
	}, 0, 0)

	if err != nil {
		return nil, err
	}

	displayNameMap := make(map[uint64]string, len(clusters))
	for _, cluster := range clusters {
		displayNameMap[cluster.ClusterId] = cluster.DisplayName
	}

	return displayNameMap, nil
}