package br

import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"math"
	"net/url"
	"path"
	"strings"
	"time"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	"github.com/prometheus/client_golang/prometheus"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	clusteragentv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	templateversion "github.com/tidbcloud/cluster-service-ng/pkg/template/version"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const ExecuteClassicRestoreActionName = "execute-classic-restore"

type executeClassicRestoreAction struct {
	dataplaneManager    dataplane.Manager
	sharedPoolRepo      repo.SharedPoolRepo
	classicRestoreRepo  repo.ClassicRestoreRepo
	templateVersionSvc  *templateversion.Service
	bizConfig           *config.BizConfig
	imageConfig         *config.ImageConfig
	dpServiceCli        dplanemgrv1.DataPlaneServiceClient
	clusterConfigClient clusterconfig.Client
	clustergatewayCli   clustergateway.Client
}

func NewExecuteClassicRestoreAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneManager dataplane.Manager,
	classicRestoreRepo repo.ClassicRestoreRepo,
	templateVersionSvc *templateversion.Service,
	bizConfig *config.BizConfig,
	imageConfig *config.ImageConfig,
	dpServiceCli dplanemgrv1.DataPlaneServiceClient,
	clusterConfigClient clusterconfig.Client,
	clustergatewayCli clustergateway.Client,
) actions.Action {
	return actions.CreateActionFromAsync(&executeClassicRestoreAction{
		sharedPoolRepo:      sharedPoolRepo,
		classicRestoreRepo:  classicRestoreRepo,
		dataplaneManager:    dataplaneManager,
		templateVersionSvc:  templateVersionSvc,
		bizConfig:           bizConfig,
		imageConfig:         imageConfig,
		dpServiceCli:        dpServiceCli,
		clusterConfigClient: clusterConfigClient,
		clustergatewayCli:   clustergatewayCli,
	}, time.Minute, time.Hour*4)
}

func (a *executeClassicRestoreAction) Name() string {
	return ExecuteClassicRestoreActionName
}

func (a *executeClassicRestoreAction) Start(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, fmt.Sprintf("start %s action...", a.Name()))

	targetSharedPool, remoteCli, restore, err := a.getCreateRestoreContextInfos(args)
	if err != nil {
		tilog.Error(ctx, "failed to get create classic restore context infos", zap.Error(err))
		return fmt.Errorf("failed to get create classic restore context infos: %w", err)
	}

	err = AddActionForClassicRestore(ctx, a.classicRestoreRepo, restore.RestoreID, schema.ClassicRestoreActionExecRestore)
	if err != nil {
		tilog.Error(ctx, "failed to add action to restore", zap.Error(err))
		return fmt.Errorf("failed to add action to restore: %w", err)
	}

	err = a.initServiceAccountIfNeeded(ctx, targetSharedPool, remoteCli)
	if err != nil {
		tilog.Error(ctx, "failed to init service account", zap.Error(err))
		return fmt.Errorf("failed to init service account: %w", err)
	}

	err = a.initAKSKSecretIfNeeded(ctx, targetSharedPool, remoteCli, restore)
	if err != nil {
		tilog.Error(ctx, "failed to init AK/SK secret", zap.Error(err))
		return fmt.Errorf("failed to init AK/SK secret: %w", err)
	}

	tenant, err := getTenantTagValueOfDataplane(ctx, a.dpServiceCli, targetSharedPool.GetDataplaneName())
	if err != nil {
		return fmt.Errorf("failed to get tenant tag value of dataplane: %w", err)
	}

	image, options, err := a.getImageAndOptions(ctx, targetSharedPool, restore, tenant)
	if err != nil {
		return fmt.Errorf("failed to get image and options for classic restore: %w", err)
	}

	yamlStr, err := a.renderRestoreCR(ctx, targetSharedPool, restore, args, tenant, image, options)
	if err != nil {
		tilog.Error(ctx, "failed to render restore CR yaml", zap.Error(err))
		return fmt.Errorf("failed to render restore CR yaml: %s", err.Error())
	}

	resourceCr, _, err := k8s.GenerateUnstructuredFromYaml(yamlStr)
	if err != nil {
		tilog.Error(ctx, "failed to generate unstructured from yaml", zap.String("yaml", yamlStr), zap.Error(err))
		return fmt.Errorf("failed to generate unstructured from yaml: %s", err.Error())
	}

	err = k8s.CreateIfNotExist(ctx, remoteCli, resourceCr)
	if err != nil {
		tilog.Error(ctx, "failed to create restore", zap.String("yaml", yamlStr), zap.Error(err))
		return fmt.Errorf("failed to create restore: %s", err.Error())
	}

	tilog.Info(ctx, " create restore success", zap.String("yaml", yamlStr))
	return nil
}

