package common

import (
	"context"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/dumpling/task"
	dutils "github.com/tidbcloud/eks-provider/controllers/dumpling/task/utils"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"

	batchv1 "k8s.io/api/batch/v1"
)

const (
	s3SecretNamePrefix   = "s3-secret-dumpling-"
	port                 = 4000
	host                 = "db-tidb"
	APIVersion           = "batch/v1"
	kind                 = "Job"
	dumplingPrefix       = "dumpling-job-"
	Finalizer            = "dumpling.cloud.pingcap.com/finalizer"
	defaultDumplingImage = "pingcap/dumpling:v20231122"
)

type Dumpling struct {
	*task.BaseTask
}

func NewDumpling(context *task.Context) *Dumpling {
	return &Dumpling{BaseTask: &task.BaseTask{Context: context}}
}

func (task *Dumpling) Name() string {
	return "Dumpling"
}

func (task *Dumpling) Create(ctx context.Context) deploy.Result {
	logger := log.FromContext(ctx)
	logger.Info("begin create in dumpling task")

	switch phase := task.Dumpling.Status.Phase; phase {
	case cloudv1.DumplingPhaseCreatedNodeGroup:
		cli, err := task.Remote.GetClient(ctx)
		if err != nil {
			return deploy.Panic(fmt.Errorf("get remote client failed: %v", err))
		}

		// create s3 secret
		s3Secret := task.createSecretForS3(ctx)
		if err := k8s.Apply(ctx, cli, s3Secret); err != nil {
			return deploy.Panic(fmt.Errorf("create s3 secret failed: %v", err))
		}
		logger.Info("create s3 secret successfully")

		dumplingJob := task.getDumplingJobCR()
		logger.Info("dumpling job", "job", dumplingJob, "pod spec", dumplingJob.Spec.Template)

		if err := k8s.Apply(ctx, cli, dumplingJob); err != nil {
			return deploy.Panic(fmt.Errorf("create dumpling job failed: %v", err))
		}

		task.Dumpling.Status.Phase = cloudv1.DumplingPhaseRunning
	}
	return deploy.Done()
}

func (task *Dumpling) createSecretForS3(ctx context.Context) *corev1.Secret {
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      s3SecretNamePrefix + task.TaskName,
			Namespace: tidbPrefix + task.Dumpling.Spec.ClusterName,
		},
		Data: map[string][]byte{
			"access_key": []byte(task.Dumpling.Spec.S3AccessKey),
			"secret_key": []byte(task.Dumpling.Spec.S3SecretKey),
		},
		Type: corev1.SecretTypeOpaque,
	}

	return secret
}

func (task *Dumpling) getDumplingJobCR() *batchv1.Job {
	args := []string{
		fmt.Sprintf("--output=%s", task.Dumpling.Spec.BucketPath),
		fmt.Sprintf("--host=%s", host),
		fmt.Sprintf("--port=%d", port),
		fmt.Sprintf("--user=%s", task.Dumpling.Spec.User),
		fmt.Sprintf("--password=%s", task.Dumpling.Spec.Password),
	}

	args = append(args, task.Dumpling.Spec.Options...)

	if len(task.Dumpling.Spec.TableFilter) != 0 {
		for _, filter := range task.Dumpling.Spec.TableFilter {
			args = append(args, fmt.Sprintf("--filter=%s", filter))
		}
	}

	env := []corev1.EnvVar{
		{
			Name: "AWS_ACCESS_KEY_ID",
			ValueFrom: &corev1.EnvVarSource{
				SecretKeyRef: &corev1.SecretKeySelector{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: s3SecretNamePrefix + task.TaskName,
					},
					Key: "access_key",
				},
			},
		},
		{
			Name: "AWS_SECRET_ACCESS_KEY",
			ValueFrom: &corev1.EnvVarSource{
				SecretKeyRef: &corev1.SecretKeySelector{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: s3SecretNamePrefix + task.TaskName,
					},
					Key: "secret_key",
				},
			},
		},
	}

	image := task.Image
	if image == "" {
		image = defaultDumplingImage
	}

	podSpec := &corev1.PodTemplateSpec{
		Spec: corev1.PodSpec{
			Containers: []corev1.Container{
				{
					Name:            "dumpling",
					Image:           image,
					Args:            args,
					Command:         []string{"/dumpling"},
					ImagePullPolicy: corev1.PullAlways,
					Env:             env,
				},
			},
			RestartPolicy: corev1.RestartPolicyNever,
			Tolerations:   utils.TolerationWithUK(cloudv1.ComponentBackup, task.Dumpling.Name),
			Affinity:      utils.NodeAffinityWithUK(dutils.DumplingNodeAffinityKey(task.Dumpling), dutils.DumplingNodeAffinityValue(task.Dumpling)),
		},
	}

	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dumplingPrefix + task.Dumpling.ObjectMeta.Name,
			Namespace: tidbPrefix + task.Dumpling.Spec.ClusterName,
		},
		Spec: batchv1.JobSpec{
			BackoffLimit: pointer.Int32(0),
			Template:     *podSpec,
		},
	}
	return job
}

func (task *Dumpling) Delete(ctx context.Context) deploy.Result {
	var dumplingJob batchv1.Job
	cli, err := task.Remote.GetClient(ctx)
	if err != nil {
		return deploy.Panic(err)
	}

	// delete secret
	var secret corev1.Secret
	if err := cli.Get(ctx, types.NamespacedName{
		Name:      s3SecretNamePrefix + task.TaskName,
		Namespace: tidbPrefix + task.Dumpling.Spec.ClusterName,
	}, &secret); err != nil {
		return deploy.Panic(fmt.Errorf("get secret failed: %v", err))
	}

	if err := k8s.DeleteObject(ctx, cli, &secret); err != nil {
		return deploy.Panic(fmt.Errorf("delete secret failed: %v", err))
	}

	// delete dumpling job cr
	if err := cli.Get(ctx, types.NamespacedName{
		Name:      dumplingPrefix + task.Dumpling.ObjectMeta.Name,
		Namespace: tidbPrefix + task.Dumpling.Spec.ClusterName,
	}, &dumplingJob); err != nil {
		return deploy.Panic(fmt.Errorf("get dumpling job failed with: %v", err))
	}

	if err := k8s.DeleteObject(ctx, cli, &dumplingJob); err != nil {
		return deploy.Panic(fmt.Errorf("delete dumpling job failed: %v", err))
	}

	controllerutil.RemoveFinalizer(task.Dumpling, Finalizer)
	if err := task.Local.Update(ctx, task.Dumpling); err != nil {
		return deploy.Panic(fmt.Errorf("update dumpling cr failed: %v", err))
	}

	return deploy.Done()
}