package master

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"time"

	"github.com/pingcap/tiflow-operator/api/config"
	"github.com/pingcap/tiflow-operator/api/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/spec"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task/cluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

type TiflowClusterMaster struct {
	TiflowClusterMasterInterface
}

type TiflowClusterMasterInterface interface {
	cluster.BaseContextGetter
	GetTiflowClusterMaster() *eksv1alpha1.TiflowClusterMaster
	FillTiflowClusterSpec(tiflowCluster *v1alpha1.TiflowCluster) error
	Ready(ctx context.Context) error
}

func (t *TiflowClusterMaster) Name() string {
	return "TiflowClusterMaster"
}

func (t *TiflowClusterMaster) Create(ctx context.Context) deploy.Result {
	err := t.waitForConditionReady(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	remote, err := t.GetBaseContext().Remote.GetClient(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	tiflowCluster, tiflowClusterMut, tiflowUpdateHook := t.generateTiflowCluster(ctx)
	_, err = k8s.CreateOrUpdate(ctx, remote, tiflowCluster, tiflowClusterMut, tiflowUpdateHook)

	return deploy.ErrorOrDone(err)
}

func (t *TiflowClusterMaster) waitForConditionReady(ctx context.Context) error {
	if err := t.Ready(ctx); err != nil {
		return err
	}
	conditions := &t.GetTiflowClusterMaster().Status.Conditions
	condition := conditions.GetCondition(eksv1alpha1.TiflowClusterBucketReady)
	if condition.Status != corev1.ConditionTrue {
		return fmt.Errorf("bucket not ready")
	}
	condition = conditions.GetCondition(eksv1alpha1.TiflowClusterNodeGroupReady)
	if condition.Status != corev1.ConditionTrue {
		return fmt.Errorf("nodegroup not ready")
	}
	if t.GetBaseContext().Config.Predicate.MatchLabels[cloudv1.ProviderTypeLabel] == consts.ProviderTypeEKS {
		// only need private link on EKS
		condition = conditions.GetCondition(eksv1alpha1.TiflowClusterPrivateLinkEndPointReady)
		if condition.Status != corev1.ConditionTrue {
			return fmt.Errorf("private link not ready")
		}
	}
	return nil
}

func (t *TiflowClusterMaster) generateTiflowCluster(ctx context.Context) (*v1alpha1.TiflowCluster, controllerutil.MutateFn, k8s.UpdateHook) {
	tiflowCluster := t.getTiflowClusterMasterObject()
	tiflowClusterMut := func() error {
		if tiflowCluster.Annotations == nil {
			tiflowCluster.Annotations = map[string]string{eksv1alpha1.TiflowMasterUpdateOperatorAnnotationKey: ""}
		}

		return t.fillSpecToTiflowCluster(tiflowCluster)
	}

	tiflowUpdateHook := func() error {
		return t.fillLastTransitionTimeAnnotationToTiflowCluster(tiflowCluster)
	}

	return tiflowCluster, tiflowClusterMut, tiflowUpdateHook
}

func (t *TiflowClusterMaster) fillLastTransitionTimeAnnotationToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) error {
	tiflowCluster.Annotations[eksv1alpha1.TiflowMasterUpdateOperatorAnnotationKey] = tiflowCluster.Status.LastTransitionTime.Format(time.RFC3339)
	return nil
}

func (t *TiflowClusterMaster) fillImageAndConfigToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) error {
	conf := t.GetBaseContext().ClusterConfig
	if conf == nil {
		return errors.New("ClusterConfig is nil, not ready")
	}

	tiflowCluster.Spec.Master.BaseImage = conf.BaseImage
	tiflowCluster.Spec.Master.Version = pointer.String(conf.Tag)
	tiflowCluster.Spec.Master.Config = &config.GenericConfig{MP: conf.Config}
	return nil
}

func (t *TiflowClusterMaster) fillSpecToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) error {
	tiflowCluster.Spec.ConfigUpdateStrategy = "RollingUpdate"
	tiflowCluster.Spec.ImagePullPolicy = "Always"
	tiflowCluster.Spec.Version = ""

	if err := t.fillMasterSpecToTiflowCluster(tiflowCluster); err != nil {
		return err
	}

	// fill specified spec here
	return t.FillTiflowClusterSpec(tiflowCluster)
}

func (t *TiflowClusterMaster) fillLabelSelectorToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) {
	if tiflowCluster.Spec.Master.NodeSelector == nil {
		tiflowCluster.Spec.Master.NodeSelector = map[string]string{}
	}
	label := t.GetBaseContext().Namer.Labels()
	for k, v := range label {
		tiflowCluster.Spec.Master.NodeSelector[k] = v
	}
}

