// Package common add by blacknc <blacknc@163.com> at 2024/08/28 17:17:20
package common

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

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	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/labels"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

const specCodeLableKey = "dmSpecCode"
const imageTagLableKey = "dmImageTag"

// AnnoKeyEnableClusterIDLabel controls whether to add cluster ID label to DM pods.
// When set to "true" on the DMJob CR, the cluster ID label will be added to
// both master and worker pod labels. For existing DMJobs, this annotation
// should be added manually after notifying customers, as it will trigger pod restart.
const AnnoKeyEnableClusterIDLabel = "dm.cloud.pingcap.com/enable-cluster-id-label"

// DMCluster manage lifecycle of dmcluster.
type DMCluster struct {
	*task.BaseTask
}

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

func NewDMCluster(ctx *task.Context) *DMCluster {
	return &DMCluster{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func GetDMClusterNamespace(dmJob *eksv1alpha1.DMJob, region string) string {
	return fmt.Sprintf("dm-%s-%s", dmJob.Namespace, region)
}

func GetDMClusterName(dmJob *eksv1alpha1.DMJob) string {
	return dmJob.Name + "-dmcluster"
}

func (t *DMCluster) describeDMCluster() (*v1alpha1.DMCluster, error) {
	reclaimPolicy := corev1.PersistentVolumeReclaimDelete
	dmToleration := corev1.Toleration{
		Key:      "dedicated",
		Operator: "Equal",
		Value:    "tidb-" + task.ComponentName,
		Effect:   corev1.TaintEffectNoSchedule,
	}

	masterConfig := v1alpha1.NewMasterConfig()
	masterConfig.Set("openapi", true)
	workerConfig := v1alpha1.NewWorkerConfig()
	workerConfig.Set("redact-info-log", true)

	workerStorageSize := t.DMWorkerSpec.StorageSize.String()
	if t.DMJob.Annotations["tags.tidbcloud.com/dm-import-mode"] == "physical" {
		workerStorageSize = "15Ti"
	}

	dmCluster := &v1alpha1.DMCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      GetDMClusterName(t.DMJob),
			Namespace: GetDMClusterNamespace(t.DMJob, t.GetRegion()),
		},
		Spec: v1alpha1.DMClusterSpec{
			PVReclaimPolicy: &reclaimPolicy,
			Master: v1alpha1.MasterSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Labels:       t.masterLable(),
					NodeSelector: t.GetMasterNodeLabels(),
					Tolerations: []corev1.Toleration{
						dmToleration,
					},
				},
				BaseImage:        t.MasterImage,
				MaxFailoverCount: ptr.To(int32(0)),
				Replicas:         int32(t.DMMasterSpec.InstanceNumber),
				StorageClassName: &t.DMMasterSpec.StorageClass.Name,
				StorageSize:      t.DMMasterSpec.StorageSize.String(),
				Config:           masterConfig,
			},
			Worker: &v1alpha1.WorkerSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					Labels:       t.workerLable(t.WorkerImage),
					NodeSelector: t.GetWorkerNodeLabels(),
					Tolerations: []corev1.Toleration{
						dmToleration,
					},
					HostNetwork: ptr.To(true),
				},
				BaseImage:        t.WorkerImage,
				MaxFailoverCount: ptr.To(int32(0)),
				Replicas:         int32(t.DMWorkerSpec.InstanceNumber),
				StorageClassName: &t.DMWorkerSpec.StorageClass.Name,
				StorageSize:      workerStorageSize,
				Config:           workerConfig,
			},
		},
	}

	return dmCluster, nil
}

func (t *DMCluster) Create(ctx context.Context) deploy.Result {

	dmCluster, err := t.describeDMCluster()
	if err != nil {
		return deploy.Panic(err)
	}

	clusterCli := t.Context.EnvProvider.GetClient()
	if err := k8s.Apply(ctx, clusterCli, dmCluster); err != nil {
		return deploy.Panic(err)
	}

	return deploy.Done()
}

