package task

import (
	"context"
	"fmt"
	"math"
	"time"

	opv2 "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"gopkg.in/yaml.v2"
	scheduling_v1 "k8s.io/api/scheduling/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/utils"
	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	gitopssdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

type restoreTask struct {
	*baseTask
}

func NewRestore(ctx *Context) deploy.Task {
	return &restoreTask{
		baseTask: &baseTask{Context: ctx},
	}
}

func (t *restoreTask) Name() string {
	return "Restore"
}

func (t *restoreTask) Create(ctx context.Context) deploy.Result {
	affinity := t.cloud.GetPodAffinity()
	restore := &opv1alpha1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getRestoreName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
			Labels:    t.restore.Labels,
			Annotations: map[string]string{
				"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
			},
		},
		Spec: opv1alpha1.RestoreSpec{
			Type: opv1alpha1.BackupTypeFull,
			Mode: opv1alpha1.RestoreModeSnapshot,
			BR: &opv1alpha1.BRConfig{
				Cluster:          t.clusterNamer.GetTiDBName(),
				ClusterNamespace: t.clusterNamer.GetTiDBNamespace(),
			},
			Affinity:    affinity,
			Tolerations: utils.Toleration(cloudv1.ComponentRestore),
			UseKMS:      false,
			Env:         t.restore.Spec.Env,
		},
	}
	if t.restore.Spec.ResourceRequirements != nil {
		restore.Spec.ResourceRequirements = *t.restore.Spec.ResourceRequirements
	}

	pitrTS := ptr.Deref(t.restore.Spec.From.PiTRRestoredTS, "")
	if pitrTS != "" {
		restore.Spec.Mode = opv1alpha1.RestoreModePiTR
		restore.Spec.PitrRestoredTs = pitrTS
		// pitr storage implemented in cloud provider

		// Compaction-accelerated PiTR: when the restore is opted in via
		// Spec.CompactBackupName, configure ReplicationConfig with the sibling CompactBackup CR name.
		// WaitTimeout is read from the RestoreAnnoWaitTimeout annotation.
		compactBackupName := t.restore.Spec.CompactBackupName
		if compactBackupName != "" {
			rc := &opv1alpha1.ReplicationConfig{
				CompactBackupName: compactBackupName,
			}
			if waitTimeout, err := t.restore.WaitTimeout(); err != nil {
				return deploy.Panic(fmt.Errorf("invalid %s annotation: %w", cloudv1.RestoreAnnoWaitTimeout, err))
			} else if waitTimeout > 0 {
				rc.WaitTimeout = &metav1.Duration{Duration: waitTimeout}
			}
			restore.Spec.ReplicationConfig = rc
		}
	}

	// override image
	baseImage := gitopssdk.OverrideBaseImage(t.restore, t.restore.Spec.Configuration.BaseImage)
	restore.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, t.restore.Spec.Configuration.Tag)
	if err := configmanager.MergeBRConfig(restore.Spec.BR, []byte(t.restore.Spec.Configuration.Config)); err != nil {
		return deploy.Panic(err)
	}
	// cloud vendor specific overlay
	if err := t.cloud.Overlay(t.restore, restore); err != nil {
		return deploy.Panic(err)
	}

	// add priority class to restore pod to avoid node-pressure eviction
	// ref: https://kubernetes.io/docs/concepts/scheduling-eviction/node-pressure-eviction/#pod-selection-for-kubelet-eviction
	if err := t.ensurePriorityClass(ctx); err != nil {
		return deploy.Error(err)
	}
	restore.Spec.PriorityClassName = utils.BRPriorityClassName

	err := t.applyOpRestore(ctx, restore)
	return deploy.PanicOrDone(err)
}