func (a *executeClassicRestoreAction) CheckFinish(args actions.ActionArgs) (bool, error) {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, fmt.Sprintf("check %s action finish...", a.Name()))

	targetSharedPool, remoteCli, restore, err := a.getCreateRestoreContextInfos(args)
	if err != nil {
		tilog.Error(ctx, "failed to get create classic restore context infos", zap.Error(err))
		return false, fmt.Errorf("failed to get create classic restore context infos: %w", err)
	}

	restoreCR := &v1alpha1br.Restore{}
	restoreName := tioperator.ClassicRestoreCRName(restore.RestoreID)
	namespace := targetSharedPool.GetDataPlaneNamespace()
	err = remoteCli.Get(ctx, types.NamespacedName{
		Name:      restoreName,
		Namespace: namespace,
	}, restoreCR)
	if err != nil {
		tilog.Error(ctx, "failed to get restore", zap.String("name", restoreName), zap.String("namespace", namespace), zap.Error(err))
		return false, fmt.Errorf("failed to get restore: %w", err)
	}

	if restoreCR.Status.Phase == v1alpha1br.RestoreComplete {
		tilog.Info(ctx, "classic restore complete", zap.String("name", restoreName), zap.String("namespace", namespace))
		return true, nil
	}

	if restoreCR.Status.Phase == v1alpha1br.RestoreFailed || restoreCR.Status.Phase == v1alpha1br.RestoreInvalid {
		tilog.Error(ctx, "classic restore failed or invalid", zap.String("name", restoreName), zap.String("namespace", namespace), zap.String("phase", string(restoreCR.Status.Phase)))
		ClassicRestoreUnretryableErrorTotal.With(prometheus.Labels{
			LabelRestoreId: restore.RestoreID,
			LabelErrorType: ErrorTypeRestore,
		}).Inc()
		message := fmt.Sprintf("restore failed or invalid: %s", restoreCR.Status.Phase)
		if err := a.classicRestoreRepo.SetStatusWithFailure(ctx, restore.RestoreID, schema.ClassicRestoreStatusFailed,
			kernelRestoreFailure("KERNEL_TASK_FAILED", message)); err != nil {
			return false, fmt.Errorf("failed to mark classic restore failed: %w", err)
		}
		return false, actions.NewTerminalError(errors.New(message))
	}

	return false, nil
}

func (a *executeClassicRestoreAction) getCreateRestoreContextInfos(args actions.ActionArgs) (targetSharedPool *entity.SharedPool, remoteCli client.Client, restore *entity.ClassicRestore, err error) {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "get create classic restore context infos...")
	restoreID, _ := GetRestoreID(args)
	restore, err = a.classicRestoreRepo.Load(ctx, restoreID, false)
	if err != nil {
		err = fmt.Errorf("failed to load classic restore: %w", err)
		return
	}
	targetSharedPool, err = a.sharedPoolRepo.Load(ctx, restore.TargetSharedPoolID, false)
	if err != nil {
		err = fmt.Errorf("failed to load target shared pool: %w", err)
		return
	}
	remoteCli, err = a.dataplaneManager.GetClient(targetSharedPool.GetDataplaneName())
	if err != nil {
		err = fmt.Errorf("failed to get remote client for target shared pool: %w", err)
		return
	}
	tilog.Info(ctx, "get create classic restore context infos success", zap.String("restoreID", restoreID))
	return targetSharedPool, remoteCli, restore, nil
}