func (t *DMCluster) PostCreate(ctx context.Context) deploy.Result {
	dmCluster := &v1alpha1.DMCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      GetDMClusterName(t.DMJob),
			Namespace: GetDMClusterNamespace(t.DMJob, t.GetRegion()),
		},
	}
	clusterCli := t.Context.EnvProvider.GetClient()
	if err := clusterCli.Get(ctx, client.ObjectKeyFromObject(dmCluster), dmCluster); err != nil {
		return deploy.Error(err)
	}
	t.DMJob.Status.MasterServerName = fmt.Sprintf("http://%s-dm-master.%s.svc.cluster.local:8261",
		dmCluster.Name, dmCluster.Namespace)
	t.DMJob.Status.MasterSpecCode = labels.Set(dmCluster.Spec.Master.Labels).Get(specCodeLableKey)
	t.DMJob.Status.WorkerSpecCode = labels.Set(dmCluster.Spec.Worker.Labels).Get(specCodeLableKey)
	t.DMJob.Status.DMImageTag = labels.Set(dmCluster.Spec.Worker.Labels).Get(imageTagLableKey)

	var isReady bool
	for _, condition := range dmCluster.Status.Conditions {
		if condition.Type == v1alpha1.DMClusterReady && condition.Status == corev1.ConditionTrue {
			isReady = true
			break
		}
	}
	if isReady {
		return deploy.Error(k8s.RequeueOnlyError).WithRequeuAfter(time.Second * 30)
	} else {
		return deploy.Error(fmt.Errorf("dmcluster not ready, namespace:%s, name:%s, master phase:%s, worker phase:%s",
			dmCluster.Namespace, dmCluster.Name, dmCluster.Status.Master.Phase, dmCluster.Status.Worker.Phase)).WithRequeuAfter(time.Second * 30)
	}
}

func (t *DMCluster) Delete(ctx context.Context) deploy.Result {

	if !t.IsDataplaneExist() {
		return deploy.Done()
	}

	dmCluster := &v1alpha1.DMCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      GetDMClusterName(t.DMJob),
			Namespace: GetDMClusterNamespace(t.DMJob, t.GetRegion()),
		},
	}

	clusterCli := t.Context.EnvProvider.GetClient()
	return deploy.PanicOrDone(k8s.DeleteObject(ctx, clusterCli, dmCluster))
}

func (t *DMCluster) PostDelete(ctx context.Context) deploy.Result {

	if !t.IsDataplaneExist() {
		return deploy.Done()
	}

	dmCluster := &v1alpha1.DMCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      GetDMClusterName(t.DMJob),
			Namespace: GetDMClusterNamespace(t.DMJob, t.GetRegion()),
		},
	}

	clusterCli := t.Context.EnvProvider.GetClient()
	if err := clusterCli.Get(ctx, client.ObjectKeyFromObject(dmCluster), dmCluster); err == nil {
		err = errors.New("dmcluster still exists:" + dmCluster.Name)
		t.Logger.Error(nil, "dmcluster still exists", "name", dmCluster.Name)
		return deploy.Error(err)
	} else if !apierrors.IsNotFound(err) {
		t.Logger.Error(err, "failed to get dmcluster", "name", dmCluster.Name)
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *DMCluster) masterLable() map[string]string {
	l := map[string]string{
		specCodeLableKey: t.DMMasterSpec.SpecCode,
	}
	if t.DMJob.Annotations[AnnoKeyEnableClusterIDLabel] == consts.AnnoValTrue {
		l[consts.ClusterLabelKey] = t.ClusterID
	}
	return l
}
func (t *DMCluster) workerLable(image string) map[string]string {
	tag := t.GetImageTag(image)
	l := map[string]string{
		specCodeLableKey: t.DMWorkerSpec.SpecCode,
		imageTagLableKey: tag,
	}
	if t.DMJob.Annotations[AnnoKeyEnableClusterIDLabel] == consts.AnnoValTrue {
		l[consts.ClusterLabelKey] = t.ClusterID
	}
	return l
}