func (t *restoreTask) PostCreate(ctx context.Context) deploy.Result {
	restore, err := t.getOpRestore(ctx)
	if err != nil {
		t.restore.Status.Phase = cloudv1.RestorePrepare
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}
	syncRestoreProgresses(t.restore, restore.Status.Progresses)
	mirrorReplicationStep(t.restore, restore)

	switch restore.Status.Phase {
	case opv1alpha1.RestoreComplete:
		t.restore.Status.Phase = cloudv1.RestoreComplete
		return deploy.Done()
	case opv1alpha1.RestoreFailed, opv1alpha1.RestoreInvalid:
		t.restore.Status.Phase = cloudv1.RestoreFailed
		t.logger.Info("restore failed", "conditions", restore.Status.Conditions)
		return deploy.Done()
	case opv1alpha1.RestoreScheduled:
		t.restore.Status.Phase = cloudv1.RestoreScheduled
	case opv1alpha1.RestoreRetryFailed:
		t.restore.Status.Phase = cloudv1.RestorePrepare // retryable error as prepare
	case opv1alpha1.RestoreRunning:
		t.restore.Status.Phase = cloudv1.RestoreRunning
	default: // empty phase as prepare
		t.restore.Status.Phase = cloudv1.RestorePrepare
	}
	return deploy.Error(fmt.Errorf("restore is unfinished")).WithRequeuAfter(time.Second * 30)
}

func (t *restoreTask) Delete(ctx context.Context) deploy.Result {
	if t.restore.Status.Phase == cloudv1.RestoreFailed {
		// Ignore deleting dataplane Restore CR for troubleshooting when restore failed
		return deploy.Done()
	}
	restore := &opv1alpha1.Restore{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getRestoreName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
		},
	}
	err := client.IgnoreNotFound(t.deleteOpRestore(ctx, restore))
	return deploy.ErrorOrDone(err)
}

func (t *restoreTask) ensurePriorityClass(ctx context.Context) error {
	priorityClass := &scheduling_v1.PriorityClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: utils.BRPriorityClassName,
		},
		Value:         utils.BRPriorityClassValue,
		GlobalDefault: false,
		Description:   "Priority class for backup and restore",
	}
	if err := k8s.CreateIfNotExist(ctx, t.remote, priorityClass); err != nil {
		return fmt.Errorf("fail to create priority class: %w", err)
	}
	return nil
}

func (t *restoreTask) applyOpRestore(ctx context.Context, tidbRestore *opv1alpha1.Restore) error {
	if t.cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		// convert restore CR from v1 to v2. As the CRD definition is similar, we can do this.
		data, err := yaml.Marshal(tidbRestore)
		if err != nil {
			return fmt.Errorf("marshal tidb restore to yaml error: %w", err)
		}
		tidbRestoreV2 := &opv2.Restore{}
		if err := yaml.Unmarshal(data, tidbRestoreV2); err != nil {
			return fmt.Errorf("unmarshal tidb restore to v2 error: %w", err)
		}
		return k8s.Apply(ctx, t.remote, tidbRestoreV2)
	}

	return k8s.Apply(ctx, t.remote, tidbRestore)
}

func (t *restoreTask) getOpRestore(ctx context.Context) (*opv1alpha1.Restore, error) {
	key := types.NamespacedName{
		Name:      t.getRestoreName(),
		Namespace: t.clusterNamer.GetTiDBNamespace(),
	}
	if t.cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		tidbRestoreV2 := opv2.Restore{}
		if err := t.remote.Get(ctx, key, &tidbRestoreV2); err != nil {
			return nil, err
		}
		// convert restore CR from v2 to v1. As the CRD definition is similar, we can do this.
		data, err := yaml.Marshal(tidbRestoreV2)
		if err != nil {
			return nil, fmt.Errorf("marshal tidb restore v2 to yaml error: %w", err)
		}
		tidbRestoreV1 := &opv1alpha1.Restore{}
		if err := yaml.Unmarshal(data, tidbRestoreV1); err != nil {
			return nil, fmt.Errorf("unmarshal tidb restore to v1 error: %w", err)
		}
		return tidbRestoreV1, nil
	}

	tidbRestore := opv1alpha1.Restore{}
	if err := t.remote.Get(ctx, key, &tidbRestore); err != nil {
		return nil, err
	}
	return &tidbRestore, nil
}

func (t *restoreTask) deleteOpRestore(ctx context.Context, tidbRestore *opv1alpha1.Restore) error {
	if t.cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		tidbRestoreV2 := &opv2.Restore{
			ObjectMeta: metav1.ObjectMeta{
				Name:      tidbRestore.Name,
				Namespace: tidbRestore.Namespace,
			},
		}
		return t.remote.Delete(ctx, tidbRestoreV2)
	}
	return t.remote.Delete(ctx, tidbRestore)
}