func (a *executeClassicRestoreAction) renderRestoreCR(ctx context.Context, targetSharedPool *entity.SharedPool, restore *entity.ClassicRestore, args actions.ActionArgs, tenant, image string, options map[string]string) (string, error) {
	restoreName := tioperator.ClassicRestoreCRName(restore.RestoreID)

	cpu, memory, err := getResourcesForClassicRestore(args)
	if err != nil {
		return "", fmt.Errorf("failed to get resources for classic restore: %w", err)
	}

	bucketURL, err := url.Parse(restore.SourceBackupStorage)
	if err != nil {
		return "", fmt.Errorf("failed to parse backup storage URL: %w", err)
	}

	s3Compatible := restore.BackupCloudProvider == "aws" || restore.BackupCloudProvider == "alicloud"

	provider := "aws"
	if restore.BackupCloudProvider == "alicloud" {
		provider = "alibaba"
	}

	endpoint := ""
	if restore.BackupCloudProvider == "alicloud" {
		if strings.EqualFold(restore.SourceRegion, targetSharedPool.GetObjectStorage().GetRegion()) {
			endpoint = fmt.Sprintf("https://oss-%s-internal.aliyuncs.com", restore.SourceRegion)
		} else { // cross region, use public endpoint
			endpoint = fmt.Sprintf("https://oss-%s.aliyuncs.com", restore.SourceRegion)
		}
	}

	secretName := ""
	if restore.ClassicBackupAuth != nil {
		secretName = classicRestoreSecretName(restore.RestoreID)
	}
	if restore.SourceRoleARN != "" {
		options["s3.role-arn"] = restore.SourceRoleARN
		if restore.SourceRoleExternalID != "" {
			options["s3.external-id"] = restore.SourceRoleExternalID
		}
	}

	tplArgs := map[string]string{
		"Name":           restoreName,
		"Namespace":      targetSharedPool.GetDataPlaneNamespace(),
		"ResourceCPU":    cpu,
		"ResourceMemory": memory,
		"GOMEMLIMIT":     halfMemory(memory),
		"Image":          image,
		"Options":        genOptionsStr(options),

		"ForS3Compatible": fmt.Sprintf("%t", s3Compatible),
		"Bucket":          bucketURL.Hostname(),
		"Prefix":          bucketURL.Path,
		"Region":          restore.SourceRegion,
		"Provider":        provider,
		"Endpoint":        endpoint,
		"SecretName":      secretName,
		"SendCredToTikv":  fmt.Sprintf("%t", restore.ClassicBackupAuth != nil),

		"Tenant":          tenant,
		"ResourcePool":    targetSharedPool.GetId(),
		"TargetClusterID": restore.TargetClusterID,
	}

	yamlStr, _, _, err := templateversion.RenderWithVersion(ctx, a.templateVersionSvc, "classic_restore", "", tplArgs)
	if err != nil {
		return "", fmt.Errorf("failed to render template: %w", err)
	}
	return yamlStr, nil
}

func halfMemory(memory string) string {
	memoryInMiB, _ := ParseToMiB(memory)
	return fmt.Sprintf("%dMiB", memoryInMiB/2)
}

func (a *executeClassicRestoreAction) initServiceAccountIfNeeded(ctx context.Context, targetSharedPool *entity.SharedPool, remoteCli client.Client) error {
	// common args
	tplArgs := map[string]string{
		"ClusterId": targetSharedPool.GetId(),
	}

	// create classic restore service account
	saYamlStr, _, _, err := templateversion.RenderWithVersion(ctx, a.templateVersionSvc, "classic_restore_sa", "", tplArgs)
	if err != nil {
		return fmt.Errorf("failed to render classic_restore_sa template: %w", err)
	}
	saCr, _, err := k8s.GenerateUnstructuredFromYaml(saYamlStr)
	if err != nil {
		tilog.Error(ctx, "failed to generate unstructured from sa yaml", zap.String("yaml", saYamlStr), zap.Error(err))
		return fmt.Errorf("failed to generate unstructured from sa yaml: %s", err.Error())
	}
	err = k8s.CreateIfNotExist(ctx, remoteCli, saCr)
	if err != nil {
		tilog.Error(ctx, "failed to create classic restore service account", zap.String("yaml", saYamlStr), zap.Error(err))
		return fmt.Errorf("failed to create classic restore service account: %s", err.Error())
	}

	// create classic restore role binding
	rbYamlStr, _, _, err := templateversion.RenderWithVersion(ctx, a.templateVersionSvc, "classic_restore_rolebinding", "", tplArgs)
	if err != nil {
		return fmt.Errorf("failed to render classic_restore_rolebinding template: %w", err)
	}
	rbCr, _, err := k8s.GenerateUnstructuredFromYaml(rbYamlStr)
	if err != nil {
		tilog.Error(ctx, "failed to generate unstructured from classic_restore_rolebinding yaml", zap.String("yaml", rbYamlStr), zap.Error(err))
		return fmt.Errorf("failed to generate unstructured from classic_restore_rolebinding yaml: %s", err.Error())
	}
	err = k8s.CreateIfNotExist(ctx, remoteCli, rbCr)
	if err != nil {
		tilog.Error(ctx, "failed to create classic restore role binding", zap.String("yaml", rbYamlStr), zap.Error(err))
		return fmt.Errorf("failed to create classic restore role binding: %s", err.Error())
	}

	return nil
}

