// Add cloud-agnostic helper functions here.

package common

import (
	"bytes"
	"context"
	"fmt"
	"net/http"

	"github.com/crossplane-contrib/provider-aws/apis/autoscaling/v1beta1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	commonv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/go-logr/logr"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/pkg/version"
)

// isNodeVersionHigher reports whether nodeVer is strictly higher than desiredVer.
// Versions are expected in "x.y" format (e.g. "1.29"). On parse/compare failure
// it returns false and a non-nil error so callers do not treat invalid versions as higher.
func isNodeVersionHigher(nodeVer, desiredVer string) (bool, error) {
	if nodeVer == "" || desiredVer == "" {
		return false, fmt.Errorf("cannot compare versions: nodeVer=%q desiredVer=%q", nodeVer, desiredVer)
	}
	higher, err := version.CompareVersions(nodeVer, ">", desiredVer)
	if err != nil {
		return false, fmt.Errorf("compare versions %q and %q: %w", nodeVer, desiredVer, err)
	}
	return higher, nil
}

func GetTiDBNodeVersion(cluster *cloudv1.Cluster, providerType string) (currentVersion string, err error) {
	switch providerType {
	case consts.ProviderTypeEKS:
		status, e := cloudv1.EKSClusterStatusFromCluster(cluster)
		if e != nil {
			err = e
			break
		}
		currentVersion = status.NodeVersion
	case consts.ProviderTypeGKE:
		status, e := cloudv1.GKEClusterStatusFromCluster(cluster)
		if e != nil {
			err = e
			break
		}
		currentVersion = status.NodeVersion
	}
	return
}

// TCComponentReplicas returns the configured replica count for the given component type.
// Returns 0 when the component is not configured in the TC spec.
func TCComponentReplicas(tc *opv1alpha1.TidbCluster, component cloudv1.ComponentType) int32 {
	switch component {
	case cloudv1.ComponentTiProxy:
		if tc.Spec.TiProxy != nil {
			return tc.Spec.TiProxy.Replicas
		}
	case cloudv1.ComponentPD:
		if tc.Spec.PD != nil {
			return tc.Spec.PD.Replicas
		}
	case cloudv1.ComponentTiKV:
		if tc.Spec.TiKV != nil {
			return tc.Spec.TiKV.Replicas
		}
	case cloudv1.ComponentTiFlash:
		if tc.Spec.TiFlash != nil {
			return tc.Spec.TiFlash.Replicas
		}
	case cloudv1.ComponentTiDB:
		if tc.Spec.TiDB != nil {
			return tc.Spec.TiDB.Replicas
		}
	}
	return 0
}

// MutateTCForRestartingComponents sets the restart annotation for the specified component types.
// It returns true if any annotation was changed. Unknown component types are silently ignored.
// Components whose spec is nil are skipped.
func MutateTCForRestartingComponents(tc *opv1alpha1.TidbCluster, version string, components ...cloudv1.ComponentType) (changed bool) {
	temp := tc.DeepCopy()
	for _, comp := range components {
		switch comp {
		case cloudv1.ComponentTiProxy:
			if tc.Spec.TiProxy != nil {
				setRestartAnnotation(&tc.Spec.TiProxy.ComponentSpec, version)
			}
		case cloudv1.ComponentPD:
			if tc.Spec.PD != nil {
				setRestartAnnotation(&tc.Spec.PD.ComponentSpec, version)
			}
		case cloudv1.ComponentTiKV:
			if tc.Spec.TiKV != nil {
				setRestartAnnotation(&tc.Spec.TiKV.ComponentSpec, version)
			}
		case cloudv1.ComponentTiFlash:
			if tc.Spec.TiFlash != nil {
				setRestartAnnotation(&tc.Spec.TiFlash.ComponentSpec, version)
			}
		case cloudv1.ComponentTiDB:
			if tc.Spec.TiDB != nil {
				setRestartAnnotation(&tc.Spec.TiDB.ComponentSpec, version)
			}
		}
	}
	return !equality.Semantic.DeepEqual(tc.Spec, temp.Spec)
}

