package task

import (
	"context"
	"fmt"
	"time"

	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/namer"
	"github.com/tidbcloud/ticdc-service/pkg/cdccluster"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	corev1 "k8s.io/api/core/v1"
	"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)

	secondarySubnets, err := t.CloudProvider.GetSecondaryNodeSubnets(ctx)
	if err != nil {
		return deploy.Error(err)
	}
	taintValue := t.getTaintValue(secondarySubnets)

	desiredNode := cdccluster.Node{
		InstanceType: t.Cdc.Spec.InstanceType,
		K8sVersion:   t.Cdc.Spec.NodeVersion,
		TaintValue:   taintValue,
	}
	needReschedule, err := t.ClusterProvider.NeedReschedule(ctx, t.Remote, &desiredNode)
	if err != nil {
		return deploy.Error(err)
	}
	if needReschedule {
		req := &cloud.NodeRequest{
			Count:        t.Cdc.Spec.Replicas,
			Component:    string(cloudv1.ComponentTiCDC),
			InstanceType: t.Cdc.Spec.InstanceType,
			NodeVersion:  t.Cdc.Spec.NodeVersion,
			TaintValue:   taintValue,
		}
		t.Logger.Info("TiCDC pods needs to be rescheduled to new nodes")
		t.Logger.Info("Creating overprovision pods to trigger node scale")
		if err := t.Scaler.CreateNodeScaleTrigger(ctx, req); err != nil {
			return deploy.Error(err)
		}
		ready, err := t.Scaler.CheckNodesReady(ctx, req)
		if err != nil {
			return deploy.Error(err)
		}
		if !ready {
			return deploy.Error(fmt.Errorf("new nodes are not ready"))
		}
	}

	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.Cdc.Spec.Configuration.BaseImage, 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)
	}

	// We select node by instance type and tidb-component label
	labelSelector := map[string]string{
		namer.ComponentSelectorKey:  string(cloudv1.ComponentTiCDC),
		consts.InstanceTypeLabelKey: t.Cdc.Spec.InstanceType,
		consts.K8sVersionLabelKey:   t.Cdc.Spec.NodeVersion,
	}
	t.ClusterProvider.SetNodeSelector(labelSelector)
	class := t.getStorageClassName()
	t.ClusterProvider.SetStorage(class, fmt.Sprintf("%dGi", t.Cdc.Spec.Storage.SizeGi))

	tolerations := []corev1.Toleration{
		{
			Key:      "dedicated",
			Operator: corev1.TolerationOpEqual,
			Value:    taintValue,
			Effect:   corev1.TaintEffectNoSchedule,
		},
	}
	t.ClusterProvider.SetTolerations(tolerations)

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

	if t.Cdc.Spec.MountCertificate && t.Config.IsFeatureEnabled(config.FeatureTiCDCCertificate) {
		if err := t.ClusterProvider.SetServiceAccount(ctx, t.Remote); err != nil {
			return deploy.Error(fmt.Errorf("failed to set service account: %w", err))
		}
		if err := t.ClusterProvider.EnableMountCertificate(ctx, t.Config.TiCDCCertSidecarImage, t.Remote); err != nil {
			return deploy.Error(fmt.Errorf("failed to enable mount certificate: %w", err))
		}
	}

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

	// Put cleanNodeScaleTrigger in Create method instead of PostCreate
	// to make the status always updated but cleanNodeScaleTrigger not executed when not ready
	rescheduled, err := t.ClusterProvider.Rescheduled(ctx, t.Remote)
	if err != nil {
		return deploy.Error(err)
	}
	if !rescheduled {
		return deploy.Error(fmt.Errorf("wait ticdc pods rescheduled to new node"))
	}
	t.Logger.Info("All TiCDC pods rescheduled to new nodes")
	if err := t.Scaler.CleanNodeScaleTrigger(ctx); 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.ErrorOrDone(t.Scaler.CleanNodeScaleTrigger(ctx))
}