package tioperator

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func ListClusters(ctx context.Context, remoteCli client.Client) ([]*entity.SharedPool, error) {
	nsLister := &corev1.NamespaceList{}
	// list namespces
	err := remoteCli.List(ctx, nsLister, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			schemav1.LabelUsedForKey: schemav1.ClusterNGServiceType,
		}),
	})
	if err != nil {
		return nil, err
	}

	var clusters []*entity.SharedPool
	for i := range nsLister.Items {
		if !strings.HasPrefix(nsLister.Items[i].Name, namespacePrefix) {
			tilog.Warn(ctx, "Namespace  does not match the tidb prefix", zap.String("namespace", nsLister.Items[i].Name))
			continue
		}

		cls, err := GetCluster(ctx, remoteCli, strings.TrimPrefix(nsLister.Items[i].Name, namespacePrefix))
		if err != nil {
			return nil, err
		}
		if cls != nil {
			clusters = append(clusters, cls)
		}
	}
	return clusters, nil
}

func GetCluster(ctx context.Context, remoteCli client.Client, clusterId string) (*entity.SharedPool, error) {
	sharedPool := &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:       clusterId,
			Provider: nil,
		},
	}

	sharedPool.Phase = getClusterStatus(ctx, remoteCli, clusterId)

	// convert components
	var err error
	sharedPool.ComponentGroups, err = getComponents(ctx, remoteCli, sharedPool.GetDataPlaneNamespace())
	if err != nil {
		return nil, err
	}

	// use pd version as cluster version
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1.ComponentType_PD {
			// set cluster version
			sharedPool.Version = comp.GetVersion()
			break
		}
	}

	if sharedPool.Phase == apiv1.SharedPool_PAUSED {
		tilog.Warn(ctx, "dataplane cluster is paused", zap.String("sharedPoolId", sharedPool.GetId()))
		return sharedPool, nil
	}

	// operator v2 marks ClusterCondAvailable=False when no TiDBGroup is available.
	// For clusters without TiDB component, we should continue evaluating
	// component statuses instead of returning unavailable directly.
	if sharedPool.Phase == apiv1.SharedPool_UNAVAILABLE {
		if shouldRelaxClusterAvailableCondition(sharedPool.GetComponentGroups()) {
			tilog.Info(ctx, "cluster condition is unavailable but base topology shape detected, continue component-based readiness evaluation",
				zap.String("sharedPoolId", sharedPool.GetId()))
			sharedPool.Phase = apiv1.SharedPool_AVAILABLE
		} else {
			tilog.Warn(ctx, "dataplane cluster is unavailable", zap.String("sharedPoolId", sharedPool.GetId()))
			return sharedPool, nil
		}
	}

	if len(sharedPool.GetComponentGroups()) == 0 {
		tilog.Warn(ctx, "dataplane cluster has no components", zap.String("sharedPoolId", sharedPool.GetId()))
		sharedPool.Phase = apiv1.SharedPool_UNAVAILABLE
		return sharedPool, nil
	}

	relaxClusterAvailable := shouldRelaxClusterAvailableCondition(sharedPool.GetComponentGroups())

	// check component status
	// if any component is modifying, we should set the cluster status to modifying
	// if any component is unavailable, we should set the cluster status to unavailable, without tidb/tidb-worker
	for _, comp := range sharedPool.GetComponentGroups() {
		if componentHasZeroNodeCount(comp) {
			// Components explicitly configured with 0 replicas should not affect shared pool readiness.
			continue
		}

		if relaxClusterAvailable && comp.GetComponentType() == apiv1.ComponentType_TIPROXY {
			// base topology shape readiness is driven by PD/TiKV;
			// tiproxy may come from failed/in-progress tenant creation and should not downgrade shared pool.
			continue
		}

		if comp.GetPhase() == apiv1.Component_UNAVAILABLE &&
			comp.GetComponentType() != apiv1.ComponentType_TIDB &&
			comp.GetComponentType() != apiv1.ComponentType_TIDB_WORKER &&
			comp.GetComponentType() != apiv1.ComponentType_TIFLASH_LEARNER &&
			comp.GetComponentType() != apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER {
			tilog.Warn(ctx, "dataplane component is unavailable",
				zap.String("id", sharedPool.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)),
				zap.String("componentGroup", entity.GetComponentGroupName(comp)))
			// if any component is unavailable, we should set the cluster status to unavailable
			sharedPool.Phase = apiv1.SharedPool_UNAVAILABLE
			break
		}

		if comp.GetPhase() == apiv1.Component_MODIFYING || comp.GetPhase() == apiv1.Component_CREATING {
			tilog.Warn(ctx, "dataplane component is modifying or creating",
				zap.String("id", sharedPool.GetId()),
				zap.String("componentType", entity.GetComponentTypeName(comp)),
				zap.String("componentGroup", entity.GetComponentGroupName(comp)))
			sharedPool.Phase = apiv1.SharedPool_MODIFYING
			break
		}
	}

	// For clusters without TiDB component, skip backup components readiness gate.
	// Backup components may not be created/ready before the first TiDB group exists.
	if relaxClusterAvailable {
		tilog.Info(ctx, "skip backup components readiness check for base topology shape", zap.String("id", sharedPool.GetId()))
		return sharedPool, nil
	}

	// if backup component is not ready, the shared pool is not ready
	backupComponentsStatus, err := GetBackupComponentsStatus(ctx, remoteCli, sharedPool.GetDataPlaneNamespace())
	if err != nil {
		return nil, fmt.Errorf("failed to get backup components status: %w", err)
	}
	if backupComponentsStatus == apiv1.Component_UNAVAILABLE {
		tilog.Warn(ctx, "backup components are not available",
			zap.String("id", sharedPool.GetId()),
			zap.String("namespace", sharedPool.GetDataPlaneNamespace()),
			zap.String("phase_before_backup_check", sharedPool.GetPhase().String()))
		sharedPool.Phase = apiv1.SharedPool_UNAVAILABLE
		return sharedPool, nil
	}

	return sharedPool, nil
}

