package imports

import (
	"context"
	"encoding/json"
	"fmt"
	"path"
	"strings"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/reconciler"
	"github.com/tidbcloud/infra-api/pkg/consts"

	"github.com/pingcap/errors"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	v1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

const (
	// Use a fixed version to prevent breaking changes in the latest AWS CLI image.
	awsPrecheckImage               = "docker.io/amazon/aws-cli:2.2.29"
	awsPrecheckCredentialsKey      = "credentials"
	awsPrecheckCredentialsFileName = "credentials"
	awsPrecheckVolumeName          = "credentials"
	awsPrecheckMountPath           = "/aws/credentials"

	gcpPrecheckImage = "gcr.io/google.com/cloudsdktool/cloud-sdk:354.0.0-slim"
)

func getPrecheckName(importName string) string {
	return importName + "-precheck"
}

func (ci *clusterImport) startPrecheck(ctx context.Context, i *cloudv1.Import, cred []byte) error {
	ci.log.Info("start precheck")
	switch ci.network.Spec.Provider {
	case consts.ProviderTypeAWS,
		consts.ProviderTypeAWSFreeTier,
		consts.ProviderTypeAWSSharedTier,
		consts.ProviderTypeEKS:
		return startAWSPrecheck(ctx, ci, i, cred)
	case consts.ProviderTypeGCP,
		consts.ProviderTypeGKE,
		consts.ProviderTypeGCPSharedTier:
		return startGCPPrecheck(ctx, ci, i, cred)
	default:
		return fmt.Errorf("provider %s is not supported yet", ci.network.Spec.Provider)
	}
}

func startAWSPrecheck(ctx context.Context, ci *clusterImport, i *cloudv1.Import, cred []byte) error {
	secret, job := awsPrecheckSecret(ci, i, cred), awsPrecheckJob(ci, i)
	if err := ci.env.PreCreateJobHook(job); err != nil {
		return err
	}
	if _, err := controllerutil.CreateOrUpdate(ctx, ci.clusterCli, secret, func() error { return nil }); err != nil {
		return err
	}
	if _, err := controllerutil.CreateOrUpdate(ctx, ci.clusterCli, job, func() error { return nil }); err != nil {
		return err
	}
	return nil
}

func awsPrecheckSecret(ci *clusterImport, i *cloudv1.Import, cred []byte) *corev1.Secret {
	return &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getPrecheckName(i.Name),
			Namespace: ci.k8sStatus.Namespace,
		},
		Data: map[string][]byte{
			awsPrecheckCredentialsKey: cred,
		},
	}
}

func awsPrecheckJob(ci *clusterImport, i *cloudv1.Import) *batchv1.Job {
	var s3Uri string
	if i.Spec.Source.Storage.S3 != nil {
		s3Uri = i.Spec.Source.Storage.S3.URL
	}
	theLightningLabels := lightningLabels(i.Name, ci.env.TCName(ci.cluster.Name))
	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getPrecheckName(i.Name),
			Namespace: ci.k8sStatus.Namespace,
			Labels:    theLightningLabels,
		},
		Spec: batchv1.JobSpec{
			BackoffLimit: pointer.Int32Ptr(0),
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: theLightningLabels, // should configure the label, because some network policies are applied based on label selectors
				},
				Spec: corev1.PodSpec{
					HostNetwork:   true,
					DNSPolicy:     corev1.DNSClusterFirstWithHostNet,
					RestartPolicy: corev1.RestartPolicyNever,
					Containers: []corev1.Container{
						{
							Env: []corev1.EnvVar{
								{
									Name:  "AWS_SDK_LOAD_CONFIG",
									Value: "true",
								},
								{
									Name:  "AWS_CONFIG_FILE",
									Value: path.Join(awsPrecheckMountPath, awsPrecheckCredentialsFileName),
								}},
							Name:            "aws-precheck",
							Image:           awsPrecheckImage,
							ImagePullPolicy: corev1.PullIfNotPresent,
							Command: []string{
								// Validate s3.listBuckets policy
								"aws", "s3", "ls", s3Uri,
							},
							VolumeMounts: []corev1.VolumeMount{
								{
									Name:      awsPrecheckVolumeName,
									MountPath: awsPrecheckMountPath,
								},
							},
						},
					},
					Volumes: []corev1.Volume{{
						Name: awsPrecheckVolumeName,
						VolumeSource: corev1.VolumeSource{
							Secret: &corev1.SecretVolumeSource{
								SecretName: getPrecheckName(i.Name),
								Items: []corev1.KeyToPath{
									{Key: awsPrecheckCredentialsKey, Path: awsPrecheckCredentialsFileName},
								},
							},
						},
					},
					},
				},
			},
		},
	}
}

