package imports

import (
	"context"
	stdErrors "errors"
	"strconv"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/pkg/keyspacemigration"
	providertypes "github.com/tidbcloud/aws-shared-provider/pkg/types"
	provutils "github.com/tidbcloud/aws-shared-provider/pkg/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-provider/util/collections"
	"github.com/tidbcloud/infra-provider/util/config"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

const (
	lightningPort = 8289
	lightningName = "tidb-lightning"

	importStorageRequest = "100Gi"
	importFinalizer      = "import.cloud.pingcap.com/finalizer"
)

const (
	defaultTimeout = 2 * time.Minute
	requeueAfter   = 15 * time.Second
)

type taskStatus uint8

const (
	taskStatusNotStarted taskStatus = 0
	taskStatusRunning    taskStatus = 1
	taskStatusCompleted  taskStatus = 2
)

// ImportReconciler reconciles a Import object
type ImportReconciler struct {
	client.Client
	Log             logr.Logger
	envBuilder      *ServerlessTierEnvBuilder
	migrationLocker keyspacemigration.Locker
	maxConcurrent   int
}

func NewImportReconciler(
	mgr ctrl.Manager,
	log logr.Logger,
	envBuilder *ServerlessTierEnvBuilder,
	migrationLocker keyspacemigration.Locker,
	maxConcurrent int,
) *ImportReconciler {
	return &ImportReconciler{
		Client:          mgr.GetClient(),
		Log:             log,
		envBuilder:      envBuilder,
		migrationLocker: migrationLocker,
		maxConcurrent:   maxConcurrent,
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *ImportReconciler) SetupWithManager(mgr ctrl.Manager, controllerName string, p ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		Named(controllerName).
		For(&cloudv1.Import{}, builder.WithPredicates(p...)).
		WithOptions(controller.Options{MaxConcurrentReconciles: r.maxConcurrent}).
		Complete(r)
}

func (r *ImportReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, errResult error) {
	logger := r.Log.WithValues("import", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)

	ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
	defer cancel()

	i := &cloudv1.Import{}
	err := r.Client.Get(ctx, req.NamespacedName, i)
	if err != nil {
		logger.Error(err, "get resource failed")
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	importSvc, err := r.NewImportService(ctx, i)
	if err != nil {
		logger.Error(err, "create import service failed")
		return ctrl.Result{}, err
	}

	if !i.GetDeletionTimestamp().IsZero() {
		// add delay to ensure logs are collected
		if time.Since(i.GetDeletionTimestamp().Time) < 60*time.Second {
			return ctrl.Result{
				RequeueAfter: 60 * time.Second,
			}, nil
		}
		err := importSvc.Delete(ctx, i)
		if err != nil {
			logger.Error(err, "Delete resource failed")
			return ctrl.Result{}, err
		}

		controllerutil.RemoveFinalizer(i, importFinalizer)
		if err := r.Client.Update(ctx, i); err != nil {
			logger.Error(err, "Remove Finalizer failed")
			return ctrl.Result{}, err
		}

		logger.Info("Successfully deleted resource")
		return ctrl.Result{}, nil
	}

	if importSvc.env.clusterObj == nil {
		logger.Info("cluster not found, waiting for import to be marked for deletion")
		return ctrl.Result{RequeueAfter: requeueAfter}, nil
	}

	// init finalizer
	if !controllerutil.ContainsFinalizer(i, importFinalizer) {
		controllerutil.AddFinalizer(i, importFinalizer)
		if err := r.Client.Update(ctx, i); err != nil {
			logger.Error(err, "Add Finalizer failed")
			return ctrl.Result{}, err
		}
	}

	switch i.Status.Phase {
	case "":
		err = importSvc.Create(ctx, i)
		if err != nil {
			if stdErrors.Is(err, provutils.ErrKeyspaceMetaServiceGroupMigrating) {
				keyspaceName := importSvc.env.clusterObj.Status.KeyspaceToken
				logger.Error(err, "keyspace is migrating meta-service group, will retry later", "keyspace-name", keyspaceName)
				return ctrl.Result{RequeueAfter: requeueAfter}, nil
			}
			logger.Error(err, "create resource failed")
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	case cloudv1.ImportFailed, cloudv1.ImportCompleted:
		return ctrl.Result{}, nil
	default:
		err = importSvc.Update(ctx, i)
		if err != nil {
			logger.Error(err, "update resource failed")
			return ctrl.Result{}, err
		}
		return ctrl.Result{RequeueAfter: requeueAfter}, nil
	}
}

func (r *ImportReconciler) NewImportService(ctx context.Context,
	i *cloudv1.Import) (_ *ImportService, errResult error) {
	logger := log.FromContext(ctx)
	env, err := r.envBuilder.BuildEnvFromImport(ctx, i)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return &ImportService{
		infraCli:        r.Client,
		env:             env,
		logger:          logger,
		migrationLocker: r.migrationLocker,
	}, nil
}

type ImportService struct {
	infraCli        client.Client
	env             *ServerlessTierEnv
	logger          logr.Logger
	migrationLocker keyspacemigration.Locker
}

func (ci *ImportService) checkKeyspaceNotMigrating() error {
	keyspaceName := ci.env.clusterObj.Status.KeyspaceToken
	if ci.migrationLocker != nil && ci.migrationLocker.IsLocked(keyspaceName) {
		return provutils.ErrKeyspaceMetaServiceGroupMigrating
	}
	return nil
}

func (ci *ImportService) Create(ctx context.Context, i *cloudv1.Import) (errResult error) {
	// do not evict workload pods with cluster-autoscaler/alicloud,
	// https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler
	podAnnotations := map[string]string{providertypes.CASafeToEvict: "false"}
	if ci.env.managerCfg.CloudProviderType == providertypes.ProviderTypeAliCloudFreeTier {
		podAnnotations = map[string]string{providertypes.GoatSafeToEvict: "false"}
	}

	if err := ci.checkKeyspaceNotMigrating(); err != nil {
		return err
	}

	var cfg *config.GenericConfig
	if len(i.Spec.LightningConfig) == 0 {
		return errors.New("lightning config is empty")
	}
	impConfigTOML := new(cloudv1.TomlConfig)
	if err := impConfigTOML.UnmarshalTOML([]byte(i.Spec.LightningConfig)); err != nil {
		return errors.Annotate(err, "parse config TOML data error")
	}
	cfg = config.New(impConfigTOML.MP)

	labels := utils.CopyLabelsForWorkloads(i.Labels)
	keyspaceName := ci.env.clusterObj.Status.KeyspaceToken
	labels[providertypes.KeyspaceNameLabel] = keyspaceName

	cfgBytes, err := cfg.MarshalTOML()
	if err != nil {
		return err
	}

	importCm, err := ci.env.getImportCm(ctx)
	if err != nil {
		return errors.Annotate(err, "get import cm")
	}

	job, secret := newLightning(ci, i.ObjectMeta.Name, cfgBytes, importCm)
	err = preCreateJobHook(ci.env, job, importCm)
	if err != nil {
		return err
	}

	svc := newLightningService(i.Name, ci.env.ImportJobNamespace())
	job.Labels = collections.MergeMaps(labels, job.Labels)
	job.Spec.Template.Labels = collections.MergeMaps(labels, job.Spec.Template.Labels)
	job.Spec.Template.Annotations = collections.MergeMaps(podAnnotations, job.Spec.Template.Annotations)
	secret.Labels = collections.MergeMaps(labels, secret.Labels)
	svc.Labels = collections.MergeMaps(labels, svc.Labels)

	err = ci.ensurePVC(ctx, i, labels)
	if err != nil {
		return err
	}
	appendVolume(job)

	if err := preCreateSecretHook(secret, ci.env, importCm); err != nil {
		return err
	}
	if _, err = controllerutil.CreateOrUpdate(ctx, ci.env.GetControllerClient(), secret, func() error {
		return nil
	}); err != nil {
		return err
	}

	if _, err = controllerutil.CreateOrUpdate(ctx, ci.env.GetControllerClient(), job, func() error {
		return nil
	}); err != nil {
		return err
	}

	if _, err = controllerutil.CreateOrUpdate(ctx, ci.env.GetControllerClient(), svc, func() error {
		return nil
	}); err != nil {
		return err
	}

	i.Status.Phase = cloudv1.ImportPrepare
	return ci.infraCli.Status().Update(ctx, i)
}

func (ci *ImportService) ensurePVC(ctx context.Context, i *cloudv1.Import, labels map[string]string) error {
	if i.Spec.StorageRequest == "" {
		i.Spec.StorageRequest = importStorageRequest
	}
	storageClassName := ci.env.StorageClassName()
	pvc := newPersistentVolumeClaim(i.Name, ci.env.ImportJobNamespace(),
		storageClassName, i.Spec.StorageRequest)
	pvc.Labels = collections.MergeMaps(labels, pvc.Labels)

	ci.logger.Info("Creating PVC", "name", pvc.Name, "storageClass", storageClassName, "request", i.Spec.StorageRequest)
	res, err := k8s.CreateIgnoreExists(ctx, ci.env.GetControllerClient(), pvc)
	if err != nil {
		ci.logger.Error(err, "Error creating PVC", "name", pvc.Name)
		return err
	}
	ci.logger.Info("PVC created or already exists", "name", pvc.Name, "result", res)
	return nil
}

// Update send task and update task Progress
func (ci *ImportService) Update(ctx context.Context, i *cloudv1.Import) (errResult error) {
	if setStatusErr := ci.setStatusFailedOrCompleteByJob(ctx, i); client.IgnoreNotFound(setStatusErr) != nil {
		ci.logger.Error(setStatusErr, "failed setStatusByJob requeue...")
		return setStatusErr
	}
	if i.Status.Phase == cloudv1.ImportFailed || i.Status.Phase == cloudv1.ImportCompleted {
		ci.logger.Info("Task info", "task", "update task status due to job reaching terminal state", "status", i.Status.Phase)
		return ci.infraCli.Status().Update(ctx, i)
	}

	lightningPods := &corev1.PodList{}
	if err := ci.env.GetControllerClient().List(ctx, lightningPods,
		client.MatchingLabels{"job-name": i.Name},
		client.InNamespace(ci.env.ImportJobNamespace())); err != nil {
		ci.logger.Error(err, "failed to get lightning pods, requeue...")
	}
	if len(lightningPods.Items) == 0 {
		ci.logger.Info("lightning pod not up, requeue...")
		return nil
	}
	if lightningPods.Items[0].Status.Phase == corev1.PodPending {
		ci.logger.Info("lightning pod not up, requeue...")
		// ensure pvc exists, pvc may be deleted because of zoneRedundantEvicter
		pvc := &corev1.PersistentVolumeClaim{}
		if err := ci.env.GetControllerClient().Get(ctx, types.NamespacedName{
			Name:      i.Name,
			Namespace: ci.env.ImportJobNamespace(),
		}, pvc); err != nil {
			if errors.IsNotFound(err) {
				ci.logger.Info("PVC not found, will attempt to recreate", "pvc", i.Name)
				labels := utils.CopyLabelsForWorkloads(i.Labels)
				err = ci.ensurePVC(ctx, i, labels)
				if err != nil {
					ci.logger.Error(err, "Failed to ensure PVC existence", "pvc", i.Name)
					return err
				}
				ci.logger.Info("Ensuring PVC existence", "pvc", i.Name)
			} else {
				ci.logger.Error(err, "Failed to get PVC", "pvc", i.Name)
				return err
			}
		}

		ci.logger.Info("PVC already exists", "pvc", pvc)
		return nil
	}

	scheme := "http"
	if ci.env.GetClusterCertSecretName() != "" {
		scheme = "https"
	}
	link := k8s.NewHTTPLinkNormal(ctx, scheme, ci.env.ImportJobNamespace(), i.Name, strconv.Itoa(lightningPort))
	lightningClient := newLightningClient(link)
	ci.logger.Info("Task info", "task", "start request to lightning",
		"k8s namespace", ci.env.ImportJobNamespace(), "import name", i.Name)
	task, err := lightningClient.GetTaskProgress()
	if err != nil {
		ci.logger.Error(err, "failed to get import task progress, requeue...")
		return err
	}

	ci.logger.Info("Task info", "taskMessage", task.Message, "tableCount", len(task.Tables))
	// start task when task not started or checkpoint enabled
	if task.Status == taskStatusNotStarted && (!i.Status.IsTaskCreated || ci.env.EnableCheckpoint(ctx)) {
		var cfg string
		cfg, err = ci.getLightningConfigFromSecret(ctx, i.Name)
		if err != nil {
			ci.logger.Error(err, "Task create", "task", "get config failed before createTask")
			i.Status.Phase = cloudv1.ImportFailed
			i.Status.Message = "unknown error, please seek support"
			return ci.infraCli.Status().Update(ctx, i)
		}
		taskId, err := lightningClient.CreateTask(cfg)
		if err != nil {
			i.Status.Phase = cloudv1.ImportFailed
			i.Status.Message = err.Error()
			ci.logger.Error(err, "Task create", "task", "task create failed")
			return ci.infraCli.Status().Update(ctx, i)
		}
		i.Status.IsTaskCreated = true
		i.Status.TaskId = taskId
	}

	saveImportStatus(&(i.Status), task)
	if i.Status.Phase == cloudv1.ImportFailed {
		ci.logger.Error(errors.New("task failed"),
			"Task info", "taskMessage", task.Message)
	}

	ci.logger.Info("Task info", "task", "update task complete", "status", i.Status.Phase)
	return ci.infraCli.Status().Update(ctx, i)
}

func (ci *ImportService) setStatusFailedOrCompleteByJob(ctx context.Context, i *cloudv1.Import) error {
	var job batchv1.Job
	if err := ci.env.GetControllerClient().Get(ctx, types.NamespacedName{
		Name:      i.Name,
		Namespace: ci.env.ImportJobNamespace(),
	}, &job); err != nil {
		return err
	}

	for _, cond := range job.Status.Conditions {
		if cond.Status != corev1.ConditionTrue {
			continue
		}

		// The Job may turn out to be completed (batchv1.JobComplete) if its pod was evicted either before starting
		// or after finishing the import task. In such cases, Lightning marks itself as having exited normally
		// and sets the Job status to Complete. However, we cannot reliably determine if the import task
		// actually succeeded in these scenarios.
		// For both failure (batchv1.JobFailed) and ambiguous completion cases, we mark the import as failed.
		if cond.Type == batchv1.JobFailed || cond.Type == batchv1.JobComplete {
			ci.logger.Error(errors.New(cond.Message), "import job terminated abnormally", "condition", cond.Type)
			i.Status.Phase = cloudv1.ImportFailed
			if i.Status.Message == "" {
				i.Status.Message = "abnormal termination, please seek support"
			}
		}
	}
	return nil
}

func (ci *ImportService) Delete(ctx context.Context, i *cloudv1.Import) (errResult error) {
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.Name,
			Namespace: ci.env.ImportJobNamespace(),
		},
	}
	err := ci.env.GetControllerClient().Delete(ctx, svc)
	if client.IgnoreNotFound(err) != nil {
		ci.logger.Error(err, "Deleting import service failed", "service name", i.Name)
		return err
	}

	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.Name,
			Namespace: ci.env.ImportJobNamespace(),
		},
	}
	background := metav1.DeletePropagationBackground
	err = ci.env.GetControllerClient().Delete(ctx, job, &client.DeleteOptions{PropagationPolicy: &background})
	if client.IgnoreNotFound(err) != nil {
		ci.logger.Error(err, "Deleting import batch job failed", "batch job name", i.Name)
		return err
	}

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.Name,
			Namespace: ci.env.ImportJobNamespace(),
		},
	}
	err = ci.env.GetControllerClient().Delete(ctx, secret)
	if client.IgnoreNotFound(err) != nil {
		ci.logger.Error(err, "Deleting import secret failed", "secret name", i.Name)
		return err
	}

	pvc := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.Name,
			Namespace: ci.env.ImportJobNamespace(),
		},
	}
	err = ci.env.GetControllerClient().Delete(ctx, pvc)
	if client.IgnoreNotFound(err) != nil {
		ci.logger.Error(err, "Deleting import pvc failed", "pvc name", i.Name)
		return err
	}

	if i.Status.TaskId == 0 || ci.env.clusterObj == nil {
		ci.logger.Info("No TaskId or cluster obj found; skipping task cancellation")
		return nil
	}
	err = ci.env.CancelLoadDataTasks(ctx, i.Status.TaskId)
	if err != nil {
		ci.logger.Error(err, "CancelLoadDataTasks failed", "task id", i.Status.TaskId)
		// This error is ignored because it does not affect the deletion flow.
		// Failing to cancel the load data task only delays resource cleanup on the LoadDataWorker side.
		return nil
	}
	return nil
}