func syncRestoreProgresses(restore *cloudv1.Restore, tidbRestoreProgresses []opv1alpha1.Progress) {
	// Strip any trailing synthetic Replication:* entry before resizing so that
	// it survives the truncation below. mirrorReplicationStep re-appends it and
	// checks for changes, keeping LastTransitionTime stable across reconciles.
	var synthetic *cloudv1.RestoreProgress
	if n := len(restore.Status.Progresses); n > 0 && hasReplicationStepPrefix(restore.Status.Progresses[n-1].Step) {
		s := restore.Status.Progresses[n-1]
		synthetic = &s
		restore.Status.Progresses = restore.Status.Progresses[:n-1]
	}

	i := 0
	if len(restore.Status.Progresses) > len(tidbRestoreProgresses) {
		restore.Status.Progresses = restore.Status.Progresses[:len(tidbRestoreProgresses)]
	}
	for ; i < len(restore.Status.Progresses); i++ {
		if restore.Status.Progresses[i].LastTransitionTime.Equal(&tidbRestoreProgresses[i].LastTransitionTime) {
			continue
		}
		restore.Status.Progresses[i].Step = tidbRestoreProgresses[i].Step
		restore.Status.Progresses[i].Progress = convertProgress(tidbRestoreProgresses[i].Progress)
		restore.Status.Progresses[i].LastTransitionTime = tidbRestoreProgresses[i].LastTransitionTime
	}
	for ; i < len(tidbRestoreProgresses); i++ {
		restore.Status.Progresses = append(restore.Status.Progresses, cloudv1.RestoreProgress{
			Step:               tidbRestoreProgresses[i].Step,
			Progress:           convertProgress(tidbRestoreProgresses[i].Progress),
			LastTransitionTime: tidbRestoreProgresses[i].LastTransitionTime,
		})
	}

	// Re-attach the synthetic entry so mirrorReplicationStep finds it and
	// avoids resetting LastTransitionTime when the step is unchanged.
	if synthetic != nil {
		restore.Status.Progresses = append(restore.Status.Progresses, *synthetic)
	}
}

func convertProgress(tidbProgress float64) int32 {
	return int32(math.Floor(tidbProgress))
}

// replicationStepProgressPrefix tags the synthetic Progress entry mirrored
// from opv1alpha1.RestoreStatus.ReplicationStep so callers can identify
// the compaction-restore phase signal in cloudv1.Restore.Status.Progresses
// without inventing a new top-level field.
const replicationStepProgressPrefix = "Replication:"

// mirrorReplicationStep surfaces opv1alpha1.RestoreStatus.ReplicationStep
// onto cloudv1.Restore.Status.Progresses as a single trailing synthetic
// entry. A new entry replaces the previous synthetic one in place rather
// than accumulating one entry per phase transition.
func mirrorReplicationStep(cloudRestore *cloudv1.Restore, opRestore *opv1alpha1.Restore) {
	step := opRestore.Status.ReplicationStep
	if step == "" {
		return
	}
	entry := cloudv1.RestoreProgress{
		Step:               replicationStepProgressPrefix + step,
		Progress:           0,
		LastTransitionTime: metav1.Time{Time: time.Now()},
	}
	n := len(cloudRestore.Status.Progresses)
	if n > 0 && hasReplicationStepPrefix(cloudRestore.Status.Progresses[n-1].Step) {
		// Replace the existing synthetic entry only if the step value
		// changed; this keeps LastTransitionTime stable across reconciles
		// and lets external observers diff phase transitions.
		if cloudRestore.Status.Progresses[n-1].Step == entry.Step {
			return
		}
		cloudRestore.Status.Progresses[n-1] = entry
		return
	}
	cloudRestore.Status.Progresses = append(cloudRestore.Status.Progresses, entry)
}

func hasReplicationStepPrefix(step string) bool {
	return len(step) >= len(replicationStepProgressPrefix) &&
		step[:len(replicationStepProgressPrefix)] == replicationStepProgressPrefix
}