func startGCPPrecheck(ctx context.Context, ci *clusterImport, i *cloudv1.Import, cred []byte) error {
	secret, job := gcpPrecheckSecret(ci, i, cred), gcpPrecheckJob(ci, i)
	if err := ci.env.PreCreateJobHook(job); err != nil {
		return err
	}
	if _, err := controllerutil.CreateOrUpdate(ctx, ci.clusterCli, secret, func() error { return nil }); err != nil {
		return err
	}
	if _, err := controllerutil.CreateOrUpdate(ctx, ci.clusterCli, job, func() error { return nil }); err != nil {
		return err
	}
	return nil
}

func gcpPrecheckSecret(ci *clusterImport, i *cloudv1.Import, cred []byte) *corev1.Secret {
	return &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getPrecheckName(i.Name),
			Namespace: ci.k8sStatus.Namespace,
		},
		Data: map[string][]byte{
			gcpCredentialsKey: cred,
		},
	}
}

func gcpPrecheckJob(ci *clusterImport, i *cloudv1.Import) *batchv1.Job {
	var gcsUri string
	if i.Spec.Source.Storage.GCS != nil {
		gcsUri = i.Spec.Source.Storage.GCS.URL
		// Import takes the scheme gcs:// while gsutil uses gs://.
		gcsUri = strings.ReplaceAll(gcsUri, "gcs://", "gs://")
	}

	theLightningLabels := lightningLabels(i.Name, ci.env.TCName(ci.cluster.Name))
	return &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getPrecheckName(i.Name),
			Namespace: ci.k8sStatus.Namespace,
			Labels:    theLightningLabels,
		},
		Spec: batchv1.JobSpec{
			BackoffLimit: pointer.Int32Ptr(0),
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: theLightningLabels, // should configure the label, because some network policies are applied based on label selectors
				},
				Spec: corev1.PodSpec{
					HostNetwork:   true,
					DNSPolicy:     corev1.DNSClusterFirstWithHostNet,
					RestartPolicy: corev1.RestartPolicyNever,
					Containers: []corev1.Container{
						{
							Name:            "gcp-precheck",
							Image:           gcpPrecheckImage,
							ImagePullPolicy: corev1.PullIfNotPresent,
							Command: []string{
								// Validate storage list permission
								"gsutil", "-o",
								fmt.Sprintf("Credentials:gs_service_key_file=%s", path.Join(gcpCredentialsBase, gcpCredentialsPath)),
								"ls", gcsUri,
							},
							VolumeMounts: []corev1.VolumeMount{
								{
									Name:      "gcp-credentials",
									MountPath: gcpCredentialsBase,
								},
							},
						},
					},
					Volumes: []corev1.Volume{{
						Name: "gcp-credentials",
						VolumeSource: corev1.VolumeSource{
							Secret: &corev1.SecretVolumeSource{
								SecretName: getPrecheckName(i.Name),
								Items: []corev1.KeyToPath{
									{Key: gcpCredentialsKey, Path: gcpCredentialsPath},
								},
							},
						},
					},
					},
				},
			},
		},
	}
}

