package crd

import (
	"context"
	"flink-kube-operator/internal/crd/v1alpha1"
	"fmt"

	"flink-kube-operator/pkg"

	"github.com/reactivex/rxgo/v2"
	"go.uber.org/zap"
	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/watch"
)

func (crd Crd) watchFlinkJobs() rxgo.Observable {

	ch := make(chan rxgo.Item)

	go func() {
		pkg.Logger.Debug("[crd] starting watch")
		watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
		if err != nil {
			panic(err)
		}
		defer watcher.Stop()
		for event := range watcher.ResultChan() {
			pkg.Logger.Debug("[crd] event received", zap.Any("type", event.Type))
			unstructuredJob := event.Object.(*unstructured.Unstructured)
			unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
			if err != nil {
				pkg.Logger.Error("cannot create unstructured map", zap.Error(err))
				continue
			}
			job := &v1alpha1.FlinkJob{}

			err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
			if err != nil {
				pkg.Logger.Error("cannot convert unstructured to structured", zap.Error(err))
				continue
			}

			ch <- rxgo.Item{
				V: &FlinkJobCrdEvent{
					EventType: event.Type,
					Job:       job,
				},
			}
			switch event.Type {
			case watch.Bookmark:
			case watch.Modified:
				pkg.Logger.Info("[crd] [watch] flink job updated")
				fmt.Printf("FlinkJob updated: %s\n", job.GetName())
				crd.repsert(job)
			case watch.Added:
				pkg.Logger.Info("[crd] [watch] new flink job created")
				crd.repsert(job)
			case watch.Deleted:
			}
		}

	}()

	return rxgo.FromChannel(ch)
}