package job

import (
	"context"
	"fmt"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"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/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/util/collections"
)

type ClusterExecutor struct {
	*BaseTask
}

func NewClusterExecutor(ctx *Context) *ClusterExecutor {
	return &ClusterExecutor{BaseTask: &BaseTask{Context: ctx}}
}

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

func (t *ClusterExecutor) Create(ctx context.Context) deploy.Result {
	obj := &eksv1alpha1.TiflowClusterExecutor{}
	obj.Namespace = t.Network.Namespace
	obj.Name = t.Namer.GetTiflowClusterName()
	_, err := controllerutil.CreateOrUpdate(ctx, t.Local, obj, func() error {
		obj.Spec.NetworkName = t.Network.Name
		obj.Spec.SpecCode = t.TiflowJob.Spec.SpecCode
		if obj.Labels == nil {
			obj.Labels = make(map[string]string)
		}
		labelOfTiDBClusterName := spec.GetTiDBClusterNameLabel(&(t.TiflowJob.ObjectMeta))
		labelFromConfig := t.Config.GenerateLabelFromConfig()
		obj.Labels = collections.MergeMaps(obj.Labels, labelFromConfig, labelOfTiDBClusterName)
		obj.Labels[eksv1alpha1.TiflowMasterManagerTypeLabelKey] = t.Namer.GetTiflowClusterMasterManagerType()
		obj.Labels[eksv1alpha1.TiflowJobLabelKey] = t.TiflowJob.Name
		return nil
	})
	return deploy.ErrorOrDone(err)
}

func (t *ClusterExecutor) PostCreate(ctx context.Context) deploy.Result {
	clusterExecutor, err := t.getClusterExecutor(ctx)
	if err != nil {
		return deploy.Error(err)
	}

	t.observe(ctx, clusterExecutor)
	isReady := t.checkClusterExecutorIsReady(clusterExecutor)
	if isReady {
		return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(time.Second * 30)
	} else {
		return deploy.Error(fmt.Errorf("namespace is %s, executor name is %s, phase is %s, is not running",
			clusterExecutor.Namespace, clusterExecutor.Name, clusterExecutor.Status.Phase))
	}
}

func (t *ClusterExecutor) observe(ctx context.Context, clusterExecutor *eksv1alpha1.TiflowClusterExecutor) {
	t.observeTiflowClusterMasterServerName(ctx, clusterExecutor)
	t.observeTceSpecCode(ctx, clusterExecutor)
}

func (t *ClusterExecutor) observeTceSpecCode(ctx context.Context, clusterExecutor *eksv1alpha1.TiflowClusterExecutor) {
	t.TiflowJob.Status.TceSpecCode = clusterExecutor.Spec.SpecCode
}

func (t *ClusterExecutor) observeTiflowClusterMasterServerName(
	ctx context.Context, clusterExecutor *eksv1alpha1.TiflowClusterExecutor) {
	if clusterExecutor.Spec.TiflowClusterMasterName == "" {
		return
	}
	tiflowClusterMaster := &eksv1alpha1.TiflowClusterMaster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      clusterExecutor.Spec.TiflowClusterMasterName,
			Namespace: t.Network.Namespace,
		},
	}
	err := t.Local.Get(ctx, client.ObjectKeyFromObject(tiflowClusterMaster), tiflowClusterMaster)
	if err != nil {
		t.Logger.Error(err, "query tiflow cluster master occur error!",
			"namespace", tiflowClusterMaster.Namespace, "name", tiflowClusterMaster.Name)
	}
	t.TiflowJob.Status.MasterServerName = tiflowClusterMaster.Status.ServerName
}

func (t *ClusterExecutor) checkClusterExecutorIsReady(clusterExecutor *eksv1alpha1.TiflowClusterExecutor) bool {
	if clusterExecutor.Status.Phase != eksv1alpha1.TiflowClusterExecutorPhaseRunning {
		return false
	} else {
		return true
	}
}

func (t *ClusterExecutor) getClusterExecutor(ctx context.Context) (*eksv1alpha1.TiflowClusterExecutor, error) {
	clusterExecutor := t.getClusterExecutorObject()
	err := t.Local.Get(ctx, client.ObjectKeyFromObject(clusterExecutor), clusterExecutor)
	if err != nil {
		return nil, err
	}
	return clusterExecutor, nil
}

func (t *ClusterExecutor) Delete(ctx context.Context) deploy.Result {
	return deploy.ErrorOrDone(k8s.DeleteObject(ctx, t.Local, t.getClusterExecutorObject()))
}

func (t *ClusterExecutor) getClusterExecutorObject() *eksv1alpha1.TiflowClusterExecutor {
	clusterExecutor := &eksv1alpha1.TiflowClusterExecutor{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: t.Network.Namespace,
			Name:      t.Namer.GetTiflowClusterName(),
		},
	}
	return clusterExecutor
}

func (t *ClusterExecutor) PostDelete(ctx context.Context) deploy.Result {
	clusterExecutor := t.getClusterExecutorObject()
	return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(clusterExecutor), clusterExecutor))
}