func (t *TiflowClusterMaster) fillMasterSpecToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) error {
	if tiflowCluster.Spec.Master == nil {
		tiflowCluster.Spec.Master = &v1alpha1.MasterSpec{}
	}
	if err := t.fillImageAndConfigToTiflowCluster(tiflowCluster); err != nil {
		return err
	}

	tiflowCluster.Spec.Master.MaxFailoverCount = 0
	tiflowCluster.Spec.Master.DNSPolicy = corev1.DNSClusterFirstWithHostNet
	tiflowCluster.Spec.Master.HostNetwork = pointer.Bool(true)
	t.fillLabelSelectorToTiflowCluster(tiflowCluster)
	t.fillTolerationToTiflowCluster(tiflowCluster)
	t.fillResourceRequireToTiflowCluster(tiflowCluster)
	t.fillEnvToTiflowCluster(tiflowCluster)
	return nil
}

func (t *TiflowClusterMaster) fillTolerationToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) {
	label := t.GetBaseContext().Namer.Labels()
	tolerations := []corev1.Toleration{}
	for k, v := range label {
		tolerations = append(tolerations, corev1.Toleration{
			Key:    k,
			Value:  v,
			Effect: corev1.TaintEffectNoSchedule,
		})
	}
	sort.Slice(tolerations,
		func(i, j int) bool {
			return tolerations[i].Key < tolerations[j].Key
		})
	tiflowCluster.Spec.Master.Tolerations = tolerations
}

func (t *TiflowClusterMaster) fillResourceRequireToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) {
	spec := t.GetBaseContext().TiflowSpec
	tiflowCluster.Spec.Master.Replicas = int32(spec.InstanceNumber)
	tiflowCluster.Spec.Master.ResourceRequirements = corev1.ResourceRequirements{
		Requests: corev1.ResourceList{
			corev1.ResourceCPU:    resource.MustParse("20m"),
			corev1.ResourceMemory: resource.MustParse("50Mi"),
		},
		Limits: corev1.ResourceList{
			corev1.ResourceCPU:    spec.InstanceType.CPU,
			corev1.ResourceMemory: spec.InstanceType.Memory,
		},
	}
}

func (t *TiflowClusterMaster) fillEnvToTiflowCluster(tiflowCluster *v1alpha1.TiflowCluster) {
	tiflowCluster.Spec.Master.Env = []corev1.EnvVar{
		{
			Name:  consts.GOMEMLIMIT,
			Value: spec.GetGoMemLimit(t.GetBaseContext().TiflowSpec.InstanceType.Memory, spec.DefaultGoMemLimitRatio),
		},
	}
}

func (t *TiflowClusterMaster) getTiflowClusterMasterObject() *v1alpha1.TiflowCluster {
	return &v1alpha1.TiflowCluster{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.GetBaseContext().Namer.GetTiflowNamespace(),
			Name:      t.GetBaseContext().Namer.GetTiflowClusterName(),
		},
	}
}

func (t *TiflowClusterMaster) PostCreate(ctx context.Context) deploy.Result {
	err := t.waitForConditionReady(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	tiflowCluster, err := t.getTiflowClusterMaster(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	t.observe(tiflowCluster)
	isReady := t.checkTiflowClusterMasterIsReady(tiflowCluster)
	cluster.UpdateCondition(&t.GetTiflowClusterMaster().Status.Conditions, eksv1alpha1.TiflowClusterReady, isReady)

	if isReady {
		return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(cluster.RequeueOnlyInterval)
	} else {
		return deploy.Error(fmt.Errorf("wait dataplane master ready"))
	}
}

func (t *TiflowClusterMaster) observe(tiflowCluster *v1alpha1.TiflowCluster) {
	t.GetTiflowClusterMaster().Status.ServerName = tiflowCluster.Status.Master.ServerAddress
}

func (t *TiflowClusterMaster) checkTiflowClusterMasterIsReady(tiflowCluster *v1alpha1.TiflowCluster) bool {
	isReady := false

	format := tiflowCluster.Status.LastTransitionTime.Format(time.RFC3339)
	if format != tiflowCluster.Annotations[eksv1alpha1.TiflowMasterUpdateOperatorAnnotationKey] &&
		tiflowCluster.Status.ClusterPhase == v1alpha1.ClusterRunning {
		isReady = true
	}

	return isReady
}

func (t *TiflowClusterMaster) getTiflowClusterMaster(ctx context.Context) (*v1alpha1.TiflowCluster, error) {
	remote, err := t.GetBaseContext().Remote.GetClient(ctx)
	if err != nil {
		return nil, err
	}

	tiflowCluster := t.getTiflowClusterMasterObject()
	err = remote.Get(ctx, client.ObjectKeyFromObject(tiflowCluster), tiflowCluster)
	if err != nil {
		return nil, err
	}
	return tiflowCluster, nil
}

func (t *TiflowClusterMaster) Delete(ctx context.Context) deploy.Result {
	remote, err := t.GetBaseContext().Remote.GetClient(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.ErrorOrDone(k8s.DeleteObject(ctx, remote, t.getTiflowClusterMasterObject()))
}

func (t *TiflowClusterMaster) PostDelete(ctx context.Context) deploy.Result {
	remote, err := t.GetBaseContext().Remote.GetClient(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	tiflowClusterMaster := t.getTiflowClusterMasterObject()
	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, remote, client.ObjectKeyFromObject(tiflowClusterMaster), tiflowClusterMaster))
}