package kubernetes

import (
	"context"
	"errors"
	"fmt"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/config"
)

type NodePodSource interface {
	ListNodes(ctx context.Context) ([]Node, error)
	ListPods(ctx context.Context) ([]Pod, error)
	ListNamespaces(ctx context.Context) ([]Namespace, error)
	ListResourceQuotas(ctx context.Context) ([]ResourceQuota, error)
}

type Source struct {
	reader client.Reader
}

func NewSource(reader client.Reader) *Source {
	return &Source{reader: reader}
}

func NewInClusterSource(ctx context.Context) (*Source, error) {
	scheme := runtime.NewScheme()
	if err := corev1.AddToScheme(scheme); err != nil {
		return nil, fmt.Errorf("add corev1 scheme: %w", err)
	}

	cfg, err := config.GetConfig()
	if err != nil {
		return nil, fmt.Errorf("get cluster config: %w", err)
	}

	reader, err := newCachedReader(ctx, cfg, scheme)
	if err != nil {
		return nil, err
	}

	return NewSource(reader), nil
}

func (s *Source) ListNodes(ctx context.Context) ([]Node, error) {
	var nodeList corev1.NodeList
	if err := s.reader.List(ctx, &nodeList); err != nil {
		return nil, fmt.Errorf("list nodes: %w", err)
	}

	nodes := make([]Node, 0, len(nodeList.Items))
	for i := range nodeList.Items {
		node := &nodeList.Items[i]
		nodes = append(nodes, Node{
			Name:        node.Name,
			Labels:      cloneLabels(node.Labels),
			CapacityCPU: float64(node.Status.Capacity.Cpu().Value()),
			NodeSpec:    node.Labels["node-spec"],
			Taints:      cloneTaints(node.Spec.Taints),
		})
	}

	return nodes, nil
}

func (s *Source) ListPods(ctx context.Context) ([]Pod, error) {
	var podList corev1.PodList
	// We intentionally scan all namespaces because the workload namespace is generated dynamically
	if err := s.reader.List(ctx, &podList); err != nil {
		return nil, fmt.Errorf("list pods: %w", err)
	}

	pods := make([]Pod, 0, len(podList.Items))
	for i := range podList.Items {
		pod := &podList.Items[i]
		pods = append(pods, Pod{
			Name:      pod.Name,
			NodeName:  pod.Spec.NodeName,
			Labels:    cloneLabels(pod.Labels),
			OwnerKind: firstOwnerKind(pod.OwnerReferences),
			Phase:     string(pod.Status.Phase),
		})
	}

	return pods, nil
}

func (s *Source) ListNamespaces(ctx context.Context) ([]Namespace, error) {
	var namespaceList corev1.NamespaceList
	// Resource pool CPU limit export needs the full shared-pool namespace set so
	// pools whose quota was deleted for vcpusLimit=0 still emit a zero-valued sample.
	if err := s.reader.List(ctx, &namespaceList); err != nil {
		return nil, fmt.Errorf("list namespaces: %w", err)
	}

	namespaces := make([]Namespace, 0, len(namespaceList.Items))
	for i := range namespaceList.Items {
		namespace := &namespaceList.Items[i]
		namespaces = append(namespaces, Namespace{
			Name:   namespace.Name,
			Labels: cloneLabels(namespace.Labels),
		})
	}

	return namespaces, nil
}

func (s *Source) ListResourceQuotas(ctx context.Context) ([]ResourceQuota, error) {
	var resourceQuotaList corev1.ResourceQuotaList
	// Resource pools live in generated dataplane namespaces, so we scan all namespaces.
	if err := s.reader.List(ctx, &resourceQuotaList); err != nil {
		return nil, fmt.Errorf("list resource quotas: %w", err)
	}

	resourceQuotas := make([]ResourceQuota, 0, len(resourceQuotaList.Items))
	for i := range resourceQuotaList.Items {
		resourceQuota := &resourceQuotaList.Items[i]
		hardCPU := 0.0
		if cpuQty, ok := resourceQuota.Spec.Hard[corev1.ResourceRequestsCPU]; ok {
			hardCPU = cpuQty.AsApproximateFloat64()
		}
		resourceQuotas = append(resourceQuotas, ResourceQuota{
			Name:      resourceQuota.Name,
			Namespace: resourceQuota.Namespace,
			HardCPU:   hardCPU,
			Labels:    cloneLabels(resourceQuota.Labels),
		})
	}

	return resourceQuotas, nil
}

type starterCache interface {
	Start(context.Context) error
	WaitForCacheSync(context.Context) bool
}

func newCachedReader(ctx context.Context, cfg *rest.Config, scheme *runtime.Scheme) (client.Reader, error) {
	cfg = rest.CopyConfig(cfg)
	cfg.Timeout = 0
	cfg.QPS = 100
	cfg.Burst = 150

	httpClient, err := rest.HTTPClientFor(cfg)
	if err != nil {
		return nil, fmt.Errorf("build Kubernetes HTTP client: %w", err)
	}

	reader, err := cache.New(cfg, cache.Options{
		Scheme:     scheme,
		HTTPClient: httpClient,
	})
	if err != nil {
		return nil, fmt.Errorf("create Kubernetes cache: %w", err)
	}

	if err := startCache(ctx, reader); err != nil {
		return nil, err
	}

	return reader, nil
}

func startCache(ctx context.Context, reader starterCache) error {
	errCh := make(chan error, 1)
	go func() {
		if err := reader.Start(ctx); err != nil && !errors.Is(err, context.Canceled) {
			errCh <- err
		}
	}()

	select {
	case err := <-errCh:
		return fmt.Errorf("start cache: %w", err)
	default:
	}

	syncCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if reader.WaitForCacheSync(syncCtx) {
		return nil
	}

	select {
	case err := <-errCh:
		return fmt.Errorf("start cache: %w", err)
	case <-time.After(10 * time.Millisecond):
	}

	return fmt.Errorf("cache is not synced: %w", syncCtx.Err())
}

func firstOwnerKind(owners []metav1.OwnerReference) string {
	if len(owners) == 0 {
		return ""
	}

	return owners[0].Kind
}

func cloneLabels(labels map[string]string) map[string]string {
	if len(labels) == 0 {
		return nil
	}

	cloned := make(map[string]string, len(labels))
	for k, v := range labels {
		cloned[k] = v
	}

	return cloned
}

func cloneTaints(taints []corev1.Taint) []Taint {
	if len(taints) == 0 {
		return nil
	}

	cloned := make([]Taint, 0, len(taints))
	for _, taint := range taints {
		cloned = append(cloned, Taint{
			Key:   taint.Key,
			Value: taint.Value,
		})
	}

	return cloned
}