package it

import (
	"bufio"
	"context"
	"errors"
	"io"
	"slices"
	"strings"
	"time"

	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/util/wait"
	"k8s.io/client-go/rest"
)

func (s *integrationSuite) logNamespaceEvents(ns string) {
	w, err := s.k8s.CoreV1().Events(ns).Watch(s.ctx, metav1.ListOptions{})
	if err != nil {
		if errors.Is(err, context.Canceled) { // test done while watch was being established
			return
		}
		s.NoError(err)
		return
	}
	defer w.Stop()
	for e := range w.ResultChan() {
		ev, ok := e.Object.(*corev1.Event)
		if !ok {
			// Might be a metav1.Status when context is canceled
			continue
		}
		s.T().Logf("%s event: %s/%s/%s: %s, %s", ev.Type, ev.InvolvedObject.Namespace, ev.InvolvedObject.Kind, ev.Name, ev.Reason, ev.Message)
	}
}

func (s *integrationSuite) slurpPodLogs(prefix string, resp rest.ResponseWrapper) {
	_ = wait.PollUntilContextCancel(s.ctx, 100*time.Millisecond, false, func(ctx context.Context) (bool, error) {
		// we want all logs. Caller will shut down agentk and we'll exit
		readCloser, err := resp.Stream(context.Background()) //nolint: contextcheck
		if err != nil {
			if !apierrors.IsBadRequest(err) { // Ignore bad request errors - container has not been started yet
				s.NoError(err)
			}
			return false, nil
		}
		defer func() {
			s.NoError(readCloser.Close())
		}()

		r := bufio.NewReader(readCloser)
		for {
			str, err := r.ReadString('\n')
			str = strings.TrimRight(str, "\n\r")
			if str != "" {
				s.T().Log(prefix, str)
			}
			switch err {
			case nil:
				continue
			case io.EOF, io.ErrUnexpectedEOF: //nolint: errorlint
			default:
				s.NoError(err)
			}
			return true, nil
		}
	})
}

func (s *integrationSuite) createNS(name, label string) *corev1.Namespace {
	ns, err := s.k8s.CoreV1().Namespaces().Create(s.ctx, &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
			Labels: map[string]string{
				"ls": label,
			},
		},
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Created Namespace:", name)
	return ns
}

func (s *integrationSuite) deleteNS(ns string) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	nss := s.k8s.CoreV1().Namespaces()
	err := nss.Delete(ctx, ns, metav1.DeleteOptions{})
	s.Require().NoError(err)
	s.T().Log("Marked Namespace for deletion:", ns)
	s.Require().Eventually(func() bool {
		_, err := nss.Get(ctx, ns, metav1.GetOptions{})
		return apierrors.IsNotFound(err)
	}, 30*time.Second, 100*time.Millisecond, "Namespace not deleted")
	s.T().Log("Namespace deleted:", ns)
}

func (s *integrationSuite) waitForPodReady(ns, name string) {
	podResourceVersion := ""
	s.Require().Eventually(func() bool {
		pod, err := s.k8s.CoreV1().Pods(ns).Get(s.ctx, name, metav1.GetOptions{})
		s.Require().NoError(err)
		if podResourceVersion != pod.ResourceVersion {
			podResourceVersion = pod.ResourceVersion
			s.T().Log("Pod phase", pod.Status.Phase, "reason", pod.Status.Reason, "conditions", pod.Status.Conditions)
			s.T().Log("Pod container statuses", pod.Status.ContainerStatuses)
		}
		return isPodReady(pod) && isPodRunning(pod)
	}, time.Minute, 100*time.Millisecond, "Pod not ready")
}

func isPodReady(pod *corev1.Pod) bool {
	return slices.ContainsFunc(pod.Status.Conditions, func(c corev1.PodCondition) bool {
		return c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue
	})
}

func isPodRunning(pod *corev1.Pod) bool {
	return pod.Status.Phase == corev1.PodRunning
}