// IsComponentRestartedAndReady checks whether a single component has been restarted for the target
// version and is fully ready. Returns true when the component is absent or has zero replicas.
// Unknown component types always return true.
func IsComponentRestartedAndReady(tc *opv1alpha1.TidbCluster, version string, component cloudv1.ComponentType) bool {
	if tc == nil {
		return true
	}
	switch component {
	case cloudv1.ComponentTiProxy:
		if tc.Spec.TiProxy == nil || tc.Spec.TiProxy.Replicas == 0 {
			return true
		}
		if tc.Spec.TiProxy.ComponentSpec.Annotations == nil || tc.Spec.TiProxy.ComponentSpec.Annotations[RestartTiDBAnnotation] != version {
			return false
		}
		if tc.TiProxyUpgrading() || tc.TiProxyScaling() {
			return false
		}
		return tc.Status.TiProxy.Synced && tc.TiProxyAllMembersReady()

	case cloudv1.ComponentPD:
		if tc.Spec.PD == nil || tc.Spec.PD.Replicas == 0 {
			return true
		}
		if tc.Spec.PD.ComponentSpec.Annotations == nil || tc.Spec.PD.ComponentSpec.Annotations[RestartTiDBAnnotation] != version {
			return false
		}
		if tc.PDUpgrading() || tc.PDScaling() {
			return false
		}
		return tc.Status.PD.Synced && tc.PDAllMembersReady()

	case cloudv1.ComponentTiKV:
		if tc.Spec.TiKV == nil || tc.Spec.TiKV.Replicas == 0 {
			return true
		}
		if tc.Spec.TiKV.ComponentSpec.Annotations == nil || tc.Spec.TiKV.ComponentSpec.Annotations[RestartTiDBAnnotation] != version {
			return false
		}
		if tc.TiKVUpgrading() || tc.TiKVScaling() {
			return false
		}
		return tc.Status.TiKV.Synced && tc.TiKVAllStoresReady()

	case cloudv1.ComponentTiFlash:
		if tc.Spec.TiFlash == nil || tc.Spec.TiFlash.Replicas == 0 {
			return true
		}
		if tc.Spec.TiFlash.ComponentSpec.Annotations == nil || tc.Spec.TiFlash.ComponentSpec.Annotations[RestartTiDBAnnotation] != version {
			return false
		}
		if tc.TiFlashUpgrading() || tc.TiFlashScaling() {
			return false
		}
		return tc.Status.TiFlash.Synced && tc.TiFlashAllStoresReady()

	case cloudv1.ComponentTiDB:
		if tc.Spec.TiDB == nil || tc.Spec.TiDB.Replicas == 0 {
			return true
		}
		if tc.Spec.TiDB.ComponentSpec.Annotations == nil || tc.Spec.TiDB.ComponentSpec.Annotations[RestartTiDBAnnotation] != version {
			return false
		}
		if tc.TiDBUpgrading() || tc.TiDBScaling() {
			return false
		}
		return tc.TiDBAllMembersReady()
	}
	return true
}

// MutateTCForRestarting add annotations to component's spec to restarts the tidb cluster gracefully.
// According to the doc, need to add an annotation like:
//
// `tidb.pingcap.com/restartedAt: 2020-04-20T12:00`
//
// To avoid the annotation being modified multiple times, we use the desired version as the value.
func MutateTCForRestarting(tc *opv1alpha1.TidbCluster, version string) (changed bool) {
	return MutateTCForRestartingComponents(tc, version,
		cloudv1.ComponentPD, cloudv1.ComponentTiKV, cloudv1.ComponentTiDB,
		cloudv1.ComponentTiFlash, cloudv1.ComponentTiProxy,
	)
}

// MutateTCForRestartingTiProxyOnly updates the restart annotation for TiProxy only.
func MutateTCForRestartingTiProxyOnly(tc *opv1alpha1.TidbCluster, version string) (changed bool) {
	return MutateTCForRestartingComponents(tc, version, cloudv1.ComponentTiProxy)
}

// MutateTCForRestartingWithoutTiProxy updates the restart annotation for all TiDB components except TiProxy.
func MutateTCForRestartingWithoutTiProxy(tc *opv1alpha1.TidbCluster, version string) (changed bool) {
	return MutateTCForRestartingComponents(tc, version,
		cloudv1.ComponentPD, cloudv1.ComponentTiKV, cloudv1.ComponentTiDB, cloudv1.ComponentTiFlash,
	)
}

// IsTiProxyRestartedAndReady checks if TiProxy has been restarted for the target version and is ready.
func IsTiProxyRestartedAndReady(tc *opv1alpha1.TidbCluster, version string) bool {
	return IsComponentRestartedAndReady(tc, version, cloudv1.ComponentTiProxy)
}

func setRestartAnnotation(compSpec *opv1alpha1.ComponentSpec, version string) {
	if compSpec.Annotations == nil {
		compSpec.Annotations = map[string]string{}
	}
	compSpec.Annotations[RestartTiDBAnnotation] = version
}

func PauseCluster(ctx context.Context, cli client.Client, cluster *cloudv1.Cluster) (updated bool, err error) {
	if cluster == nil {
		return updated, fmt.Errorf("cluster is nil")
	}
	if cluster.Status.Phase != cloudv1.ClusterPhaseRunning && cluster.Status.Phase != cloudv1.ClusterPhaseSuspend {
		return updated, fmt.Errorf("cluster is not ready for upgrading, cluster: %s, namespace: %s, phase: %s", cluster.Name, cluster.Namespace, cluster.Status.Phase)
	}

	if cluster.Spec.Paused == nil || !*cluster.Spec.Paused {
		cluster.Spec.Paused = pointer.Bool(true)
		util.SetAnnotation(cluster, cloudv1.AnnoPauseReconciliationReason, PauseReason)
		if err := k8s.Apply(ctx, cli, cluster); err != nil {
			return updated, fmt.Errorf("failed to pause cluster: %w", err)
		}
		updated = true
	} else if reason := util.GetAnnotation(cluster, cloudv1.AnnoPauseReconciliationReason); reason != "" && reason != PauseReason {
		return updated, fmt.Errorf("another one is pausing this cr, reason: %s", reason)
	}

	return updated, nil
}