func UpdateClusterUpgradePolicy(
	ctx context.Context,
	remoteCli client.Client,
	clusterId string,
	policy v1alpha1.UpgradePolicy,
) error {
	cluster := &v1alpha1.Cluster{}
	key := types.NamespacedName{
		Name:      clusterCRName,
		Namespace: dataplaneNamespace(clusterId),
	}
	if err := remoteCli.Get(ctx, key, cluster); err != nil {
		return fmt.Errorf("failed to get cluster %s: %w", clusterId, err)
	}
	if cluster.Spec.UpgradePolicy == policy {
		return nil
	}

	patch := client.MergeFrom(cluster.DeepCopy())
	cluster.Spec.UpgradePolicy = policy
	if err := remoteCli.Patch(ctx, cluster, patch); err != nil {
		return fmt.Errorf("failed to update cluster %s upgrade policy: %w", clusterId, err)
	}
	return nil
}

func IsClusterUpgradePolicyObserved(
	ctx context.Context,
	remoteCli client.Client,
	clusterId string,
	policy v1alpha1.UpgradePolicy,
) (bool, error) {
	cluster := &v1alpha1.Cluster{}
	key := types.NamespacedName{
		Name:      clusterCRName,
		Namespace: dataplaneNamespace(clusterId),
	}
	if err := remoteCli.Get(ctx, key, cluster); err != nil {
		return false, fmt.Errorf("failed to get cluster %s: %w", clusterId, err)
	}

	policyMatches := cluster.Spec.UpgradePolicy == policy
	generationObserved := cluster.Status.ObservedGeneration == cluster.Generation
	return policyMatches && generationObserved, nil
}

func SetClusterSuspendCompute(ctx context.Context, remoteCli client.Client, clusterID string, desired bool) error {
	cluster := &v1alpha1.Cluster{}
	key := types.NamespacedName{Name: clusterCRName, Namespace: dataplaneNamespace(clusterID)}
	if err := remoteCli.Get(ctx, key, cluster); err != nil {
		return fmt.Errorf("get cluster %s: %w", clusterID, err)
	}
	if cluster.Spec.SuspendAction != nil && cluster.Spec.SuspendAction.SuspendCompute == desired {
		return nil
	}
	patch := client.MergeFrom(cluster.DeepCopy())
	if cluster.Spec.SuspendAction == nil {
		cluster.Spec.SuspendAction = &v1alpha1.SuspendAction{}
	}
	cluster.Spec.SuspendAction.SuspendCompute = desired
	if err := remoteCli.Patch(ctx, cluster, patch); err != nil {
		return fmt.Errorf("patch cluster %s suspendCompute=%t: %w", clusterID, desired, err)
	}
	return nil
}

// ClusterSuspendComputeStatus reports the Operator spec and whether that spec
// has converged in the current Cluster generation.
type ClusterSuspendComputeStatus struct {
	SpecDesired bool
	Observed    bool
}