func (ci *ImportService) getLightningConfigFromSecret(ctx context.Context, name string) (string, error) {
	secret := &corev1.Secret{}
	namespacedName := types.NamespacedName{
		Namespace: ci.env.ImportJobNamespace(),
		Name:      name,
	}
	err := ci.env.GetControllerClient().Get(ctx, namespacedName, secret)
	if err != nil {
		return "", err
	}
	bytes := secret.Data[lightningConfigFileName]
	cfg := string(bytes)
	return cfg, nil
}

func newPersistentVolumeClaim(name string, ns string,
	storageClassName string, requestLimit string) *corev1.PersistentVolumeClaim {
	v := resource.MustParse(requestLimit)

	return &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: ns,
			Labels:    lightningLabels(name),
		},
		Spec: corev1.PersistentVolumeClaimSpec{
			AccessModes: []corev1.PersistentVolumeAccessMode{
				corev1.ReadWriteOnce,
			},
			StorageClassName: pointer.String(storageClassName),
			Resources: corev1.VolumeResourceRequirements{
				Requests: corev1.ResourceList{
					corev1.ResourceStorage: v,
				},
			},
		},
	}
}

func appendVolume(job *batchv1.Job) {
	volume := corev1.Volume{
		Name: job.Name,
		VolumeSource: corev1.VolumeSource{
			PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
				ClaimName: job.Name,
				ReadOnly:  false,
			},
		},
	}
	job.Spec.Template.Spec.Volumes = append(job.Spec.Template.Spec.Volumes, volume)

	mount := corev1.VolumeMount{
		Name:      job.Name,
		MountPath: "/etc/endpoint",
	}

	for i := range job.Spec.Template.Spec.Containers {
		c := job.Spec.Template.Spec.Containers[i]
		c.VolumeMounts = append(c.VolumeMounts, mount)
		job.Spec.Template.Spec.Containers[i] = c
	}
}