package export

import (
	"context"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	clientv3 "go.etcd.io/etcd/client/v3"
	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"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)

func (r *reconcileContext) cleanExport() error {
	r.Info("clean export")
	if r.export.Annotations[types.AnnoExportDumplingJob] != "" {
		if err := r.tryCleanDumplingJob(r.ctx); err != nil {
			return errors.Annotate(err, "error clean dumpling job")
		}
	}

	if r.export.Status.FinishedSize != "" && r.export.Annotations[types.AnnoExportSendToMetering] == "" {
		err := r.sendToMetering(r.export.Status.FinishedSize, r.export.Status.CompressedSize)
		if err != nil {
			r.Error(err, "Failed to send to metering")
			return err
		}
		r.export.Annotations[types.AnnoExportSendToMetering] = "true"
		return r.InfraClient.Update(r.ctx, r.export)
	}

	if r.export.Annotations[types.AnnoExportFinishTime] == "" {
		r.export.Annotations[types.AnnoExportFinishTime] = time.Now().Format(time.RFC3339)
		return r.InfraClient.Update(r.ctx, r.export)
	} else {
		finishTimeStr := r.export.Annotations[types.AnnoExportFinishTime]
		finishTime, _ := time.Parse(time.RFC3339, finishTimeStr)
		// wait at least 1 minute because vm-agent scrapes interval is 1 minute
		sleepTime := 1*time.Minute - time.Since(finishTime)
		if sleepTime > 0 {
			return &utils.RequeueAfterError{
				Cause:        errors.New("waiting for export metrics to be scraped"),
				RequeueAfter: sleepTime,
			}
		}
	}

	if r.export.Annotations[types.AnnoExportEtcdKey] != "" {
		if err := r.tryDeleteEtcdToken(); err != nil {
			return errors.Annotate(err, "error delete etcd token")
		}
	}

	if r.export.Annotations[types.AnnoExportResourceGroup] != "" {
		if err := r.tryDeleteResourceGroup(); err != nil {
			return errors.Annotate(err, "error delete resource group")
		}
	}

	return nil
}

func (r *reconcileContext) tryCleanDumplingJob(ctx context.Context) error {
	// Step 1: delete job
	err := r.Client.Delete(ctx, &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      r.BuildDumplingJobName(),
			Namespace: r.ExportJobNS,
		},
	}, client.PropagationPolicy(metav1.DeletePropagationBackground))

	if err != nil && !apierrors.IsNotFound(err) {
		return err
	}

	// Step 2: get pod termination message
	pod, err := r.getJobPod()
	if err != nil {
		if apierrors.IsNotFound(err) {
			r.Info("Pod not found for job", "job", r.BuildDumplingJobName())
			return nil
		}
		r.Error(err, "Failed to get job pod")
		return err
	}
	if r.export.Status.Phase != cloudv1.ExportPhaseFailed && r.export.Status.FinishedSize == "" {
		terminationMessage := ""
		maxWaitTime := 30 * time.Second // value as same as TerminationGracePeriodSeconds
		pollInterval := 1 * time.Second
		startTime := time.Now()
		for time.Since(startTime) < maxWaitTime {
			err := r.Client.Get(ctx, client.ObjectKey{Name: pod.Name, Namespace: pod.Namespace}, pod)
			if err != nil {
				r.Error(err, "Failed to get pod")
				return err
			}
			for _, cs := range pod.Status.ContainerStatuses {
				if cs.Name == DumplingContainerName && cs.State.Terminated != nil {
					if cs.State.Terminated.Message != "" {
						terminationMessage = cs.State.Terminated.Message
						finishedSize, compressedSize, err := extractExportSize(terminationMessage)
						if err != nil {
							r.Error(err, "failed to extract export size")
							metrics.FailedGetExportSizeCounter.WithLabelValues(r.export.Spec.ClusterName, r.export.Name, strconv.Itoa(int(r.keyspaceID))).Inc()
							return err
						}
						r.export.Status.FinishedSize = finishedSize
						r.export.Status.CompressedSize = compressedSize

						err = r.InfraClient.Status().Update(r.ctx, r.export)
						if err != nil {
							r.Error(err, "Failed to update export status with size")
							return err
						}
						break
					}
				}
			}
			if terminationMessage != "" {
				break
			}
			time.Sleep(pollInterval)
		}
		if terminationMessage == "" {
			r.Error(errors.New("empty termination message"), "Failed to get termination message from pod after timeout", "job", r.BuildDumplingJobName())
			metrics.FailedGetExportSizeCounter.WithLabelValues(r.export.Spec.ClusterName, r.export.Name, strconv.Itoa(int(r.keyspaceID))).Inc()
		}
	}

	// Step 3: remove finalizer for pod(failed, succeeded, timeout, canceled)
	err = r.removePodFinalizer(ctx, pod)
	if err != nil {
		r.Error(err, "Failed to remove finalizers from pod")
		return err
	}

	return nil
}

func (r *reconcileContext) tryDeleteEtcdToken() error {
	if r.keyspaceToken == "" {
		return nil
	}

	exportType := r.export.Annotations[types.AnnoExportType]
	exportPoolNs := ExportTiDBPoolNS
	if exportType == string(ExportTypeVIP) {
		exportPoolNs = ExportVipTiDBPoolNS
	} else if exportType == string(ExportTypeSuperVIP) {
		exportPoolNs = ExportSuperVipTiDBPoolNS
	}
	etcdTokenPrefix := r.EtcdTokenPrefixMap[exportPoolNs]

	token := r.keyspaceToken + ":" + r.export.Name
	_, err := r.EtcdKV.Delete(r.ctx, etcdTokenPrefix+token+"/", clientv3.WithPrefix())
	return err
}

func (r *reconcileContext) tryDeleteResourceGroup() error {
	name := r.export.Annotations[types.AnnoExportResourceGroup]
	return r.PdClient.DeleteResourceGroup(r.ctx, name)
}

func (r *reconcileContext) removePodFinalizer(ctx context.Context, pod *corev1.Pod) error {
	copyPod := pod.DeepCopy()
	controllerutil.RemoveFinalizer(copyPod, exportFinalizer)

	err := r.Client.Update(ctx, copyPod)
	if err != nil {
		return err
	}
	return nil
}