func (a *executeClassicRestoreAction) initAKSKSecretIfNeeded(ctx context.Context, targetSharedPool *entity.SharedPool, remoteCli client.Client, restore *entity.ClassicRestore) error {
	if restore.ClassicBackupAuth == nil {
		tilog.Info(ctx, "no classic backup auth found, skip init AKSK secret", zap.String("restoreID", restore.RestoreID))
		return nil
	}
	accessKeyId, err := base64.StdEncoding.DecodeString(restore.ClassicBackupAuth.AccessKeyId)
	if err != nil {
		return fmt.Errorf("failed to decode access key id: %w", err)
	}
	accessKeySecret, err := base64.StdEncoding.DecodeString(restore.ClassicBackupAuth.AccessKeySecret)
	if err != nil {
		return fmt.Errorf("failed to decode access key secret: %w", err)
	}

	secretName := classicRestoreSecretName(restore.RestoreID)
	secret := &corev1.Secret{
		Type: corev1.SecretTypeOpaque,
		ObjectMeta: metav1.ObjectMeta{
			Name:      secretName,
			Namespace: targetSharedPool.GetDataPlaneNamespace(),
		},
		Data: map[string][]byte{
			"access_key": accessKeyId,
			"secret_key": accessKeySecret,
		},
	}
	err = k8s.CreateIfNotExist(ctx, remoteCli, secret)
	if err != nil {
		tilog.Error(ctx, "failed to create secret", zap.String("secretName", secretName), zap.Error(err))
		return fmt.Errorf("failed to create secret: %w", err)
	}
	tilog.Info(ctx, "create AKSK secret success", zap.String("secretName", secretName))
	return nil
}

func getResourcesForClassicRestore(args actions.ActionArgs) (cpu, memory string, err error) {
	cpuStr, ok := args[workflow.TiBRCPUKey]
	if !ok {
		return "", "", fmt.Errorf("cpu is required, but not found in args: %v", args)
	}
	milliCPU, err := ParseCPU(cpuStr.String())
	if err != nil {
		return "", "", fmt.Errorf("failed to parse cpu %s: %w", cpuStr.String(), err)
	}
	cpu = fmt.Sprintf("%d", ComputeAvailableCPU(uint64(milliCPU)/1000))

	memoryStr, ok := args[workflow.TiBRMemoryKey]
	if !ok {
		return "", "", fmt.Errorf("memory is required, but not found in args: %v", args)
	}
	memoryInMiB, err := ParseToMiB(memoryStr.String())
	if err != nil {
		return "", "", fmt.Errorf("failed to parse memory %s: %w", memoryStr.String(), err)
	}
	memory = fmt.Sprintf("%dMi", tioperator.ComputeAvailableMemory(memoryInMiB))

	return cpu, memory, nil
}

