package task

import (
	"context"
	"encoding/json"
	"fmt"
	"regexp"
	"strings"
	"time"
	"unicode/utf8"

	opv2 "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"sigs.k8s.io/yaml"

	corev1 "k8s.io/api/core/v1"
	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"
)

const maxBackupFailedErrorAnnotationBytes = 4000

var quotedBRErrorPattern = regexp.MustCompile(`error="([^"]+)"`)

type backupTask struct {
	*baseTask
}

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

func (t *backupTask) Name() string {
	return "Backup"
}

func (t *backupTask) Create(ctx context.Context) deploy.Result {
	// make the backup pod use the desired nodepool
	affinity := t.cloud.GetPodAffinity()
	backup := &opv1alpha1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getBackupName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
			Labels:    t.backup.Labels,
			Annotations: map[string]string{
				"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
			},
		},
		Spec: opv1alpha1.BackupSpec{
			Type:        opv1alpha1.BackupTypeFull,
			TableFilter: t.backup.Spec.TableFilter,
			// We always use Retain policy for tidb-operator and let dataflow-svc to clean data
			CleanPolicy: opv1alpha1.CleanPolicyTypeRetain,
			CommitTs:    ptr.Deref(t.backup.Spec.CommitTS, ""),
			Mode:        opv1alpha1.BackupMode(*t.backup.Spec.Mode),
			BR: &opv1alpha1.BRConfig{
				Cluster:          t.clusterNamer.GetTiDBName(),
				ClusterNamespace: t.clusterNamer.GetTiDBNamespace(),
				StatusAddr:       "0.0.0.0:8089",
			},
			Affinity:    affinity,
			Tolerations: utils.Toleration(cloudv1.ComponentBackup),
			UseKMS:      false,
			Env:         t.backup.Spec.Env,
		},
	}
	if t.backup.Spec.ResourceRequirements != nil {
		backup.Spec.ResourceRequirements = *t.backup.Spec.ResourceRequirements
	}

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

	// fix log backup does not need this param, or it will cause log backup failed
	// refer to: https://github.com/tidbcloud/infra-provider/blob/c2e1ded28a51e3aefa97335e942d85252eb3b75a/cluster-controller/controllers/br/backup_controller.go#L366-L367
	if t.backup.Spec.Mode != nil && cloudv1.BackupModeLog == *t.backup.Spec.Mode {
		backup.Spec.BR.Concurrency = nil
	}

	// log backup need to update the log truncate until and log stop to the dataplane backup CR
	if err := t.syncLogBackupSpec(backup); err != nil {
		return deploy.Error(err)
	}

	// add priority class to backup 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)
	}
	backup.Spec.PriorityClassName = utils.BRPriorityClassName

	err := t.applyOpBackup(ctx, backup)
	return deploy.PanicOrDone(err)
}

func (t *backupTask) PostCreate(ctx context.Context) deploy.Result {
	backup, err := t.getOpBackup(ctx)
	if err != nil {
		t.backup.Status.Phase = cloudv1.BackupPrepare
		return deploy.Panic(err).WithRequeuAfter(time.Second * 30)
	}

	// status sync
	t.backup.Status.Phase = convertBackupPhase(backup.Status.Phase)
	if backup.Status.Phase == opv1alpha1.BackupFailed {
		if errMsg := extractBackupFailedErrorForAnnotation(backup); errMsg != "" {
			if t.backup.Annotations == nil {
				t.backup.Annotations = make(map[string]string)
			}
			t.backup.Annotations[cloudv1.BackupAnnotationKeyLastFailedError] = errMsg
		}
		t.logger.Info("backup failed", "conditions", backup.Status.Conditions)
		return deploy.Done()
	}

	t.backup.Status.BackupPath = backup.Status.BackupPath
	t.backup.Status.BackupSize = backup.Status.BackupSize
	t.backup.Status.BackupTS = backup.Status.CommitTs
	t.backup.Status.LogCheckpointTs = backup.Status.LogCheckpointTs
	t.backup.Status.LogTruncateUntil = backup.Status.LogSuccessTruncateUntil
	if err := t.cloud.OverlayStatus(t.backup, backup); err != nil {
		return deploy.Error(err)
	}

	if t.backup.Spec.Mode != nil && *t.backup.Spec.Mode == cloudv1.BackupModeLog {
		// sync log sub command statuses
		bytes, err := json.Marshal(backup.Status.LogSubCommandStatuses)
		if err != nil {
			return deploy.Error(fmt.Errorf("marshal log sub command status: %w", err))
		}
		if err := json.Unmarshal(bytes, &t.backup.Status.LogSubCommandStatuses); err != nil {
			return deploy.Error(fmt.Errorf("unmarshal log sub command status: %w", err))
		}
	} else if t.backup.Spec.Mode != nil && *t.backup.Spec.Mode == cloudv1.BackupModeSnapshot {
		switch backup.Status.Phase {
		case opv1alpha1.BackupComplete:
			return deploy.Done()
		default: // we treat all other status as running
			t.backup.Status.Phase = cloudv1.BackupRunning
			return deploy.Error(fmt.Errorf("backup is running")).WithRequeuAfter(time.Second * 30)
		}
	}

	return deploy.Error(fmt.Errorf("backup is unfinished")).WithRequeuAfter(time.Second * 30)
}

