package watcher

import (
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	tidbClusterInformers "github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
	tidbClusterV1alpha1listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/controller"
	tidbClusterLabel "github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	kubeinformers "k8s.io/client-go/informers"
	"k8s.io/client-go/util/workqueue"

	"github.com/tidbcloud/billing-reporter/report"
)

type PodWatcher struct {
	queue      workqueue.RateLimitingInterface
	tcLister   tidbClusterV1alpha1listers.TidbClusterLister
	sqsManager *report.SqsManager
}

func NewPodWatcher(kubeFactory kubeinformers.SharedInformerFactory,
	informerFactory tidbClusterInformers.SharedInformerFactory,
	sqsManager *report.SqsManager) *PodWatcher {
	queue := workqueue.NewNamedRateLimitingQueue(
		workqueue.DefaultControllerRateLimiter(),
		"pod")
	podInformer := kubeFactory.Core().V1().Pods().Informer()
	pw := &PodWatcher{
		sqsManager: sqsManager,
		queue:      queue,
		tcLister:   informerFactory.Pingcap().V1alpha1().TidbClusters().Lister(),
	}
	watchForPod(podInformer, queue, time.Now())
	return pw
}

func (pw *PodWatcher) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer pw.queue.ShutDown()

	log.Info("Starting PodWatcher")
	defer log.Info("Shutting down PodWatcher")

	for i := 0; i < workers; i++ {
		go wait.Until(pw.worker, time.Second, stopCh)
	}

	<-stopCh
}

func (pw *PodWatcher) worker() {
	for pw.processNestWorkItem() {
		// revive:disable:empty-block
	}
}

func (pw *PodWatcher) processNestWorkItem() bool {
	key, quit := pw.queue.Get()
	if quit {
		return false
	}
	defer pw.queue.Done(key)
	if err := pw.sync(key.(Key)); err != nil {
		if errors.Find(err, controller.IsRequeueError) != nil {
			log.Info("PodWatcher: still need sync and requeue", zap.String("key", key.(string)), zap.Error(err))
		} else {
			utilruntime.HandleError(errors.Errorf("PodWatcher: %v, sync failed, err: %v", key.(string), err))
		}
		pw.queue.AddRateLimited(key)
	} else {
		pw.queue.Forget(key)
	}
	return true
}

func (pw *PodWatcher) sync(key Key) error {
	ns, name, err := splitMetaNamespaceKey(key.MetaNamespaceKey)
	if err != nil {
		return err
	}
	pwe := PodWatchEvent{
		Name:      name,
		Namespace: ns,
		Type:      key.Type,
		Timestamp: key.Timestamp,
	}
	// key.Pod is no need to check Labels nil or not here, it is filtered before put in the queue
	tcName, ok := key.Pod.GetLabels()[tidbClusterLabel.InstanceLabelKey]
	if !ok {
		return nil
	}
	tcr := &TidbClusterRef{
		Namespace: ns,
		Name:      tcName,
	}
	tc, err := pw.tcLister.TidbClusters(ns).Get(tcName)
	if err != nil {
		if k8sErrors.IsNotFound(err) {
			tcr.Existed = false
			pwe.Cluster = tcr
		} else {
			log.Error("failed to fetch tc", zap.String("namespace", ns), zap.String("tcName", tcName), zap.Error(err))
			return err
		}
	} else {
		tcr.Existed = true
		pwe.Cluster = tcr
		fillTenantIdAndClusterId(tcr, tc)
	}
	return pw.sendPodWatcherEvent(pwe)
}

func (pw *PodWatcher) sendPodWatcherEvent(pwe PodWatchEvent) error {
	msg, err := buildSqsMsg(pwe)
	if err != nil {
		return err
	}
	return pw.sqsManager.SendSqsMessage(msg)
}