package keyspace

import (
	"context"
	"errors"
	"fmt"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	batchv1 "k8s.io/api/batch/v1"
	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/util/wait"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	defaultKeyspaceGCLifetime = "6000"

	backoffLimit          = 3
	activeDeadlineSeconds = 60
)

var (
	keyspaceEnabled  = keyspacepb.KeyspaceState_ENABLED.String()
	keyspaceDisabled = keyspacepb.KeyspaceState_DISABLED.String()
)

// UpdateStateParam represents parameters needed to modify target the state of some keyspace.
type UpdateStateParam struct {
	State string `json:"state"`
}

func (r *Reconciler) createKeyspace(
	ctx context.Context,
	keyspace *serverlessv1alpha1.Keyspace,
) (*pd.KeyspaceMeta, error) {
	var req pd.KeyspaceMeta
	req.Name = keyspace.Spec.KeyspaceName
	req.Config.GCLifeTime = defaultKeyspaceGCLifetime
	if keyspace.Spec.Encrypt {
		req.Config.Encryption = `{"enabled":true}`
	}

	k, err := r.PdClient.CreateKeyspace(ctx, &req)
	if err != nil {
		if pd.IsKeyspaceAlreadyExists(err) {
			return r.PdClient.GetKeyspaceByName(ctx, keyspace.Spec.KeyspaceName)
		}
		return nil, fmt.Errorf("fail to create keyspace: %w", err)
	}

	return k, nil
}

func (r *Reconciler) removeKeyspace(ctx context.Context, name string) error {
	keyspace, err := r.PdClient.GetKeyspaceByName(ctx, name)
	if err != nil {
		if pd.IsKeyspaceNotExists(err) {
			return nil
		}
		return err
	}
	if keyspace != nil {
		switch keyspace.State {
		case keyspaceEnabled:
			if err := r.PdClient.UpdateKeyspaceState(ctx, name, keyspacepb.KeyspaceState_DISABLED); err != nil {
				return err
			}
			fallthrough
		case keyspaceDisabled:
			if err := r.PdClient.UpdateKeyspaceState(ctx, name, keyspacepb.KeyspaceState_ARCHIVED); err != nil {
				return err
			}
		}
	}
	return nil
}

func (r *Reconciler) buildActivateJob(keyspace *serverlessv1alpha1.Keyspace) *batchv1.Job {
	volumes := []corev1.Volume{
		{
			Name: "config",
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: "activate-tidb-config",
					},
					Items: []corev1.KeyToPath{
						{
							Key:  "config-file",
							Path: "tidb.toml",
						},
					},
				},
			},
		},
		{
			Name: "startup-script",
			VolumeSource: corev1.VolumeSource{
				ConfigMap: &corev1.ConfigMapVolumeSource{
					LocalObjectReference: corev1.LocalObjectReference{
						Name: "activate-tidb-config",
					},
					Items: []corev1.KeyToPath{
						{
							Key:  "startup-script",
							Path: "tidb_start_script.sh",
						},
					},
				},
			},
		},
	}
	volumeMounts := []corev1.VolumeMount{
		{
			Name:      "config",
			ReadOnly:  true,
			MountPath: "/etc/tidb",
		},
		{
			Name:      "startup-script",
			ReadOnly:  true,
			MountPath: "/usr/local/bin",
		},
	}

	if r.EnableTLS {
		volumes = append(volumes, corev1.Volume{
			Name: "tidb-tls",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: utils.TiDBClusterSecretName,
				},
			},
		})
		volumeMounts = append(volumeMounts, corev1.VolumeMount{
			Name:      "tidb-tls",
			MountPath: "/var/lib/tidb-tls",
			ReadOnly:  true,
		})
	}

	containers := []corev1.Container{
		{
			Name:            "activate-tidb-job",
			Image:           r.ActivateTidbImage,
			ImagePullPolicy: corev1.PullIfNotPresent,
			Command:         []string{"/bin/sh", "/usr/local/bin/tidb_start_script.sh"},
			VolumeMounts:    volumeMounts,
			Env: []corev1.EnvVar{
				{Name: "TZ", Value: "UTC"},
				{Name: "KEYSPACE_NAME", Value: keyspace.Spec.KeyspaceName},
			},
		},
	}
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      keyspace.Name + "-activate-tidb-job",
			Namespace: r.Namespace,
		},
		Spec: batchv1.JobSpec{
			TTLSecondsAfterFinished: ptr.To[int32](600), //nolint: gomnd // 10 minutes
			BackoffLimit:            ptr.To[int32](backoffLimit),
			ActiveDeadlineSeconds:   ptr.To[int64](activeDeadlineSeconds),
			Template: corev1.PodTemplateSpec{
				Spec: corev1.PodSpec{
					Containers:    containers,
					Volumes:       volumes,
					RestartPolicy: corev1.RestartPolicyNever,
				},
			},
		},
	}
	return job
}

func (r *Reconciler) activateKeyspace(ctx context.Context, keyspace *serverlessv1alpha1.Keyspace) error {
	activateTidbJob := r.buildActivateJob(keyspace)
	if err := r.Client.Create(ctx, activateTidbJob); err != nil && !apierrors.IsAlreadyExists(err) {
		return err
	}

	err := wait.PollUntilContextCancel(ctx, defaultWaitInterval, false, func(ctx context.Context) (done bool, err error) {
		if err = r.Client.Get(ctx, client.ObjectKeyFromObject(activateTidbJob), activateTidbJob); err != nil {
			return
		}
		for _, cond := range activateTidbJob.Status.Conditions {
			if cond.Type == batchv1.JobComplete && cond.Status == corev1.ConditionTrue {
				return true, nil
			}
			if cond.Type == batchv1.JobFailed && cond.Status == corev1.ConditionTrue {
				return false, errors.New("activate tidb job failed")
			}
		}
		return
	})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return &utils.RequeueAfterError{RequeueAfter: defaultWaitInterval}
		}
		background := metav1.DeletePropagationBackground
		_ = r.Client.Delete(ctx, activateTidbJob, &client.DeleteOptions{PropagationPolicy: &background})
		return &utils.RequeueAfterError{Cause: err, RequeueAfter: defaultWaitInterval}
	}

	return nil
}