// InspectClusterSuspendCompute reads the Operator suspend spec and its observed
// state from one Cluster snapshot.
func InspectClusterSuspendCompute(ctx context.Context, remoteCli client.Client, clusterID string) (ClusterSuspendComputeStatus, error) {
	cluster := &v1alpha1.Cluster{}
	key := types.NamespacedName{Name: clusterCRName, Namespace: dataplaneNamespace(clusterID)}
	if err := remoteCli.Get(ctx, key, cluster); err != nil {
		return ClusterSuspendComputeStatus{}, fmt.Errorf("get cluster %s: %w", clusterID, err)
	}
	suspendCompute := cluster.Spec.SuspendAction != nil && cluster.Spec.SuspendAction.SuspendCompute
	status := ClusterSuspendComputeStatus{SpecDesired: suspendCompute}
	if cluster.Status.ObservedGeneration != cluster.Generation {
		return status, nil
	}
	conditions := map[string]metav1.Condition{}
	for _, condition := range cluster.Status.Conditions {
		conditions[condition.Type] = condition
	}
	conditionMatches := func(conditionType string, status metav1.ConditionStatus) bool {
		condition, ok := conditions[conditionType]
		return ok && condition.Status == status && condition.ObservedGeneration == cluster.Generation
	}
	if suspendCompute {
		status.Observed = conditionMatches(v1alpha1.ClusterCondSuspended, metav1.ConditionTrue)
		return status, nil
	}
	// Ready and Synced are Cluster-level aggregates for all observed Groups.
	// Do not gate resume on Available: it can remain false for PD/TiKV-only pools.
	status.Observed = conditionMatches(v1alpha1.ClusterCondSuspended, metav1.ConditionFalse) &&
		conditionMatches(v1alpha1.CondReady, metav1.ConditionTrue) &&
		conditionMatches(v1alpha1.CondSynced, metav1.ConditionTrue)
	return status, nil
}

func IsClusterSuspendComputeObserved(ctx context.Context, remoteCli client.Client, clusterID string, desired bool) (bool, error) {
	status, err := InspectClusterSuspendCompute(ctx, remoteCli, clusterID)
	if err != nil {
		return false, err
	}
	return status.SpecDesired == desired && status.Observed, nil
}

func shouldRelaxClusterAvailableCondition(components []*entity.Component) bool {
	hasPD := false
	hasTiKV := false

	for _, comp := range components {
		if componentHasZeroNodeCount(comp) {
			continue
		}

		switch comp.GetComponentType() {
		case apiv1.ComponentType_PD:
			hasPD = true
		case apiv1.ComponentType_TIKV:
			hasTiKV = true
		case apiv1.ComponentType_TIDB, apiv1.ComponentType_TIDB_WORKER:
			return false
		case apiv1.ComponentType_TIKV_WORKER, apiv1.ComponentType_COPROCESSOR_WORKER:
			// allowed
		case apiv1.ComponentType_TIPROXY:
			// allowed
		default:
			// any other component type means this is not the base topology shape
			return false
		}
	}

	return hasPD && hasTiKV
}

func componentHasZeroNodeCount(comp *entity.Component) bool {
	if comp == nil || comp.GetNodeQuota() == nil {
		return false
	}
	return comp.GetNodeQuota().GetNodeCount() == 0
}

func getClusterStatus(ctx context.Context, remoteCli client.Client, id string) apiv1.SharedPool_Phase {
	// operator v2 cluster
	var clusterCr v1alpha1.Cluster
	key := types.NamespacedName{Name: clusterCRName, Namespace: dataplaneNamespace(id)}

	if err := remoteCli.Get(ctx, key, &clusterCr); err != nil {
		tilog.Warn(ctx, "failed to get sharedPool from dataplane", zap.String("id", id), zap.Error(err))
		return apiv1.SharedPool_UNAVAILABLE
	}

	if clusterCr.GetGeneration() != clusterCr.Status.ObservedGeneration {
		tilog.Warn(ctx, "cluster generation is not match", zap.String("id", id), zap.Any("generation", clusterCr.GetGeneration()), zap.Any("observed generation", clusterCr.Status.ObservedGeneration))
		return apiv1.SharedPool_UNAVAILABLE
	}

	clusterStatus := apiv1.SharedPool_AVAILABLE
	for _, condition := range clusterCr.Status.Conditions {
		// check if cluster is available
		switch condition.Type {
		case v1alpha1.ClusterCondAvailable:
			if condition.Status != metav1.ConditionTrue {
				tilog.Warn(ctx, "cluster is not available", zap.String("id", id), zap.Any("condition", condition))
				clusterStatus = apiv1.SharedPool_UNAVAILABLE
			}
		case v1alpha1.ClusterCondSuspended:
			if condition.Status != metav1.ConditionFalse {
				tilog.Warn(ctx, "cluster is suspended", zap.String("id", id), zap.Any("condition", condition))
				clusterStatus = apiv1.SharedPool_PAUSED
			}
		default:
			// do nothing
			// only log the condition
			tilog.Info(ctx, "unknown cluster condition", zap.String("id", id), zap.Any("condition", condition))
		}

		if clusterStatus == apiv1.SharedPool_PAUSED {
			break
		}
	}

	// if cluster set paused
	if clusterCr.Spec.Paused && clusterStatus != apiv1.SharedPool_PAUSED {
		clusterStatus = apiv1.SharedPool_PAUSING
	}

	// overlay paused status
	if clusterStatus == apiv1.SharedPool_PAUSING || clusterStatus == apiv1.SharedPool_PAUSED {
		tilog.Warn(ctx, "cluster is suspended, return unavailable status", zap.String("id", id), zap.String("originalStatus", clusterStatus.String()))
		return apiv1.SharedPool_UNAVAILABLE
	}

	return clusterStatus
}

