package k8s

import (
	"bytes"
	"context"
	"fmt"
	"time"

	httptransport "github.com/go-openapi/runtime/client"
	"github.com/go-openapi/strfmt"
	"github.com/imdario/mergo"
	"github.com/pingcap/errors"
	"golang.org/x/time/rate"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/retry"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/utils/pointer"

	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/dbaas_central/client"
	"github.com/pingcap/test-infra/sdk/dbaas_central/client/backup_service"
	"github.com/pingcap/test-infra/sdk/dbaas_central/client/cluster_service"
	"github.com/pingcap/test-infra/sdk/pkg/tidbcloud"
	"github.com/pingcap/test-infra/sdk/pkg/util"
)

const (
	kubectlExecConfigAPIVersion = "client.authentication.k8s.io/v1beta1"
	baseKubeClientLease         = 15 * time.Minute // token used by baseKubeClient has expiration, simply set it to 15 mins
)

type RequeueError struct {
	reason string
}

func (re *RequeueError) Error() string {
	return re.reason
}

type TiDBCloudObjectController struct {
	tidbCloudConfiguration tidbcloud.TiDBCloudConfiguration
	clientSet              kubernetes.Interface
	secretInformer         cache.SharedIndexInformer
	informerFactory        informers.SharedInformerFactory
	queue                  workqueue.RateLimitingInterface
	tidbCloudClient        *client.DbaasCentral
	tidbCloudKubeClientSet TiDBCloudKubeClientSet
}

func (t *TiDBCloudObjectController) StartUntil(context context.Context, workers int) error {
	t.informerFactory.Start(context.Done())
	if !cache.WaitForCacheSync(context.Done(), t.secretInformer.HasSynced) {
		return fmt.Errorf("fail to cache sync")
	}
	sugar.Infof("TiDBCloudObjectController informerFactory cache sync successfully")
	wait.Until(func() {
		t.run(context, workers)
	}, 5*time.Second, context.Done())
	return nil
}

func (t *TiDBCloudObjectController) run(context context.Context, workers int) {
	defer utilruntime.HandleCrash()
	defer t.queue.ShutDown()

	for i := 0; i < workers; i++ {
		go wait.Until(t.worker, time.Second, context.Done())
	}
	<-context.Done()
}

func (t *TiDBCloudObjectController) worker() {
	for t.processNextWorkItem() {
	}
}

func (t *TiDBCloudObjectController) processNextWorkItem() bool {
	key, quit := t.queue.Get()
	if quit {
		return false
	}
	defer t.queue.Done(key)

	if err := t.sync(key.(string)); err != nil {
		if _, ok := err.(*RequeueError); ok {
			sugar.Infof("tidbcloud cluster: %v, still need sync: %v, requeuing", key.(string), err)
		} else {
			utilruntime.HandleError(fmt.Errorf("tidbcloud cluster: %v, sync failed %v, requeuing", key.(string), err))
		}
		t.queue.AddRateLimited(key)
	} else {
		t.queue.Forget(key)
	}
	return true
}

var tidbCloudClusterFinalizer = "test-infra/tidbCloudCluster-protection"

func (t *TiDBCloudObjectController) sync(key string) error {
	ns, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}
	bookSecret, err := t.clientSet.CoreV1().Secrets(ns).Get(context.TODO(), name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		return nil
	}
	if err != nil {
		return err
	}
	if !bookSecret.ObjectMeta.DeletionTimestamp.IsZero() && stringsContains(bookSecret.ObjectMeta.Finalizers, tidbCloudClusterFinalizer) {
		return t.deleteCluster(ns, bookSecret)
	}
	if val, ok := bookSecret.Data["update_ready"]; ok && string(val) == "false" {
		return t.syncConfig(ns, bookSecret)
	}
	return nil
}

