package k8s

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"strings"

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	rdutil "gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/module/remote_development/agent/util"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/logz"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/cli-runtime/pkg/resource"
	"k8s.io/client-go/dynamic"
	"k8s.io/kubectl/pkg/cmd/util"
	"sigs.k8s.io/cli-utils/pkg/apply"
	"sigs.k8s.io/cli-utils/pkg/apply/event"
	"sigs.k8s.io/cli-utils/pkg/common"
	"sigs.k8s.io/cli-utils/pkg/inventory"
	"sigs.k8s.io/cli-utils/pkg/kstatus/watcher"
	"sigs.k8s.io/cli-utils/pkg/object/validation"
)

var (
	errNoInventoryFound       = errors.New("no inventory found")
	errNoOwningInventoryFound = errors.New("no owning inventory found")
)

type K8sClient struct {
	log           *slog.Logger
	dynamicClient dynamic.Interface
	applier       *apply.Applier
	factory       util.Factory
}

// verify interface compliance for K8sClient
var _ Client = (*K8sClient)(nil)

// applierInfo contains the information that is needed to run an applier command to Kubernetes.
// It contains the inventory object and the objects tracked by that inventory.
type applierInfo struct {
	invInfo *unstructured.Unstructured
	objects []*unstructured.Unstructured
}

func New(log *slog.Logger, factory util.Factory) (*K8sClient, error) {
	dynamicClient, err := factory.DynamicClient()
	if err != nil {
		return nil, err
	}

	invClient, err := inventory.ConfigMapClientFactory{}.NewClient(factory)
	if err != nil {
		return nil, err
	}

	applier, err := apply.NewApplierBuilder().
		WithFactory(factory).
		WithInventoryClient(invClient).
		Build()
	if err != nil {
		return nil, err
	}

	return &K8sClient{
		log:           log,
		dynamicClient: dynamicClient,
		applier:       applier,
		factory:       factory,
	}, nil
}

func (k *K8sClient) convertToUnstructured(obj runtime.Object) (*unstructured.Unstructured, error) {
	objMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
	if err != nil {
		return nil, err
	}
	return &unstructured.Unstructured{
		Object: objMap,
	}, nil
}

func (k *K8sClient) Get(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) (*unstructured.Unstructured, error) {
	return k.dynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
}

func (k *K8sClient) UpdateOrCreate(ctx context.Context, gvr schema.GroupVersionResource, namespace string, obj runtime.Object, force bool) (*unstructured.Unstructured, error) {
	unstructuredObj, err := k.convertToUnstructured(obj)
	if err != nil {
		return nil, err
	}
	options := metav1.ApplyOptions{
		FieldManager: api.FieldManager,
		Force:        force,
	}

	return k.dynamicClient.Resource(gvr).Namespace(namespace).Apply(ctx, unstructuredObj.GetName(), unstructuredObj, options)
}

func (k *K8sClient) Delete(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) error {
	return k.dynamicClient.Resource(gvr).Namespace(namespace).Delete(ctx, name, metav1.DeleteOptions{})
}

func (k *K8sClient) Apply(ctx context.Context, config string) <-chan error {
	objs, err := k.Decode(strings.NewReader(config))
	if err != nil {
		return rdutil.ToAsync(err)
	}

	parsedApplierInfo, err := k.groupObjectsByInventory(objs)
	if err != nil {
		return rdutil.ToAsync(err)
	}

	errorChannels := make([]<-chan error, 0, len(parsedApplierInfo))
	for _, applierInfo := range parsedApplierInfo {
		inventoryName := applierInfo.invInfo.GetName()
		namespace := applierInfo.invInfo.GetNamespace()

		// process work - apply to cluster
		k.log.Debug("Applying work to cluster", logz.InventoryName(inventoryName), logz.InventoryNamespace(namespace))

		invInfo, err := inventory.ConfigMapToInventoryInfo(applierInfo.invInfo)
		if err != nil {
			errorChannels = append(errorChannels, rdutil.ToAsync(err))
			continue
		}
		applierOptions := apply.ApplierOptions{
			ServerSideOptions: common.ServerSideOptions{
				ServerSideApply: true,
				ForceConflicts:  true,
				FieldManager:    api.FieldManager,
			},
			ReconcileTimeout:         0,
			EmitStatusEvents:         true,
			PruneTimeout:             0,
			ValidationPolicy:         validation.ExitEarly,
			WatcherRESTScopeStrategy: watcher.RESTScopeNamespace,
		}
		events := k.applier.Run(ctx, invInfo, applierInfo.objects, applierOptions)
		errCh := rdutil.RunWithAsyncResult[error](func(outCh chan<- error) {
			for e := range events {
				k.log.Debug("Applied event", applyEvent(e))
				if e.Type == event.ErrorType {
					k.log.Error(
						"Error when applying config",
						logz.Error(e.ErrorEvent.Err),
						logz.InventoryName(inventoryName),
						logz.InventoryNamespace(namespace),
					)
					outCh <- e.ErrorEvent.Err
				}
			}
		})
		errorChannels = append(errorChannels, errCh)

		k.log.Debug("Applied work to cluster", logz.InventoryName(inventoryName), logz.InventoryNamespace(namespace))
	}

	return k.combineChannelsAndJoinErrors(errorChannels)
}

// combineChannelsAndJoinErrors combines all the errors published in input slice of channels,
// merges them into one error and makes it available in the returned channel. The channel is closed
// automatically after the error is published
func (k *K8sClient) combineChannelsAndJoinErrors(errorChannels []<-chan error) <-chan error {
	return rdutil.RunWithAsyncResult[error](func(outCh chan<- error) {
		combinedChannels := rdutil.CombineChannels(errorChannels)

		var allErrors []error
		for e := range combinedChannels {
			allErrors = append(allErrors, e)
		}

		combinedErr := errors.Join(allErrors...)
		if combinedErr != nil {
			outCh <- combinedErr
		}
	})
}

func (k *K8sClient) Decode(data io.Reader) ([]*unstructured.Unstructured, error) {
	// parse in local mode to retrieve objects.
	builder := resource.NewBuilder(k.factory).
		ContinueOnError().
		Flatten().
		Unstructured().
		Local()

	builder.Stream(data, "main")

	result := builder.Do()
	var objs []*unstructured.Unstructured
	err := result.Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}
		objs = append(objs, info.Object.(*unstructured.Unstructured))
		return nil
	})
	if err != nil {
		return nil, err
	}

	return objs, nil
}

func (k *K8sClient) groupObjectsByInventory(objs []*unstructured.Unstructured) (map[string]*applierInfo, error) {
	info := map[string]*applierInfo{}
	for _, obj := range objs {
		if inventory.IsInventoryObject(obj) {
			info[obj.GetName()] = &applierInfo{
				invInfo: obj,
				objects: make([]*unstructured.Unstructured, 0),
			}
		}
	}
	if len(info) == 0 {
		return nil, errNoInventoryFound
	}

	for _, obj := range objs {
		if inventory.IsInventoryObject(obj) {
			continue
		}
		annotations := obj.GetAnnotations()
		key, ok := annotations[inventory.OwningInventoryKey]
		if !ok {
			return nil, errNoOwningInventoryFound
		}
		info[key].objects = append(info[key].objects, obj)
	}

	return info, nil
}

func applyEvent(event event.Event) slog.Attr {
	return logz.LazyValue(logz.ApplyEvent, func() slog.Value {
		return slog.StringValue(event.String())
	})
}
