package task

import (
	"context"
	"fmt"
	"time"

	"github.com/tidbcloud/ticdc-service-ng/pkg/cdccluster"
	"github.com/tidbcloud/ticdc-service-ng/pkg/cloud"
	"github.com/tidbcloud/ticdc-service-ng/pkg/consts"
	"github.com/tidbcloud/ticdc-service-ng/pkg/deploy"
	"k8s.io/apimachinery/pkg/types"
)

const (
	clusterRequeueAfter = 30 * time.Second
)

type cluster struct {
	*baseTask
}

func NewClusterTask(ctx *Context) deploy.Task {
	return &cluster{
		baseTask: &baseTask{Context: ctx},
	}
}

func (t *cluster) Name() string {
	return "Cluster"
}

func (t *cluster) Create(ctx context.Context) deploy.Result {
	crName := t.Cdc.Name
	crNamespace := t.Namespace
	t.ClusterProvider.Init(crNamespace, crName)

	t.ClusterProvider.SetUpstreamCluster(crNamespace, "db")
	t.ClusterProvider.SetReplicas(int32(t.Cdc.Spec.Replicas))

	if t.Cdc.Spec.Configuration != nil {
		if err := t.ClusterProvider.SetTomlConfig(t.Cdc.Spec.Configuration.Config); err != nil {
			return deploy.Error(err)
		}
		t.ClusterProvider.SetVersion(t.Config.TiCDCBaseImage,
			t.Cdc.Spec.Configuration.Version,
			t.Cdc.Spec.Configuration.Tag)
	}
	// override some deployment specific config fields
	// use cr name as cdc cluster-id
	if err := t.ClusterProvider.SetConfigField("cluster-id", crName); err != nil {
		return deploy.Error(err)
	}
	// dir should always set by deployment instead of gitops
	if err := t.ClusterProvider.SetConfigField("data-dir", cdccluster.TiCDCDefaultDirPath); err != nil {
		return deploy.Error(err)
	}
	// Make the cdc log print to stdout, container logs should not be output to a file, otherwise it will cause disk exhaustion.
	// Docker log drive will collect log and rotate the log.
	if err := t.ClusterProvider.SetConfigField("log-file", ""); err != nil {
		return deploy.Error(err)
	}

	labelSelector := map[string]string{
		consts.NodeSpecLabelKey: cdccluster.GenNodeSpecLabelValue(t.Cdc.Spec.NodeSpec.CpuCore, t.Cdc.Spec.NodeSpec.MemoryGi),
	}
	t.ClusterProvider.SetNodeSelector(labelSelector)
	t.ClusterProvider.SetRequest(t.Cdc.Spec.NodeSpec.CpuCore, t.Cdc.Spec.NodeSpec.MemoryGi)
	vac, err := t.CloudProvider.BuildVolumeAttributesClass(types.NamespacedName{Name: t.getVolumeAttributesClassName()}, &cloud.VolumeAttributes{})
	if err != nil {
		return deploy.Error(err)
	}
	var vacName = ""
	// for now alicloud don't support vac
	if vac != nil {
		vacName = t.getVolumeAttributesClassName()
	}
	t.ClusterProvider.SetStorage(fmt.Sprintf("%dGi", t.Cdc.Spec.Storage.SizeGi), vacName)

	if len(t.Cdc.Annotations) > 0 {
		if restartedAt, ok := t.Cdc.Annotations[consts.RestartTiCDCAnnotation]; ok {
			t.ClusterProvider.SetRestart(restartedAt)
		}
	}

	if err := t.ClusterProvider.ApplyCR(ctx, t.Remote); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *cluster) PostCreate(ctx context.Context) deploy.Result {
	status, err := t.ClusterProvider.GetClusterStatus(ctx, t.Remote)
	if err != nil {
		t.Logger.Error(err, "failed to get cluster status")
		return deploy.Error(err)
	}
	t.Cdc.Status = *status
	// requeue cluster reconcile to sync status
	return deploy.Done().WithRequeuAfter(clusterRequeueAfter)
}

func (t *cluster) Delete(ctx context.Context) deploy.Result {
	name := types.NamespacedName{
		Name:      t.Cdc.Name,
		Namespace: t.Namespace,
	}
	if err := t.ClusterProvider.DeleteCR(ctx, t.Remote, name); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}