func getComponents(ctx context.Context, remoteClient client.Client, dataplaneNs string) ([]*entity.Component, error) {
	clusterComponents := make([]*entity.Component, 0)

	// pd
	pdComps, err := listPDGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get pd groups: %w", err)
	}
	clusterComponents = append(clusterComponents, pdComps...)

	// tso
	pdTsoComps, err := listPDTsoGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get pd tso groups: %w", err)
	}
	clusterComponents = append(clusterComponents, pdTsoComps...)

	// pd scheduling
	pdScheComps, err := listPDSchedulingGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get pd scheduling groups: %w", err)
	}
	clusterComponents = append(clusterComponents, pdScheComps...)

	// resource manager
	resourceManagerComps, err := listResourceManagerGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get resource manager groups: %w", err)
	}
	clusterComponents = append(clusterComponents, resourceManagerComps...)

	// tikv
	tikvComps, err := listTiKVGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get tikv groups: %w", err)
	}
	clusterComponents = append(clusterComponents, tikvComps...)

	// tiflash
	tiflashComps, err := listTiFlashGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get tiflash groups: %w", err)
	}
	clusterComponents = append(clusterComponents, tiflashComps...)

	// tikv worker
	tikvWorkerComps, err := listTiKVWorkerGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get tikv worker groups: %w", err)
	}
	clusterComponents = append(clusterComponents, tikvWorkerComps...)

	// coprocessor worker
	coprocessorWorkerComps, err := listCoprocessorWorkerGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get coprocessor worker groups: %w", err)
	}
	clusterComponents = append(clusterComponents, coprocessorWorkerComps...)

	// schema manager
	schemaManagerComps, err := listSchemaManagerGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get schema manager groups: %w", err)
	}
	clusterComponents = append(clusterComponents, schemaManagerComps...)

	// tidb
	tidbComps, err := listTiDBGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get tidb groups: %w", err)
	}
	clusterComponents = append(clusterComponents, tidbComps...)

	// tiproxy
	tiproxyComps, err := listTiProxyGroups(ctx, remoteClient, dataplaneNs)
	if err != nil {
		return nil, fmt.Errorf("failed to get tiproxy groups: %w", err)
	}
	clusterComponents = append(clusterComponents, tiproxyComps...)

	return clusterComponents, nil
}

func cvtStoragesToEn(ctx context.Context, volumes []v1alpha1.Volume) ([]*entity.Storage, error) {
	if len(volumes) == 0 {
		return nil, nil
	}

	clusterStorages := make([]*entity.Storage, 0, len(volumes))

	for _, volume := range volumes {
		if volume.StorageClassName == nil {
			continue
		}
		name := *volume.StorageClassName
		if volume.VolumeAttributesClassName != nil && *volume.VolumeAttributesClassName != "" {
			name = *volume.VolumeAttributesClassName
		}

		storageName, err := cvtStorageNameToEn(volume.Name)
		if err != nil {
			return nil, err
		}
		storageType := getStorageType(name)
		iops := getStorageIops(name)

		// local disk not support iops
		// cloud auto can set iops to 0
		if iops == 0 && storageType != localSSD && storageType != cloudAuto {
			tilog.Error(ctx, "invalid iops", zap.Uint32("iops", iops), zap.Any("volume", volume), zap.Any("volume", volumes))
			return nil, errors.New("invalid iops")
		}
		clusterStorages = append(clusterStorages, &entity.Storage{
			Name:          storageName,
			Type:          storageType,
			SizeGi:        uint32(float64(volume.Storage.ScaledValue(resource.Giga)) / 1.073741824),
			Iops:          iops,
			ThroughputMbs: getStorageThroughput(name),
			MountPath:     volume.Mounts[0].MountPath,
		})
	}

	if len(clusterStorages) != len(volumes) {
		tilog.Error(ctx, "convert storage failed", zap.Any("entity storage", clusterStorages), zap.Any("operator storage", volumes))
		return nil, errors.New("convert storage failed")
	}

	return clusterStorages, nil
}