// Copyright 2026 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//go:build e2e

// Package waiter contains all polling and readiness operations used by the e2e
// environment, workloads, and component suites.
package waiter

import (
	"context"
	"fmt"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	ResourceTimeout = 8 * time.Minute
	PollInterval    = 200 * time.Millisecond
)

// DeploymentAvailable waits for the named Deployment's Available condition.
func DeploymentAvailable(ctx context.Context, c client.Client, key types.NamespacedName) error {
	return k8swait.PollUntilContextCancel(ctx, PollInterval, true, func(ctx context.Context) (bool, error) {
		deployment := &appsv1.Deployment{}
		if err := c.Get(ctx, key, deployment); err != nil {
			return false, client.IgnoreNotFound(err)
		}
		for _, condition := range deployment.Status.Conditions {
			if condition.Type == appsv1.DeploymentAvailable && condition.Status == corev1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	})
}

// CRDEstablished waits for an apiextensions CRD Established condition.
func CRDEstablished(
	ctx context.Context,
	dynamicClient dynamic.Interface,
	resource schema.GroupVersionResource,
	name string,
	timeout time.Duration,
) error {
	return k8swait.PollUntilContextTimeout(ctx, PollInterval, timeout, true, func(ctx context.Context) (bool, error) {
		crd, err := dynamicClient.Resource(resource).Get(ctx, name, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		value, found, err := unstructured.NestedFieldNoCopy(crd.Object, "status", "conditions")
		if err != nil || !found || value == nil {
			return false, err
		}
		conditions, ok := value.([]interface{})
		if !ok {
			return false, fmt.Errorf("CRD %q status.conditions has type %T", name, value)
		}
		for _, item := range conditions {
			condition, ok := item.(map[string]interface{})
			if ok && condition["type"] == "Established" && condition["status"] == "True" {
				return true, nil
			}
		}
		return false, nil
	})
}

// APIServiceAvailable waits for an aggregated APIService Available condition.
func APIServiceAvailable(
	ctx context.Context,
	dynamicClient dynamic.Interface,
	resource schema.GroupVersionResource,
	name string,
) error {
	return k8swait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
		apiService, err := dynamicClient.Resource(resource).Get(ctx, name, metav1.GetOptions{})
		if apierrors.IsNotFound(err) {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		conditions, found, err := unstructured.NestedSlice(apiService.Object, "status", "conditions")
		if err != nil || !found {
			return false, err
		}
		for _, item := range conditions {
			condition, ok := item.(map[string]interface{})
			if ok && condition["type"] == "Available" && condition["status"] == "True" {
				return true, nil
			}
		}
		return false, nil
	})
}

// NamespaceDeleted waits until a namespace disappears.
func NamespaceDeleted(ctx context.Context, c client.Client, namespace string) error {
	return k8swait.PollUntilContextCancel(ctx, PollInterval, true, func(ctx context.Context) (bool, error) {
		current := &corev1.Namespace{}
		err := c.Get(ctx, client.ObjectKey{Name: namespace}, current)
		return apierrors.IsNotFound(err), client.IgnoreNotFound(err)
	})
}

// ObjectDeleted waits until an arbitrary typed object disappears.
func ObjectDeleted(ctx context.Context, c client.Client, object client.Object) error {
	return k8swait.PollUntilContextCancel(ctx, PollInterval, true, func(ctx context.Context) (bool, error) {
		current := object.DeepCopyObject().(client.Object)
		err := c.Get(ctx, client.ObjectKeyFromObject(object), current)
		return apierrors.IsNotFound(err), client.IgnoreNotFound(err)
	})
}

// PodsDeleted waits until no Pod matching the supplied labels remains.
func PodsDeleted(
	ctx context.Context,
	c client.Client,
	namespace string,
	labels map[string]string,
) error {
	return k8swait.PollUntilContextCancel(ctx, PollInterval, true, func(ctx context.Context) (bool, error) {
		remaining := &corev1.PodList{}
		if err := c.List(ctx, remaining,
			client.InNamespace(namespace),
			client.MatchingLabels(labels),
		); err != nil {
			return false, err
		}
		return len(remaining.Items) == 0, nil
	})
}

// PodReady waits until the named Pod has a true Ready condition and returns
// its latest representation.
func PodReady(ctx context.Context, c client.Client, pod *corev1.Pod) (*corev1.Pod, error) {
	last := &corev1.Pod{}
	err := k8swait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
		if err := c.Get(ctx, client.ObjectKeyFromObject(pod), last); err != nil {
			return false, err
		}
		return IsPodReady(last), nil
	})
	if err != nil {
		return nil, fmt.Errorf("wait for Pod %s/%s Ready (phase=%s, message=%s): %w",
			pod.Namespace, pod.Name, last.Status.Phase, last.Status.Message, err)
	}
	return last.DeepCopy(), nil
}

// PodCPU waits for a fresh metrics-server sample at or above the requested CPU
// usage.
func PodCPU(
	ctx context.Context,
	metrics metricsclient.Interface,
	namespace, name string,
	minimumMilliCPU int64,
	notBefore time.Time,
) (int64, time.Time, error) {
	var (
		observedMilliCPU int64
		observedAt       time.Time
	)
	err := k8swait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
		podMetrics, err := metrics.MetricsV1beta1().PodMetricses(namespace).Get(ctx, name, metav1.GetOptions{})
		if apierrors.IsNotFound(err) || apierrors.IsServiceUnavailable(err) {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		observedMilliCPU = 0
		for _, container := range podMetrics.Containers {
			observedMilliCPU += container.Usage.Cpu().MilliValue()
		}
		observedAt = podMetrics.Timestamp.Time
		return !observedAt.IsZero() && !observedAt.Before(notBefore) && observedMilliCPU >= minimumMilliCPU, nil
	})
	if err != nil {
		return observedMilliCPU, observedAt, fmt.Errorf(
			"wait for PodMetrics %s/%s at or after %s with CPU >= %dm (last=%dm at %s): %w",
			namespace, name, notBefore.Format(time.RFC3339), minimumMilliCPU,
			observedMilliCPU, observedAt.Format(time.RFC3339), err,
		)
	}
	return observedMilliCPU, observedAt, nil
}

// IsPodReady reports whether a Pod has a true Ready condition.
func IsPodReady(pod *corev1.Pod) bool {
	for _, condition := range pod.Status.Conditions {
		if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
			return true
		}
	}
	return false
}