package apiserver

import (
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/operator"
	pingcapV1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	clusterClientSet "github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
	clusterListers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	"time"
)

// Controller is the controller implementation for tidb cluster resources
type Controller struct {
	// kubeclientset is a standard kubernetes clusterClientSet
	kubeclientset    kubernetes.Interface
	clusterClientset clusterClientSet.Interface
	clusterLister    clusterListers.TidbClusterLister
	clustersSynced   cache.InformerSynced

	workqueue workqueue.RateLimitingInterface

	deleteFunc DeleteFunc
	updateFunc UpdateFunc
}

type DeleteFunc func(string)
type UpdateFunc func(*pingcapV1alpha1.TidbCluster)

func startController(store *Store, deleteFunc DeleteFunc, updateFunc UpdateFunc, stop chan struct{}) {

	tidbInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(store.TidbClusterClientWithoutTimeout,
		time.Duration(time.Second*30), externalversions.WithTweakListOptions(func(options *v1.ListOptions) {
			options.LabelSelector = fmt.Sprintf("%s=%s", operator.CreatedByLabelKey, operator.CreatedByLabelVal)
		}))

	tidbClusterInformer := tidbInformerFactory.Pingcap().V1alpha1().TidbClusters()

	controller := &Controller{
		kubeclientset:    store.KubeClient,
		clusterClientset: store.TidbClusterClientWithoutTimeout,
		clusterLister:    tidbClusterInformer.Lister(),
		clustersSynced:   tidbClusterInformer.Informer().HasSynced,
		workqueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "TiDBClusters"),
		updateFunc:       updateFunc,
		deleteFunc:       deleteFunc,
	}
	go tidbInformerFactory.Start(stop)

	log.Info("Setting up event handlers")
	// Set up an event handler for when tidb cluster resources change
	tidbClusterInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueCluster,
		UpdateFunc: func(old, new interface{}) {
			oldCluster := old.(*pingcapV1alpha1.TidbCluster)
			newCluster := new.(*pingcapV1alpha1.TidbCluster)
			if oldCluster.ResourceVersion == newCluster.ResourceVersion {
				// Periodic resync will send update events for all known tidb clusters.
				// Two different versions of the same cluster will always have different RVs.
				return
			}
			controller.enqueueCluster(new)
		},
		DeleteFunc: controller.enqueueClusterForDelete,
	})
	go controller.Run(2, stop)
}

// Run will set up the event handlers for types we are interested in, as well
// as syncing informer caches and starting workers. It will block until stopCh
// is closed, at which point it will shutdown the workqueue and wait for
// workers to finish processing their current work items.
func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
	defer runtime.HandleCrash()
	defer c.workqueue.ShutDown()

	// Start the informer factories to begin populating the informer caches
	log.Info("Starting tidb cluster control loop")

	// Wait for the caches to be synced before starting workers
	log.Info("Waiting for informer caches to sync")
	if ok := cache.WaitForCacheSync(stopCh, c.clustersSynced); !ok {
		return fmt.Errorf("failed to wait for caches to sync")
	}

	log.Info("Starting workers")
	// Launch two workers to process tidb cluster resources
	for i := 0; i < threadiness; i++ {
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	log.Info("Started workers")
	<-stopCh
	log.Info("Shutting down workers")

	return nil
}

// runWorker is a long-running function that will continually call the
// processNextWorkItem function in order to read and process a message on the
// workqueue.
func (c *Controller) runWorker() {
	for c.processNextWorkItem() {
		// revive:disable:empty-block
	}
}

// processNextWorkItem will read a single work item off the workqueue and
// attempt to process it, by calling the syncHandler.
func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.workqueue.Get()

	if shutdown {
		return false
	}

	// We wrap this block in a func so we can defer c.workqueue.Done.
	err := func(obj interface{}) error {
		// We call Done here so the workqueue knows we have finished
		// processing this item. We also must remember to call Forget if we
		// do not want this work item being re-queued. For example, we do
		// not call Forget if a transient error occurs, instead the item is
		// put back on the workqueue and attempted again after a back-off
		// period.
		defer c.workqueue.Done(obj)
		var key string
		var ok bool
		// We expect strings to come off the workqueue. These are of the
		// form namespace/name. We do this as the delayed nature of the
		// workqueue means the items in the informer cache may actually be
		// more up to date that when the item was initially put onto the
		// workqueue.
		if key, ok = obj.(string); !ok {
			// As the item in the workqueue is actually invalid, we call
			// Forget here else we'd go into a loop of attempting to
			// process a work item that is invalid.
			c.workqueue.Forget(obj)
			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
			return nil
		}
		// Run the syncHandler, passing it the namespace/name string of the
		// tidb cluster resource to be synced.
		if err := c.syncHandler(key); err != nil {
			return fmt.Errorf("error syncing '%s': %s", key, err.Error())
		}
		// Finally, if no error occurs we Forget this item so it does not
		// get queued again until another change happens.
		c.workqueue.Forget(obj)
		log.Info("Successfully synced", zap.String("cluster", key))
		return nil
	}(obj)

	if err != nil {
		runtime.HandleError(err)
		return true
	}

	return true
}

func (c *Controller) syncHandler(key string) error {
	// Convert the namespace/name string into a distinct namespace and name
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
		return nil
	}

	cluster, err := c.clusterLister.TidbClusters(namespace).Get(name)
	if err != nil {
		// The tidb cluster resource may no longer exist, in which case we stop processing.
		if errors.IsNotFound(err) {
			log.Warn("cluster does not exist in local cache, will delete it from status cache ...",
				zap.String("name", name))

			c.deleteFunc(name)
			return nil
		}

		runtime.HandleError(fmt.Errorf("failed to list cluster by: %s/%s", namespace, name))

		return err
	}

	log.Info("Try to process cluster", zap.String("cluster", name))
	c.updateFunc(cluster)
	return nil
}

func (c *Controller) enqueueCluster(obj interface{}) {
	var key string
	var err error
	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
		runtime.HandleError(err)
		return
	}
	c.workqueue.AddRateLimited(key)
}

func (c *Controller) enqueueClusterForDelete(obj interface{}) {
	var key string
	var err error
	key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		runtime.HandleError(err)
		return
	}
	c.workqueue.AddRateLimited(key)
}