package router

import (
	"context"
	"fmt"
	"net"
	"net/url"
	"slices"
	"time"

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/syncz"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tool/grpcz"
	"gitlab.com/gitlab-org/cluster-integration/tunnel/tunserver"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
)

const (
	// We don't want any resyncs.
	resyncPeriod     time.Duration = 0
	podsIPsIndexName               = "IPs"
)

type GatewayURLQuerier struct {
	kubeClient       kubernetes.Interface
	namespace        string
	podLabelSelector string
	urlScheme        string
	urlPort          string
	subs             syncz.Subscriptions[[]grpcz.URLTarget]

	listFuncSet chan struct{}
	listFunc    func() []grpcz.URLTarget
}

func NewGatewayURLQuerier(kubeClient kubernetes.Interface, namespace, podLabelSelector, urlScheme, urlPort string) *GatewayURLQuerier {
	return &GatewayURLQuerier{
		kubeClient:       kubeClient,
		namespace:        namespace,
		podLabelSelector: podLabelSelector,
		urlScheme:        urlScheme,
		urlPort:          urlPort,
		subs:             *syncz.NewSubscriptions[[]grpcz.URLTarget](),
		listFuncSet:      make(chan struct{}),
	}
}

func (q *GatewayURLQuerier) Run(ctx context.Context) error {
	pods := q.kubeClient.CoreV1().Pods(q.namespace)
	inf := cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListWithContextFunc: func(ctx context.Context, o metav1.ListOptions) (runtime.Object, error) {
				o.LabelSelector = q.podLabelSelector
				return pods.List(context.WithoutCancel(ctx), o) // Watch will be closed via Stop()
			},
			WatchFuncWithContext: func(ctx context.Context, o metav1.ListOptions) (watch.Interface, error) {
				o.LabelSelector = q.podLabelSelector
				return pods.Watch(context.WithoutCancel(ctx), o) // Watch will be closed via Stop()
			},
		},
		&corev1.Pod{},
		resyncPeriod,
		cache.Indexers{
			podsIPsIndexName: q.readyPodIPsIndexFunc,
		},
	)
	indexer := inf.GetIndexer()
	q.listFunc = func() []grpcz.URLTarget {
		vals := indexer.ListIndexFuncValues(podsIPsIndexName)
		res := make([]grpcz.URLTarget, 0, len(vals))
		for _, val := range vals {
			res = append(res, grpcz.URLTarget(val))
		}
		return res
	}
	_, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj any) {
			q.subs.Dispatch(ctx, q.listFunc())
		},
		UpdateFunc: func(oldObj, newObj any) {
			q.subs.Dispatch(ctx, q.listFunc())
		},
		DeleteFunc: func(obj any) {
			q.subs.Dispatch(ctx, q.listFunc())
		},
	})
	if err != nil {
		return err
	}
	go func() {
		pollErr := wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool /* done */, error) {
			return inf.HasSynced(), nil
		})
		if pollErr == nil {
			close(q.listFuncSet) // unblock all CachedGatewayURLs() calls once q.listFunc is set and informer has synced.
		}
	}()
	inf.RunWithContext(ctx)
	return nil
}

func (q *GatewayURLQuerier) PollGatewayURLs(ctx context.Context, _ api.AgentKey, cb tunserver.PollGatewayURLsCallback[grpcz.URLTarget]) {
	// wait for the field to be set and informer synced
	select {
	case <-ctx.Done():
		return
	case <-q.listFuncSet:
	}
	// Subscribe, get the current list, listen for any new URLs.
	// This ensures we don't miss any URLs.
	listen := q.subs.Subscribe(ctx)
	cb(q.listFunc())
	listen(func(ctx context.Context, agentkPodsURLs []grpcz.URLTarget) {
		cb(agentkPodsURLs)
	})
}

func (q *GatewayURLQuerier) CachedGatewayURLs(_ api.AgentKey) []grpcz.URLTarget {
	// Polling is not expensive - there is no polling at all.
	// So, we want the caller to use PollGatewayURLs() straight away.
	return nil
}

func (q *GatewayURLQuerier) readyPodIPsIndexFunc(obj any) ([]string, error) {
	pod, ok := obj.(*corev1.Pod)
	if !ok {
		// This should never happen and informer will panic but that's the right thing to do.
		return nil, fmt.Errorf("object is not a Pod but %T", obj)
	}

	if isPodRunning(pod) && isPodReady(pod) && isPodHasIP(pod) {
		u := url.URL{
			Scheme: q.urlScheme,
			Host:   net.JoinHostPort(pod.Status.PodIP, q.urlPort),
		}
		return []string{u.String()}, nil
	}

	return nil, nil
}

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
}

func isPodHasIP(pod *corev1.Pod) bool {
	return pod.Status.PodIP != ""
}