func (t *TiDBCloudObjectController) deleteCluster(ns string, bookSecret *corev1.Secret) error {
	// TODO: try to use client.Delete() here
	clusterID := string(bookSecret.Data["cluster_id"])
	token := string(bookSecret.Data["auth_token"])
	orgID := string(bookSecret.Data["org_id"])
	projectID := string(bookSecret.Data["project_id"])

	var resp *cluster_service.GetClusterOK
	resp, err := t.tidbCloudClient.ClusterService.GetCluster(&cluster_service.GetClusterParams{
		ID:        clusterID,
		OrgID:     orgID,
		ProjectID: projectID,
		Context:   context.TODO(),
	}, httptransport.BearerToken(token))
	if err != nil {
		return errors.Trace(err)
	}
	lists, err := t.tidbCloudClient.BackupService.ListBackups(&backup_service.ListBackupsParams{
		ClusterID: clusterID,
		OrgID:     orgID,
		Page:      pointer.Int64(1),
		PerPage:   pointer.Int64(100),
		ProjectID: projectID,
		Context:   context.TODO(),
	}, httptransport.BearerToken(token))
	if err != nil {
		return errors.Trace(err)
	}
	for _, backup := range lists.GetPayload().Items {
		_, err = t.tidbCloudClient.BackupService.DeleteBackup(&backup_service.DeleteBackupParams{
			ClusterID: clusterID,
			ID:        *backup.ID,
			OrgID:     orgID,
			ProjectID: projectID,
			Context:   context.TODO(),
		}, httptransport.BearerToken(token))
		if err != nil {
			return errors.Trace(err)
		}
	}
	if resp.GetPayload().Cluster != nil {
		_, err = t.tidbCloudClient.ClusterService.DeleteCluster(&cluster_service.DeleteClusterParams{
			ClusterID: clusterID,
			OrgID:     orgID,
			ProjectID: projectID,
			Context:   context.TODO(),
		}, httptransport.BearerToken(token))
		if err != nil {
			return errors.Trace(err)
		}
	}

	baseKubeClient, err := t.tidbCloudKubeClientSet.getBaseKubeClient()
	if err != nil {
		return errors.Trace(err)
	}
	if err := baseKubeClient.DynamicClient.Resource(clusterConfigGVR).
		Delete(context.TODO(), util.TiDBCloudClusterConfigName(orgID, clusterID), metav1.DeleteOptions{}); err != nil {
		return fmt.Errorf("delete cluster configs error: %v", err)
	}

	bookSecret.Finalizers = stringsRemove(bookSecret.Finalizers, tidbCloudClusterFinalizer)
	_, err = t.clientSet.CoreV1().Secrets(ns).Update(context.TODO(), bookSecret, metav1.UpdateOptions{})
	if apierrors.IsConflict(err) {
		return &RequeueError{reason: err.Error()}
	}
	return errors.Trace(err)
}

func (t *TiDBCloudObjectController) syncConfig(ns string, bookSecret *corev1.Secret) error {
	clusterConfig := new(unstructured.Unstructured)
	newClusterConfig := new(unstructured.Unstructured)

	baseKubeClient, err := t.tidbCloudKubeClientSet.getBaseKubeClient()
	if err != nil {
		return errors.Trace(err)
	}
	if _, ok := bookSecret.Data["cluster_config"]; !ok {
		// empty cluster config in secret
		newClusterConfig, err = baseKubeClient.DynamicClient.
			Resource(clusterConfigGVR).Get(context.TODO(), apiresource.TiDBCloudBaseClusterConfig, metav1.GetOptions{})
		if err != nil {
			return errors.Annotate(err, "get base cluster-config")
		}

		ccLabels := newClusterConfig.GetLabels()
		delete(newClusterConfig.Object, "metadata")
		delete(newClusterConfig.Object, "status")
		newClusterConfig.SetName(util.TiDBCloudClusterConfigName(string(bookSecret.Data["org_id"]), string(bookSecret.Data["cluster_id"])))
		if err := mergo.Merge(
			&ccLabels,
			map[string]string{
				apiresource.TiDBCloudClusterLabel: string(bookSecret.Data["cluster_id"]),
				"tidbcloud.com/managed-by":        "test-infra",
				apiresource.TiDBCloudTenentLabel:  string(bookSecret.Data["org_id"]),
			},
			mergo.WithOverride); err != nil {
			return errors.Trace(err)
		}
		newClusterConfig.SetLabels(ccLabels)
		newClusterConfig.Object["spec"].(map[string]interface{})["configMeta"].(map[string]interface{})["gitCommitID"] = "hacked-by-test-infra"
		bClusterConfig := new(bytes.Buffer)
		if err := unstructured.UnstructuredJSONScheme.Encode(newClusterConfig, bClusterConfig); err != nil {
			return errors.Trace(err)
		}
		bookSecret.Data["cluster_config"] = bClusterConfig.Bytes()
	} else if _, _, err := unstructured.UnstructuredJSONScheme.Decode(bookSecret.Data["cluster_config"], nil, newClusterConfig); err != nil {
		return errors.Trace(err)
	}

	if clusterConfig, err = baseKubeClient.DynamicClient.Resource(clusterConfigGVR).
		Get(context.TODO(),
			util.TiDBCloudClusterConfigName(string(bookSecret.Data["org_id"]), string(bookSecret.Data["cluster_id"])),
			metav1.GetOptions{},
		); err != nil {
		if !errors.IsNotFound(err) {
			return errors.Annotate(err, "get cluster config object error")
		}
	}

	if clusterConfig == nil {
		_, err = baseKubeClient.DynamicClient.
			Resource(clusterConfigGVR).Create(context.TODO(), newClusterConfig, metav1.CreateOptions{})
		if err != nil {
			return errors.Annotate(err, "create cluster-config")
		}
	} else {
		spec, _, err := unstructured.NestedFieldCopy(newClusterConfig.Object, "spec")
		if err == nil {
			return nil
		}
		if err := unstructured.SetNestedField(clusterConfig.Object, spec, "spec"); err != nil {
			return errors.Trace(err)
		}
		err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
			_, updateErr := baseKubeClient.DynamicClient.Resource(clusterConfigGVR).Update(context.Background(), clusterConfig, metav1.UpdateOptions{})
			if updateErr == nil {
				return nil
			}
			if _, err := baseKubeClient.DynamicClient.Resource(clusterConfigGVR).
				Get(context.TODO(),
					util.TiDBCloudClusterConfigName(string(bookSecret.Data["org_id"]), string(bookSecret.Data["cluster_id"])), metav1.GetOptions{}); err != nil {
				return errors.Annotate(err, "get cluster config object error")
			}
			return updateErr
		})
		if err != nil {
			return err
		}
	}

	bookSecret.Data["update_ready"] = []byte("true")
	_, err = t.clientSet.CoreV1().Secrets(ns).Update(context.TODO(), bookSecret, metav1.UpdateOptions{})
	if apierrors.IsConflict(err) {
		return &RequeueError{reason: err.Error()}
	}
	return err
}