type podExitInfo struct {
	ExitCode *int32  `json:"exit_code"`
	Logs     *string `json:"logs"`
}

func (ci *clusterImport) syncPrecheckStatus(ctx context.Context, i *cloudv1.Import) error {
	ci.log.Info("syncing precheck")
	namespacedName := types.NamespacedName{
		Namespace: ci.k8sStatus.Namespace,
		Name:      getPrecheckName(i.Name),
	}
	var job batchv1.Job
	err := ci.clusterCli.Get(ctx, namespacedName, &job)
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		i.Status.Phase = ""
		goto updatestatus
	}
	for _, cond := range job.Status.Conditions {
		if cond.Type == batchv1.JobComplete && cond.Status == corev1.ConditionTrue {
			i.Status.Phase = cloudv1.ImportCompleted
			goto updatestatus
		}
		if cond.Type == batchv1.JobFailed && cond.Status == corev1.ConditionTrue {
			i.Status.Phase = cloudv1.ImportFailed
			pods := &corev1.PodList{}
			if err := ci.clusterCli.List(ctx, pods,
				client.MatchingLabels{
					"job-name":       getPrecheckName(i.Name),
					"controller-uid": string(job.UID),
				},
				client.InNamespace(job.Namespace),
			); err != nil {
				return errors.Trace(err)
			}
			if pods.Size() > 0 {
				thePod := pods.Items[0]

				var thePodExitInfo podExitInfo
				theContainerState := thePod.Status.ContainerStatuses[0].State
				if theContainerState.Terminated != nil {
					exitCode := thePod.Status.ContainerStatuses[0].State.Terminated.ExitCode
					thePodExitInfo.ExitCode = &exitCode
				}

				thePodExitInfo.Logs = func() *string {
					cliSet, err := kubernetes.NewForConfig(ci.restCfg)
					if err != nil {
						ci.log.Error(err, "generate kube client from kube config error. Skip setting the log", "kubeconfig", ci.restCfg)
						return nil
					}
					getLogReq := cliSet.CoreV1().Pods(thePod.Namespace).GetLogs(thePod.Name, &corev1.PodLogOptions{})
					logData, err := getLogReq.DoRaw(ctx)
					if err != nil {
						ci.log.Error(err, "get log from Pod error.  Skip setting the log", "pod", thePod.Name)
						return nil
					}
					logStr := string(logData)
					return &logStr
				}()
				statusMsg, err := json.Marshal(&thePodExitInfo)
				if err != nil {
					return errors.Trace(err)
				}
				i.Status.Message = string(statusMsg)
			}

			goto updatestatus
		}
	}
	i.Status.Phase = cloudv1.ImportRunning
updatestatus:
	ci.log.Info("update import precheck status", "status", i.Status.Phase)
	return ci.cli.Status().Update(ctx, i)
}

func (ci *clusterImport) deletePrecheck(ctx context.Context, i *cloudv1.Import) (reconciler.ExternalDeletion, error) {
	background := metav1.DeletePropagationBackground
	jobDeleted, err := deleteK8SResource(ctx, ci.clusterCli, types.NamespacedName{
		Name:      getPrecheckName(i.Name),
		Namespace: ci.k8sStatus.Namespace,
	}, &batchv1.Job{}, &client.DeleteOptions{PropagationPolicy: &background})
	if err != nil {
		return reconciler.ExternalDeletion{}, err
	}

	secretDeleted, err := deleteK8SResource(ctx, ci.clusterCli, types.NamespacedName{
		Name:      getPrecheckName(i.Name),
		Namespace: ci.k8sStatus.Namespace,
	}, &corev1.Secret{})
	if err != nil {
		return reconciler.ExternalDeletion{}, err
	}
	return reconciler.ExternalDeletion{ResourceDeleted: jobDeleted.ResourceDeleted && secretDeleted.ResourceDeleted}, nil
}