func (a *executeClassicRestoreAction) getImageAndOptions(ctx context.Context, targetSharedPool *entity.SharedPool, restore *entity.ClassicRestore, tenantId string) (string, map[string]string, error) {
	rateLimitMBps, err := a.calculateRateLimitMBps(ctx, targetSharedPool)
	if err != nil {
		return "", nil, fmt.Errorf("failed to calculate rate limit MBps: %w", err)
	}

	if a.bizConfig.DefaultTiBRConfig.IntegrateConfigCenter {
		image, options, err := a.getImageAndOptionsFromConfigCenter(ctx, targetSharedPool, restore, tenantId)
		if err != nil {
			return "", nil, fmt.Errorf("failed to get image and options from config center: %w", err)
		}

		if options["ratelimit"] == "" && rateLimitMBps > 0 {
			options["ratelimit"] = fmt.Sprintf("%d", rateLimitMBps)
		}
		return image, options, nil
	}

	classicRestoreConfig := a.bizConfig.ClassicRestore
	imageTag := classicRestoreConfig.ImageTag
	if imageTag == "" {
		imageTag = DefaultClassicRestoreImageTag
	}
	options := classicRestoreConfig.Options
	if len(options) == 0 {
		options = DefaultClassicRestoreOptions
	}
	options["keyspace-name"] = resolveTiDBClusterKeyspaceForBackup(targetSharedPool, restore.TargetClusterID)
	if options["ratelimit"] == "" && rateLimitMBps > 0 {
		options["ratelimit"] = fmt.Sprintf("%d", rateLimitMBps)
	}

	return fmt.Sprintf("%s/%s/%s:%s", a.imageConfig.Registry, a.imageConfig.RepoPrefix, "br", imageTag), options, nil
}

func (a *executeClassicRestoreAction) getImageAndOptionsFromConfigCenter(ctx context.Context, targetSharedPool *entity.SharedPool, restore *entity.ClassicRestore, tenantId string) (string, map[string]string, error) {
	_, cfg, err := a.clusterConfigClient.GetBRComponentVersionAndConfig(ctx, targetSharedPool, tenantId, apiv1.ComponentType_CLASSIC_RESTORE)
	if err != nil {
		return "", nil, fmt.Errorf("failed to get %s version and config: %w", apiv1.ComponentType_CLASSIC_RESTORE.String(), err)
	}
	image := fmt.Sprintf("%s/%s/%s:%s",
		a.imageConfig.Registry,
		a.imageConfig.RepoPrefix,
		path.Base(cfg.GetImageRepo()),
		cfg.GetImageTag(),
	)
	tomlConfig := schemav1.NewTomlConfig(map[string]any{})
	if err := tomlConfig.UnmarshalTOML([]byte(cfg.GetConfigurationFileContent())); err != nil {
		return "", nil, fmt.Errorf("failed to unmarshal options: %w", err)
	}
	options := map[string]string{}
	for key, value := range tomlConfig.MP {
		options[key] = fmt.Sprintf("%v", value)
	}
	options["keyspace-name"] = resolveTiDBClusterKeyspaceForBackup(targetSharedPool, restore.TargetClusterID)
	// Notice: no allow user config this in configure center
	delete(options, "send-credentials-to-tikv")

	return image, options, nil
}

func (a *executeClassicRestoreAction) calculateRateLimitMBps(ctx context.Context, targetSharedPool *entity.SharedPool) (int, error) {
	// only handle rate limit on alicloud now
	if targetSharedPool.Provider.Type != apiv1.ProviderType_ALICLOUD {
		return 0, nil
	}

	topology, err := a.clustergatewayCli.GetClusterTopology(ctx, targetSharedPool, &clusteragentv1.GetClusterTopologyReq{
		ClusterId:        targetSharedPool.GetId(),
		IncludeTikvUsage: false,
	})
	if err != nil {
		return 0, fmt.Errorf("failed to get cluster topology: %w", err)
	}
	if len(topology.TikvNodes) == 0 {
		return 0, fmt.Errorf("no tikv nodes found")
	}

	totalRateLimitGbps := 30.0
	if a.bizConfig.DefaultTiBRConfig.OSSRateLimitGbps > 0 {
		totalRateLimitGbps = a.bizConfig.DefaultTiBRConfig.OSSRateLimitGbps
	}

	totalRateLimitMBps := totalRateLimitGbps * 1000.0 / 8.0
	perTiKVRateLimit := int(math.Ceil(totalRateLimitMBps / float64(len(topology.TikvNodes))))
	if perTiKVRateLimit < 1 {
		perTiKVRateLimit = 1
	}
	return perTiKVRateLimit, nil
}