func NewTiDBCloudClusterController(clientSet kubernetes.Interface, tidbCloudKubeClientSet TiDBCloudKubeClientSet, configuration tidbcloud.TiDBCloudConfiguration) (*TiDBCloudObjectController, error) {
	labelSelector := labels.Set(map[string]string{
		"app.kubernetes.io/managed-by": "test-infra",
		"app.kubernetes.io/name":       "tidbcloud_cluster",
	}).AsSelector()

	informerFactory := informers.NewSharedInformerFactoryWithOptions(clientSet, 30*time.Second, informers.WithTweakListOptions(func(options *metav1.ListOptions) {
		if len(options.LabelSelector) > 0 {
			options.LabelSelector += "," + labelSelector.String()
		} else {
			options.LabelSelector = labelSelector.String()
		}
	}))
	tidbCloudClient := client.NewHTTPClientWithConfig(strfmt.Default, client.DefaultTransportConfig().WithHost(configuration.TiDBCloudURL).WithSchemes([]string{"https"}))

	t := &TiDBCloudObjectController{
		tidbCloudConfiguration: configuration,
		clientSet:              clientSet,
		informerFactory:        informerFactory,
		secretInformer:         informerFactory.Core().V1().Secrets().Informer(),
		queue: workqueue.NewNamedRateLimitingQueue(
			workqueue.NewMaxOfRateLimiter(
				workqueue.NewItemExponentialFailureRateLimiter(time.Second, 100*time.Second),
				// 10 qps, 100 bucket size.  This is only for retry speed and its only the overall factor (not per item)
				&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
			),
			"tidbcloud cluster",
		),
		tidbCloudClient:        tidbCloudClient,
		tidbCloudKubeClientSet: tidbCloudKubeClientSet,
	}
	t.secretInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		UpdateFunc: func(oldObj, newObj interface{}) {
			sugar.Debugf("in update func\n")
			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(newObj)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("could't get key for object %+v: %v", newObj, err))
				return
			}
			t.queue.Add(key)
		},
		DeleteFunc: func(obj interface{}) {
			sugar.Debugf("in delete func\n")
			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err != nil {
				utilruntime.HandleError(fmt.Errorf("could't get key for object %+v: %v", obj, err))
				return
			}
			t.queue.Add(key)
		},
	})
	return t, nil
}