// UnpauseCluster unpauses the cluster CR's reconcile.
func UnpauseCluster(ctx context.Context, cli client.Client, uo *cloudv1.UpgradeOperation) error {
	for _, step := range uo.Spec.Steps {
		if step.Cluster != "" {
			var cluster cloudv1.Cluster
			if err := cli.Get(ctx, client.ObjectKey{Name: step.Cluster, Namespace: uo.Namespace}, &cluster); err != nil {
				return fmt.Errorf("failed to get cluster %s/%s: %w", uo.Namespace, step.Cluster, err)
			}
			if cluster.Spec.Paused != nil && *cluster.Spec.Paused {
				cluster.Spec.Paused = pointer.Bool(false)

				annotations := cluster.GetAnnotations()
				delete(annotations, cloudv1.AnnoPauseReconciliationReason)
				cluster.SetAnnotations(annotations)

				if err := k8s.Apply(ctx, cli, &cluster); err != nil {
					return fmt.Errorf("failed to unpause cluster: %w", err)
				}
			}
		}
	}

	return nil
}

// FinishUpgrading unpauses network and cluster CRs' reconcile and updates UpgradeOperation's status.
func FinishUpgrading(ctx context.Context, cli client.Client, uo *cloudv1.UpgradeOperation) error {
	uo.FinishCurrentStep()
	return cli.Status().Update(ctx, uo)
}

// IsCrossplaneCRReadyAndSynced checks if the crossplane CR is ready and synced.
func IsCrossplaneCRReadyAndSynced(conditions []commonv1.Condition) bool {
	ready, synced := false, false

	for _, c := range conditions {
		switch c.Type {
		case commonv1.TypeReady:
			ready = c.Status == corev1.ConditionTrue
		case commonv1.TypeSynced:
			synced = c.Status == corev1.ConditionTrue
		}
	}

	return ready && synced
}

func IsTCReady(tc *opv1alpha1.TidbCluster) bool {
	for _, cond := range tc.Status.Conditions {
		if cond.Type == opv1alpha1.TidbClusterReady && cond.Status == corev1.ConditionTrue {
			return true
		}
	}
	return false
}

func UpdateClusterStatus(ctx context.Context, infraCli client.Client, cluster *cloudv1.Cluster, desiredVersion string) error {
	if err := cloudv1.UpdateClusterStatus(cluster, func(s *cloudv1.CommonK8sClusterStatus) bool {
		if s.NodeVersion == desiredVersion {
			return false
		}
		s.NodeVersion = desiredVersion
		return true
	}); err != nil {
		return fmt.Errorf("failed to update cluster CR status's node version: %w", err)
	}

	return infraCli.Status().Update(ctx, cluster)
}

func GetCRNameList(objects interface{}) []string {
	var names []string

	switch objs := objects.(type) {
	case []*manualv1alpha1.NodeGroup:
		for _, o := range objs {
			names = append(names, o.GetName())
		}
	case []*v1beta1.AutoScalingGroup:
		for _, o := range objs {
			names = append(names, o.GetName())
		}
	case []*gkev1beta1.NodePool:
		for _, o := range objs {
			names = append(names, o.GetName())
		}
	}

	return names
}

// LogAndEmitEvent logs the event and emits it using the provided logger and eventRecorder.
// The event type determines how the event is logged.
func LogAndEmitEvent(logger logr.Logger, eventRecorder record.EventRecorder, obj runtime.Object, eventType, msg string, err error) {
	switch eventType {
	case NormalEvent:
		logger.Info(msg)

	case WarningEvent:
		logger.Error(err, msg)
	}
	eventRecorder.Event(obj, eventType, "", msg)
}

func RestartTiDBCluster(cluster *cloudv1.Cluster, client *http.Client, version string) error {
	if cluster == nil || cluster.Status.Gateway == nil {
		return fmt.Errorf("cluster or gateway is nil")
	}

	url := cluster.Status.Gateway.Endpoint + fmt.Sprintf("/cluster-agent/tidb%s/api/v1/clusters/%s:restart", cluster.Name, cluster.Name)
	body := []byte(fmt.Sprintf(`{"operation_id": "%s", "cluster_id": "%s", "target_type": "CLUSTER"}`, version, cluster.Name))

	req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
	if err != nil {
		return fmt.Errorf("failed to create http request: %w", err)
	}
	req.Header.Add("Content-Type", "application/json")

	res, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("failed to send http request: %w", err)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		return fmt.Errorf("unexpected response status code: %d", res.StatusCode)
	}
	return nil
}