func convertBackupPhase(conditionType opv1alpha1.BackupConditionType) cloudv1.BackupConditionType {
	switch conditionType {
	case opv1alpha1.BackupScheduled:
		return cloudv1.BackupScheduled
	case opv1alpha1.BackupPrepare:
		return cloudv1.BackupPrepare
	case opv1alpha1.BackupRunning:
		return cloudv1.BackupRunning
	case opv1alpha1.BackupComplete:
		return cloudv1.BackupComplete
	case opv1alpha1.BackupClean:
		return cloudv1.BackupClean
	case opv1alpha1.BackupStopped:
		return cloudv1.BackupStopped
	case opv1alpha1.BackupFailed, opv1alpha1.BackupInvalid:
		return cloudv1.BackupFailed
	case opv1alpha1.BackupRetryTheFailed:
		return cloudv1.BackupPrepare // retryable error as prepare
	default:
		return cloudv1.BackupPrepare // empty phase as prepare
	}
}

func (t *backupTask) Delete(ctx context.Context) deploy.Result {
	if t.backup.Status.Phase == cloudv1.BackupFailed {
		// Ignore deleting dataplane Backup CR for troubleshooting when backup failed
		return deploy.Done()
	}
	backup := &opv1alpha1.Backup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      t.getBackupName(),
			Namespace: t.clusterNamer.GetTiDBNamespace(),
		},
	}
	err := client.IgnoreNotFound(t.deleteOpBackup(ctx, backup))
	return deploy.ErrorOrDone(err)
}

func (t *backupTask) 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 *backupTask) applyOpBackup(ctx context.Context, tidbBackup *opv1alpha1.Backup) error {
	if t.cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		// convert backup CR from v1 to v2. As the CRD definition is similar, we can do this.
		data, err := yaml.Marshal(tidbBackup)
		if err != nil {
			return fmt.Errorf("marshal tidb backup to yaml error: %w", err)
		}
		tidbBackupV2 := &opv2.Backup{}
		if err := yaml.Unmarshal(data, tidbBackupV2); err != nil {
			return fmt.Errorf("unmarshal tidb backup to v2 error: %w", err)
		}
		return k8s.Apply(ctx, t.remote, tidbBackupV2)
	}

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

func extractBackupFailedErrorForAnnotation(backup *opv1alpha1.Backup) string {
	if backup == nil {
		return ""
	}
	if errMsg := extractBackupFailedErrorFromConditions(backup.Status.Conditions); errMsg != "" {
		return errMsg
	}
	for _, subStatus := range backup.Status.LogSubCommandStatuses {
		if errMsg := extractBackupFailedErrorFromConditions(subStatus.Conditions); errMsg != "" {
			return errMsg
		}
	}
	return ""
}

func extractBackupFailedErrorFromConditions(conditions []opv1alpha1.BackupCondition) string {
	for i := len(conditions) - 1; i >= 0; i-- {
		condition := conditions[i]
		if condition.Type != opv1alpha1.BackupFailed && condition.Type != opv1alpha1.BackupInvalid {
			continue
		}
		if condition.Status != corev1.ConditionTrue && condition.Status != "" {
			continue
		}
		return buildStoredBackupError(condition.Reason, sanitizeBackupErrorMessage(condition.Message))
	}
	return ""
}

func buildStoredBackupError(reason, message string) string {
	reason = strings.TrimSpace(reason)
	message = strings.TrimSpace(message)
	parts := make([]string, 0, 2)
	if reason != "" {
		parts = append(parts, "reason="+reason)
	}
	if message != "" {
		parts = append(parts, "message="+message)
	}
	if len(parts) == 0 {
		return ""
	}
	return truncateToMaxBytes(strings.Join(parts, "; "), maxBackupFailedErrorAnnotationBytes)
}

func sanitizeBackupErrorMessage(raw string) string {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return ""
	}
	if idx := strings.LastIndex(raw, "Error:"); idx >= 0 {
		candidate := strings.TrimSpace(raw[idx+len("Error:"):])
		if candidate != "" {
			return candidate
		}
	}
	if match := quotedBRErrorPattern.FindStringSubmatch(raw); len(match) == 2 {
		return strings.TrimSpace(match[1])
	}
	return raw
}

func truncateToMaxBytes(s string, maxBytes int) string {
	if len(s) <= maxBytes {
		return s
	}
	for len(s) > maxBytes {
		_, size := utf8.DecodeLastRuneInString(s)
		if size <= 0 {
			break
		}
		s = s[:len(s)-size]
	}
	return strings.TrimSpace(s)
}

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

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

func (t *backupTask) deleteOpBackup(ctx context.Context, tidbBackup *opv1alpha1.Backup) error {
	if t.cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		tidbBackupV2 := &opv2.Backup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      tidbBackup.Name,
				Namespace: tidbBackup.Namespace,
			},
		}
		return t.remote.Delete(ctx, tidbBackupV2)
	}
	return t.remote.Delete(ctx, tidbBackup)
}

func (t *backupTask) syncLogBackupSpec(tidbBackup *opv1alpha1.Backup) error {
	if t.backup.Spec.LogTruncateUntil != nil && tidbBackup.Spec.LogTruncateUntil != *t.backup.Spec.LogTruncateUntil {
		tidbBackup.Spec.LogTruncateUntil = *t.backup.Spec.LogTruncateUntil
	}
	if t.backup.Spec.LogStop != nil && tidbBackup.Spec.LogStop != *t.backup.Spec.LogStop {
		tidbBackup.Spec.LogStop = *t.backup.Spec.LogStop
	}

	return nil
}